kevy-persist 7.0.0

//! Stream AOF-rewrite encoding — split from `rewrite_fmt.rs` when the
//! seg-collection arms pushed it against the 500-LOC cap. The XADD /
//! id-fixup / group reconstruction command stream for one stream value.

use crate::rewrite_fmt::emit;
use kevy_resp::Argv;
use kevy_store::{StreamData, StreamId};
use std::io::{self, Write};

/// Render one stream as commands: one XADD per entry (slow on huge
/// streams but correct — a multi-entry XADD batch is a future parser
/// feature), then `XSETID` whenever a bare replay of those XADDs would
/// not reproduce the scalar state (deleted tail, deleted-only stream,
/// non-zero `entries_added` drift), then the consumer-group section.
/// Returns the number of command frames written so callers that ship
/// rebuild frames elsewhere (scope migration) can report a frame count.
///
/// ```
/// use kevy_store::{MissingStream, Store, XAddIdSpec};
///
/// let mut store = Store::new();
/// let fields = vec![(b"f".to_vec(), b"v".to_vec())];
/// store.xadd(b"s", XAddIdSpec::AutoSeq(1), fields, MissingStream::Create, 0)?;
/// let stream = store.stream_view(b"s")?.ok_or("no stream")?;
/// let mut out = Vec::new();
/// let frames = kevy_persist::write_stream_as_commands(&mut out, b"s", stream)?;
/// assert_eq!(frames, 1, "one XADD, no fixups");
/// let (xadd, _) = kevy_resp::parse_command(&out)?.ok_or("incomplete")?;
/// assert_eq!(&xadd[0], b"XADD");
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
pub fn write_stream_as_commands<W: Write>(
    mut w: W,
    key: &[u8],
    s: &StreamData,
) -> io::Result<usize> {
    stream_as_commands(&mut w, key, s, crate::AofFormat::V1, &mut Vec::new())
}

/// Format-aware body of [`write_stream_as_commands`].
pub(crate) fn stream_as_commands<W: Write>(
    w: &mut W,
    key: &[u8],
    s: &StreamData,
    fmt: crate::AofFormat,
    scratch: &mut Vec<u8>,
) -> io::Result<usize> {
    let mut frames = 0usize;
    for (id, fv) in s.entries() {
        let mut argv: Vec<Vec<u8>> = Vec::with_capacity(3 + fv.len() * 2);
        argv.push(b"XADD".to_vec());
        argv.push(key.to_vec());
        argv.push(id.encode());
        for (f, v) in fv {
            argv.push(f.to_vec());
            argv.push(v.to_vec());
        }
        emit(w, &Argv::from(argv), fmt, scratch)?;
        frames += 1;
    }
    frames += write_stream_id_fixup(w, key, s, fmt, scratch)?;
    frames += write_stream_group_commands(w, key, s, fmt, scratch)?;
    Ok(frames)
}

/// The stream-ID bookkeeping tail of a stream rewrite: recreate an
/// emptied stream's advanced ID clock, then XSETID when the natural
/// replay outcome differs from the stored scalars.
pub(crate) fn write_stream_id_fixup<W: Write>(
    w: &mut W,
    key: &[u8],
    s: &StreamData,
    fmt: crate::AofFormat,
    scratch: &mut Vec<u8>,
) -> io::Result<usize> {
    let mut frames = 0usize;
    let (len, last, mxd, added) = (s.length(), s.last_id(), s.max_deleted_id(), s.entries_added());
    if len == 0 && last != StreamId::MIN {
        // Empty stream whose ID clock advanced (all entries deleted):
        // re-create the key with the right `last_id` via the
        // `XADD MAXLEN 0` trick — the inline trim wipes the dummy row.
        let argv = vec![
            b"XADD".to_vec(),
            key.to_vec(),
            b"MAXLEN".to_vec(),
            b"0".to_vec(),
            last.encode(),
            b"x".to_vec(),
            b"x".to_vec(),
        ];
        emit(w, &Argv::from(argv), fmt, scratch)?;
        frames += 1;
    }
    // What replaying the commands emitted so far yields: a trim leaves the
    // highest deleted ID alone, so the MAXLEN 0 trick leaves it at 0-0.
    // The only no-key case left is the virgin empty stream (groups-only)
    // — its scalars are all zero by construction, so skipping XSETID
    // there is exact.
    let natural = if len > 0 {
        (s.last_entry().map_or(StreamId::MIN, |(id, _)| id), len, StreamId::MIN)
    } else {
        (last, u64::from(last != StreamId::MIN), StreamId::MIN)
    };
    if natural != (last, added, mxd) {
        let argv = vec![
            b"XSETID".to_vec(),
            key.to_vec(),
            last.encode(),
            b"ENTRIESADDED".to_vec(),
            added.to_string().into_bytes(),
            b"MAXDELETEDID".to_vec(),
            mxd.encode(),
        ];
        emit(w, &Argv::from(argv), fmt, scratch)?;
        frames += 1;
    }
    Ok(frames)
}

/// The XCLAIM that restores one pending entry to its owner on replay.
///
/// `FORCE` because the entry is already in the stream and `JUSTID` because
/// the payload is not being re-delivered — only the ownership and the
/// delivery bookkeeping are. Split from `write_stream_group_commands` for
/// the 50-line rule; it was the longest of that function's three argv
/// literals and the only one with arguments to explain.
fn xclaim_argv(
    key: &[u8],
    group: &[u8],
    consumer: &[u8],
    id: StreamId,
    delivery_time_ms: impl core::fmt::Display,
    delivery_count: impl core::fmt::Display,
) -> Vec<Vec<u8>> {
    vec![
        b"XCLAIM".to_vec(),
        key.to_vec(),
        group.to_vec(),
        consumer.to_vec(),
        b"0".to_vec(),
        id.encode(),
        b"TIME".to_vec(),
        delivery_time_ms.to_string().into_bytes(),
        b"RETRYCOUNT".to_vec(),
        delivery_count.to_string().into_bytes(),
        b"FORCE".to_vec(),
        b"JUSTID".to_vec(),
    ]
}

