kevy 6.4.0

//! `CONFIG GET / SET / REWRITE / RESETSTAT` subhandlers and the
//! `Config` → key-value flattener they need. Split out of
//! `super::mod` so the parent file stays under the project's 500-LOC cap.
//!
//! `CONFIG SET` and `CONFIG REWRITE` are now real:
//! - SET validates against the **hot-settable matrix** locked in
//!   `V1.0-BOUNDARY.md`. Hot-settable knobs (`maxmemory`,
//!   `maxmemory-policy`, `appendfsync`, `auto-aof-rewrite-*`, `hz`,
//!   `maxmemory-samples`, `loglevel`, `logfile`-stdout/stderr) build
//!   a fresh `Arc<Config>` and atomically swap the live config via
//!   `RuntimeState::config_replace`. Per-shard re-application happens
//!   lazily on the next tick via
//!   `kevy_rt::Commands::live_runtime_config` (~100 ms upper bound on
//!   propagation; well under Redis's "best-effort" semantics).
//! - Non-hot-settable fields (`bind`, `port`, `threads`, `dir`,
//!   `appendonly`, `logfile`-with-path) return Redis's canonical
//!   `ERR ... can't be changed at runtime` form.
//! - REWRITE re-emits the live config via `Config::to_toml_string`
//!   and rename-overwrites the source file atomically. Per the
//!   hot-settable matrix, inline comments are NOT preserved; the
//!   reply notes this.

use std::path::PathBuf;
use std::sync::Arc;

use kevy_config::{AppendFsync, Config, EvictionPolicy, LogLevel, LogOutput, parse_size};
use kevy_resp::{
    ArgvView, RespVersion, encode_array_len, encode_bulk, encode_error, encode_map_header,
    encode_simple_string,
};

use crate::state::Ctx;

use super::{appendfsync_str, eviction_str, log_level_str, wrong_args};

pub(crate) fn cmd_config<A: ArgvView + ?Sized>(
    ctx: &Ctx<'_>,
    args: &A,
    out: &mut Vec<u8>,
    proto: RespVersion,
) {
    let sub = match args.get(1) {
        Some(s) => s.to_ascii_uppercase(),
        None => return wrong_args(out, "config"),
    };
    match sub.as_slice() {
        b"GET" => cmd_config_get(&ctx.state.config(), args, out, proto),
        b"SET" => cmd_config_set(ctx, args, out),
        b"REWRITE" => cmd_config_rewrite(ctx, out),
        b"RESETSTAT" => encode_simple_string(out, "OK"),
        _ => encode_error(
            out,
            &format!(
                "ERR unknown CONFIG subcommand '{}'",
                String::from_utf8_lossy(args.get(1).unwrap_or(&[][..]))
            ),
        ),
    }
}

