tilezz 0.2.0

//! On-disk verdict store for the classifier: the JSONL
//! `<dafsa index>\t<Classified>` format plus the machinery around it.
//!
//! Three concerns live here. The **coverage sidecar** ([`CoveredRange`] /
//! [`read_ranges`] / `append_range` / [`covered_union`]) records which index
//! ranges a run has finished, so `read_resume` can skip them on restart
//! (self-healing: it drops torn / duplicate trailing lines). **Packing**
//! ([`run_pack`]) rewrites the store in sorted, deduplicated `(index asc)`
//! order; **random access** ([`store_lookup`]) is then an O(log n) byte
//! bisection over that packed store. [`run_merge`] applies a deep-pass overlay
//! onto the store -- a cert overrides an Undecided line, never the reverse. The
//! drivers in [`super::classify_tiles`] produce and consume this store.

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

use crate::classify::cert::{Classified, Verdict};

/// One covered index range with its run-time verdict tallies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CoveredRange {
    pub start: u64,
    /// Exclusive.
    pub end: u64,
    /// `[periodic, cannot_tile, undecided]` counted at emission time.
    pub tally: [usize; 3],
}

/// The coverage sidecar path for a store: `<store>.ranges`.
pub fn ranges_path(store: &Path) -> std::path::PathBuf {
    store.with_extension("ranges")
}

/// Read the coverage sidecar (missing file = no coverage). Malformed lines
/// (a torn tail) are skipped defensively; ranges are returned sorted by
/// start. Overlapping records are legal (a re-run may re-cover a range with
/// zero new tallies); consumers use `covered` / [`covered_union`].
pub fn read_ranges(store: &Path) -> Vec<CoveredRange> {
    let Ok(txt) = std::fs::read_to_string(ranges_path(store)) else {
        return Vec::new();
    };
    let mut out: Vec<CoveredRange> = txt
        .lines()
        .filter_map(|l| {
            let mut it = l.split('\t');
            let start = it.next()?.parse().ok()?;
            let end = it.next()?.parse().ok()?;
            let p = it.next()?.parse().ok()?;
            let n = it.next()?.parse().ok()?;
            let u = it.next()?.parse().ok()?;
            (start < end).then_some(CoveredRange {
                start,
                end,
                tally: [p, n, u],
            })
        })
        .collect();
    out.sort_by_key(|r| (r.start, r.end));
    out
}

/// Append one completed-window record to the sidecar (flushed -- a crash
/// after the append can at worst re-cover the window with zero tallies).
pub(crate) fn append_range(store: &Path, r: CoveredRange) {
    use std::io::Write;
    let mut f = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(ranges_path(store))
        .unwrap();
    writeln!(
        f,
        "{}\t{}\t{}\t{}\t{}",
        r.start, r.end, r.tally[0], r.tally[1], r.tally[2]
    )
    .unwrap();
    f.flush().unwrap();
}

/// Whether `idx` lies in the covered union (disjoint intervals sorted by
/// start, as produced by [`covered_union`]). O(log n).
pub fn in_covered_union(union: &[(u64, u64)], idx: u64) -> bool {
    match union.partition_point(|&(s, _)| s <= idx) {
        0 => false,
        i => union[i - 1].1 > idx,
    }
}

/// Merge covered ranges into a disjoint sorted union of `(start, end)`.
pub fn covered_union(ranges: &[CoveredRange]) -> Vec<(u64, u64)> {
    let mut out: Vec<(u64, u64)> = Vec::new();
    for r in ranges {
        match out.last_mut() {
            Some((_, e)) if r.start <= *e => *e = (*e).max(r.end),
            _ => out.push((r.start, r.end)),
        }
    }
    out
}

