grit-core 0.2.3

//! The single-writer actor (see AGENTS.md "Write path"). SQLite has one
//! writer; we embrace it: all mutations flow through one thread owning the
//! write connection. Callers talk to it over an mpsc channel.
//!
//! Op application is **idempotent** (an op id already in the oplog is skipped)
//! and **commutative for concurrent ops** (conflicting row versions resolve by
//! HLC, invalidations converge to the minimum) — the sync guarantee of Design
//! Invariant 5, property-tested in `tests/convergence.rs`.

use std::sync::Arc;
use std::sync::mpsc::{Receiver, SyncSender, sync_channel};
use std::thread::JoinHandle;

use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
use uuid::Uuid;

use crate::clock::Clock;
use crate::error::{Error, Result};
use crate::ops::{GraphOp, OplogEntry};
use crate::vecext::f32s_as_bytes;

// Apply is by far the most common message; boxing OplogEntry to appease the
// variant-size lint would add an allocation to every write for a transient
// message that lives microseconds.
#[allow(clippy::large_enum_variant)]
pub(crate) enum WriteMsg {
    Apply {
        entry: OplogEntry,
        reply: SyncSender<Result<bool>>,
    },
    RegisterModel {
        model_id: String,
        dim: usize,
        version: String,
        reply: SyncSender<Result<()>>,
    },
    SetEmbedding {
        table: VecTable,
        id: Uuid,
        vector: Vec<f32>,
        reply: SyncSender<Result<()>>,
    },
    Shutdown,
}

#[derive(Debug, Clone, Copy)]
pub(crate) enum VecTable {
    Nodes,
    Edges,
}

impl VecTable {
    fn name(self) -> &'static str {
        match self {
            VecTable::Nodes => "vec_nodes",
            VecTable::Edges => "vec_edges",
        }
    }
}

pub(crate) fn spawn(
    mut conn: Connection,
    clock: Arc<dyn Clock>,
    rx: Receiver<WriteMsg>,
) -> JoinHandle<()> {
    std::thread::Builder::new()
        .name("grit-writer".into())
        .spawn(move || {
            while let Ok(msg) = rx.recv() {
                match msg {
                    WriteMsg::Apply { entry, reply } => {
                        let _ = reply.send(apply_entry(&mut conn, clock.as_ref(), &entry));
                    }
                    WriteMsg::RegisterModel {
                        model_id,
                        dim,
                        version,
                        reply,
                    } => {
                        let _ = reply.send(register_model(&mut conn, &model_id, dim, &version));
                    }
                    WriteMsg::SetEmbedding {
                        table,
                        id,
                        vector,
                        reply,
                    } => {
                        let _ = reply.send(set_embedding(&mut conn, table, id, &vector));
                    }
                    WriteMsg::Shutdown => break,
                }
            }
        })
        .expect("failed to spawn grit-writer thread")
}

/// Blocking round-trip to the writer thread.
pub(crate) fn call<T>(
    tx: &std::sync::mpsc::Sender<WriteMsg>,
    make: impl FnOnce(SyncSender<Result<T>>) -> WriteMsg,
) -> Result<T> {
    let (reply_tx, reply_rx) = sync_channel(1);
    tx.send(make(reply_tx)).map_err(|_| Error::Closed)?;
    reply_rx.recv().map_err(|_| Error::Closed)?
}

/// Append `entry` to the oplog and apply its derived state in one transaction.
/// Returns `false` (and changes nothing) if the op id was already applied.
fn apply_entry(conn: &mut Connection, clock: &dyn Clock, entry: &OplogEntry) -> Result<bool> {
    let op_json = serde_json::to_string(&entry.op)?;
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let inserted = tx.execute(
        "INSERT OR IGNORE INTO oplog (id, hlc, device_id, op, applied_at)
         VALUES (?1, ?2, ?3, ?4, ?5)",
        params![
            entry.id.to_string(),
            entry.hlc.encode(),
            entry.device_id,
            op_json,
            clock.now_ms(),
        ],
    )?;
    if inserted == 0 {
        // Idempotency: this op has already been applied here.
        return Ok(false);
    }
    apply_op(&tx, entry)?;
    tx.commit()?;
    Ok(true)
}

