hayson-engine 0.11.0

use std::io;
use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::atomic::write_atomic;
use crate::registry::table_registry::TableRegistry;
use crate::schema::{HaysonRecord, UNPARTITIONED};
use crate::storage::{LogFile, LogSet};

const SNAPSHOT_VERSION: u8 = 1;
const KIND: &str = "index";
const PAIRING_WINDOW: u64 = 4096;
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;

#[derive(Debug, Clone, PartialEq)]
pub struct SnapshotHeader {
    pub table: String,
    /// The log generation this snapshot was taken of — see [`crate::table::TableStats::gen`].
    /// `0` for a snapshot written before this engine wrote the field at all (`docs/format.md`).
    pub gen: u64,
    pub schema_hash: String,
    pub seq: u64,
    pub watermark: u64,
    pub dead_bytes: u64,
    /// The whole log's fingerprint — `None` on a partitioned table, where every partition carries
    /// its own in [`SnapshotHeader::parts`] and no single file spans the table.
    pub log_hash: Option<String>,
    /// The log's leading window fingerprint, length-independent — see [`log_prefix_fingerprint`].
    /// `None` on a snapshot written before this engine wrote the field, and on a partitioned table
    /// (`docs/format.md`).
    pub prefix_hash: Option<String>,
    /// One entry per partition, and `Some` exactly for a partitioned table (`docs/format.md`).
    pub parts: Option<Vec<SnapshotPart>>,
}

/// One partition's pairing data inside a snapshot header: its name, its watermark, its dead bytes
/// and its own content fingerprint.
#[derive(Debug, Clone, PartialEq)]
pub struct SnapshotPart {
    pub part: String,
    pub watermark: u64,
    pub dead_bytes: u64,
    pub log_hash: String,
}

#[derive(Debug, Clone, PartialEq)]
pub struct SnapshotEntry {
    pub key: String,
    pub offset: u64,
    pub length: u64,
    pub seq: u64,
    pub projection: HaysonRecord,
    pub part: Option<String>,
}

#[derive(Debug, Clone, PartialEq)]
pub struct Snapshot {
    pub header: SnapshotHeader,
    pub entries: Vec<SnapshotEntry>,
}

#[derive(Serialize, Deserialize)]
struct RawHeader {
    #[serde(rename = "snapshotVersion")]
    snapshot_version: u8,
    kind: String,
    table: String,
    #[serde(default, deserialize_with = "lenient_generation")]
    gen: u64,
    #[serde(rename = "schemaHash")]
    schema_hash: String,
    seq: u64,
    watermark: u64,
    #[serde(rename = "deadBytes")]
    dead_bytes: u64,
    #[serde(rename = "logHash", skip_serializing_if = "Option::is_none", default)]
    log_hash: Option<String>,
    #[serde(rename = "prefixHash", skip_serializing_if = "Option::is_none", default)]
    prefix_hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none", default)]
    parts: Option<Vec<RawPart>>,
}

#[derive(Serialize, Deserialize)]
struct RawPart {
    p: String,
    w: u64,
    d: u64,
    h: String,
}

#[derive(Serialize, Deserialize)]
struct RawEntry {
    k: String,
    o: u64,
    l: u64,
    s: u64,
    p: HaysonRecord,
    #[serde(skip_serializing_if = "Option::is_none", default)]
    pt: Option<String>,
}

/// Reads the header's `gen` the way the TS reference's `generationOf` does: anything that is not a
/// non-negative whole number — a missing field in a snapshot written before this engine carried
/// one, a float, a string, `null` — counts as generation `0` rather than making the whole header
/// unparseable. The counter is a hint for the compaction bookkeeping, never a correctness input,
/// so a snapshot is worth keeping even when it is the one field that is junk.
fn lenient_generation<'de, D>(deserializer: D) -> Result<u64, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value = Option::<serde_json::Value>::deserialize(deserializer)?;
    Ok(value.as_ref().and_then(serde_json::Value::as_u64).unwrap_or(0))
}