/// Read back an existing store for resume: tally + collect the already-done
/// indices within `[start, end)`, and truncate a partial trailing line (a
/// killed run's torn write, a corrupt line, a duplicate) so appends stay
/// line-aligned and every tile occurs at most once. Returns the done set and
/// the `[periodic, cannot_tile, undecided]` counts already on disk. Lines
/// inside `covered` (the sidecar union) still join `done` but are EXCLUDED
/// from the tally -- the sidecar already counted them at emission time.
///
/// Malformed lines and duplicate-index lines (any perimeter) are DROPPED by
/// an atomic rewrite -- their tiles are simply not in `done`, so the run
/// re-does them; leaving the bad lines in place would instead accumulate a
/// duplicate per resume (the old behaviour handled only a torn trailing
/// line).
pub(crate) fn read_resume(
    out: &Path,
    start: u64,
    end: u64,
    covered: &[(u64, u64)],
) -> (std::collections::HashSet<u64>, [usize; 3]) {
    let mut done = std::collections::HashSet::new();
    let mut tally = [0usize; 3];
    if !out.exists() {
        return (done, tally);
    }
    // A read failure (IO error, invalid UTF-8 from disk corruption) must
    // ABORT: defaulting to an empty string would make the torn-tail branch
    // truncate the whole store -- every perimeter's already-computed verdicts.
    let txt = std::fs::read_to_string(out).unwrap_or_else(|e| {
        panic!(
            "resume: store {} unreadable ({e}); refusing to touch it",
            out.display()
        )
    });
    let valid = txt.rfind('\n').map_or(0, |i| i + 1);
    let mut keep: Vec<&str> = Vec::new();
    let mut seen_any: std::collections::HashSet<u64> = std::collections::HashSet::new();
    let mut dropped = 0usize;
    for line in txt[..valid].lines() {
        let parsed = line.split_once('\t').and_then(|(idx_s, json)| {
            let idx = idx_s.parse::<u64>().ok()?;
            let c = serde_json::from_str::<Classified>(json).ok()?;
            Some((idx, c))
        });
        let Some((idx, c)) = parsed else {
            dropped += 1;
            continue;
        };
        if !seen_any.insert(idx) {
            dropped += 1; // duplicate index: keep the first occurrence only
            continue;
        }
        keep.push(line);
        // Coverage-union lines are wholly the sidecar's: not tallied (their
        // range record counted them at emission) and not in `done` (the
        // window loop skips covered indices before ever consulting it, and
        // the total accounting counts them via the union).
        if (start..end).contains(&idx) && !in_covered_union(covered, idx) {
            match c {
                Classified::Decided(Verdict::Periodic(_)) => tally[0] += 1,
                Classified::Decided(Verdict::CannotTile(_)) => tally[1] += 1,
                Classified::Undecided { .. } => tally[2] += 1,
            }
            done.insert(idx);
        }
    }
    let torn = valid as u64
        != std::fs::metadata(out)
            .map(|m| m.len())
            .unwrap_or(valid as u64);
    if dropped > 0 {
        // Atomic rewrite without the bad/duplicate lines (also drops a torn tail).
        eprintln!("resume: dropping {dropped} malformed/duplicate store line(s)");
        let tmp = out.with_extension("jsonl.tmp");
        {
            use std::io::Write;
            let mut f = std::io::BufWriter::new(std::fs::File::create(&tmp).unwrap());
            for l in &keep {
                writeln!(f, "{l}").unwrap();
            }
            f.flush().unwrap();
        }
        std::fs::rename(&tmp, out).unwrap();
    } else if torn {
        let f = std::fs::OpenOptions::new().write(true).open(out).unwrap();
        f.set_len(valid as u64).unwrap();
    }
    (done, tally)
}
/// The deep pass's sidecar path: results land in `<store>.deep.jsonl` next to
/// the store, NOT in the store itself ([`run_merge`] applies them).
pub fn deep_overlay_path(store: &Path) -> std::path::PathBuf {
    store.with_extension("deep.jsonl")
}
/// Result of [`run_merge`]: how the overlay's entries landed.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct MergeStats {
    /// Store Undecided lines replaced by a Decided overlay verdict.
    pub overridden: usize,
    /// Overlay entries whose store line is already Decided (kept; a
    /// certificate is never overwritten).
    pub kept_decided: usize,
    /// Overlay entries that stayed Undecided (store line untouched).
    pub still_undecided: usize,
    /// Overlay entries with no store line at all (appended).
    pub appended: usize,
}