fn apply_op(tx: &Transaction<'_>, entry: &OplogEntry) -> Result<()> {
    let hlc = entry.hlc.encode();
    // Derived-state timestamps come from the op's HLC wall time, not the local
    // clock, so replaying an oplog reproduces identical rows on every device.
    let wall = entry.hlc.wall_ms;
    match &entry.op {
        GraphOp::AddNode {
            id,
            kind,
            name,
            summary,
            attrs,
            group_id,
        } => {
            // On id collision (two devices minting the same UUIDv7 is
            // practically impossible, but replays make it reachable), the
            // lowest HLC wins deterministically regardless of apply order.
            // A tombstoned id stays dead (Purge commutes with adds).
            tx.execute(
                "INSERT INTO nodes (id, kind, name, summary, attrs, group_id, created_at, hlc)
                 SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8
                 WHERE NOT EXISTS (SELECT 1 FROM purged WHERE id = ?1)
                 ON CONFLICT(id) DO UPDATE SET
                     kind = excluded.kind, name = excluded.name,
                     summary = excluded.summary, attrs = excluded.attrs,
                     group_id = excluded.group_id, created_at = excluded.created_at,
                     hlc = excluded.hlc
                 WHERE excluded.hlc < nodes.hlc",
                params![
                    id.to_string(),
                    kind,
                    name,
                    summary,
                    serde_json::to_string(attrs)?,
                    group_id,
                    wall,
                    hlc,
                ],
            )?;
            // An UpdateNode may have been applied before this row arrived
            // (concurrent ops), and a lower-HLC AddNode replay rewrites row
            // content wholesale — re-fold the per-field winners either way.
            fold_node_updates(tx, &id.to_string())?;
        }
        GraphOp::UpdateNode {
            id,
            name,
            summary,
            kind,
            attrs,
        } => {
            // Append-only per-field record first (survives even if the node
            // row hasn't arrived yet), then fold into the row. Concurrent
            // updates of the same field converge to the highest HLC —
            // commutative and idempotent. A tombstoned id stays dead.
            let id_s = id.to_string();
            let attrs_s = attrs.as_ref().map(serde_json::to_string).transpose()?;
            let fields: [(&str, Option<&str>); 4] = [
                ("name", name.as_deref()),
                ("summary", summary.as_deref()),
                ("kind", kind.as_deref()),
                ("attrs", attrs_s.as_deref()),
            ];
            for (field, value) in fields {
                let Some(value) = value else { continue };
                let mut stmt = tx.prepare_cached(
                    "INSERT INTO node_updates (node_id, field, value, hlc)
                     SELECT ?1, ?2, ?3, ?4
                     WHERE NOT EXISTS (SELECT 1 FROM purged WHERE id = ?1)
                     ON CONFLICT(node_id, field) DO UPDATE SET
                         value = excluded.value, hlc = excluded.hlc
                     WHERE excluded.hlc > node_updates.hlc",
                )?;
                stmt.execute(params![id_s, field, value, hlc])?;
            }
            fold_node_updates(tx, &id_s)?;
        }
        GraphOp::AddEdge {
            id,
            src,
            dst,
            rel,
            fact,
            attrs,
            group_id,
            valid_at,
            invalid_at,
        } => {
            // Endpoints resolve through any already-applied merges, and every
            // merge re-points existing edges when it applies — so add/merge
            // interleavings converge on the same endpoints either way.
            let src = resolve_canonical(tx, &src.to_string())?;
            let dst = resolve_canonical(tx, &dst.to_string())?;
            tx.execute(
                "INSERT INTO edges (id, src, dst, rel, fact, attrs, group_id, valid_at,
                                    invalid_at, created_at, hlc)
                 SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11
                 WHERE NOT EXISTS (SELECT 1 FROM purged WHERE id = ?1)
                 ON CONFLICT(id) DO UPDATE SET
                     src = excluded.src, dst = excluded.dst, rel = excluded.rel,
                     fact = excluded.fact, attrs = excluded.attrs,
                     group_id = excluded.group_id, valid_at = excluded.valid_at,
                     invalid_at = excluded.invalid_at,
                     created_at = excluded.created_at, hlc = excluded.hlc
                 WHERE excluded.hlc < edges.hlc",
                params![
                    id.to_string(),
                    src,
                    dst,
                    rel,
                    fact,
                    serde_json::to_string(attrs)?,
                    group_id,
                    valid_at,
                    invalid_at,
                    wall,
                    hlc,
                ],
            )?;
            // An InvalidateEdge for this edge may have been applied before the
            // edge itself arrived (concurrent ops, arbitrary merge order):
            // re-fold the append-only invalidation records over the row.
            fold_edge_invalidations(tx, &id.to_string())?;
        }
        GraphOp::AddEpisode {
            id,
            source,
            kind,
            content,
            occurred_at,
            group_id,
            mentions,
        } => {
            tx.execute(
                "INSERT INTO episodes (id, source, kind, content, occurred_at, group_id,
                                       created_at, hlc)
                 SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8
                 WHERE NOT EXISTS (SELECT 1 FROM purged WHERE id = ?1)
                 ON CONFLICT(id) DO UPDATE SET
                     source = excluded.source, kind = excluded.kind,
                     content = excluded.content,
                     occurred_at = excluded.occurred_at, group_id = excluded.group_id,
                     created_at = excluded.created_at, hlc = excluded.hlc
                 WHERE excluded.hlc < episodes.hlc",
                params![
                    id.to_string(),
                    source,
                    kind,
                    content,
                    occurred_at,
                    group_id,
                    wall,
                    hlc
                ],
            )?;
            for target in mentions {
                // Mentions keep the ORIGINAL target id, are never re-pointed
                // by merges (reads expand merge aliases instead), and purge of
                // either endpoint deletes + tombstones them. Storing anything
                // resolution-dependent here would make the final row depend on
                // which merges had already been applied — order-dependent.
                let mut stmt = tx.prepare_cached(
                    "INSERT OR IGNORE INTO mentions (episode_id, target_id)
                     SELECT ?1, ?2
                     WHERE NOT EXISTS (SELECT 1 FROM purged WHERE id IN (?1, ?2))",
                )?;
                stmt.execute(params![id.to_string(), target.to_string()])?;
            }
        }
        GraphOp::InvalidateEdge {
            edge_id,
            invalid_at,
        } => {
            // Append-only record first (survives even if the edge row hasn't
            // arrived yet), then fold into the edge. Concurrent invalidations
            // converge to MIN(invalid_at) — commutative and idempotent. Two
            // devices invalidating the same edge at the same event time
            // collide on the (edge_id, invalid_at) PK: field-wise MIN keeps
            // the earliest recording deterministically in every apply order
            // (recorded_at is the HLC wall, so the two fields stay coherent).
            tx.execute(
                "INSERT INTO edge_invalidations
                     (edge_id, invalid_at, recorded_at, hlc)
                 SELECT ?1, ?2, ?3, ?4
                 WHERE NOT EXISTS (SELECT 1 FROM purged WHERE id = ?1)
                 ON CONFLICT(edge_id, invalid_at) DO UPDATE SET
                     recorded_at = MIN(recorded_at, excluded.recorded_at),
                     hlc = MIN(hlc, excluded.hlc)",
                params![edge_id.to_string(), invalid_at, wall, hlc],
            )?;
            fold_edge_invalidations(tx, &edge_id.to_string())?;
        }
        GraphOp::MergeNodes { from, into } => {
            if from == into {
                // Defensive: Grit::apply rejects this before it reaches the
                // oplog, but a remote entry could still carry it.
                return Ok(());
            }
            let (from_s, into_s) = (from.to_string(), into.to_string());
            // If `from` hasn't arrived yet (concurrent op order), record the
            // merge on a stub row; the max-HLC sentinel guarantees the real
            // AddNode's content wins the upsert when it lands, while the
            // merge marker survives (AddNode never touches merged_into).
            tx.execute(
                "INSERT INTO nodes (id, kind, name, summary, attrs, group_id,
                                    created_at, hlc)
                 SELECT ?1, '', '', '', '{}', '', ?2, ?3
                 WHERE NOT EXISTS (SELECT 1 FROM nodes WHERE id = ?1)
                   AND NOT EXISTS (SELECT 1 FROM purged WHERE id = ?1)",
                params![from_s, wall, MAX_HLC_SENTINEL],
            )?;
            // First merge of a node wins the audit pointer; concurrent merges
            // of the SAME `from` node into different targets are a Layer 2
            // protocol violation (it is the single merge decider) and are not
            // part of the convergence contract.
            tx.execute(
                "UPDATE nodes
                 SET merged_into = ?2,
                     expired_at = CASE
                         WHEN expired_at IS NULL OR expired_at > ?3 THEN ?3
                         ELSE expired_at END
                 WHERE id = ?1 AND merged_into IS NULL",
                params![from_s, into_s, wall],
            )?;
            // A merge that lands after `from` was purged still records its
            // pointer on the tombstone — resolution stays order-independent.
            tx.execute(
                "UPDATE purged SET merged_into = ?2
                 WHERE id = ?1 AND merged_into IS NULL",
                params![from_s, into_s],
            )?;
            // A stub row created above must still carry any already-applied
            // updates, or Merge-then-Update and Update-then-Merge would
            // leave different rows until the real AddNode lands.
            fold_node_updates(tx, &from_s)?;
            repoint_aliases(tx, &into_s)?;
        }
        GraphOp::Purge { ids } => {
            purge(tx, ids)?;
        }
    }
    Ok(())
}

