rindle_wire/

normalize_protocol.rs

//! The **normalized subscription protocol** (`NORMALIZED-CHANGES-DESIGN.md` §3) — the
//! path-free twin of the engine's `flat_protocol`. The envelope semantics are identical
//! (a once-per-subscription [`NormalizedHello`], a logical-seq-0 snapshot, then
//! gap-free incremental [`NormalizedBatch`]es at seq 1, 2, …); only the **payload**
//! differs — table-tagged [`NormalizedOp`]s instead of path-tagged `FlatChange`s — and
//! the **hello is slimmer**: flat per-table schemas (for positional-row alignment + a
//! `normalized_fp`), not a nested hierarchical view schema (§3).
//!
//! Why a sibling here rather than a generic reuse of `flat_protocol`: that module's
//! `Publisher`/`Subscriber`/`Batch` are concrete over `FlatChange` + a hierarchical view
//! `Schema`, and `flat_protocol` lives in the wasm-clean `rindle` core; normalize lives
//! beside this module in `rindle-wire` (also wasm-clean, but the protocol crate, not
//! the engine — see the placement note in [`crate::normalize`]). So this re-implements
//! the *small* epoch/seq/gap machinery over the normalized payload rather than
//! generic-ifying core. The contract it enforces is the flat one:
//!
//! - **Sender** ([`NormalizedPublisher`]) wraps a [`NormalizeFold`], drives the seq
//!   counter, and stamps every frame with the subscription `epoch` + `normalized_fp`.
//!   An **empty transaction emits no batch and consumes no seq** (so seq stays gap-free
//!   over *emitted* batches).
//! - **Receiver** ([`NormalizedSubscriber`], the validation half — the fold itself is
//!   the TS `NormalizedSync`, Slice 4) enforces the comparator contract at the hello and,
//!   per batch, epoch + fingerprint + strict in-order seq. A **duplicate** seq is
//!   discarded idempotently (`rc` add/remove are not idempotent). A **gap** is fatal: the
//!   only repair is a full re-hydrate under a **new epoch** (§5.3), after which stale
//!   old-epoch frames are rejected by [`NormalizedProtocolError::EpochMismatch`].
//!
//! Snapshot **chunking** (the `flat_protocol` `SnapshotChunk` path) is deferred: this
//! slice ships the single-shot seq-0 snapshot. Chunking is mechanical to add later (the
//! design §3 notes it as a capability, not a v1 requirement).

use std::collections::{BTreeMap, BTreeSet};

use rindle::{Ast, CaughtChange, WireRow, COMPARATOR_VERSION};

use crate::normalize::{
    required_columns_by_table, table_tree, NormalizeFold, NormalizedOp, PkMap, ReqCols, TableNode,
};

// ---------------------------------------------------------------------------
// Wire types (§3)
// ---------------------------------------------------------------------------

/// One base table's flat schema on the wire: its ordered column **names** (wire rows are
/// positional against them) and the primary-key column **indices**. The client validates
/// these against its own typed schema and registers any table it hasn't yet (e.g. an
/// EXISTS witness table). No `sort` — base tables sort by PK in the memory source, and the
/// result sort is the client's local engine's concern (§3).
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TableWireSchema {
    pub name: Box<str>,
    pub columns: Vec<Box<str>>,
    pub primary_key: Vec<u32>,
}

/// The subscription handshake (§3), sent once before any [`NormalizedBatch`]. Slimmer than
/// the flat `Hello`: flat per-table schemas, no nested view schema or per-level sort.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct NormalizedHello {
    pub epoch: u64,
    pub comparator_version: u32,
    /// The base tables this subscription's footprint can surface (the query's table tree),
    /// in a stable order (sorted by name) so the wire form is deterministic.
    pub tables: Vec<TableWireSchema>,
    /// FNV-1a-64 fingerprint over `{ table → (ordered columns, PK) }` by **name** — a
    /// sibling of the engine's `schema_fp`. Drift ⇒ re-subscribe.
    pub normalized_fp: u64,
}

/// One committed transaction's normalized ops (or the seq-0 hydrate snapshot). Ops apply
/// in order into one client `Db.write()` transaction.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct NormalizedBatch {
    pub epoch: u64,
    pub seq: u64,
    /// The global commit version this frame's data reflects (`OPTIMISTIC-WRITES-DESIGN.md`
    /// §8.3/§8.6) — the replica's `TxId` for the commit (the hydrate watermark for the
    /// seq-0 snapshot). Data frames ship eagerly per query; an optimistic client buffers
    /// them by `cv` and applies all `cv ≤ cv_min` as one coherent transaction (§8.5).
    /// Non-optimistic consumers may ignore it.
    pub cv: u64,
    pub normalized_fp: u64,
    pub ops: Vec<NormalizedOp>,
}

/// The first `Int` in this batch's rows outside `Number.MAX_SAFE_INTEGER`, if any —
/// the 09.8 `strict_i64` walk for the normalized wire (design 226 Stage A). A JS-facing
/// boundary with strict mode on refuses the batch with a typed error instead of letting
/// the cell encode round it (`wire_json`'s `Int → f64` collapse).
pub fn unsafe_int_in_normalized_batch(batch: &NormalizedBatch) -> Option<i64> {
    use rindle::js_safe::unsafe_int_in_cells;
    batch.ops.iter().find_map(|op| match op {
        NormalizedOp::Add { row, .. } | NormalizedOp::Remove { row, .. } => {
            unsafe_int_in_cells(row)
        }
        NormalizedOp::Edit { old, new, .. } => {
            unsafe_int_in_cells(old).or_else(|| unsafe_int_in_cells(new))
        }
    })
}

