use super::{
    AedbConfig, AedbError, AedbErrorCode, AedbInstance, CallerContext, ColumnDef, ColumnType,
    ConsistencyMode, DdlOperation, DurabilityMode, Expr, IdempotencyKey, Mutation, Permission,
    Query, QueryError, QueryOptions, RECOVERY_CACHE_TTL, RecoveryCache, RecoveryMode, Row,
    TransactionEnvelope, Value, WriteClass, WriteIntent, create_table, mk_recovery_view,
};
use std::fs;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::tempdir;

#[test]
fn recovery_cache_refresh_keeps_recent_entry() {
    let mut cache = RecoveryCache::default();
    for seq in 1..=16 {
        cache.put(seq, mk_recovery_view(seq));
    }
    assert!(
        cache.get(1).is_some(),
        "first entry should exist before refresh"
    );

    cache.put(17, mk_recovery_view(17));

    assert!(
        cache.get(1).is_some(),
        "refreshing an entry must protect it from immediate eviction"
    );
    assert!(
        cache.get(2).is_none(),
        "oldest non-refreshed entry should be evicted first"
    );
}

#[test]
fn recovery_cache_prunes_expired_entries() {
    let mut cache = RecoveryCache::default();
    cache.put(7, mk_recovery_view(7));
    {
        let entry = cache.entries.get_mut(&7).expect("entry");
        entry.created = Instant::now() - RECOVERY_CACHE_TTL - Duration::from_secs(1);
    }
    cache.prune_expired();
    assert!(cache.get(7).is_none(), "expired entry must be removed");
}

#[tokio::test]
async fn existence_and_introspection_apis_report_catalog_state() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    db.create_project("p").await.expect("project");
    db.create_scope("p", "s1").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "s1".into(),
        table_name: "users".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    db.commit(Mutation::Ddl(DdlOperation::CreateIndex {
        project_id: "p".into(),
        scope_id: "s1".into(),
        table_name: "users".into(),
        index_name: "idx_users_name".into(),
        if_not_exists: false,
        columns: vec!["name".into()],
        index_type: crate::catalog::schema::IndexType::BTree,
        partial_filter: None,
    }))
    .await
    .expect("index");

    assert!(db.project_exists("p").await.expect("project_exists"));
    assert!(db.scope_exists("p", "s1").await.expect("scope_exists"));
    assert!(
        db.table_exists("p", "s1", "users")
            .await
            .expect("table_exists")
    );
    assert!(
        db.index_exists("p", "s1", "users", "idx_users_name")
            .await
            .expect("index_exists")
    );

    let projects = db.list_projects().await.expect("list projects");
    assert!(projects.iter().any(|p| p.project_id == "p"));

    let scopes = db.list_scopes_info("p").await.expect("list scopes");
    let scope = scopes
        .iter()
        .find(|s| s.scope_id == "s1")
        .expect("scope info");
    assert_eq!(scope.table_count, 1);

    let tables = db.list_tables_info("p", "s1").await.expect("list tables");
    let table = tables
        .iter()
        .find(|t| t.table_name == "users")
        .expect("table info");
    assert_eq!(table.column_count, 2);
    assert_eq!(table.index_count, 1);
}

#[tokio::test]
async fn dependency_aware_ddl_batch_reorders_create_dependencies() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    let batch = db
        .commit_ddl_batch_dependency_aware(vec![
            DdlOperation::CreateIndex {
                project_id: "arcana".into(),
                scope_id: "ops".into(),
                table_name: "users".into(),
                index_name: "idx_users_name".into(),
                if_not_exists: true,
                columns: vec!["name".into()],
                index_type: crate::catalog::schema::IndexType::BTree,
                partial_filter: None,
            },
            DdlOperation::CreateTable {
                owner_id: None,
                if_not_exists: true,
                project_id: "arcana".into(),
                scope_id: "ops".into(),
                table_name: "users".into(),
                columns: vec![
                    ColumnDef {
                        name: "id".into(),
                        col_type: ColumnType::Integer,
                        nullable: false,
                    },
                    ColumnDef {
                        name: "name".into(),
                        col_type: ColumnType::Text,
                        nullable: false,
                    },
                ],
                primary_key: vec!["id".into()],
            },
            DdlOperation::CreateScope {
                owner_id: None,
                project_id: "arcana".into(),
                scope_id: "ops".into(),
                if_not_exists: true,
            },
            DdlOperation::CreateProject {
                owner_id: None,
                project_id: "arcana".into(),
                if_not_exists: true,
            },
        ])
        .await
        .expect("dependency-aware batch");

    assert_eq!(batch.results.len(), 4);
    assert!(
        db.index_exists("arcana", "ops", "users", "idx_users_name")
            .await
            .expect("index exists")
    );
}

