rindle_wire/

wire_json.rs

//! The normalized protocol's JSON wire rendering — **byte-for-byte the shape the
//! `@rindle/remote` client validates** (bare cells, camelCase keys, lowercase `op` tags,
//! hex fingerprints, f64-safe counters). Moved here from `rindle-replica` so every party
//! on the wire shares ONE codec: the napi addon and the Rust daemon's ws front *emit*
//! with it (via `rindle-replica`'s re-export), and the room — the first Rust *consumer*
//! of these frames (`RINDLE-REALTIME-DESIGN.md` §3) — *parses* with it. The
//! `..._from_json` decoders are new with the room's upstream leg; until they landed,
//! decode existed only in TS (`@rindle/remote/src/normalized.ts`).
//!
//! Two conventions are deliberate and load-bearing:
//!
//! - **Counters are JS numbers** (`epoch`/`seq`/`cv`/`comparatorVersion` as f64 — the
//!   emitters write `as f64`, so a Rust producer renders `3` as `3.0` while a JS producer
//!   writes `3`; the decoders accept both), and **fingerprints are 16-hex-char strings**
//!   (a u64 would lose precision as a JS number).
//! - **Cells are bare** (`number | string | boolean | null`): `Int` collapses to a JS
//!   number and `Json` to its string form on encode, so decode yields `Float`/`Str` —
//!   the same inference the wasm `Db`'s `js_to_owned_value` applies. The wire is
//!   value-faithful, not variant-faithful, by design (the engine's single numeric model).
//!
//! The serde derives on the protocol types are **not** this wire — they are the in-Rust
//! round-trip form (oracles, journals). Sockets carry this rendering.
//!
//! Shape-only by design: [`normalized_hello_from_json`] checks structure, not semantics —
//! PK bounds, duplicate tables, and fingerprint verification belong to
//! `RoomStore::open` (in `rindle-room-core`) / the
//! [`NormalizedSubscriber`](crate::normalize_protocol::NormalizedSubscriber), which
//! return the protocol's own errors.

use std::sync::Arc;

use rindle::change::SourceChange;
use rindle::value::{OwnedRow, OwnedValue};
use serde_json::{json, Map, Value};

use crate::normalize::NormalizedOp;
use crate::normalize_protocol::{NormalizedBatch, NormalizedHello, TableWireSchema};

/// `OwnedValue` → a bare JS value. `Int`/`Float` collapse to a JS `number` (the engine's
/// single numeric model); `Str`/`Json` to a string (a json column is parsed at read time by
/// the view's typed schema, exactly as on the local path).
pub fn owned_to_json(v: &OwnedValue) -> Value {
    match v {
        // The server projects each query to its own columns at the emit boundary
        // (PROJECTION-SUPPORT-DESIGN.md §5.2), so a non-selected column is *dropped* from
        // the wire row — `Absent` never reaches the JSON renderer. Assert that; map to
        // `null` defensively.
        OwnedValue::Absent => {
            debug_assert!(false, "OwnedValue::Absent must never reach the server wire");
            Value::Null
        }
        OwnedValue::Null => Value::Null,
        OwnedValue::Bool(b) => Value::Bool(*b),
        OwnedValue::Int(i) => json!(*i as f64),
        OwnedValue::Float(f) => json!(*f),
        OwnedValue::Str(s) => Value::String(s.to_string()),
        OwnedValue::Json(s) => Value::String(s.to_string()),
    }
}

/// A bare JS cell → `OwnedValue`, inferring from the runtime JSON type (matches the wasm
/// `js_to_owned_value`: number→Float always, string→Str, bool→Bool, null→Null). A json
/// column already crossed as its stringified form, so it lands here as a `String`.
pub fn json_to_owned(v: &Value) -> OwnedValue {
    match v {
        Value::Null => OwnedValue::Null,
        Value::Bool(b) => OwnedValue::Bool(*b),
        Value::Number(n) => OwnedValue::Float(n.as_f64().unwrap_or(0.0)),
        Value::String(s) => OwnedValue::Str(Arc::from(s.as_str())),
        // Defensive: a non-stringified object/array cell isn't expected on the bare wire.
        other => OwnedValue::Str(Arc::from(other.to_string().as_str())),
    }
}

