Skip to main content

reddb_server/runtime/
impl_ddl.rs

1//! DDL execution: CREATE TABLE, DROP TABLE, ALTER TABLE via SQL AST
2//!
3//! Translates DDL statements into collection-level operations on the
4//! underlying `UnifiedStore`.  RedDB uses a flexible schema-on-read
5//! model, so column definitions are advisory metadata rather than
6//! rigid constraints.
7
8use super::*;
9use crate::catalog::CollectionModel;
10use crate::runtime::audit_log::{AuditAuthSource, AuditEvent, AuditFieldEscaper, Outcome};
11use crate::runtime::ddl::polymorphic_resolver;
12use reddb_rql::analyzer::{analyze_create_table, resolve_declared_data_type};
13use reddb_rql::ast::CreateColumnDef;
14use reddb_rql::ast::{
15    CreateVcsRefQuery, DropForkQuery, DropVcsRefQuery, ForkStoreQuery, VcsRefKind,
16};
17use std::collections::{BTreeSet, HashMap, HashSet};
18
19fn operational_wal_retention_floor(durable_lsn: u64) -> u64 {
20    if durable_lsn == 0 {
21        0
22    } else {
23        1
24    }
25}
26
27fn validate_store_fork_lsn(
28    fork_lsn: u64,
29    retention_floor: u64,
30    durable_lsn: u64,
31) -> RedDBResult<()> {
32    if fork_lsn < retention_floor {
33        return Err(RedDBError::Query(format!(
34            "cannot fork store at LSN {fork_lsn}: below operational WAL retention floor \
35             {retention_floor}; restore from backup for older recovery points"
36        )));
37    }
38    if fork_lsn > durable_lsn {
39        return Err(RedDBError::Query(format!(
40            "cannot fork store at LSN {fork_lsn}: current durable LSN is {durable_lsn}; \
41             restore from backup for recovery points outside the retained WAL window"
42        )));
43    }
44    Ok(())
45}
46
47fn vault_master_key_ref(collection: &str) -> String {
48    format!("red.vault.{collection}.master_key")
49}
50
51impl RedDBRuntime {
52    /// Execute CREATE TABLE
53    ///
54    /// Creates a new collection in the store.  Column definitions are
55    /// recorded for introspection but do not enforce rigid schema
56    /// constraints.
57    pub fn execute_create_table(
58        &self,
59        raw_query: &str,
60        query: &CreateTableQuery,
61    ) -> RedDBResult<RuntimeQueryResult> {
62        if query.collection_model != CollectionModel::Table {
63            if query
64                .columns
65                .iter()
66                .any(|column| column.generated.is_some() || column.check.is_some())
67            {
68                return Err(RedDBError::Query(
69                    "schema expressions currently require TABLE collections".to_string(),
70                ));
71            }
72            return self.execute_create_keyed_collection(raw_query, query);
73        }
74        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
75        let store = self.inner.db.store();
76        analyze_create_table(query).map_err(|err| RedDBError::Query(err.to_string()))?;
77        // Reject an AI policy that wires a provider/model to a modality it
78        // cannot serve — fail at DDL time, before the collection exists,
79        // using the capability matrix from #1269.
80        if let Some(policy) = &query.ai_policy {
81            validate_ai_policy_modalities(policy)?;
82        }
83        crate::reserved_fields::ensure_no_reserved_public_item_fields(
84            query.columns.iter().map(|column| column.name.as_str()),
85            &format!("table '{}'", query.name),
86        )?;
87        // Check if the collection already exists.
88        let exists = store.get_collection(&query.name).is_some();
89        if exists {
90            if query.if_not_exists {
91                return Ok(RuntimeQueryResult::ok_message(
92                    raw_query.to_string(),
93                    &format!("table '{}' already exists", query.name),
94                    "create",
95                ));
96            }
97            return Err(RedDBError::Query(format!(
98                "table '{}' already exists",
99                query.name
100            )));
101        }
102
103        // Build and validate the contract before mutating storage so invalid
104        // SQL types / duplicate columns do not leave partial side effects.
105        let contract = collection_contract_from_create_table(query)?;
106        crate::application::collection_contract_enforcer::validate_contract_expressions(&contract)?;
107        validate_event_subscriptions(self, &query.name, &contract.subscriptions)?;
108        // Create the collection.
109        store
110            .create_collection(&query.name)
111            .map_err(|err| RedDBError::Internal(err.to_string()))?;
112        for subscription in &contract.subscriptions {
113            ensure_event_target_queue(self, &subscription.target_queue)?;
114        }
115        if let Some(default_ttl_ms) = query.default_ttl_ms {
116            self.inner
117                .db
118                .set_collection_default_ttl_ms(&query.name, default_ttl_ms);
119        }
120        self.inner
121            .db
122            .save_collection_contract(contract)
123            .map_err(|err| RedDBError::Internal(err.to_string()))?;
124        if let Some(tenant_id) = crate::runtime::impl_core::current_tenant() {
125            store.set_config_tree(
126                &format!("red.collection_tenants.{}", query.name),
127                &crate::serde_json::Value::String(tenant_id),
128            );
129        }
130        self.inner
131            .db
132            .persist_metadata()
133            .map_err(|err| RedDBError::Internal(err.to_string()))?;
134        self.refresh_table_planner_stats(&query.name);
135        self.invalidate_result_cache();
136        // Issue #120 — feed the create into the schema-vocabulary
137        // reverse index so AskPipeline (#121) sees this collection.
138        let columns: Vec<String> = query.columns.iter().map(|col| col.name.clone()).collect();
139        self.schema_vocabulary_apply(
140            crate::runtime::schema_vocabulary::DdlEvent::CreateCollection {
141                collection: query.name.clone(),
142                columns,
143                type_tags: Vec::new(),
144                description: None,
145            },
146        );
147        // Partition metadata (Phase 2.2 PG parity).
148        //
149        // When the CREATE TABLE carries a `PARTITION BY RANGE|LIST|HASH (col)`
150        // clause, stamp the partition config into `red_config` under
151        // `partition.{table}.{by,column}`. Children are registered separately
152        // via `ALTER TABLE parent ATTACH PARTITION child ...`.
153        if let Some(spec) = &query.partition_by {
154            let kind_str = match spec.kind {
155                reddb_rql::ast::PartitionKind::Range => "range",
156                reddb_rql::ast::PartitionKind::List => "list",
157                reddb_rql::ast::PartitionKind::Hash => "hash",
158            };
159            store.set_config_tree(
160                &format!("partition.{}.by", query.name),
161                &crate::serde_json::Value::String(kind_str.to_string()),
162            );
163            store.set_config_tree(
164                &format!("partition.{}.column", query.name),
165                &crate::serde_json::Value::String(spec.column.clone()),
166            );
167        }
168
169        // Table-scoped multi-tenancy (Phase 2.5.4).
170        //
171        // `CREATE TABLE t (...) TENANT BY (col)` declaration:
172        //   1. Persists the `tenant_tables.{table}.column` marker so
173        //      INSERTs can auto-fill and future opens re-hydrate.
174        //   2. Registers the table in the in-memory `tenant_tables`
175        //      HashMap used by the DML auto-fill path.
176        //   3. Installs an implicit RLS policy equivalent to
177        //      `USING (col = CURRENT_TENANT())` across all actions.
178        //   4. Flips `rls_enabled_tables` on so the policy applies.
179        if let Some(col) = &query.tenant_by {
180            store.set_config_tree(
181                &format!("tenant_tables.{}.column", query.name),
182                &crate::serde_json::Value::String(col.clone()),
183            );
184            self.register_tenant_table(&query.name, col);
185        }
186
187        let ttl_suffix = query
188            .default_ttl_ms
189            .map(|ttl_ms| format!(" with default TTL {}ms", ttl_ms))
190            .unwrap_or_default();
191
192        let tenant_suffix = query
193            .tenant_by
194            .as_ref()
195            .map(|col| format!(" (tenant-scoped by {col})"))
196            .unwrap_or_default();
197
198        Ok(RuntimeQueryResult::ok_message(
199            raw_query.to_string(),
200            &format!(
201                "table '{}' created{}{}",
202                query.name, ttl_suffix, tenant_suffix
203            ),
204            "create",
205        ))
206    }
207
208    fn execute_create_keyed_collection(
209        &self,
210        raw_query: &str,
211        query: &CreateTableQuery,
212    ) -> RedDBResult<RuntimeQueryResult> {
213        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
214        if is_system_schema_name(&query.name) {
215            return Err(RedDBError::Query("system schema is read-only".to_string()));
216        }
217        let store = self.inner.db.store();
218        let label = polymorphic_resolver::model_name(query.collection_model);
219        if store.get_collection(&query.name).is_some() {
220            if query.if_not_exists {
221                return Ok(RuntimeQueryResult::ok_message(
222                    raw_query.to_string(),
223                    &format!("{label} '{}' already exists", query.name),
224                    "create",
225                ));
226            }
227            return Err(RedDBError::Query(format!(
228                "{label} '{}' already exists",
229                query.name
230            )));
231        }
232
233        if query.collection_model == CollectionModel::Vault {
234            self.reject_vault_transaction()?;
235            self.provision_vault_key_material(&query.name, query.vault_own_master_key)?;
236        }
237        store
238            .create_collection(&query.name)
239            .map_err(|err| RedDBError::Internal(err.to_string()))?;
240        if query.collection_model == CollectionModel::Vault {
241            let key_scope = if query.vault_own_master_key {
242                "own"
243            } else {
244                "cluster"
245            };
246            store.set_config_tree(
247                &format!("red.vault.{}.key_scope", query.name),
248                &crate::serde_json::Value::String(key_scope.to_string()),
249            );
250            store.set_config_tree(
251                &format!("red.vault.{}.status", query.name),
252                &crate::serde_json::Value::String("sealed".to_string()),
253            );
254        }
255        if query.collection_model == CollectionModel::Metrics {
256            for spec in &query.metrics_rollup_policies {
257                let policy = crate::storage::timeseries::retention::DownsamplePolicy::parse(spec)
258                    .ok_or_else(|| {
259                    RedDBError::Query(format!("invalid metrics rollup policy '{}'", spec))
260                })?;
261                if policy.source != "raw" {
262                    return Err(RedDBError::Query(format!(
263                        "invalid metrics rollup policy '{}': metrics v0 rollups must use raw as source",
264                        spec
265                    )));
266                }
267                if !matches!(
268                    policy.aggregation.as_str(),
269                    "avg" | "sum" | "min" | "max" | "count"
270                ) {
271                    return Err(RedDBError::Query(format!(
272                        "invalid metrics rollup policy '{}': supported aggregations are avg, sum, min, max, count",
273                        spec
274                    )));
275                }
276            }
277            if let Some(raw_retention_ms) = query.default_ttl_ms {
278                self.inner
279                    .db
280                    .set_collection_default_ttl_ms(&query.name, raw_retention_ms);
281                store.set_config_tree(
282                    &format!("red.metrics.{}.raw_retention_ms", query.name),
283                    &crate::serde_json::Value::Number(raw_retention_ms as f64),
284                );
285            }
286            let tenant_identity = query
287                .tenant_by
288                .clone()
289                .unwrap_or_else(|| "current_tenant".to_string());
290            store.set_config_tree(
291                &format!("red.metrics.{}.tenant_identity", query.name),
292                &crate::serde_json::Value::String(tenant_identity),
293            );
294            store.set_config_tree(
295                &format!("red.metrics.{}.namespace", query.name),
296                &crate::serde_json::Value::String("default".to_string()),
297            );
298            if !query.metrics_rollup_policies.is_empty() {
299                store.set_config_tree(
300                    &format!("red.metrics.{}.rollup_policies", query.name),
301                    &crate::serde_json::Value::Array(
302                        query
303                            .metrics_rollup_policies
304                            .iter()
305                            .cloned()
306                            .map(crate::serde_json::Value::String)
307                            .collect(),
308                    ),
309                );
310            }
311        }
312        let contract = if query.collection_model == CollectionModel::Metrics {
313            metrics_collection_contract(query)
314        } else {
315            keyed_collection_contract(
316                &query.name,
317                query.collection_model,
318                query.analytics_config.clone(),
319            )
320        };
321        self.inner
322            .db
323            .save_collection_contract(contract)
324            .map_err(|err| RedDBError::Internal(err.to_string()))?;
325        if let Some(tenant_id) = crate::runtime::impl_core::current_tenant() {
326            store.set_config_tree(
327                &format!("red.collection_tenants.{}", query.name),
328                &crate::serde_json::Value::String(tenant_id),
329            );
330        }
331        self.inner
332            .db
333            .persist_metadata()
334            .map_err(|err| RedDBError::Internal(err.to_string()))?;
335        self.invalidate_result_cache();
336
337        Ok(RuntimeQueryResult::ok_message(
338            raw_query.to_string(),
339            &format!("{label} '{}' created", query.name),
340            "create",
341        ))
342    }
343
344    /// Bootstrap an internal graph collection declared `WITH ANALYTICS`
345    /// (issue #803). The SQL DDL path refuses `red.*` names via
346    /// [`is_system_schema_name`], so built-ins like `red.topology.cluster`
347    /// cannot be created through `CREATE GRAPH`; this method drives the same
348    /// contract-build / save / persist path directly, bypassing only that
349    /// guard. Idempotent: a no-op when the collection already exists (its
350    /// analytics contract is recovered from the WAL on restart), so it is safe
351    /// to call unconditionally on every boot.
352    pub fn ensure_system_graph_with_analytics(
353        &self,
354        name: &str,
355        outputs: &[crate::catalog::AnalyticsOutput],
356    ) -> RedDBResult<()> {
357        let store = self.inner.db.store();
358        if store.get_collection(name).is_some() {
359            return Ok(());
360        }
361        let analytics_config = outputs
362            .iter()
363            .map(|output| crate::catalog::AnalyticsViewDescriptor {
364                output: *output,
365                algorithm: None,
366                resolution: None,
367                max_iterations: None,
368                tolerance: None,
369            })
370            .collect();
371        store
372            .create_collection(name)
373            .map_err(|err| RedDBError::Internal(err.to_string()))?;
374        let contract = keyed_collection_contract(
375            name,
376            crate::catalog::CollectionModel::Graph,
377            analytics_config,
378        );
379        self.inner
380            .db
381            .save_collection_contract(contract)
382            .map_err(|err| RedDBError::Internal(err.to_string()))?;
383        self.inner
384            .db
385            .persist_metadata()
386            .map_err(|err| RedDBError::Internal(err.to_string()))?;
387        self.invalidate_result_cache_process_only();
388        Ok(())
389    }
390
391    pub fn execute_create_collection(
392        &self,
393        raw_query: &str,
394        query: &CreateCollectionQuery,
395    ) -> RedDBResult<RuntimeQueryResult> {
396        let model = match query.kind.as_str() {
397            "graph" => CollectionModel::Graph,
398            "document" => CollectionModel::Document,
399            "metrics" => CollectionModel::Metrics,
400            "vector.turbo" => {
401                let dimension = query.vector_dimension.ok_or_else(|| {
402                    RedDBError::Query(
403                        "CREATE COLLECTION KIND vector.turbo requires DIM".to_string(),
404                    )
405                })?;
406                let create = CreateVectorQuery {
407                    name: query.name.clone(),
408                    dimension,
409                    metric: query
410                        .vector_metric
411                        .unwrap_or(crate::storage::engine::distance::DistanceMetric::Cosine),
412                    if_not_exists: query.if_not_exists,
413                };
414                let result = self.execute_create_vector(raw_query, &create)?;
415                // Issue #693 — durable kind marker that distinguishes
416                // this collection from the legacy `vector` path on
417                // every restart. Runtime state (TurboQuantIndex +
418                // TurboExtent) is lazily materialised by
419                // `RedDB::turbo_state` on first INSERT/SEARCH so the
420                // marker is the only thing that needs to survive.
421                let store = self.inner.db.store();
422                crate::runtime::vector_turbo_kind::mark_as_turbo(&store, &query.name);
423                self.inner
424                    .db
425                    .persist_metadata()
426                    .map_err(|err| RedDBError::Internal(err.to_string()))?;
427                // Eagerly construct the state so the TurboExtent (if
428                // any) is reserved at creation time rather than on
429                // first INSERT. Errors are swallowed: the lazy path in
430                // `turbo_state` will retry on the next access.
431                let _ = self.inner.db.turbo_state(&query.name);
432                return Ok(result);
433            }
434            // KIND blockchain — issue #523 foundation: stored on top of a
435            // Table-shaped collection. The `chain` marker + reserved-column
436            // discipline make the difference. Schema validation, conflict
437            // retries, and `verify_chain` come in later iterations.
438            "blockchain" => CollectionModel::Table,
439            other => {
440                return Err(RedDBError::Query(format!(
441                    "NOT_YET_SUPPORTED: CREATE COLLECTION KIND {other} is not implemented"
442                )));
443            }
444        };
445        let create = CreateTableQuery {
446            collection_model: model,
447            name: query.name.clone(),
448            columns: Vec::new(),
449            unique_constraints: Vec::new(),
450            if_not_exists: query.if_not_exists,
451            default_ttl_ms: None,
452            metrics_rollup_policies: Vec::new(),
453            context_index_fields: Vec::new(),
454            context_index_enabled: false,
455            timestamps: false,
456            partition_by: None,
457            tenant_by: None,
458            append_only: false,
459            subscriptions: Vec::new(),
460            analytics_config: Vec::new(),
461            vault_own_master_key: false,
462            ai_policy: None,
463        };
464        let result = self.execute_create_table(raw_query, &create)?;
465        if query.kind == "blockchain" {
466            self.install_blockchain_kind(&query.name)?;
467        }
468        // Issue #522 — wire `SIGNED_BY (...)` into the runtime. The parser
469        // already produces a validated 32-byte pubkey list; installing
470        // the registry stamps the per-collection signer set into
471        // `red_config` so the INSERT path can load it cheaply.
472        if !query.allowed_signers.is_empty() {
473            let actor = crate::runtime::impl_core::current_user_projected()
474                .unwrap_or_else(|| "@system/create-collection".to_string());
475            crate::runtime::signed_writes_kind::install(
476                &self.inner.db.store(),
477                &query.name,
478                &query.allowed_signers,
479                &actor,
480            );
481        }
482        Ok(result)
483    }
484
485    /// Stamp `red.collection.{name}.kind = "chain"` and append the genesis
486    /// row. Idempotent against `IF NOT EXISTS`: if the collection already
487    /// has a row at height 0 we leave it.
488    fn install_blockchain_kind(&self, name: &str) -> RedDBResult<()> {
489        use crate::runtime::blockchain_kind;
490        use crate::storage::unified::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
491        use std::sync::Arc;
492
493        let store = self.inner.db.store();
494        blockchain_kind::mark_as_chain(&store, name);
495
496        let existing_tip = blockchain_kind::chain_tip(&store, name);
497        if existing_tip.height.is_some() {
498            return Ok(());
499        }
500
501        let fields = blockchain_kind::genesis_fields(blockchain_kind::now_ms());
502        let named: std::collections::HashMap<String, reddb_types::Value> =
503            fields.into_iter().collect();
504        let entity = UnifiedEntity::new(
505            EntityId::new(0),
506            EntityKind::TableRow {
507                table: Arc::from(name),
508                row_id: 0,
509            },
510            EntityData::Row(RowData {
511                columns: Vec::new(),
512                named: Some(named),
513                schema: None,
514            }),
515        );
516        store
517            .insert_auto(name, entity)
518            .map_err(|err| RedDBError::Internal(err.to_string()))?;
519        // #524: prime the in-memory tip cache so the chain-tip endpoint and
520        // subsequent INSERTs don't have to scan the collection to find genesis.
521        if let Some(tip) = blockchain_kind::chain_tip_full(&store, name) {
522            self.inner
523                .chain_tip_cache
524                .lock()
525                .insert(name.to_string(), tip);
526        }
527        Ok(())
528    }
529
530    pub fn execute_create_vector(
531        &self,
532        raw_query: &str,
533        query: &CreateVectorQuery,
534    ) -> RedDBResult<RuntimeQueryResult> {
535        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
536        if is_system_schema_name(&query.name) {
537            return Err(RedDBError::Query("system schema is read-only".to_string()));
538        }
539        let store = self.inner.db.store();
540        if store.get_collection(&query.name).is_some() {
541            if query.if_not_exists {
542                return Ok(RuntimeQueryResult::ok_message(
543                    raw_query.to_string(),
544                    &format!("vector '{}' already exists", query.name),
545                    "create",
546                ));
547            }
548            return Err(RedDBError::Query(format!(
549                "vector '{}' already exists",
550                query.name
551            )));
552        }
553
554        store
555            .create_collection(&query.name)
556            .map_err(|err| RedDBError::Internal(err.to_string()))?;
557        self.inner
558            .db
559            .save_collection_contract(vector_collection_contract(query))
560            .map_err(|err| RedDBError::Internal(err.to_string()))?;
561        if let Some(tenant_id) = crate::runtime::impl_core::current_tenant() {
562            store.set_config_tree(
563                &format!("red.collection_tenants.{}", query.name),
564                &crate::serde_json::Value::String(tenant_id),
565            );
566        }
567        // Slice G (#675) — register `vector.turbo` as the built-in
568        // baseline for every newly-created vector collection. The
569        // marker is the durable signal `RedDB::turbo_state` reads to
570        // materialise the TurboQuantIndex + TurboExtent on first
571        // INSERT/SEARCH. Pre-existing vector collections were created
572        // without the marker and stay on the brute-force fallback path
573        // until they are explicitly migrated — there is no implicit
574        // backfill in this slice.
575        crate::runtime::vector_turbo_kind::mark_as_turbo(&store, &query.name);
576        self.inner
577            .db
578            .persist_metadata()
579            .map_err(|err| RedDBError::Internal(err.to_string()))?;
580        // Eagerly materialise the per-collection turbo state so the
581        // TurboExtent (when the store is paged) is reserved at CREATE
582        // time rather than on the first INSERT. Failures are swallowed
583        // — the lazy path in `turbo_state` retries on next access.
584        let _ = self.inner.db.turbo_state(&query.name);
585        self.invalidate_result_cache();
586
587        Ok(RuntimeQueryResult::ok_message(
588            raw_query.to_string(),
589            &format!("vector '{}' created", query.name),
590            "create",
591        ))
592    }
593
594    fn provision_vault_key_material(
595        &self,
596        collection: &str,
597        own_master_key: bool,
598    ) -> RedDBResult<()> {
599        let auth_store = self.inner.auth_store.read().clone().ok_or_else(|| {
600            RedDBError::Query("CREATE VAULT requires an enabled, unsealed vault".to_string())
601        })?;
602        if !auth_store.is_vault_backed() {
603            return Err(RedDBError::Query(
604                "CREATE VAULT requires an enabled, unsealed vault".to_string(),
605            ));
606        }
607
608        auth_store
609            .ensure_vault_secret_key_result()
610            .map_err(|err| RedDBError::Query(err.to_string()))?;
611
612        if own_master_key
613            && auth_store
614                .vault_kv_get(&vault_master_key_ref(collection))
615                .is_none()
616        {
617            let key = crate::auth::store::random_bytes(32);
618            auth_store
619                .vault_kv_try_set(vault_master_key_ref(collection), hex::encode(key))
620                .map_err(|err| RedDBError::Query(err.to_string()))?;
621        }
622
623        Ok(())
624    }
625
626    /// Execute DROP TABLE
627    ///
628    /// Drops the collection and all its data from the store.
629    pub fn execute_drop_table(
630        &self,
631        raw_query: &str,
632        query: &DropTableQuery,
633    ) -> RedDBResult<RuntimeQueryResult> {
634        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
635        let store = self.inner.db.store();
636
637        if is_system_schema_name(&query.name) {
638            return Err(RedDBError::Query("system schema is read-only".to_string()));
639        }
640
641        let exists = store.get_collection(&query.name).is_some();
642        if !exists {
643            if query.if_exists {
644                return Ok(RuntimeQueryResult::ok_message(
645                    raw_query.to_string(),
646                    &format!("table '{}' does not exist", query.name),
647                    "drop",
648                ));
649            }
650            return Err(RedDBError::NotFound(format!(
651                "table '{}' not found",
652                query.name
653            )));
654        }
655        let actual =
656            polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
657        polymorphic_resolver::ensure_model_match(CollectionModel::Table, actual)?;
658
659        // Emit 1 collection_dropped event before storage is wiped.
660        // Queue is preserved; subscription is removed with the contract below.
661        let final_count = store
662            .get_collection(&query.name)
663            .map(|manager| manager.query_all(|_| true).len() as u64)
664            .unwrap_or(0);
665        crate::runtime::mutation::emit_collection_dropped_event_for_collection(
666            self,
667            &query.name,
668            final_count,
669        )?;
670
671        let topology_lock = self.inner.index_store.collection_topology_lock(&query.name);
672        let _topology_guard = topology_lock.write();
673        let orphaned_indices: Vec<String> = self
674            .inner
675            .index_store
676            .list_indices(&query.name)
677            .into_iter()
678            .map(|index| index.name)
679            .collect();
680        for name in &orphaned_indices {
681            self.inner.index_store.drop_index(name, &query.name);
682        }
683
684        store
685            .drop_collection(&query.name)
686            .map_err(|err| RedDBError::Internal(err.to_string()))?;
687        self.inner.db.invalidate_vector_index(&query.name);
688        self.inner.db.clear_collection_default_ttl_ms(&query.name);
689        self.inner
690            .db
691            .remove_collection_contract(&query.name)
692            .map_err(|err| RedDBError::Internal(err.to_string()))?;
693        self.clear_table_planner_stats(&query.name);
694        self.invalidate_result_cache();
695        // Issue #119: a dropped collection vanishes from every
696        // (tenant, role)'s visible-collections set. Auth is optional in
697        // embedded mode so guard the call.
698        if let Some(store) = self.inner.auth_store.read().clone() {
699            store.invalidate_visible_collections_cache();
700        }
701        self.inner
702            .db
703            .persist_metadata()
704            .map_err(|err| RedDBError::Internal(err.to_string()))?;
705        // Issue #120 — drop both the collection entries *and* every
706        // index entry that was scoped to this collection.  Dropping the
707        // collection wipes columns + collection-name + type-tags +
708        // index hits in one pass via `purge_collection_entries`, so
709        // the explicit `DropIndex` calls would be redundant.
710        self.schema_vocabulary_apply(
711            crate::runtime::schema_vocabulary::DdlEvent::DropCollection {
712                collection: query.name.clone(),
713            },
714        );
715
716        Ok(RuntimeQueryResult::ok_message(
717            raw_query.to_string(),
718            &format!("table '{}' dropped", query.name),
719            "drop",
720        ))
721    }
722
723    pub fn execute_drop_graph(
724        &self,
725        raw_query: &str,
726        query: &DropGraphQuery,
727    ) -> RedDBResult<RuntimeQueryResult> {
728        self.execute_drop_typed_collection(
729            raw_query,
730            &query.name,
731            query.if_exists,
732            CollectionModel::Graph,
733            "graph",
734        )
735    }
736
737    pub fn execute_drop_vector(
738        &self,
739        raw_query: &str,
740        query: &DropVectorQuery,
741    ) -> RedDBResult<RuntimeQueryResult> {
742        self.execute_drop_typed_collection(
743            raw_query,
744            &query.name,
745            query.if_exists,
746            CollectionModel::Vector,
747            "vector",
748        )
749    }
750
751    pub fn execute_drop_document(
752        &self,
753        raw_query: &str,
754        query: &DropDocumentQuery,
755    ) -> RedDBResult<RuntimeQueryResult> {
756        self.execute_drop_typed_collection(
757            raw_query,
758            &query.name,
759            query.if_exists,
760            CollectionModel::Document,
761            "document",
762        )
763    }
764
765    pub fn execute_drop_kv(
766        &self,
767        raw_query: &str,
768        query: &DropKvQuery,
769    ) -> RedDBResult<RuntimeQueryResult> {
770        let label = polymorphic_resolver::model_name(query.model);
771        self.execute_drop_typed_collection(
772            raw_query,
773            &query.name,
774            query.if_exists,
775            query.model,
776            label,
777        )
778    }
779
780    pub fn execute_drop_collection(
781        &self,
782        raw_query: &str,
783        query: &DropCollectionQuery,
784    ) -> RedDBResult<RuntimeQueryResult> {
785        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
786        if is_system_schema_name(&query.name) {
787            return Err(RedDBError::Query("system schema is read-only".to_string()));
788        }
789        let store = self.inner.db.store();
790        if store.get_collection(&query.name).is_none() {
791            if query.if_exists {
792                return Ok(RuntimeQueryResult::ok_message(
793                    raw_query.to_string(),
794                    &format!("collection '{}' does not exist", query.name),
795                    "drop",
796                ));
797            }
798            return Err(RedDBError::NotFound(format!(
799                "collection '{}' not found",
800                query.name
801            )));
802        }
803
804        let actual =
805            polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
806        if let Some(expected) = query.model {
807            polymorphic_resolver::ensure_model_match(expected, actual)?;
808        }
809
810        match actual {
811            CollectionModel::Table => self.execute_drop_table(
812                raw_query,
813                &DropTableQuery {
814                    name: query.name.clone(),
815                    if_exists: query.if_exists,
816                },
817            ),
818            CollectionModel::TimeSeries => self.execute_drop_timeseries(
819                raw_query,
820                &DropTimeSeriesQuery {
821                    name: query.name.clone(),
822                    if_exists: query.if_exists,
823                },
824            ),
825            CollectionModel::Queue => self.execute_drop_queue(
826                raw_query,
827                &DropQueueQuery {
828                    name: query.name.clone(),
829                    if_exists: query.if_exists,
830                },
831            ),
832            CollectionModel::Graph => self.execute_drop_graph(
833                raw_query,
834                &DropGraphQuery {
835                    name: query.name.clone(),
836                    if_exists: query.if_exists,
837                },
838            ),
839            CollectionModel::Vector => self.execute_drop_vector(
840                raw_query,
841                &DropVectorQuery {
842                    name: query.name.clone(),
843                    if_exists: query.if_exists,
844                },
845            ),
846            CollectionModel::Document => self.execute_drop_document(
847                raw_query,
848                &DropDocumentQuery {
849                    name: query.name.clone(),
850                    if_exists: query.if_exists,
851                },
852            ),
853            CollectionModel::Kv => self.execute_drop_kv(
854                raw_query,
855                &DropKvQuery {
856                    name: query.name.clone(),
857                    if_exists: query.if_exists,
858                    model: CollectionModel::Kv,
859                },
860            ),
861            CollectionModel::Config => self.execute_drop_kv(
862                raw_query,
863                &DropKvQuery {
864                    name: query.name.clone(),
865                    if_exists: query.if_exists,
866                    model: CollectionModel::Config,
867                },
868            ),
869            CollectionModel::Vault => self.execute_drop_kv(
870                raw_query,
871                &DropKvQuery {
872                    name: query.name.clone(),
873                    if_exists: query.if_exists,
874                    model: CollectionModel::Vault,
875                },
876            ),
877            CollectionModel::Hll => self.execute_probabilistic_command(
878                raw_query,
879                &ProbabilisticCommand::DropHll {
880                    name: query.name.clone(),
881                    if_exists: query.if_exists,
882                },
883            ),
884            CollectionModel::Sketch => self.execute_probabilistic_command(
885                raw_query,
886                &ProbabilisticCommand::DropSketch {
887                    name: query.name.clone(),
888                    if_exists: query.if_exists,
889                },
890            ),
891            CollectionModel::Filter => self.execute_probabilistic_command(
892                raw_query,
893                &ProbabilisticCommand::DropFilter {
894                    name: query.name.clone(),
895                    if_exists: query.if_exists,
896                },
897            ),
898            CollectionModel::Metrics => self.execute_drop_typed_collection(
899                raw_query,
900                &query.name,
901                query.if_exists,
902                CollectionModel::Metrics,
903                "metrics",
904            ),
905            CollectionModel::Mixed => self.execute_drop_typed_collection(
906                raw_query,
907                &query.name,
908                query.if_exists,
909                CollectionModel::Mixed,
910                "collection",
911            ),
912        }
913    }
914
915    /// Issue #801 — guard `ALTER GRAPH ... ADD|DROP ANALYTICS` so analytics
916    /// lifecycle mutations only land on a collection declared as a graph. The
917    /// `<graph>.<output>` resolver only consults `analytics_config` on
918    /// graph-model contracts, so allowing the op on any other model would write
919    /// dead config the reader can never surface.
920    fn require_graph_for_analytics(&self, name: &str) -> RedDBResult<()> {
921        let is_graph = self
922            .inner
923            .db
924            .collection_contract(name)
925            .map(|c| c.declared_model == crate::catalog::CollectionModel::Graph)
926            .unwrap_or(false);
927        if !is_graph {
928            return Err(RedDBError::Query(format!(
929                "ALTER GRAPH ... ANALYTICS: '{name}' is not a graph collection"
930            )));
931        }
932        Ok(())
933    }
934
935    /// Execute ALTER TABLE
936    ///
937    /// In RedDB's schema-on-read model, ALTER TABLE operations are advisory.
938    /// ADD COLUMN records the schema intent, DROP COLUMN removes it, and
939    /// RENAME COLUMN is a metadata rename.  Existing data is not rewritten.
940    pub fn execute_alter_table(
941        &self,
942        raw_query: &str,
943        query: &AlterTableQuery,
944    ) -> RedDBResult<RuntimeQueryResult> {
945        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
946        let expression_contract =
947            self.db()
948                .collection_contract(&query.name)
949                .is_some_and(|contract| {
950                    crate::application::collection_contract_enforcer::has_contract_expressions(
951                        &contract,
952                    )
953                });
954        if query.operations.iter().any(|operation| match operation {
955            AlterOperation::AddColumn(column) => {
956                expression_contract || column.generated.is_some() || column.check.is_some()
957            }
958            AlterOperation::DropColumn(_) | AlterOperation::RenameColumn { .. } => {
959                expression_contract
960            }
961            _ => false,
962        }) {
963            return Err(RedDBError::Query("ALTER of expression schemas requires validated backfill; create a new collection and import explicitly".to_string()));
964        }
965
966        let store = self.inner.db.store();
967
968        // Verify the table exists.
969        if store.get_collection(&query.name).is_none() {
970            return Err(RedDBError::NotFound(format!(
971                "table '{}' not found",
972                query.name
973            )));
974        }
975
976        let mut messages = Vec::new();
977
978        // Collect column-level changes upfront for schema-change event emission below.
979        let fields_added: Vec<String> = query
980            .operations
981            .iter()
982            .filter_map(|op| {
983                if let AlterOperation::AddColumn(col) = op {
984                    Some(col.name.clone())
985                } else {
986                    None
987                }
988            })
989            .collect();
990        let fields_removed: Vec<String> = query
991            .operations
992            .iter()
993            .filter_map(|op| {
994                if let AlterOperation::DropColumn(name) = op {
995                    Some(name.clone())
996                } else {
997                    None
998                }
999            })
1000            .collect();
1001
1002        for op in &query.operations {
1003            match op {
1004                AlterOperation::AddColumn(col) => {
1005                    // Schema-on-read: column will be available on next insert.
1006                    messages.push(format!("column '{}' added", col.name));
1007                }
1008                AlterOperation::DropColumn(name) => {
1009                    messages.push(format!("column '{}' dropped", name));
1010                }
1011                AlterOperation::RenameColumn { from, to } => {
1012                    messages.push(format!("column '{}' renamed to '{}'", from, to));
1013                }
1014                AlterOperation::AttachPartition { child, bound } => {
1015                    // Persist child → parent binding in red_config so the
1016                    // future planner-side pruner can enumerate children and
1017                    // evaluate their bounds.
1018                    store.set_config_tree(
1019                        &format!("partition.{}.children.{}", query.name, child),
1020                        &crate::serde_json::Value::String(bound.clone()),
1021                    );
1022                    messages.push(format!(
1023                        "partition '{child}' attached to '{}' ({bound})",
1024                        query.name
1025                    ));
1026                }
1027                AlterOperation::DetachPartition { child } => {
1028                    store.set_config_tree(
1029                        &format!("partition.{}.children.{}", query.name, child),
1030                        &crate::serde_json::Value::Null,
1031                    );
1032                    messages.push(format!(
1033                        "partition '{child}' detached from '{}'",
1034                        query.name
1035                    ));
1036                }
1037                AlterOperation::EnableRowLevelSecurity => {
1038                    self.inner
1039                        .rls_enabled_tables
1040                        .write()
1041                        .insert(query.name.clone());
1042                    // Persist flag so RLS survives restart via red_config.
1043                    store.set_config_tree(
1044                        &format!("rls.enabled.{}", query.name),
1045                        &crate::serde_json::Value::Bool(true),
1046                    );
1047                    self.invalidate_plan_cache();
1048                    messages.push(format!("row level security enabled on '{}'", query.name));
1049                    self.invalidate_result_cache();
1050                }
1051                AlterOperation::DisableRowLevelSecurity => {
1052                    self.inner.rls_enabled_tables.write().remove(&query.name);
1053                    store.set_config_tree(
1054                        &format!("rls.enabled.{}", query.name),
1055                        &crate::serde_json::Value::Null,
1056                    );
1057                    self.invalidate_plan_cache();
1058                    messages.push(format!("row level security disabled on '{}'", query.name));
1059                    self.invalidate_result_cache();
1060                }
1061                // Phase 2.5.4: retrofit tenancy onto an existing table.
1062                AlterOperation::EnableTenancy { column } => {
1063                    store.set_config_tree(
1064                        &format!("tenant_tables.{}.column", query.name),
1065                        &crate::serde_json::Value::String(column.clone()),
1066                    );
1067                    self.register_tenant_table(&query.name, column);
1068                    self.invalidate_plan_cache();
1069                    messages.push(format!(
1070                        "tenancy enabled on '{}' by column '{column}'",
1071                        query.name
1072                    ));
1073                }
1074                AlterOperation::DisableTenancy => {
1075                    store.set_config_tree(
1076                        &format!("tenant_tables.{}.column", query.name),
1077                        &crate::serde_json::Value::Null,
1078                    );
1079                    self.unregister_tenant_table(&query.name);
1080                    self.invalidate_plan_cache();
1081                    messages.push(format!("tenancy disabled on '{}'", query.name));
1082                }
1083                AlterOperation::SetAppendOnly(on) => {
1084                    // Contract is the single source of truth for the
1085                    // UPDATE/DELETE parse-time guard. The flag lands
1086                    // below via `apply_alter_operations_to_contract`;
1087                    // here we only publish the human-readable message.
1088                    messages.push(format!(
1089                        "append_only {} on '{}'",
1090                        if *on { "enabled" } else { "disabled" },
1091                        query.name
1092                    ));
1093                }
1094                AlterOperation::SetVersioned(on) => {
1095                    // Opt the collection into (or out of) Git-for-Data.
1096                    // Persists a row in red_vcs_settings; next AS OF /
1097                    // merge / diff against this table honours the
1098                    // flag. Retroactive: existing row versions whose
1099                    // xmins are still pinned by commits become
1100                    // reachable via AS OF COMMIT immediately.
1101                    self.vcs_set_versioned(&query.name, *on)?;
1102                    messages.push(format!(
1103                        "versioned {} on '{}'",
1104                        if *on { "enabled" } else { "disabled" },
1105                        query.name
1106                    ));
1107                }
1108                AlterOperation::EnableEvents(subscription) => {
1109                    let mut subscription = subscription.clone();
1110                    subscription.source = query.name.clone();
1111                    validate_event_subscriptions(
1112                        self,
1113                        &query.name,
1114                        std::slice::from_ref(&subscription),
1115                    )?;
1116                    ensure_event_target_queue(self, &subscription.target_queue)?;
1117                    messages.push(format!(
1118                        "events enabled on '{}' to '{}'",
1119                        query.name, subscription.target_queue
1120                    ));
1121                }
1122                AlterOperation::DisableEvents => {
1123                    messages.push(format!("events disabled on '{}'", query.name));
1124                }
1125                AlterOperation::AddSubscription { name, descriptor } => {
1126                    let mut sub = descriptor.clone();
1127                    sub.name = name.clone();
1128                    sub.source = query.name.clone();
1129                    validate_event_subscriptions(self, &query.name, std::slice::from_ref(&sub))?;
1130                    ensure_event_target_queue(self, &sub.target_queue)?;
1131                    messages.push(format!(
1132                        "subscription '{}' added on '{}' to '{}'",
1133                        name, query.name, sub.target_queue
1134                    ));
1135                }
1136                AlterOperation::DropSubscription { name } => {
1137                    messages.push(format!(
1138                        "subscription '{}' dropped on '{}'",
1139                        name, query.name
1140                    ));
1141                }
1142                AlterOperation::AddSigner { pubkey } => {
1143                    // Issue #522 — admin-gated registry mutation. The
1144                    // standard DDL `check_write` above gates by role; we
1145                    // additionally verify the collection actually has a
1146                    // signed-writes registry installed so this isn't a
1147                    // covert way to retrofit one (use `CREATE COLLECTION
1148                    // ... SIGNED_BY (...)` for that).
1149                    if !crate::runtime::signed_writes_kind::is_signed(&store, &query.name) {
1150                        return Err(RedDBError::Query(format!(
1151                            "ALTER COLLECTION ADD SIGNER: '{}' has no signer registry; \
1152                             recreate it with CREATE COLLECTION ... SIGNED_BY (...)",
1153                            query.name
1154                        )));
1155                    }
1156                    let actor = crate::runtime::impl_core::current_user_projected()
1157                        .unwrap_or_else(|| "@system/alter".to_string());
1158                    let changed = crate::runtime::signed_writes_kind::add_signer(
1159                        &store,
1160                        &query.name,
1161                        *pubkey,
1162                        &actor,
1163                    );
1164                    messages.push(format!(
1165                        "signer {} on '{}'",
1166                        if changed { "added" } else { "already present" },
1167                        query.name
1168                    ));
1169                }
1170                AlterOperation::RevokeSigner { pubkey } => {
1171                    if !crate::runtime::signed_writes_kind::is_signed(&store, &query.name) {
1172                        return Err(RedDBError::Query(format!(
1173                            "ALTER COLLECTION REVOKE SIGNER: '{}' has no signer registry",
1174                            query.name
1175                        )));
1176                    }
1177                    let actor = crate::runtime::impl_core::current_user_projected()
1178                        .unwrap_or_else(|| "@system/alter".to_string());
1179                    let changed = crate::runtime::signed_writes_kind::revoke_signer(
1180                        &store,
1181                        &query.name,
1182                        pubkey,
1183                        &actor,
1184                    );
1185                    messages.push(format!(
1186                        "signer {} on '{}'",
1187                        if changed {
1188                            "revoked"
1189                        } else {
1190                            "already revoked"
1191                        },
1192                        query.name
1193                    ));
1194                }
1195                AlterOperation::SetRetention { duration_ms } => {
1196                    // Issue #580 — validate that the collection has a
1197                    // timestamp column the lazy-on-scan filter can key
1198                    // off. Without one there is no anchor for "older
1199                    // than now - duration"; reject at ALTER time so the
1200                    // policy can never silently hide all rows.
1201                    let existing = self.inner.db.collection_contract(&query.name);
1202                    let has_ts_column = existing
1203                        .as_ref()
1204                        .map(retention_timestamp_column_exists)
1205                        .unwrap_or(false);
1206                    if !has_ts_column {
1207                        return Err(RedDBError::Query(format!(
1208                            "ALTER COLLECTION SET RETENTION: '{}' has no timestamp \
1209                             column — declare a TIMESTAMP/TIMESTAMPMS/DATETIME column \
1210                             or enable WITH timestamps = true before setting a \
1211                             retention policy",
1212                            query.name
1213                        )));
1214                    }
1215                    messages.push(format!(
1216                        "retention set to {duration_ms} ms on '{}'",
1217                        query.name
1218                    ));
1219                }
1220                AlterOperation::UnsetRetention => {
1221                    messages.push(format!("retention cleared on '{}'", query.name));
1222                }
1223                AlterOperation::AddAnalytics(views) => {
1224                    // Issue #801 — analytics lifecycle is graph-only. Reject the
1225                    // op on a non-graph collection so a misrouted ALTER fails
1226                    // loudly instead of silently writing analytics_config onto a
1227                    // model whose reader never resolves `<name>.<output>`.
1228                    self.require_graph_for_analytics(&query.name)?;
1229                    let existing = self.inner.db.collection_contract(&query.name);
1230                    let enabled: std::collections::BTreeSet<crate::catalog::AnalyticsOutput> =
1231                        existing
1232                            .as_ref()
1233                            .map(|c| c.analytics_config.iter().map(|v| v.output).collect())
1234                            .unwrap_or_default();
1235                    for view in views {
1236                        if enabled.contains(&view.output) {
1237                            // Idempotent: adding an already-enabled output is a
1238                            // no-op — no error, no duplicate state.
1239                            messages.push(format!(
1240                                "analytics '{}' already enabled on '{}'",
1241                                view.output.as_str(),
1242                                query.name
1243                            ));
1244                        } else {
1245                            messages.push(format!(
1246                                "analytics '{}' enabled on '{}'",
1247                                view.output.as_str(),
1248                                query.name
1249                            ));
1250                        }
1251                    }
1252                }
1253                AlterOperation::DropAnalytics(output) => {
1254                    self.require_graph_for_analytics(&query.name)?;
1255                    let enabled = self
1256                        .inner
1257                        .db
1258                        .collection_contract(&query.name)
1259                        .map(|c| c.analytics_config.iter().any(|v| v.output == *output))
1260                        .unwrap_or(false);
1261                    if !enabled {
1262                        // Dropping an output that was never enabled is a clean,
1263                        // explicit error (AC4) rather than a silent no-op.
1264                        return Err(RedDBError::Query(format!(
1265                            "ALTER GRAPH DROP ANALYTICS: analytics output '{}' is not enabled on graph '{}'",
1266                            output.as_str(),
1267                            query.name
1268                        )));
1269                    }
1270                    messages.push(format!(
1271                        "analytics '{}' disabled on '{}'",
1272                        output.as_str(),
1273                        query.name
1274                    ));
1275                }
1276            }
1277        }
1278
1279        let mut contract = self
1280            .inner
1281            .db
1282            .collection_contract(&query.name)
1283            .unwrap_or_else(|| default_collection_contract_for_existing_table(&query.name));
1284        apply_alter_operations_to_contract(&mut contract, &query.operations);
1285        contract.version = contract.version.saturating_add(1);
1286        contract.updated_at_unix_ms = current_unix_ms();
1287        self.inner
1288            .db
1289            .save_collection_contract(contract)
1290            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1291        // Issue #301 — emit OperatorEvent when column schema changes on a
1292        // collection that has active event subscriptions, so operators know
1293        // downstream consumers may see a different payload shape.
1294        if !fields_added.is_empty() || !fields_removed.is_empty() {
1295            let sub_names: Vec<String> = self
1296                .inner
1297                .db
1298                .collection_contract(&query.name)
1299                .map(|c| {
1300                    c.subscriptions
1301                        .iter()
1302                        .filter(|s| s.enabled)
1303                        .map(|s| s.name.clone())
1304                        .collect()
1305                })
1306                .unwrap_or_default();
1307            if !sub_names.is_empty() {
1308                crate::telemetry::operator_event::OperatorEvent::SubscriptionSchemaChange {
1309                    collection: query.name.clone(),
1310                    subscription_names: sub_names.join(", "),
1311                    fields_added: fields_added.join(", "),
1312                    fields_removed: fields_removed.join(", "),
1313                    lsn: self.cdc_current_lsn(),
1314                }
1315                .emit_global();
1316            }
1317        }
1318
1319        self.clear_table_planner_stats(&query.name);
1320        self.invalidate_result_cache();
1321        // Issue #120 — refresh the schema-vocabulary entries from the
1322        // post-ALTER contract. Drop+recreate inside the index keeps
1323        // the invalidation guarantee complete (no stale columns from
1324        // before an ALTER ... DROP COLUMN).
1325        let post_alter_columns: Vec<String> = self
1326            .inner
1327            .db
1328            .collection_contract(&query.name)
1329            .map(|contract| {
1330                contract
1331                    .declared_columns
1332                    .iter()
1333                    .map(|col| col.name.clone())
1334                    .collect()
1335            })
1336            .unwrap_or_default();
1337        self.schema_vocabulary_apply(
1338            crate::runtime::schema_vocabulary::DdlEvent::AlterCollection {
1339                collection: query.name.clone(),
1340                columns: post_alter_columns,
1341                type_tags: Vec::new(),
1342                description: None,
1343            },
1344        );
1345
1346        let message = if messages.is_empty() {
1347            format!("table '{}' altered (no operations)", query.name)
1348        } else {
1349            format!("table '{}' altered: {}", query.name, messages.join(", "))
1350        };
1351
1352        Ok(RuntimeQueryResult::ok_message(
1353            raw_query.to_string(),
1354            &message,
1355            "alter",
1356        ))
1357    }
1358
1359    pub fn execute_create_vcs_ref(
1360        &self,
1361        raw_query: &str,
1362        query: &CreateVcsRefQuery,
1363    ) -> RedDBResult<RuntimeQueryResult> {
1364        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1365        let connection_id = crate::runtime::impl_core::current_connection_id();
1366        match query.kind {
1367            VcsRefKind::Branch => {
1368                self.vcs_branch_create(crate::application::vcs::CreateBranchInput {
1369                    name: query.name.clone(),
1370                    from: query.target.clone(),
1371                    connection_id,
1372                })?;
1373            }
1374            VcsRefKind::Tag => {
1375                let target = match &query.target {
1376                    Some(target) => target.clone(),
1377                    None => self
1378                        .vcs_status(crate::application::vcs::StatusInput { connection_id })?
1379                        .head_commit
1380                        .filter(|head| !head.is_empty())
1381                        .ok_or_else(|| {
1382                            RedDBError::InvalidConfig(
1383                                "cannot create tag: HEAD has no commits".to_string(),
1384                            )
1385                        })?,
1386                };
1387                self.vcs_tag_create(crate::application::vcs::CreateTagInput {
1388                    name: query.name.clone(),
1389                    target,
1390                    annotation: None,
1391                })?;
1392            }
1393        }
1394        let kind = match query.kind {
1395            VcsRefKind::Branch => "branch",
1396            VcsRefKind::Tag => "tag",
1397        };
1398        Ok(RuntimeQueryResult::ok_message(
1399            raw_query.to_string(),
1400            &format!("{kind} '{}' created", query.name),
1401            "create",
1402        ))
1403    }
1404
1405    pub fn execute_drop_vcs_ref(
1406        &self,
1407        raw_query: &str,
1408        query: &DropVcsRefQuery,
1409    ) -> RedDBResult<RuntimeQueryResult> {
1410        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1411        match query.kind {
1412            VcsRefKind::Branch => self.vcs_branch_delete(&query.name)?,
1413            VcsRefKind::Tag => self.vcs_tag_delete(&query.name)?,
1414        }
1415        let kind = match query.kind {
1416            VcsRefKind::Branch => "branch",
1417            VcsRefKind::Tag => "tag",
1418        };
1419        Ok(RuntimeQueryResult::ok_message(
1420            raw_query.to_string(),
1421            &format!("{kind} '{}' dropped", query.name),
1422            "drop",
1423        ))
1424    }
1425
1426    pub fn execute_fork_store(
1427        &self,
1428        raw_query: &str,
1429        query: &ForkStoreQuery,
1430    ) -> RedDBResult<RuntimeQueryResult> {
1431        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1432        let path = self.inner.db.path().ok_or_else(|| {
1433            RedDBError::Query("FORK STORE requires a persistent store".to_string())
1434        })?;
1435        let durable_lsn = self.cdc_current_lsn();
1436        let fork_lsn = query.at_lsn.unwrap_or(durable_lsn);
1437        let retention_floor = operational_wal_retention_floor(durable_lsn);
1438        validate_store_fork_lsn(fork_lsn, retention_floor, durable_lsn)?;
1439
1440        let manifest = reddb_file::OperationalManifest::for_db_path(path);
1441        let mut collections = self.inner.db.store().list_collections();
1442        collections.sort();
1443        manifest.recover_or_bootstrap(&collections).map_err(|err| {
1444            RedDBError::Query(format!("failed to prepare store fork manifest: {err}"))
1445        })?;
1446        manifest
1447            .create_fork(&query.name, fork_lsn)
1448            .map_err(|err| RedDBError::Query(format!("failed to create store fork: {err}")))?;
1449        Ok(RuntimeQueryResult::ok_message(
1450            raw_query.to_string(),
1451            &format!("store fork '{}' created at LSN {fork_lsn}", query.name),
1452            "fork_store",
1453        ))
1454    }
1455
1456    pub fn execute_drop_fork(
1457        &self,
1458        raw_query: &str,
1459        query: &DropForkQuery,
1460    ) -> RedDBResult<RuntimeQueryResult> {
1461        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1462        let path = self.inner.db.path().ok_or_else(|| {
1463            RedDBError::Query("DROP FORK requires a persistent store".to_string())
1464        })?;
1465        let manifest = reddb_file::OperationalManifest::for_db_path(path);
1466        let dropped = manifest
1467            .drop_fork(&query.name)
1468            .map_err(|err| RedDBError::Query(format!("failed to drop store fork: {err}")))?;
1469        if !dropped && !query.if_exists {
1470            return Err(RedDBError::NotFound(format!("store fork '{}'", query.name)));
1471        }
1472        Ok(RuntimeQueryResult::ok_message(
1473            raw_query.to_string(),
1474            &format!("store fork '{}' dropped", query.name),
1475            "drop_fork",
1476        ))
1477    }
1478
1479    pub fn execute_promote_fork(
1480        &self,
1481        raw_query: &str,
1482        query: &PromoteForkQuery,
1483    ) -> RedDBResult<RuntimeQueryResult> {
1484        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1485        let path = self.inner.db.path().ok_or_else(|| {
1486            RedDBError::Query("PROMOTE FORK requires a persistent store".to_string())
1487        })?;
1488        self.flush()?;
1489        let manifest = reddb_file::OperationalManifest::for_db_path(path);
1490        let outcome = manifest
1491            .promote_fork(&query.name)
1492            .map_err(|err| RedDBError::Query(format!("failed to promote store fork: {err}")))?;
1493        let outcome =
1494            outcome.ok_or_else(|| RedDBError::NotFound(format!("store fork '{}'", query.name)))?;
1495        Ok(RuntimeQueryResult::ok_message(
1496            raw_query.to_string(),
1497            &format!(
1498                "store fork '{}' promoted at LSN {}; retired parent archived at {}",
1499                outcome.name,
1500                outcome.fork_lsn,
1501                outcome.archived_parent.store_identity()
1502            ),
1503            "promote_fork",
1504        ))
1505    }
1506
1507    /// Execute EXPLAIN ALTER FOR CREATE TABLE
1508    ///
1509    /// Pure read: computes the schema diff between the target table's
1510    /// current `CollectionContract` and the embedded `CREATE TABLE` body,
1511    /// and returns it as SQL `ALTER TABLE` text (default) or structured
1512    /// JSON. Never mutates storage.
1513    pub fn execute_explain_alter(
1514        &self,
1515        raw_query: &str,
1516        query: &ExplainAlterQuery,
1517    ) -> RedDBResult<RuntimeQueryResult> {
1518        // Validate the target CREATE TABLE body so syntactically valid
1519        // but semantically broken targets (bad SQL types, duplicate
1520        // columns) are caught here rather than inside the diff engine.
1521        let analyzed = analyze_create_table(&query.target)
1522            .map_err(|err| RedDBError::Query(err.to_string()))?;
1523
1524        let current_contract = self.inner.db.collection_contract(&query.target.name);
1525
1526        // The column-diff planner cannot emit ADD/DROP CONSTRAINT. Do not
1527        // silently report a complete migration when table-level UNIQUE changes.
1528        let current_unique = current_contract
1529            .as_ref()
1530            .map(|contract| {
1531                contract
1532                    .table_def
1533                    .iter()
1534                    .flat_map(|table| &table.constraints)
1535                    .filter(|constraint| {
1536                        constraint.constraint_type == reddb_types::ConstraintType::Unique
1537                            && !contract.declared_columns.iter().any(|column| {
1538                                column.unique
1539                                    && constraint.columns == [column.name.clone()]
1540                                    && constraint.name == format!("uniq_{}", column.name)
1541                            })
1542                    })
1543                    .map(|constraint| (constraint.name.clone(), constraint.columns.clone()))
1544                    .collect::<BTreeSet<_>>()
1545            })
1546            .unwrap_or_default();
1547        let target_unique = analyzed
1548            .unique_constraints
1549            .into_iter()
1550            .map(|constraint| (constraint.name, constraint.columns))
1551            .collect::<BTreeSet<_>>();
1552        if current_unique != target_unique {
1553            return Err(RedDBError::Query("NOT_YET_SUPPORTED: EXPLAIN ALTER cannot plan changes to table-level UNIQUE constraints".to_string()));
1554        }
1555
1556        let current_columns: Vec<crate::physical::DeclaredColumnContract> = current_contract
1557            .as_ref()
1558            .map(|c| c.declared_columns.clone())
1559            .unwrap_or_default();
1560
1561        let diff = super::schema_diff::compute_column_diff(
1562            &query.target.name,
1563            &current_columns,
1564            &query.target.columns,
1565        );
1566
1567        let rendered = match query.format {
1568            ExplainFormat::Sql => super::schema_diff::format_as_sql(&diff),
1569            ExplainFormat::Json => super::schema_diff::format_as_json(&diff),
1570        };
1571
1572        let format_label = match query.format {
1573            ExplainFormat::Sql => "sql",
1574            ExplainFormat::Json => "json",
1575        };
1576
1577        let columns = vec![
1578            "table".to_string(),
1579            "format".to_string(),
1580            "diff".to_string(),
1581        ];
1582        let row = vec![
1583            ("table".to_string(), Value::text(query.target.name.clone())),
1584            ("format".to_string(), Value::text(format_label.to_string())),
1585            ("diff".to_string(), Value::text(rendered)),
1586        ];
1587
1588        Ok(RuntimeQueryResult::ok_records(
1589            raw_query.to_string(),
1590            columns,
1591            vec![row],
1592            "explain",
1593        ))
1594    }
1595
1596    /// Execute CREATE INDEX
1597    ///
1598    /// Registers a new index on a collection, builds it from existing data,
1599    /// and makes it available to the query executor for O(1) lookups.
1600    pub fn execute_create_index(
1601        &self,
1602        raw_query: &str,
1603        query: &CreateIndexQuery,
1604    ) -> RedDBResult<RuntimeQueryResult> {
1605        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1606        let topology_lock = self
1607            .inner
1608            .index_store
1609            .collection_topology_lock(&query.table);
1610        let _topology_guard = topology_lock.write();
1611        let store = self.inner.db.store();
1612
1613        // Verify the table exists
1614        let manager = store
1615            .get_collection(&query.table)
1616            .ok_or_else(|| RedDBError::NotFound(format!("table '{}' not found", query.table)))?;
1617
1618        if query.if_not_exists
1619            && self
1620                .inner
1621                .index_store
1622                .list_indices(&query.table)
1623                .iter()
1624                .any(|index| index.name == query.name)
1625        {
1626            return Ok(RuntimeQueryResult::ok_message(
1627                raw_query.to_string(),
1628                &format!(
1629                    "index '{}' already exists on '{}'; skipped",
1630                    query.name, query.table
1631                ),
1632                "create",
1633            ));
1634        }
1635
1636        let method_kind = match query.method {
1637            IndexMethod::Hash => super::index_store::IndexMethodKind::Hash,
1638            IndexMethod::BTree => super::index_store::IndexMethodKind::BTree,
1639            IndexMethod::Bitmap => super::index_store::IndexMethodKind::Bitmap,
1640            // Generic spatial request → engine default spatial backend
1641            // (disk-resident H3 as of PRD #1574 slice 4, #1578).
1642            IndexMethod::Spatial => super::index_store::IndexMethodKind::default_spatial(),
1643            IndexMethod::H3 { resolution } => {
1644                super::index_store::IndexMethodKind::H3 { resolution }
1645            }
1646        };
1647
1648        // Extract fields from existing entities for indexing. Row
1649        // entities may arrive in either the "named" HashMap layout
1650        // (gRPC `BulkInsertBinary` path) OR the columnar layout
1651        // (HTTP `POST /collections/X/bulk/rows` fast path, which uses
1652        // `schema: Some(Arc<Vec<String>>)` + `columns: Vec<Value>` and
1653        // leaves `named == None`). Prior to this commit the columnar
1654        // branch returned an empty field list, so `CREATE INDEX` built
1655        // a zero-entity index over HTTP-inserted data even though the
1656        // data was queryable via `SELECT`.
1657        let entities = manager.query_all(|_| true);
1658        let mut entity_fields: Vec<(crate::storage::unified::EntityId, Vec<(String, Value)>)> =
1659            entities
1660                .iter()
1661                .map(|e| {
1662                    let fields = match &e.data {
1663                        crate::storage::EntityData::Row(row) => {
1664                            if let Some(ref named) = row.named {
1665                                named.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
1666                            } else if let Some(ref schema) = row.schema {
1667                                // Columnar layout — pair each column
1668                                // with its positional name from the
1669                                // shared schema Arc.
1670                                schema
1671                                    .iter()
1672                                    .zip(row.columns.iter())
1673                                    .map(|(k, v)| (k.clone(), v.clone()))
1674                                    .collect()
1675                            } else {
1676                                Vec::new()
1677                            }
1678                        }
1679                        crate::storage::EntityData::Node(node) => node
1680                            .properties
1681                            .iter()
1682                            .map(|(k, v)| (k.clone(), v.clone()))
1683                            .collect(),
1684                        crate::storage::EntityData::TimeSeries(ts) => ts
1685                            .fields
1686                            .iter()
1687                            .map(|(k, v)| (k.clone(), v.clone()))
1688                            .collect(),
1689                        _ => Vec::new(),
1690                    };
1691                    (e.id, fields)
1692                })
1693                .collect();
1694
1695        if query.unique
1696            && matches!(
1697                method_kind,
1698                super::index_store::IndexMethodKind::Hash
1699                    | super::index_store::IndexMethodKind::BTree
1700            )
1701        {
1702            // Only live rows reserve a key: the dead versions of an updated or
1703            // deleted row share its key but are not duplicates. This is DDL
1704            // only — reopening must still rebuild an index over data that an
1705            // older release let duplicate.
1706            let snapshots = self.snapshot_manager();
1707            let own_xids = self.own_transaction_xids();
1708            let mut live = entities.iter().map(|entity| {
1709                snapshots.row_reserves_unique_key(entity.xmin, entity.xmax, &own_xids)
1710            });
1711            entity_fields.retain(|_| live.next().unwrap_or(false));
1712            if let Some((earlier, later)) =
1713                super::index_store::first_duplicate_unique_key(&query.columns, &entity_fields)
1714            {
1715                return Err(RedDBError::Query(format!(
1716                    "cannot create unique index '{}' on '{}': rows {} and {} share the same ({}) key",
1717                    query.name,
1718                    query.table,
1719                    earlier.raw(),
1720                    later.raw(),
1721                    query.columns.join(", ")
1722                )));
1723            }
1724        }
1725
1726        let index_growth_rows = entity_fields
1727            .iter()
1728            .map(|(_, fields)| fields.clone())
1729            .collect::<Vec<_>>();
1730        let _reservation = self.admit_non_evictable_growth(
1731            crate::storage::memory_pools::MemoryPool::IndexMemory,
1732            &format!("create index {}", query.name),
1733            crate::runtime::memory_admission::estimate_index_growth(
1734                &index_growth_rows,
1735                &query.columns,
1736            ),
1737        )?;
1738
1739        // Build the index
1740        let indexed_count = self
1741            .inner
1742            .index_store
1743            .create_index(
1744                &query.name,
1745                &query.table,
1746                &query.columns,
1747                method_kind,
1748                query.unique,
1749                &entity_fields,
1750            )
1751            .map_err(RedDBError::Internal)?;
1752
1753        let analyzed = crate::storage::query::planner::stats_catalog::analyze_entity_fields(
1754            &query.table,
1755            &entity_fields,
1756        );
1757        crate::storage::query::planner::stats_catalog::persist_table_stats(&store, &analyzed);
1758        self.invalidate_plan_cache();
1759
1760        // Register metadata
1761        self.inner
1762            .index_store
1763            .register(super::index_store::RegisteredIndex {
1764                name: query.name.clone(),
1765                collection: query.table.clone(),
1766                columns: query.columns.clone(),
1767                method: method_kind,
1768                unique: query.unique,
1769            });
1770        self.persist_runtime_index_descriptor(super::index_store::RegisteredIndex {
1771            name: query.name.clone(),
1772            collection: query.table.clone(),
1773            columns: query.columns.clone(),
1774            method: method_kind,
1775            unique: query.unique,
1776        })?;
1777        // Issue #120 — surface the index name + indexed columns in
1778        // the schema-vocabulary so AskPipeline (#121) can resolve
1779        // "the email index" back to its collection.
1780        self.schema_vocabulary_apply(crate::runtime::schema_vocabulary::DdlEvent::CreateIndex {
1781            collection: query.table.clone(),
1782            index: query.name.clone(),
1783            columns: query.columns.clone(),
1784        });
1785
1786        let method_str = format!("{}", query.method);
1787        let unique_str = if query.unique { "unique " } else { "" };
1788        let cols = query.columns.join(", ");
1789        let total_count = entity_fields.len();
1790        let coverage_detail =
1791            if matches!(method_kind, super::index_store::IndexMethodKind::H3 { .. })
1792                && total_count > 0
1793                && indexed_count == 0
1794            {
1795                format!(
1796                    "{} of {} entities indexed — no indexable geo value in '{}'; expected {}",
1797                    indexed_count,
1798                    total_count,
1799                    cols,
1800                    crate::geo::RECOGNIZED_GEO_SHAPES
1801                )
1802            } else if matches!(method_kind, super::index_store::IndexMethodKind::H3 { .. }) {
1803                format!("{} of {} entities indexed", indexed_count, total_count)
1804            } else {
1805                format!("{} entities indexed", indexed_count)
1806            };
1807
1808        Ok(RuntimeQueryResult::ok_message(
1809            raw_query.to_string(),
1810            &format!(
1811                "{}index '{}' created on '{}' ({}) using {} ({})",
1812                unique_str, query.name, query.table, cols, method_str, coverage_detail
1813            ),
1814            "create",
1815        ))
1816    }
1817
1818    /// Execute DROP INDEX
1819    ///
1820    /// Removes an index from a collection.
1821    pub fn execute_drop_index(
1822        &self,
1823        raw_query: &str,
1824        query: &DropIndexQuery,
1825    ) -> RedDBResult<RuntimeQueryResult> {
1826        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1827        let topology_lock = self
1828            .inner
1829            .index_store
1830            .collection_topology_lock(&query.table);
1831        let _topology_guard = topology_lock.write();
1832        let store = self.inner.db.store();
1833
1834        // Verify the table exists
1835        if store.get_collection(&query.table).is_none() {
1836            if query.if_exists {
1837                return Ok(RuntimeQueryResult::ok_message(
1838                    raw_query.to_string(),
1839                    &format!("table '{}' does not exist", query.table),
1840                    "drop",
1841                ));
1842            }
1843            return Err(RedDBError::NotFound(format!(
1844                "table '{}' not found",
1845                query.table
1846            )));
1847        }
1848
1849        // Remove from IndexStore
1850        self.inner.index_store.drop_index(&query.name, &query.table);
1851        self.persist_runtime_index_drop(&query.table, &query.name)?;
1852        self.invalidate_plan_cache();
1853        // Issue #120 — keep the schema-vocabulary index entry in sync.
1854        self.schema_vocabulary_apply(crate::runtime::schema_vocabulary::DdlEvent::DropIndex {
1855            collection: query.table.clone(),
1856            index: query.name.clone(),
1857        });
1858
1859        Ok(RuntimeQueryResult::ok_message(
1860            raw_query.to_string(),
1861            &format!("index '{}' dropped from '{}'", query.name, query.table),
1862            "drop",
1863        ))
1864    }
1865
1866    fn execute_drop_typed_collection(
1867        &self,
1868        raw_query: &str,
1869        name: &str,
1870        if_exists: bool,
1871        expected_model: CollectionModel,
1872        label: &str,
1873    ) -> RedDBResult<RuntimeQueryResult> {
1874        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1875        if is_system_schema_name(name) {
1876            return Err(RedDBError::Query("system schema is read-only".to_string()));
1877        }
1878        let store = self.inner.db.store();
1879        if store.get_collection(name).is_none() {
1880            if if_exists {
1881                return Ok(RuntimeQueryResult::ok_message(
1882                    raw_query.to_string(),
1883                    &format!("{label} '{name}' does not exist"),
1884                    "drop",
1885                ));
1886            }
1887            return Err(RedDBError::NotFound(format!("{label} '{name}' not found")));
1888        }
1889
1890        let actual = self
1891            .inner
1892            .db
1893            .collection_contract(name)
1894            .map(|contract| contract.declared_model)
1895            .map(Ok)
1896            .unwrap_or_else(|| {
1897                polymorphic_resolver::resolve(name, &self.inner.db.catalog_model_snapshot())
1898            })?;
1899        polymorphic_resolver::ensure_model_match(expected_model, actual)?;
1900        self.drop_collection_storage(raw_query, name, label)
1901    }
1902
1903    pub fn execute_truncate(
1904        &self,
1905        raw_query: &str,
1906        query: &TruncateQuery,
1907    ) -> RedDBResult<RuntimeQueryResult> {
1908        self.check_write(crate::runtime::write_gate::WriteKind::Ddl)?;
1909        if is_system_schema_name(&query.name) {
1910            return Err(RedDBError::Query("system schema is read-only".to_string()));
1911        }
1912
1913        let label = query
1914            .model
1915            .map(polymorphic_resolver::model_name)
1916            .unwrap_or("collection");
1917        let store = self.inner.db.store();
1918        if store.get_collection(&query.name).is_none() {
1919            if query.if_exists {
1920                return Ok(RuntimeQueryResult::ok_message(
1921                    raw_query.to_string(),
1922                    &format!("{label} '{}' does not exist", query.name),
1923                    "truncate",
1924                ));
1925            }
1926            return Err(RedDBError::NotFound(format!(
1927                "{label} '{}' not found",
1928                query.name
1929            )));
1930        }
1931
1932        let actual =
1933            polymorphic_resolver::resolve(&query.name, &self.inner.db.catalog_model_snapshot())?;
1934        if let Some(expected) = query.model {
1935            polymorphic_resolver::ensure_model_match(expected, actual)?;
1936        }
1937
1938        if actual == CollectionModel::Queue {
1939            return self.execute_queue_command(
1940                raw_query,
1941                &QueueCommand::Purge {
1942                    queue: query.name.clone(),
1943                },
1944            );
1945        }
1946
1947        // Count before wiping so we can emit the aggregated truncate event.
1948        let affected = self.truncate_collection_entities(&query.name)?;
1949        // Emit 1 truncate event (not N delete events) for event-enabled collections.
1950        crate::runtime::mutation::emit_truncate_event_for_collection(self, &query.name, affected)?;
1951        self.inner.db.invalidate_vector_index(&query.name);
1952        self.clear_table_planner_stats(&query.name);
1953        self.invalidate_result_cache();
1954
1955        Ok(RuntimeQueryResult::ok_message(
1956            raw_query.to_string(),
1957            &format!(
1958                "{affected} entities truncated from {label} '{}'",
1959                query.name
1960            ),
1961            "truncate",
1962        ))
1963    }
1964
1965    fn truncate_collection_entities(&self, name: &str) -> RedDBResult<u64> {
1966        let store = self.inner.db.store();
1967        let Some(manager) = store.get_collection(name) else {
1968            return Ok(0);
1969        };
1970        let entities = manager.query_all(|_| true);
1971        if entities.is_empty() {
1972            return Ok(0);
1973        }
1974
1975        for entity in &entities {
1976            let fields = entity_index_fields(&entity.data);
1977            self.inner
1978                .index_store
1979                .index_entity_delete(name, entity.id, &fields)
1980                .map_err(RedDBError::Internal)?;
1981        }
1982
1983        let ids = entities.iter().map(|entity| entity.id).collect::<Vec<_>>();
1984        let deleted_ids = store
1985            .delete_batch(name, &ids)
1986            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1987        for id in &deleted_ids {
1988            store.context_index().remove_entity(*id);
1989        }
1990        Ok(deleted_ids.len() as u64)
1991    }
1992
1993    fn drop_collection_storage(
1994        &self,
1995        raw_query: &str,
1996        name: &str,
1997        label: &str,
1998    ) -> RedDBResult<RuntimeQueryResult> {
1999        let store = self.inner.db.store();
2000
2001        // Emit 1 collection_dropped event before storage is wiped.
2002        // Queue is preserved; subscription is removed with the contract below.
2003        let final_count = store
2004            .get_collection(name)
2005            .map(|manager| manager.query_all(|_| true).len() as u64)
2006            .unwrap_or(0);
2007        crate::runtime::mutation::emit_collection_dropped_event_for_collection(
2008            self,
2009            name,
2010            final_count,
2011        )?;
2012
2013        let topology_lock = self.inner.index_store.collection_topology_lock(name);
2014        let _topology_guard = topology_lock.write();
2015        let orphaned_indices: Vec<String> = self
2016            .inner
2017            .index_store
2018            .list_indices(name)
2019            .into_iter()
2020            .map(|index| index.name)
2021            .collect();
2022        for index_name in &orphaned_indices {
2023            self.inner.index_store.drop_index(index_name, name);
2024        }
2025
2026        store
2027            .drop_collection(name)
2028            .map_err(|err| RedDBError::Internal(err.to_string()))?;
2029        self.inner.db.invalidate_vector_index(name);
2030        self.inner.db.clear_collection_default_ttl_ms(name);
2031        self.inner
2032            .db
2033            .remove_collection_contract(name)
2034            .map_err(|err| RedDBError::Internal(err.to_string()))?;
2035        self.clear_table_planner_stats(name);
2036        self.invalidate_result_cache();
2037        if let Some(store) = self.inner.auth_store.read().clone() {
2038            store.invalidate_visible_collections_cache();
2039        }
2040        self.inner
2041            .db
2042            .persist_metadata()
2043            .map_err(|err| RedDBError::Internal(err.to_string()))?;
2044        self.schema_vocabulary_apply(
2045            crate::runtime::schema_vocabulary::DdlEvent::DropCollection {
2046                collection: name.to_string(),
2047            },
2048        );
2049
2050        Ok(RuntimeQueryResult::ok_message(
2051            raw_query.to_string(),
2052            &format!("{label} '{name}' dropped"),
2053            "drop",
2054        ))
2055    }
2056}
2057
2058pub(crate) fn is_system_schema_name(name: &str) -> bool {
2059    name == "red" || name.starts_with("red.") || name.starts_with("__red_schema_")
2060}
2061
2062fn entity_index_fields(data: &EntityData) -> Vec<(String, Value)> {
2063    match data {
2064        EntityData::Row(row) => {
2065            if let Some(ref named) = row.named {
2066                named
2067                    .iter()
2068                    .map(|(key, value)| (key.clone(), value.clone()))
2069                    .collect()
2070            } else if let Some(ref schema) = row.schema {
2071                schema
2072                    .iter()
2073                    .zip(row.columns.iter())
2074                    .map(|(key, value)| (key.clone(), value.clone()))
2075                    .collect()
2076            } else {
2077                Vec::new()
2078            }
2079        }
2080        EntityData::Node(node) => node
2081            .properties
2082            .iter()
2083            .map(|(key, value)| (key.clone(), value.clone()))
2084            .collect(),
2085        _ => Vec::new(),
2086    }
2087}
2088
2089/// DDL-time gate for a declared AI policy (issue #1271). Each declared
2090/// modality block is checked against the provider capability matrix
2091/// (#1269): a provider/model that cannot serve the requested modality is
2092/// rejected before the collection is created. Uses the built-in registry
2093/// (no per-deployment overrides at DDL time).
2094fn validate_ai_policy_modalities(policy: &crate::catalog::AiPolicy) -> RedDBResult<()> {
2095    use crate::runtime::ai::provider_capabilities::{Modality, Registry};
2096    let registry = Registry::new();
2097    let check = |provider: &str, model: &str, modality: Modality| -> RedDBResult<()> {
2098        registry
2099            .validate_policy_modality(provider, model, modality)
2100            .map_err(|err| RedDBError::Query(err.to_string()))
2101    };
2102    if let Some(embed) = &policy.embed {
2103        check(&embed.provider, &embed.model, Modality::Embed)?;
2104    }
2105    if let Some(moderate) = &policy.moderate {
2106        check(&moderate.provider, &moderate.model, Modality::Moderate)?;
2107    }
2108    if let Some(vision) = &policy.vision {
2109        check(&vision.provider, &vision.model, Modality::Vision)?;
2110    }
2111    Ok(())
2112}
2113
2114fn collection_contract_from_create_table(
2115    query: &CreateTableQuery,
2116) -> RedDBResult<crate::physical::CollectionContract> {
2117    let now = current_unix_ms();
2118    let mut declared_columns: Vec<crate::physical::DeclaredColumnContract> = query
2119        .columns
2120        .iter()
2121        .map(declared_column_contract_from_ddl)
2122        .collect();
2123    if query.timestamps {
2124        // Opt-in `WITH timestamps = true` auto-adds two user-visible
2125        // columns that the write path populates from
2126        // UnifiedEntity::created_at/updated_at. BIGINT unix-ms, NOT NULL.
2127        declared_columns.push(crate::physical::DeclaredColumnContract {
2128            name: "created_at".to_string(),
2129            data_type: "BIGINT".to_string(),
2130            sql_type: Some(reddb_types::SqlTypeName::simple("BIGINT")),
2131            not_null: true,
2132            default: None,
2133            generated: None,
2134            check: None,
2135            compress: None,
2136            unique: false,
2137            primary_key: false,
2138            enum_variants: Vec::new(),
2139            array_element: None,
2140            decimal_precision: None,
2141        });
2142        declared_columns.push(crate::physical::DeclaredColumnContract {
2143            name: "updated_at".to_string(),
2144            data_type: "BIGINT".to_string(),
2145            sql_type: Some(reddb_types::SqlTypeName::simple("BIGINT")),
2146            not_null: true,
2147            default: None,
2148            generated: None,
2149            check: None,
2150            compress: None,
2151            unique: false,
2152            primary_key: false,
2153            enum_variants: Vec::new(),
2154            array_element: None,
2155            decimal_precision: None,
2156        });
2157    }
2158    Ok(crate::physical::CollectionContract {
2159        name: query.name.clone(),
2160        declared_model: crate::catalog::CollectionModel::Table,
2161        schema_mode: crate::catalog::SchemaMode::SemiStructured,
2162        origin: crate::physical::ContractOrigin::Explicit,
2163        version: 1,
2164        created_at_unix_ms: now,
2165        updated_at_unix_ms: now,
2166        default_ttl_ms: query.default_ttl_ms,
2167        vector_dimension: None,
2168        vector_metric: None,
2169        context_index_fields: query.context_index_fields.clone(),
2170        declared_columns,
2171        table_def: Some(build_table_def_from_create_table(query)?),
2172        timestamps_enabled: query.timestamps,
2173        context_index_enabled: query.context_index_enabled
2174            || !query.context_index_fields.is_empty(),
2175        metrics_raw_retention_ms: None,
2176        metrics_rollup_policies: Vec::new(),
2177        metrics_tenant_identity: None,
2178        metrics_namespace: None,
2179        append_only: query.append_only,
2180        subscriptions: query.subscriptions.clone(),
2181        analytics_config: Vec::new(),
2182        session_key: None,
2183        session_gap_ms: None,
2184        retention_duration_ms: None,
2185        analytical_storage: None,
2186        ai_policy: query.ai_policy.clone(),
2187    })
2188}
2189
2190fn default_collection_contract_for_existing_table(
2191    name: &str,
2192) -> crate::physical::CollectionContract {
2193    let now = current_unix_ms();
2194    crate::physical::CollectionContract {
2195        name: name.to_string(),
2196        declared_model: crate::catalog::CollectionModel::Table,
2197        schema_mode: crate::catalog::SchemaMode::SemiStructured,
2198        origin: crate::physical::ContractOrigin::Explicit,
2199        version: 0,
2200        created_at_unix_ms: now,
2201        updated_at_unix_ms: now,
2202        default_ttl_ms: None,
2203        vector_dimension: None,
2204        vector_metric: None,
2205        context_index_fields: Vec::new(),
2206        declared_columns: Vec::new(),
2207        table_def: Some(reddb_types::TableDef::new(name.to_string())),
2208        timestamps_enabled: false,
2209        context_index_enabled: false,
2210        metrics_raw_retention_ms: None,
2211        metrics_rollup_policies: Vec::new(),
2212        metrics_tenant_identity: None,
2213        metrics_namespace: None,
2214        append_only: false,
2215        subscriptions: Vec::new(),
2216        analytics_config: Vec::new(),
2217        session_key: None,
2218        session_gap_ms: None,
2219        retention_duration_ms: None,
2220        analytical_storage: None,
2221
2222        ai_policy: None,
2223    }
2224}
2225
2226fn keyed_collection_contract(
2227    name: &str,
2228    model: crate::catalog::CollectionModel,
2229    analytics_config: Vec<crate::catalog::AnalyticsViewDescriptor>,
2230) -> crate::physical::CollectionContract {
2231    let now = current_unix_ms();
2232    crate::physical::CollectionContract {
2233        name: name.to_string(),
2234        declared_model: model,
2235        schema_mode: crate::catalog::SchemaMode::Dynamic,
2236        origin: crate::physical::ContractOrigin::Explicit,
2237        version: 1,
2238        created_at_unix_ms: now,
2239        updated_at_unix_ms: now,
2240        default_ttl_ms: None,
2241        vector_dimension: None,
2242        vector_metric: None,
2243        context_index_fields: Vec::new(),
2244        declared_columns: Vec::new(),
2245        table_def: None,
2246        timestamps_enabled: false,
2247        context_index_enabled: false,
2248        metrics_raw_retention_ms: None,
2249        metrics_rollup_policies: Vec::new(),
2250        metrics_tenant_identity: None,
2251        metrics_namespace: None,
2252        append_only: false,
2253        subscriptions: Vec::new(),
2254        analytics_config,
2255        session_key: None,
2256        session_gap_ms: None,
2257        retention_duration_ms: None,
2258        analytical_storage: None,
2259
2260        ai_policy: None,
2261    }
2262}
2263
2264fn metrics_collection_contract(query: &CreateTableQuery) -> crate::physical::CollectionContract {
2265    let now = current_unix_ms();
2266    crate::physical::CollectionContract {
2267        name: query.name.clone(),
2268        declared_model: crate::catalog::CollectionModel::Metrics,
2269        schema_mode: crate::catalog::SchemaMode::SemiStructured,
2270        origin: crate::physical::ContractOrigin::Explicit,
2271        version: 1,
2272        created_at_unix_ms: now,
2273        updated_at_unix_ms: now,
2274        default_ttl_ms: query.default_ttl_ms,
2275        vector_dimension: None,
2276        vector_metric: None,
2277        context_index_fields: Vec::new(),
2278        declared_columns: Vec::new(),
2279        table_def: None,
2280        timestamps_enabled: false,
2281        context_index_enabled: false,
2282        metrics_raw_retention_ms: query.default_ttl_ms,
2283        metrics_rollup_policies: query.metrics_rollup_policies.clone(),
2284        metrics_tenant_identity: Some(
2285            query
2286                .tenant_by
2287                .clone()
2288                .unwrap_or_else(|| "current_tenant".to_string()),
2289        ),
2290        metrics_namespace: Some("default".to_string()),
2291        append_only: true,
2292        subscriptions: Vec::new(),
2293        analytics_config: Vec::new(),
2294        session_key: None,
2295        session_gap_ms: None,
2296        retention_duration_ms: None,
2297        analytical_storage: None,
2298
2299        ai_policy: None,
2300    }
2301}
2302
2303fn vector_collection_contract(query: &CreateVectorQuery) -> crate::physical::CollectionContract {
2304    let now = current_unix_ms();
2305    crate::physical::CollectionContract {
2306        name: query.name.clone(),
2307        declared_model: crate::catalog::CollectionModel::Vector,
2308        schema_mode: crate::catalog::SchemaMode::Dynamic,
2309        origin: crate::physical::ContractOrigin::Explicit,
2310        version: 1,
2311        created_at_unix_ms: now,
2312        updated_at_unix_ms: now,
2313        default_ttl_ms: None,
2314        vector_dimension: Some(query.dimension),
2315        vector_metric: Some(query.metric),
2316        context_index_fields: Vec::new(),
2317        declared_columns: Vec::new(),
2318        table_def: None,
2319        timestamps_enabled: false,
2320        context_index_enabled: false,
2321        metrics_raw_retention_ms: None,
2322        metrics_rollup_policies: Vec::new(),
2323        metrics_tenant_identity: None,
2324        metrics_namespace: None,
2325        append_only: false,
2326        subscriptions: Vec::new(),
2327        analytics_config: Vec::new(),
2328        session_key: None,
2329        session_gap_ms: None,
2330        retention_duration_ms: None,
2331        analytical_storage: None,
2332
2333        ai_policy: None,
2334    }
2335}
2336
2337fn declared_column_contract_from_ddl(
2338    column: &CreateColumnDef,
2339) -> crate::physical::DeclaredColumnContract {
2340    crate::physical::DeclaredColumnContract {
2341        name: column.name.clone(),
2342        data_type: column.data_type.clone(),
2343        sql_type: Some(column.sql_type.clone()),
2344        not_null: column.not_null,
2345        default: column.default.clone(),
2346        generated: column.generated.clone(),
2347        check: column.check.clone(),
2348        compress: column.compress,
2349        unique: column.unique,
2350        primary_key: column.primary_key,
2351        enum_variants: column.enum_variants.clone(),
2352        array_element: column.array_element.clone(),
2353        decimal_precision: column.decimal_precision,
2354    }
2355}
2356
2357fn apply_alter_operations_to_contract(
2358    contract: &mut crate::physical::CollectionContract,
2359    operations: &[AlterOperation],
2360) {
2361    if contract.table_def.is_none() {
2362        contract.table_def = Some(reddb_types::TableDef::new(contract.name.clone()));
2363    }
2364    for operation in operations {
2365        match operation {
2366            AlterOperation::AddColumn(column) => {
2367                if !contract
2368                    .declared_columns
2369                    .iter()
2370                    .any(|existing| existing.name == column.name)
2371                {
2372                    contract
2373                        .declared_columns
2374                        .push(declared_column_contract_from_ddl(column));
2375                }
2376                if let Some(table_def) = contract.table_def.as_mut() {
2377                    if table_def.get_column(&column.name).is_none() {
2378                        if let Ok(column_def) = column_def_from_ddl(column) {
2379                            if column.primary_key {
2380                                table_def.primary_key.push(column.name.clone());
2381                                table_def.constraints.push(
2382                                    reddb_types::Constraint::new(
2383                                        format!("pk_{}", column.name),
2384                                        reddb_types::ConstraintType::PrimaryKey,
2385                                    )
2386                                    .on_columns(vec![column.name.clone()]),
2387                                );
2388                            }
2389                            if column.unique {
2390                                table_def.constraints.push(
2391                                    reddb_types::Constraint::new(
2392                                        format!("uniq_{}", column.name),
2393                                        reddb_types::ConstraintType::Unique,
2394                                    )
2395                                    .on_columns(vec![column.name.clone()]),
2396                                );
2397                            }
2398                            if column.not_null {
2399                                table_def.constraints.push(
2400                                    reddb_types::Constraint::new(
2401                                        format!("not_null_{}", column.name),
2402                                        reddb_types::ConstraintType::NotNull,
2403                                    )
2404                                    .on_columns(vec![column.name.clone()]),
2405                                );
2406                            }
2407                            table_def.columns.push(column_def);
2408                        }
2409                    }
2410                }
2411            }
2412            AlterOperation::DropColumn(name) => {
2413                contract
2414                    .declared_columns
2415                    .retain(|column| column.name != *name);
2416                if let Some(table_def) = contract.table_def.as_mut() {
2417                    if let Some(index) = table_def.column_index(name) {
2418                        table_def.columns.remove(index);
2419                    }
2420                    table_def.primary_key.retain(|column| column != name);
2421                    table_def.constraints.retain(|constraint| {
2422                        !constraint.columns.iter().any(|column| column == name)
2423                    });
2424                    table_def
2425                        .indexes
2426                        .retain(|index| !index.columns.iter().any(|column| column == name));
2427                }
2428            }
2429            AlterOperation::RenameColumn { from, to } => {
2430                if contract
2431                    .declared_columns
2432                    .iter()
2433                    .any(|column| column.name == *to)
2434                {
2435                    continue;
2436                }
2437                if let Some(column) = contract
2438                    .declared_columns
2439                    .iter_mut()
2440                    .find(|column| column.name == *from)
2441                {
2442                    column.name = to.clone();
2443                }
2444                if let Some(table_def) = contract.table_def.as_mut() {
2445                    if let Some(column) = table_def
2446                        .columns
2447                        .iter_mut()
2448                        .find(|column| column.name == *from)
2449                    {
2450                        column.name = to.clone();
2451                    }
2452                    for primary_key in &mut table_def.primary_key {
2453                        if *primary_key == *from {
2454                            *primary_key = to.clone();
2455                        }
2456                    }
2457                    for constraint in &mut table_def.constraints {
2458                        for column in &mut constraint.columns {
2459                            if *column == *from {
2460                                *column = to.clone();
2461                            }
2462                        }
2463                        if let Some(ref_columns) = constraint.ref_columns.as_mut() {
2464                            for column in ref_columns {
2465                                if *column == *from {
2466                                    *column = to.clone();
2467                                }
2468                            }
2469                        }
2470                    }
2471                    for index in &mut table_def.indexes {
2472                        for column in &mut index.columns {
2473                            if *column == *from {
2474                                *column = to.clone();
2475                            }
2476                        }
2477                    }
2478                }
2479            }
2480            // Partition ops don't touch the column contract — metadata is
2481            // persisted separately via `red_config.partition.*`.
2482            AlterOperation::AttachPartition { .. } | AlterOperation::DetachPartition { .. } => {}
2483            // RLS toggles don't touch the column contract — flag is persisted
2484            // separately via `red_config.rls.enabled.{table}` and enforced
2485            // through the in-memory `rls_enabled_tables` set.
2486            AlterOperation::EnableRowLevelSecurity | AlterOperation::DisableRowLevelSecurity => {}
2487            // Phase 2.5.4: tenancy toggles persist via `red_config.tenant_tables.*`
2488            // and are enforced through `tenant_tables` + RLS auto-policy.
2489            AlterOperation::EnableTenancy { .. } | AlterOperation::DisableTenancy => {}
2490            AlterOperation::SetAppendOnly(on) => {
2491                contract.append_only = *on;
2492            }
2493            // VCS opt-in is persisted to red_vcs_settings by the
2494            // executor, not the contract — nothing to do here.
2495            AlterOperation::SetVersioned(_) => {}
2496            AlterOperation::EnableEvents(subscription) => {
2497                let mut subscription = subscription.clone();
2498                subscription.source = contract.name.clone();
2499                subscription.enabled = true;
2500                if let Some(existing) = contract
2501                    .subscriptions
2502                    .iter_mut()
2503                    .find(|existing| existing.target_queue == subscription.target_queue)
2504                {
2505                    *existing = subscription;
2506                } else {
2507                    contract.subscriptions.push(subscription);
2508                }
2509            }
2510            AlterOperation::DisableEvents => {
2511                for subscription in &mut contract.subscriptions {
2512                    subscription.enabled = false;
2513                }
2514            }
2515            AlterOperation::AddSubscription { name, descriptor } => {
2516                let mut sub = descriptor.clone();
2517                sub.name = name.clone();
2518                sub.source = contract.name.clone();
2519                sub.enabled = true;
2520                if let Some(existing) = contract.subscriptions.iter_mut().find(|s| s.name == *name)
2521                {
2522                    *existing = sub;
2523                } else {
2524                    contract.subscriptions.push(sub);
2525                }
2526            }
2527            AlterOperation::DropSubscription { name } => {
2528                contract.subscriptions.retain(|s| s.name != *name);
2529            }
2530            // Signer registry mutations live in `red_config` outside the
2531            // contract surface — the executor applied them directly via
2532            // `signed_writes_kind::{add,revoke}_signer`. Nothing to fold
2533            // into the column-shaped contract.
2534            AlterOperation::AddSigner { .. } | AlterOperation::RevokeSigner { .. } => {}
2535            AlterOperation::SetRetention { duration_ms } => {
2536                contract.retention_duration_ms = Some(*duration_ms);
2537            }
2538            AlterOperation::UnsetRetention => {
2539                contract.retention_duration_ms = None;
2540            }
2541            // Issue #801 — fold analytics lifecycle into the WAL-backed
2542            // `analytics_config` so the change is durable and the
2543            // `<graph>.<output>` resolver picks it up on the next read.
2544            AlterOperation::AddAnalytics(views) => {
2545                for view in views {
2546                    // Idempotent: skip outputs already enabled so a repeated
2547                    // ADD never duplicates state. The first declaration wins —
2548                    // re-declaring options requires DROP then ADD.
2549                    if !contract
2550                        .analytics_config
2551                        .iter()
2552                        .any(|existing| existing.output == view.output)
2553                    {
2554                        contract.analytics_config.push(view.clone());
2555                    }
2556                }
2557            }
2558            AlterOperation::DropAnalytics(output) => {
2559                contract
2560                    .analytics_config
2561                    .retain(|view| view.output != *output);
2562            }
2563        }
2564    }
2565}
2566
2567/// Issue #580 — returns true if the contract carries at least one
2568/// column the retention filter can use as a timestamp anchor:
2569/// either `WITH timestamps = true` (auto `created_at` / `updated_at`)
2570/// or a user-declared column with a temporal data_type.
2571pub(crate) fn retention_timestamp_column_exists(
2572    contract: &crate::physical::CollectionContract,
2573) -> bool {
2574    if contract.timestamps_enabled {
2575        return true;
2576    }
2577    if matches!(
2578        contract.declared_model,
2579        crate::catalog::CollectionModel::TimeSeries | crate::catalog::CollectionModel::Metrics
2580    ) {
2581        // Time-series and metrics collections carry an intrinsic
2582        // timestamp axis on every row even without a declared column;
2583        // retention has a natural anchor.
2584        return true;
2585    }
2586    contract
2587        .declared_columns
2588        .iter()
2589        .any(|column| is_temporal_data_type(&column.data_type))
2590}
2591
2592fn is_temporal_data_type(data_type: &str) -> bool {
2593    let upper = data_type.to_ascii_uppercase();
2594    matches!(
2595        upper.as_str(),
2596        "TIMESTAMP" | "TIMESTAMPMS" | "TIMESTAMP_MS" | "DATETIME" | "DATE"
2597    )
2598}
2599
2600fn validate_event_subscriptions(
2601    runtime: &RedDBRuntime,
2602    source: &str,
2603    subscriptions: &[crate::catalog::SubscriptionDescriptor],
2604) -> RedDBResult<()> {
2605    for subscription in subscriptions
2606        .iter()
2607        .filter(|subscription| subscription.enabled)
2608    {
2609        if subscription.all_tenants && crate::runtime::impl_core::current_tenant().is_some() {
2610            return Err(RedDBError::Query(
2611                "cross-tenant subscription requires cluster-admin capability (events:cluster_subscribe)".to_string(),
2612            ));
2613        }
2614        validate_subscription_auth(runtime, source, subscription)?;
2615        if subscription.target_queue == source
2616            || subscription_would_create_cycle(
2617                &runtime.inner.db,
2618                source,
2619                &subscription.target_queue,
2620            )
2621        {
2622            return Err(RedDBError::Query(
2623                "subscription would create cycle".to_string(),
2624            ));
2625        }
2626        audit_subscription_redact_gap(runtime, source, subscription);
2627    }
2628    Ok(())
2629}
2630
2631fn validate_subscription_auth(
2632    runtime: &RedDBRuntime,
2633    source: &str,
2634    subscription: &crate::catalog::SubscriptionDescriptor,
2635) -> RedDBResult<()> {
2636    let auth_store = match runtime.inner.auth_store.read().clone() {
2637        Some(store) => store,
2638        None => return Ok(()),
2639    };
2640    let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
2641        Some(identity) => identity,
2642        None => return Ok(()),
2643    };
2644    let tenant = crate::runtime::impl_core::current_tenant();
2645    let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
2646
2647    if auth_store.iam_authorization_enabled() {
2648        let ctx = crate::auth::policies::EvalContext {
2649            principal_tenant: tenant.clone(),
2650            current_tenant: tenant.clone(),
2651            peer_ip: None,
2652            mfa_present: false,
2653            now_ms: crate::auth::now_ms(),
2654            principal_is_admin_role: role == crate::auth::Role::Admin,
2655            principal_is_platform_scoped: principal.tenant.is_none(),
2656        };
2657        let mut source_resource = crate::auth::policies::ResourceRef::new("table", source);
2658        if let Some(t) = tenant.as_deref() {
2659            source_resource = source_resource.with_tenant(t.to_string());
2660        }
2661        if !auth_store.check_policy_authz_with_role(
2662            &principal,
2663            "select",
2664            &source_resource,
2665            &ctx,
2666            role,
2667        ) {
2668            return Err(RedDBError::Query(format!(
2669                "permission denied: principal=`{}` action=`select` resource=`{}:{}` denied by IAM policy",
2670                principal, source_resource.kind, source_resource.name
2671            )));
2672        }
2673
2674        let mut target_resource =
2675            crate::auth::policies::ResourceRef::new("queue", subscription.target_queue.clone());
2676        if let Some(t) = tenant.as_deref() {
2677            target_resource = target_resource.with_tenant(t.to_string());
2678        }
2679        if !auth_store.check_policy_authz_with_role(
2680            &principal,
2681            "write",
2682            &target_resource,
2683            &ctx,
2684            role,
2685        ) {
2686            return Err(RedDBError::Query(format!(
2687                "permission denied: principal=`{}` action=`write` resource=`{}:{}` denied by IAM policy",
2688                principal, target_resource.kind, target_resource.name
2689            )));
2690        }
2691        return Ok(());
2692    }
2693
2694    let ctx = crate::auth::privileges::AuthzContext {
2695        principal: &username,
2696        effective_role: role,
2697        tenant: tenant.as_deref(),
2698    };
2699    auth_store
2700        .check_grant(
2701            &ctx,
2702            crate::auth::privileges::Action::Select,
2703            &crate::auth::privileges::Resource::table_from_name(source),
2704        )
2705        .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))?;
2706    auth_store
2707        .check_grant(
2708            &ctx,
2709            crate::auth::privileges::Action::Insert,
2710            &crate::auth::privileges::Resource::table_from_name(&subscription.target_queue),
2711        )
2712        .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))?;
2713    Ok(())
2714}
2715
2716fn audit_subscription_redact_gap(
2717    runtime: &RedDBRuntime,
2718    source: &str,
2719    subscription: &crate::catalog::SubscriptionDescriptor,
2720) {
2721    let auth_store = match runtime.inner.auth_store.read().clone() {
2722        Some(store) if store.iam_authorization_enabled() => store,
2723        _ => return,
2724    };
2725    let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
2726        Some(identity) => identity,
2727        None => return,
2728    };
2729    let tenant = crate::runtime::impl_core::current_tenant();
2730    let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
2731    let missing = subscription_redact_gap_columns(&auth_store, &principal, source, subscription);
2732    if missing.is_empty() {
2733        return;
2734    }
2735
2736    let columns = missing.into_iter().collect::<Vec<_>>().join(", ");
2737    tracing::warn!(
2738        target: "reddb::operator",
2739        "subscription_redact_gap: source={} target_queue={} columns=[{}]",
2740        source,
2741        subscription.target_queue,
2742        columns
2743    );
2744    let mut event = AuditEvent::builder("subscription_redact_gap")
2745        .principal(username)
2746        .source(AuditAuthSource::System)
2747        .resource(format!(
2748            "subscription:{}->{}",
2749            source, subscription.target_queue
2750        ))
2751        .outcome(Outcome::Success)
2752        .field(AuditFieldEscaper::field("source", source))
2753        .field(AuditFieldEscaper::field(
2754            "target_queue",
2755            subscription.target_queue.clone(),
2756        ))
2757        .field(AuditFieldEscaper::field(
2758            "subscription",
2759            subscription.name.clone(),
2760        ))
2761        .field(AuditFieldEscaper::field("columns", columns))
2762        .field(AuditFieldEscaper::field("role", role.as_str()));
2763    if let Some(t) = tenant {
2764        event = event.tenant(t);
2765    }
2766    runtime.inner.audit_log.record_event(event.build());
2767}
2768
2769fn subscription_redact_gap_columns(
2770    auth_store: &crate::auth::store::AuthStore,
2771    principal: &crate::auth::UserId,
2772    source: &str,
2773    subscription: &crate::catalog::SubscriptionDescriptor,
2774) -> BTreeSet<String> {
2775    let redacted: HashSet<String> = subscription
2776        .redact_fields
2777        .iter()
2778        .map(|field| field.to_ascii_lowercase())
2779        .collect();
2780    auth_store
2781        .effective_policies(principal)
2782        .iter()
2783        .flat_map(|policy| policy.statements.iter())
2784        .filter(|statement| statement.effect == crate::auth::policies::Effect::Deny)
2785        .filter(|statement| statement.actions.iter().any(action_pattern_matches_select))
2786        .flat_map(|statement| statement.resources.iter())
2787        .filter_map(|resource| denied_column_for_source(resource, source))
2788        .filter(|column| !redact_covers_column(&redacted, source, column))
2789        .collect()
2790}
2791
2792fn action_pattern_matches_select(pattern: &crate::auth::policies::ActionPattern) -> bool {
2793    match pattern {
2794        crate::auth::policies::ActionPattern::Wildcard => true,
2795        crate::auth::policies::ActionPattern::Exact(action) => action == "select",
2796        crate::auth::policies::ActionPattern::Prefix(prefix) => {
2797            "select".len() > prefix.len() + 1
2798                && "select".starts_with(prefix)
2799                && "select".as_bytes()[prefix.len()] == b':'
2800        }
2801    }
2802}
2803
2804fn denied_column_for_source(
2805    resource: &crate::auth::policies::ResourcePattern,
2806    source: &str,
2807) -> Option<String> {
2808    let crate::auth::policies::ResourcePattern::Exact { kind, name } = resource else {
2809        return None;
2810    };
2811    if kind != "column" {
2812        return None;
2813    }
2814    let column = crate::auth::ColumnRef::parse_resource_name(name).ok()?;
2815    (column.table_resource_name() == source).then_some(column.column)
2816}
2817
2818fn redact_covers_column(redacted: &HashSet<String>, source: &str, column: &str) -> bool {
2819    let column = column.to_ascii_lowercase();
2820    let qualified = format!("{}.{}", source.to_ascii_lowercase(), column);
2821    redacted.contains("*") || redacted.contains(&column) || redacted.contains(&qualified)
2822}
2823
2824fn subscription_would_create_cycle(
2825    db: &crate::storage::unified::devx::RedDB,
2826    source: &str,
2827    target: &str,
2828) -> bool {
2829    let mut graph: HashMap<String, Vec<String>> = HashMap::new();
2830    for contract in db.collection_contracts() {
2831        for subscription in contract
2832            .subscriptions
2833            .into_iter()
2834            .filter(|subscription| subscription.enabled)
2835        {
2836            graph
2837                .entry(subscription.source)
2838                .or_default()
2839                .push(subscription.target_queue);
2840        }
2841    }
2842    graph
2843        .entry(source.to_string())
2844        .or_default()
2845        .push(target.to_string());
2846
2847    let mut stack = vec![target.to_string()];
2848    let mut seen = HashSet::new();
2849    while let Some(node) = stack.pop() {
2850        if node == source {
2851            return true;
2852        }
2853        if !seen.insert(node.clone()) {
2854            continue;
2855        }
2856        if let Some(next) = graph.get(&node) {
2857            stack.extend(next.iter().cloned());
2858        }
2859    }
2860    false
2861}
2862
2863pub(crate) fn ensure_event_target_queue_pub(
2864    runtime: &RedDBRuntime,
2865    queue: &str,
2866) -> RedDBResult<()> {
2867    ensure_event_target_queue(runtime, queue)
2868}
2869
2870fn ensure_event_target_queue(runtime: &RedDBRuntime, queue: &str) -> RedDBResult<()> {
2871    let store = runtime.inner.db.store();
2872    if store.get_collection(queue).is_some() {
2873        return Ok(());
2874    }
2875    store
2876        .create_collection(queue)
2877        .map_err(|err| RedDBError::Internal(err.to_string()))?;
2878    runtime
2879        .inner
2880        .db
2881        .save_collection_contract(event_queue_collection_contract(queue))
2882        .map_err(|err| RedDBError::Internal(err.to_string()))?;
2883    store.set_config_tree(
2884        &format!("queue.{queue}.mode"),
2885        &crate::serde_json::Value::String("fanout".to_string()),
2886    );
2887    Ok(())
2888}
2889
2890fn event_queue_collection_contract(queue: &str) -> crate::physical::CollectionContract {
2891    let now = current_unix_ms();
2892    crate::physical::CollectionContract {
2893        name: queue.to_string(),
2894        declared_model: crate::catalog::CollectionModel::Queue,
2895        schema_mode: crate::catalog::SchemaMode::Dynamic,
2896        origin: crate::physical::ContractOrigin::Implicit,
2897        version: 1,
2898        created_at_unix_ms: now,
2899        updated_at_unix_ms: now,
2900        default_ttl_ms: None,
2901        vector_dimension: None,
2902        vector_metric: None,
2903        context_index_fields: Vec::new(),
2904        declared_columns: Vec::new(),
2905        table_def: None,
2906        timestamps_enabled: false,
2907        context_index_enabled: false,
2908        metrics_raw_retention_ms: None,
2909        metrics_rollup_policies: Vec::new(),
2910        metrics_tenant_identity: None,
2911        metrics_namespace: None,
2912        append_only: true,
2913        subscriptions: Vec::new(),
2914        analytics_config: Vec::new(),
2915        session_key: None,
2916        session_gap_ms: None,
2917        retention_duration_ms: None,
2918        analytical_storage: None,
2919
2920        ai_policy: None,
2921    }
2922}
2923
2924fn build_table_def_from_create_table(
2925    query: &CreateTableQuery,
2926) -> RedDBResult<reddb_types::TableDef> {
2927    let mut table = reddb_types::TableDef::new(query.name.clone());
2928    for column in &query.columns {
2929        if column.primary_key {
2930            table.primary_key.push(column.name.clone());
2931            table.constraints.push(
2932                reddb_types::Constraint::new(
2933                    format!("pk_{}", column.name),
2934                    reddb_types::ConstraintType::PrimaryKey,
2935                )
2936                .on_columns(vec![column.name.clone()]),
2937            );
2938        }
2939        if column.unique {
2940            table.constraints.push(
2941                reddb_types::Constraint::new(
2942                    format!("uniq_{}", column.name),
2943                    reddb_types::ConstraintType::Unique,
2944                )
2945                .on_columns(vec![column.name.clone()]),
2946            );
2947        }
2948        if column.not_null {
2949            table.constraints.push(
2950                reddb_types::Constraint::new(
2951                    format!("not_null_{}", column.name),
2952                    reddb_types::ConstraintType::NotNull,
2953                )
2954                .on_columns(vec![column.name.clone()]),
2955            );
2956        }
2957        table.columns.push(column_def_from_ddl(column)?);
2958    }
2959    table.constraints.extend(
2960        analyze_create_table(query)
2961            .map_err(|err| RedDBError::Query(err.to_string()))?
2962            .unique_constraints,
2963    );
2964    // WITH timestamps = true: append the two runtime-managed columns
2965    // to the schema so resolved_contract_columns exposes them to the
2966    // normalize/validate path. Declared as UnsignedInteger (unix-ms),
2967    // not-nullable; the write path auto-fills them.
2968    if query.timestamps {
2969        table.columns.push(
2970            reddb_types::ColumnDef::new(
2971                "created_at".to_string(),
2972                reddb_types::DataType::UnsignedInteger,
2973            )
2974            .not_null(),
2975        );
2976        table.columns.push(
2977            reddb_types::ColumnDef::new(
2978                "updated_at".to_string(),
2979                reddb_types::DataType::UnsignedInteger,
2980            )
2981            .not_null(),
2982        );
2983        table.constraints.push(
2984            reddb_types::Constraint::new(
2985                "not_null_created_at".to_string(),
2986                reddb_types::ConstraintType::NotNull,
2987            )
2988            .on_columns(vec!["created_at".to_string()]),
2989        );
2990        table.constraints.push(
2991            reddb_types::Constraint::new(
2992                "not_null_updated_at".to_string(),
2993                reddb_types::ConstraintType::NotNull,
2994            )
2995            .on_columns(vec!["updated_at".to_string()]),
2996        );
2997    }
2998    table
2999        .validate()
3000        .map_err(|err| RedDBError::Query(format!("invalid table definition: {err}")))?;
3001    Ok(table)
3002}
3003
3004fn column_def_from_ddl(column: &CreateColumnDef) -> RedDBResult<reddb_types::ColumnDef> {
3005    let data_type = resolve_declared_data_type(&column.data_type)
3006        .map_err(|err| RedDBError::Query(err.to_string()))?;
3007    let mut column_def = reddb_types::ColumnDef::new(column.name.clone(), data_type);
3008    if column.not_null {
3009        column_def = column_def.not_null();
3010    }
3011    if let Some(default) = &column.default {
3012        column_def = column_def.with_default(default.as_bytes().to_vec());
3013    }
3014    if column.compress.unwrap_or(0) > 0 {
3015        column_def = column_def.compressed();
3016    }
3017    if !column.enum_variants.is_empty() {
3018        column_def = column_def.with_variants(column.enum_variants.clone());
3019    }
3020    if let Some(precision) = column.decimal_precision {
3021        column_def = column_def.with_precision(precision);
3022    }
3023    if let Some(element_type) = &column.array_element {
3024        column_def = column_def.with_element_type(
3025            resolve_declared_data_type(element_type)
3026                .map_err(|err| RedDBError::Query(err.to_string()))?,
3027        );
3028    }
3029    column_def = column_def.with_metadata("ddl_data_type", column.data_type.clone());
3030    if column.unique {
3031        column_def = column_def.with_metadata("unique", "true");
3032    }
3033    if column.primary_key {
3034        column_def = column_def.with_metadata("primary_key", "true");
3035    }
3036    Ok(column_def)
3037}
3038
3039fn current_unix_ms() -> u128 {
3040    std::time::SystemTime::now()
3041        .duration_since(std::time::UNIX_EPOCH)
3042        .unwrap_or_default()
3043        .as_millis()
3044}
3045
3046#[cfg(test)]
3047mod tests {
3048    use crate::auth::policies::{ActionPattern, Effect, Policy, ResourcePattern, Statement};
3049    use crate::auth::store::{AuthStore, PrincipalRef};
3050    use crate::auth::UserId;
3051    use crate::auth::{AuthConfig, Role};
3052    use crate::runtime::impl_core::{clear_current_auth_identity, set_current_auth_identity};
3053    use crate::{RedDBOptions, RedDBRuntime};
3054    use reddb_types::Value;
3055    use std::sync::Arc;
3056
3057    fn make_allow_policy(id: &str, action: &str, collection: &str) -> Policy {
3058        Policy {
3059            id: id.to_string(),
3060            version: 1,
3061            tenant: None,
3062            created_at: 0,
3063            updated_at: 0,
3064            statements: vec![Statement {
3065                sid: None,
3066                effect: Effect::Allow,
3067                actions: vec![ActionPattern::Exact(action.to_string())],
3068                resources: vec![ResourcePattern::Exact {
3069                    kind: "collection".to_string(),
3070                    name: collection.to_string(),
3071                }],
3072                condition: None,
3073            }],
3074        }
3075    }
3076
3077    fn wire_auth_store(rt: &RedDBRuntime) -> Arc<AuthStore> {
3078        let store = Arc::new(AuthStore::new(AuthConfig::default()));
3079        *rt.inner.auth_store.write() = Some(store.clone());
3080        store
3081    }
3082
3083    #[test]
3084    fn drop_denied_without_iam_policy() {
3085        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3086        rt.execute_query("CREATE TABLE foo (id INT)").unwrap();
3087        let store = wire_auth_store(&rt);
3088        // Put a select-only policy so IAM mode activates, but give alice no drop policy.
3089        let select_only = Policy {
3090            id: "select-only".to_string(),
3091            version: 1,
3092            tenant: None,
3093            created_at: 0,
3094            updated_at: 0,
3095            statements: vec![Statement {
3096                sid: None,
3097                effect: Effect::Allow,
3098                actions: vec![ActionPattern::Exact("select".to_string())],
3099                resources: vec![ResourcePattern::Wildcard],
3100                condition: None,
3101            }],
3102        };
3103        store.put_policy_internal(select_only).unwrap();
3104        let alice = UserId::from_parts(None, "alice");
3105        store
3106            .attach_policy(PrincipalRef::User(alice), "select-only")
3107            .unwrap();
3108        set_current_auth_identity("alice".to_string(), Role::Write);
3109        let err = rt.execute_query("DROP TABLE foo").unwrap_err();
3110        clear_current_auth_identity();
3111        assert!(
3112            format!("{err}").contains("denied by IAM policy"),
3113            "got: {err}"
3114        );
3115    }
3116
3117    #[test]
3118    fn drop_allowed_with_explicit_iam_policy() {
3119        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3120        rt.execute_query("CREATE TABLE bar (id INT)").unwrap();
3121        let store = wire_auth_store(&rt);
3122        let policy = make_allow_policy("allow-drop-bar", "drop", "bar");
3123        store.put_policy_internal(policy).unwrap();
3124        let bob = UserId::from_parts(None, "bob");
3125        store
3126            .attach_policy(PrincipalRef::User(bob), "allow-drop-bar")
3127            .unwrap();
3128        set_current_auth_identity("bob".to_string(), Role::Write);
3129        rt.execute_query("DROP TABLE bar").unwrap();
3130        clear_current_auth_identity();
3131    }
3132
3133    #[test]
3134    fn drop_allowed_with_wildcard_iam_policy() {
3135        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3136        rt.execute_query("CREATE TABLE baz (id INT)").unwrap();
3137        let store = wire_auth_store(&rt);
3138        let policy = Policy {
3139            id: "allow-drop-all".to_string(),
3140            version: 1,
3141            tenant: None,
3142            created_at: 0,
3143            updated_at: 0,
3144            statements: vec![Statement {
3145                sid: None,
3146                effect: Effect::Allow,
3147                actions: vec![ActionPattern::Exact("drop".to_string())],
3148                resources: vec![ResourcePattern::Wildcard],
3149                condition: None,
3150            }],
3151        };
3152        store.put_policy_internal(policy).unwrap();
3153        let carl = UserId::from_parts(None, "carl");
3154        store
3155            .attach_policy(PrincipalRef::User(carl), "allow-drop-all")
3156            .unwrap();
3157        set_current_auth_identity("carl".to_string(), Role::Write);
3158        rt.execute_query("DROP TABLE baz").unwrap();
3159        clear_current_auth_identity();
3160    }
3161
3162    #[test]
3163    fn truncate_denied_without_iam_policy() {
3164        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3165        rt.execute_query("CREATE TABLE qux (id INT)").unwrap();
3166        let store = wire_auth_store(&rt);
3167        // Acceptance #2 (#712 / S5A): in `policy_only` mode a
3168        // principal with no matching policy is denied even if their
3169        // role would have permitted the action. The pre-#712 default
3170        // of "no policy → deny" only holds under `policy_only`, so
3171        // pin the mode explicitly here so the assertion holds
3172        // regardless of the construction-time default.
3173        store
3174            .set_enforcement_mode(crate::auth::enforcement_mode::PolicyEnforcementMode::PolicyOnly);
3175        // A policy exists (IAM active) but gives no truncate right.
3176        let select_only = Policy {
3177            id: "select-only-2".to_string(),
3178            version: 1,
3179            tenant: None,
3180            created_at: 0,
3181            updated_at: 0,
3182            statements: vec![Statement {
3183                sid: None,
3184                effect: Effect::Allow,
3185                actions: vec![ActionPattern::Exact("select".to_string())],
3186                resources: vec![ResourcePattern::Wildcard],
3187                condition: None,
3188            }],
3189        };
3190        store.put_policy_internal(select_only).unwrap();
3191        let dana = UserId::from_parts(None, "dana");
3192        store
3193            .attach_policy(PrincipalRef::User(dana), "select-only-2")
3194            .unwrap();
3195        set_current_auth_identity("dana".to_string(), Role::Write);
3196        let err = rt.execute_query("TRUNCATE TABLE qux").unwrap_err();
3197        clear_current_auth_identity();
3198        assert!(
3199            format!("{err}").contains("denied by IAM policy"),
3200            "got: {err}"
3201        );
3202    }
3203
3204    #[test]
3205    fn truncate_table_clears_rows_and_preserves_schema_and_indexes() {
3206        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3207        rt.execute_query("CREATE TABLE users (id INT, name TEXT)")
3208            .unwrap();
3209        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'ana'), (2, 'bob')")
3210            .unwrap();
3211        rt.execute_query("CREATE INDEX idx_users_id ON users (id) USING HASH")
3212            .unwrap();
3213
3214        let truncated = rt.execute_query("TRUNCATE TABLE users").unwrap();
3215        assert_eq!(truncated.statement_type, "truncate");
3216        assert_eq!(truncated.affected_rows, 0);
3217
3218        let empty = rt.execute_query("SELECT id FROM users").unwrap();
3219        assert!(empty.result.records.is_empty());
3220
3221        rt.execute_query("INSERT INTO users (id, name) VALUES (3, 'cy')")
3222            .unwrap();
3223        let selected = rt
3224            .execute_query("SELECT name FROM users WHERE id = 3")
3225            .unwrap();
3226        let name = selected.result.records[0].get("name").unwrap();
3227        assert_eq!(name, &Value::text("cy"));
3228        assert!(rt.db().collection_contract("users").is_some());
3229        assert!(rt
3230            .inner
3231            .index_store
3232            .list_indices("users")
3233            .iter()
3234            .any(|index| index.name == "idx_users_id"));
3235    }
3236
3237    #[test]
3238    fn truncate_collection_is_polymorphic_and_typed_mismatch_fails() {
3239        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3240        rt.execute_query("CREATE QUEUE tasks").unwrap();
3241        rt.execute_query("QUEUE PUSH tasks {'job':'a'}").unwrap();
3242
3243        let err = rt.execute_query("TRUNCATE TABLE tasks").unwrap_err();
3244        assert!(format!("{err}").contains("model mismatch: expected table, got queue"));
3245
3246        rt.execute_query("TRUNCATE COLLECTION tasks").unwrap();
3247        let len = rt.execute_query("QUEUE LEN tasks").unwrap();
3248        assert_eq!(
3249            len.result.records[0].get("len"),
3250            Some(&Value::UnsignedInteger(0))
3251        );
3252    }
3253
3254    #[test]
3255    fn truncate_system_schema_is_read_only() {
3256        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3257        let err = rt
3258            .execute_query("TRUNCATE COLLECTION red.collections")
3259            .unwrap_err();
3260        assert!(format!("{err}").contains("system schema is read-only"));
3261    }
3262
3263    // ── #302 / #310: TRUNCATE / DROP single-event semantics ────────────────
3264
3265    fn queue_payloads(rt: &RedDBRuntime, queue: &str) -> Vec<crate::json::Value> {
3266        let result = rt
3267            .execute_query(&format!("QUEUE PEEK {queue} 100"))
3268            .expect("peek queue");
3269        result
3270            .result
3271            .records
3272            .iter()
3273            .map(
3274                |record| match record.get("payload").expect("payload column") {
3275                    Value::Json(bytes) => crate::json::from_slice(bytes).expect("json payload"),
3276                    other => panic!("expected JSON queue payload, got {other:?}"),
3277                },
3278            )
3279            .collect()
3280    }
3281
3282    /// `TRUNCATE users` on an event-enabled collection emits exactly 1
3283    /// `truncate` event, not one delete event per row.
3284    #[test]
3285    fn truncate_event_enabled_table_emits_single_truncate_event() {
3286        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3287        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3288            .unwrap();
3289        rt.execute_query(
3290            "INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob'), (3, 'carol')",
3291        )
3292        .unwrap();
3293
3294        // Drain the 3 insert events so we start clean.
3295        rt.execute_query("QUEUE POP users_events COUNT 10").unwrap();
3296
3297        rt.execute_query("TRUNCATE TABLE users").unwrap();
3298
3299        let events = queue_payloads(&rt, "users_events");
3300        // Must be exactly 1 truncate event, not 3 delete events.
3301        assert_eq!(
3302            events.len(),
3303            1,
3304            "expected 1 truncate event, got {}",
3305            events.len()
3306        );
3307        let ev = events[0].as_object().expect("event is object");
3308        assert_eq!(
3309            ev.get("op").and_then(crate::json::Value::as_str),
3310            Some("truncate")
3311        );
3312        assert_eq!(
3313            ev.get("collection").and_then(crate::json::Value::as_str),
3314            Some("users")
3315        );
3316        assert_eq!(
3317            ev.get("entities_count")
3318                .and_then(crate::json::Value::as_u64),
3319            Some(3)
3320        );
3321        assert!(ev.get("ts").and_then(crate::json::Value::as_u64).is_some());
3322        assert!(ev.get("lsn").and_then(crate::json::Value::as_u64).is_some());
3323        assert!(ev
3324            .get("event_id")
3325            .and_then(crate::json::Value::as_str)
3326            .is_some_and(|s| !s.is_empty()));
3327    }
3328
3329    /// `TRUNCATE users` on a collection without event subscription emits no events.
3330    #[test]
3331    fn truncate_no_events_collection_emits_nothing() {
3332        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3333        rt.execute_query("CREATE TABLE plain (id INT, val TEXT)")
3334            .unwrap();
3335        rt.execute_query("INSERT INTO plain (id, val) VALUES (1, 'a'), (2, 'b')")
3336            .unwrap();
3337        // No EVENTS subscription — truncate must work without touching any queue.
3338        rt.execute_query("TRUNCATE TABLE plain").unwrap();
3339        // No crash, no queue to check. Just verify truncation happened.
3340        let rows = rt.execute_query("SELECT id FROM plain").unwrap();
3341        assert!(rows.result.records.is_empty());
3342    }
3343
3344    /// `DROP TABLE users` on an event-enabled collection emits exactly 1
3345    /// `collection_dropped` event. The subscription is removed from the
3346    /// source contract but the target queue is preserved for consumer drain.
3347    #[test]
3348    fn drop_event_enabled_table_emits_single_collection_dropped_event() {
3349        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3350        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3351            .unwrap();
3352        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob')")
3353            .unwrap();
3354
3355        // Drain insert events so we start clean.
3356        rt.execute_query("QUEUE POP users_events COUNT 10").unwrap();
3357
3358        rt.execute_query("DROP TABLE users").unwrap();
3359
3360        // Queue must still exist with 1 collection_dropped event.
3361        let events = queue_payloads(&rt, "users_events");
3362        assert_eq!(
3363            events.len(),
3364            1,
3365            "expected 1 collection_dropped event, got {}",
3366            events.len()
3367        );
3368        let ev = events[0].as_object().expect("event is object");
3369        assert_eq!(
3370            ev.get("op").and_then(crate::json::Value::as_str),
3371            Some("collection_dropped")
3372        );
3373        assert_eq!(
3374            ev.get("collection").and_then(crate::json::Value::as_str),
3375            Some("users")
3376        );
3377        assert_eq!(
3378            ev.get("final_entities_count")
3379                .and_then(crate::json::Value::as_u64),
3380            Some(2)
3381        );
3382        assert!(ev.get("ts").and_then(crate::json::Value::as_u64).is_some());
3383        assert!(ev.get("lsn").and_then(crate::json::Value::as_u64).is_some());
3384        assert!(ev
3385            .get("event_id")
3386            .and_then(crate::json::Value::as_str)
3387            .is_some_and(|s| !s.is_empty()));
3388
3389        // Source collection is gone.
3390        let err = rt.execute_query("SELECT id FROM users").unwrap_err();
3391        assert!(
3392            format!("{err}").contains("users"),
3393            "expected not-found error"
3394        );
3395    }
3396
3397    /// `DROP TABLE users` on a collection without event subscription works
3398    /// normally with no event emitted.
3399    #[test]
3400    fn drop_no_events_collection_emits_nothing() {
3401        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3402        rt.execute_query("CREATE TABLE plain (id INT, val TEXT)")
3403            .unwrap();
3404        rt.execute_query("INSERT INTO plain (id, val) VALUES (1, 'a')")
3405            .unwrap();
3406        rt.execute_query("DROP TABLE plain").unwrap();
3407        // No crash and collection is gone.
3408        let err = rt.execute_query("SELECT id FROM plain").unwrap_err();
3409        assert!(format!("{err}").contains("plain"));
3410    }
3411
3412    // ── #297: ops_filter + WHERE filter ────────────────────────────────────
3413
3414    /// `WITH EVENTS (INSERT)` — UPDATE and DELETE events must NOT be emitted.
3415    #[test]
3416    fn ops_filter_insert_only_ignores_update_and_delete() {
3417        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3418        rt.execute_query(
3419            "CREATE TABLE items (id INT, val TEXT) WITH EVENTS (INSERT) TO items_events",
3420        )
3421        .unwrap();
3422        rt.execute_query("INSERT INTO items (id, val) VALUES (1, 'a')")
3423            .unwrap();
3424        rt.execute_query("UPDATE items SET val = 'b' WHERE id = 1")
3425            .unwrap();
3426        rt.execute_query("DELETE FROM items WHERE id = 1").unwrap();
3427
3428        let events = queue_payloads(&rt, "items_events");
3429        // Only the INSERT should have fired.
3430        assert_eq!(
3431            events.len(),
3432            1,
3433            "expected 1 insert event, got {}",
3434            events.len()
3435        );
3436        assert_eq!(
3437            events[0]
3438                .as_object()
3439                .unwrap()
3440                .get("op")
3441                .and_then(crate::json::Value::as_str),
3442            Some("insert")
3443        );
3444    }
3445
3446    /// `WITH EVENTS WHERE status = 'active'` — only rows matching the predicate generate events.
3447    #[test]
3448    fn where_filter_skips_rows_that_do_not_match() {
3449        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3450        rt.execute_query(
3451            "CREATE TABLE users (id INT, status TEXT) WITH EVENTS WHERE status = 'active' TO users_events",
3452        )
3453        .unwrap();
3454
3455        // This row should generate an event.
3456        rt.execute_query("INSERT INTO users (id, status) VALUES (1, 'active')")
3457            .unwrap();
3458        // This row should NOT generate an event.
3459        rt.execute_query("INSERT INTO users (id, status) VALUES (2, 'inactive')")
3460            .unwrap();
3461
3462        let events = queue_payloads(&rt, "users_events");
3463        assert_eq!(
3464            events.len(),
3465            1,
3466            "expected 1 event (only active), got {}",
3467            events.len()
3468        );
3469        let ev = events[0].as_object().unwrap();
3470        assert_eq!(
3471            ev.get("op").and_then(crate::json::Value::as_str),
3472            Some("insert")
3473        );
3474        let after = ev.get("after").unwrap().as_object().unwrap();
3475        assert_eq!(
3476            after.get("status").and_then(crate::json::Value::as_str),
3477            Some("active")
3478        );
3479    }
3480
3481    /// `WITH EVENTS (INSERT, UPDATE) WHERE status = 'active'` — combination functional.
3482    #[test]
3483    fn ops_filter_and_where_filter_combined() {
3484        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3485        rt.execute_query(
3486            "CREATE TABLE items (id INT, status TEXT) WITH EVENTS (INSERT, UPDATE) WHERE status = 'active' TO items_events",
3487        )
3488        .unwrap();
3489
3490        // INSERT active → event
3491        rt.execute_query("INSERT INTO items (id, status) VALUES (1, 'active')")
3492            .unwrap();
3493        // INSERT inactive → no event
3494        rt.execute_query("INSERT INTO items (id, status) VALUES (2, 'inactive')")
3495            .unwrap();
3496        // UPDATE row 1 to inactive → after = inactive, no event
3497        rt.execute_query("UPDATE items SET status = 'inactive' WHERE id = 1")
3498            .unwrap();
3499        // DELETE → ops_filter excludes it
3500        rt.execute_query("DELETE FROM items WHERE id = 2").unwrap();
3501
3502        let events = queue_payloads(&rt, "items_events");
3503        // Only the first INSERT (active) fires; UPDATE result is inactive so skipped; DELETE excluded by ops_filter.
3504        assert_eq!(
3505            events.len(),
3506            1,
3507            "expected 1 event, got {}: {events:?}",
3508            events.len()
3509        );
3510        assert_eq!(
3511            events[0]
3512                .as_object()
3513                .unwrap()
3514                .get("op")
3515                .and_then(crate::json::Value::as_str),
3516            Some("insert")
3517        );
3518    }
3519
3520    /// WHERE filter on DELETE events — the before-state (pre-image) is evaluated.
3521    #[test]
3522    fn where_filter_on_delete_checks_before_state() {
3523        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3524        rt.execute_query(
3525            "CREATE TABLE users (id INT, status TEXT) WITH EVENTS (DELETE) WHERE status = 'active' TO users_events",
3526        )
3527        .unwrap();
3528
3529        rt.execute_query("INSERT INTO users (id, status) VALUES (1, 'active'), (2, 'inactive')")
3530            .unwrap();
3531
3532        // Delete active row → event (before-state was active)
3533        rt.execute_query("DELETE FROM users WHERE id = 1").unwrap();
3534        // Delete inactive row → no event (before-state was inactive)
3535        rt.execute_query("DELETE FROM users WHERE id = 2").unwrap();
3536
3537        let events = queue_payloads(&rt, "users_events");
3538        assert_eq!(
3539            events.len(),
3540            1,
3541            "expected 1 delete event, got {}",
3542            events.len()
3543        );
3544        let ev = events[0].as_object().unwrap();
3545        assert_eq!(
3546            ev.get("op").and_then(crate::json::Value::as_str),
3547            Some("delete")
3548        );
3549    }
3550
3551    // ── #301: schema evolution OperatorEvent on ALTER ───────────────────────
3552
3553    /// ADD COLUMN on event-enabled table must succeed (OperatorEvent is best-effort).
3554    #[test]
3555    fn alter_add_column_on_event_enabled_table_succeeds() {
3556        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3557        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO users_events")
3558            .unwrap();
3559        // Must not error — OperatorEvent emission is best-effort (no global sink in tests).
3560        rt.execute_query("ALTER TABLE users ADD COLUMN phone TEXT")
3561            .unwrap();
3562        // The column is now in the contract.
3563        let contract = rt.db().collection_contract("users").unwrap();
3564        assert!(
3565            contract.declared_columns.iter().any(|c| c.name == "phone"),
3566            "phone column should be in contract"
3567        );
3568        // Subscription still enabled after the alter.
3569        assert!(
3570            contract.subscriptions.iter().any(|s| s.enabled),
3571            "subscription should remain enabled"
3572        );
3573    }
3574
3575    /// DROP COLUMN on event-enabled table must succeed; non-column ALTERs
3576    /// (like ENABLE ROW LEVEL SECURITY) must also succeed without emitting.
3577    #[test]
3578    fn alter_drop_column_and_rls_on_event_enabled_table_succeeds() {
3579        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3580        rt.execute_query(
3581            "CREATE TABLE items (id INT, secret TEXT, status TEXT) WITH EVENTS TO items_events",
3582        )
3583        .unwrap();
3584        // DROP COLUMN — schema change event path exercises, must not error.
3585        rt.execute_query("ALTER TABLE items DROP COLUMN secret")
3586            .unwrap();
3587        let contract = rt.db().collection_contract("items").unwrap();
3588        assert!(
3589            !contract.declared_columns.iter().any(|c| c.name == "secret"),
3590            "secret column should be removed"
3591        );
3592        // ENABLE RLS — non-column op, no schema-change event (coverage).
3593        rt.execute_query("ALTER TABLE items ENABLE ROW LEVEL SECURITY")
3594            .unwrap();
3595        // Collection and subscription still intact.
3596        assert!(
3597            contract.subscriptions.iter().any(|s| s.enabled),
3598            "subscription should remain enabled"
3599        );
3600    }
3601
3602    /// Slice G (#675) — every newly-created `CollectionModel::Vector`
3603    /// collection is marked `vector.turbo` and gains a materialised
3604    /// `TurboCollectionState`. The marker is what the SEARCH/INSERT
3605    /// hot paths read to branch between TurboQuant and the legacy
3606    /// brute-force fallback.
3607    #[test]
3608    fn create_vector_marks_collection_as_turbo_baseline() {
3609        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3610        rt.execute_query("CREATE VECTOR embeddings DIM 4").unwrap();
3611        let store = rt.db().store();
3612        assert!(
3613            crate::runtime::vector_turbo_kind::is_turbo(&store, "embeddings"),
3614            "new vector collections must be turbo-marked baseline"
3615        );
3616        assert!(
3617            rt.db().turbo_state("embeddings").is_some(),
3618            "turbo_state must materialise after CREATE VECTOR"
3619        );
3620    }
3621
3622    /// Non-vector collections must NOT be turbo-marked. Guards against
3623    /// the always-baseline change accidentally leaking the marker onto
3624    /// CREATE TABLE / CREATE DOCUMENT / etc.
3625    #[test]
3626    fn create_table_does_not_mark_turbo() {
3627        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3628        rt.execute_query("CREATE TABLE plain (id INT)").unwrap();
3629        let store = rt.db().store();
3630        assert!(
3631            !crate::runtime::vector_turbo_kind::is_turbo(&store, "plain"),
3632            "non-vector collections must not gain the turbo marker"
3633        );
3634        assert!(rt.db().turbo_state("plain").is_none());
3635    }
3636
3637    /// `CREATE COLLECTION KIND vector.turbo` continues to mark the
3638    /// collection (idempotent with the new always-baseline path).
3639    #[test]
3640    fn create_collection_kind_vector_turbo_still_marked() {
3641        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3642        rt.execute_query("CREATE COLLECTION turbo_v KIND vector.turbo DIM 4")
3643            .unwrap();
3644        let store = rt.db().store();
3645        assert!(crate::runtime::vector_turbo_kind::is_turbo(
3646            &store, "turbo_v"
3647        ));
3648        assert!(rt.db().turbo_state("turbo_v").is_some());
3649    }
3650
3651    /// Issue #1271: a declared AI policy is persisted in the catalog and
3652    /// reads back identical via the collection contract (introspection).
3653    #[test]
3654    fn create_table_persists_and_introspects_ai_policy() {
3655        use crate::catalog::{ModerateDegradedMode, ModerateRejectAction};
3656        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3657        rt.execute_query(
3658            "CREATE TABLE posts (id INT, title TEXT, body TEXT, photo TEXT) WITH ( \
3659               EMBED (fields = ('title', 'body'), provider = 'openai', model = 'text-embedding-3-small'), \
3660               MODERATE (fields = ('body'), provider = 'openai', model = 'omni-moderation-latest', sync = true, degraded = closed, on_reject = flag), \
3661               VISION (image_field = 'photo', outputs = ('caption'), provider = 'openai', model = 'gpt-4o') \
3662             )",
3663        )
3664        .expect("create table with ai policy");
3665
3666        let contracts = rt.db().collection_contracts();
3667        let contract = contracts
3668            .iter()
3669            .find(|c| c.name == "posts")
3670            .expect("contract persisted");
3671        let policy = contract.ai_policy.as_ref().expect("ai policy persisted");
3672
3673        let embed = policy.embed.as_ref().expect("embed");
3674        assert_eq!(embed.fields, vec!["title".to_string(), "body".to_string()]);
3675        assert_eq!(embed.model, "text-embedding-3-small");
3676
3677        let moderate = policy.moderate.as_ref().expect("moderate");
3678        assert!(moderate.sync_gate);
3679        assert_eq!(moderate.degraded_mode, ModerateDegradedMode::Closed);
3680        assert_eq!(moderate.reject_action, ModerateRejectAction::Flag);
3681
3682        let vision = policy.vision.as_ref().expect("vision");
3683        assert_eq!(vision.image_field, "photo");
3684        assert_eq!(vision.model, "gpt-4o");
3685    }
3686
3687    /// Issue #1271 / #1269: a policy wiring a provider to a modality it
3688    /// cannot serve is rejected at DDL time, and the collection is not
3689    /// created. Anthropic has no embeddings product in the matrix.
3690    #[test]
3691    fn create_table_rejects_ai_policy_incapable_provider() {
3692        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
3693        let err = rt
3694            .execute_query(
3695                "CREATE TABLE bad (id INT, body TEXT) WITH ( \
3696                   EMBED (fields = ('body'), provider = 'anthropic', model = 'claude-3-5-sonnet') \
3697                 )",
3698            )
3699            .unwrap_err();
3700        assert!(
3701            err.to_string()
3702                .contains("cannot serve the 'embed' modality"),
3703            "{err}"
3704        );
3705        assert!(
3706            rt.db().store().get_collection("bad").is_none(),
3707            "rejected DDL must not create the collection"
3708        );
3709    }
3710}