/// Consumer-group section of a stream rewrite: `XGROUP CREATE … MKSTREAM`
/// (MKSTREAM covers groups on a virgin empty stream), `XGROUP SETID …
/// ENTRIESREAD n` when the read counter is known (its own frame, which a
/// 6.4 reader skips and still has the group), one `XGROUP CREATECONSUMER`
/// per known consumer (so a 6.4 reader has the consumers the internal
/// frames below would give it), one `XCLAIM … TIME t RETRYCOUNT n FORCE
/// JUSTID` per live PEL row — full delivery_time/count fidelity, the same
/// technique Redis's own AOF rewrite uses — then one internal
/// `XINTERNAL.CONSUMERSEEN key group consumer t [a]` per known consumer,
/// which makes it with its last contact and its last active time, after
/// the claims that stamp it with the replay's clock. A pending row whose
/// entry is gone (deleted or trimmed while pending) is put back by an
/// internal `XINTERNAL.PENDING key group consumer t n id` instead: no
/// client command makes such a row again, and a history read and
/// `XAUTOCLAIM` still answer it.
pub(crate) fn write_stream_group_commands<W: Write>(
    w: &mut W,
    key: &[u8],
    s: &StreamData,
    fmt: crate::AofFormat,
    scratch: &mut Vec<u8>,
) -> io::Result<usize> {
    let mut frames = 0usize;
    for g in s.export_groups() {
        frames += write_group_head(w, key, &g, fmt, scratch)?;
        for (ms, seq, consumer, delivery_time_ms, delivery_count) in &g.pel {
            let id = StreamId::new(*ms, *seq);
            let argv = if s.contains_entry(id) {
                xclaim_argv(key, &g.name, consumer, id, *delivery_time_ms, *delivery_count)
            } else {
                pending_argv(key, &g.name, consumer, id, *delivery_time_ms, *delivery_count)
            };
            emit(w, &Argv::from(argv), fmt, scratch)?;
            frames += 1;
        }
        frames += write_consumer_times(w, key, &g, fmt, scratch)?;
    }
    Ok(frames)
}

/// The frames that make a group and its consumers, before its pending
/// rows: `XGROUP CREATE`, the read counter, one `XGROUP CREATECONSUMER`
/// per consumer.
fn write_group_head<W: Write>(
    w: &mut W,
    key: &[u8],
    g: &kevy_store::LoadedGroup,
    fmt: crate::AofFormat,
    scratch: &mut Vec<u8>,
) -> io::Result<usize> {
    let last_delivered = StreamId::new(g.last_delivered.0, g.last_delivered.1).encode();
    let named = |sub: &[u8], tail: &[&[u8]]| {
        let mut argv = vec![b"XGROUP".to_vec(), sub.to_vec(), key.to_vec(), g.name.clone()];
        argv.extend(tail.iter().map(|t| t.to_vec()));
        Argv::from(argv)
    };
    emit(w, &named(b"CREATE", &[&last_delivered, b"MKSTREAM"]), fmt, scratch)?;
    let mut frames = 1;
    if let Some(n) = g.entries_read {
        let n = n.to_string();
        emit(w, &named(b"SETID", &[&last_delivered, b"ENTRIESREAD", n.as_bytes()]), fmt, scratch)?;
        frames += 1;
    }
    for (consumer, _) in &g.consumers {
        emit(w, &named(b"CREATECONSUMER", &[consumer]), fmt, scratch)?;
        frames += 1;
    }
    Ok(frames)
}

/// The internal frame that puts back a pending row whose entry is gone.
fn pending_argv(
    key: &[u8],
    group: &[u8],
    consumer: &[u8],
    id: StreamId,
    delivery_time_ms: u64,
    delivery_count: u64,
) -> Vec<Vec<u8>> {
    vec![
        kevy_resp::ops_table::PENDING.as_bytes().to_vec(),
        key.to_vec(),
        group.to_vec(),
        consumer.to_vec(),
        delivery_time_ms.to_string().into_bytes(),
        delivery_count.to_string().into_bytes(),
        id.encode(),
    ]
}

/// One `XINTERNAL.CONSUMERSEEN key group consumer t [a]` per consumer of
/// `g`: its last contact, and its last active time when it has one.
fn write_consumer_times<W: Write>(
    w: &mut W,
    key: &[u8],
    g: &kevy_store::LoadedGroup,
    fmt: crate::AofFormat,
    scratch: &mut Vec<u8>,
) -> io::Result<usize> {
    let mut active: Vec<&(Vec<u8>, u64)> = g.active.iter().collect();
    active.sort_unstable();
    for (consumer, last_seen_ms) in &g.consumers {
        let mut argv = vec![
            kevy_resp::ops_table::CONSUMER_SEEN.as_bytes().to_vec(),
            key.to_vec(),
            g.name.clone(),
            consumer.clone(),
            last_seen_ms.to_string().into_bytes(),
        ];
        if let Ok(i) = active.binary_search_by(|(name, _)| name.as_slice().cmp(consumer)) {
            argv.push(active[i].1.to_string().into_bytes());
        }
        emit(w, &Argv::from(argv), fmt, scratch)?;
    }
    Ok(g.consumers.len())
}