/// One positional wire row → its JSON array of bare cells.
pub fn wire_row_to_json(row: &[OwnedValue]) -> Value {
    Value::Array(row.iter().map(owned_to_json).collect())
}

/// One JSON array of bare cells → a positional wire row.
pub fn wire_row_from_json(v: &Value) -> Result<Vec<OwnedValue>, String> {
    Ok(v.as_array()
        .ok_or("wire row: not an array")?
        .iter()
        .map(json_to_owned)
        .collect())
}

/// A `NormalizedOp` → its camelCase JS object: `{ table, op, row | old/new }` (bare cells),
/// the path-free twin of a flat change (NORMALIZED-CHANGES-DESIGN.md §3).
pub fn normalized_op_to_json(op: &NormalizedOp) -> Value {
    match op {
        NormalizedOp::Add { table, row } => {
            json!({ "table": table.to_string(), "op": "add", "row": wire_row_to_json(row) })
        }
        NormalizedOp::Remove { table, row } => {
            json!({ "table": table.to_string(), "op": "remove", "row": wire_row_to_json(row) })
        }
        NormalizedOp::Edit { table, old, new } => json!({
            "table": table.to_string(),
            "op": "edit",
            "old": wire_row_to_json(old),
            "new": wire_row_to_json(new),
        }),
    }
}

/// Parse one `{ table, op, … }` wire object into a [`NormalizedOp`]. Note the shape is the
/// OUTBOUND batch wire — `remove` carries `row` and `edit` carries `old`/`new` — distinct
/// from the change-source ingest shape (`rindle-replica`'s `mutation_from_json`), where
/// `remove` carries `old` and `edit`'s new row rides as `row`.
pub fn normalized_op_from_json(v: &Value) -> Result<NormalizedOp, String> {
    let obj = v.as_object().ok_or("normalized op: not an object")?;
    let table: Box<str> = obj
        .get("table")
        .and_then(Value::as_str)
        .ok_or("normalized op missing table")?
        .into();
    let row = |key: &str| -> Result<Vec<OwnedValue>, String> {
        wire_row_from_json(
            obj.get(key)
                .ok_or_else(|| format!("normalized op missing {key}"))?,
        )
    };
    match obj.get("op").and_then(Value::as_str) {
        Some("add") => Ok(NormalizedOp::Add {
            table,
            row: row("row")?,
        }),
        Some("remove") => Ok(NormalizedOp::Remove {
            table,
            row: row("row")?,
        }),
        Some("edit") => Ok(NormalizedOp::Edit {
            table,
            old: row("old")?,
            new: row("new")?,
        }),
        other => Err(format!("unknown normalized op: {other:?}")),
    }
}

/// A `NormalizedBatch` → `{ epoch, seq, cv, normalizedFp, ops }`. The fingerprint is a
/// hex string (a u64 would lose precision as a JS number); `cv` is the commit version
/// the optimistic client buffers by (OPTIMISTIC-WRITES-DESIGN.md §8.6).
pub fn normalized_batch_to_json(b: &NormalizedBatch) -> Value {
    json!({
        "epoch": b.epoch as f64,
        "seq": b.seq as f64,
        "cv": b.cv as f64,
        "normalizedFp": format!("{:016x}", b.normalized_fp),
        "ops": b.ops.iter().map(normalized_op_to_json).collect::<Vec<_>>(),
    })
}

/// Parse an `nbatch` frame's `batch` object into a [`NormalizedBatch`].
pub fn normalized_batch_from_json(v: &Value) -> Result<NormalizedBatch, String> {
    let obj = v.as_object().ok_or("nbatch: not an object")?;
    let ops = obj
        .get("ops")
        .and_then(Value::as_array)
        .ok_or("nbatch missing ops")?
        .iter()
        .map(normalized_op_from_json)
        .collect::<Result<Vec<_>, _>>()?;
    Ok(NormalizedBatch {
        epoch: counter(obj, "epoch")?,
        seq: counter(obj, "seq")?,
        cv: counter(obj, "cv")?,
        normalized_fp: fingerprint(obj, "normalizedFp")?,
        ops,
    })
}

