macrame/temporal/

replay.rs

use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};

use crate::error::{DbError, Result};
use crate::temporal::as_of::NodeAttributes;

/// Full materialized state reconstructed from transaction_log replay (§5.5).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MaterializedState {
    pub seq_anchor: i64,
    pub timestamp: String,
    pub concepts: HashMap<String, NodeAttributes>,
    pub edges: Vec<(String, String, String, String, String)>,
    /// **Nothing had been recorded yet at `timestamp`** (0.8.0, B5, D-121).
    ///
    /// An empty state has two meanings and a caller can act differently on
    /// them. *Everything was retired by then* is a fact about the data;
    /// *the ledger had not started* is a fact about the question. Both come
    /// back as zero concepts and zero edges, so the difference has to be
    /// carried rather than inferred.
    ///
    /// Set only when the log was verified **intact** — see
    /// `hot_log_reach`. If rows had been archived away, `ts` below the hot
    /// floor is not "before history", it is "the history is in the other file",
    /// and that path raises instead of answering.
    ///
    /// `#[serde(default)]` so the field is additive: a snapshot written without
    /// it deserialises with `false`, which is the right answer for any state
    /// that had rows to fold. Old snapshots cannot actually reach this code —
    /// the container carries `SCHEMA_VERSION` and v8 refused every v7 file
    /// (D-043) — but the tolerance costs nothing and the next field to arrive
    /// may not land in a release that bumps the schema.
    #[serde(default)]
    pub predates_recorded_history: bool,
}

impl MaterializedState {
    /// The state before any log row has been applied.
    fn empty(ts: &str) -> Self {
        Self {
            seq_anchor: 0,
            timestamp: ts.to_string(),
            concepts: HashMap::new(),
            edges: Vec::new(),
            predates_recorded_history: false,
        }
    }
}

/// The newest log payload shape this build writes and the highest it can read.
///
/// Kept beside the folds because they are the only readers, and bumped in step
/// with the `json_object('v', …)` literals in `schema::ddl` — a test asserts the
/// two agree, since nothing else would notice them drifting apart.
pub(crate) const PAYLOAD_VERSION: u8 = 2;

/// Every fold partitions on `(table_name, entity_id)`, never `entity_id` alone.
///
/// The two namespaces are not disjoint and nothing makes them so. A link's
/// `entity_id` is the synthetic `source|target|type|valid_from`; a concept's is
/// whatever the caller passed, unvalidated (defect AD). Partitioning on the id
/// alone therefore lets a concept and a link contend for one window, and
/// `ROW_NUMBER() = 1` hands the whole partition to whichever has the greater
/// `seq_id` — so the loser vanishes from the reconstruction while sitting
/// plainly in both `concepts` and `transaction_log`. Silent, and on the read
/// path the ledger exists to make trustworthy.
///
/// Validating identifiers would make the collision unreachable and is the
/// durable fix; this makes it harmless regardless, which is the property worth
/// having at the fold. `table_name` leads the partition because the log is
/// already indexed on `entity_id` and the discriminator is two values wide.
const HOT_FOLD: &str = r#"
    SELECT seq_id, table_name, entity_id, operation, payload
    FROM (
        SELECT seq_id, table_name, entity_id, operation, payload,
               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id ORDER BY seq_id DESC) as rn
        FROM transaction_log
        WHERE recorded_at <= ?1
    ) WHERE rn = 1
"#;

/// Fold over hot and cold together (§5.5, D-026). Requires `cold` to be ATTACHed.
///
/// The hot entry wins for entities present in both files because its `seq_id` is
/// greater — the same last-writer-wins rule as snapshot composition.
const COLD_FOLD: &str = r#"
    SELECT seq_id, table_name, entity_id, operation, payload
    FROM (
        SELECT seq_id, table_name, entity_id, operation, payload,
               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id ORDER BY seq_id DESC) as rn
        FROM (
            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at FROM main.transaction_log
            UNION ALL
            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at FROM cold.transaction_log
        ) WHERE recorded_at <= ?1
    ) WHERE rn = 1
"#;

/// Fold over the hot log *above a snapshot anchor* (§5.5, D-049).
///
/// `seq_id > ?2` is an inequality, and deliberately so: `AUTOINCREMENT` leaves
/// gaps whenever a transaction rolls back, so successor arithmetic
/// (`seq_id = :anchor + 1`) would stop at the first gap and silently truncate
/// the delta. This is the first anchored fold in the crate, which makes it the
/// first code D-024's rule has ever bound — before this the rule was vacuous,
/// not satisfied.
const ANCHORED_HOT_FOLD: &str = r#"
    SELECT seq_id, table_name, entity_id, operation, payload
    FROM (
        SELECT seq_id, table_name, entity_id, operation, payload,
               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id ORDER BY seq_id DESC) as rn
        FROM transaction_log
        WHERE recorded_at <= ?1 AND seq_id > ?2
    ) WHERE rn = 1
"#;