/// The connection-level progress frame (§8.6): advances the client's coherent release
/// point (`cv_min`). Pure release signal — mutation confirmation does NOT ride it:
/// `lmid` is a row in the client-mutations table (`_rindle_client_mutations`,
/// `rindle-replica`'s `CLIENT_MUTATIONS_TABLE`), delivered through the client's
/// own per-client system query like any other data, so it is released by the same
/// `cv_min` that releases the commit's effects (transactionally coherent by
/// construction). Emitted per the poke rule (§8.4); standalone only to advance
/// `cv_min` during a quiet window.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ProgressFrame {
    /// The commit version every one of this connection's live queries has processed
    /// through — the client's coherent-apply release point (§8.3).
    pub cv_min: u64,
}

/// A protocol violation the [`NormalizedSubscriber`] surfaces. All but a duplicate are
/// fatal — the only repair is a re-hydrate under a new epoch (§5.3).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum NormalizedProtocolError {
    /// The hello's table set is not a valid normalized wire schema. Validation happens
    /// before fingerprinting so an attacker-controlled primary-key index can never reach
    /// `columns[pk]` and panic.
    InvalidSchema(NormalizedSchemaError),
    /// The hello's comparator-contract version differs — reconstruction would silently
    /// corrupt, so the subscription is refused (a code contract, not data).
    ComparatorMismatch { expected: u32, got: u32 },
    /// The advertised / batch fingerprint differs from the subscribed schema.
    FpMismatch { expected: u64, got: u64 },
    /// A batch from a different (usually stale) subscription epoch.
    EpochMismatch { expected: u64, got: u64 },
    /// A seq beyond the next expected ⇒ a batch was lost. Re-hydrate. (`expected` is the
    /// next seq the receiver wanted.)
    Gap { expected: u64, got: u64 },
}

/// A malformed table set advertised by a [`NormalizedHello`]. `table_index` identifies
/// the offending entry in [`NormalizedHello::tables`]; the table name itself stays in the
/// hello so this error remains small and `Copy` while callers can still render a precise
/// diagnostic.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct NormalizedSchemaError {
    pub table_index: usize,
    pub kind: NormalizedSchemaErrorKind,
}

/// The semantic wire-schema rules required by [`normalized_fp`] and by consumers that
/// register the advertised tables as keyed sources.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum NormalizedSchemaErrorKind {
    EmptyPrimaryKey,
    PrimaryKeyOutOfBounds,
    TablesNotSortedByUniqueName,
}

impl NormalizedSchemaErrorKind {
    /// Stable human-readable reason used by room diagnostics.
    pub fn reason(self) -> &'static str {
        match self {
            Self::EmptyPrimaryKey => "empty primary key",
            Self::PrimaryKeyOutOfBounds => "primary-key index out of bounds",
            Self::TablesNotSortedByUniqueName => "tables not sorted by unique name",
        }
    }
}

/// The outcome of applying a [`NormalizedBatch`].
#[derive(Clone, Debug)]
pub enum NormalizedApplied {
    /// A newly-applied batch: apply these ops to the base store (in order).
    Live(Vec<NormalizedOp>),
    /// An already-applied seq (re-delivery) — discard idempotently, base store unchanged.
    Duplicate,
}

// ---------------------------------------------------------------------------
// normalized_fp — the by-name content fingerprint (§3)
// ---------------------------------------------------------------------------

/// Validate the table-set invariants required by [`normalized_fp`] and normalized-store
/// construction. This accepts wire-deserialized input, so every failure is typed and no
/// malformed schema may panic.
pub fn validate_normalized_schema(tables: &[TableWireSchema]) -> Result<(), NormalizedSchemaError> {
    for (table_index, table) in tables.iter().enumerate() {
        let error = |kind| NormalizedSchemaError { table_index, kind };
        if table.primary_key.is_empty() {
            return Err(error(NormalizedSchemaErrorKind::EmptyPrimaryKey));
        }
        if table
            .primary_key
            .iter()
            .any(|&pk| pk as usize >= table.columns.len())
        {
            return Err(error(NormalizedSchemaErrorKind::PrimaryKeyOutOfBounds));
        }
        if table_index > 0 && tables[table_index - 1].name >= table.name {
            return Err(error(
                NormalizedSchemaErrorKind::TablesNotSortedByUniqueName,
            ));
        }
    }
    Ok(())
}

/// FNV-1a-64 accumulator. The exact length-prefixed byte protocol is the wire contract —
/// any peer must hash identically. (A local twin of the core `wire_schema::Fnv`, which is
/// private; the byte protocol below is normalized-specific so it does not need to match
/// `schema_fp`.)
struct Fnv(u64);
impl Fnv {
    fn new() -> Fnv {
        Fnv(0xcbf29ce4_84222325) // FNV-1a 64 offset basis
    }
    fn byte(&mut self, b: u8) {
        self.0 ^= b as u64;
        self.0 = self.0.wrapping_mul(0x0000_0100_0000_01b3); // FNV-1a 64 prime
    }
    fn u32(&mut self, v: u32) {
        for b in v.to_le_bytes() {
            self.byte(b);
        }
    }
    /// Length-prefixed string — disambiguates concatenations (`"ab"+"c"` ≠ `"a"+"bc"`).
    fn s(&mut self, s: &str) {
        self.u32(s.len() as u32);
        for &b in s.as_bytes() {
            self.byte(b);
        }
    }
}