/// A `NormalizedHello` → `{ epoch, comparatorVersion, tables: [{ name, columns, primaryKey }],
/// normalizedFp }` — the slim per-table-schema handshake (§3).
pub fn normalized_hello_to_json(h: &NormalizedHello) -> Value {
    let tables: Vec<Value> = h
        .tables
        .iter()
        .map(|t| {
            json!({
                "name": t.name.to_string(),
                "columns": t.columns.iter().map(|c| Value::String(c.to_string())).collect::<Vec<_>>(),
                "primaryKey": t.primary_key.iter().map(|&i| json!(i as f64)).collect::<Vec<_>>(),
            })
        })
        .collect();
    json!({
        "epoch": h.epoch as f64,
        "comparatorVersion": h.comparator_version as f64,
        "tables": tables,
        "normalizedFp": format!("{:016x}", h.normalized_fp),
    })
}

/// Parse an `nhello` frame's `hello` object into a [`NormalizedHello`] (shape only — see
/// the module docs for where semantic validation lives).
pub fn normalized_hello_from_json(v: &Value) -> Result<NormalizedHello, String> {
    let obj = v.as_object().ok_or("nhello: not an object")?;
    let tables = obj
        .get("tables")
        .and_then(Value::as_array)
        .ok_or("nhello missing tables")?
        .iter()
        .map(table_wire_schema_from_json)
        .collect::<Result<Vec<_>, _>>()?;
    let comparator_version = counter(obj, "comparatorVersion")?;
    let comparator_version = u32::try_from(comparator_version)
        .map_err(|_| format!("comparatorVersion out of range: {comparator_version}"))?;
    Ok(NormalizedHello {
        epoch: counter(obj, "epoch")?,
        comparator_version,
        tables,
        normalized_fp: fingerprint(obj, "normalizedFp")?,
    })
}

fn table_wire_schema_from_json(v: &Value) -> Result<TableWireSchema, String> {
    let obj = v.as_object().ok_or("table schema: not an object")?;
    let name: Box<str> = obj
        .get("name")
        .and_then(Value::as_str)
        .ok_or("table schema missing name")?
        .into();
    let columns = obj
        .get("columns")
        .and_then(Value::as_array)
        .ok_or("table schema missing columns")?
        .iter()
        .map(|c| {
            c.as_str()
                .map(Box::from)
                .ok_or_else(|| "table schema: non-string column".to_string())
        })
        .collect::<Result<Vec<_>, _>>()?;
    let primary_key = obj
        .get("primaryKey")
        .and_then(Value::as_array)
        .ok_or("table schema missing primaryKey")?
        .iter()
        .map(|i| {
            let idx = counter_value(i, "primaryKey index")?;
            u32::try_from(idx).map_err(|_| format!("primaryKey index out of range: {idx}"))
        })
        .collect::<Result<Vec<_>, _>>()?;
    Ok(TableWireSchema {
        name,
        columns,
        primary_key,
    })
}

/// One wire counter field: a non-negative integer-valued JS number. Rust producers write
/// `3.0` (the emitters cast `as f64`), JS producers write `3` — accept both; reject
/// fractions, negatives, and anything past 2^53 (not exactly representable, so a producer
/// could never have meant it).
fn counter(obj: &Map<String, Value>, key: &str) -> Result<u64, String> {
    counter_value(obj.get(key).ok_or_else(|| format!("missing {key}"))?, key)
}

fn counter_value(v: &Value, what: &str) -> Result<u64, String> {
    if let Some(u) = v.as_u64() {
        return Ok(u);
    }
    let f = v.as_f64().ok_or_else(|| format!("{what}: not a number"))?;
    const MAX_SAFE: f64 = 9007199254740992.0; // 2^53
    if f.fract() != 0.0 || !(0.0..=MAX_SAFE).contains(&f) {
        return Err(format!("{what}: not a wire-safe counter: {f}"));
    }
    Ok(f as u64)
}

