lix 0.15.1

#[cfg(test)]
use std::collections::BTreeMap;
use std::collections::HashMap;
#[cfg(test)]
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, OnceLock};

use serde_json::Value as JsonValue;

use crate::catalog::revision::CatalogRevision;
use crate::catalog::snapshot::{
    CatalogFingerprint, fingerprint_schema_facts, hash_fingerprint_part,
};
use crate::catalog::{CatalogSnapshot, SchemaCatalogFact};
use crate::domain::{Domain, committed_row_ref_is_exact_branch_scoped};
use crate::hot_state::{
    HotStateFilter, HotStateReader, HotStateScanRequest, MaterializedHotStateBatch,
    MaterializedHotStateRowRef,
};
use crate::{LixError, NullableKeyFilter};

const REGISTERED_SCHEMA_KEY: &str = "lix_registered_schema";

/// Compiled catalog snapshots are cached at most this many fact sets deep.
/// Schema catalogs churn rarely; the bound only guards against pathological
/// schema-mutation workloads growing the cache without limit.
const COMPILED_CATALOG_CACHE_LIMIT: usize = 64;

/// Engine schema visibility boundary.
///
/// Dynamic SQL planning receives an effective schema snapshot composed from
/// immutable engine built-ins plus custom schemas in live state. Historical
/// repositories can retain built-in `lix_registered_schema` rows for
/// introspection, but those mirrors never participate in schema authority. The
/// context also owns engine-wide caches keyed by effective fact content or the
/// atomic storage revision captured when a transaction opens.
pub(crate) struct CatalogContext {
    compiled_catalogs: Mutex<HashMap<CatalogFingerprint, Arc<CatalogSnapshot>>>,
    transaction_opening_catalogs:
        Mutex<HashMap<TransactionOpeningCatalogKey, Arc<CatalogSnapshot>>>,
    #[cfg(test)]
    sql_read_schema_loads: AtomicUsize,
    #[cfg(test)]
    pub(crate) committed_catalog_warms: AtomicUsize,
}

/// Fingerprint of the raw catalog rows visible to a domain, hashed before any
/// JSON decoding. The raw `snapshot_content` bytes uniquely determine the
/// decoded facts, so this key is at least as discriminating as the facts
/// fingerprint: textual or ordering variations can only cause a conservative
/// cache miss, never a wrong hit.
#[derive(Clone, PartialEq, Eq, Hash)]
struct CatalogRowsFingerprint(String);

#[derive(Clone, PartialEq, Eq, Hash)]
struct TransactionOpeningCatalogKey {
    domain: Domain,
    revision: CatalogRevision,
}

impl CatalogContext {
    pub(crate) fn new() -> Self {
        Self {
            compiled_catalogs: Mutex::new(HashMap::new()),
            transaction_opening_catalogs: Mutex::new(HashMap::new()),
            #[cfg(test)]
            sql_read_schema_loads: AtomicUsize::new(0),
            #[cfg(test)]
            committed_catalog_warms: AtomicUsize::new(0),
        }
    }

    /// Returns the catalog captured by a transaction-opening storage snapshot.
    ///
    /// The revision is loaded from the same pinned read as `hot_state`. A
    /// matching `(domain, revision)` therefore proves that registered-schema
    /// visibility is identical without rescanning its tracked and untracked
    /// rows. Missing revisions conservatively retain the scan path for stores
    /// initialized by older engines.
    pub(crate) async fn compiled_catalog_for_transaction_open<R>(
        &self,
        hot_state: &R,
        domain: &Domain,
        revision: Option<&CatalogRevision>,
    ) -> Result<Arc<CatalogSnapshot>, LixError>
    where
        R: HotStateReader + ?Sized,
    {
        let Some(revision) = revision else {
            return self.compiled_catalog_for_domain(hot_state, domain).await;
        };
        let key = TransactionOpeningCatalogKey {
            domain: domain.clone(),
            revision: revision.clone(),
        };
        if let Some(snapshot) = self
            .transaction_opening_catalogs
            .lock()
            .expect("transaction opening catalog cache lock should not be poisoned")
            .get(&key)
        {
            return Ok(Arc::clone(snapshot));
        }

        let snapshot = self.compiled_catalog_for_domain(hot_state, domain).await?;
        let mut cache = self
            .transaction_opening_catalogs
            .lock()
            .expect("transaction opening catalog cache lock should not be poisoned");
        if cache.len() >= COMPILED_CATALOG_CACHE_LIMIT {
            if let Some(evicted) = cache.keys().find(|entry| **entry != key).cloned() {
                cache.remove(&evicted);
            }
        }
        cache.insert(key, Arc::clone(&snapshot));
        Ok(snapshot)
    }