/// Fingerprint a `tables` set by **name** (resolving PK to column names), independent of
/// internal column numbering. `tables` must satisfy [`validate_normalized_schema`]; the
/// publisher constructors guarantee that invariant, and [`NormalizedSubscriber::open`]
/// validates received schemas before calling this function.
pub fn normalized_fp(tables: &[TableWireSchema]) -> u64 {
    let mut h = Fnv::new();
    h.u32(tables.len() as u32);
    for t in tables {
        h.s(&t.name);
        h.u32(t.columns.len() as u32);
        for c in &t.columns {
            h.s(c);
        }
        h.u32(t.primary_key.len() as u32);
        for &pk in &t.primary_key {
            // Resolve PK to the column NAME so the identity is numbering-independent.
            h.s(&t.columns[pk as usize]);
        }
    }
    h.0
}

// ---------------------------------------------------------------------------
// Publisher (sender)
// ---------------------------------------------------------------------------

/// Sender side: wraps a [`NormalizeFold`], stamps batches with the subscription `epoch` +
/// `normalized_fp`, and drives the gap-free seq. The caller drains the change-sink (the
/// replica's per-query `CaughtChange`s) and hands them to [`snapshot`](Self::snapshot) /
/// [`commit`](Self::commit).
pub struct NormalizedPublisher {
    epoch: u64,
    fold: NormalizeFold,
    hello: NormalizedHello,
    next_seq: u64,
    /// Project-at-emit map (`PROJECTION-SUPPORT-DESIGN.md` §5.2): table name → the base
    /// column indices this query syncs for it, in ascending (= projected wire) order. Only
    /// **projected** tables appear; a table absent here syncs full. Empty ⇒ no projection,
    /// so emission is a pass-through (a `'*'` query is byte-identical, §7).
    proj: BTreeMap<Box<str>, Vec<usize>>,
}

/// Project-at-emit map (`PROJECTION-SUPPORT-DESIGN.md` §5.2): table name → the base column
/// indices this query syncs for it, in ascending (= projected wire) order. Only **projected**
/// tables appear; a table absent here syncs full. An empty map ⇒ no projection, so emission is
/// a pass-through (a `'*'` query is byte-identical, §7).
pub type ProjMap = BTreeMap<Box<str>, Vec<usize>>;

/// The query-wide pieces a normalized subscription needs, derived once from the AST + the
/// surfaced-table schemas: the [`NormalizeFold`] (the shared footprint serializer over the
/// pipeline's **full** rows), the **projected** wire table set and its [`normalized_fp`], and
/// the [`ProjMap`] that drives project-at-emit. Splitting this out lets a **shared** engine
/// query own ONE fold + projection while each subscriber builds its own [`NormalizedHello`]
/// from the same projected table set ([`hello_from_parts`]) — fold, projection, and hello agree
/// by construction.
///
/// Panics if a surfaced table lacks a schema — a build-time config invariant (mirrors
/// [`NormalizeFold`]'s `validate_pk`).
pub fn build_query_parts(
    ast: &Ast,
    schemas: Vec<TableWireSchema>,
) -> (NormalizeFold, Vec<TableWireSchema>, u64, ProjMap) {
    // Every table the tree surfaces must have a schema.
    let mut needed = BTreeSet::new();
    collect_tables(&table_tree(ast), &mut needed);
    for t in &needed {
        assert!(
            schemas.iter().any(|s| &s.name == t),
            "normalized query: no schema supplied for table {t:?}"
        );
    }

    // Sort tables by name for a deterministic wire form + fingerprint.
    let mut tables = schemas;
    tables.sort_by(|a, b| a.name.cmp(&b.name));

    // Derive the fold's PK map (table → PK column indices) from the FULL schemas: the fold runs
    // over the pipeline's full rows; projection is applied AFTER it, at emit (§2.1).
    let pk: PkMap = tables
        .iter()
        .map(|t| {
            (
                t.name.clone(),
                t.primary_key.iter().map(|&i| i as usize).collect(),
            )
        })
        .collect();
    let fold = NormalizeFold::new(ast, pk);

    // Project each surfaced table to the columns this query needs (§5.2). A table whose required
    // set covers every column (or is `'*'`) stays full; otherwise its wire schema declares the
    // projected columns (in ascending base-index order) with the PK remapped, and the base
    // indices are recorded in `proj` for project-at-emit. The wire fingerprint covers the
    // PROJECTED schema so the subscriber validates against exactly what it receives.
    let required = required_columns_by_table(ast);
    let mut proj: ProjMap = BTreeMap::new();
    let wire_tables: Vec<TableWireSchema> = tables
        .iter()
        .map(|t| match project_columns(t, required.get(&t.name)) {
            Some(idxs) => {
                let wire = TableWireSchema {
                    name: t.name.clone(),
                    columns: idxs.iter().map(|&i| t.columns[i].clone()).collect(),
                    primary_key: t
                        .primary_key
                        .iter()
                        .map(|&pk| {
                            idxs.iter()
                                .position(|&i| i == pk as usize)
                                .expect("PK is forced into the projection")
                                as u32
                        })
                        .collect(),
                };
                proj.insert(t.name.clone(), idxs);
                wire
            }
            None => t.clone(),
        })
        .collect();
    let fp = normalized_fp(&wire_tables);

    (fold, wire_tables, fp, proj)
}

