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

feox_server/protocol/command/
executor.rs