/// Fold over hot **and cold** above a snapshot anchor (§5.5, 0.5.5).
///
/// The union is what lets composition survive an archive. Rows keep their
/// `seq_id` when they move to cold — the cold schema declares a plain `INTEGER
/// PRIMARY KEY` precisely so history is not renumbered — so `seq_id > ?2`
/// partitions the two files consistently and last-writer-wins across them by the
/// same rule the unanchored folds use.
const ANCHORED_COLD_FOLD: &str = r#"
    SELECT seq_id, table_name, entity_id, operation, payload
    FROM (
        SELECT seq_id, table_name, entity_id, operation, payload,
               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id ORDER BY seq_id DESC) as rn
        FROM (
            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at FROM main.transaction_log
            UNION ALL
            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at FROM cold.transaction_log
        ) WHERE recorded_at <= ?1 AND seq_id > ?2
    ) WHERE rn = 1
"#;

/// The winning log rows for one fold, before they are applied to a base state.
///
/// Absence and disappearance are different facts, and a merge is where the
/// difference starts to matter. A full fold from nothing can treat "this entity
/// went away" and "there is no row for it" identically — both end as absence.
/// Composed onto a snapshot they are opposites: a disappearance must *remove*
/// the entity the snapshot carries, and skipping it leaves the snapshot's stale
/// row standing as though nothing had happened. So they are collected rather
/// than dropped, and the full fold applies them to an empty base, which keeps
/// one code path for both cases (D-049).
///
/// **There is one such set, not two (D-072).** It used to carry `edges_gone`
/// beside `concepts_gone`, and both were populated only from the `'D'` branch of
/// [`fold_delta`] — so when that branch became an error, `edges_gone` was left
/// reachable by nothing. Closing one unreachable path by opening another is not
/// a fix, so it went too.
///
/// The asymmetry is real and worth stating, because "concepts can vanish and
/// edges cannot" looks like an oversight until you follow it:
///
/// * A **concept** disappears by being *retired*, which writes a `'U'` row whose
///   payload has `retired = 1`. That is a genuine removal from a composed state
///   and `concepts_gone` carries it.
/// * An **edge** never disappears. It is retired by asserting a successor over
///   the same interval key — same `source|target|type|valid_from`, later
///   `recorded_at` — so the log row is an `'I'` under the *same* `entity_id`, and
///   last-writer-wins in [`Self::apply_to`] replaces the tuple in place. There is
///   nothing to remove because nothing left; the interval simply closed.
///
/// That is Doctrine III showing through: an edge assertion is immutable and
/// superseded, never deleted.
#[derive(Default)]
struct Delta {
    concepts: HashMap<String, NodeAttributes>,
    /// Keyed by `transaction_log.entity_id`: `source|target|type|valid_from`.
    edges: HashMap<String, (String, String, String, String, String)>,
    /// Concepts retired as of the fold's instant. See the type's note for why
    /// there is no edge equivalent.
    concepts_gone: HashSet<String>,
    max_seq: i64,
}

/// The log's `entity_id` for a link, rebuilt from a materialised edge tuple.
///
/// Must match `trg_links_log_i`'s
/// `source_id || '|' || target_id || '|' || edge_type || '|' || valid_from`
/// exactly, or a delta row will fail to replace the snapshot row it supersedes.
/// Safe because ULIDs are Crockford base32 and edge types are `[A-Z0-9]+`, so
/// `|` cannot occur inside a component (§4.3).
fn edge_key(e: &(String, String, String, String, String)) -> String {
    format!("{}|{}|{}|{}", e.0, e.1, e.2, e.3)
}

/// Release a `cold` handle left attached by an earlier call (§5.5, D-044).
///
/// Both ATTACH sites pair with an unconditional DETACH on the way out, so in
/// the normal course this finds nothing and the statement fails harmlessly with
/// "no such database: cold". It exists for the case the pairing cannot cover: a
/// panic unwinding between the two, which skips the DETACH no matter which exit
/// path the `Result` would have taken.
///
/// A `Drop` guard is the reflex here and does not work — `execute` is `async`,
/// and a `Drop` impl cannot await, so it would build a future, discard it, and
/// leave the handle attached while looking like it had cleaned up. Recovering
/// on the way *in* needs no destructor, works regardless of how the handle
/// leaked, and turns permanent poisoning of the connection into one failed
/// statement nobody sees.
pub(crate) async fn detach_stale_cold(conn: &libsql::Connection) {
    let _ = conn.execute("DETACH DATABASE cold", ()).await;
}

