feox_server/protocol/command/
executor.rs

1use super::client::ClientOperations;
2use super::list::ListOperations;
3use super::Command;
4use crate::client_registry::ClientRegistry;
5use crate::config::Config;
6use crate::protocol::resp::RespValue;
7use bytes::Bytes;
8use feoxdb::FeoxStore;
9use std::sync::Arc;
10
11/// Match a key against a glob pattern
12fn match_pattern(key: &[u8], pattern: &str) -> bool {
13    let key_str = String::from_utf8_lossy(key);
14    glob_match(pattern, &key_str)
15}
16
17/// Simple glob pattern matching (* and ? support)
18fn glob_match(pattern: &str, text: &str) -> bool {
19    let mut p_idx = 0;
20    let mut t_idx = 0;
21    let mut star_idx = None;
22    let mut star_match = None;
23
24    let pattern_bytes = pattern.as_bytes();
25    let text_bytes = text.as_bytes();
26
27    while t_idx < text_bytes.len() {
28        if p_idx < pattern_bytes.len() {
29            match pattern_bytes[p_idx] {
30                b'*' => {
31                    star_idx = Some(p_idx);
32                    star_match = Some(t_idx);
33                    p_idx += 1;
34                }
35                b'?' => {
36                    p_idx += 1;
37                    t_idx += 1;
38                }
39                _ => {
40                    if pattern_bytes[p_idx] == text_bytes[t_idx] {
41                        p_idx += 1;
42                        t_idx += 1;
43                    } else if let Some(star) = star_idx {
44                        p_idx = star + 1;
45                        star_match = Some(star_match.unwrap() + 1);
46                        t_idx = star_match.unwrap();
47                    } else {
48                        return false;
49                    }
50                }
51            }
52        } else if let Some(star) = star_idx {
53            p_idx = star + 1;
54            star_match = Some(star_match.unwrap() + 1);
55            t_idx = star_match.unwrap();
56        } else {
57            return false;
58        }
59    }
60
61    // Check remaining pattern characters (should only be *)
62    while p_idx < pattern_bytes.len() && pattern_bytes[p_idx] == b'*' {
63        p_idx += 1;
64    }
65
66    p_idx == pattern_bytes.len()
67}
68
69/// Extract prefix from a pattern (everything before the first wildcard)
70fn extract_prefix(pattern: &str) -> &str {
71    for (i, ch) in pattern.char_indices() {
72        if ch == '*' || ch == '?' || ch == '[' {
73            return &pattern[..i];
74        }
75    }
76    pattern
77}
78
79/// Executes parsed Redis commands against a FeoxStore
80///
81/// Translates between Redis protocol semantics and FeOx operations.
82#[derive(Clone)]
83pub struct CommandExecutor {
84    store: Arc<FeoxStore>,
85    list_ops: ListOperations,
86    client_ops: ClientOperations,
87    config: Config, // Store config for auth checking
88    start_time: std::time::Instant,
89    commands_processed: Arc<std::sync::atomic::AtomicU64>,
90    connection_id: Option<usize>,
91}
92
93impl CommandExecutor {
94    /// Create a new command executor with the given store and config
95    pub fn new(store: Arc<FeoxStore>, config: &Config) -> Self {
96        let list_ops = ListOperations::new(Arc::clone(&store));
97        Self {
98            store,
99            list_ops,
100            client_ops: ClientOperations::new(),
101            config: config.clone(),
102            start_time: std::time::Instant::now(),
103            commands_processed: Arc::new(std::sync::atomic::AtomicU64::new(0)),
104            connection_id: None,
105        }
106    }
107
108    /// Set the client registry and connection ID for CLIENT command support
109    pub fn with_client_info(mut self, registry: Arc<ClientRegistry>, connection_id: usize) -> Self {
110        self.client_ops = ClientOperations::with_registry(registry);
111        self.connection_id = Some(connection_id);
112        self
113    }
114
115    /// Check if password is correct
116    pub fn check_auth(&self, password: &str) -> bool {
117        self.config.check_password(password)
118    }
119
120    /// Execute a command and return RESP response
121    #[inline]
122    pub fn execute(&self, cmd: Command) -> RespValue {
123        use std::sync::atomic::Ordering;
124
125        // Increment command counter
126        self.commands_processed.fetch_add(1, Ordering::Relaxed);
127
128        match cmd {
129            Command::Get(key) => match self.store.get_bytes(&key) {
130                Ok(value) => RespValue::BulkString(Some(value)),
131                Err(feoxdb::FeoxError::KeyNotFound) => RespValue::BulkString(None),
132                Err(e) => RespValue::Error(format!("ERR {}", e)),
133            },
134
135            Command::Set { key, value, ex, px } => {
136                // Pass None to let FeOx generate a new timestamp
137                let result = if let Some(seconds) = ex {
138                    self.store
139                        .insert_with_ttl_and_timestamp(&key, &value, seconds, None)
140                } else if let Some(millis) = px {
141                    self.store
142                        .insert_with_ttl_and_timestamp(&key, &value, millis / 1000, None)
143                } else {
144                    self.store.insert_with_timestamp(&key, &value, None)
145                };
146
147                match result {
148                    Ok(_) => RespValue::SimpleString(Bytes::from_static(b"OK")),
149                    Err(e) => RespValue::Error(format!("ERR {}", e)),
150                }
151            }
152
153            Command::Del(keys) => {
154                let mut count = 0i64;
155                for key in keys {
156                    if self.store.delete(&key).is_ok() {
157                        count += 1;
158                    }
159                }
160                RespValue::Integer(count)
161            }
162
163            Command::Exists(keys) => {
164                let count = keys
165                    .iter()
166                    .filter(|key| self.store.contains_key(key))
167                    .count() as i64;
168                RespValue::Integer(count)
169            }
170
171            Command::Incr(key) => match self.store.atomic_increment(&key, 1) {
172                Ok(val) => RespValue::Integer(val),
173                Err(e) => RespValue::Error(format!("ERR {}", e)),
174            },
175
176            Command::IncrBy { key, delta } => match self.store.atomic_increment(&key, delta) {
177                Ok(val) => RespValue::Integer(val),
178                Err(e) => RespValue::Error(format!("ERR {}", e)),
179            },
180
181            Command::Decr(key) => match self.store.atomic_increment(&key, -1) {
182                Ok(val) => RespValue::Integer(val),
183                Err(e) => RespValue::Error(format!("ERR {}", e)),
184            },
185
186            Command::DecrBy { key, delta } => match self.store.atomic_increment(&key, -delta) {
187                Ok(val) => RespValue::Integer(val),
188                Err(e) => RespValue::Error(format!("ERR {}", e)),
189            },
190
191            Command::Expire { key, seconds } => match self.store.update_ttl(&key, seconds) {
192                Ok(_) => RespValue::Integer(1),
193                Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(0),
194                Err(e) => RespValue::Error(format!("ERR {}", e)),
195            },
196
197            Command::PExpire { key, milliseconds } => {
198                match self.store.update_ttl(&key, milliseconds / 1000) {
199                    Ok(_) => RespValue::Integer(1),
200                    Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(0),
201                    Err(e) => RespValue::Error(format!("ERR {}", e)),
202                }
203            }
204
205            Command::Ttl(key) => {
206                match self.store.get_ttl(&key) {
207                    Ok(Some(ttl)) => RespValue::Integer(ttl as i64),
208                    Ok(None) => RespValue::Integer(-1), // No TTL
209                    Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(-2),
210                    Err(e) => RespValue::Error(format!("ERR {}", e)),
211                }
212            }
213
214            Command::PTtl(key) => {
215                match self.store.get_ttl(&key) {
216                    Ok(Some(ttl)) => RespValue::Integer((ttl * 1000) as i64),
217                    Ok(None) => RespValue::Integer(-1), // No TTL
218                    Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(-2),
219                    Err(e) => RespValue::Error(format!("ERR {}", e)),
220                }
221            }
222
223            Command::Persist(key) => match self.store.persist(&key) {
224                Ok(_) => RespValue::Integer(1),
225                Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(0),
226                Err(e) => RespValue::Error(format!("ERR {}", e)),
227            },
228
229            Command::MGet(keys) => {
230                let values: Vec<RespValue> = keys
231                    .into_iter()
232                    .map(|key| match self.store.get_bytes(&key) {
233                        Ok(value) => RespValue::BulkString(Some(value)),
234                        Err(_) => RespValue::BulkString(None),
235                    })
236                    .collect();
237                RespValue::Array(Some(values))
238            }
239
240            Command::MSet(pairs) => {
241                for (key, value) in pairs {
242                    // Pass None to let FeOx generate a new timestamp
243                    if let Err(e) = self.store.insert_with_timestamp(&key, &value, None) {
244                        return RespValue::Error(format!("ERR {}", e));
245                    }
246                }
247                RespValue::SimpleString(Bytes::from_static(b"OK"))
248            }
249
250            Command::Ping(msg) => match msg {
251                Some(m) => RespValue::BulkString(Some(m)),
252                None => RespValue::SimpleString(Bytes::from_static(b"PONG")),
253            },
254
255            Command::Echo(msg) => RespValue::BulkString(Some(msg)),
256
257            Command::Config { action, args } => {
258                match action.to_uppercase().as_str() {
259                    "GET" => {
260                        // Return empty config for compatibility with redis-benchmark
261                        if args.is_empty() {
262                            RespValue::Array(Some(vec![]))
263                        } else {
264                            // Return nil for any specific config request
265                            let mut results = Vec::new();
266                            for arg in args {
267                                results.push(RespValue::BulkString(Some(arg)));
268                                results.push(RespValue::BulkString(None)); // nil value
269                            }
270                            RespValue::Array(Some(results))
271                        }
272                    }
273                    "SET" => {
274                        // Pretend to set config successfully
275                        RespValue::SimpleString(Bytes::from_static(b"OK"))
276                    }
277                    _ => RespValue::Error(format!("ERR Unknown CONFIG subcommand '{}'", action)),
278                }
279            }
280
281            Command::Command => {
282                // Return supported commands in Redis COMMAND format
283                // Each command entry: [name, arity, flags, first_key, last_key, step]
284                let commands = vec![
285                    // Basic commands
286                    vec![
287                        RespValue::BulkString(Some(Bytes::from_static(b"GET"))),
288                        RespValue::Integer(2), // arity (command + 1 key)
289                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
290                            Bytes::from_static(b"readonly"),
291                        ))])),
292                        RespValue::Integer(1), // first key position
293                        RespValue::Integer(1), // last key position
294                        RespValue::Integer(1), // step
295                    ],
296                    vec![
297                        RespValue::BulkString(Some(Bytes::from_static(b"SET"))),
298                        RespValue::Integer(-3), // arity (variable, min 3)
299                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
300                            Bytes::from_static(b"write"),
301                        ))])),
302                        RespValue::Integer(1),
303                        RespValue::Integer(1),
304                        RespValue::Integer(1),
305                    ],
306                    vec![
307                        RespValue::BulkString(Some(Bytes::from_static(b"DEL"))),
308                        RespValue::Integer(-2), // arity (variable, min 2)
309                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
310                            Bytes::from_static(b"write"),
311                        ))])),
312                        RespValue::Integer(1),
313                        RespValue::Integer(-1), // all args are keys
314                        RespValue::Integer(1),
315                    ],
316                    vec![
317                        RespValue::BulkString(Some(Bytes::from_static(b"EXISTS"))),
318                        RespValue::Integer(-2),
319                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
320                            Bytes::from_static(b"readonly"),
321                        ))])),
322                        RespValue::Integer(1),
323                        RespValue::Integer(-1),
324                        RespValue::Integer(1),
325                    ],
326                    // Atomic operations
327                    vec![
328                        RespValue::BulkString(Some(Bytes::from_static(b"INCR"))),
329                        RespValue::Integer(2),
330                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
331                            Bytes::from_static(b"write"),
332                        ))])),
333                        RespValue::Integer(1),
334                        RespValue::Integer(1),
335                        RespValue::Integer(1),
336                    ],
337                    vec![
338                        RespValue::BulkString(Some(Bytes::from_static(b"DECR"))),
339                        RespValue::Integer(2),
340                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
341                            Bytes::from_static(b"write"),
342                        ))])),
343                        RespValue::Integer(1),
344                        RespValue::Integer(1),
345                        RespValue::Integer(1),
346                    ],
347                    // TTL commands
348                    vec![
349                        RespValue::BulkString(Some(Bytes::from_static(b"EXPIRE"))),
350                        RespValue::Integer(3),
351                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
352                            Bytes::from_static(b"write"),
353                        ))])),
354                        RespValue::Integer(1),
355                        RespValue::Integer(1),
356                        RespValue::Integer(1),
357                    ],
358                    vec![
359                        RespValue::BulkString(Some(Bytes::from_static(b"TTL"))),
360                        RespValue::Integer(2),
361                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
362                            Bytes::from_static(b"readonly"),
363                        ))])),
364                        RespValue::Integer(1),
365                        RespValue::Integer(1),
366                        RespValue::Integer(1),
367                    ],
368                    // Bulk operations
369                    vec![
370                        RespValue::BulkString(Some(Bytes::from_static(b"MGET"))),
371                        RespValue::Integer(-2),
372                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
373                            Bytes::from_static(b"readonly"),
374                        ))])),
375                        RespValue::Integer(1),
376                        RespValue::Integer(-1),
377                        RespValue::Integer(1),
378                    ],
379                    vec![
380                        RespValue::BulkString(Some(Bytes::from_static(b"MSET"))),
381                        RespValue::Integer(-3),
382                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
383                            Bytes::from_static(b"write"),
384                        ))])),
385                        RespValue::Integer(1),
386                        RespValue::Integer(-1),
387                        RespValue::Integer(2), // key-value pairs
388                    ],
389                    // Server commands
390                    vec![
391                        RespValue::BulkString(Some(Bytes::from_static(b"PING"))),
392                        RespValue::Integer(-1),
393                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
394                            Bytes::from_static(b"fast"),
395                        ))])),
396                        RespValue::Integer(0),
397                        RespValue::Integer(0),
398                        RespValue::Integer(0),
399                    ],
400                    // FeOx-specific
401                    vec![
402                        RespValue::BulkString(Some(Bytes::from_static(b"JSONPATCH"))),
403                        RespValue::Integer(3),
404                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
405                            Bytes::from_static(b"write"),
406                        ))])),
407                        RespValue::Integer(1),
408                        RespValue::Integer(1),
409                        RespValue::Integer(1),
410                    ],
411                    vec![
412                        RespValue::BulkString(Some(Bytes::from_static(b"CAS"))),
413                        RespValue::Integer(4),
414                        RespValue::Array(Some(vec![RespValue::BulkString(Some(
415                            Bytes::from_static(b"write"),
416                        ))])),
417                        RespValue::Integer(1),
418                        RespValue::Integer(1),
419                        RespValue::Integer(1),
420                    ],
421                ];
422
423                RespValue::Array(Some(
424                    commands
425                        .into_iter()
426                        .map(|cmd| RespValue::Array(Some(cmd)))
427                        .collect(),
428                ))
429            }
430
431            Command::Quit => RespValue::SimpleString(Bytes::from_static(b"OK")),
432
433            Command::FlushDb => {
434                // FeOx doesn't have a direct flush method
435                // For in-memory mode: would need to recreate the store
436                // For persistent mode: would need to delete files and recreate
437                // Since we can't recreate the store from here, return error
438                RespValue::Error("ERR FLUSHDB requires server restart. For persistent mode, also delete data files.".to_string())
439            }
440
441            Command::Keys(pattern) => {
442                // Use range_query to get all keys, then filter by pattern
443                let prefix = extract_prefix(&pattern);
444
445                // Calculate end key for prefix scan
446                let (start_key, end_key) = if prefix.is_empty() {
447                    // Scan all keys
448                    (vec![], vec![0xFF; 255])
449                } else if pattern == prefix {
450                    // Exact match, no wildcards
451                    return match self.store.get_bytes(prefix.as_bytes()) {
452                        Ok(_) => {
453                            let keys =
454                                vec![RespValue::BulkString(Some(Bytes::from(prefix.to_string())))];
455                            RespValue::Array(Some(keys))
456                        }
457                        Err(_) => RespValue::Array(Some(vec![])),
458                    };
459                } else {
460                    // Prefix scan with pattern matching
461                    let mut end = prefix.as_bytes().to_vec();
462                    end.push(b'~'); // Use tilde as upper bound
463                    (prefix.as_bytes().to_vec(), end)
464                };
465
466                // Get keys using range_query
467                match self.store.range_query(&start_key, &end_key, 100000) {
468                    Ok(pairs) => {
469                        let keys: Vec<RespValue> = pairs
470                            .into_iter()
471                            .filter(|(key, _)| match_pattern(key, &pattern))
472                            .map(|(key, _)| RespValue::BulkString(Some(Bytes::from(key))))
473                            .collect();
474                        RespValue::Array(Some(keys))
475                    }
476                    Err(e) => RespValue::Error(format!("ERR {}", e)),
477                }
478            }
479
480            Command::Scan {
481                cursor,
482                count,
483                pattern,
484            } => {
485                // Parse cursor (empty or "0" means start from beginning)
486                let start_key = if cursor.is_empty() || cursor == b"0" {
487                    vec![]
488                } else {
489                    cursor.clone()
490                };
491
492                // For prefix patterns, optimize the scan range
493                let (scan_start, scan_end) = if let Some(ref pat) = pattern {
494                    let prefix = extract_prefix(pat);
495                    if !prefix.is_empty() && pat.starts_with(prefix) && pat.contains('*') {
496                        // Optimize for prefix patterns like "user:*"
497                        let mut end = prefix.as_bytes().to_vec();
498                        end.push(b'~');
499
500                        // Adjust start if cursor is past the prefix
501                        let actual_start = if start_key.len() > prefix.len()
502                            && start_key.starts_with(prefix.as_bytes())
503                        {
504                            start_key
505                        } else if start_key.is_empty() {
506                            prefix.as_bytes().to_vec()
507                        } else {
508                            start_key
509                        };
510
511                        (actual_start, end)
512                    } else {
513                        (start_key, vec![0xFF; 255])
514                    }
515                } else {
516                    (start_key, vec![0xFF; 255])
517                };
518
519                // Get keys using range_query (get a bit more than requested to ensure we have enough after filtering)
520                let fetch_count = if pattern.is_some() { count * 2 } else { count };
521                match self
522                    .store
523                    .range_query(&scan_start, &scan_end, fetch_count + 1)
524                {
525                    Ok(pairs) => {
526                        let mut keys = Vec::new();
527                        let mut next_cursor = None;
528
529                        for (key, _) in pairs.into_iter() {
530                            // Skip if we've collected enough
531                            if keys.len() >= count {
532                                next_cursor = Some(key.clone());
533                                break;
534                            }
535
536                            // Apply pattern filter if specified
537                            if let Some(ref pat) = pattern {
538                                if !match_pattern(&key, pat) {
539                                    continue;
540                                }
541                            }
542
543                            keys.push(RespValue::BulkString(Some(Bytes::from(key.clone()))));
544                        }
545
546                        // Format response: [cursor, [keys...]]
547                        let cursor_str = if let Some(next) = next_cursor {
548                            Bytes::from(next)
549                        } else {
550                            Bytes::from_static(b"0") // End of iteration
551                        };
552
553                        RespValue::Array(Some(vec![
554                            RespValue::BulkString(Some(cursor_str)),
555                            RespValue::Array(Some(keys)),
556                        ]))
557                    }
558                    Err(e) => RespValue::Error(format!("ERR {}", e)),
559                }
560            }
561
562            Command::Info(section) => {
563                use std::sync::atomic::Ordering;
564
565                // Get actual stats
566                let uptime = self.start_time.elapsed().as_secs();
567                let commands = self.commands_processed.load(Ordering::Relaxed);
568                let stats = self.store.stats();
569
570                // Format memory size
571                let format_bytes = |bytes: usize| -> String {
572                    if bytes < 1024 {
573                        format!("{}B", bytes)
574                    } else if bytes < 1024 * 1024 {
575                        format!("{:.1}K", bytes as f64 / 1024.0)
576                    } else if bytes < 1024 * 1024 * 1024 {
577                        format!("{:.1}M", bytes as f64 / (1024.0 * 1024.0))
578                    } else {
579                        format!("{:.1}G", bytes as f64 / (1024.0 * 1024.0 * 1024.0))
580                    }
581                };
582
583                let mut info = String::new();
584
585                // Server section
586                if section.is_none()
587                    || section
588                        .as_ref()
589                        .map(|s| s.eq_ignore_ascii_case("server"))
590                        .unwrap_or(false)
591                {
592                    info.push_str(&format!(
593                        "# Server\r\n\
594                        redis_version:feox-{}\r\n\
595                        redis_mode:standalone\r\n\
596                        process_id:{}\r\n\
597                        tcp_port:6379\r\n\
598                        uptime_in_seconds:{}\r\n",
599                        env!("CARGO_PKG_VERSION"),
600                        std::process::id(),
601                        uptime
602                    ));
603                }
604
605                // Memory section
606                if section.is_none()
607                    || section
608                        .as_ref()
609                        .map(|s| s.eq_ignore_ascii_case("memory"))
610                        .unwrap_or(false)
611                {
612                    info.push_str(&format!(
613                        "# Memory\r\n\
614                        used_memory:{}\r\n\
615                        used_memory_human:{}\r\n\
616                        used_memory_cache:{}\r\n\
617                        used_memory_cache_human:{}\r\n",
618                        stats.memory_usage,
619                        format_bytes(stats.memory_usage),
620                        stats.cache_memory,
621                        format_bytes(stats.cache_memory)
622                    ));
623                }
624
625                // Stats section
626                if section.is_none()
627                    || section
628                        .as_ref()
629                        .map(|s| s.eq_ignore_ascii_case("stats"))
630                        .unwrap_or(false)
631                {
632                    info.push_str(&format!(
633                        "# Stats\r\n\
634                        total_connections_received:0\r\n\
635                        total_commands_processed:{}\r\n\
636                        instantaneous_ops_per_sec:0\r\n\
637                        total_net_input_bytes:0\r\n\
638                        total_net_output_bytes:0\r\n\
639                        total_operations:{}\r\n\
640                        total_gets:{}\r\n\
641                        total_inserts:{}\r\n\
642                        keyspace_hits:{}\r\n\
643                        keyspace_misses:{}\r\n\
644                        cache_hit_rate:{:.2}\r\n",
645                        commands,
646                        stats.total_operations,
647                        stats.total_gets,
648                        stats.total_inserts,
649                        stats.cache_hits,
650                        stats.cache_misses,
651                        stats.cache_hit_rate * 100.0
652                    ));
653                }
654
655                // Keyspace section
656                if section.is_none()
657                    || section
658                        .as_ref()
659                        .map(|s| s.eq_ignore_ascii_case("keyspace"))
660                        .unwrap_or(false)
661                {
662                    info.push_str(&format!(
663                        "# Keyspace\r\n\
664                        db0:keys={},expires=0,avg_ttl=0\r\n",
665                        stats.record_count
666                    ));
667                }
668
669                RespValue::BulkString(Some(Bytes::from(info)))
670            }
671
672            Command::JsonPatch { key, patch } => {
673                // Use FeOx's native json_patch method
674                match self.store.json_patch(&key, &patch) {
675                    Ok(_) => RespValue::SimpleString(Bytes::from_static(b"OK")),
676                    Err(e) => RespValue::Error(format!("ERR {}", e)),
677                }
678            }
679
680            Command::Cas {
681                key,
682                expected,
683                new_value,
684            } => {
685                // Use FeOx's native compare_and_swap method
686                match self.store.compare_and_swap(&key, &expected, &new_value) {
687                    Ok(swapped) => RespValue::Integer(if swapped { 1 } else { 0 }),
688                    Err(e) => RespValue::Error(format!("ERR {}", e)),
689                }
690            }
691
692            Command::LPush { key, values } => match self.list_ops.lpush(&key, values) {
693                Ok(count) => RespValue::Integer(count),
694                Err(e) => RespValue::Error(format!("ERR {}", e)),
695            },
696
697            Command::RPush { key, values } => match self.list_ops.rpush(&key, values) {
698                Ok(count) => RespValue::Integer(count),
699                Err(e) => RespValue::Error(format!("ERR {}", e)),
700            },
701
702            Command::LPop { key, count } => match self.list_ops.lpop(&key, count) {
703                Ok(values) => {
704                    if values.is_empty() {
705                        RespValue::BulkString(None)
706                    } else if values.len() == 1 {
707                        RespValue::BulkString(Some(values.into_iter().next().unwrap()))
708                    } else {
709                        RespValue::Array(Some(
710                            values
711                                .into_iter()
712                                .map(|v| RespValue::BulkString(Some(v)))
713                                .collect(),
714                        ))
715                    }
716                }
717                Err(e) => RespValue::Error(format!("ERR {}", e)),
718            },
719
720            Command::RPop { key, count } => match self.list_ops.rpop(&key, count) {
721                Ok(values) => {
722                    if values.is_empty() {
723                        RespValue::BulkString(None)
724                    } else if values.len() == 1 {
725                        RespValue::BulkString(Some(values.into_iter().next().unwrap()))
726                    } else {
727                        RespValue::Array(Some(
728                            values
729                                .into_iter()
730                                .map(|v| RespValue::BulkString(Some(v)))
731                                .collect(),
732                        ))
733                    }
734                }
735                Err(e) => RespValue::Error(format!("ERR {}", e)),
736            },
737
738            Command::LLen(key) => match self.list_ops.llen(&key) {
739                Ok(count) => RespValue::Integer(count),
740                Err(e) => RespValue::Error(format!("ERR {}", e)),
741            },
742
743            Command::LRange { key, start, stop } => match self.list_ops.lrange(&key, start, stop) {
744                Ok(values) => RespValue::Array(Some(
745                    values
746                        .into_iter()
747                        .map(|v| RespValue::BulkString(Some(v)))
748                        .collect(),
749                )),
750                Err(e) => RespValue::Error(format!("ERR {}", e)),
751            },
752
753            Command::LIndex { key, index } => match self.list_ops.lindex(&key, index) {
754                Ok(Some(value)) => RespValue::BulkString(Some(value)),
755                Ok(None) => RespValue::BulkString(None),
756                Err(e) => RespValue::Error(format!("ERR {}", e)),
757            },
758
759            Command::Auth(_) => {
760                // This should be handled in connection.rs
761                // If we get here, it means auth is not configured
762                if self.config.requirepass.is_none() {
763                    RespValue::Error("-ERR Client sent AUTH, but no password is set".to_string())
764                } else {
765                    // Should not reach here
766                    RespValue::Error("-ERR AUTH failed".to_string())
767                }
768            }
769
770            Command::Client {
771                ref subcommand,
772                ref args,
773            } => self
774                .client_ops
775                .execute(subcommand, args, self.connection_id),
776
777            // Pub/Sub commands are handled in connection.rs
778            Command::Subscribe(_)
779            | Command::Unsubscribe(_)
780            | Command::PSubscribe(_)
781            | Command::PUnsubscribe(_)
782            | Command::Publish { .. }
783            | Command::PubSub { .. } => RespValue::Error(
784                "-ERR Pub/Sub commands should be handled in connection layer".to_string(),
785            ),
786        }
787    }
788}
feox_server/protocol/command/executor.rs

feox_server/protocol/command/
executor.rs