fn fnv1a_update(mut hash: u64, bytes: &[u8]) -> u64 {
    for &b in bytes {
        hash ^= b as u64;
        hash = hash.wrapping_mul(FNV_PRIME);
    }
    hash
}

/// Non-cryptographic content fingerprint of the log up to `up_to`: format label, `up_to` itself,
/// and the first/last `PAIRING_WINDOW` bytes before it. A fixed-size window keeps the cost flat
/// regardless of table size — this is a cache-validity check, not a security boundary, so FNV-1a
/// over that window is enough and avoids a `sha2`/`blake3` dependency.
pub fn log_fingerprint(log: &LogFile, up_to: u64) -> io::Result<String> {
    let mut hash = fnv1a_update(FNV_OFFSET, format!("hayson-rust/log/1\n{up_to}\n").as_bytes());
    let head_len = PAIRING_WINDOW.min(up_to);
    if head_len > 0 {
        hash = fnv1a_update(hash, &log.read_bytes(0, head_len)?);
        if up_to > head_len {
            let tail_len = PAIRING_WINDOW.min(up_to - head_len);
            hash = fnv1a_update(hash, &log.read_bytes(up_to - tail_len, tail_len)?);
        }
    }
    Ok(format!("{hash:016x}"))
}

/// How many leading bytes [`log_prefix_fingerprint`] covers for a log (or watermark) of `size`.
pub fn prefix_window(size: u64) -> u64 {
    PAIRING_WINDOW.min(size)
}

/// Fingerprint of the log's first `len` bytes, with `len` deliberately kept OUT of the preimage.
/// That is the one difference from [`log_fingerprint`], and the whole point: the same value can be
/// recomputed from a log whose length has since changed, so a log SHORTER than the watermark can
/// still be told apart from a different log altogether. `len` is always
/// [`prefix_window`] of a length both sides agree on, so both sides hash the same byte range.
pub fn log_prefix_fingerprint(log: &LogFile, len: u64) -> io::Result<String> {
    let mut hash = fnv1a_update(FNV_OFFSET, b"hayson-rust/log-prefix/1\n");
    if len > 0 {
        hash = fnv1a_update(hash, &log.read_bytes(0, len)?);
    }
    Ok(format!("{hash:016x}"))
}

pub fn write_snapshot(
    path: &Path,
    registry: &TableRegistry,
    table: &str,
    schema_hash: &str,
    gen: u64,
    log: &LogFile,
) -> io::Result<()> {
    let watermark = log.size();
    let log_hash = Some(log_fingerprint(log, watermark)?);
    let prefix_hash = Some(log_prefix_fingerprint(log, prefix_window(watermark))?);
    write_with_header(
        path,
        registry,
        table,
        schema_hash,
        gen,
        watermark,
        log_hash,
        prefix_hash,
        None,
    )
}

/// Same as [`write_snapshot`], for a table whose records are spread over one log file per partition
/// value: the header carries a `parts` entry per partition — its own watermark, dead bytes and
/// fingerprint — instead of one `logHash` for a single file (`docs/format.md`). An unpartitioned
/// set writes exactly what [`write_snapshot`] would.
pub fn write_snapshot_set(
    path: &Path,
    registry: &TableRegistry,
    table: &str,
    schema_hash: &str,
    gen: u64,
    logs: &LogSet,
) -> io::Result<()> {
    if !logs.is_partitioned() {
        let Some(log) = logs.log_of(UNPARTITIONED) else {
            return Err(io::Error::other(format!("table \"{table}\" has no log to snapshot")));
        };
        return write_snapshot(path, registry, table, schema_hash, gen, log);
    }

    let mut parts = Vec::new();
    for part in logs.parts() {
        let Some(log) = logs.log_of(&part) else { continue };
        parts.push(RawPart {
            p: part.clone(),
            w: log.size(),
            d: registry.stats_of(&part).dead_bytes,
            h: log_fingerprint(log, log.size())?,
        });
    }
    write_with_header(
        path,
        registry,
        table,
        schema_hash,
        gen,
        logs.total_size(),
        None,
        None,
        Some(parts),
    )
}