/// Reconstruct database state as believed at past instant `ts` using window-function log fold (§5.5, D-026).
///
/// When `ts` predates the hot log's horizon the cold database is ATTACHed for
/// exactly one fold and DETACHed unconditionally on the way out, error paths
/// included. ATTACH is not transactional and survives ROLLBACK, so a handle
/// leaked by an early return would make every later `reconstruct` *and* every
/// later `archive` fail with "database cold is already in use" — one corrupt
/// payload would permanently poison the connection. This is the same failure
/// mode `archive()` carries a note about, and the two now share a shape.
/// Snapshot composition (§5.5, D-049) applies when `snapshots_dir` holds a
/// snapshot at or before `ts` and no archive database exists — see
/// `snapshot_anchor` for why archiving disables it. Otherwise the fold runs
/// from genesis, which is correct and costs what the whole log costs.
pub async fn reconstruct(
    conn: &libsql::Connection,
    ts: &str,
    archive_path: Option<&Path>,
    snapshots_dir: Option<&Path>,
) -> Result<MaterializedState> {
    match hot_log_reach(conn, ts, archive_path).await? {
        HotLogReach::Covers => {
            if let Some(base) = snapshot_anchor(snapshots_dir, ts).await {
                let anchor = base.seq_anchor;
                let delta =
                    fold_delta(conn, ANCHORED_HOT_FOLD, libsql::params![ts, anchor]).await?;
                return Ok(delta.apply_to(base, ts));
            }
            return fold(conn, ts, HOT_FOLD).await;
        }
        HotLogReach::PredatesRecordedHistory => {
            // Nothing had been recorded by `ts`, and nothing has been removed
            // from the log, so there is no history anywhere to go looking for.
            // The empty state is the answer, flagged so a caller can tell it
            // from a state that is empty because everything was retired.
            let mut state = MaterializedState::empty(ts);
            state.predates_recorded_history = true;
            return Ok(state);
        }
        HotLogReach::NeedsArchive => {}
    }

    // The delta lives in the cold archive database. Both ways of failing to
    // reach it carry `archive_hint`, which is the message the rejected hot-side
    // marker was wanted for — see that function for why no marker is needed.
    //
    // **Computed inside the error arms, not before them.** `NeedsArchive` is the
    // ordinary path to a cold fold and usually succeeds; an eager hint would put
    // an extra query on it for a string almost every caller discards. An
    // injection probe caught this — `a_failed_cold_reconstruct_still_detaches`
    // reached the hint on a run that raised nothing from here.
    let archive = match archive_path {
        Some(p) => p,
        None => {
            return Err(DbError::ReplayCorrupt {
                seq: 0,
                reason: format!(
                    "state at {ts} predates the hot log and no archive path was given; {}",
                    archive_hint(conn).await
                ),
            })
        }
    };
    if !archive.exists() {
        return Err(DbError::ReplayCorrupt {
            seq: 0,
            reason: format!(
                "archive database file {archive:?} does not exist; {}",
                archive_hint(conn).await
            ),
        });
    }

    detach_stale_cold(conn).await;

    // Bound, not interpolated: a path is caller data, and hand-rolled quote
    // doubling is a worse version of what the driver already does correctly.
    conn.execute(
        "ATTACH DATABASE ?1 AS cold",
        libsql::params![archive.to_string_lossy().as_ref()],
    )
    .await?;

    // Composition works across the archive boundary because the anchored fold
    // unions both files; before 0.5.5 it was refused here rather than made to
    // work, and the refusal was the only thing keeping the answer right.
    let result = match snapshot_anchor(snapshots_dir, ts).await {
        Some(base) => {
            let anchor = base.seq_anchor;
            fold_delta(conn, ANCHORED_COLD_FOLD, libsql::params![ts, anchor])
                .await
                .map(|delta| delta.apply_to(base, ts))
        }
        None => fold(conn, ts, COLD_FOLD).await,
    };

    // Unconditional: see the ATTACH note above.
    if let Err(e) = conn.execute("DETACH DATABASE cold", ()).await {
        tracing::warn!("reconstruct: failed to DETACH cold database: {e}");
    }

    result
}

/// Fold from genesis and compare against the composed answer (§5.5, T5.3,
/// D-092).
///
/// # The problem this exists for
///
/// [`crate::temporal::save_snapshot`] is written by `write_final`, which calls
/// [`reconstruct`] — and `reconstruct` composes onto the *previous* snapshot
/// whenever one is usable. So snapshot *n* is derived from snapshot *n−1*, and
/// there is no periodic full fold anywhere in the chain. An error introduced at
/// any link is copied forward indefinitely, and every subsequent read agrees
/// with it, because they are all reading the same descendant.
///
/// The project's own open item names the difficulty honestly: a full fold is
/// exactly the cost snapshots exist to avoid, so this cannot run on every read.
/// It is a **scheduling** problem, and this function is the thing to schedule.
///
/// # It reports; it does not repair
///
/// Deliberate, and not merely conservative. Under [Doctrine VI] a snapshot is
/// derivative and disposable, so the repair is *delete the snapshots* — one
/// line, available to the caller, and correct without this function's help.
/// What the caller cannot get for themselves is the knowledge that the chain
/// diverged, and silently rewriting the file would destroy the only evidence of
/// a bug in composition. A divergence here is not a corrupt database; it is a
/// wrong **cache**, and it means composition has a defect worth finding.
///
/// # Cost
///
/// One fold from genesis over the whole log, plus one composed reconstruction.
/// That is the expensive path by construction — see [`crate::Database::
/// verify_snapshot_chain`] for the handle-level entry point and the note on
/// when to run it.
///
/// [Doctrine VI]: ../../../docs/architecture/s0-s3-foundations.md#doctrine-vi
pub async fn verify_snapshot_chain(
    conn: &libsql::Connection,
    ts: &str,
    archive_path: Option<&Path>,
    snapshots_dir: &Path,
) -> Result<ChainCheck> {
    // The composed answer: what every reader gets today.
    let composed = reconstruct(conn, ts, archive_path, Some(snapshots_dir)).await?;
    // The authority: the same instant, with the snapshot directory withheld, so
    // `snapshot_anchor` finds nothing and the fold runs from genesis. Passing
    // `None` is what makes this an independent computation rather than a second
    // call to the thing under test.
    let folded = reconstruct(conn, ts, archive_path, None).await?;
    Ok(ChainCheck::compare(ts, &composed, &folded))
}

