feox_server/protocol/command/
executor.rs

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

feox_server/protocol/command/
executor.rs
