feox_server/protocol/command/
executor.rs

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

feox_server/protocol/command/
executor.rs