/// The result of a [`verify_snapshot_chain`] cross-check.
///
/// Carries the disagreements rather than a bool, because "the chain diverged" is
/// not actionable and "these three concepts differ, and this edge is present in
/// one and not the other" is. Bounded — see [`ChainCheck::SAMPLE_LIMIT`] — since
/// a chain that went wrong early can disagree about every row, and a report that
/// is the size of the database is one nobody reads.
#[derive(Debug, Clone)]
pub struct ChainCheck {
    pub timestamp: String,
    /// `seq_anchor` of the composed answer and of the genesis fold. These
    /// **may legitimately differ**: the composed answer anchors at the snapshot
    /// it started from plus its delta, and the fold anchors at the newest row it
    /// saw. Reported for diagnosis, never compared.
    pub composed_anchor: i64,
    pub folded_anchor: i64,
    pub composed_concepts: usize,
    pub folded_concepts: usize,
    pub composed_edges: usize,
    pub folded_edges: usize,
    /// Concept ids present in one and not the other, or whose attributes differ.
    pub concept_disagreements: Vec<String>,
    /// Edge keys present in one and not the other.
    pub edge_disagreements: Vec<String>,
    /// True when either list was truncated at [`ChainCheck::SAMPLE_LIMIT`].
    pub truncated: bool,
}

impl ChainCheck {
    /// How many disagreements of each kind to carry.
    pub const SAMPLE_LIMIT: usize = 32;

    pub fn diverged(&self) -> bool {
        !self.concept_disagreements.is_empty() || !self.edge_disagreements.is_empty()
    }

    fn compare(ts: &str, composed: &MaterializedState, folded: &MaterializedState) -> Self {
        let mut concept_disagreements = Vec::new();
        let mut truncated = false;

        let mut ids: Vec<&String> = composed.concepts.keys().collect();
        ids.extend(folded.concepts.keys());
        ids.sort_unstable();
        ids.dedup();
        for id in ids {
            let a = composed.concepts.get(id);
            let b = folded.concepts.get(id);
            let same = match (a, b) {
                (Some(a), Some(b)) => {
                    a.title == b.title
                        && a.content == b.content
                        && a.embedding_model == b.embedding_model
                }
                (None, None) => true,
                _ => false,
            };
            if !same {
                if concept_disagreements.len() < Self::SAMPLE_LIMIT {
                    concept_disagreements.push(id.clone());
                } else {
                    truncated = true;
                }
            }
        }

        // Edges are a `Vec` of tuples with no declared order, so the comparison
        // is on the set. Comparing the vectors directly would report a
        // divergence for a reordering, which is not one — and that false
        // positive is worse than useless here, because the whole point of this
        // check is that a report means "go and find the bug".
        let key = |e: &(String, String, String, String, String)| {
            format!("{}|{}|{}|{}|{}", e.0, e.1, e.2, e.3, e.4)
        };
        let ca: HashSet<String> = composed.edges.iter().map(key).collect();
        let fa: HashSet<String> = folded.edges.iter().map(key).collect();
        let mut edge_disagreements: Vec<String> = ca.symmetric_difference(&fa).cloned().collect();
        edge_disagreements.sort_unstable();
        if edge_disagreements.len() > Self::SAMPLE_LIMIT {
            edge_disagreements.truncate(Self::SAMPLE_LIMIT);
            truncated = true;
        }

        Self {
            timestamp: ts.to_string(),
            composed_anchor: composed.seq_anchor,
            folded_anchor: folded.seq_anchor,
            composed_concepts: composed.concepts.len(),
            folded_concepts: folded.concepts.len(),
            composed_edges: ca.len(),
            folded_edges: fa.len(),
            concept_disagreements,
            edge_disagreements,
            truncated,
        }
    }
}

impl std::fmt::Display for ChainCheck {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if !self.diverged() {
            return write!(
                f,
                "snapshot chain agrees with a genesis fold at {}: {} concepts, {} edges",
                self.timestamp, self.folded_concepts, self.folded_edges
            );
        }
        write!(
            f,
            "snapshot chain DIVERGED at {}: composed {} concepts / {} edges, \
             genesis fold {} concepts / {} edges; {} concept and {} edge \
             disagreements{}. The snapshots are a wrong cache, not a corrupt \
             ledger — deleting the snapshot directory restores correctness and \
             loses only speed (Doctrine VI). concepts: {:?} edges: {:?}",
            self.timestamp,
            self.composed_concepts,
            self.composed_edges,
            self.folded_concepts,
            self.folded_edges,
            self.concept_disagreements.len(),
            self.edge_disagreements.len(),
            if self.truncated { " (truncated)" } else { "" },
            self.concept_disagreements,
            self.edge_disagreements,
        )
    }
}