/// Apply a deep-pass overlay onto the store (atomic rewrite): a DECIDED
/// overlay verdict replaces the store's Undecided line for that index --
/// a cert overrides unknown, never the reverse (a store line that is already
/// Decided is kept, counted in `kept_decided`). Overlay entries for indices
/// absent from the store are appended (breaking a packed store's order --
/// re-run [`run_pack`] then; pure replacements preserve it). Line order is
/// otherwise preserved.
pub fn run_merge(store: &Path, overlay: &Path) -> MergeStats {
    use std::collections::HashMap;
    use std::io::Write;

    let otxt = std::fs::read_to_string(overlay).expect("overlay readable");
    // Last entry per index wins (an overlay only grows by appends).
    let mut over: HashMap<u64, (&str, bool)> = HashMap::new(); // idx -> (line, decided)
    for line in otxt.lines() {
        let Some((idx_s, json)) = line.split_once('\t') else {
            continue;
        };
        let Ok(idx) = idx_s.parse::<u64>() else {
            continue;
        };
        let Ok(c) = serde_json::from_str::<Classified>(json) else {
            continue;
        };
        over.insert(idx, (line, matches!(c, Classified::Decided(_))));
    }

    let stxt = std::fs::read_to_string(store).expect("store readable");
    let mut stats = MergeStats::default();
    let tmp = store.with_extension("jsonl.tmp");
    {
        let mut f = std::io::BufWriter::new(std::fs::File::create(&tmp).unwrap());
        for line in stxt.lines() {
            let entry = line
                .split_once('\t')
                .and_then(|(idx_s, json)| {
                    let idx = idx_s.parse::<u64>().ok()?;
                    let c = serde_json::from_str::<Classified>(json).ok()?;
                    Some((idx, c))
                })
                .and_then(|(idx, c)| over.remove(&idx).map(|o| (c, o)));
            match entry {
                Some((Classified::Undecided { .. }, (oline, true))) => {
                    stats.overridden += 1;
                    writeln!(f, "{oline}").unwrap();
                }
                Some((Classified::Undecided { .. }, (_, false))) => {
                    stats.still_undecided += 1;
                    writeln!(f, "{line}").unwrap();
                }
                Some((Classified::Decided(_), _)) => {
                    stats.kept_decided += 1;
                    writeln!(f, "{line}").unwrap();
                }
                None => writeln!(f, "{line}").unwrap(),
            }
        }
        // Overlay entries with no store line: append.
        for (_, (oline, _)) in over {
            stats.appended += 1;
            writeln!(f, "{oline}").unwrap();
        }
        f.flush().unwrap();
    }
    std::fs::rename(&tmp, store).unwrap();
    eprintln!(
        "merge: overridden {} / kept-decided {} / still-undecided {} / appended {}{}",
        stats.overridden,
        stats.kept_decided,
        stats.still_undecided,
        stats.appended,
        if stats.appended > 0 {
            " (appends break packing; re-run --pack)"
        } else {
            ""
        }
    );
    stats
}

// ---------------------------------------------------------------------------
// Store packing + point lookup.
// ---------------------------------------------------------------------------