#[tokio::test]
async fn ddl_errors_expose_stable_error_codes() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");

    let err = db
        .commit(Mutation::Ddl(DdlOperation::CreateIndex {
            project_id: "arcana".into(),
            scope_id: "app".into(),
            table_name: "missing_users".into(),
            index_name: "idx_missing".into(),
            if_not_exists: false,
            columns: vec!["name".into()],
            index_type: crate::catalog::schema::IndexType::BTree,
            partial_filter: None,
        }))
        .await
        .expect_err("create index on missing table should fail");

    assert_eq!(err.code(), AedbErrorCode::TableNotFound);
}

#[tokio::test]
async fn snapshot_limit_enforced_on_read_path() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(
        AedbConfig {
            max_concurrent_snapshots: 1,
            ..AedbConfig::default()
        },
        dir.path(),
    )
    .expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: None,
        },
    }))
    .await
    .expect("grant");
    let caller = CallerContext::new("alice");

    let handle = {
        let mut mgr = db.snapshot_manager.lock();
        mgr.acquire_bounded(
            crate::snapshot::reader::SnapshotReadView {
                keyspace: Arc::new(crate::storage::keyspace::Keyspace::default().snapshot()),
                catalog: Arc::new(crate::catalog::Catalog::default()),
                seq: 0,
            },
            1,
        )
        .expect("occupy")
    };

    let err = db
        .kv_get("p", "app", b"k", ConsistencyMode::AtLatest, &caller)
        .await
        .expect_err("snapshot cap");
    assert!(matches!(
        err,
        crate::query::error::QueryError::SnapshotLimitReached
    ));

    let mut mgr = db.snapshot_manager.lock();
    mgr.release(handle);
    let _ = mgr.gc();
}

#[tokio::test]
async fn idempotent_retry_does_not_double_apply_non_idempotent_mutation() {
    let dir = tempdir().expect("temp");
    let db =
        Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    let key = IdempotencyKey([42u8; 16]);
    let mut tasks = Vec::new();
    for _ in 0..8 {
        let db = Arc::clone(&db);
        let key = key.clone();
        tasks.push(tokio::spawn(async move {
            db.commit_envelope(TransactionEnvelope {
                caller: None,
                idempotency_key: Some(key),
                write_class: WriteClass::Standard,
                assertions: Vec::new(),
                read_set: Default::default(),
                write_intent: WriteIntent {
                    mutations: vec![Mutation::KvIncU256 {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: b"idem-counter".to_vec(),
                        amount_be: {
                            let mut out = [0u8; 32];
                            out[31] = 1;
                            out
                        },
                    }],
                },
                base_seq: 0,
            })
            .await
            .expect("idempotent commit")
        }));
    }

    let mut seqs = std::collections::BTreeSet::new();
    let mut outcomes = Vec::new();
    for t in tasks {
        let res = t.await.expect("join");
        seqs.insert(res.commit_seq);
        outcomes.push(res.idempotency);
    }
    assert_eq!(seqs.len(), 1, "all retries must resolve to one commit_seq");
    assert!(
        outcomes
            .iter()
            .any(|o| matches!(o, crate::commit::executor::IdempotencyOutcome::Duplicate)),
        "at least one retry should report duplicate outcome"
    );

    let entry = db
        .kv_get_no_auth("p", "app", b"idem-counter", ConsistencyMode::AtLatest)
        .await
        .expect("kv_get")
        .expect("counter exists");
    assert_eq!(
        primitive_types::U256::from_big_endian(&entry.value),
        primitive_types::U256::one(),
        "idempotent retries must apply mutation exactly once"
    );
}