    /// Returns the compiled snapshot for the catalog rows visible to `domain`.
    ///
    /// This is the transaction-opening cache-miss path and the authoritative
    /// path for validation overlays. Identical native rows avoid JSON decoding
    /// and canonicalization, but still require the live-state scans.
    pub(crate) async fn compiled_catalog_for_domain<R>(
        &self,
        hot_state: &R,
        domain: &Domain,
    ) -> Result<Arc<CatalogSnapshot>, LixError>
    where
        R: HotStateReader + ?Sized,
    {
        let catalog_rows = scan_catalog_rows(hot_state, domain).await?;
        let mut hasher = blake3::Hasher::new();
        for (schema_domain, row) in catalog_rows.iter() {
            hash_fingerprint_part(&mut hasher, &schema_domain.fingerprint_component());
            let payload = row
                .raw_snapshot()
                .expect("catalog rows are filtered to raw native payloads");
            hasher.update(&(payload.len() as u64).to_le_bytes());
            hasher.update(payload);
        }
        let fingerprint = CatalogRowsFingerprint(hasher.finalize().to_hex().to_string());

        if let Some(snapshot) = compiled_catalogs_by_rows()
            .lock()
            .expect("compiled catalog rows cache lock should not be poisoned")
            .get(&fingerprint)
        {
            return Ok(Arc::clone(snapshot));
        }

        let facts = facts_from_catalog_rows(&catalog_rows)?;
        let snapshot = self.compiled_catalog_for_facts(&facts)?;
        let mut cache = compiled_catalogs_by_rows()
            .lock()
            .expect("compiled catalog rows cache lock should not be poisoned");
        if cache.len() >= COMPILED_CATALOG_CACHE_LIMIT {
            if let Some(evicted) = cache.keys().find(|key| **key != fingerprint).cloned() {
                cache.remove(&evicted);
            }
        }
        cache.insert(fingerprint, Arc::clone(&snapshot));
        Ok(snapshot)
    }

    /// Returns the compiled snapshot for `facts`, building it on first use.
    ///
    /// The cache key is the content fingerprint of the facts, so a cached
    /// snapshot can never go stale: changed schema rows produce different
    /// facts and therefore a different key. The lock is not held while
    /// compiling, so two racing callers may both compile the same facts; the
    /// results are identical and the last insert wins.
    pub(crate) fn compiled_catalog_for_facts(
        &self,
        facts: &[SchemaCatalogFact],
    ) -> Result<Arc<CatalogSnapshot>, LixError> {
        let fingerprint = fingerprint_schema_facts(facts)?;
        if let Some(snapshot) = self
            .compiled_catalogs
            .lock()
            .expect("compiled catalog cache lock should not be poisoned")
            .get(&fingerprint)
        {
            return Ok(Arc::clone(snapshot));
        }
        let snapshot = Arc::new(CatalogSnapshot::from_schema_facts(facts)?);
        #[cfg(feature = "storage-benches")]
        crate::storage_bench::record_transaction_schema_catalog_compile();
        let mut cache = self
            .compiled_catalogs
            .lock()
            .expect("compiled catalog cache lock should not be poisoned");
        if cache.len() >= COMPILED_CATALOG_CACHE_LIMIT {
            // Evict one other entry instead of clearing: identical schema
            // content on N branches occupies N keys, so a full clear would
            // recompile every hot catalog whenever the bound is reached.
            if let Some(evicted) = cache.keys().find(|key| **key != fingerprint).cloned() {
                cache.remove(&evicted);
            }
        }
        cache.insert(fingerprint, Arc::clone(&snapshot));
        Ok(snapshot)
    }

