kevy-persist 7.0.0

use super::*;
use crate::snapshot_fmt::VERSION;
use std::borrow::Cow;
use std::time::Duration;

/// An `apply` for a staging ring whose records a test does not need.
pub(crate) fn ignore_owed(_: &mut Argv) {}

pub(crate) fn temp_file(name: &str) -> std::path::PathBuf {
    unique_file(name, "rdb")
}

// named after a directory unique by construction across threads and
// processes; the directory goes at once, so the caller's remove_file
// still leaves nothing behind
pub(crate) fn unique_file(name: &str, ext: &str) -> std::path::PathBuf {
    let dir = kevy_tmpdir::unique_dir(name);
    std::fs::remove_dir(&dir).unwrap();
    std::path::PathBuf::from(format!("{}.{ext}", dir.display()))
}

#[test]
fn snapshot_round_trip() {
    let path = temp_file("rt");

    let mut src = Store::new();
    src.set(b"plain", b"value".to_vec(), None, kevy_store::SetCondition::Always);
    src.set(b"empty", Vec::new(), None, kevy_store::SetCondition::Always);
    src.set(b"binary", vec![0u8, 1, 2, 255, 254], None, kevy_store::SetCondition::Always);
    src.set(
        b"withttl",
        b"soon".to_vec(),
        Some(Duration::from_secs(100)),
        kevy_store::SetCondition::Always,
    );

    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();

    assert_eq!(dst.dbsize(), 4);
    assert_eq!(dst.get(b"plain").unwrap(), Some(Cow::Borrowed(&b"value"[..])));
    assert_eq!(dst.get(b"empty").unwrap(), Some(Cow::Borrowed(&b""[..])));
    assert_eq!(dst.get(b"binary").unwrap(), Some(Cow::Borrowed(&[0u8, 1, 2, 255, 254][..])));
    assert_eq!(dst.get(b"withttl").unwrap(), Some(Cow::Borrowed(&b"soon"[..])));
    // TTL survived (stored as an absolute Unix-ms deadline, v3 format).
    assert!(dst.pttl(b"withttl") > 90_000);

    let _ = std::fs::remove_file(&path);
}

/// TTL re-anchoring regression: a snapshot stores the **absolute** deadline, so
/// time elapsed between save and load is subtracted from the restored TTL.
/// The pre-fix v2 format stored remaining-ms and re-anchored on load, so the
/// TTL would read back ~unchanged regardless of the gap.
#[test]
fn snapshot_ttl_is_absolute_across_delay() {
    let path = temp_file("ttl-abs");
    let mut src = Store::new();
    src.set(b"k", b"v".to_vec(), Some(Duration::from_secs(100)), kevy_store::SetCondition::Always);
    save_snapshot(&src, &path).unwrap();

    std::thread::sleep(Duration::from_millis(1500));

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    let pttl = dst.pttl(b"k");
    // ~1.5 s of the 100 s elapsed while "down": deadline preserved => < 99 s.
    assert!(
        (0..99_000).contains(&pttl),
        "PTTL after delayed load = {pttl} ms; absolute deadline not preserved"
    );
    assert!(pttl > 90_000, "PTTL {pttl} implausibly low");
    let _ = std::fs::remove_file(&path);
}

#[test]
fn bad_magic_is_rejected() {
    let path = temp_file("bad");
    std::fs::write(&path, b"NOTKEVY!....").unwrap();
    let mut dst = Store::new();
    assert!(load_snapshot(&mut dst, &path).is_err());
    let _ = std::fs::remove_file(&path);
}

#[test]
fn expired_keys_are_not_saved() {
    let path = temp_file("exp");
    let mut src = Store::new();
    src.set(b"live", b"1".to_vec(), None, kevy_store::SetCondition::Always);
    src.set(
        b"dead",
        b"2".to_vec(),
        Some(Duration::from_millis(1)),
        kevy_store::SetCondition::Always,
    );
    std::thread::sleep(Duration::from_millis(8));

    save_snapshot(&src, &path).unwrap();
    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();

    assert_eq!(dst.dbsize(), 1);
    assert_eq!(dst.get(b"live").unwrap(), Some(Cow::Borrowed(&b"1"[..])));
    assert_eq!(dst.get(b"dead").unwrap(), None);
    let _ = std::fs::remove_file(&path);
}