/// The per-subscriber handshake for a (possibly shared) normalized query at `epoch`, from
/// the query's deterministic table set + [`normalized_fp`] (see [`build_query_parts`]).
pub fn hello_from_parts(
    epoch: u64,
    tables: Vec<TableWireSchema>,
    normalized_fp: u64,
) -> NormalizedHello {
    NormalizedHello {
        epoch,
        comparator_version: COMPARATOR_VERSION,
        tables,
        normalized_fp,
    }
}

impl NormalizedPublisher {
    /// Open a publisher at `epoch` for `ast`, given the flat schema of **every** table the
    /// query's tree can surface (root + every `related`/EXISTS child table). Bump `epoch`
    /// on a re-hydrate (§5.3). Panics if a surfaced table lacks a schema — a build-time
    /// config invariant (mirrors `NormalizeFold`'s `validate_pk`).
    pub fn new(epoch: u64, ast: &Ast, schemas: Vec<TableWireSchema>) -> NormalizedPublisher {
        // The fold (over full rows), the projected wire table set + fingerprint, and the
        // project-at-emit map all come from one place (§5.2) — the same primitive the shared
        // engine-query path uses, so a `NormalizedPublisher` and a shared query project identically.
        let (fold, wire_tables, fp, proj) = build_query_parts(ast, schemas);
        NormalizedPublisher {
            epoch,
            fold,
            hello: hello_from_parts(epoch, wire_tables, fp),
            next_seq: 0,
            proj,
        }
    }

    /// The handshake to send before any batch.
    pub fn hello(&self) -> &NormalizedHello {
        &self.hello
    }

    /// The subscription epoch.
    pub fn epoch(&self) -> u64 {
        self.epoch
    }

    /// The hydrate snapshot (the pipeline's `CaughtChange::Add`s from
    /// `Graph::hydrate_change_sink`) folded into a single seq-0 batch of `add` ops, at
    /// commit version `cv` (the watermark the hydrate reflects — `Update::Hydrated`'s
    /// `tx_id`). Always emitted — even for an empty result — so the receiver learns the
    /// baseline is set. Reserves the seq-0 slot; increments start at seq 1. Call once,
    /// before any `commit`.
    pub fn snapshot(&mut self, caught: &[CaughtChange], cv: u64) -> NormalizedBatch {
        let ops = project_ops(self.fold.fold(caught), &self.proj);
        self.next_seq = 1;
        NormalizedBatch {
            epoch: self.epoch,
            seq: 0,
            cv,
            normalized_fp: self.hello.normalized_fp,
            ops,
        }
    }

    /// Wrap one transaction's drained changes, stamped with the transaction's commit
    /// version `cv` (`Update::Changed`'s `tx_id`). Folds them through the persistent
    /// footprint and emits the net membership deltas. Returns `None` for a transaction
    /// that produced **no** net op — no batch, no seq consumed — keeping seq gap-free
    /// over emitted batches (an empty source tx, or one whose effects cancel within the
    /// footprint).
    pub fn commit(&mut self, caught: &[CaughtChange], cv: u64) -> Option<NormalizedBatch> {
        let ops = project_ops(self.fold.fold(caught), &self.proj);
        let seq = take_emitted_seq(&mut self.next_seq, !ops.is_empty())?;
        Some(NormalizedBatch {
            epoch: self.epoch,
            seq,
            cv,
            normalized_fp: self.hello.normalized_fp,
            ops,
        })
    }
}

/// Reserve the next sequence only when a fold produced an emitted batch. Kept as a pure
/// transition both to make the gap-free rule explicit and to verify it without pulling
/// `NormalizeFold`'s randomized `HashMap` state into the model checker.
pub(crate) fn take_emitted_seq(next_seq: &mut u64, has_ops: bool) -> Option<u64> {
    if !has_ops {
        return None;
    }
    let seq = *next_seq;
    *next_seq += 1;
    Some(seq)
}

/// Every table `node`'s subtree can surface (`pub`: `rindle-room-core`'s downstream
/// materialization walks the same set to assemble its schemas).
pub fn collect_tables(node: &TableNode, out: &mut BTreeSet<Box<str>>) {
    out.insert(node.table.clone());
    // Pruned (`exists_noSync`) slots are `None`: their permission tables are never synced.
    for c in node.children.iter().flatten() {
        collect_tables(c, out);
    }
}

/// The base column indices to sync for `t`, in ascending order, or `None` if the table syncs
/// full (its required set is `All`, absent, or already covers every column). The PK is always
/// forced present — the client keys rows by it (PROJECTION-SUPPORT-DESIGN.md §4.1).
fn project_columns(t: &TableWireSchema, req: Option<&ReqCols>) -> Option<Vec<usize>> {
    let names = match req {
        Some(ReqCols::Names(names)) => names,
        _ => return None, // `All` / missing ⇒ full table, no projection
    };
    let mut idxs: Vec<usize> = (0..t.columns.len())
        .filter(|&i| names.contains(&t.columns[i]))
        .collect();
    for &pk in &t.primary_key {
        let pk = pk as usize;
        if !idxs.contains(&pk) {
            idxs.push(pk);
        }
    }
    idxs.sort_unstable();
    // A projection that covers every column is just the full table.
    if idxs.len() == t.columns.len() {
        None
    } else {
        Some(idxs)
    }
}