    /// Projects schema definitions for SQL surface-planning cache tests.
    ///
    /// SQL surfaces are a read-planning projection over the active untracked
    /// schema catalog. Validation must use `schema_facts_for_domain` instead so
    /// schema durability remains explicit.
    #[cfg(test)]
    pub(crate) async fn schema_jsons_for_sql_read_planning<R>(
        &self,
        hot_state: &R,
        branch_id: &str,
    ) -> Result<Vec<JsonValue>, LixError>
    where
        R: HotStateReader + ?Sized,
    {
        self.sql_read_schema_loads.fetch_add(1, Ordering::Relaxed);
        let facts = self
            .schema_facts_for_domain(hot_state, &Domain::schema_catalog(branch_id, true))
            .await?;
        let mut schemas = crate::schema::seed_schema_definitions()
            .into_iter()
            .map(|schema| {
                let key = crate::schema::schema_key_from_definition(schema)?;
                Ok((key.schema_key, schema.clone()))
            })
            .collect::<Result<BTreeMap<String, JsonValue>, LixError>>()?;
        for fact in facts {
            let schema_key = fact.catalog_key().schema_key.clone();
            if crate::schema::seed_schema_definition(&schema_key).is_some() {
                continue;
            }
            if schemas
                .insert(schema_key.clone(), fact.schema().clone())
                .is_some()
            {
                return Err(LixError::new(
                    LixError::CODE_SCHEMA_DEFINITION,
                    format!(
                        "SQL surface schema '{schema_key}' is visible from more than one schema catalog fact"
                    ),
                )
                .with_hint("SQL schema surfaces are named by schema_key. Keep exactly one visible schema per schema_key for SQL planning."));
            }
        }
        Ok(schemas.into_values().collect())
    }

    #[cfg(test)]
    pub(crate) fn sql_read_schema_load_count_for_test(&self) -> usize {
        self.sql_read_schema_loads.load(Ordering::Relaxed)
    }

    /// Loads schema facts reachable from a row domain for catalog tests.
    #[cfg(test)]
    pub(crate) async fn schema_facts_for_domain<R>(
        &self,
        hot_state: &R,
        domain: &Domain,
    ) -> Result<Vec<SchemaCatalogFact>, LixError>
    where
        R: HotStateReader + ?Sized,
    {
        let catalog_rows = scan_catalog_rows(hot_state, domain).await?;
        facts_from_catalog_rows(&catalog_rows)
    }
}

fn compiled_catalogs_by_rows()
-> &'static Mutex<HashMap<CatalogRowsFingerprint, Arc<CatalogSnapshot>>> {
    static CACHE: OnceLock<Mutex<HashMap<CatalogRowsFingerprint, Arc<CatalogSnapshot>>>> =
        OnceLock::new();
    CACHE.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Scans the raw registered-schema rows reachable from `domain`, without
/// decoding their snapshots.
struct CatalogRows {
    domains: Vec<CatalogDomainRows>,
}

impl CatalogRows {
    fn iter(&self) -> impl Iterator<Item = (&Domain, MaterializedHotStateRowRef<'_>)> {
        self.domains.iter().flat_map(|domain_rows| {
            domain_rows.rows.iter().filter_map(move |row| {
                row_belongs_to_schema_catalog_domain(row, &domain_rows.domain)
                    .then_some((&domain_rows.domain, row))
            })
        })
    }
}

struct CatalogDomainRows {
    domain: Domain,
    rows: MaterializedHotStateBatch,
}

async fn scan_catalog_rows<R>(hot_state: &R, domain: &Domain) -> Result<CatalogRows, LixError>
where
    R: HotStateReader + ?Sized,
{
    let schema_domains = domain.schema_catalog_domains();
    let mut catalog_rows = Vec::with_capacity(schema_domains.len());
    for schema_domain in schema_domains {
        let request = HotStateScanRequest {
            filter: HotStateFilter {
                schema_keys: vec![REGISTERED_SCHEMA_KEY.to_string()],
                branch_ids: vec![schema_domain.branch_id().to_string()],
                file_ids: vec![NullableKeyFilter::Null],
                untracked: Some(schema_domain.untracked()),
                include_tombstones: false,
                ..HotStateFilter::default()
            },
            projection: crate::hot_state::HotStateProjection {
                columns: vec!["raw_snapshot".to_owned()],
            },
            ..HotStateScanRequest::default()
        };
        let rows = if schema_domain.untracked() {
            hot_state.scan_batch(&request).await?
        } else {
            hot_state.scan_tracked_batch(&request).await?
        };
        catalog_rows.push(CatalogDomainRows {
            domain: schema_domain,
            rows,
        });
    }
    Ok(CatalogRows {
        domains: catalog_rows,
    })
}