#[allow(clippy::too_many_arguments)]
fn write_with_header(
    path: &Path,
    registry: &TableRegistry,
    table: &str,
    schema_hash: &str,
    gen: u64,
    watermark: u64,
    log_hash: Option<String>,
    prefix_hash: Option<String>,
    parts: Option<Vec<RawPart>>,
) -> io::Result<()> {
    let header = RawHeader {
        snapshot_version: SNAPSHOT_VERSION,
        kind: KIND.to_string(),
        table: table.to_string(),
        gen,
        schema_hash: schema_hash.to_string(),
        seq: registry.max_seq(),
        watermark,
        dead_bytes: registry.stats().dead_bytes,
        log_hash,
        prefix_hash,
        parts,
    };
    let mut lines = Vec::with_capacity(registry.records() + 1);
    lines.push(serde_json::to_string(&header)?);

    // NOTE: arrival order, not key order -- a restore re-ranks keys as it reads them, so the file carries it.
    for key in registry.keys_in_arrival_order() {
        let entry = registry
            .slot_of(key)
            .expect("key came from registry.keys_in_arrival_order()");
        let raw = RawEntry {
            k: key.clone(),
            o: entry.offset,
            l: entry.length,
            s: entry.seq,
            p: registry.projection_of(key).cloned().unwrap_or_default(),
            pt: registry.part_of(key).filter(|p| !p.is_empty()).map(str::to_string),
        };
        lines.push(serde_json::to_string(&raw)?);
    }
    let payload = format!("{}\n", lines.join("\n"));
    write_atomic(path, payload.as_bytes(), None)
}

/// Reads and parses `path`. Returns `Ok(None)` for anything that makes the file unusable as a
/// snapshot — missing, truncated, wrong format marker, unparseable line — since a checkpoint file
/// is a cache and none of those are ever surfaced as an error to the caller. `Err` is reserved for
/// a genuine IO failure (e.g. permission denied) reading a file that does exist.
pub fn read_snapshot(path: &Path) -> io::Result<Option<Snapshot>> {
    let raw = match std::fs::read_to_string(path) {
        Ok(content) => content,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error),
    };
    let mut lines = raw.split('\n').filter(|line| !line.is_empty());
    let Some(first) = lines.next() else {
        return Ok(None);
    };
    let Ok(header) = serde_json::from_str::<RawHeader>(first) else {
        return Ok(None);
    };
    if header.snapshot_version != SNAPSHOT_VERSION || header.kind != KIND {
        return Ok(None);
    }

    let mut entries = Vec::new();
    for line in lines {
        let Ok(raw_entry) = serde_json::from_str::<RawEntry>(line) else {
            return Ok(None);
        };
        entries.push(SnapshotEntry {
            key: raw_entry.k,
            offset: raw_entry.o,
            length: raw_entry.l,
            seq: raw_entry.s,
            projection: raw_entry.p,
            part: raw_entry.pt.filter(|p| !p.is_empty()),
        });
    }

    Ok(Some(Snapshot {
        header: SnapshotHeader {
            table: header.table,
            gen: header.gen,
            schema_hash: header.schema_hash,
            seq: header.seq,
            watermark: header.watermark,
            dead_bytes: header.dead_bytes,
            log_hash: header.log_hash,
            prefix_hash: header.prefix_hash,
            parts: header.parts.map(|parts| {
                parts
                    .into_iter()
                    .map(|raw| SnapshotPart {
                        part: raw.p,
                        watermark: raw.w,
                        dead_bytes: raw.d,
                        log_hash: raw.h,
                    })
                    .collect()
            }),
        },
        entries,
    }))
}

pub fn remove_snapshot(path: &Path) -> io::Result<()> {
    match std::fs::remove_file(path) {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error),
    }
}