/// Project every op's rows to the per-table column set (§5.2). A no-op when `proj` is empty.
/// `pub` (not `pub(crate)`): `rindle-replica`'s `Drain` applies the same projection on its
/// own fanout path, and the room's downstream publisher will too.
pub fn project_ops(ops: Vec<NormalizedOp>, proj: &ProjMap) -> Vec<NormalizedOp> {
    if proj.is_empty() {
        return ops;
    }
    ops.into_iter().map(|op| project_op(op, proj)).collect()
}

/// Project one op's row(s) positional over the table's projected columns; a table absent from
/// `proj` passes through full.
fn project_op(op: NormalizedOp, proj: &BTreeMap<Box<str>, Vec<usize>>) -> NormalizedOp {
    fn pick(row: &WireRow, cols: &[usize]) -> WireRow {
        cols.iter().map(|&c| row[c].clone()).collect()
    }
    match op {
        NormalizedOp::Add { table, row } => {
            let row = match proj.get(&table) {
                Some(cols) => pick(&row, cols),
                None => row,
            };
            NormalizedOp::Add { table, row }
        }
        NormalizedOp::Remove { table, row } => {
            let row = match proj.get(&table) {
                Some(cols) => pick(&row, cols),
                None => row,
            };
            NormalizedOp::Remove { table, row }
        }
        NormalizedOp::Edit { table, old, new } => match proj.get(&table) {
            Some(cols) => NormalizedOp::Edit {
                old: pick(&old, cols),
                new: pick(&new, cols),
                table,
            },
            None => NormalizedOp::Edit { table, old, new },
        },
    }
}

// ---------------------------------------------------------------------------
// Subscriber (receiver — the validation half)
// ---------------------------------------------------------------------------

/// The subscriber's lifecycle: establish the seq-0 baseline, then accept live increments.
#[derive(Clone, Copy, Debug)]
enum Phase {
    /// Awaiting the seq-0 snapshot batch.
    Snapshot,
    /// Baseline established; only incremental batches (seq ≥ 1) are accepted.
    Live { last_seq: u64 },
}

/// Receiver side: the protocol state machine that **validates** the envelope and emits
/// clean ops for the caller (the base-store fold / TS `NormalizedSync`) to apply. It does
/// not itself hold the base store — it owns only epoch/seq/fp state.
#[derive(Debug)]
pub struct NormalizedSubscriber {
    epoch: u64,
    normalized_fp: u64,
    phase: Phase,
}

impl NormalizedSubscriber {
    /// Open from a [`NormalizedHello`]: validate the wire-constructed table set before
    /// fingerprinting it, hard-reject a comparator-contract mismatch (§5.5), and verify
    /// the advertised fingerprint matches the shipped table set. Apply the seq-0 snapshot
    /// batch next.
    pub fn open(hello: &NormalizedHello) -> Result<NormalizedSubscriber, NormalizedProtocolError> {
        validate_normalized_schema(&hello.tables)
            .map_err(NormalizedProtocolError::InvalidSchema)?;
        if hello.comparator_version != COMPARATOR_VERSION {
            return Err(NormalizedProtocolError::ComparatorMismatch {
                expected: COMPARATOR_VERSION,
                got: hello.comparator_version,
            });
        }
        let computed = normalized_fp(&hello.tables);
        if computed != hello.normalized_fp {
            return Err(NormalizedProtocolError::FpMismatch {
                expected: hello.normalized_fp,
                got: computed,
            });
        }
        Ok(NormalizedSubscriber {
            epoch: hello.epoch,
            normalized_fp: hello.normalized_fp,
            phase: Phase::Snapshot,
        })
    }

    fn check_frame(&self, epoch: u64, fp: u64) -> Result<(), NormalizedProtocolError> {
        if epoch != self.epoch {
            return Err(NormalizedProtocolError::EpochMismatch {
                expected: self.epoch,
                got: epoch,
            });
        }
        if fp != self.normalized_fp {
            return Err(NormalizedProtocolError::FpMismatch {
                expected: self.normalized_fp,
                got: fp,
            });
        }
        Ok(())
    }

    /// Apply one batch (the seq-0 snapshot or an incremental batch). Validates epoch +
    /// fingerprint + strict in-order seq, returning the ops to apply on success. A
    /// duplicate (already-applied seq) is [`NormalizedApplied::Duplicate`] (no ops); a gap
    /// is a fatal [`NormalizedProtocolError::Gap`] (re-hydrate under a new epoch).
    pub fn apply(
        &mut self,
        batch: &NormalizedBatch,
    ) -> Result<NormalizedApplied, NormalizedProtocolError> {
        self.check_frame(batch.epoch, batch.normalized_fp)?;
        match &mut self.phase {
            Phase::Snapshot => {
                if batch.seq != 0 {
                    // The baseline must come first; a seq ≥ 1 before it means seq 0 was lost.
                    return Err(NormalizedProtocolError::Gap {
                        expected: 0,
                        got: batch.seq,
                    });
                }
                self.phase = Phase::Live { last_seq: 0 };
                Ok(NormalizedApplied::Live(batch.ops.clone()))
            }
            Phase::Live { last_seq } => {
                if batch.seq <= *last_seq {
                    return Ok(NormalizedApplied::Duplicate);
                }
                if batch.seq != *last_seq + 1 {
                    return Err(NormalizedProtocolError::Gap {
                        expected: *last_seq + 1,
                        got: batch.seq,
                    });
                }
                *last_seq = batch.seq;
                Ok(NormalizedApplied::Live(batch.ops.clone()))
            }
        }
    }
}