fn facts_from_catalog_rows(catalog_rows: &CatalogRows) -> Result<Vec<SchemaCatalogFact>, LixError> {
    let row_count = catalog_rows
        .domains
        .iter()
        .map(|domain| domain.rows.len())
        .sum();
    let mut facts = Vec::with_capacity(row_count);
    for (schema_domain, row) in catalog_rows.iter() {
        let Some((key, schema)) = decode_registered_schema_row(row)? else {
            continue;
        };
        facts.push(SchemaCatalogFact::new(schema_domain.clone(), key, schema));
    }
    Ok(facts)
}

fn row_belongs_to_schema_catalog_domain(
    row: MaterializedHotStateRowRef<'_>,
    domain: &Domain,
) -> bool {
    row.schema_key() == REGISTERED_SCHEMA_KEY
        && row.file_id().is_none()
        && row.raw_snapshot().is_some()
        && row.branch_id() == domain.branch_id()
        && row.untracked() == domain.untracked()
        && committed_row_ref_is_exact_branch_scoped(row, domain.branch_id())
}

fn decode_registered_schema_row(
    row: MaterializedHotStateRowRef<'_>,
) -> Result<Option<(crate::schema::SchemaKey, JsonValue)>, LixError> {
    if row.schema_key() != REGISTERED_SCHEMA_KEY {
        return Err(LixError::new(
            "LIX_ERROR_UNKNOWN",
            format!(
                "expected lix_registered_schema row, got schema_key={}",
                row.schema_key()
            ),
        ));
    }

    let decoded;
    let typed = match row.decoded_snapshot() {
        Some(typed) => typed.as_ref(),
        None => {
            let Some(payload) = row.raw_snapshot() else {
                return Ok(None);
            };
            decoded = crate::plugin::runtime::WasmTypedRow::decode_durable_payload(
                Arc::from(payload.as_ref()),
                REGISTERED_SCHEMA_KEY,
                row.row_pk(),
            )?;
            &decoded
        }
    };
    let (_, plan) = CatalogSnapshot::builtin()
        .plan_for_key(REGISTERED_SCHEMA_KEY)
        .expect("embedded catalog contains lix_registered_schema");
    typed.validate_resolved_schema_binding(
        REGISTERED_SCHEMA_KEY,
        REGISTERED_SCHEMA_KEY,
        &plan.fingerprint().bytes(),
    )?;
    plan.compiled_schema
        .validate_complete_row(&typed.row)
        .map_err(|error| {
            LixError::new(
                LixError::CODE_SCHEMA_VALIDATION,
                format!("invalid typed registered schema row: {error}"),
            )
        })?;
    let stored_schema_key = match typed.row.get("schema_key") {
        Some(lix_schema::Value::Text(value)) => value,
        _ => {
            return Err(LixError::new(
                LixError::CODE_SCHEMA_VALIDATION,
                "typed registered schema row is missing schema_key",
            ));
        }
    };
    let schema = match typed.row.get("value") {
        // Cloning binary JSONB retains its shared byte owner. Consume that
        // cheap view directly into the catalog DOM; calling `as_value` first
        // would populate the row's lazy DOM and then deep-clone the schema.
        Some(lix_schema::Value::Jsonb(value)) => value.clone().into_value(),
        _ => {
            return Err(LixError::new(
                LixError::CODE_SCHEMA_VALIDATION,
                "typed registered schema row is missing schema value",
            ));
        }
    };
    let key = crate::schema::schema_key_from_definition(&schema)?;
    if key.schema_key != *stored_schema_key {
        return Err(LixError::new(
            LixError::CODE_SCHEMA_VALIDATION,
            "typed registered schema row key does not match its schema definition",
        ));
    }
    Ok(Some((key, schema)))
}