#[test]
fn hash_snapshot_round_trip() {
    let path = temp_file("hashrt");
    let mut src = Store::new();
    src.hset(b"h", &[(b"a".as_slice(), b"1".as_slice()), (b"b".as_slice(), b"two".as_slice())])
        .unwrap();
    src.set(b"s", b"str".to_vec(), None, kevy_store::SetCondition::Always);
    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    assert_eq!(dst.type_of(b"h"), "hash");
    assert_eq!(dst.hget(b"h", b"a").unwrap(), Some(&b"1"[..]));
    assert_eq!(dst.hget(b"h", b"b").unwrap(), Some(&b"two"[..]));
    assert_eq!(dst.hlen(b"h"), Ok(2));
    assert_eq!(dst.get(b"s").unwrap(), Some(Cow::Borrowed(&b"str"[..])));
    let _ = std::fs::remove_file(&path);
}

#[test]
fn list_snapshot_round_trip() {
    let path = temp_file("listrt");
    let mut src = Store::new();
    src.rpush(b"l", &[b"a".as_slice(), b"b".as_slice(), b"c".as_slice()]).unwrap();
    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    assert_eq!(dst.type_of(b"l"), "list");
    assert_eq!(dst.llen(b"l"), Ok(3));
    assert_eq!(dst.lrange(b"l", 0, -1).unwrap(), vec![b"a".to_vec(), b"b".to_vec(), b"c".to_vec()]);
    let _ = std::fs::remove_file(&path);
}

#[test]
fn set_snapshot_round_trip() {
    let path = temp_file("setrt");
    let mut src = Store::new();
    src.sadd(b"s", &[b"x".as_slice(), b"y".as_slice(), b"z".as_slice()]).unwrap();
    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    assert_eq!(dst.type_of(b"s"), "set");
    assert_eq!(dst.scard(b"s"), Ok(3));
    let mut members = dst.smembers(b"s").unwrap();
    members.sort();
    assert_eq!(members, vec![b"x".to_vec(), b"y".to_vec(), b"z".to_vec()]);
    let _ = std::fs::remove_file(&path);
}

#[test]
fn zset_snapshot_round_trip() {
    let path = temp_file("zsetrt");
    let mut src = Store::new();
    src.zadd(b"z", &[(1.0, b"a".as_slice()), (2.0, b"b".as_slice()), (0.5, b"c".as_slice())])
        .unwrap();
    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    assert_eq!(dst.type_of(b"z"), "zset");
    assert_eq!(dst.zcard(b"z"), Ok(3));
    // Ascending score order: c(0.5), a(1.0), b(2.0)
    let range = dst.zrange(b"z", 0, -1).unwrap();
    assert_eq!(range, vec![(b"c".to_vec(), 0.5), (b"a".to_vec(), 1.0), (b"b".to_vec(), 2.0),]);
    let _ = std::fs::remove_file(&path);
}

#[test]
fn all_types_snapshot_round_trip() {
    let path = temp_file("allrt");
    let mut src = Store::new();
    src.set(b"str", b"hello".to_vec(), None, kevy_store::SetCondition::Always);
    src.hset(b"hash", &[(b"f".as_slice(), b"v".as_slice())]).unwrap();
    src.rpush(b"list", &[b"i".as_slice()]).unwrap();
    src.sadd(b"set", &[b"m".as_slice()]).unwrap();
    src.zadd(b"zset", &[(1.0, b"k".as_slice())]).unwrap();
    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    assert_eq!(dst.dbsize(), 5);
    assert_eq!(dst.type_of(b"str"), "string");
    assert_eq!(dst.type_of(b"hash"), "hash");
    assert_eq!(dst.type_of(b"list"), "list");
    assert_eq!(dst.type_of(b"set"), "set");
    assert_eq!(dst.type_of(b"zset"), "zset");
    let _ = std::fs::remove_file(&path);
}