/// The newest usable snapshot at or before `ts`, or `None` to fold from genesis.
///
/// **Composition used to be disabled once an archive database existed, and as of
/// 0.5.5 it is not.** The reason for the refusal was real: `LOG_ARCHIVABLE`
/// (§5.7) removes superseded rows scattered through the sequence, so a row above
/// the anchor and at or before `ts` could be in cold while a newer row for the
/// same entity — recorded *after* `ts`, invisible to the fold — kept it out of
/// the hot log. The delta missed it and the snapshot answered with a stale
/// value. The fix is the one that note named: the cold log is now in the delta,
/// via [`ANCHORED_COLD_FOLD`], so the archived row is visible again and there is
/// nothing left to refuse.
///
/// Selection loads candidates newest-first and stops at the first whose
/// timestamp is at or before `ts`, so the common case — `reconstruct(now)` —
/// reads exactly one file. A snapshot this build cannot read
/// ([`DbError::SnapshotIncompatible`], D-043) is skipped, not raised: an
/// incompatible snapshot is an ordinary consequence of upgrading, and the whole
/// point of distinguishing it from corruption is that the answer is to carry on
/// without it.
///
/// # It runs on a blocking thread, and a lost one costs speed only (0.13.11, W8.1, D-184)
///
/// The scan is a directory listing plus one or more full
/// [`load_snapshot`](super::snapshot::load_snapshot) calls — decompression and
/// bincode over the whole state, on a worker that has other tasks waiting. The
/// *whole scan* is offloaded rather than each file, because the loop is
/// sequential by construction (it stops at the first usable file) and a hop per
/// candidate would add scheduling to a path whose common case reads exactly one.
///
/// A [`tokio::task::JoinError`] means the loader panicked, and the answer is the
/// same one this function already gives for every other kind of unusable file:
/// `None`, and fold from genesis. That is not leniency, it is what a snapshot
/// *is* — derivative and disposable under [Doctrine VI], so the cost of ignoring
/// one is a slower reconstruction and never a wrong one. It is also a real
/// improvement over the previous arrangement: inline, a panic in the loader
/// unwound through [`reconstruct`] and took the caller's task with it, which
/// meant a single corrupt file could stop a process that had a correct answer
/// available the whole time. W8.4 fuzzes for exactly those panics; this is what
/// happens to the ones it has not found yet.
///
/// [Doctrine VI]: ../../../docs/architecture/s0-s3-foundations.md#doctrine-vi
async fn snapshot_anchor(snapshots_dir: Option<&Path>, ts: &str) -> Option<MaterializedState> {
    let dir = snapshots_dir?.to_path_buf();
    let ts = ts.to_string();
    match tokio::task::spawn_blocking(move || newest_usable_snapshot(&dir, &ts)).await {
        Ok(found) => found,
        Err(e) => {
            tracing::warn!("the snapshot scan did not finish ({e}); folding from genesis");
            None
        }
    }
}

/// The blocking half of [`snapshot_anchor`]: read the directory, load
/// newest-first, stop at the first snapshot at or before `ts`.
fn newest_usable_snapshot(dir: &Path, ts: &str) -> Option<MaterializedState> {
    let mut candidates: Vec<(i64, PathBuf)> = std::fs::read_dir(dir)
        .ok()?
        .flatten()
        .map(|e| e.path())
        .filter_map(|p| super::snapshot::seq_from_filename(&p).map(|s| (s, p)))
        .collect();
    candidates.sort_by_key(|(seq, _)| std::cmp::Reverse(*seq));

    for (_, path) in candidates {
        match super::snapshot::load_snapshot(&path) {
            // Sound as a string comparison because every timestamp is the
            // canonical fixed width (D-029).
            Ok(state) if state.timestamp.as_str() <= ts => return Some(state),
            Ok(_) => continue,
            Err(DbError::SnapshotIncompatible { reason, .. }) => {
                tracing::warn!("skipping snapshot {path:?}: {reason}");
                continue;
            }
            Err(e) => {
                tracing::warn!("skipping unreadable snapshot {path:?}: {e}");
                continue;
            }
        }
    }
    None
}

/// Where the answer for `ts` lives.
///
/// Three cases, not two (0.8.0, B5, D-121). This used to be a `bool`, and the
/// missing third case is the whole of B5: *below the log's floor* was folded in
/// with *the delta is elsewhere*, so a question about a time before the ledger
/// started came back as [`DbError::ReplayCorrupt`] — the class meaning the
/// ledger is damaged — naming an archive file the caller had never created.
enum HotLogReach {
    /// The hot log holds everything needed at `ts`. Fold it.
    Covers,
    /// Nothing had been recorded by `ts`, and nothing has ever been removed
    /// from the log, so no other file could hold it either. The empty state is
    /// the correct answer, not a failure to find one.
    PredatesRecordedHistory,
    /// The delta is in the cold archive. If it cannot be reached, that is an
    /// error and stays one.
    NeedsArchive,
}