fn cmd_config_get<A: ArgvView + ?Sized>(
    cfg: &Config,
    args: &A,
    out: &mut Vec<u8>,
    proto: RespVersion,
) {
    if args.len() < 3 {
        return wrong_args(out, "config|get");
    }
    // CONFIG GET pattern1 [pattern2 ...] — collect all (key, value) pairs
    // whose key matches any of the requested glob patterns. Reply shape:
    // V2 — flat `*2N\r\n[k1, v1, k2, v2, ...]` array (Redis legacy).
    // V3 — `%N\r\n[k1, v1, k2, v2, ...]` Map (per the RESP3 spec — kv
    //      replies are real maps).
    let mut hits: Vec<(&'static str, String)> = Vec::new();
    for i in 2..args.len() {
        let pat = args[i].to_ascii_lowercase();
        for (key, val) in config_pairs(cfg) {
            if glob_match(&pat, key.as_bytes()) && !hits.iter().any(|(k, _)| *k == key) {
                hits.push((key, val));
            }
        }
    }
    match proto {
        RespVersion::V2 => encode_array_len(out, (hits.len() * 2) as i64),
        RespVersion::V3 => encode_map_header(out, hits.len() as i64),
    }
    for (k, v) in hits {
        encode_bulk(out, k.as_bytes());
        encode_bulk(out, v.as_bytes());
    }
}

fn cmd_config_set<A: ArgvView + ?Sized>(ctx: &Ctx<'_>, args: &A, out: &mut Vec<u8>) {
    if args.len() != 4 {
        return wrong_args(out, "config|set");
    }
    let key = args[2].to_ascii_lowercase();
    let value = &args[3];
    // Record the CONFIG SET event to the audit log (if enabled).
    let v_slice: &[u8] = value;
    ctx.state.obs.audit_record(&[&b"CONFIG"[..], &b"SET"[..], &key[..], v_slice]);
    let live = ctx.state.config();
    let mut new_cfg = (*live).clone();
    match apply_hot_set(&mut new_cfg, &key, value) {
        Ok(()) => {
            ctx.state.config_replace(Arc::new(new_cfg));
            encode_simple_string(out, "OK");
        }
        Err(SetError::ReadOnly(k)) => encode_error(
            out,
            &format!("ERR config setting '{k}' can't be changed at runtime, restart required"),
        ),
        Err(SetError::Unknown(k)) => {
            encode_error(out, &format!("ERR Unknown CONFIG SET parameter: '{k}'"))
        }
        Err(SetError::BadValue { key, reason }) => {
            encode_error(out, &format!("ERR CONFIG SET failed for '{key}': {reason}"))
        }
    }
}

fn cmd_config_rewrite(ctx: &Ctx<'_>, out: &mut Vec<u8>) {
    // Audit the admin event.
    ctx.state.obs.audit_record(&[&b"CONFIG"[..], &b"REWRITE"[..]]);
    let cfg = ctx.state.config();
    let Some(path) = cfg.source_path.clone() else {
        return encode_error(out, "ERR The server is running without a config file");
    };
    let text = rewrite_text(&cfg, &path);
    match atomic_write(&path, text.as_bytes()) {
        Ok(()) => encode_simple_string(out, "OK"),
        Err(e) => encode_error(
            out,
            &format!("ERR CONFIG REWRITE could not write {}: {e}", path.display()),
        ),
    }
}

/// Build the rewrite payload, preferring the comment-preserving path
/// (re-parse original source line-by-line, splice values in place) and
/// falling back to [`Config::to_toml_string`] when the source can't be
/// read or re-parsed.
fn rewrite_text(cfg: &Config, path: &PathBuf) -> String {
    match std::fs::read_to_string(path) {
        Ok(src) => match cfg.to_toml_string_preserving(&src) {
            Ok(t) => t,
            Err(e) => {
                eprintln!(
                    "kevy: CONFIG REWRITE comment-preserving re-parse of {} \
                     failed ({e}); falling back to standard template (comments lost)",
                    path.display(),
                );
                cfg.to_toml_string()
            }
        },
        Err(e) => {
            eprintln!(
                "kevy: CONFIG REWRITE could not read {} for comment preservation \
                 ({e}); falling back to standard template (comments lost)",
                path.display(),
            );
            cfg.to_toml_string()
        }
    }
}

/// Atomic-rename file write: dump to `<path>.rewrite` with fsync,
/// then `rename(2)` over the live path. Tolerates the temp-file
/// existing from a prior crashed rewrite (overwritten on next try).
fn atomic_write(path: &PathBuf, bytes: &[u8]) -> std::io::Result<()> {
    use std::io::Write;
    let mut tmp = path.clone();
    let new_name = match path.file_name() {
        Some(n) => {
            let mut s = n.to_os_string();
            s.push(".rewrite");
            s
        }
        None => return Err(std::io::Error::other("CONFIG REWRITE path has no file name")),
    };
    tmp.set_file_name(new_name);
    let mut f = std::fs::OpenOptions::new().create(true).write(true).truncate(true).open(&tmp)?;
    f.write_all(bytes)?;
    f.sync_data()?;
    drop(f);
    std::fs::rename(&tmp, path)?;
    Ok(())
}

#[derive(Debug)]
enum SetError {
    /// Field exists but the hot-settable matrix marks it as
    /// requiring a restart (bind, port, threads, dir, appendonly,
    /// logfile-with-path).
    ReadOnly(String),
    /// Field name is not recognised by the schema at all.
    Unknown(String),
    /// Field exists + is hot-settable, but the value didn't parse.
    BadValue { key: String, reason: String },
}

/// Hot-set one field on `cfg` based on its Redis-style key. Returns
/// `Ok(())` on a clean apply or an [`SetError`] describing the refusal.
fn apply_hot_set(cfg: &mut Config, key: &[u8], value: &[u8]) -> Result<(), SetError> {
    let key_str = std::str::from_utf8(key)
        .map_err(|_| SetError::Unknown(String::from_utf8_lossy(key).into_owned()))?;
    let value_str = std::str::from_utf8(value).map_err(|_| SetError::BadValue {
        key: key_str.to_string(),
        reason: "value is not valid UTF-8".to_string(),
    })?;
    match key_str {
        "maxmemory" | "maxmemory-policy" => set_memory(cfg, key_str, value_str),
        "appendfsync"
        | "auto-aof-rewrite-percentage"
        | "auto-aof-rewrite-min-size"
        | "auto-aof-rewrite-bytes"
        | "auto-aof-rewrite-interval-secs" => set_persistence(cfg, key_str, value_str),
        "hz" | "maxmemory-samples" => set_expiry(cfg, key_str, value_str),
        "loglevel" | "logfile" => set_log(cfg, key_str, value_str),
        // Redis spells it with hyphens on the wire and kevy's TOML uses
        // underscores; the engine supported the feature from the config file
        // and simply had no wire path to it. Spring Data Redis's key-expiry
        // listener, the socket.io redis adapter and several job queues send
        // `CONFIG SET notify-keyspace-events Ex` as their first act on a new
        // connection, so "unknown parameter" met them in the first second.
        "notify-keyspace-events" => set_notification(cfg, key_str, value_str),
        // Hot-settable ONLY as a budget change: the shard tick
        // re-resolves + re-applies it (the maxmemory precedent).
        // Turning tiering on/off needs the vlog lifecycle — a restart;
        // the spill dir likewise.
        "tiering-budget" => {
            cfg.tiering.budget = Some(
                kevy_config::TierBudgetSpec::parse(value_str)
                    .map_err(|reason| SetError::BadValue { key: key_str.to_string(), reason })?,
            );
            Ok(())
        }
        // The storage form of a declared row. Hot-settable because the two
        // representations must be comparable on ONE running server rather
        // than two builds, and because a replica has to store rows the way
        // its primary does — a failover otherwise changes the memory
        // profile. Turning it on converts existing rows through the tick
        // backfill; turning it off leaves packed rows packed until the next
        // write to each, which is a form, not a behaviour.
        "packed-rows" => set_packed_rows(cfg, key_str, value_str),
        "bind" | "port" | "io-threads" | "dir" | "appendonly" | "tiering-spill-dir" => {
            Err(SetError::ReadOnly(key_str.to_string()))
        }
        other => Err(SetError::Unknown(other.to_string())),
    }
}

/// `packed-rows yes|no` — the storage form of a declared row.
fn set_packed_rows(cfg: &mut Config, key: &str, value: &str) -> Result<(), SetError> {
    cfg.server.packed_rows = match value {
        "yes" | "on" | "true" | "1" => true,
        "no" | "off" | "false" | "0" => false,
        _ => {
            return Err(SetError::BadValue {
                key: key.to_string(),
                reason: "expected one of yes / no".to_string(),
            });
        }
    };
    Ok(())
}

fn set_memory(cfg: &mut Config, key: &str, value: &str) -> Result<(), SetError> {
    match key {
        "maxmemory" => {
            cfg.memory.maxmemory = parse_size(value)
                .map_err(|reason| SetError::BadValue { key: key.to_string(), reason })?;
        }
        "maxmemory-policy" => {
            cfg.memory.maxmemory_policy =
                EvictionPolicy::parse(value).ok_or_else(|| SetError::BadValue {
                    key: key.to_string(),
                    reason: "expected one of noeviction / allkeys-lru / \
                             allkeys-lfu / allkeys-random / volatile-lru / \
                             volatile-lfu / volatile-random / volatile-ttl"
                        .to_string(),
                })?;
        }
        _ => return Err(SetError::Unknown(key.to_string())),
    }
    Ok(())
}

fn set_persistence(cfg: &mut Config, key: &str, value: &str) -> Result<(), SetError> {
    match key {
        "appendfsync" => {
            cfg.persistence.appendfsync =
                AppendFsync::parse(value).ok_or_else(|| SetError::BadValue {
                    key: key.to_string(),
                    reason: "expected one of always / everysec / no".to_string(),
                })?;
        }
        "auto-aof-rewrite-percentage" => {
            cfg.persistence.auto_aof_rewrite_percentage =
                value.parse::<u32>().map_err(|_| SetError::BadValue {
                    key: key.to_string(),
                    reason: "expected a non-negative integer".to_string(),
                })?;
        }
        "auto-aof-rewrite-min-size" => {
            cfg.persistence.auto_aof_rewrite_min_size = parse_size(value)
                .map_err(|reason| SetError::BadValue { key: key.to_string(), reason })?;
        }
        "auto-aof-rewrite-bytes" => {
            cfg.persistence.auto_aof_rewrite_bytes = parse_size(value)
                .map_err(|reason| SetError::BadValue { key: key.to_string(), reason })?;
        }
        "auto-aof-rewrite-interval-secs" => {
            cfg.persistence.auto_aof_rewrite_interval_secs =
                value.parse::<u64>().map_err(|_| SetError::BadValue {
                    key: key.to_string(),
                    reason: "expected a non-negative integer".to_string(),
                })?;
        }
        _ => return Err(SetError::Unknown(key.to_string())),
    }
    Ok(())
}

fn set_expiry(cfg: &mut Config, key: &str, value: &str) -> Result<(), SetError> {
    let n = value.parse::<u32>().map_err(|_| SetError::BadValue {
        key: key.to_string(),
        reason: "expected a non-negative integer".to_string(),
    })?;
    match key {
        "hz" => cfg.expiry.hz = n,
        "maxmemory-samples" => cfg.expiry.sample = n,
        _ => return Err(SetError::Unknown(key.to_string())),
    }
    Ok(())
}

/// `notify-keyspace-events` — Redis spells it hyphenated on the wire, kevy's
/// TOML uses underscores, and nothing bridged the two until 6.3.0: the engine
/// supported keyspace notifications from the config file and had no wire path
/// to them. Spring Data's expiry listener sets this on connect.
fn set_notification(cfg: &mut Config, key: &str, value: &str) -> Result<(), SetError> {
    kevy_config::parse_notification_flags(value).map_err(|c| SetError::BadValue {
        key: key.to_string(),
        reason: format!("unknown flag char {c:?}"),
    })?;
    cfg.notification.notify_keyspace_events = value.to_string();
    Ok(())
}

fn set_log(cfg: &mut Config, key: &str, value: &str) -> Result<(), SetError> {
    match key {
        "loglevel" => {
            cfg.log.level = LogLevel::parse(value).ok_or_else(|| SetError::BadValue {
                key: key.to_string(),
                reason: "expected one of trace / debug / info / warning / error".to_string(),
            })?;
        }
        "logfile" => {
            // Redis names this `logfile`; kevy's TOML calls it `log.output`.
            // Per the hot-settable matrix, only stdout / stderr are
            // hot-settable. Any file path requires opening a handle the
            // shards can write to, which needs a restart.
            match LogOutput::parse(value) {
                LogOutput::Stdout => cfg.log.output = LogOutput::Stdout,
                LogOutput::Stderr => cfg.log.output = LogOutput::Stderr,
                LogOutput::File(_) => return Err(SetError::ReadOnly(key.to_string())),
            }
        }
        _ => return Err(SetError::Unknown(key.to_string())),
    }
    Ok(())
}

/// Flat list of redis-style `(key, value-as-string)` pairs the current
/// [`Config`] exposes via `CONFIG GET`. Keys use Redis convention
/// (lowercase, hyphenated) so the names match valkey docs verbatim.
fn config_pairs(cfg: &Config) -> Vec<(&'static str, String)> {
    let mut v: Vec<(&'static str, String)> = Vec::new();
    let [a, b, c, d] = cfg.server.bind;
    v.push(("bind", format!("{a}.{b}.{c}.{d}")));
    v.push(("port", cfg.server.port.to_string()));
    v.push(("io-threads", cfg.server.threads.to_string()));
    v.push(("packed-rows", if cfg.server.packed_rows { "yes" } else { "no" }.to_string()));
    v.push(("dir", cfg.server.data_dir.display().to_string()));
    v.push(("appendonly", yes_no(cfg.persistence.aof)));
    // kevy has no RDB save schedule (snapshots are explicit SAVE/BGSAVE);
    // the empty string is Redis's own "no save points" value. Standard
    // tooling (redis-benchmark's per-node config fetch) requires the key
    // to exist and treats its absence as "could not fetch CONFIG".
    v.push(("save", String::new()));
    v.push(("appendfsync", appendfsync_str(cfg.persistence.appendfsync).to_string()));
    v.push((
        "auto-aof-rewrite-percentage",
        cfg.persistence.auto_aof_rewrite_percentage.to_string(),
    ));
    v.push(("auto-aof-rewrite-min-size", cfg.persistence.auto_aof_rewrite_min_size.to_string()));
    v.push(("maxmemory", cfg.memory.maxmemory.to_string()));
    v.push(("maxmemory-policy", eviction_str(cfg.memory.maxmemory_policy).to_string()));
    v.push(("hz", cfg.expiry.hz.to_string()));
    // A client that CONFIG SETs this then reads it back — Spring Data's
    // listener does exactly that to confirm the flags took — needs it here
    // too, or the write appears to have vanished.
    v.push(("notify-keyspace-events", cfg.notification.notify_keyspace_events.clone()));
    v.push(("maxmemory-samples", cfg.expiry.sample.to_string()));
    v.push(("loglevel", log_level_str(cfg.log.level).to_string()));
    v.push(("cluster-enabled", yes_no(cfg.cluster.enabled)));
    v.push(("cluster-port-base", crate::cluster_port_base(cfg).to_string()));
    push_tiering_pairs(&mut v, cfg);
    v
}

/// Tiering: the budget in its configured form (`auto` / `N%` /
/// bytes); empty string = tiering off (the `save`-style "off" value).
fn push_tiering_pairs(v: &mut Vec<(&'static str, String)>, cfg: &Config) {
    v.push((
        "tiering-budget",
        cfg.tiering.budget.map(kevy_config::TierBudgetSpec::as_config_string).unwrap_or_default(),
    ));
    v.push((
        "tiering-spill-dir",
        cfg.tiering.spill_dir.as_ref().map(|p| p.display().to_string()).unwrap_or_default(),
    ));
}

/// Minimal glob matcher — `*` matches any run of bytes; everything else
/// matches literally. Sufficient for CONFIG GET patterns
/// (`maxmemory*`, `*`, exact names). Doesn't handle `?` or `[...]` —
/// real Redis does but we've never seen a CONFIG GET use them in
/// production traffic.
fn glob_match(pat: &[u8], s: &[u8]) -> bool {
    fn go(pat: &[u8], s: &[u8]) -> bool {
        match pat.split_first() {
            None => s.is_empty(),
            Some((&b'*', rest)) => {
                if rest.is_empty() {
                    return true;
                }
                (0..=s.len()).any(|i| go(rest, &s[i..]))
            }
            Some((&c, rest)) => match s.split_first() {
                Some((&first, srest)) if first == c => go(rest, srest),
                _ => false,
            },
        }
    }
    go(pat, s)
}

fn yes_no(b: bool) -> String {
    if b { "yes".into() } else { "no".into() }
}

#[cfg(test)]
#[path = "config_tests.rs"]
mod tests;