#[tokio::test]
#[ignore = "manual perf probe: durability knob sweep (batch/coalescing)"]
async fn benchmark_durability_knob_sweep() {
    fn percentile(sorted: &[u128], p: f64) -> u128 {
        if sorted.is_empty() {
            return 0;
        }
        let percentile_index = ((sorted.len() as f64 - 1.0) * p).round() as usize;
        sorted[percentile_index.min(sorted.len() - 1)]
    }

    #[derive(Clone)]
    struct Profile {
        name: &'static str,
        batch_interval_ms: u64,
        batch_max_bytes: usize,
        coalesce_enabled: bool,
        coalesce_window_us: u64,
    }

    async fn run_profile(profile: &Profile) -> (f64, u128, u128, u64, u64) {
        let dir = tempdir().expect("temp");
        let mut config = AedbConfig {
            durability_mode: DurabilityMode::Batch,
            batch_interval_ms: profile.batch_interval_ms,
            batch_max_bytes: profile.batch_max_bytes,
            recovery_mode: RecoveryMode::Permissive,
            hash_chain_required: false,
            durable_ack_coalescing_enabled: profile.coalesce_enabled,
            durable_ack_coalesce_window_us: profile.coalesce_window_us,
            ..AedbConfig::default()
        };
        config.manifest_hmac_key = None;
        let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
        db.create_project("p").await.expect("project");

        for i in 0..8_000usize {
            db.commit(Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: format!("sweep-seed:{i:05}").into_bytes(),
                value: vec![b's'; 128],
            })
            .await
            .expect("seed");
        }

        let workers = 8usize;
        let commits_per_worker = 500usize;
        let started = Instant::now();
        let mut tasks = Vec::with_capacity(workers);
        for worker in 0..workers {
            let db = Arc::clone(&db);
            tasks.push(tokio::spawn(async move {
                let mut lats = Vec::with_capacity(commits_per_worker);
                for i in 0..commits_per_worker {
                    let t0 = Instant::now();
                    db.commit(Mutation::KvSet {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: format!("sweep:{worker:02}:{i:06}").into_bytes(),
                        value: vec![b'x'; 256],
                    })
                    .await
                    .expect("commit");
                    lats.push(t0.elapsed().as_micros());
                }
                lats
            }));
        }

        let mut all_lat = Vec::with_capacity(workers * commits_per_worker);
        for task in tasks {
            let mut lats = task.await.expect("worker join");
            all_lat.append(&mut lats);
        }
        all_lat.sort_unstable();

        let elapsed = started.elapsed().as_secs_f64().max(0.000_001);
        let tps = (workers * commits_per_worker) as f64 / elapsed;
        let p50 = percentile(&all_lat, 0.50);
        let p99 = percentile(&all_lat, 0.99);
        let op = db.operational_metrics().await;
        (tps, p50, p99, op.wal_sync_ops, op.avg_wal_sync_micros)
    }

    let profiles = vec![
        Profile {
            name: "baseline_10ms_1mb_no_coalesce",
            batch_interval_ms: 10,
            batch_max_bytes: 1024 * 1024,
            coalesce_enabled: false,
            coalesce_window_us: 0,
        },
        Profile {
            name: "trial_20ms_4mb_coalesce_1000us",
            batch_interval_ms: 20,
            batch_max_bytes: 4 * 1024 * 1024,
            coalesce_enabled: true,
            coalesce_window_us: 1000,
        },
        Profile {
            name: "trial_20ms_8mb_coalesce_1500us",
            batch_interval_ms: 20,
            batch_max_bytes: 8 * 1024 * 1024,
            coalesce_enabled: true,
            coalesce_window_us: 1500,
        },
        Profile {
            name: "trial_40ms_8mb_coalesce_1500us",
            batch_interval_ms: 40,
            batch_max_bytes: 8 * 1024 * 1024,
            coalesce_enabled: true,
            coalesce_window_us: 1500,
        },
    ];

    for profile in &profiles {
        let (tps, p50, p99, wal_sync_ops, avg_wal_sync_us) = run_profile(profile).await;
        eprintln!(
            "durability_sweep: profile={} tps={:.2} p50_us={} p99_us={} wal_sync_ops={} avg_wal_sync_us={}",
            profile.name, tps, p50, p99, wal_sync_ops, avg_wal_sync_us
        );
    }
}

#[tokio::test]
async fn strict_open_rejects_directory_previously_opened_in_non_strict_mode() {
    let dir = tempdir().expect("temp");
    let mut permissive = AedbConfig::production([7u8; 32]);
    permissive.recovery_mode = RecoveryMode::Permissive;
    permissive.hash_chain_required = false;

    let db = AedbInstance::open_anonymous(permissive, dir.path()).expect("open permissive");
    db.shutdown().await.expect("shutdown permissive");
    drop(db); // release the data-directory lock before reopening

    let strict = AedbConfig::production([7u8; 32]);
    let err = match AedbInstance::open_anonymous(strict, dir.path()) {
        Ok(db) => {
            db.shutdown().await.expect("shutdown unexpected strict db");
            panic!("strict open should fail closed");
        }
        Err(err) => err,
    };
    assert!(
        matches!(err, AedbError::Validation(ref msg) if msg.contains("strict open denied")),
        "unexpected error: {err}"
    );
}