#[cfg(test)]
mod projection_tests {
    use super::*;
    use rindle::value::{owned_row, OwnedValue as V};
    use rindle::{table, CaughtChange, CaughtNode};

    fn tbl(name: &str, cols: &[&str], pk: &[u32]) -> TableWireSchema {
        TableWireSchema {
            name: name.into(),
            columns: cols.iter().map(|c| (*c).into()).collect(),
            primary_key: pk.to_vec(),
        }
    }

    fn leaf(row: Vec<V>) -> CaughtNode {
        CaughtNode {
            row: owned_row(row),
            relationships: Default::default(),
        }
    }

    fn ints(row: &WireRow) -> Vec<i64> {
        row.iter()
            .map(|v| match v {
                V::Int(i) => *i,
                other => panic!("expected Int, got {other:?}"),
            })
            .collect()
    }

    // issue(id, title, priority); PK id.
    fn issue_schema() -> Vec<TableWireSchema> {
        vec![tbl("issue", &["id", "title", "priority"], &[0])]
    }

    #[test]
    fn projected_hello_declares_only_required_columns_with_remapped_pk() {
        // select priority, order by id → required {id, priority}; title is dropped. Projected
        // wire columns are in ascending base-index order: [id, priority]; PK remaps to 0.
        let ast = table("issue")
            .select("priority")
            .order_by("id", "asc")
            .build();
        let pubr = NormalizedPublisher::new(1, &ast, issue_schema());
        let hello = pubr.hello();
        assert_eq!(hello.tables.len(), 1);
        let t = &hello.tables[0];
        assert_eq!(&*t.name, "issue");
        assert_eq!(
            t.columns.iter().map(|c| &**c).collect::<Vec<_>>(),
            vec!["id", "priority"]
        );
        assert_eq!(t.primary_key, vec![0]);
    }

    #[test]
    fn projected_snapshot_rows_drop_unselected_cells() {
        let ast = table("issue")
            .select("priority")
            .order_by("id", "asc")
            .build();
        let mut pubr = NormalizedPublisher::new(1, &ast, issue_schema());
        // The pipeline emits a FULL row; project-at-emit drops `title` (index 1).
        let batch = pubr.snapshot(
            &[CaughtChange::Add(leaf(vec![
                V::Int(1),
                V::str("a"),
                V::Int(5),
            ]))],
            0,
        );
        assert_eq!(batch.ops.len(), 1);
        match &batch.ops[0] {
            NormalizedOp::Add { table, row } => {
                assert_eq!(&**table, "issue");
                assert_eq!(ints(row), vec![1, 5]); // [id, priority] — title gone
            }
            other => panic!("expected Add, got {other:?}"),
        }
    }

    #[test]
    fn projected_edit_rows_are_projected_on_both_sides() {
        let ast = table("issue")
            .select("priority")
            .order_by("id", "asc")
            .build();
        let mut pubr = NormalizedPublisher::new(1, &ast, issue_schema());
        // Seed (snapshot), then edit priority 5→9 — both old/new project to [id, priority].
        pubr.snapshot(
            &[CaughtChange::Add(leaf(vec![
                V::Int(1),
                V::str("a"),
                V::Int(5),
            ]))],
            0,
        );
        let batch = pubr
            .commit(
                &[CaughtChange::Edit {
                    old: owned_row(vec![V::Int(1), V::str("a"), V::Int(5)]),
                    row: owned_row(vec![V::Int(1), V::str("a"), V::Int(9)]),
                }],
                1,
            )
            .expect("non-empty commit");
        match &batch.ops[0] {
            NormalizedOp::Edit { table, old, new } => {
                assert_eq!(&**table, "issue");
                assert_eq!(ints(old), vec![1, 5]);
                assert_eq!(ints(new), vec![1, 9]);
            }
            other => panic!("expected Edit, got {other:?}"),
        }
    }

    #[test]
    fn empty_publisher_commit_consumes_no_seq() {
        let ast = table("issue").order_by("id", "asc").build();
        let mut pubr = NormalizedPublisher::new(1, &ast, issue_schema());
        assert_eq!(pubr.snapshot(&[], 0).seq, 0);
        assert!(pubr.commit(&[], 1).is_none());
        assert!(pubr.commit(&[], 2).is_none());

        let emitted = pubr
            .commit(
                &[CaughtChange::Add(leaf(vec![
                    V::Int(1),
                    V::str("a"),
                    V::Int(5),
                ]))],
                3,
            )
            .expect("non-empty commit");
        assert_eq!(emitted.seq, 1);
    }

