feox_server/protocol/command/
executor.rs

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

feox_server/protocol/command/
executor.rs