// ───────────── stream consumer groups (snapshot v4) ─────────────

use kevy_store::{GroupCreateMode, ReadGroupId, StreamId, XAddIdSpec};

/// Entries 1-1/2-1/3-1; group `g`; `c1` holds 1-1+2-1 (t=1000), `c2`
/// holds 3-1 (t=2000); 2-1 XDEL'd → its PEL row is a tombstone.
fn grouped_stream_store() -> Store {
    let mut src = Store::new();
    for ms in [1u64, 2, 3] {
        src.xadd(
            b"st",
            XAddIdSpec::Explicit(StreamId::new(ms, 1)),
            vec![(b"f".to_vec(), b"v".to_vec())],
            kevy_store::MissingStream::Create,
            0,
        )
        .unwrap();
    }
    src.xgroup_create(
        b"st",
        b"g",
        GroupCreateMode::AtId(StreamId::MIN),
        kevy_store::MissingStream::Refuse,
    )
    .unwrap();
    src.xreadgroup(
        b"st",
        b"g",
        b"c1",
        ReadGroupId::New,
        Some(2),
        kevy_store::AckMode::Pending,
        1000,
    )
    .unwrap();
    src.xreadgroup(b"st", b"g", b"c2", ReadGroupId::New, None, kevy_store::AckMode::Pending, 2000)
        .unwrap();
    src.xdel(b"st", &[StreamId::new(2, 1)]).unwrap();
    src
}

#[test]
fn stream_groups_snapshot_round_trip() {
    let path = temp_file("groups");
    let src = grouped_stream_store();
    save_snapshot(&src, &path).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();

    let view = dst.stream_view(b"st").unwrap().unwrap();
    assert_eq!(view.length(), 2);
    assert_eq!(view.last_id(), StreamId::new(3, 1));
    assert_eq!(view.entries_added(), 3);
    assert_eq!(view.max_deleted_id(), StreamId::new(2, 1));
    let g = view.group(b"g").expect("group must survive the snapshot");
    assert_eq!(g.last_delivered_id(), StreamId::new(3, 1));
    // Snapshot is the full-fidelity path: the 2-1 tombstone survives.
    assert_eq!(g.pending_count(), 3);
    let p2 = g.pending_entry(StreamId::new(2, 1)).unwrap();
    assert_eq!(
        (p2.consumer.as_slice(), p2.delivery_time_ms, p2.delivery_count),
        (&b"c1"[..], 1000, 1)
    );
    let mut consumers: Vec<(Vec<u8>, u64, usize)> =
        g.consumers().map(|(n, c)| (n.to_vec(), c.last_seen_ms(), c.pending_count())).collect();
    consumers.sort();
    assert_eq!(consumers, vec![(b"c1".to_vec(), 1000, 2), (b"c2".to_vec(), 2000, 1)]);
    let _ = std::fs::remove_file(&path);
}

/// A v3 file (pre-groups format) must still load — entries + scalars
/// present, group map empty. Hand-encoded bytes pin the v3 layout.
#[test]
fn v3_snapshot_without_groups_still_loads() {
    let path = temp_file("v3compat");
    let mut b: Vec<u8> = Vec::new();
    b.extend_from_slice(b"KEVYSNAP");
    b.push(3); // version 3: no group section after the entries block
    b.push(6); // OP_STREAM
    b.push(0); // no TTL
    b.extend_from_slice(&2u32.to_le_bytes());
    b.extend_from_slice(b"st");
    for v in [1u64, 1, 0, 0, 1] {
        // last(1,1), max_deleted(0,0), entries_added=1
        b.extend_from_slice(&v.to_le_bytes());
    }
    b.extend_from_slice(&1u32.to_le_bytes()); // 1 entry
    b.extend_from_slice(&1u64.to_le_bytes()); // ms
    b.extend_from_slice(&1u64.to_le_bytes()); // seq
    b.extend_from_slice(&1u32.to_le_bytes()); // 1 field
    b.extend_from_slice(&1u32.to_le_bytes());
    b.push(b'f');
    b.extend_from_slice(&1u32.to_le_bytes());
    b.push(b'v');
    b.push(0); // OP_EOF
    std::fs::write(&path, &b).unwrap();

    let mut dst = Store::new();
    load_snapshot(&mut dst, &path).unwrap();
    let view = dst.stream_view(b"st").unwrap().unwrap();
    assert_eq!(view.length(), 1);
    assert_eq!(view.last_id(), StreamId::new(1, 1));
    assert_eq!(view.group_count(), 0);
    let _ = std::fs::remove_file(&path);
}