    #[test]
    fn star_query_is_unprojected_and_byte_identical() {
        let ast = table("issue").order_by("id", "asc").build();
        let pubr = NormalizedPublisher::new(1, &ast, issue_schema());
        let t = &pubr.hello().tables[0];
        assert_eq!(
            t.columns.iter().map(|c| &**c).collect::<Vec<_>>(),
            vec!["id", "title", "priority"]
        );
        assert_eq!(t.primary_key, vec![0]);
        let mut pubr = pubr;
        let batch = pubr.snapshot(
            &[CaughtChange::Add(leaf(vec![
                V::Int(1),
                V::str("a"),
                V::Int(5),
            ]))],
            0,
        );
        match &batch.ops[0] {
            NormalizedOp::Add { row, .. } => assert_eq!(row.len(), 3), // full row kept
            other => panic!("expected Add, got {other:?}"),
        }
    }

    #[test]
    fn required_columns_force_in_filter_and_sort_not_just_select() {
        // select title; where priority > 2; order by id → required {title, priority, id}.
        let ast = table("issue")
            .select("title")
            .where_op("priority", ">", 2)
            .order_by("id", "asc")
            .build();
        let req = required_columns_by_table(&ast);
        match req.get("issue").expect("issue required") {
            ReqCols::Names(names) => {
                let mut got: Vec<&str> = names.iter().map(|n| &**n).collect();
                got.sort_unstable();
                assert_eq!(got, vec!["id", "priority", "title"]);
            }
            ReqCols::All => panic!("expected a projected (Names) requirement"),
        }
        // And the wire schema keeps all three (so a projection that needs every column is full).
        let pubr = NormalizedPublisher::new(1, &ast, issue_schema());
        assert_eq!(pubr.hello().tables[0].columns.len(), 3);
    }

    #[test]
    fn materialized_relationship_correlation_forces_unselected_parent_column() {
        // select id; .sub_as(watchers: watch where watch.issuePriority = issue.priority). The
        // relationship is MATERIALIZED (a slot) — its child IS synced and the client re-runs the
        // join — so the parent correlation key `priority` MUST be forced into issue's required
        // set even though the user only selected `id`. Otherwise the client would receive the
        // child rows but lack the parent key to join them (PROJECTION-SUPPORT-DESIGN.md §3.3).
        // This is the projection × relationship interaction the rebase first fused; the existing
        // `parity_projection_keeps_relationships` correlates on `id` (already selected), so it
        // never exercises the force-an-UNSELECTED-column path (`collect_required` slot walk).
        let ast = table("issue")
            .select("id")
            .sub_as("watchers", |row| {
                table("watch").r#where("issuePriority", row.col("priority"))
            })
            .order_by("id", "asc")
            .build();
        let req = required_columns_by_table(&ast);
        match req.get("issue").expect("issue required") {
            ReqCols::Names(names) => {
                assert!(
                    names.iter().any(|n| &**n == "priority"),
                    "the relationship correlation column `priority` must be forced in despite \
                     `select id`, got {names:?}"
                );
                assert!(
                    names.iter().any(|n| &**n == "id"),
                    "the selected pk `id` is present, got {names:?}"
                );
            }
            ReqCols::All => panic!("expected a projected (Names) requirement, not All"),
        }
        // The child `watch` is synced too (it has a slot), so it appears in the required map.
        assert!(
            req.contains_key("watch"),
            "the materialized child is synced: {req:?}"
        );
    }