/// Whether the hot log alone can answer for `ts` — a *completeness* test.
///
/// **This replaces a reach test that was not one (0.5.5).** The previous version
/// asked `MIN(recorded_at) <= ts`: whether the hot log stretches back far enough
/// to contain `ts`. That is a different question from whether it still contains
/// everything needed to answer at `ts`, and `LOG_ARCHIVABLE` (§5.7) is exactly
/// what pulls the two apart — it removes *superseded* rows, scattered through
/// the sequence rather than forming a prefix. One entity archived and another
/// not is enough: the unarchived one keeps `MIN` pointing before the cutoff
/// while the archived one's winning row is gone, and the fold silently returns a
/// state missing an entity. Measured, not theorised — see
/// `reconstructing_before_the_archive_cutoff_keeps_every_entity`.
///
/// The sound test rests on the one guarantee the archive does make: **the newest
/// row per entity is never archivable**, because archivability requires a later
/// row to exist. So if `ts` is at or after the newest hot stamp, every entity's
/// winning row at `ts` is its newest row overall, and every such row is hot.
/// That covers `reconstruct(now)` — the common case, and the case §5.7 designed
/// `LOG_ARCHIVABLE` around — and nothing else.
///
/// Anything earlier goes to the cold file. That is more ATTACHes than the old
/// rule performed, and the trade is not close: the old rule was cheaper because
/// it was answering a question nobody asked.
///
/// With no archive database in play the reach test *is* the completeness test —
/// nothing has been removed, so the hot log is the whole log — and it is kept,
/// because it is also what distinguishes "before recorded history" from "the
/// cold file is missing" (D-026).
async fn hot_log_reach(
    conn: &libsql::Connection,
    ts: &str,
    archive_path: Option<&Path>,
) -> Result<HotLogReach> {
    let row = conn
        .query(
            "SELECT MIN(recorded_at), MAX(recorded_at) FROM transaction_log",
            (),
        )
        .await?
        .next()
        .await?;
    let (min_recorded_at, max_recorded_at): (Option<String>, Option<String>) = match row {
        Some(r) => (r.get(0).ok(), r.get(1).ok()),
        None => (None, None),
    };

    // Sound as string comparisons because every recorded_at is the canonical
    // fixed width (D-029).
    if archive_path.is_some_and(|p| p.exists()) {
        // An empty hot log beside an archive is the fully-archived case and
        // covers nothing. It cannot arise from `archive()` itself — the newest
        // row per entity always stays — but answering "covered" here would make
        // such a file reconstruct to the empty state with no error at all.
        return Ok(match max_recorded_at {
            Some(max_ts) if max_ts.as_str() <= ts => HotLogReach::Covers,
            _ => HotLogReach::NeedsArchive,
        });
    }

    match min_recorded_at {
        Some(min_ts) if min_ts.as_str() <= ts => Ok(HotLogReach::Covers),
        // No log at all: a genuinely empty database, and the empty state has
        // always been the answer here.
        None => Ok(HotLogReach::PredatesRecordedHistory),
        // `ts` is below the hot log's floor, and there is no archive file to
        // consult. Which of the two meanings that has is decided by whether
        // anything was ever removed from the log — see `hot_log_is_intact`.
        Some(_) => Ok(if hot_log_is_intact(conn).await? {
            HotLogReach::PredatesRecordedHistory
        } else {
            HotLogReach::NeedsArchive
        }),
    }
}

/// What the caller needs to know when the cold delta cannot be reached —
/// **assembled from the hot file alone** (0.9.0, C4).
///
/// # This is the message the hot-side marker was wanted for
///
/// [D-121](../../docs/architecture/s13-decision-register.md) rejected a hot-side
/// marker recording *archived at* and *horizon*, then left the door open: 0.9.0
/// was to adopt it "only if it wants the richer message". C4 asked for the
/// message and found the marker cannot supply it, because the proposed message —
/// *"this database was archived on X; pass the archive path"* — is **weaker**
/// than what the hot log already carries:
///
/// * *how many rows went* is `MAX(seq_id) - COUNT(*)`, exact for the reason
///   [`hot_log_is_intact`] gives;
/// * *how far back the hot file still reaches* is `MIN(seq_id)` and its
///   `recorded_at` — which is the fact that actually tells a caller whether the
///   archive is worth fetching, and which a marker's archive **timestamp** does
///   not give them;
/// * *that archiving happened at all* is the one bit [`hot_log_is_intact`]
///   already answers.
///
/// The only datum a marker would add is the wall-clock instant of the last
/// archive run, and no branch and no caller needs it. So the marker is refused
/// outright rather than deferred again: under
/// [D-036](../../docs/architecture/s13-decision-register.md) a hot-table addition
/// lands pre-1.0 or not at all, and a table whose whole content is a timestamp
/// used in one error string is not worth a rung.
///
/// # There is no "nothing was archived" case, and that was settled by injection
///
/// This first carried a branch for `removed == 0`, on the reasoning that the
/// `NeedsArchive` arm is reachable without any archiving. That reasoning was
/// **wrong about where the cost lands and right about the branch**, and only a
/// probe told the two apart: replacing the branch body with a panic showed it
/// firing from `a_failed_cold_reconstruct_still_detaches`, a test that raises
/// nothing from here — because the hint was being computed *before* the two
/// arms that use it, on every cold fold. Made lazy, the probe went quiet across
/// all 27 targets.
///
/// So the branch was dead at the use sites: both arms require
/// [`hot_log_is_intact`] to have returned false, or an archive file to have
/// existed when `hot_log_reach` looked and to have gone by the time this did.
/// Rows really were removed in every case that gets here, and the message may
/// say so without qualification. Deleted rather than kept as a defensive
/// fallback, for the reason `delete_guarded` records about
/// `classify_archive_violation`: unreachable code that looks reasonable is
/// harder to remove later than now.
///
/// Best-effort by construction: this runs on the error path, where a second
/// failure must not replace the diagnosis with its own. A query that does not
/// answer yields a hint that says so, and the caller still gets the error it came
/// for.
async fn archive_hint(conn: &libsql::Connection) -> String {
    // `COUNT(*)` always returns a row, so `None` here means the query itself
    // failed and there is nothing to say beyond that.
    let row = match conn
        .query(
            "SELECT COUNT(*), MIN(seq_id), MAX(seq_id), MIN(recorded_at) FROM transaction_log",
            (),
        )
        .await
    {
        Ok(mut rows) => rows.next().await.ok().flatten(),
        Err(_) => None,
    };

    let Some(row) = row else {
        return "the hot log could not be inspected for an archive horizon".into();
    };
    let count: i64 = row.get(0).unwrap_or(0);
    if count == 0 {
        return "the hot log is empty".into();
    }
    let min: i64 = row.get(1).unwrap_or(0);
    let max: i64 = row.get(2).unwrap_or(0);
    let floor: String = row.get(3).unwrap_or_default();
    let removed = max - count;

    format!(
        "{removed} log rows have been archived out of this database; the hot log \
         now begins at seq_id {min} ({floor})"
    )
}