/// A sentinel that sorts after every real HLC (real wall clocks are far below
/// `0xffff_ffff_ffff_ffff` ms). Used for merge stub rows so any real op's
/// content wins the min-HLC upsert.
const MAX_HLC_SENTINEL: &str = "ffffffffffffffff-ffffffff-";

/// Fold the append-only invalidation records over an edge row: event-time
/// `invalid_at` converges to the earliest of the AddEdge-provided bound
/// (the row value the upsert just wrote) and every recorded invalidation.
/// `expired_at` is deliberately NOT touched — in grit's read model an
/// expired row is retracted from belief entirely, whereas an invalidation
/// only bounds the fact's event time; the belief history of invalidations
/// lives in `edge_invalidations.recorded_at`.
fn fold_edge_invalidations(tx: &Transaction<'_>, edge_id: &str) -> Result<()> {
    let mut stmt = tx.prepare_cached(
        "UPDATE edges SET
             invalid_at = MIN(COALESCE(invalid_at, 9223372036854775807),
                              (SELECT MIN(invalid_at) FROM edge_invalidations
                               WHERE edge_id = ?1))
         WHERE id = ?1
           AND EXISTS (SELECT 1 FROM edge_invalidations WHERE edge_id = ?1)",
    )?;
    stmt.execute(params![edge_id])?;
    Ok(())
}

