rindle_wire/

query_key.rs

//! [`QueryKey`] — the canonical identity of a subscription query, the dedup key of
//! every materialization map (`rindled`'s `MaterializationManager` and the room's
//! materialize-on-presentation bookkeeping, `RINDLE-REALTIME-DESIGN.md` §4/§10.1).
//! Two subscribers share one materialized pipeline iff their keys are equal; the
//! `fingerprint_hex` form is the `queryKey` string that crosses the control planes.
//! Moved here from `rindle-server` so the daemon and the room compute identical keys
//! by construction.

use std::collections::BTreeMap;
use std::fmt;
use std::hash::{Hash, Hasher};

use rindle::{canonicalize_wire_number_lits, Ast};
use serde_json::Value;

/// Why a [`QueryKey`] could not be computed: the AST failed to serialize to JSON
/// (the canonical byte form). Practically unreachable for a well-formed [`Ast`].
#[derive(Debug)]
pub struct QueryKeyError(pub serde_json::Error);

impl fmt::Display for QueryKeyError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "query key canonicalization failed: {}", self.0)
    }
}

impl std::error::Error for QueryKeyError {}

/// Daemon output stream shape. Daemon subscriptions prefer [`StreamMode::Normalized`]
/// for v1, but the mode participates in the query key so future flat streams never
/// accidentally share a materialization.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum StreamMode {
    Normalized,
    Flat,
}

impl StreamMode {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            StreamMode::Normalized => "normalized",
            StreamMode::Flat => "flat",
        }
    }
}

/// Every daemon-side input that can affect emitted rows for a materialized query.
#[derive(Clone, Debug, Eq)]
pub struct QueryKey {
    schema_version: String,
    stream_mode: StreamMode,
    visibility_key: String,
    canonical_ast: Vec<u8>,
    fingerprint: u64,
}

impl QueryKey {
    pub fn new(
        schema_version: impl Into<String>,
        stream_mode: StreamMode,
        ast: &Ast,
        visibility_key: impl Into<String>,
    ) -> Result<QueryKey, QueryKeyError> {
        let schema_version = schema_version.into();
        let visibility_key = visibility_key.into();
        // Design 310 impl plan F2: the wire-token number canonicalization
        // (`canonicalize_wire_number_lits`) used to run only at the wasm boundary, so
        // `Int(1)` and `Number(1.0)` spellings of one query forked into two
        // materializations here. Canonicalize the clone the bytes are taken from, so
        // the key — like `FamilyKey` — is over the wire-token identity.
        let mut ast = ast.clone();
        canonicalize_wire_number_lits(&mut ast);
        let canonical_ast = canonical_ast_bytes(&ast)?;

        let mut hasher = Fnv1a64::new();
        hash_field(&mut hasher, schema_version.as_bytes());
        hash_field(&mut hasher, stream_mode.as_str().as_bytes());
        hash_field(&mut hasher, visibility_key.as_bytes());
        hash_field(&mut hasher, &canonical_ast);
        let fingerprint = hasher.finish();

        Ok(QueryKey {
            schema_version,
            stream_mode,
            visibility_key,
            canonical_ast,
            fingerprint,
        })
    }

    pub fn schema_version(&self) -> &str {
        &self.schema_version
    }

    pub fn stream_mode(&self) -> StreamMode {
        self.stream_mode
    }

    pub fn visibility_key(&self) -> &str {
        &self.visibility_key
    }

    pub fn canonical_ast(&self) -> &[u8] {
        &self.canonical_ast
    }

    pub fn fingerprint(&self) -> u64 {
        self.fingerprint
    }

    pub fn fingerprint_hex(&self) -> String {
        format!("{:016x}", self.fingerprint)
    }
}

impl PartialEq for QueryKey {
    fn eq(&self, other: &Self) -> bool {
        self.schema_version == other.schema_version
            && self.stream_mode == other.stream_mode
            && self.visibility_key == other.visibility_key
            && self.canonical_ast == other.canonical_ast
    }
}

impl Hash for QueryKey {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.schema_version.hash(state);
        self.stream_mode.hash(state);
        self.visibility_key.hash(state);
        self.canonical_ast.hash(state);
    }
}

/// Canonical bytes for a typed AST. This is intentionally not caller-provided raw
/// JSON text: serde defaults normalize omitted optional fields, and object keys are
/// sorted recursively before encoding.
pub fn canonical_ast_bytes(ast: &Ast) -> Result<Vec<u8>, QueryKeyError> {
    let mut value = serde_json::to_value(ast).map_err(QueryKeyError)?;
    normalize_select_arrays(&mut value);
    let mut out = Vec::new();
    write_canonical_json(&value, &mut out)?;
    Ok(out)
}