/// Sort a verdict store by dafsa index (atomic rewrite), enabling the
/// O(log n) byte-bisection of [`store_lookup`]. Every store consumer keys on
/// the index and tolerates any line order (resume, `--deep`, `--verify`), so
/// packing is safe at any point and permanent (append-resume after packing
/// just un-sorts the tail until the next pack). Panics on a malformed line --
/// pack after a resume/verify has seen the store. In-memory sort: fine
/// through n=14 (28.5M lines, a few GB); n>=15 stores need an external run
/// merge (the fast-pass windows emit roughly index-clustered runs, so a
/// k-way merge is the natural upgrade).
pub fn run_pack(store: &Path) -> usize {
    use std::io::Write;
    let txt = std::fs::read_to_string(store).expect("store readable");
    let mut lines: Vec<(u64, &str)> = txt
        .lines()
        .map(|l| {
            let idx = l
                .split_once('\t')
                .and_then(|(i, _)| i.parse::<u64>().ok())
                .unwrap_or_else(|| panic!("pack: malformed store line {l:.80}"));
            (idx, l)
        })
        .collect();
    lines.sort_unstable_by_key(|&(idx, _)| idx);
    let tmp = store.with_extension("jsonl.tmp");
    {
        let mut f = std::io::BufWriter::new(std::fs::File::create(&tmp).unwrap());
        for (_, l) in &lines {
            writeln!(f, "{l}").unwrap();
        }
        f.flush().unwrap();
    }
    std::fs::rename(&tmp, store).unwrap();
    lines.len()
}

/// The first full line starting at or after byte `pos`, as
/// `(line_start, line)`; `None` when no full line starts there. Seeking to
/// `pos - 1` and discarding through the next newline lands exactly on a line
/// start even when `pos` itself is one.
fn line_at(f: &mut std::fs::File, pos: u64, file_len: u64) -> io::Result<Option<(u64, String)>> {
    use std::io::{BufRead, BufReader, Seek, SeekFrom};
    if pos >= file_len {
        return Ok(None);
    }
    let mut start = pos;
    let mut r = if pos == 0 {
        f.seek(SeekFrom::Start(0))?;
        BufReader::new(f)
    } else {
        f.seek(SeekFrom::Start(pos - 1))?;
        let mut r = BufReader::new(f);
        let mut skip = String::new();
        let n = r.read_line(&mut skip)? as u64;
        start = pos - 1 + n;
        if start >= file_len {
            return Ok(None);
        }
        r
    };
    let mut line = String::new();
    if r.read_line(&mut line)? == 0 {
        return Ok(None);
    }
    while line.ends_with('\n') || line.ends_with('\r') {
        line.pop();
    }
    Ok(Some((start, line)))
}

/// Parse a store line's leading dafsa index.
fn line_index(line: &str) -> io::Result<u64> {
    line.split_once('\t')
        .and_then(|(i, _)| i.parse::<u64>().ok())
        .ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::InvalidData,
                format!("malformed store line: {line:.80}"),
            )
        })
}

/// O(log n) point lookup of one dafsa index in a PACKED (index-sorted)
/// store: byte bisection narrows to a small window, then a linear scan
/// settles it -- no sidecar index, no full read. Returns `None` when the
/// index has no line. On an unpacked store the answer is meaningless; run
/// [`run_pack`] first (the CLI guards the common case).
pub fn store_lookup(store: &Path, idx: u64) -> io::Result<Option<Classified>> {
    const SCAN_WINDOW: u64 = 64 * 1024;
    let mut f = std::fs::File::open(store)?;
    let file_len = f.metadata()?.len();
    // Bisect on byte positions: keep the invariant that any line with index
    // `idx` starts at or after `lo` and strictly before `hi`.
    let (mut lo, mut hi) = (0u64, file_len);
    while hi - lo > SCAN_WINDOW {
        let mid = lo + (hi - lo) / 2;
        match line_at(&mut f, mid, file_len)? {
            None => hi = mid,
            Some((ls, line)) => {
                if line_index(&line)? <= idx {
                    lo = ls; // target starts at or after this line's start
                } else {
                    hi = mid; // probe line starts at ls >= mid, already > idx
                }
            }
        }
    }
    // Linear scan of the remaining window.
    let mut pos = lo;
    while let Some((ls, line)) = line_at(&mut f, pos, file_len)? {
        let li = line_index(&line)?;
        if li == idx {
            let json = line.split_once('\t').unwrap().1;
            let c = serde_json::from_str::<Classified>(json)
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
            return Ok(Some(c));
        }
        if li > idx {
            return Ok(None);
        }
        pos = ls + line.len() as u64 + 1;
    }
    Ok(None)
}