/// Fold the per-field `node_updates` winners over the node row's AddNode
/// base content. An update wins over base unconditionally (HLC causality:
/// a device can only update a node it has seen, so update HLCs follow the
/// creation HLC); among updates per field, the highest HLC won at insert.
/// No-op when the row hasn't arrived yet — AddNode re-folds on arrival.
fn fold_node_updates(tx: &Transaction<'_>, node_id: &str) -> Result<()> {
    let mut stmt = tx.prepare_cached(
        "UPDATE nodes SET
             name    = COALESCE((SELECT value FROM node_updates
                                 WHERE node_id = ?1 AND field = 'name'), name),
             summary = COALESCE((SELECT value FROM node_updates
                                 WHERE node_id = ?1 AND field = 'summary'), summary),
             kind    = COALESCE((SELECT value FROM node_updates
                                 WHERE node_id = ?1 AND field = 'kind'), kind),
             attrs   = COALESCE((SELECT value FROM node_updates
                                 WHERE node_id = ?1 AND field = 'attrs'), attrs)
         WHERE id = ?1
           AND EXISTS (SELECT 1 FROM node_updates WHERE node_id = ?1)",
    )?;
    stmt.execute(params![node_id])?;
    Ok(())
}

/// Follow `merged_into` pointers to the canonical node id. Concurrent merges
/// can form a cycle (A→B on one device, B→A on another); the canonical member
/// of a cycle is the minimum id — deterministic on every device.
fn resolve_canonical(tx: &Transaction<'_>, id: &str) -> Result<String> {
    let mut chain: Vec<String> = vec![id.to_owned()];
    let mut seen: std::collections::HashSet<String> = chain.iter().cloned().collect();
    loop {
        let current = chain.last().expect("chain never empty");
        // A purged node keeps its merge pointer on the tombstone.
        let mut stmt = tx.prepare_cached(
            "SELECT merged_into FROM nodes WHERE id = ?1
             UNION ALL
             SELECT merged_into FROM purged WHERE id = ?1
             LIMIT 1",
        )?;
        let next: Option<Option<String>> =
            stmt.query_row(params![current], |r| r.get(0)).optional()?;
        match next {
            None | Some(None) => return Ok(chain.pop().expect("chain never empty")),
            Some(Some(next)) => {
                if seen.contains(&next) {
                    let pos = chain
                        .iter()
                        .position(|x| x == &next)
                        .expect("seen implies present");
                    return Ok(chain[pos..]
                        .iter()
                        .min()
                        .expect("cycle slice non-empty")
                        .clone());
                }
                seen.insert(next.clone());
                chain.push(next);
            }
        }
    }
}