/// Normalize every `select` projection list in the AST JSON — sort + dedup the column
/// names — so `select=[title,priority]` and `[priority,title]` canonicalize to the same
/// [`QueryKey`] (column order does not affect results; view column order is fixed by the
/// schema — `PROJECTION-SUPPORT-DESIGN.md` §5.1 / OQ-5). An **omitted** `select` (`'*'`)
/// stays distinct from any explicit list: it is simply absent from the JSON, never folded
/// to a full-column list (keeps the key schema-agnostic, avoiding a schema-evolution
/// hazard). Recurses so nested subqueries (`related`, correlated `where`) normalize too.
pub(crate) fn normalize_select_arrays(value: &mut Value) {
    match value {
        Value::Object(map) => {
            if let Some(Value::Array(items)) = map.get_mut("select") {
                if items.iter().all(Value::is_string) {
                    items.sort_by(|a, b| a.as_str().cmp(&b.as_str()));
                    items.dedup();
                }
            }
            for item in map.values_mut() {
                normalize_select_arrays(item);
            }
        }
        Value::Array(items) => {
            for item in items.iter_mut() {
                normalize_select_arrays(item);
            }
        }
        _ => {}
    }
}

pub(crate) fn write_canonical_json(value: &Value, out: &mut Vec<u8>) -> Result<(), QueryKeyError> {
    match value {
        Value::Null => out.extend_from_slice(b"null"),
        Value::Bool(v) => out.extend_from_slice(if *v { b"true" } else { b"false" }),
        Value::Number(n) => out.extend_from_slice(n.to_string().as_bytes()),
        Value::String(s) => {
            let quoted = serde_json::to_string(s).map_err(QueryKeyError)?;
            out.extend_from_slice(quoted.as_bytes());
        }
        Value::Array(items) => {
            out.push(b'[');
            for (idx, item) in items.iter().enumerate() {
                if idx > 0 {
                    out.push(b',');
                }
                write_canonical_json(item, out)?;
            }
            out.push(b']');
        }
        Value::Object(map) => {
            let sorted: BTreeMap<_, _> = map.iter().collect();
            out.push(b'{');
            for (idx, (key, item)) in sorted.iter().enumerate() {
                if idx > 0 {
                    out.push(b',');
                }
                let quoted = serde_json::to_string(key).map_err(QueryKeyError)?;
                out.extend_from_slice(quoted.as_bytes());
                out.push(b':');
                write_canonical_json(item, out)?;
            }
            out.push(b'}');
        }
    }
    Ok(())
}

pub(crate) fn hash_field(hasher: &mut Fnv1a64, bytes: &[u8]) {
    hasher.write(&(bytes.len() as u64).to_le_bytes());
    hasher.write(bytes);
}

pub(crate) struct Fnv1a64(u64);

impl Fnv1a64 {
    pub(crate) fn new() -> Fnv1a64 {
        Fnv1a64(0xcbf29ce484222325)
    }
}

impl Hasher for Fnv1a64 {
    fn write(&mut self, bytes: &[u8]) {
        for b in bytes {
            self.0 ^= u64::from(*b);
            self.0 = self.0.wrapping_mul(0x100000001b3);
        }
    }

    fn finish(&self) -> u64 {
        self.0
    }
}

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

    fn key(ast: &Ast) -> QueryKey {
        QueryKey::new("v1", StreamMode::Normalized, ast, "vis").expect("key builds")
    }

    #[test]
    fn select_column_order_does_not_fork_the_key() {
        // `[title, priority]` and `[priority, title]` select the same columns; the view
        // column order is schema-fixed, so they must share a materialization (OQ-5).
        let a = table("issue").select("title").select("priority").build();
        let b = table("issue").select("priority").select("title").build();
        assert_eq!(key(&a), key(&b));
        assert_eq!(key(&a).fingerprint(), key(&b).fingerprint());
    }

    #[test]
    fn duplicate_selected_columns_are_deduped() {
        let a = table("issue").select("title").select("title").build();
        let b = table("issue").select("title").build();
        assert_eq!(key(&a), key(&b));
    }

    #[test]
    fn different_projections_get_distinct_keys() {
        // A narrower subscriber must NOT share a materialization with a wider one
        // (a column-permission leak / over-send, §5.1).
        let one = table("issue").select("title").build();
        let two = table("issue").select("title").select("priority").build();
        assert_ne!(one, two);
    }

    #[test]
    fn int_and_integral_number_spellings_share_a_key() {
        // F2 (design 310 impl plan): `where x = 1` and `where x = 1.0` are one query
        // text on every JSON home, so they must share a materialization; a non-integral
        // number stays distinct.
        let int = table("issue").r#where("x", 1i64).build();
        let num = table("issue").r#where("x", 1.0f64).build();
        let frac = table("issue").r#where("x", 1.5f64).build();
        assert_eq!(key(&int), key(&num));
        assert_ne!(key(&int), key(&frac));
    }

    #[test]
    fn star_is_distinct_from_an_explicit_list() {
        // `'*'` (omitted select) means "whatever columns exist now"; an explicit list
        // means "exactly these". They must stay distinct even if the list equals the
        // current schema (§5.1 — keeps QueryKey schema-agnostic).
        let star = table("issue").build();
        let explicit = table("issue").select("id").select("title").build();
        assert_ne!(star, explicit);
    }
}