/// One wire fingerprint field: the emitter's `{:016x}` hex string.
fn fingerprint(obj: &Map<String, Value>, key: &str) -> Result<u64, String> {
    let s = obj
        .get(key)
        .and_then(Value::as_str)
        .ok_or_else(|| format!("missing {key}"))?;
    u64::from_str_radix(s, 16).map_err(|_| format!("{key}: not a hex fingerprint: {s:?}"))
}

/// A packed engine row → its JSON array of bare cells (the [`wire_row_to_json`] twin
/// for [`OwnedRow`]).
pub fn packed_row_to_json(row: &OwnedRow) -> Value {
    Value::Array(row.cells().map(|v| owned_to_json(&v.to_owned())).collect())
}

/// One row change in the change-source **ingest** shape — the `changes[]` element
/// `rindled`'s `/apply-row-change-txn` consumes (`rindle-replica`'s `mutation_from_json`
/// is the decode dual): `add → {row}`, `remove → {old}`, `edit → {old, row}` where `row`
/// is the NEW image. Deliberately distinct from [`normalized_op_to_json`] — that is the
/// OUTBOUND batch wire, where `remove` carries `row` and `edit` carries `new`. The
/// room's write-behind flush emits this shape (`RINDLE-REALTIME-DESIGN.md` §5.3), with
/// `old` carrying the **base** image the CAS precondition asserts. NOTE: this emitter
/// renders cells via [`owned_to_json`], which is FLOAT-FORM for `Int` — it is NOT the
/// integer-exact ingest codec (`rindle-replica::wire_json::mutation_to_json`, design
/// 226 Stage D). Room cells are Number-plane today so nothing is lost; unify onto the
/// exact codec before any exact-int producer feeds this path.
pub fn row_change_to_json(table: &str, change: &SourceChange) -> Value {
    match change {
        SourceChange::Add(row) => {
            json!({ "table": table, "op": "add", "row": packed_row_to_json(row) })
        }
        SourceChange::Remove(row) => {
            json!({ "table": table, "op": "remove", "old": packed_row_to_json(row) })
        }
        SourceChange::Edit { row, old } => json!({
            "table": table,
            "op": "edit",
            "old": packed_row_to_json(old),
            "row": packed_row_to_json(row),
        }),
    }
}

/// FNV-1a-64 over raw bytes — the flush batch identity (`batch_hash`, §5.3 step 4:
/// computed once by the emitter over its canonical batch bytes; the authority stores
/// and *compares* it, never recomputes). Render with `{:016x}`.
pub fn fnv1a64(bytes: &[u8]) -> u64 {
    let mut h: u64 = 0xcbf29ce4_84222325; // FNV-1a 64 offset basis
    for &b in bytes {
        h ^= b as u64;
        h = h.wrapping_mul(0x0000_0100_0000_01b3); // FNV-1a 64 prime
    }
    h
}

// ---------------------------------------------------------------------------
// Direct-to-bytes `nbatch` encoding — the server delivery fan-out hot path
// ---------------------------------------------------------------------------
//
// `RouterSink::batch` (in `rindle-server`) fans one commit's derived delta out to every
// subscribing connection. The obvious path — build a `serde_json::Value` tree per
// subscriber (`normalized_batch_to_json`) and `.to_string()` it — pays a fresh `Value` tree
// (a heap `Value` per cell + a `Map`/`Vec` per op) for every subscriber. [`nbatch_frame`] /
// [`nbatch_ops_fragment`] instead write the frame DIRECTLY into a byte buffer, no `Value`
// tree: encode the `ops` array once (outside the router lock), then splice it into each
// subscriber's cheap scalar envelope.
//
// (An earlier revision also memoized the encoded `ops` fragment within a tick to collapse
// identical-delta fan-out, but an isolated before/after showed no measurable win — delivery
// is not the drain thread's bottleneck in the measured regimes — so it was dropped for the
// simpler stateless path. The structural fix for wide identical-delta fan-out is subscriber
// dedup, RINDLE-SERVER-DESIGN.md §6, which collapses the derive too.)
//
// **Byte-for-byte identical** to the `json!` renderer it replaces, by construction: every
// scalar (counters as f64, cells, strings) is emitted by `serde_json` itself
// (`to_writer` → `serialize_f64` / `serialize_str`, the same calls `Value::to_string`
// makes), and object keys are written in the alphabetical order the `BTreeMap`-backed
// `serde_json::Map` produces. `nbatch_frame_matches_json_macro` pins this against the old
// renderer across every op/cell shape — a divergence fails there, not in a browser.