/// After a merge lands, re-point every edge/mention endpoint that resolves to
/// `into`'s canonical node directly at it. Runs on every merge apply, so an
/// edge added before the merge (on any device) ends up identical to one added
/// after it.
fn repoint_aliases(tx: &Transaction<'_>, into: &str) -> Result<()> {
    let canon = resolve_canonical(tx, into)?;
    // All ids whose merged_into chain reaches the canonical node (cycle-safe:
    // UNION, not UNION ALL).
    const ALIASES: &str = "WITH RECURSIVE aliases (id) AS (
         SELECT ?1
         UNION
         SELECT n.id FROM nodes AS n JOIN aliases AS a ON n.merged_into = a.id
         UNION
         SELECT p.id FROM purged AS p JOIN aliases AS a ON p.merged_into = a.id
     )";
    tx.execute(
        &format!(
            "{ALIASES}
             UPDATE edges SET src = ?1
             WHERE src IN (SELECT id FROM aliases) AND src != ?1"
        ),
        params![canon],
    )?;
    tx.execute(
        &format!(
            "{ALIASES}
             UPDATE edges SET dst = ?1
             WHERE dst IN (SELECT id FROM aliases) AND dst != ?1"
        ),
        params![canon],
    )?;
    // Mentions are deliberately NOT re-pointed: they keep the original target
    // id (reads expand merge aliases). Re-pointing them here while Purge
    // deletes them by original id would make the final row set depend on
    // whether the merge or the purge applied first.
    Ok(())
}

/// Hard-delete exactly the listed ids and tombstone them forever. No cascade:
/// purging a node does NOT delete its incident edges (a cascade would make
/// Purge order-dependent against concurrent AddEdge ops); Layer 2 lists edge
/// ids explicitly — `node_history` hands them over. Ghost edges whose
/// endpoint was purged are filtered out of every read. The purge op itself
/// remains in the oplog as the audit record.
fn purge(tx: &Transaction<'_>, ids: &[Uuid]) -> Result<()> {
    let has_vec_nodes = table_exists(tx, "vec_nodes")?;
    let has_vec_edges = table_exists(tx, "vec_edges")?;
    for id in ids {
        let id_s = id.to_string();
        // Capture the node's merge pointer (if any) into the tombstone so
        // resolution through this id keeps working after the row is gone.
        tx.execute(
            "INSERT OR IGNORE INTO purged (id, merged_into)
             VALUES (?1, (SELECT merged_into FROM nodes WHERE id = ?1))",
            params![id_s],
        )?;
        tx.execute("DELETE FROM nodes WHERE id = ?1", params![id_s])?;
        tx.execute("DELETE FROM edges WHERE id = ?1", params![id_s])?;
        tx.execute("DELETE FROM episodes WHERE id = ?1", params![id_s])?;
        tx.execute(
            "DELETE FROM edge_invalidations WHERE edge_id = ?1",
            params![id_s],
        )?;
        tx.execute("DELETE FROM node_updates WHERE node_id = ?1", params![id_s])?;
        tx.execute(
            "DELETE FROM mentions WHERE episode_id = ?1 OR target_id = ?1",
            params![id_s],
        )?;
        if has_vec_nodes {
            tx.execute("DELETE FROM vec_nodes WHERE id = ?1", params![id_s])?;
        }
        if has_vec_edges {
            tx.execute("DELETE FROM vec_edges WHERE id = ?1", params![id_s])?;
        }
    }
    Ok(())
}