/// Was any row ever removed from `transaction_log`? — answered exactly, from
/// the hot file alone (0.8.0, B5, D-121).
///
/// # Why this question needs answering at all
///
/// With `ts` below the hot log's floor and no archive file present, the state
/// on disk is consistent with two very different histories: **nothing was ever
/// archived**, in which case the hot log is the whole log and the answer to
/// *what was believed at `ts`* is "nothing yet"; or **rows were archived and
/// the cold file is gone**, in which case the answer is unknowable and saying
/// "nothing" would be inventing one. Before this, the two were conflated and
/// both raised — which made an ordinary question about a young database report
/// the ledger as damaged.
///
/// # Why `seq_id` settles it, with no marker and no schema change
///
/// `transaction_log.seq_id` is `INTEGER PRIMARY KEY AUTOINCREMENT`, so values
/// are allocated 1, 2, 3, … and **never reused**. A rolled-back transaction
/// leaves no gap — `sqlite_sequence` rolls back with it, which
/// [D-049](../../docs/architecture/s13-decision-register.md) established by
/// measurement after assuming the opposite. So the only thing that can perturb
/// the sequence is deletion, and `trg_txlog_guard_delete` confines deletion to
/// an archive session.
///
/// Therefore: if nothing was removed, the ids are exactly `1..=MAX` and
/// `COUNT(*) == MAX(seq_id)` with `MIN(seq_id) == 1`. And conversely — this is
/// the half that makes it a proof rather than a heuristic — those two equalities
/// force the set of `COUNT` distinct ids inside `[1, MAX]` to be all of it, so
/// nothing is missing. The test is exact in both directions, not merely
/// suggestive.
///
/// **It does not depend on the archive removing a contiguous block**, which it
/// does not: `archive()` removes *superseded* rows scattered through the
/// sequence. Scattered removal leaves interior gaps, which fails the count
/// equality; removal from the front raises `MIN` above 1. Removal from the end
/// cannot happen, because the newest row per entity is never archivable.
///
/// # What it deliberately does not claim
///
/// Nothing about *when* the archiving happened or *what* went, which is what
/// the rejected hot-side marker would have carried. It answers one bit, and one
/// bit is what the branch above needs.
async fn hot_log_is_intact(conn: &libsql::Connection) -> Result<bool> {
    let row = conn
        .query(
            "SELECT COUNT(*), MIN(seq_id), MAX(seq_id) FROM transaction_log",
            (),
        )
        .await?
        .next()
        .await?;
    let Some(row) = row else {
        return Ok(true);
    };
    let count: i64 = row.get(0).unwrap_or(0);
    if count == 0 {
        return Ok(true);
    }
    let min: i64 = row.get(1).unwrap_or(0);
    let max: i64 = row.get(2).unwrap_or(0);
    Ok(min == 1 && count == max)
}

/// Whether a connection alone can fold `transaction_log` at `ts` (W7.1, D-174).
///
/// The completeness question [`hot_log_reach`] answers, minus the archive file
/// it does not have. Both callers take a `Connection`, so when the hot log is
/// short they have nowhere to go and must refuse rather than fold what is left:
/// [`crate::graph::TraversalBuilder::as_of_recorded`] folds for topology, and
/// [`crate::temporal::hydrate_attributes`] folds for the text (0.13.16, W9.1,
/// [D-189](../../docs/architecture/s13-decision-register.md#d-189)). The second
/// was folding without asking, which is what §3.2 was.
///
/// **One bit, and the conservative one.** `hot_log_is_intact` says whether
/// anything was ever removed, not whether *this* instant survived the removal.
/// The archive cutoff is not recorded hot-side — that is the marker D-132
/// refused — so an archived database refuses every instant here, including ones
/// a fold would have got right. `ts` is taken anyway rather than dropped from the
/// signature, because the refusal names it and because a cutoff-aware version
/// would need it.
pub(crate) async fn hot_log_answers_for(conn: &libsql::Connection, _ts: &str) -> Result<bool> {
    hot_log_is_intact(conn).await
}

/// Run one fold query from nothing — the unanchored path.
async fn fold(conn: &libsql::Connection, ts: &str, query: &str) -> Result<MaterializedState> {
    let delta = fold_delta(conn, query, libsql::params![ts]).await?;
    Ok(delta.apply_to(MaterializedState::empty(ts), ts))
}