// A corrupt/hostile snapshot declaring a huge element count with no
// bytes behind it must fail with a plain `io::Error` (truncated
// stream) — NOT drive a multi-gigabyte `with_capacity` / `vec![0;n]`
// that aborts the process. This is the malicious-primary path
// (`replication_apply` feeds primary bytes straight into the loader).
#[test]
fn forged_count_fails_cleanly_not_alloc_abort() {
    // OP_SET (a set) declaring ~4.29 billion members, then EOF.
    let mut b: Vec<u8> = Vec::new();
    b.extend_from_slice(b"KEVYSNAP");
    b.push(VERSION);
    b.push(4); // OP_SET
    b.push(0); // no TTL
    b.extend_from_slice(&2u32.to_le_bytes()); // key len
    b.extend_from_slice(b"sk");
    b.extend_from_slice(&u32::MAX.to_le_bytes()); // member count: forged
    // ...no members follow. The loader must hit end-of-stream, not OOM.
    let mut dst = Store::new();
    let err = load_snapshot_from(&mut dst, std::io::Cursor::new(b)).unwrap_err();
    assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);

    // Same for a forged single bulk length.
    let mut b2: Vec<u8> = Vec::new();
    b2.extend_from_slice(b"KEVYSNAP");
    b2.push(VERSION);
    b2.push(1); // OP_STRING
    b2.push(0); // no TTL
    b2.extend_from_slice(&2u32.to_le_bytes());
    b2.extend_from_slice(b"sk");
    b2.extend_from_slice(&u32::MAX.to_le_bytes()); // value len: forged
    let mut dst2 = Store::new();
    let err2 = load_snapshot_from(&mut dst2, std::io::Cursor::new(b2)).unwrap_err();
    assert_eq!(err2.kind(), std::io::ErrorKind::UnexpectedEof);
}

// ─────────────── SnapshotView serialization (COW E-3) ───────────────

fn populated_store() -> Store {
    let mut s = Store::new();
    s.set(b"s1", b"plain".to_vec(), None, kevy_store::SetCondition::Always);
    s.set(b"s2", vec![b'x'; 100], None, kevy_store::SetCondition::Always); // heap str
    s.hset(b"h", &[(b"f".as_slice(), b"v".as_slice())]).unwrap();
    s.rpush(b"l", &[b"a".as_slice(), b"b".as_slice()]).unwrap();
    s.sadd(b"set", &[b"m1".as_slice(), b"m2".as_slice()]).unwrap();
    s.zadd(b"z", &[(1.5, b"one".as_slice())]).unwrap();
    s
}

/// A frozen view serializes byte-identically to the live store it froze
/// (no-TTL data: TTL deadlines are stamped at write time and would
/// legitimately differ between two writes).
#[test]
fn view_snapshot_bytes_match_store_snapshot() {
    let s = populated_store();
    let view = s.collect_snapshot();
    let dir = std::env::temp_dir();
    let p_store = dir.join(format!("kevy-e3-store-{}.rdb", std::process::id()));
    let p_view = dir.join(format!("kevy-e3-view-{}.rdb", std::process::id()));
    save_snapshot(&s, &p_store).unwrap();
    save_snapshot(&view, &p_view).unwrap();
    // each file ends with its own id; the entries before it are the same
    let body = |p: &std::path::Path| {
        let mut b = std::fs::read(p).unwrap();
        b.truncate(b.len() - 16);
        b
    };
    assert_eq!(body(&p_store), body(&p_view));
    let _ = std::fs::remove_file(&p_store);
    let _ = std::fs::remove_file(&p_view);
}