#[tokio::test]
async fn strict_open_rejects_tampered_trust_mode_marker() {
    let dir = tempdir().expect("temp");
    let mut permissive = AedbConfig::production([9u8; 32]);
    permissive.recovery_mode = RecoveryMode::Permissive;
    permissive.hash_chain_required = false;

    let db = AedbInstance::open_anonymous(permissive, dir.path()).expect("open permissive");
    db.shutdown().await.expect("shutdown permissive");
    drop(db); // release the data-directory lock before reopening

    fs::write(
        dir.path().join("trust_mode.json"),
        r#"{"ever_non_strict_recovery":false,"ever_hash_chain_disabled":false}"#,
    )
    .expect("tamper trust mode marker");

    let err = match AedbInstance::open_anonymous(AedbConfig::production([9u8; 32]), dir.path()) {
        Ok(db) => {
            db.shutdown().await.expect("shutdown unexpected strict db");
            panic!("tampered trust marker should fail closed");
        }
        Err(err) => err,
    };
    assert!(
        matches!(err, AedbError::IntegrityError { ref message } if message.contains("trust mode marker hmac mismatch")),
        "unexpected error: {err}"
    );
}

#[tokio::test]
async fn queries_reject_oversized_in_lists_and_like_patterns() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;

    let oversized_in = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id"]).from("items").where_(Expr::In(
                "id".into(),
                (0..10_001).map(Value::Integer).collect(),
            )),
            QueryOptions::default(),
        )
        .await
        .expect_err("oversized IN list should be rejected");
    assert!(
        matches!(oversized_in, QueryError::InvalidQuery { reason } if reason.contains("IN list"))
    );

    let oversized_like = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["name"])
                .from("items")
                .where_(Expr::Like("name".into(), "a".repeat(257))),
            QueryOptions::default(),
        )
        .await
        .expect_err("oversized LIKE should be rejected");
    assert!(
        matches!(oversized_like, QueryError::InvalidQuery { reason } if reason.contains("LIKE pattern"))
    );
}

#[tokio::test]
async fn managed_system_tables_reject_direct_user_mutations() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    let err = db
        .commit(Mutation::Upsert {
            project_id: crate::catalog::SYSTEM_PROJECT_ID.into(),
            scope_id: "app".into(),
            table_name: "reactive_processor_checkpoints".into(),
            primary_key: vec![Value::Text("processor".into())],
            row: Row {
                values: vec![
                    Value::Text("processor".into()),
                    Value::Integer(42),
                    Value::Timestamp(1),
                ],
            },
        })
        .await
        .expect_err("managed system tables must reject direct writes");

    assert!(
        matches!(err, AedbError::Validation(message) if message.contains("managed and read-only"))
    );
}

// ---------------------------------------------------------------------------
// Mutation::UpdateFields — primary-key-addressed partial-column update.
// ---------------------------------------------------------------------------

async fn open_with_entity_table(dir: &std::path::Path) -> AedbInstance {
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "ent",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "x".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "y".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;
    db
}

async fn read_ent(db: &AedbInstance, id: i64) -> Vec<Value> {
    let result = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id", "x", "y"])
                .from("ent")
                .where_(Expr::Eq("id".into(), Value::Integer(id))),
            QueryOptions::default(),
        )
        .await
        .expect("query");
    result.rows.into_iter().next().expect("row present").values
}

#[tokio::test]
async fn update_fields_overwrites_only_named_columns() {
    let dir = tempdir().expect("temp");
    let db = open_with_entity_table(dir.path()).await;
    db.commit(Mutation::Insert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row::from_values(vec![
            Value::Integer(1),
            Value::Integer(10),
            Value::Integer(20),
        ]),
    })
    .await
    .expect("insert");

    // Update only x; y must be preserved.
    db.commit(Mutation::UpdateFields {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        fields: vec![("x".into(), Value::Integer(99))],
    })
    .await
    .expect("update fields");

    assert_eq!(
        read_ent(&db, 1).await,
        vec![Value::Integer(1), Value::Integer(99), Value::Integer(20)]
    );
}