/// Append `serde_json`'s rendering of the f64 `f` to `buf`. Delegating to `serde_json`
/// (not a hand-rolled `ryu`/`write!`) is what makes the output byte-identical to the
/// `json!(x as f64)` path: `5u64 as f64` → `5.0` (never `5`), non-finite → `null`.
fn write_f64(buf: &mut Vec<u8>, f: f64) {
    serde_json::to_writer(&mut *buf, &f).expect("Vec<u8> write is infallible");
}

/// Append the JSON string form of `s` — quoted and escaped exactly as `serde_json`
/// escapes it (the same `serialize_str` `Value::String` uses).
fn write_json_str(buf: &mut Vec<u8>, s: &str) {
    serde_json::to_writer(&mut *buf, &s).expect("Vec<u8> write is infallible");
}

/// Append `v` as a 16-char lowercase zero-padded hex run — identical to `{:016x}`, with
/// no `format!` allocation. Hex digits never need JSON escaping, so the caller wraps it
/// in quotes directly.
fn write_hex16(buf: &mut Vec<u8>, mut v: u64) {
    let mut out = [0u8; 16];
    for slot in out.iter_mut().rev() {
        let nibble = (v & 0xf) as u8;
        *slot = if nibble < 10 {
            b'0' + nibble
        } else {
            b'a' + (nibble - 10)
        };
        v >>= 4;
    }
    buf.extend_from_slice(&out);
}

/// One bare cell → its JS value, byte-identical to [`owned_to_json`] + `.to_string()`.
fn write_cell(buf: &mut Vec<u8>, v: &OwnedValue) {
    match v {
        OwnedValue::Absent => {
            debug_assert!(false, "OwnedValue::Absent must never reach the server wire");
            buf.extend_from_slice(b"null");
        }
        OwnedValue::Null => buf.extend_from_slice(b"null"),
        OwnedValue::Bool(true) => buf.extend_from_slice(b"true"),
        OwnedValue::Bool(false) => buf.extend_from_slice(b"false"),
        // `Int` collapses to a JS number exactly as `json!(*i as f64)` does.
        OwnedValue::Int(i) => write_f64(buf, *i as f64),
        OwnedValue::Float(f) => write_f64(buf, *f),
        OwnedValue::Str(s) | OwnedValue::Json(s) => write_json_str(buf, s),
    }
}

/// One positional row → its `[cells…]` array.
fn write_row(buf: &mut Vec<u8>, row: &[OwnedValue]) {
    buf.push(b'[');
    for (i, c) in row.iter().enumerate() {
        if i > 0 {
            buf.push(b',');
        }
        write_cell(buf, c);
    }
    buf.push(b']');
}