/// View-serialized AOF replays into an equivalent store, and reflects the
/// collect instant — not mutations that landed during/after serialization.
#[test]
fn view_aof_round_trips_at_the_collect_instant() {
    let mut s = populated_store();
    let view = s.collect_snapshot();
    // Post-collect mutations must not appear in the dump.
    s.set(b"s1", b"mutated".to_vec(), None, kevy_store::SetCondition::Always);
    s.hset(b"h", &[(b"f2".as_slice(), b"late".as_slice())]).unwrap();

    // dump_aof writes a FILE at this path — the unique dir is its parent.
    let p = kevy_tmpdir::unique_dir("persist").join("view.aof");
    let crate::RewriteStats { keys, bytes } = dump_aof(&p, &view).unwrap();
    assert_eq!(keys, 6);
    assert!(bytes > 0);

    let mut restored = Store::new();
    replay_aof(&p, |args| {
        crate::tests_rewrite::apply_for_test(&mut restored, &args);
    })
    .unwrap();
    assert_eq!(restored.get(b"s1").unwrap(), Some(Cow::Borrowed(b"plain".as_slice())));
    assert_eq!(restored.hget(b"h", b"f2").unwrap(), None);
    assert_eq!(restored.hget(b"h", b"f").unwrap(), Some(b"v".as_slice()));
    let _ = std::fs::remove_file(&p);
}

/// A row sealed into a segment is not a value the rewrite re-emits as
/// commands: the estimate leaves it out, where the rewrite writes a frame
/// naming its segment.
#[test]
fn a_row_sealed_to_a_segment_is_left_out_of_the_estimate() {
    let d = kevy_tmpdir::TmpDir::new("persist-estimate-seg");
    let mut s = Store::new();
    s.enable_seg_rows(d.path()).unwrap();
    s.hset(b"user:1", &[(b"name".as_slice(), b"ada".as_slice())]).unwrap();
    let sealed = s.seal_rows_to_seg(b"user", &[b"user:1".to_vec()]).unwrap().expect("sealed");
    assert_eq!(s.commit_row_eviction(&sealed), 1);
    let empty = crate::estimate_rewrite_size(&Store::new());
    assert_eq!(crate::estimate_rewrite_size(&s), empty);
    let written = dump_aof(&d.path().join("seg.aof"), &s).unwrap();
    assert!(written.bytes > empty, "the segment frame is written");
}

/// A shard's snapshot that carries a frame hands it to the merge like a
/// logged command, after the keys; a snapshot that cannot be read fails it.
#[test]
fn a_merge_replays_a_snapshot_frame_and_fails_on_an_unreadable_one() {
    use crate::reshard::{StdLayout, merge_sources};
    let dir = kevy_tmpdir::TmpDir::new("persist-merge-aux");
    let mut shard = Store::new();
    shard.set(b"k", b"v".to_vec(), None, kevy_store::SetCondition::Always);
    let frame = Argv::from(vec![b"XINTERNAL.EXAMPLE".to_vec(), b"1".to_vec()]);
    let src = crate::WithAux::new(&shard, Some(&frame));
    save_snapshot(&src, &crate::layout::snapshot_path(dir.path(), 0)).unwrap();
    let mut temp = Store::new();
    let mut replayed = Vec::new();
    let mut on_frame = |store: &mut Store, args: Argv| replayed.push((store.dbsize(), args));
    let sources = merge_sources(dir.path(), 1, &StdLayout, &mut temp, &mut on_frame).unwrap();
    assert_eq!(sources.len(), 1);
    assert_eq!(temp.dbsize(), 1);
    // a second shard whose snapshot path is a directory
    std::fs::create_dir(crate::layout::snapshot_path(dir.path(), 1)).unwrap();
    let mut again = Store::new();
    assert!(merge_sources(dir.path(), 2, &StdLayout, &mut again, &mut on_frame).is_err());
    assert_eq!(replayed, [(1, frame.clone()), (1, frame)], "the frame arrives after the keys");
}