/// Run one fold query and collect the winning rows, deletions included.
async fn fold_delta(
    conn: &libsql::Connection,
    query: &str,
    params: impl libsql::params::IntoParams,
) -> Result<Delta> {
    let mut rows = conn.query(query, params).await?;
    let mut d = Delta::default();
    let (concepts, edges, max_seq) = (&mut d.concepts, &mut d.edges, &mut d.max_seq);

    while let Some(row) = rows.next().await? {
        let seq_id: i64 = row.get(0)?;
        let table_name: String = row.get(1)?;
        let _entity_id: String = row.get(2)?;
        let op: String = row.get(3)?;
        let payload_str: String = row.get(4)?;

        if seq_id > *max_seq {
            *max_seq = seq_id;
        }

        // A `'D'` row is corruption, not a tombstone (D-072).
        //
        // Doctrine V permits no physical delete outside an archive session, and
        // the archive *moves* rows rather than logging their removal — so no
        // trigger in the schema writes a `'D'`, and no code path in the crate
        // can produce one. This arm used to treat it as a tombstone, which read
        // as a claim that deletions are recorded and reconstructible. They are
        // not. Refusing here makes the doctrine enforced at the fold rather than
        // assumed by it, and is the same call D-060 made for overlap: the layer
        // that can notice should.
        //
        // Retirement is unaffected and is the mechanism that actually removes a
        // concept from a composed state — see the `retired != 0` branch below,
        // which is where `concepts_gone` is populated in practice.
        if op == "D" {
            return Err(DbError::ReplayCorrupt {
                seq: seq_id,
                reason: format!(
                    "transaction_log carries a 'D' operation for {table_name} \
                     entity {_entity_id:?}; Doctrine V permits no physical delete \
                     outside an archive session, and the archive logs none. This \
                     row was not written by this crate."
                ),
            });
        }

        let payload: serde_json::Value =
            serde_json::from_str(&payload_str).map_err(|e| DbError::ReplayCorrupt {
                seq: seq_id,
                reason: format!("Failed to parse payload JSON: {e}"),
            })?;

        // v1 and v2 differ by one added field, so v1 folds by reading it as
        // absent — which is what `Option` already means here. A future shape
        // that *removes* or *retypes* a field would not be able to share this
        // path, and would want a match on `v` rather than a ceiling.
        let v = payload.get("v").and_then(|v| v.as_u64()).unwrap_or(1);
        if v > PAYLOAD_VERSION as u64 {
            return Err(DbError::PayloadVersion {
                got: v as u8,
                max: PAYLOAD_VERSION,
            });
        }

        if table_name == "concepts" {
            let id = _entity_id;
            let retired = payload.get("retired").and_then(|r| r.as_i64()).unwrap_or(0);
            if retired == 0 {
                let title = payload
                    .get("title")
                    .and_then(|s| s.as_str())
                    .unwrap_or("")
                    .to_string();
                let content = payload
                    .get("content")
                    .and_then(|s| s.as_str())
                    .unwrap_or("")
                    .to_string();
                let embedding_model = payload
                    .get("embedding_model")
                    .and_then(|s| s.as_str())
                    .map(|s| s.to_string());
                concepts.insert(
                    id.clone(),
                    NodeAttributes {
                        id,
                        title,
                        content,
                        embedding_model,
                    },
                );
            } else {
                // Retirement is the application axis (§4.1), and a reconstruction
                // shows what was visible. Onto a snapshot that means removing
                // the concept, not declining to add it.
                d.concepts_gone.insert(id);
            }
        } else if table_name == "links" {
            let src = payload
                .get("source_id")
                .and_then(|s| s.as_str())
                .unwrap_or("")
                .to_string();
            let tgt = payload
                .get("target_id")
                .and_then(|s| s.as_str())
                .unwrap_or("")
                .to_string();
            let edge_type = payload
                .get("edge_type")
                .and_then(|s| s.as_str())
                .unwrap_or("")
                .to_string();
            let vf = payload
                .get("valid_from")
                .and_then(|s| s.as_str())
                .unwrap_or("")
                .to_string();
            let vt = payload
                .get("valid_to")
                .and_then(|s| s.as_str())
                .unwrap_or("")
                .to_string();
            edges.insert(_entity_id, (src, tgt, edge_type, vf, vt));
        }
    }

    Ok(d)
}

impl Delta {
    /// Compose onto `base` under last-writer-wins by `seq_id` (§5.5).
    ///
    /// The delta is by construction newer than the base — it is the fold of
    /// everything above the base's anchor — so every row it carries wins, and
    /// every retirement it carries removes. This is the same rule
    /// `trg_links_current_sync`'s upsert applies and the same rule the cold
    /// fold applies; that the three agree is asserted by test rather than by
    /// this comment (§8).
    fn apply_to(self, base: MaterializedState, ts: &str) -> MaterializedState {
        let mut concepts = base.concepts;
        let mut edges: HashMap<String, (String, String, String, String, String)> =
            base.edges.into_iter().map(|e| (edge_key(&e), e)).collect();

        for id in self.concepts_gone {
            concepts.remove(&id);
        }
        // No edge equivalent: an edge is superseded in place under the same
        // `entity_id`, never removed — see [`Delta`] (D-072).
        concepts.extend(self.concepts);
        edges.extend(self.edges);

        // Sorted so the result is a function of the state and not of hash
        // iteration order — `reconstruct` is compared against itself by the
        // property suite, and two runs must be equal, not merely equivalent.
        let mut edges: Vec<_> = edges.into_values().collect();
        edges.sort();

        MaterializedState {
            seq_anchor: self.max_seq.max(base.seq_anchor),
            timestamp: ts.to_string(),
            concepts,
            edges,
            // A delta was applied, so there was history to fold. `reconstruct`
            // sets the flag on the one path that never gets here.
            predates_recorded_history: false,
        }
    }
}