#[cfg(test)]
mod tests {
    use async_trait::async_trait;
    use serde_json::json;

    use super::*;
    use crate::GLOBAL_BRANCH_ID;
    use crate::changelog::ChangeId;
    use crate::common::LixTimestamp;
    use crate::hot_state::MaterializedHotStateRow;

    #[tokio::test]
    async fn compiled_catalog_for_domain_hits_cache_without_decoding() {
        let context = CatalogContext::new();
        let domain = Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", true);
        let reader = RowsHotStateReader::new(vec![
            registered_schema_row("alpha_schema"),
            registered_schema_row("beta_schema"),
        ]);

        let first = context
            .compiled_catalog_for_domain(&reader, &domain)
            .await
            .expect("catalog should compile");
        let second = context
            .compiled_catalog_for_domain(&reader, &domain)
            .await
            .expect("catalog should hit the raw-rows cache");
        assert!(
            Arc::ptr_eq(&first, &second),
            "identical raw rows must return the cached snapshot"
        );

        let changed_reader = RowsHotStateReader::new(vec![
            registered_schema_row("alpha_schema"),
            registered_schema_row("gamma_schema"),
        ]);
        let changed = context
            .compiled_catalog_for_domain(&changed_reader, &domain)
            .await
            .expect("changed catalog should compile");
        assert!(
            !Arc::ptr_eq(&first, &changed),
            "changed raw rows must compile a different snapshot"
        );
        assert!(changed.contains("gamma_schema"));
        assert!(!first.contains("gamma_schema"));
    }

    #[tokio::test]
    async fn transaction_opening_revision_skips_catalog_row_scans_until_it_changes() {
        let context = CatalogContext::new();
        let domain = Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", true);
        let revision = CatalogRevision::for_test(b"revision-one");
        let reader = RowsHotStateReader::new(vec![registered_schema_row("alpha_schema")]);

        let first = context
            .compiled_catalog_for_transaction_open(&reader, &domain, Some(&revision))
            .await
            .expect("opening catalog should compile");
        assert_eq!(
            reader.scan_count(),
            2,
            "cold open scans both durability scopes"
        );
        let second = context
            .compiled_catalog_for_transaction_open(&reader, &domain, Some(&revision))
            .await
            .expect("opening catalog should hit by revision");
        assert!(Arc::ptr_eq(&first, &second));
        assert_eq!(
            reader.scan_count(),
            2,
            "hot open must not rescan registered-schema rows"
        );

        let changed_reader = RowsHotStateReader::new(vec![registered_schema_row("beta_schema")]);
        let changed = context
            .compiled_catalog_for_transaction_open(
                &changed_reader,
                &domain,
                Some(&CatalogRevision::for_test(b"revision-two")),
            )
            .await
            .expect("changed revision should reload the catalog");
        assert_eq!(changed_reader.scan_count(), 2);
        assert!(changed.contains("beta_schema"));
        assert!(!changed.contains("alpha_schema"));
    }

    #[tokio::test]
    async fn transaction_opening_revision_caches_tracked_and_sql_catalogs_separately() {
        let context = CatalogContext::new();
        let sql_domain = Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", true);
        let tracked_domain = Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", false);
        let revision = CatalogRevision::for_test(b"revision-one");
        let reader = RowsHotStateReader::new(vec![registered_schema_row("alpha_schema")]);

        context
            .compiled_catalog_for_transaction_open(&reader, &sql_domain, Some(&revision))
            .await
            .expect("SQL catalog should compile");
        context
            .compiled_catalog_for_transaction_open(&reader, &tracked_domain, Some(&revision))
            .await
            .expect("tracked catalog should compile");
        assert_eq!(
            reader.scan_count(),
            3,
            "a cold SQL catalog scans two scopes and a cold tracked catalog scans one"
        );

        context
            .compiled_catalog_for_transaction_open(&reader, &sql_domain, Some(&revision))
            .await
            .expect("SQL catalog should hit by revision");
        context
            .compiled_catalog_for_transaction_open(&reader, &tracked_domain, Some(&revision))
            .await
            .expect("tracked catalog should hit by revision");
        assert_eq!(
            reader.scan_count(),
            3,
            "hot transaction opens must not rescan either schema catalog"
        );
    }

