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