fn table_exists(tx: &Transaction<'_>, name: &str) -> Result<bool> {
    let n: i64 = tx.query_row(
        "SELECT COUNT(*) FROM sqlite_master WHERE type IN ('table', 'view') AND name = ?1",
        params![name],
        |r| r.get(0),
    )?;
    Ok(n > 0)
}

fn register_model(conn: &mut Connection, model_id: &str, dim: usize, version: &str) -> Result<()> {
    if dim == 0 || dim > 8192 {
        return Err(Error::Embedding(format!("unsupported embedding dim {dim}")));
    }
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let existing: Option<(String, i64, String)> = tx
        .query_row(
            "SELECT model_id, dim, model_version FROM embedding_meta WHERE id = 1",
            [],
            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
        )
        .map(Some)
        .or_else(|e| match e {
            rusqlite::Error::QueryReturnedNoRows => Ok(None),
            other => Err(other),
        })?;
    if let Some((_, existing_dim, _)) = existing {
        if existing_dim != dim as i64 {
            return Err(Error::Embedding(format!(
                "a model with dim {existing_dim} is already registered; \
                 changing dimensions requires a re-embedding flow (not in v0.1)"
            )));
        }
        tx.execute(
            "UPDATE embedding_meta SET model_id = ?1, model_version = ?2 WHERE id = 1",
            params![model_id, version],
        )?;
    } else {
        tx.execute(
            "INSERT INTO embedding_meta (id, model_id, dim, model_version)
             VALUES (1, ?1, ?2, ?3)",
            params![model_id, dim as i64, version],
        )?;
        // Dimension is model-dependent, so vec tables are created here, not in
        // schema.sql. `dim` is validated numeric above; no user strings are
        // interpolated.
        tx.execute_batch(&format!(
            "CREATE VIRTUAL TABLE IF NOT EXISTS vec_nodes USING vec0(
                 id TEXT PRIMARY KEY, embedding FLOAT[{dim}] distance_metric=cosine);
             CREATE VIRTUAL TABLE IF NOT EXISTS vec_edges USING vec0(
                 id TEXT PRIMARY KEY, embedding FLOAT[{dim}] distance_metric=cosine);"
        ))?;
    }
    tx.commit()?;
    Ok(())
}

fn set_embedding(conn: &mut Connection, table: VecTable, id: Uuid, vector: &[f32]) -> Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    // Right-to-forget: a vector is derived from its row's content, so an
    // embed racing a purge must not re-materialize purged data. No-op, like
    // every other write against a tombstone.
    let is_purged: i64 = tx.query_row(
        "SELECT COUNT(*) FROM purged WHERE id = ?1",
        params![id.to_string()],
        |r| r.get(0),
    )?;
    if is_purged > 0 {
        return Ok(());
    }
    let dim: i64 = tx
        .query_row("SELECT dim FROM embedding_meta WHERE id = 1", [], |r| {
            r.get(0)
        })
        .map_err(|_| Error::Embedding("no embedding model registered".into()))?;
    if vector.len() as i64 != dim {
        return Err(Error::Embedding(format!(
            "vector has dim {}, registered model has dim {dim}",
            vector.len()
        )));
    }
    // vec0 has no upsert; delete + insert.
    tx.execute(
        &format!("DELETE FROM {} WHERE id = ?1", table.name()),
        params![id.to_string()],
    )?;
    tx.execute(
        &format!(
            "INSERT INTO {} (id, embedding) VALUES (?1, ?2)",
            table.name()
        ),
        params![id.to_string(), f32s_as_bytes(vector)],
    )?;
    tx.commit()?;
    Ok(())
}