    #[tokio::test]
    async fn missing_transaction_opening_revision_conservatively_rescans() {
        let context = CatalogContext::new();
        let domain = Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", true);
        let reader = RowsHotStateReader::new(vec![registered_schema_row("alpha_schema")]);

        context
            .compiled_catalog_for_transaction_open(&reader, &domain, None)
            .await
            .expect("first opening catalog should compile");
        context
            .compiled_catalog_for_transaction_open(&reader, &domain, None)
            .await
            .expect("second opening catalog should compile");

        assert_eq!(reader.scan_count(), 4);
    }

    #[test]
    fn compiled_catalog_cache_shares_snapshots_for_equal_facts() {
        let context = CatalogContext::new();
        let parent = catalog_fact("parent_schema");
        let child = catalog_fact("child_schema");

        let first = context
            .compiled_catalog_for_facts(&[parent.clone(), child.clone()])
            .expect("catalog should compile");
        let reordered = context
            .compiled_catalog_for_facts(&[child, parent.clone()])
            .expect("catalog should compile");
        let different = context
            .compiled_catalog_for_facts(&[parent])
            .expect("catalog should compile");

        assert!(
            Arc::ptr_eq(&first, &reordered),
            "equal facts in any order must hit the same cached snapshot"
        );
        assert!(
            !Arc::ptr_eq(&first, &different),
            "different facts must compile a different snapshot"
        );
    }

    fn catalog_fact(schema_key: &str) -> SchemaCatalogFact {
        SchemaCatalogFact::new(
            Domain::schema_catalog("main", false),
            crate::schema::SchemaKey::new(schema_key),
            json!({
                "$schema": "https://lix.dev/schema-v1.json",
                "key": schema_key,
                "columns": [{ "name": "id", "type": "text", "nullable": false }],
                "primary_key": ["id"]
            }),
        )
    }