#[tokio::test]
async fn update_fields_maintains_secondary_index() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "ent",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;
    db.commit_ddl(DdlOperation::CreateIndex {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        index_name: "idx_ent_name".into(),
        if_not_exists: false,
        columns: vec!["name".into()],
        index_type: crate::catalog::schema::IndexType::BTree,
        partial_filter: None,
    })
    .await
    .expect("index");
    db.commit(Mutation::Insert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row::from_values(vec![Value::Integer(1), Value::Text("alice".into())]),
    })
    .await
    .expect("insert");

    db.commit(Mutation::UpdateFields {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        fields: vec![("name".into(), Value::Text("bob".into()))],
    })
    .await
    .expect("update fields");

    // Index lookup by the new value finds the row; the old value does not.
    let by_bob = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id"])
                .from("ent")
                .where_(Expr::Eq("name".into(), Value::Text("bob".into()))),
            QueryOptions::default(),
        )
        .await
        .expect("query bob");
    assert_eq!(by_bob.rows.len(), 1, "index must resolve the updated value");

    let by_alice = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id"])
                .from("ent")
                .where_(Expr::Eq("name".into(), Value::Text("alice".into()))),
            QueryOptions::default(),
        )
        .await
        .expect("query alice");
    assert_eq!(
        by_alice.rows.len(),
        0,
        "stale index entry for the old value must be gone"
    );
}

#[tokio::test]
async fn update_fields_rejects_primary_key_and_unknown_and_missing() {
    let dir = tempdir().expect("temp");
    let db = open_with_entity_table(dir.path()).await;
    db.commit(Mutation::Insert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row::from_values(vec![
            Value::Integer(1),
            Value::Integer(10),
            Value::Integer(20),
        ]),
    })
    .await
    .expect("insert");

    // Cannot rewrite a primary-key column.
    db.commit(Mutation::UpdateFields {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        fields: vec![("id".into(), Value::Integer(2))],
    })
    .await
    .expect_err("must reject primary-key column update");

    // Unknown column.
    db.commit(Mutation::UpdateFields {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(1)],
        fields: vec![("zzz".into(), Value::Integer(1))],
    })
    .await
    .expect_err("must reject unknown column");

    // Missing row.
    db.commit(Mutation::UpdateFields {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        primary_key: vec![Value::Integer(404)],
        fields: vec![("x".into(), Value::Integer(1))],
    })
    .await
    .expect_err("must reject update of a non-existent row");
}

#[tokio::test]
async fn update_where_leading_pk_prefix_scopes_to_one_instance() {
    // Mirrors the ECS shared-table layout: composite PK (instance_id, entity_id).
    // A keyed UpdateWhere(instance_id = X) must update only that instance's rows —
    // and (post-optimization) scan only its PK-prefix band, not the whole table.
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "ent",
        vec![
            ColumnDef {
                name: "instance_id".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "entity_id".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "val".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
        ],
        vec!["instance_id", "entity_id"],
    )
    .await;
    for (inst, ent) in [("a", "e1"), ("a", "e2"), ("b", "e1")] {
        db.commit(Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "ent".into(),
            primary_key: vec![Value::Text(inst.into()), Value::Text(ent.into())],
            row: Row::from_values(vec![
                Value::Text(inst.into()),
                Value::Text(ent.into()),
                Value::Integer(1),
            ]),
        })
        .await
        .expect("insert");
    }

    db.commit(Mutation::UpdateWhere {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "ent".into(),
        predicate: Expr::Eq("instance_id".into(), Value::Text("a".into())),
        updates: vec![("val".into(), Value::Integer(99))],
        limit: None,
    })
    .await
    .expect("update where");

    let rows = |inst: &str, ent: &str| {
        let db = &db;
        let inst = inst.to_string();
        let ent = ent.to_string();
        async move {
            db.query_no_auth(
                "p",
                "app",
                Query::select(&["val"]).from("ent").where_(
                    Expr::Eq("instance_id".into(), Value::Text(inst.into()))
                        .and(Expr::Eq("entity_id".into(), Value::Text(ent.into()))),
                ),
                QueryOptions::default(),
            )
            .await
            .expect("query")
            .rows
            .into_iter()
            .next()
            .expect("row")
            .values[0]
                .clone()
        }
    };

    assert_eq!(rows("a", "e1").await, Value::Integer(99));
    assert_eq!(rows("a", "e2").await, Value::Integer(99));
    // Instance "b" must be untouched — proves the update didn't spill outside the band.
    assert_eq!(rows("b", "e1").await, Value::Integer(1));
}