    #[test]
    fn nonmaterialized_exists_is_server_side_and_syncs_no_child() {
        // A bare top-level `whereExists` (semi-join, NOT flipped/materialized) is evaluated
        // entirely server-side: it produces no query-local slot, its child table is NOT synced,
        // and the client trusts the server-filtered parent membership without re-evaluating it.
        // Therefore the EXISTS correlation column is intentionally NOT forced into the projected
        // wire — `priority` here stays dropped. This characterizes WHY the correlation column is
        // absent (a future change must not "fix" it by force-syncing an unsynced child's key).
        let ast = table("issue")
            .select("id")
            .where_exists(|row| table("comment").r#where("issueID", row.col("priority")))
            .order_by("id", "asc")
            .build();
        assert!(
            rindle::query_local_slot_names(&ast).is_empty(),
            "a non-materialized EXISTS creates no slot"
        );
        let mut needed = BTreeSet::new();
        collect_tables(&table_tree(&ast), &mut needed);
        assert_eq!(
            needed.iter().map(|t| &**t).collect::<Vec<_>>(),
            vec!["issue"],
            "only the root table syncs; the EXISTS child is server-side-only"
        );
        let req = required_columns_by_table(&ast);
        match req.get("issue").expect("issue required") {
            ReqCols::Names(names) => assert!(
                !names.iter().any(|n| &**n == "priority"),
                "a server-side EXISTS does not force its correlation column into the wire: {names:?}"
            ),
            ReqCols::All => panic!("expected a projected (Names) requirement, not All"),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tbl(name: &str, cols: &[&str], pk: &[u32]) -> TableWireSchema {
        TableWireSchema {
            name: name.into(),
            columns: cols.iter().map(|c| (*c).into()).collect(),
            primary_key: pk.to_vec(),
        }
    }

    #[test]
    fn fp_is_deterministic_and_order_independent_after_sort() {
        let a = vec![
            tbl("issues", &["id", "title"], &[0]),
            tbl("comments", &["id", "issue_id", "body"], &[0]),
        ];
        let mut b = a.clone();
        b.reverse();
        // Sort both (as the publisher does) → identical fingerprint.
        let mut a_sorted = a.clone();
        a_sorted.sort_by(|x, y| x.name.cmp(&y.name));
        let mut b_sorted = b;
        b_sorted.sort_by(|x, y| x.name.cmp(&y.name));
        assert_eq!(normalized_fp(&a_sorted), normalized_fp(&b_sorted));
    }

    #[test]
    fn fp_detects_drift() {
        let base = vec![tbl("issues", &["id", "title"], &[0])];
        let base_fp = normalized_fp(&base);
        // Column rename.
        assert_ne!(
            normalized_fp(&[tbl("issues", &["id", "name"], &[0])]),
            base_fp
        );
        // PK change.
        assert_ne!(
            normalized_fp(&[tbl("issues", &["id", "title"], &[1])]),
            base_fp
        );
        // Added column.
        assert_ne!(
            normalized_fp(&[tbl("issues", &["id", "title", "open"], &[0])]),
            base_fp
        );
    }

    fn mk_batch(epoch: u64, seq: u64, fp: u64) -> NormalizedBatch {
        NormalizedBatch {
            epoch,
            seq,
            cv: seq, // any monotone stand-in; the subscriber doesn't validate cv
            normalized_fp: fp,
            ops: vec![],
        }
    }

    fn open_sub() -> (NormalizedSubscriber, u64) {
        let tables = vec![tbl("issues", &["id", "title"], &[0])];
        let fp = normalized_fp(&tables);
        let hello = NormalizedHello {
            epoch: 7,
            comparator_version: COMPARATOR_VERSION,
            tables,
            normalized_fp: fp,
        };
        (NormalizedSubscriber::open(&hello).unwrap(), fp)
    }

    #[test]
    fn subscriber_enforces_order_dedup_and_gap() {
        let (mut sub, fp) = open_sub();
        // Baseline (seq 0), then 1, 2.
        assert!(matches!(
            sub.apply(&mk_batch(7, 0, fp)).unwrap(),
            NormalizedApplied::Live(_)
        ));
        assert!(matches!(
            sub.apply(&mk_batch(7, 1, fp)).unwrap(),
            NormalizedApplied::Live(_)
        ));
        // Re-delivery of seq 1 ⇒ idempotent duplicate.
        assert!(matches!(
            sub.apply(&mk_batch(7, 1, fp)).unwrap(),
            NormalizedApplied::Duplicate
        ));
        assert!(matches!(
            sub.apply(&mk_batch(7, 2, fp)).unwrap(),
            NormalizedApplied::Live(_)
        ));
        // A skip to seq 4 ⇒ fatal gap (expected 3).
        assert_eq!(
            sub.apply(&mk_batch(7, 4, fp)).unwrap_err(),
            NormalizedProtocolError::Gap {
                expected: 3,
                got: 4
            }
        );
    }

    #[test]
    fn subscriber_rejects_stale_epoch_and_comparator() {
        let (mut sub, fp) = open_sub();
        sub.apply(&mk_batch(7, 0, fp)).unwrap();
        // A stale-epoch frame is rejected (the re-hydrate guard, §5.3).
        assert_eq!(
            sub.apply(&mk_batch(6, 1, fp)).unwrap_err(),
            NormalizedProtocolError::EpochMismatch {
                expected: 7,
                got: 6
            }
        );
        // A comparator mismatch is refused at open.
        let hello = NormalizedHello {
            epoch: 1,
            comparator_version: COMPARATOR_VERSION + 1,
            tables: vec![tbl("issues", &["id"], &[0])],
            normalized_fp: 0,
        };
        assert!(matches!(
            NormalizedSubscriber::open(&hello),
            Err(NormalizedProtocolError::ComparatorMismatch { .. })
        ));
    }

    #[test]
    fn subscriber_rejects_malformed_schema_before_fingerprinting() {
        let hello = NormalizedHello {
            epoch: 1,
            comparator_version: COMPARATOR_VERSION,
            tables: vec![tbl("issues", &["id"], &[99])],
            // The malformed schema cannot be fingerprinted safely. `open` must reject it
            // before `normalized_fp` indexes the attacker-controlled PK.
            normalized_fp: 0,
        };
        assert_eq!(
            NormalizedSubscriber::open(&hello).unwrap_err(),
            NormalizedProtocolError::InvalidSchema(NormalizedSchemaError {
                table_index: 0,
                kind: NormalizedSchemaErrorKind::PrimaryKeyOutOfBounds,
            })
        );

        let hello = NormalizedHello {
            tables: vec![tbl("issues", &["id"], &[])],
            ..hello
        };
        assert_eq!(
            NormalizedSubscriber::open(&hello).unwrap_err(),
            NormalizedProtocolError::InvalidSchema(NormalizedSchemaError {
                table_index: 0,
                kind: NormalizedSchemaErrorKind::EmptyPrimaryKey,
            })
        );

        let hello = NormalizedHello {
            tables: vec![tbl("issues", &["id"], &[0]), tbl("issues", &["id"], &[0])],
            ..hello
        };
        assert_eq!(
            NormalizedSubscriber::open(&hello).unwrap_err(),
            NormalizedProtocolError::InvalidSchema(NormalizedSchemaError {
                table_index: 1,
                kind: NormalizedSchemaErrorKind::TablesNotSortedByUniqueName,
            })
        );
    }
}