use super::client::ClientOperations;
use super::command_table::{CommandDef, COMMAND_TABLE};
use super::hash::HashOperations;
use super::list::ListOperations;
use super::Command;
use crate::client_registry::ClientRegistry;
use crate::config::Config;
use crate::protocol::resp::RespValue;
use crate::watch_registry::WatchRegistry;
use bytes::Bytes;
use feoxdb::FeoxStore;
use std::sync::Arc;
fn match_pattern(key: &[u8], pattern: &str) -> bool {
let key_str = String::from_utf8_lossy(key);
glob_match(pattern, &key_str)
}
fn glob_match(pattern: &str, text: &str) -> bool {
let mut p_idx = 0;
let mut t_idx = 0;
let mut star_idx = None;
let mut star_match = None;
let pattern_bytes = pattern.as_bytes();
let text_bytes = text.as_bytes();
while t_idx < text_bytes.len() {
if p_idx < pattern_bytes.len() {
match pattern_bytes[p_idx] {
b'*' => {
star_idx = Some(p_idx);
star_match = Some(t_idx);
p_idx += 1;
}
b'?' => {
p_idx += 1;
t_idx += 1;
}
_ => {
if pattern_bytes[p_idx] == text_bytes[t_idx] {
p_idx += 1;
t_idx += 1;
} else if let Some(star) = star_idx {
p_idx = star + 1;
star_match = Some(star_match.unwrap() + 1);
t_idx = star_match.unwrap();
} else {
return false;
}
}
}
} else if let Some(star) = star_idx {
p_idx = star + 1;
star_match = Some(star_match.unwrap() + 1);
t_idx = star_match.unwrap();
} else {
return false;
}
}
while p_idx < pattern_bytes.len() && pattern_bytes[p_idx] == b'*' {
p_idx += 1;
}
p_idx == pattern_bytes.len()
}
fn extract_prefix(pattern: &str) -> &str {
for (i, ch) in pattern.char_indices() {
if ch == '*' || ch == '?' || ch == '[' {
return &pattern[..i];
}
}
pattern
}
#[derive(Clone)]
pub struct CommandExecutor {
store: Arc<FeoxStore>,
list_ops: ListOperations,
hash_ops: HashOperations,
client_ops: ClientOperations,
config: Config, start_time: std::time::Instant,
commands_processed: Arc<std::sync::atomic::AtomicU64>,
connection_id: Option<usize>,
}
impl CommandExecutor {
pub fn new(store: Arc<FeoxStore>, config: &Config) -> Self {
let list_ops = ListOperations::new(Arc::clone(&store));
let hash_ops = HashOperations::new(Arc::clone(&store));
Self {
store,
list_ops,
hash_ops,
client_ops: ClientOperations::new(),
config: config.clone(),
start_time: std::time::Instant::now(),
commands_processed: Arc::new(std::sync::atomic::AtomicU64::new(0)),
connection_id: None,
}
}
pub fn with_client_info(mut self, registry: Arc<ClientRegistry>, connection_id: usize) -> Self {
self.client_ops = ClientOperations::with_registry(registry);
self.connection_id = Some(connection_id);
self
}
pub fn check_auth(&self, password: &str) -> bool {
self.config.check_password(password)
}
#[inline(always)]
pub fn fast_set(&self, key: &[u8], value: &[u8]) -> Result<(), feoxdb::FeoxError> {
self.store.insert_with_timestamp(key, value, None)?;
WatchRegistry::notify_key_modified(key);
Ok(())
}
#[inline(always)]
pub fn fast_set_bytes(&self, key: &[u8], value: bytes::Bytes) -> Result<(), feoxdb::FeoxError> {
self.store.insert_bytes_with_timestamp(key, value, None)?;
WatchRegistry::notify_key_modified(key);
Ok(())
}
#[inline(always)]
pub fn fast_get(&self, key: &[u8]) -> Result<bytes::Bytes, feoxdb::FeoxError> {
self.store.get_bytes(key)
}
#[inline]
pub fn execute(&self, cmd: Command) -> RespValue {
use std::sync::atomic::Ordering;
self.commands_processed.fetch_add(1, Ordering::Relaxed);
match cmd {
Command::Get(key) => match self.store.get_bytes(&key) {
Ok(value) => RespValue::BulkString(Some(value)),
Err(feoxdb::FeoxError::KeyNotFound) => RespValue::BulkString(None),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::Set { key, value, ex, px } => {
let result = if let Some(seconds) = ex {
self.store
.insert_bytes_with_ttl_and_timestamp(&key, value, seconds, None)
} else if let Some(millis) = px {
self.store
.insert_bytes_with_ttl_and_timestamp(&key, value, millis / 1000, None)
} else {
self.store.insert_bytes_with_timestamp(&key, value, None)
};
match result {
Ok(_) => {
WatchRegistry::notify_key_modified(&key);
RespValue::SimpleString(Bytes::from_static(b"OK"))
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Del(keys) => {
let mut count = 0i64;
for key in &keys {
if self.store.delete(key).is_ok() {
WatchRegistry::notify_key_modified(key);
count += 1;
}
}
RespValue::Integer(count)
}
Command::Exists(keys) => {
let count = keys
.iter()
.filter(|key| self.store.contains_key(key))
.count() as i64;
RespValue::Integer(count)
}
Command::Incr(key) => self.string_incr(&key, 1),
Command::IncrBy { key, delta } => self.string_incr(&key, delta),
Command::Decr(key) => self.string_incr(&key, -1),
Command::DecrBy { key, delta } => self.string_incr(&key, -delta),
Command::Expire { key, seconds } => match self.store.update_ttl(&key, seconds) {
Ok(_) => RespValue::Integer(1),
Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(0),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::PExpire { key, milliseconds } => {
match self.store.update_ttl(&key, milliseconds / 1000) {
Ok(_) => RespValue::Integer(1),
Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(0),
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Ttl(key) => {
match self.store.get_ttl(&key) {
Ok(Some(ttl)) => RespValue::Integer(ttl as i64),
Ok(None) => RespValue::Integer(-1), Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(-2),
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::PTtl(key) => {
match self.store.get_ttl(&key) {
Ok(Some(ttl)) => RespValue::Integer((ttl * 1000) as i64),
Ok(None) => RespValue::Integer(-1), Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(-2),
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Persist(key) => match self.store.persist(&key) {
Ok(_) => RespValue::Integer(1),
Err(feoxdb::FeoxError::KeyNotFound) => RespValue::Integer(0),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::MGet(keys) => {
let values: Vec<RespValue> = keys
.into_iter()
.map(|key| match self.store.get_bytes(&key) {
Ok(value) => RespValue::BulkString(Some(value)),
Err(_) => RespValue::BulkString(None),
})
.collect();
RespValue::Array(Some(values))
}
Command::MSet(pairs) => {
for (key, value) in &pairs {
if let Err(e) = self.store.insert_with_timestamp(key, value, None) {
return RespValue::Error(format!("ERR {}", e));
}
WatchRegistry::notify_key_modified(key);
}
RespValue::SimpleString(Bytes::from_static(b"OK"))
}
Command::Ping(msg) => match msg {
Some(m) => RespValue::BulkString(Some(m)),
None => RespValue::SimpleString(Bytes::from_static(b"PONG")),
},
Command::Echo(msg) => RespValue::BulkString(Some(msg)),
Command::Config { action, args } => {
match action.to_uppercase().as_str() {
"GET" => {
if args.is_empty() {
RespValue::Array(Some(vec![]))
} else {
let mut results = Vec::new();
for arg in args {
let key = String::from_utf8_lossy(&arg).to_lowercase();
let value = match key.as_str() {
"maxmemory" => "0",
"maxclients" => "10000",
"timeout" => "0",
"tcp-keepalive" => "300",
"databases" => "16",
_ => "0",
};
results.push(RespValue::BulkString(Some(arg)));
results.push(RespValue::BulkString(Some(Bytes::from(value))));
}
RespValue::Array(Some(results))
}
}
"SET" => {
RespValue::SimpleString(Bytes::from_static(b"OK"))
}
_ => RespValue::Error(format!("ERR Unknown CONFIG subcommand '{}'", action)),
}
}
Command::Command { subcommand, args } => {
Self::handle_command_subcommand(subcommand, args)
}
Command::Quit => RespValue::SimpleString(Bytes::from_static(b"OK")),
Command::FlushDb => {
RespValue::Error("ERR FLUSHDB requires server restart. For persistent mode, also delete data files.".to_string())
}
Command::Keys(pattern) => {
let prefix = extract_prefix(&pattern);
let (start_key, end_key) = if prefix.is_empty() {
(vec![], vec![0xFF; 255])
} else if pattern == prefix {
return match self.store.get_bytes(prefix.as_bytes()) {
Ok(_) => {
let keys =
vec![RespValue::BulkString(Some(Bytes::from(prefix.to_string())))];
RespValue::Array(Some(keys))
}
Err(_) => RespValue::Array(Some(vec![])),
};
} else {
let mut end = prefix.as_bytes().to_vec();
end.push(b'~'); (prefix.as_bytes().to_vec(), end)
};
match self.store.range_query(&start_key, &end_key, 100000) {
Ok(pairs) => {
let keys: Vec<RespValue> = pairs
.into_iter()
.filter(|(key, _)| match_pattern(key, &pattern))
.map(|(key, _)| RespValue::BulkString(Some(Bytes::from(key))))
.collect();
RespValue::Array(Some(keys))
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Scan {
cursor,
count,
pattern,
} => {
let start_key = if cursor.is_empty() || cursor == b"0" {
vec![]
} else {
cursor.clone()
};
let (scan_start, scan_end) = if let Some(ref pat) = pattern {
let prefix = extract_prefix(pat);
if !prefix.is_empty() && pat.starts_with(prefix) && pat.contains('*') {
let mut end = prefix.as_bytes().to_vec();
end.push(b'~');
let actual_start = if start_key.len() > prefix.len()
&& start_key.starts_with(prefix.as_bytes())
{
start_key
} else if start_key.is_empty() {
prefix.as_bytes().to_vec()
} else {
start_key
};
(actual_start, end)
} else {
(start_key, vec![0xFF; 255])
}
} else {
(start_key, vec![0xFF; 255])
};
let fetch_count = if pattern.is_some() { count * 2 } else { count };
match self
.store
.range_query(&scan_start, &scan_end, fetch_count + 1)
{
Ok(pairs) => {
let mut keys = Vec::new();
let mut next_cursor = None;
for (key, _) in pairs.into_iter() {
if keys.len() >= count {
next_cursor = Some(key.clone());
break;
}
if let Some(ref pat) = pattern {
if !match_pattern(&key, pat) {
continue;
}
}
keys.push(RespValue::BulkString(Some(Bytes::from(key.clone()))));
}
let cursor_str = if let Some(next) = next_cursor {
Bytes::from(next)
} else {
Bytes::from_static(b"0") };
RespValue::Array(Some(vec![
RespValue::BulkString(Some(cursor_str)),
RespValue::Array(Some(keys)),
]))
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Info(section) => {
use std::sync::atomic::Ordering;
let uptime = self.start_time.elapsed().as_secs();
let commands = self.commands_processed.load(Ordering::Relaxed);
let stats = self.store.stats();
let format_bytes = |bytes: usize| -> String {
if bytes < 1024 {
format!("{}B", bytes)
} else if bytes < 1024 * 1024 {
format!("{:.1}K", bytes as f64 / 1024.0)
} else if bytes < 1024 * 1024 * 1024 {
format!("{:.1}M", bytes as f64 / (1024.0 * 1024.0))
} else {
format!("{:.1}G", bytes as f64 / (1024.0 * 1024.0 * 1024.0))
}
};
let mut info = String::new();
if section.is_none()
|| section
.as_ref()
.map(|s| s.eq_ignore_ascii_case("server"))
.unwrap_or(false)
{
info.push_str(&format!(
"# Server\r\n\
redis_version:feox-{}\r\n\
redis_mode:standalone\r\n\
arch_bits:{}\r\n\
process_id:{}\r\n\
tcp_port:6379\r\n\
uptime_in_seconds:{}\r\n",
env!("CARGO_PKG_VERSION"),
std::mem::size_of::<usize>() * 8,
std::process::id(),
uptime
));
}
if section.is_none()
|| section
.as_ref()
.map(|s| s.eq_ignore_ascii_case("memory"))
.unwrap_or(false)
{
info.push_str(&format!(
"# Memory\r\n\
used_memory:{}\r\n\
used_memory_human:{}\r\n\
used_memory_cache:{}\r\n\
used_memory_cache_human:{}\r\n",
stats.memory_usage,
format_bytes(stats.memory_usage),
stats.cache_memory,
format_bytes(stats.cache_memory)
));
}
if section.is_none()
|| section
.as_ref()
.map(|s| s.eq_ignore_ascii_case("stats"))
.unwrap_or(false)
{
info.push_str(&format!(
"# Stats\r\n\
total_connections_received:0\r\n\
total_commands_processed:{}\r\n\
instantaneous_ops_per_sec:0\r\n\
total_net_input_bytes:0\r\n\
total_net_output_bytes:0\r\n\
total_operations:{}\r\n\
total_gets:{}\r\n\
total_inserts:{}\r\n\
keyspace_hits:{}\r\n\
keyspace_misses:{}\r\n\
cache_hit_rate:{:.2}\r\n",
commands,
stats.total_operations,
stats.total_gets,
stats.total_inserts,
stats.cache_hits,
stats.cache_misses,
stats.cache_hit_rate * 100.0
));
}
if section.is_none()
|| section
.as_ref()
.map(|s| s.eq_ignore_ascii_case("keyspace"))
.unwrap_or(false)
{
info.push_str(&format!(
"# Keyspace\r\n\
db0:keys={},expires=0,avg_ttl=0\r\n",
stats.record_count
));
}
RespValue::BulkString(Some(Bytes::from(info)))
}
Command::JsonPatch { key, patch } => {
match self.store.json_patch(&key, &patch) {
Ok(_) => RespValue::SimpleString(Bytes::from_static(b"OK")),
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Cas {
key,
expected,
new_value,
} => {
match self.store.compare_and_swap(&key, &expected, &new_value) {
Ok(swapped) => RespValue::Integer(if swapped { 1 } else { 0 }),
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::LPush { key, values } => match self.list_ops.lpush(&key, values) {
Ok(count) => {
WatchRegistry::notify_key_modified(&key);
RespValue::Integer(count)
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::RPush { key, values } => match self.list_ops.rpush(&key, values) {
Ok(count) => {
WatchRegistry::notify_key_modified(&key);
RespValue::Integer(count)
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::LPop { key, count } => match self.list_ops.lpop(&key, count) {
Ok(values) => {
if values.is_empty() {
RespValue::BulkString(None)
} else {
WatchRegistry::notify_key_modified(&key);
if values.len() == 1 {
RespValue::BulkString(Some(values.into_iter().next().unwrap()))
} else {
RespValue::Array(Some(
values
.into_iter()
.map(|v| RespValue::BulkString(Some(v)))
.collect(),
))
}
}
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::RPop { key, count } => match self.list_ops.rpop(&key, count) {
Ok(values) => {
if values.is_empty() {
RespValue::BulkString(None)
} else {
WatchRegistry::notify_key_modified(&key);
if values.len() == 1 {
RespValue::BulkString(Some(values.into_iter().next().unwrap()))
} else {
RespValue::Array(Some(
values
.into_iter()
.map(|v| RespValue::BulkString(Some(v)))
.collect(),
))
}
}
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::LLen(key) => match self.list_ops.llen(&key) {
Ok(count) => RespValue::Integer(count),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::LRange { key, start, stop } => match self.list_ops.lrange(&key, start, stop) {
Ok(values) => RespValue::Array(Some(
values
.into_iter()
.map(|v| RespValue::BulkString(Some(v)))
.collect(),
)),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::LIndex { key, index } => match self.list_ops.lindex(&key, index) {
Ok(Some(value)) => RespValue::BulkString(Some(value)),
Ok(None) => RespValue::BulkString(None),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HSet { key, fields } => {
let field_refs = fields.iter().map(|(f, v)| (f.as_slice(), v.clone()));
match self.hash_ops.hset(&key, field_refs) {
Ok(count) => {
WatchRegistry::notify_key_modified(&key);
RespValue::Integer(count)
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::HGet { key, field } => match self.hash_ops.hget(&key, &field) {
Ok(Some(value)) => RespValue::BulkString(Some(value)),
Ok(None) => RespValue::BulkString(None),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HMGet { key, fields } => match self.hash_ops.hmget(&key, fields) {
Ok(values) => RespValue::Array(Some(
values
.into_iter()
.map(|v| match v {
Some(val) => RespValue::BulkString(Some(val)),
None => RespValue::BulkString(None),
})
.collect(),
)),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HDel { key, fields } => match self.hash_ops.hdel(&key, fields) {
Ok(count) => {
if count > 0 {
WatchRegistry::notify_key_modified(&key);
}
RespValue::Integer(count)
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HExists { key, field } => match self.hash_ops.hexists(&key, &field) {
Ok(exists) => RespValue::Integer(if exists { 1 } else { 0 }),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HGetAll(key) => match self.hash_ops.hgetall(&key) {
Ok(pairs) => {
let mut result = Vec::new();
for (field, value) in pairs {
result.push(RespValue::BulkString(Some(Bytes::from(field))));
result.push(RespValue::BulkString(Some(value)));
}
RespValue::Array(Some(result))
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HLen(key) => match self.hash_ops.hlen(&key) {
Ok(count) => RespValue::Integer(count),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HKeys(key) => match self.hash_ops.hkeys(&key) {
Ok(keys) => RespValue::Array(Some(
keys.into_iter()
.map(|k| RespValue::BulkString(Some(Bytes::from(k))))
.collect(),
)),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HVals(key) => match self.hash_ops.hvals(&key) {
Ok(vals) => RespValue::Array(Some(
vals.into_iter()
.map(|v| RespValue::BulkString(Some(v)))
.collect(),
)),
Err(e) => RespValue::Error(format!("ERR {}", e)),
},
Command::HIncrBy { key, field, delta } => {
match self.hash_ops.hincrby(&key, &field, delta) {
Ok(new_value) => {
WatchRegistry::notify_key_modified(&key);
RespValue::Integer(new_value)
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
Command::Auth(_) => {
if self.config.requirepass.is_none() {
RespValue::Error("-ERR Client sent AUTH, but no password is set".to_string())
} else {
RespValue::Error("-ERR AUTH failed".to_string())
}
}
Command::Client {
ref subcommand,
ref args,
} => self
.client_ops
.execute(subcommand, args, self.connection_id),
Command::Subscribe(_)
| Command::Unsubscribe(_)
| Command::PSubscribe(_)
| Command::PUnsubscribe(_)
| Command::Publish { .. }
| Command::PubSub { .. } => RespValue::Error(
"-ERR Pub/Sub commands should be handled in connection layer".to_string(),
),
Command::Multi
| Command::Exec
| Command::Discard
| Command::Watch(_)
| Command::Unwatch => RespValue::Error(
"-ERR Transaction commands should be handled in connection layer".to_string(),
),
}
}
fn string_incr(&self, key: &[u8], delta: i64) -> RespValue {
if let Ok(val) = self.store.atomic_increment(key, delta) {
WatchRegistry::notify_key_modified(key);
return RespValue::Integer(val);
}
let current: i64 = match self.store.get_bytes(key) {
Ok(bytes) => {
let s = String::from_utf8_lossy(&bytes);
match s.trim().parse::<i64>() {
Ok(v) => v,
Err(_) => {
return RespValue::Error(
"ERR value is not an integer or out of range".to_string(),
)
}
}
}
Err(feoxdb::FeoxError::KeyNotFound) => 0,
Err(e) => return RespValue::Error(format!("ERR {}", e)),
};
let new_val = current + delta;
match self
.store
.insert_bytes_with_timestamp(key, Bytes::from(new_val.to_string()), None)
{
Ok(_) => {
WatchRegistry::notify_key_modified(key);
RespValue::Integer(new_val)
}
Err(e) => RespValue::Error(format!("ERR {}", e)),
}
}
fn handle_command_subcommand(subcommand: Option<String>, args: Vec<Vec<u8>>) -> RespValue {
match subcommand.as_deref() {
None => {
let commands: Vec<RespValue> = COMMAND_TABLE
.iter()
.map(|cmd| RespValue::Array(Some(cmd.to_resp_array())))
.collect();
RespValue::Array(Some(commands))
}
Some(subcmd) => match subcmd.to_uppercase().as_str() {
"COUNT" => {
let count = COMMAND_TABLE.len() as i64;
RespValue::Integer(count)
}
"INFO" => {
if args.is_empty() {
let commands: Vec<RespValue> = COMMAND_TABLE
.iter()
.map(|cmd| RespValue::Array(Some(cmd.to_resp_array())))
.collect();
RespValue::Array(Some(commands))
} else {
let mut results = Vec::new();
for arg in args {
let cmd_name = String::from_utf8_lossy(&arg);
if let Some(cmd_def) = CommandDef::find(&cmd_name) {
results.push(RespValue::Array(Some(cmd_def.to_resp_array())));
} else {
results.push(RespValue::BulkString(None));
}
}
RespValue::Array(Some(results))
}
}
"DOCS" => RespValue::Array(Some(vec![])),
"LIST" => {
let names: Vec<RespValue> = COMMAND_TABLE
.iter()
.map(|cmd| RespValue::BulkString(Some(Bytes::from(cmd.name))))
.collect();
RespValue::Array(Some(names))
}
"HELP" => {
let help_text = vec![
"COMMAND",
"COMMAND COUNT",
"COMMAND INFO <command-name> [<command-name> ...]",
"COMMAND LIST",
"COMMAND DOCS [<command-name> ...]",
];
RespValue::Array(Some(
help_text
.into_iter()
.map(|s| RespValue::BulkString(Some(Bytes::from(s))))
.collect(),
))
}
_ => RespValue::Error(format!(
"ERR Unknown subcommand '{}'. Try COMMAND HELP.",
subcmd
)),
},
}
}
}