    #[tokio::test]
    async fn visible_schemas_are_loaded_from_registered_schema_rows() {
        let context = CatalogContext::new();

        let schemas = context
            .schema_jsons_for_sql_read_planning(
                &RowsHotStateReader::new(vec![
                    registered_schema_row("lix_registered_schema"),
                    registered_schema_row("lix_key_value"),
                ]),
                "ffffffff-ffff-7fff-bfff-ffffffffffff",
            )
            .await
            .expect("schema visibility should load");

        assert!(schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("lix_registered_schema")
        }));
        assert!(schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("lix_key_value")
        }));
    }

    #[tokio::test]
    async fn compiled_catalog_projects_the_same_sql_visible_schemas() {
        let context = CatalogContext::new();
        let mut tracked = registered_schema_row("zeta_tracked_schema");
        tracked.untracked = false;
        let reader = RowsHotStateReader::new(vec![
            registered_schema_row("alpha_untracked_schema"),
            tracked,
        ]);
        let domain = Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", true);

        let durable_projection = context
            .schema_jsons_for_sql_read_planning(&reader, "ffffffff-ffff-7fff-bfff-ffffffffffff")
            .await
            .expect("SQL schema visibility should load");
        let compiled_projection = context
            .compiled_catalog_for_domain(&reader, &domain)
            .await
            .expect("catalog should compile")
            .schema_jsons();

        assert_eq!(compiled_projection, durable_projection);
    }

    #[tokio::test]
    async fn visible_schemas_include_registered_schema_rows() {
        let context = CatalogContext::new();

        let schemas = context
            .schema_jsons_for_sql_read_planning(
                &RowsHotStateReader::new(vec![registered_schema_row("engine_dynamic_schema")]),
                "ffffffff-ffff-7fff-bfff-ffffffffffff",
            )
            .await
            .expect("schema visibility should load");

        assert!(schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("engine_dynamic_schema")
        }));
    }

    #[tokio::test]
    async fn sql_read_planning_rejects_multiple_visible_schemas_for_same_surface() {
        let context = CatalogContext::new();
        let error = context
            .schema_jsons_for_sql_read_planning(
                &RowsHotStateReader::new(vec![
                    registered_schema_row("engine_dynamic_schema"),
                    registered_schema_row("engine_dynamic_schema"),
                ]),
                "ffffffff-ffff-7fff-bfff-ffffffffffff",
            )
            .await
            .expect_err("SQL surfaces must not choose a schema identity implicitly");

        assert_eq!(error.code, LixError::CODE_SCHEMA_DEFINITION);
        assert!(error.message.contains("SQL surface schema"));
    }

    #[tokio::test]
    async fn tracked_domain_sees_tracked_seed_schemas_but_not_user_untracked_schemas() {
        let context = CatalogContext::new();
        let mut seed_schema = registered_schema_row("lix_key_value");
        seed_schema.untracked = false;

        let facts = context
            .schema_facts_for_domain(
                &RowsHotStateReader::new(vec![
                    seed_schema,
                    registered_schema_row("engine_dynamic_schema"),
                ]),
                &Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", false),
            )
            .await
            .expect("schema visibility should load");
        let schemas = facts
            .iter()
            .map(SchemaCatalogFact::schema)
            .collect::<Vec<_>>();

        assert!(schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("lix_key_value")
        }));
        assert!(!schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("engine_dynamic_schema")
        }));
    }

    #[tokio::test]
    async fn tracked_domain_does_not_see_untracked_seed_schemas() {
        let context = CatalogContext::new();

        let facts = context
            .schema_facts_for_domain(
                &RowsHotStateReader::new(vec![registered_schema_row("lix_key_value")]),
                &Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", false),
            )
            .await
            .expect("schema visibility should load");
        let schemas = facts
            .iter()
            .map(SchemaCatalogFact::schema)
            .collect::<Vec<_>>();

        assert!(!schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("lix_key_value")
        }));
    }

    #[tokio::test]
    async fn visible_schemas_ignore_projected_global_schema_rows_for_branch_scope() {
        let context = CatalogContext::new();
        let mut global_only = registered_schema_row("global_only_schema");
        global_only.global = true;
        global_only.branch_id = "main".into();

        let schemas = context
            .schema_jsons_for_sql_read_planning(&RowsHotStateReader::new(vec![global_only]), "main")
            .await
            .expect("schema visibility should load");

        assert!(schemas.iter().all(|schema| {
            schema.get("key").and_then(JsonValue::as_str) != Some("global_only_schema")
        }));
        assert!(schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("lix_file_descriptor")
        }));
    }

    #[tokio::test]
    async fn schema_facts_post_filter_non_catalog_rows_even_if_reader_returns_them() {
        let context = CatalogContext::new();
        let valid_schema = registered_schema_row("valid_schema");
        let mut file_scoped_schema = registered_schema_row("file_scoped_schema");
        file_scoped_schema.file_id = Some("01920000-0000-7000-8000-0000000000a2".to_string());
        let mut tombstoned_schema = registered_schema_row("tombstoned_schema");
        tombstoned_schema.snapshot_content = None;

        let facts = context
            .schema_facts_for_domain(
                &RowsHotStateReader::new(vec![valid_schema, file_scoped_schema, tombstoned_schema]),
                &Domain::schema_catalog("ffffffff-ffff-7fff-bfff-ffffffffffff", true),
            )
            .await
            .expect("schema facts should load");
        let schema_keys = facts
            .iter()
            .filter_map(|fact| fact.schema().get("key").and_then(JsonValue::as_str))
            .collect::<Vec<_>>();

        assert_eq!(schema_keys, vec!["valid_schema"]);
    }

    #[tokio::test]
    async fn visible_schemas_still_include_engine_builtins_when_no_rows_are_visible() {
        let context = CatalogContext::new();

        let schemas = context
            .schema_jsons_for_sql_read_planning(
                &RowsHotStateReader::new(Vec::new()),
                "ffffffff-ffff-7fff-bfff-ffffffffffff",
            )
            .await
            .expect("schema visibility should load");

        assert_eq!(schemas.len(), crate::schema::seed_schema_definitions().len());
        assert!(schemas.iter().any(|schema| {
            schema.get("key").and_then(JsonValue::as_str) == Some("lix_file_descriptor")
        }));
    }

    struct RowsHotStateReader {
        rows: Vec<MaterializedHotStateRow>,
        scan_count: AtomicUsize,
    }

    impl RowsHotStateReader {
        fn new(rows: Vec<MaterializedHotStateRow>) -> Self {
            Self {
                rows,
                scan_count: AtomicUsize::new(0),
            }
        }

        fn scan_count(&self) -> usize {
            self.scan_count.load(Ordering::Relaxed)
        }
    }

    #[async_trait]
    impl HotStateReader for RowsHotStateReader {
        async fn load_exact_batch(
            &self,
            request: &crate::hot_state::HotStateExactBatchRequest,
        ) -> Result<crate::hot_state::MaterializedHotStateExactBatch, LixError> {
            crate::hot_state::load_exact_batch_via_scan_for_test(self, request).await
        }

        async fn scan_batch(
            &self,
            request: &HotStateScanRequest,
        ) -> Result<MaterializedHotStateBatch, LixError> {
            self.scan_count.fetch_add(1, Ordering::Relaxed);
            let rows = self
                .rows
                .iter()
                .filter(|row| {
                    request.filter.schema_keys.is_empty()
                        || request.filter.schema_keys.contains(&row.schema_key)
                })
                .filter(|row| {
                    request.filter.branch_ids.is_empty()
                        || request
                            .filter
                            .branch_ids
                            .iter()
                            .any(|branch_id| branch_id.as_str() == row.branch_id.as_ref())
                })
                .filter(|row| {
                    request
                        .filter
                        .untracked
                        .is_none_or(|untracked| row.untracked == untracked)
                })
                .cloned()
                .collect::<Vec<_>>();
            let mut builder =
                crate::hot_state::MaterializedHotStateBatchBuilder::with_capacity(rows.len());
            for row in rows {
                let typed = row.snapshot_content.as_deref().and_then(|snapshot| {
                    let value = serde_json::from_str(snapshot).ok()?;
                    crate::plugin::runtime::WasmTypedRow::from_builtin_json(
                        &row.schema_key,
                        &row.row_pk,
                        &value,
                    )
                    .ok()
                    .map(Arc::new)
                });
                let ordinal = builder.len();
                builder.push_owned(row);
                let raw = typed.as_ref().and_then(|typed| {
                    typed
                        .durable_payload()
                        .ok()
                        .map(|payload| bytes::Bytes::copy_from_slice(&payload))
                });
                builder.set_decoded_snapshot(ordinal, typed);
                builder.set_raw_snapshot(ordinal, raw);
            }
            Ok(builder.finish())
        }
    }

    fn registered_schema_row(schema_key: &str) -> MaterializedHotStateRow {
        MaterializedHotStateRow {
            row_pk: registered_schema_row_pk(schema_key),
            file_id: None,
            schema_key: REGISTERED_SCHEMA_KEY.to_string(),
            branch_id: GLOBAL_BRANCH_ID.into(),
            metadata: None,
            deleted: false,
            change_id: Some(ChangeId::for_test_label("change-registered-schema")),
            commit_id: None,
            global: true,
            untracked: true,
            created_at: LixTimestamp::expect_parse(
                "registered schema test created_at",
                "2026-04-23T00:00:00Z",
            ),
            updated_at: LixTimestamp::expect_parse(
                "registered schema test updated_at",
                "2026-04-23T01:00:00Z",
            ),
            snapshot_content: Some(
                json!({
                    "schema_key": schema_key,
                    "value": {
                        "$schema": "https://lix.dev/schema-v1.json",
                        "key": schema_key,
                        "columns": [{ "name": "id", "type": "text", "nullable": false }],
                        "primary_key": ["id"]
                    }
                })
                .to_string()
                .into(),
            ),
        }
    }

    fn registered_schema_row_pk(schema_key: &str) -> crate::row_pk::RowPk {
        crate::row_pk::RowPk::from_primary_key_paths(
            &json!({ "schema_key": schema_key }),
            &[vec!["schema_key".to_string()]],
        )
        .expect("registered schema identity should derive")
    }
}