/// One op → its `{…}` object, keys in the alphabetical order `serde_json` emits.
fn write_op(buf: &mut Vec<u8>, op: &NormalizedOp) {
    match op {
        // keys: op, row, table
        NormalizedOp::Add { table, row } => {
            buf.extend_from_slice(br#"{"op":"add","row":"#);
            write_row(buf, row);
            buf.extend_from_slice(br#","table":"#);
            write_json_str(buf, table);
            buf.push(b'}');
        }
        NormalizedOp::Remove { table, row } => {
            buf.extend_from_slice(br#"{"op":"remove","row":"#);
            write_row(buf, row);
            buf.extend_from_slice(br#","table":"#);
            write_json_str(buf, table);
            buf.push(b'}');
        }
        // keys: new, old, op, table
        NormalizedOp::Edit { table, old, new } => {
            buf.extend_from_slice(br#"{"new":"#);
            write_row(buf, new);
            buf.extend_from_slice(br#","old":"#);
            write_row(buf, old);
            buf.extend_from_slice(br#","op":"edit","table":"#);
            write_json_str(buf, table);
            buf.push(b'}');
        }
    }
}

/// Encode `ops` as its `[…]` JSON array — the memoizable fragment.
fn write_ops(buf: &mut Vec<u8>, ops: &[NormalizedOp]) {
    buf.push(b'[');
    for (i, op) in ops.iter().enumerate() {
        if i > 0 {
            buf.push(b',');
        }
        write_op(buf, op);
    }
    buf.push(b']');
}

/// Assemble one full `nbatch` frame for `client_qid`, splicing the already-encoded
/// `ops_fragment` into the scalar envelope. Byte-identical to the `json!` frame in
/// `RouterSink::batch`: `{"batch":{"cv":…,"epoch":…,"normalizedFp":"…","ops":…,"seq":…},
/// "queryId":…,"t":"nbatch"}` (all keys alphabetical).
pub fn nbatch_frame(client_qid: u64, batch: &NormalizedBatch, ops_fragment: &str) -> String {
    let mut buf = Vec::with_capacity(ops_fragment.len() + 96);
    buf.extend_from_slice(br#"{"batch":{"cv":"#);
    write_f64(&mut buf, batch.cv as f64);
    buf.extend_from_slice(br#","epoch":"#);
    write_f64(&mut buf, batch.epoch as f64);
    buf.extend_from_slice(br#","normalizedFp":""#);
    write_hex16(&mut buf, batch.normalized_fp);
    buf.extend_from_slice(br#"","ops":"#);
    buf.extend_from_slice(ops_fragment.as_bytes());
    buf.extend_from_slice(br#","seq":"#);
    write_f64(&mut buf, batch.seq as f64);
    buf.extend_from_slice(br#"},"queryId":"#);
    write_f64(&mut buf, client_qid as f64);
    buf.extend_from_slice(br#","t":"nbatch"}"#);
    String::from_utf8(buf).expect("nbatch frame is valid UTF-8")
}

/// Encode `batch.ops` as its `[…]` JSON array — the fragment [`nbatch_frame`] splices into
/// each subscriber's envelope. `RouterSink::batch` encodes it ONCE per batch, outside the
/// router lock, then assembles a frame per subscriber (only the envelope's
/// `queryId`/`seq`/`epoch` differ across a fan-out).
pub fn nbatch_ops_fragment(batch: &NormalizedBatch) -> String {
    let mut buf = Vec::new();
    write_ops(&mut buf, &batch.ops);
    String::from_utf8(buf).expect("ops fragment is valid UTF-8")
}

#[cfg(test)]
mod tests {
    use super::*;
    use rindle::value::OwnedValue as V;

    fn hello() -> NormalizedHello {
        NormalizedHello {
            epoch: 3,
            comparator_version: 1,
            tables: vec![
                TableWireSchema {
                    name: "comment".into(),
                    columns: vec!["id".into(), "issue_id".into(), "body".into()],
                    primary_key: vec![0],
                },
                TableWireSchema {
                    name: "issue".into(),
                    columns: vec!["id".into(), "title".into()],
                    primary_key: vec![0, 1],
                },
            ],
            normalized_fp: 0xdead_beef_0102_0304,
        }
    }

    fn batch() -> NormalizedBatch {
        NormalizedBatch {
            epoch: 3,
            seq: 7,
            cv: 41,
            normalized_fp: 0xdead_beef_0102_0304,
            ops: vec![
                NormalizedOp::Add {
                    table: "issue".into(),
                    row: vec![V::Float(1.0), V::str("a")],
                },
                NormalizedOp::Edit {
                    table: "issue".into(),
                    old: vec![V::Float(1.0), V::str("a")],
                    new: vec![V::Float(1.0), V::Null],
                },
                NormalizedOp::Remove {
                    table: "comment".into(),
                    row: vec![V::Float(9.0), V::Float(1.0), V::Bool(true)],
                },
            ],
        }
    }

    /// The golden vector: pins camelCase keys, hex fp, and alphabetical serde_json key
    /// order, so an accidental rename breaks HERE and not in a browser.
    #[test]
    fn hello_golden_wire_form() {
        let json = normalized_hello_to_json(&hello()).to_string();
        assert_eq!(
            json,
            r#"{"comparatorVersion":1.0,"epoch":3.0,"normalizedFp":"deadbeef01020304","tables":[{"columns":["id","issue_id","body"],"name":"comment","primaryKey":[0.0]},{"columns":["id","title"],"name":"issue","primaryKey":[0.0,1.0]}]}"#
        );
    }

    /// decode(encode(x)) == x for wire-faithful values (Float/Str/Bool/Null cells — the
    /// forms decode can produce), and the encoding is stable across a second round trip.
    #[test]
    fn hello_and_batch_round_trip() {
        let h = hello();
        let hj = normalized_hello_to_json(&h);
        assert_eq!(normalized_hello_from_json(&hj).unwrap(), h);

        let b = batch();
        let bj = normalized_batch_to_json(&b);
        let back = normalized_batch_from_json(&bj).unwrap();
        // NormalizedBatch has no PartialEq; the wire form IS the contract — compare there.
        assert_eq!(normalized_batch_to_json(&back), bj);
        assert_eq!(back.epoch, 3);
        assert_eq!(back.seq, 7);
        assert_eq!(back.cv, 41);
        assert_eq!(back.normalized_fp, 0xdead_beef_0102_0304);
        assert_eq!(back.ops.len(), 3);
    }

    /// `Int` and `Json` cells are value-faithful, not variant-faithful: they cross as
    /// number/string and come back as `Float`/`Str` with the same wire form.
    #[test]
    fn int_and_json_cells_collapse_to_wire_forms() {
        let row = vec![V::Int(5), V::Json(Arc::from(r#"{"k":1}"#))];
        let j = wire_row_to_json(&row);
        assert_eq!(j.to_string(), r#"[5.0,"{\"k\":1}"]"#);
        let back = wire_row_from_json(&j).unwrap();
        assert!(matches!(back[0], V::Float(f) if f == 5.0));
        assert!(matches!(&back[1], V::Str(s) if &**s == r#"{"k":1}"#));
        // And the collapsed row re-encodes identically.
        assert_eq!(wire_row_to_json(&back), j);
    }

    /// The direct-to-bytes [`NbatchEncoder`] must be byte-for-byte identical to the
    /// `json!` frame `RouterSink::batch` used to build — over every op shape, every cell
    /// variant, edge floats (−0.0, a huge float, a past-2^53 int), and an escapable +
    /// non-ASCII string. This is the wire-compat contract; a drift breaks a browser.
    #[test]
    fn nbatch_frame_matches_json_macro() {
        let cases = [
            batch(),
            // Empty ops (a fold that nets to a bare envelope).
            NormalizedBatch {
                epoch: 0,
                seq: 0,
                cv: 0,
                normalized_fp: 0,
                ops: vec![],
            },
            // Every cell variant + escaping/formatting edges.
            NormalizedBatch {
                epoch: 2,
                seq: 9,
                cv: 100,
                normalized_fp: 0xff,
                ops: vec![
                    NormalizedOp::Add {
                        table: "w*ird \"tbl\"\n".into(),
                        row: vec![
                            V::Null,
                            V::Bool(false),
                            V::Int(-7),
                            V::Float(-0.0),
                            V::Float(1.5e300),
                            V::str("héllo \"q\"\t/\\"),
                            V::Json(Arc::from(r#"{"a":[1,2]}"#)),
                        ],
                    },
                    // A past-2^53 int collapses through `as f64` exactly as the old path did.
                    NormalizedOp::Remove {
                        table: "t".into(),
                        row: vec![V::Int(9_007_199_254_740_993)],
                    },
                ],
            },
        ];
        for (i, b) in cases.iter().enumerate() {
            let qid = (i as u64) * 7 + 3;
            let expect = json!({
                "t": "nbatch",
                "queryId": qid as f64,
                "batch": normalized_batch_to_json(b),
            })
            .to_string();
            let got = nbatch_frame(qid, b, &nbatch_ops_fragment(b));
            assert_eq!(got, expect, "frame {i} diverged");
        }
    }

    /// A JS producer writes bare integers (`3`, not `3.0`) — both parse.
    #[test]
    fn accepts_js_integer_counters() {
        let b = normalized_batch_from_json(
            &serde_json::from_str(
                r#"{"epoch":1,"seq":0,"cv":12,"normalizedFp":"00000000000000ff","ops":[]}"#,
            )
            .unwrap(),
        )
        .unwrap();
        assert_eq!((b.epoch, b.seq, b.cv, b.normalized_fp), (1, 0, 12, 0xff));

        let h = normalized_hello_from_json(
            &serde_json::from_str(
                r#"{"epoch":1,"comparatorVersion":1,"tables":[],"normalizedFp":"0"}"#,
            )
            .unwrap(),
        )
        .unwrap();
        assert_eq!(h.epoch, 1);
        assert_eq!(h.comparator_version, 1);
    }

    #[test]
    fn malformed_frames_error_loudly() {
        let cases = [
            // fractional counter
            r#"{"epoch":1.5,"seq":0,"cv":0,"normalizedFp":"0","ops":[]}"#,
            // negative counter
            r#"{"epoch":-1,"seq":0,"cv":0,"normalizedFp":"0","ops":[]}"#,
            // fp not hex
            r#"{"epoch":1,"seq":0,"cv":0,"normalizedFp":"zz","ops":[]}"#,
            // fp as a number
            r#"{"epoch":1,"seq":0,"cv":0,"normalizedFp":12,"ops":[]}"#,
            // missing cv
            r#"{"epoch":1,"seq":0,"normalizedFp":"0","ops":[]}"#,
            // op with unknown tag
            r#"{"epoch":1,"seq":0,"cv":0,"normalizedFp":"0","ops":[{"table":"t","op":"upsert","row":[]}]}"#,
            // edit missing new
            r#"{"epoch":1,"seq":0,"cv":0,"normalizedFp":"0","ops":[{"table":"t","op":"edit","old":[]}]}"#,
            // row not an array
            r#"{"epoch":1,"seq":0,"cv":0,"normalizedFp":"0","ops":[{"table":"t","op":"add","row":5}]}"#,
            // op missing table
            r#"{"epoch":1,"seq":0,"cv":0,"normalizedFp":"0","ops":[{"op":"add","row":[]}]}"#,
        ];
        for c in cases {
            let v: Value = serde_json::from_str(c).unwrap();
            assert!(
                normalized_batch_from_json(&v).is_err(),
                "should reject: {c}"
            );
        }

        let hellos = [
            // non-string column
            r#"{"epoch":1,"comparatorVersion":1,"tables":[{"name":"t","columns":[1],"primaryKey":[0]}],"normalizedFp":"0"}"#,
            // pk index fractional
            r#"{"epoch":1,"comparatorVersion":1,"tables":[{"name":"t","columns":["a"],"primaryKey":[0.5]}],"normalizedFp":"0"}"#,
            // comparatorVersion overflows u32
            r#"{"epoch":1,"comparatorVersion":4294967296,"tables":[],"normalizedFp":"0"}"#,
            // tables not an array
            r#"{"epoch":1,"comparatorVersion":1,"tables":{},"normalizedFp":"0"}"#,
        ];
        for c in hellos {
            let v: Value = serde_json::from_str(c).unwrap();
            assert!(
                normalized_hello_from_json(&v).is_err(),
                "should reject: {c}"
            );
        }
    }
}