1use std::collections::HashMap;
4use std::sync::Arc;
5
6use crate::catalog::{CollectionModel, SchemaMode};
7use crate::physical::{CollectionContract, ContractOrigin};
8use crate::storage::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
9use reddb_rql::ast::ConfigValueType;
10
11use super::impl_core::{current_auth_identity, current_connection_id, current_tenant};
12use super::*;
13
14const CONFIG_HISTORY_LIMIT: usize = 16;
15
16#[derive(Clone)]
17struct ConfigVersion {
18 id: EntityId,
19 key: String,
20 version: i64,
21 value: Value,
22 tombstone: bool,
23 created_at_ms: i64,
24 op: String,
25 value_type: Option<ConfigValueType>,
26 schema_version: Option<i64>,
27 tags: Vec<String>,
28}
29
30impl super::keyed_spine::KeyedVersion for ConfigVersion {
31 fn key(&self) -> &str {
32 &self.key
33 }
34
35 fn version(&self) -> i64 {
36 self.version
37 }
38}
39
40impl ConfigVersion {
41 fn from_keyed_row(version: super::keyed_spine::KeyedRowVersion, row: &RowData) -> Self {
42 Self {
43 id: version.id,
44 key: version.key,
45 version: version.version,
46 value: version.value,
47 tombstone: version.tombstone,
48 created_at_ms: version.created_at_ms,
49 op: version.op,
50 value_type: row
51 .get_field("value_type")
52 .and_then(config_value_type_from_value),
53 schema_version: super::keyed_spine::value_i64(row.get_field("schema_version")),
54 tags: config_tags_from_value(row.get_field("tags")),
55 }
56 }
57}
58
59struct ConfigSecretRef {
60 collection: String,
61 key: String,
62}
63
64struct ConfigMutationEvidence {
65 id: String,
66 resource_type: String,
67 managed: bool,
68 mutability: crate::auth::registry::Mutability,
69 matched_action: Option<String>,
70 matched_resource: Option<String>,
71 payload: Option<Value>,
72}
73
74enum ConfigMutationAuthz {
75 Allowed(ConfigMutationEvidence),
76 Denied {
77 reason: String,
78 evidence: ConfigMutationEvidence,
79 },
80}
81
82impl RedDBRuntime {
83 pub fn execute_config_command(
84 &self,
85 raw_query: &str,
86 cmd: &reddb_rql::ast::ConfigCommand,
87 ) -> RedDBResult<RuntimeQueryResult> {
88 use reddb_rql::ast::ConfigCommand;
89
90 match cmd {
91 ConfigCommand::Put {
92 collection,
93 key,
94 value,
95 value_type,
96 tags,
97 } => self.config_write_result(
98 raw_query,
99 collection,
100 key,
101 value.clone(),
102 *value_type,
103 tags,
104 "put",
105 ),
106 ConfigCommand::Rotate {
107 collection,
108 key,
109 value,
110 value_type,
111 tags,
112 } => self.config_write_result(
113 raw_query,
114 collection,
115 key,
116 value.clone(),
117 *value_type,
118 tags,
119 "rotate",
120 ),
121 ConfigCommand::Get { collection, key } => {
122 self.config_get_result(raw_query, collection, key)
123 }
124 ConfigCommand::Resolve { collection, key } => {
125 self.config_resolve_result(raw_query, collection, key)
126 }
127 ConfigCommand::Delete { collection, key } => {
128 self.config_delete_result(raw_query, collection, key)
129 }
130 ConfigCommand::History { collection, key } => {
131 self.config_history_result(raw_query, collection, key)
132 }
133 ConfigCommand::List {
134 collection,
135 prefix,
136 limit,
137 offset,
138 } => self.config_list_result(raw_query, collection, prefix.as_deref(), *limit, *offset),
139 ConfigCommand::Watch {
140 collection,
141 key,
142 prefix,
143 from_lsn,
144 } => self.config_watch_result(raw_query, collection, key, *prefix, *from_lsn),
145 ConfigCommand::InvalidVolatileOperation { operation, .. } => {
146 Err(invalid_config_volatility(operation))
147 }
148 }
149 }
150
151 pub(crate) fn validate_config_command_before_auth(
152 &self,
153 cmd: &reddb_rql::ast::ConfigCommand,
154 ) -> RedDBResult<()> {
155 use reddb_rql::ast::ConfigCommand;
156 match cmd {
157 ConfigCommand::InvalidVolatileOperation { operation, .. } => {
158 Err(invalid_config_volatility(operation))
159 }
160 ConfigCommand::Put { collection, .. }
161 | ConfigCommand::Get { collection, .. }
162 | ConfigCommand::Resolve { collection, .. }
163 | ConfigCommand::Rotate { collection, .. }
164 | ConfigCommand::Delete { collection, .. }
165 | ConfigCommand::History { collection, .. }
166 | ConfigCommand::List { collection, .. }
167 | ConfigCommand::Watch { collection, .. } => {
168 let snapshot = self.inner.db.catalog_model_snapshot();
169 let Some(actual_model) = snapshot
170 .collections
171 .iter()
172 .find(|c| c.name == *collection)
173 .map(|c| c.declared_model.unwrap_or(c.model))
174 else {
175 return Ok(());
176 };
177 crate::runtime::ddl::polymorphic_resolver::ensure_model_match(
178 CollectionModel::Config,
179 actual_model,
180 )
181 }
182 }
183 }
184
185 fn config_resolve_result(
186 &self,
187 raw_query: &str,
188 collection: &str,
189 key: &str,
190 ) -> RedDBResult<RuntimeQueryResult> {
191 let latest = self.latest_config_version(collection, key)?;
192 if let Err(reason) = self.check_config_capability("config:read", collection, key) {
193 self.audit_config_resolve(
194 collection,
195 key,
196 None,
197 crate::runtime::audit_log::Outcome::Denied,
198 &reason,
199 );
200 return Err(RedDBError::Query(reason));
201 }
202
203 let Some(version) = latest else {
204 let reason = "not_found";
205 self.audit_config_resolve(
206 collection,
207 key,
208 None,
209 crate::runtime::audit_log::Outcome::Denied,
210 reason,
211 );
212 return Err(RedDBError::NotFound(format!(
213 "config '{}.{}' not found",
214 collection, key
215 )));
216 };
217 if version.tombstone {
218 let reason = "deleted";
219 self.audit_config_resolve(
220 collection,
221 key,
222 None,
223 crate::runtime::audit_log::Outcome::Denied,
224 reason,
225 );
226 return Err(RedDBError::NotFound(format!(
227 "config '{}.{}' is deleted",
228 collection, key
229 )));
230 }
231
232 let secret_ref = parse_config_secret_ref(&version.value).inspect_err(|err| {
233 self.audit_config_resolve(
234 collection,
235 key,
236 None,
237 crate::runtime::audit_log::Outcome::Error,
238 &err.to_string(),
239 );
240 })?;
241
242 match self.resolve_vault_secret_value(&secret_ref.collection, &secret_ref.key) {
243 Ok(value) => {
244 if value_looks_like_secret_ref(&value) {
245 let err = secret_ref_chain_error(
246 collection,
247 key,
248 &secret_ref.collection,
249 &secret_ref.key,
250 );
251 let reason = err.to_string();
252 self.audit_config_resolve(
253 collection,
254 key,
255 Some(&secret_ref),
256 crate::runtime::audit_log::Outcome::Error,
257 &reason,
258 );
259 return Err(err);
260 }
261 self.audit_config_resolve(
262 collection,
263 key,
264 Some(&secret_ref),
265 crate::runtime::audit_log::Outcome::Success,
266 "ok",
267 );
268 let mut result = UnifiedResult::with_columns(vec![
269 "collection".into(),
270 "key".into(),
271 "value".into(),
272 "resolved_store".into(),
273 "resolved_collection".into(),
274 "resolved_key".into(),
275 ]);
276 let mut record = UnifiedRecord::new();
277 record.set("collection", Value::text(collection.to_string()));
278 record.set("key", Value::text(key.to_string()));
279 record.set("value", value);
280 record.set("resolved_store", Value::text("vault"));
281 record.set("resolved_collection", Value::text(secret_ref.collection));
282 record.set("resolved_key", Value::text(secret_ref.key));
283 result.push(record);
284 Ok(RuntimeQueryResult {
285 query: raw_query.to_string(),
286 mode: reddb_rql::modes::QueryMode::Sql,
287 statement: "config_resolve",
288 engine: "config",
289 result,
290 affected_rows: 0,
291 statement_type: "select",
292 bookmark: None,
293 notice: None,
294 })
295 }
296 Err(err) => {
297 let reason = err.to_string();
298 let outcome = if reason.contains("denied") {
299 crate::runtime::audit_log::Outcome::Denied
300 } else {
301 crate::runtime::audit_log::Outcome::Error
302 };
303 self.audit_config_resolve(collection, key, Some(&secret_ref), outcome, &reason);
304 Err(err)
305 }
306 }
307 }
308
309 fn config_write_result(
310 &self,
311 raw_query: &str,
312 collection: &str,
313 key: &str,
314 value: Value,
315 requested_type: Option<ConfigValueType>,
316 tags: &[String],
317 op: &str,
318 ) -> RedDBResult<RuntimeQueryResult> {
319 let mut evidence = match self.authorize_config_write_for_event(collection, key) {
320 ConfigMutationAuthz::Allowed(evidence) => evidence,
321 ConfigMutationAuthz::Denied {
322 reason,
323 mut evidence,
324 } => {
325 evidence.payload = Some(value.clone());
326 let _ = self.emit_config_mutation_event(
327 crate::runtime::control_events::EventKind::ConfigWrite,
328 crate::runtime::control_events::Outcome::Denied,
329 "config:write",
330 collection,
331 key,
332 Some(reason.clone()),
333 &evidence,
334 );
335 return Err(RedDBError::Query(reason));
336 }
337 };
338 if let Err(err) = self.check_write(crate::runtime::write_gate::WriteKind::Dml) {
339 let _ = self.emit_config_mutation_event(
340 crate::runtime::control_events::EventKind::ConfigWrite,
341 crate::runtime::control_events::Outcome::Error,
342 "config:write",
343 collection,
344 key,
345 Some(err.to_string()),
346 &evidence,
347 );
348 return Err(err);
349 }
350 if is_enforcement_mode_config(collection, key) {
355 if let Err(err) = validate_enforcement_mode_value(&value) {
356 let _ = self.emit_config_mutation_event(
357 crate::runtime::control_events::EventKind::ConfigWrite,
358 crate::runtime::control_events::Outcome::Denied,
359 "config:write",
360 collection,
361 key,
362 Some(err.to_string()),
363 &evidence,
364 );
365 return Err(err);
366 }
367 }
368 if let Err(err) = self.ensure_config_collection(collection) {
369 let _ = self.emit_config_mutation_event(
370 crate::runtime::control_events::EventKind::ConfigWrite,
371 crate::runtime::control_events::Outcome::Error,
372 "config:write",
373 collection,
374 key,
375 Some(err.to_string()),
376 &evidence,
377 );
378 return Err(err);
379 }
380 let latest = match self.latest_config_version(collection, key) {
381 Ok(latest) => latest,
382 Err(err) => {
383 let _ = self.emit_config_mutation_event(
384 crate::runtime::control_events::EventKind::ConfigWrite,
385 crate::runtime::control_events::Outcome::Error,
386 "config:write",
387 collection,
388 key,
389 Some(err.to_string()),
390 &evidence,
391 );
392 return Err(err);
393 }
394 };
395 let version = latest.as_ref().map(|version| version.version).unwrap_or(0) + 1;
396 let (value_type, schema_version) = resolve_config_schema(latest.as_ref(), requested_type);
397 if let Some(value_type) = value_type {
398 if let Err(err) = validate_config_value_type(&value, value_type) {
399 let _ = self.emit_config_mutation_event(
400 crate::runtime::control_events::EventKind::ConfigWrite,
401 crate::runtime::control_events::Outcome::Error,
402 "config:write",
403 collection,
404 key,
405 Some(err.to_string()),
406 &evidence,
407 );
408 return Err(err);
409 }
410 }
411 evidence.payload = Some(value.clone());
412 if let Some(reason) = self.secret_ref_guard_write_check(collection, key, &value) {
413 let _ = self.emit_config_mutation_event(
414 crate::runtime::control_events::EventKind::ConfigWrite,
415 crate::runtime::control_events::Outcome::Denied,
416 "config:write",
417 collection,
418 key,
419 Some(reason.to_string()),
420 &evidence,
421 );
422 return Err(reason);
423 }
424 let before = latest.as_ref().and_then(|version| {
425 if version.tombstone {
426 None
427 } else {
428 Some(reddb_types::Value::to_json(&version.value))
429 }
430 });
431 let after = Some(reddb_types::Value::to_json(&value));
432 let change_op = if latest.is_some() {
433 crate::replication::cdc::ChangeOperation::Update
434 } else {
435 crate::replication::cdc::ChangeOperation::Insert
436 };
437 let id = match self.append_config_version(
438 collection,
439 key,
440 value,
441 version,
442 false,
443 op,
444 value_type,
445 schema_version,
446 tags,
447 ) {
448 Ok(id) => id,
449 Err(err) => {
450 let _ = self.emit_config_mutation_event(
451 crate::runtime::control_events::EventKind::ConfigWrite,
452 crate::runtime::control_events::Outcome::Error,
453 "config:write",
454 collection,
455 key,
456 Some(err.to_string()),
457 &evidence,
458 );
459 return Err(err);
460 }
461 };
462 self.record_kv_watch_event(change_op, collection, key, id.raw(), before, after);
463 if let Err(err) = self.prune_config_history(collection, key) {
464 let _ = self.emit_config_mutation_event(
465 crate::runtime::control_events::EventKind::ConfigWrite,
466 crate::runtime::control_events::Outcome::Error,
467 "config:write",
468 collection,
469 key,
470 Some(err.to_string()),
471 &evidence,
472 );
473 return Err(err);
474 }
475 self.invalidate_result_cache();
476 if let Err(err) = self.emit_config_mutation_event(
477 crate::runtime::control_events::EventKind::ConfigWrite,
478 crate::runtime::control_events::Outcome::Allowed,
479 "config:write",
480 collection,
481 key,
482 None,
483 &evidence,
484 ) {
485 let _ = self.inner.db.store().delete(collection, id);
486 self.invalidate_result_cache();
487 return Err(err);
488 }
489 if is_enforcement_mode_config(collection, key) {
493 if let Some(auth_store) = self.inner.auth_store.read().clone() {
494 if let Value::Text(text) =
495 &evidence.payload.as_ref().cloned().unwrap_or(Value::Null)
496 {
497 if let Some(mode) =
498 crate::auth::enforcement_mode::PolicyEnforcementMode::parse(text)
499 {
500 auth_store.set_enforcement_mode(mode);
501 }
502 }
503 }
504 }
505 Ok(config_write_output(
506 raw_query,
507 collection,
508 key,
509 version,
510 id,
511 value_type,
512 schema_version,
513 tags,
514 match op {
515 "rotate" => "config_rotate",
516 _ => "config_put",
517 },
518 1,
519 ))
520 }
521
522 fn config_delete_result(
523 &self,
524 raw_query: &str,
525 collection: &str,
526 key: &str,
527 ) -> RedDBResult<RuntimeQueryResult> {
528 let mut evidence = match self.authorize_config_write_for_event(collection, key) {
529 ConfigMutationAuthz::Allowed(evidence) => evidence,
530 ConfigMutationAuthz::Denied { reason, evidence } => {
531 let _ = self.emit_config_mutation_event(
532 crate::runtime::control_events::EventKind::ConfigDelete,
533 crate::runtime::control_events::Outcome::Denied,
534 "config:delete",
535 collection,
536 key,
537 Some(reason.clone()),
538 &evidence,
539 );
540 return Err(RedDBError::Query(reason));
541 }
542 };
543 if let Err(err) = self.check_write(crate::runtime::write_gate::WriteKind::Dml) {
544 let _ = self.emit_config_mutation_event(
545 crate::runtime::control_events::EventKind::ConfigDelete,
546 crate::runtime::control_events::Outcome::Error,
547 "config:delete",
548 collection,
549 key,
550 Some(err.to_string()),
551 &evidence,
552 );
553 return Err(err);
554 }
555 if let Err(err) = self.ensure_config_collection(collection) {
556 let _ = self.emit_config_mutation_event(
557 crate::runtime::control_events::EventKind::ConfigDelete,
558 crate::runtime::control_events::Outcome::Error,
559 "config:delete",
560 collection,
561 key,
562 Some(err.to_string()),
563 &evidence,
564 );
565 return Err(err);
566 }
567 let latest = match self.latest_config_version(collection, key) {
568 Ok(latest) => latest,
569 Err(err) => {
570 let _ = self.emit_config_mutation_event(
571 crate::runtime::control_events::EventKind::ConfigDelete,
572 crate::runtime::control_events::Outcome::Error,
573 "config:delete",
574 collection,
575 key,
576 Some(err.to_string()),
577 &evidence,
578 );
579 return Err(err);
580 }
581 };
582 evidence.payload = latest.as_ref().map(|version| version.value.clone());
583 let version = latest.as_ref().map(|version| version.version).unwrap_or(0) + 1;
584 let value_type = latest.as_ref().and_then(|version| version.value_type);
585 let schema_version = latest.as_ref().and_then(|version| version.schema_version);
586 let id = match self.append_config_version(
587 collection,
588 key,
589 Value::Null,
590 version,
591 true,
592 "delete",
593 value_type,
594 schema_version,
595 &[],
596 ) {
597 Ok(id) => id,
598 Err(err) => {
599 let _ = self.emit_config_mutation_event(
600 crate::runtime::control_events::EventKind::ConfigDelete,
601 crate::runtime::control_events::Outcome::Error,
602 "config:delete",
603 collection,
604 key,
605 Some(err.to_string()),
606 &evidence,
607 );
608 return Err(err);
609 }
610 };
611 if let Some(before) = latest.as_ref().and_then(|version| {
612 if version.tombstone {
613 None
614 } else {
615 Some(reddb_types::Value::to_json(&version.value))
616 }
617 }) {
618 self.record_kv_watch_event(
619 crate::replication::cdc::ChangeOperation::Delete,
620 collection,
621 key,
622 id.raw(),
623 Some(before),
624 None,
625 );
626 }
627 if let Err(err) = self.prune_config_history(collection, key) {
628 let _ = self.emit_config_mutation_event(
629 crate::runtime::control_events::EventKind::ConfigDelete,
630 crate::runtime::control_events::Outcome::Error,
631 "config:delete",
632 collection,
633 key,
634 Some(err.to_string()),
635 &evidence,
636 );
637 return Err(err);
638 }
639 self.invalidate_result_cache();
640 if let Err(err) = self.emit_config_mutation_event(
641 crate::runtime::control_events::EventKind::ConfigDelete,
642 crate::runtime::control_events::Outcome::Allowed,
643 "config:delete",
644 collection,
645 key,
646 None,
647 &evidence,
648 ) {
649 let _ = self.inner.db.store().delete(collection, id);
650 self.invalidate_result_cache();
651 return Err(err);
652 }
653 Ok(config_write_output(
654 raw_query,
655 collection,
656 key,
657 version,
658 id,
659 value_type,
660 schema_version,
661 &[],
662 "delete",
663 1,
664 ))
665 }
666
667 fn config_get_result(
668 &self,
669 raw_query: &str,
670 collection: &str,
671 key: &str,
672 ) -> RedDBResult<RuntimeQueryResult> {
673 self.check_system_config_capability("config:read", collection, key)
674 .map_err(RedDBError::Query)?;
675 let latest = self.latest_config_version(collection, key)?;
676 let mut result = UnifiedResult::with_columns(vec![
677 "collection".into(),
678 "key".into(),
679 "value".into(),
680 "version".into(),
681 "value_type".into(),
682 "schema_version".into(),
683 "tags".into(),
684 "tombstone".into(),
685 ]);
686 let mut record = UnifiedRecord::new();
687 record.set("collection", Value::text(collection.to_string()));
688 record.set("key", Value::text(key.to_string()));
689 if let Some(version) = latest {
690 record.set("value", version.value);
691 record.set("version", Value::Integer(version.version));
692 record.set("value_type", config_value_type_value(version.value_type));
693 record.set(
694 "schema_version",
695 version
696 .schema_version
697 .map(Value::Integer)
698 .unwrap_or(Value::Null),
699 );
700 record.set("tags", config_tags_value(&version.tags));
701 record.set("tombstone", Value::Boolean(version.tombstone));
702 } else {
703 record.set("value", Value::Null);
704 record.set("version", Value::Null);
705 record.set("value_type", Value::Null);
706 record.set("schema_version", Value::Null);
707 record.set("tags", Value::Null);
708 record.set("tombstone", Value::Boolean(false));
709 }
710 result.push(record);
711 Ok(RuntimeQueryResult {
712 query: raw_query.to_string(),
713 mode: reddb_rql::modes::QueryMode::Sql,
714 statement: "config_get",
715 engine: "config",
716 result,
717 affected_rows: 0,
718 statement_type: "select",
719 bookmark: None,
720 notice: None,
721 })
722 }
723
724 fn config_history_result(
725 &self,
726 raw_query: &str,
727 collection: &str,
728 key: &str,
729 ) -> RedDBResult<RuntimeQueryResult> {
730 self.check_system_config_capability("config:read", collection, key)
731 .map_err(RedDBError::Query)?;
732 let versions = super::keyed_spine::history_versions(self.config_versions(collection, key)?);
733 let mut result = UnifiedResult::with_columns(vec![
734 "collection".into(),
735 "key".into(),
736 "version".into(),
737 "value".into(),
738 "value_type".into(),
739 "schema_version".into(),
740 "tags".into(),
741 "tombstone".into(),
742 "op".into(),
743 "created_at_ms".into(),
744 ]);
745 for version in versions {
746 let mut record = UnifiedRecord::new();
747 record.set("collection", Value::text(collection.to_string()));
748 record.set("key", Value::text(key.to_string()));
749 record.set("version", Value::Integer(version.version));
750 record.set("value", version.value);
751 record.set("value_type", config_value_type_value(version.value_type));
752 record.set(
753 "schema_version",
754 version
755 .schema_version
756 .map(Value::Integer)
757 .unwrap_or(Value::Null),
758 );
759 record.set("tags", Value::Null);
760 record.set("tombstone", Value::Boolean(version.tombstone));
761 record.set("op", Value::text(version.op));
762 record.set("created_at_ms", Value::Integer(version.created_at_ms));
763 result.push(record);
764 }
765 Ok(RuntimeQueryResult {
766 query: raw_query.to_string(),
767 mode: reddb_rql::modes::QueryMode::Sql,
768 statement: "config_history",
769 engine: "config",
770 result,
771 affected_rows: 0,
772 statement_type: "select",
773 bookmark: None,
774 notice: None,
775 })
776 }
777
778 fn config_list_result(
779 &self,
780 raw_query: &str,
781 collection: &str,
782 prefix: Option<&str>,
783 limit: Option<usize>,
784 offset: usize,
785 ) -> RedDBResult<RuntimeQueryResult> {
786 let mut versions = self.latest_config_versions(collection, prefix)?;
787 versions.sort_by(|left, right| left.key.cmp(&right.key));
788 let mut result = UnifiedResult::with_columns(vec![
789 "collection".into(),
790 "key".into(),
791 "value".into(),
792 "version".into(),
793 "value_type".into(),
794 "schema_version".into(),
795 "tags".into(),
796 "tombstone".into(),
797 "op".into(),
798 "created_at_ms".into(),
799 ]);
800 for version in versions
801 .into_iter()
802 .filter(|version| {
803 self.check_config_capability("config:read", collection, &version.key)
804 .is_ok()
805 })
806 .skip(offset)
807 .take(limit.unwrap_or(usize::MAX))
808 {
809 let mut record = UnifiedRecord::new();
810 record.set("collection", Value::text(collection.to_string()));
811 record.set("key", Value::text(version.key));
812 record.set("value", version.value);
813 record.set("version", Value::Integer(version.version));
814 record.set("value_type", config_value_type_value(version.value_type));
815 record.set(
816 "schema_version",
817 version
818 .schema_version
819 .map(Value::Integer)
820 .unwrap_or(Value::Null),
821 );
822 record.set("tags", config_tags_value(&version.tags));
823 record.set("tombstone", Value::Boolean(version.tombstone));
824 record.set("op", Value::text(version.op));
825 record.set("created_at_ms", Value::Integer(version.created_at_ms));
826 result.push(record);
827 }
828 Ok(RuntimeQueryResult {
829 query: raw_query.to_string(),
830 mode: reddb_rql::modes::QueryMode::Sql,
831 statement: "config_list",
832 engine: "config",
833 result,
834 affected_rows: 0,
835 statement_type: "select",
836 bookmark: None,
837 notice: None,
838 })
839 }
840
841 fn config_watch_result(
842 &self,
843 raw_query: &str,
844 collection: &str,
845 key: &str,
846 prefix: bool,
847 from_lsn: Option<u64>,
848 ) -> RedDBResult<RuntimeQueryResult> {
849 let watch_key = if prefix {
850 format!("{key}.*")
851 } else {
852 key.to_string()
853 };
854 let endpoint = match from_lsn {
855 Some(lsn) => {
856 format!("/collections/{collection}/config/{watch_key}/watch?since_lsn={lsn}")
857 }
858 None => format!("/collections/{collection}/config/{watch_key}/watch"),
859 };
860 let mut result = UnifiedResult::with_columns(vec![
861 "collection".into(),
862 "key".into(),
863 "prefix".into(),
864 "from_lsn".into(),
865 "watch_url".into(),
866 "streaming".into(),
867 ]);
868 let mut record = UnifiedRecord::new();
869 record.set("collection", Value::text(collection.to_string()));
870 record.set("key", Value::text(watch_key));
871 record.set("prefix", Value::Boolean(prefix));
872 record.set(
873 "from_lsn",
874 from_lsn
875 .map(Value::UnsignedInteger)
876 .unwrap_or(reddb_types::Value::Null),
877 );
878 record.set("watch_url", Value::text(endpoint));
879 record.set("streaming", Value::Boolean(true));
880 result.push(record);
881 Ok(RuntimeQueryResult {
882 query: raw_query.to_string(),
883 mode: reddb_rql::modes::QueryMode::Sql,
884 statement: "config_watch",
885 engine: "config",
886 result,
887 affected_rows: 0,
888 statement_type: "stream",
889 bookmark: None,
890 notice: None,
891 })
892 }
893
894 fn ensure_config_collection(&self, collection: &str) -> RedDBResult<()> {
895 let store = self.inner.db.store();
896 if store.get_collection(collection).is_none() {
897 store
898 .create_collection(collection)
899 .map_err(|err| RedDBError::Internal(err.to_string()))?;
900 }
901 if let Some(contract) = self.inner.db.collection_contract(collection) {
902 crate::runtime::ddl::polymorphic_resolver::ensure_model_match(
903 CollectionModel::Config,
904 contract.declared_model,
905 )?;
906 return Ok(());
907 }
908 let now = current_unix_ms();
909 self.inner
910 .db
911 .save_collection_contract(CollectionContract {
912 name: collection.to_string(),
913 declared_model: CollectionModel::Config,
914 schema_mode: SchemaMode::Dynamic,
915 origin: ContractOrigin::Explicit,
916 version: 1,
917 created_at_unix_ms: now as u128,
918 updated_at_unix_ms: now as u128,
919 default_ttl_ms: None,
920 vector_dimension: None,
921 vector_metric: None,
922 context_index_fields: Vec::new(),
923 declared_columns: Vec::new(),
924 table_def: None,
925 timestamps_enabled: false,
926 context_index_enabled: false,
927 metrics_raw_retention_ms: None,
928 metrics_rollup_policies: Vec::new(),
929 metrics_tenant_identity: None,
930 metrics_namespace: None,
931 append_only: false,
932 subscriptions: Vec::new(),
933 analytics_config: Vec::new(),
934 session_key: None,
935 session_gap_ms: None,
936 retention_duration_ms: None,
937 analytical_storage: None,
938
939 ai_policy: None,
940 })
941 .map(|_| ())
942 .map_err(|err| RedDBError::Internal(err.to_string()))
943 }
944
945 fn append_config_version(
946 &self,
947 collection: &str,
948 key: &str,
949 value: Value,
950 version: i64,
951 tombstone: bool,
952 op: &str,
953 value_type: Option<ConfigValueType>,
954 schema_version: Option<i64>,
955 tags: &[String],
956 ) -> RedDBResult<EntityId> {
957 let now = current_unix_ms() as i64;
958 let fields = vec![
959 ("key".to_string(), Value::text(key.to_string())),
960 ("value".to_string(), value),
961 ("version".to_string(), Value::Integer(version)),
962 (
963 "value_type".to_string(),
964 config_value_type_value(value_type),
965 ),
966 (
967 "schema_version".to_string(),
968 schema_version.map(Value::Integer).unwrap_or(Value::Null),
969 ),
970 ("tombstone".to_string(), Value::Boolean(tombstone)),
971 ("op".to_string(), Value::text(op.to_string())),
972 ("created_at_ms".to_string(), Value::Integer(now)),
973 ("tags".to_string(), config_tags_value(tags)),
974 ];
975 let mut row = RowData::new(Vec::new());
976 row.named = Some(fields.into_iter().collect());
977 let entity = UnifiedEntity::new(
978 EntityId::new(0),
979 EntityKind::TableRow {
980 table: Arc::from(collection),
981 row_id: 0,
982 },
983 EntityData::Row(row),
984 );
985 self.inner
986 .db
987 .store()
988 .insert(collection, entity)
989 .map_err(|err| RedDBError::Internal(err.to_string()))
990 }
991
992 fn latest_config_version(
993 &self,
994 collection: &str,
995 key: &str,
996 ) -> RedDBResult<Option<ConfigVersion>> {
997 Ok(super::keyed_spine::latest_version(
998 self.config_versions(collection, key)?,
999 ))
1000 }
1001
1002 fn config_versions(&self, collection: &str, key: &str) -> RedDBResult<Vec<ConfigVersion>> {
1003 let store = self.inner.db.store();
1004 let Some(manager) = store.get_collection(collection) else {
1005 return Ok(Vec::new());
1006 };
1007 let mut versions = Vec::new();
1008 for entity in manager.query_all(|_| true) {
1009 let EntityData::Row(row) = &entity.data else {
1010 continue;
1011 };
1012 let Some(version) = super::keyed_spine::row_version(entity.id, row, 0) else {
1013 continue;
1014 };
1015 if version.key != key {
1016 continue;
1017 }
1018 versions.push(ConfigVersion::from_keyed_row(version, row));
1019 }
1020 Ok(versions)
1021 }
1022
1023 fn latest_config_versions(
1024 &self,
1025 collection: &str,
1026 prefix: Option<&str>,
1027 ) -> RedDBResult<Vec<ConfigVersion>> {
1028 let store = self.inner.db.store();
1029 let Some(manager) = store.get_collection(collection) else {
1030 return Ok(Vec::new());
1031 };
1032 let mut versions = Vec::new();
1033 for entity in manager.query_all(|_| true) {
1034 let EntityData::Row(row) = &entity.data else {
1035 continue;
1036 };
1037 let Some(version) = super::keyed_spine::row_version(entity.id, row, 0) else {
1038 continue;
1039 };
1040 versions.push(ConfigVersion::from_keyed_row(version, row));
1041 }
1042 Ok(super::keyed_spine::latest_versions(versions, prefix))
1043 }
1044
1045 fn prune_config_history(&self, collection: &str, key: &str) -> RedDBResult<()> {
1046 let mut versions = self.config_versions(collection, key)?;
1047 if versions.len() <= CONFIG_HISTORY_LIMIT {
1048 return Ok(());
1049 }
1050 versions = super::keyed_spine::history_versions(versions);
1051 let drop_count = versions.len() - CONFIG_HISTORY_LIMIT;
1052 let store = self.inner.db.store();
1053 for version in versions.into_iter().take(drop_count) {
1054 store
1055 .delete(collection, version.id)
1056 .map_err(|err| RedDBError::Internal(err.to_string()))?;
1057 }
1058 Ok(())
1059 }
1060
1061 fn authorize_config_write_for_event(&self, collection: &str, key: &str) -> ConfigMutationAuthz {
1062 let default_evidence = self.default_config_mutation_evidence(collection, key);
1063 let Some(auth_store) = self.inner.auth_store.read().clone() else {
1064 return ConfigMutationAuthz::Allowed(default_evidence);
1065 };
1066 if !auth_store.iam_authorization_enabled() {
1067 return ConfigMutationAuthz::Allowed(default_evidence);
1068 }
1069 let Some((principal, role)) = current_auth_identity() else {
1070 return ConfigMutationAuthz::Denied {
1071 reason:
1072 "IAM authorization is enabled; config capability check requires an authenticated principal"
1073 .to_string(),
1074 evidence: default_evidence,
1075 };
1076 };
1077 let tenant = current_tenant();
1078 let principal_id = crate::auth::UserId::from_parts(tenant.as_deref(), &principal);
1079 let ctx = crate::auth::policies::EvalContext {
1080 principal_tenant: tenant.clone(),
1081 current_tenant: tenant.clone(),
1082 peer_ip: None,
1083 mfa_present: false,
1084 now_ms: crate::utils::now_unix_millis() as u128,
1085 principal_is_admin_role: role == crate::auth::Role::Admin,
1086 principal_is_platform_scoped: principal_id.tenant.is_none(),
1087 };
1088 let managed_key = if collection == "red.config" {
1089 format!("red.config.{key}")
1090 } else {
1091 key.to_string()
1092 };
1093 let gate = crate::auth::managed_config::ManagedConfigGate::new(
1094 self.inner.config_registry.as_ref(),
1095 );
1096 match gate.check_write(&auth_store, &principal_id, &ctx, &managed_key) {
1097 crate::auth::managed_config::ManagedConfigDecision::Allow {
1098 entry_id,
1099 resource_type,
1100 managed,
1101 mutability,
1102 matched_action,
1103 matched_resource,
1104 ..
1105 } => {
1106 return ConfigMutationAuthz::Allowed(ConfigMutationEvidence {
1107 id: entry_id,
1108 resource_type,
1109 managed,
1110 mutability,
1111 matched_action: Some(matched_action),
1112 matched_resource: Some(matched_resource),
1113 payload: None,
1114 });
1115 }
1116 crate::auth::managed_config::ManagedConfigDecision::Deny {
1117 entry_id,
1118 resource_type,
1119 managed,
1120 mutability,
1121 matched_action,
1122 matched_resource,
1123 reason,
1124 ..
1125 } => {
1126 return ConfigMutationAuthz::Denied {
1127 reason: format!(
1128 "permission denied: managed config mutation blocked for `{managed_key}`: {reason}"
1129 ),
1130 evidence: ConfigMutationEvidence {
1131 id: entry_id,
1132 resource_type,
1133 managed,
1134 mutability,
1135 matched_action: Some(matched_action),
1136 matched_resource: Some(matched_resource),
1137 payload: None,
1138 },
1139 };
1140 }
1141 crate::auth::managed_config::ManagedConfigDecision::PassThrough { .. } => {}
1142 }
1143
1144 let mut resource = crate::auth::policies::ResourceRef::new(
1145 "config",
1146 config_target_resource(collection, key),
1147 );
1148 if let Some(ref tenant) = tenant {
1149 resource = resource.with_tenant(tenant.clone());
1150 }
1151 if auth_store.check_policy_authz_with_role(
1152 &principal_id,
1153 "config:write",
1154 &resource,
1155 &ctx,
1156 role,
1157 ) {
1158 ConfigMutationAuthz::Allowed(default_evidence)
1159 } else {
1160 ConfigMutationAuthz::Denied {
1161 reason: format!(
1162 "principal=`{}` action=`config:write` resource=`config:{}` denied by IAM policy",
1163 principal,
1164 config_target_resource(collection, key)
1165 ),
1166 evidence: default_evidence,
1167 }
1168 }
1169 }
1170
1171 fn default_config_mutation_evidence(
1172 &self,
1173 collection: &str,
1174 key: &str,
1175 ) -> ConfigMutationEvidence {
1176 let id = if collection == "red.config" {
1177 format!("red.config.{key}")
1178 } else {
1179 key.to_string()
1180 };
1181 ConfigMutationEvidence {
1182 id,
1183 resource_type: crate::auth::managed_config::RESOURCE_TYPE_CONFIG_KEY.to_string(),
1184 managed: false,
1185 mutability: crate::auth::registry::Mutability::MutableViaGovernance,
1186 matched_action: None,
1187 matched_resource: None,
1188 payload: None,
1189 }
1190 }
1191
1192 fn emit_config_mutation_event(
1193 &self,
1194 kind: crate::runtime::control_events::EventKind,
1195 outcome: crate::runtime::control_events::Outcome,
1196 action: &'static str,
1197 collection: &str,
1198 key: &str,
1199 reason: Option<String>,
1200 evidence: &ConfigMutationEvidence,
1201 ) -> RedDBResult<()> {
1202 use crate::runtime::control_events::{
1203 ActorRef, ControlEvent, ControlEventCtx, Sensitivity,
1204 };
1205
1206 let tenant = current_tenant();
1207 let principal = current_auth_identity();
1208 let actor_user = principal
1209 .as_ref()
1210 .map(|(principal, _)| crate::auth::UserId::from_parts(tenant.as_deref(), principal));
1211 let actor = actor_user
1212 .as_ref()
1213 .map(ActorRef::User)
1214 .unwrap_or(ActorRef::Anonymous);
1215 let ctx = ControlEventCtx {
1216 actor,
1217 scope: tenant
1218 .as_ref()
1219 .map(|scope| std::borrow::Cow::Borrowed(scope.as_str())),
1220 request_id: Some(std::borrow::Cow::Owned(format!(
1221 "conn-{}",
1222 current_connection_id()
1223 ))),
1224 trace_id: None,
1225 };
1226
1227 let mut fields = HashMap::new();
1228 fields.insert("id".to_string(), Sensitivity::raw(evidence.id.clone()));
1229 fields.insert(
1230 "resource_type".to_string(),
1231 Sensitivity::raw(evidence.resource_type.clone()),
1232 );
1233 fields.insert(
1234 "managed".to_string(),
1235 Sensitivity::raw(evidence.managed.to_string()),
1236 );
1237 fields.insert(
1238 "mutability".to_string(),
1239 Sensitivity::raw(config_mutability_label(evidence.mutability)),
1240 );
1241 fields.insert("collection".to_string(), Sensitivity::raw(collection));
1242 fields.insert("key".to_string(), Sensitivity::raw(key));
1243 fields.insert(
1244 "connection_id".to_string(),
1245 Sensitivity::raw(current_connection_id().to_string()),
1246 );
1247 if let Some((_, role)) = principal {
1248 fields.insert("actor_role".to_string(), Sensitivity::raw(role.as_str()));
1249 }
1250 if let Some(matched_action) = &evidence.matched_action {
1251 fields.insert(
1252 "matched_action".to_string(),
1253 Sensitivity::raw(matched_action.clone()),
1254 );
1255 }
1256 if let Some(matched_resource) = &evidence.matched_resource {
1257 fields.insert(
1258 "matched_resource".to_string(),
1259 Sensitivity::raw(matched_resource.clone()),
1260 );
1261 }
1262 if let Some(payload) = &evidence.payload {
1263 fields.insert(
1264 "payload".to_string(),
1265 config_payload_sensitivity(&evidence.resource_type, "payload", payload),
1266 );
1267 }
1268
1269 let event = ControlEvent {
1270 kind,
1271 outcome,
1272 action: std::borrow::Cow::Borrowed(action),
1273 resource: Some(format!(
1274 "config:{}",
1275 config_target_resource(collection, key)
1276 )),
1277 reason,
1278 matched_policy_id: None,
1279 fields,
1280 };
1281 let ledger = self.inner.control_event_ledger.read();
1282 match ledger.emit(&ctx, event) {
1283 Ok(_) => Ok(()),
1284 Err(err) if self.inner.control_event_config.require_persistence() => {
1285 Err(RedDBError::Internal(err.to_string()))
1286 }
1287 Err(_) => Ok(()),
1288 }
1289 }
1290
1291 fn check_config_capability(
1292 &self,
1293 action: &str,
1294 collection: &str,
1295 key: &str,
1296 ) -> Result<(), String> {
1297 let Some(auth_store) = self.inner.auth_store.read().clone() else {
1298 return Ok(());
1299 };
1300 if !auth_store.iam_authorization_enabled() {
1301 return Ok(());
1302 }
1303 let Some((principal, role)) = current_auth_identity() else {
1304 return Err(
1305 "IAM authorization is enabled; config capability check requires an authenticated principal"
1306 .to_string(),
1307 );
1308 };
1309 let tenant = current_tenant();
1310 let principal_id = crate::auth::UserId::from_parts(tenant.as_deref(), &principal);
1311 let mut resource = crate::auth::policies::ResourceRef::new(
1312 "config",
1313 config_target_resource(collection, key),
1314 );
1315 if let Some(ref tenant) = tenant {
1316 resource = resource.with_tenant(tenant.clone());
1317 }
1318 let ctx = crate::auth::policies::EvalContext {
1319 principal_tenant: tenant.clone(),
1320 current_tenant: tenant,
1321 peer_ip: None,
1322 mfa_present: false,
1323 now_ms: crate::utils::now_unix_millis() as u128,
1324 principal_is_admin_role: role == crate::auth::Role::Admin,
1325 principal_is_platform_scoped: principal_id.tenant.is_none(),
1326 };
1327 if action == "config:write" {
1328 let managed_key = if collection == "red.config" {
1329 format!("red.config.{key}")
1330 } else {
1331 key.to_string()
1332 };
1333 let gate = crate::auth::managed_config::ManagedConfigGate::new(
1334 self.inner.config_registry.as_ref(),
1335 );
1336 match gate.check_write(&auth_store, &principal_id, &ctx, &managed_key) {
1337 crate::auth::managed_config::ManagedConfigDecision::PassThrough { .. } => {}
1338 crate::auth::managed_config::ManagedConfigDecision::Allow { .. } => return Ok(()),
1339 crate::auth::managed_config::ManagedConfigDecision::Deny { reason, .. } => {
1340 return Err(format!(
1341 "permission denied: managed config mutation blocked for `{managed_key}`: {reason}"
1342 ));
1343 }
1344 }
1345 }
1346 if auth_store.check_policy_authz_with_role(&principal_id, action, &resource, &ctx, role) {
1347 Ok(())
1348 } else {
1349 Err(format!(
1350 "principal=`{}` action=`{}` resource=`config:{}` denied by IAM policy",
1351 principal,
1352 action,
1353 config_target_resource(collection, key)
1354 ))
1355 }
1356 }
1357
1358 fn check_system_config_capability(
1359 &self,
1360 action: &str,
1361 collection: &str,
1362 key: &str,
1363 ) -> Result<(), String> {
1364 if collection != "red.config" {
1365 return Ok(());
1366 }
1367 self.check_config_capability(action, collection, key)
1368 }
1369
1370 pub fn config_watch_events_since(
1371 &self,
1372 collection: &str,
1373 key: &str,
1374 since_lsn: u64,
1375 max_count: usize,
1376 ) -> Vec<crate::replication::cdc::KvWatchEvent> {
1377 self.kv_watch_events_since(collection, key, since_lsn, max_count)
1378 .into_iter()
1379 .map(|event| self.policy_filter_config_watch_event(event))
1380 .collect()
1381 }
1382
1383 pub fn config_watch_events_since_prefix(
1384 &self,
1385 collection: &str,
1386 prefix: &str,
1387 since_lsn: u64,
1388 max_count: usize,
1389 ) -> Vec<crate::replication::cdc::KvWatchEvent> {
1390 self.kv_watch_events_since_prefix(collection, prefix, since_lsn, max_count)
1391 .into_iter()
1392 .map(|event| self.policy_filter_config_watch_event(event))
1393 .collect()
1394 }
1395
1396 fn policy_filter_config_watch_event(
1397 &self,
1398 mut event: crate::replication::cdc::KvWatchEvent,
1399 ) -> crate::replication::cdc::KvWatchEvent {
1400 if self
1401 .check_config_capability("config:read", &event.collection, &event.key)
1402 .is_err()
1403 {
1404 event.before = None;
1405 event.after = None;
1406 }
1407 event
1408 }
1409
1410 fn secret_ref_guard_write_check(
1416 &self,
1417 collection: &str,
1418 key: &str,
1419 value: &Value,
1420 ) -> Option<RedDBError> {
1421 if !value_looks_like_secret_ref(value) {
1422 return None;
1423 }
1424 let Ok(secret_ref) = parse_config_secret_ref(value) else {
1425 return None;
1426 };
1427 let unsealed = match self.peek_vault_unsealed(&secret_ref.collection, &secret_ref.key) {
1428 Ok(Some(value)) => value,
1429 Ok(None) => return None,
1430 Err(_) => return None,
1431 };
1432 if value_looks_like_secret_ref(&unsealed) {
1433 return Some(secret_ref_chain_error(
1434 collection,
1435 key,
1436 &secret_ref.collection,
1437 &secret_ref.key,
1438 ));
1439 }
1440 None
1441 }
1442
1443 fn audit_config_resolve(
1444 &self,
1445 collection: &str,
1446 key: &str,
1447 secret_ref: Option<&ConfigSecretRef>,
1448 outcome: crate::runtime::audit_log::Outcome,
1449 reason: &str,
1450 ) {
1451 let actor = current_auth_identity()
1452 .map(|(principal, _)| principal)
1453 .unwrap_or_else(|| "anonymous".to_string());
1454 let request_id = match current_connection_id() {
1455 0 => "embedded".to_string(),
1456 id => format!("conn-{id}"),
1457 };
1458 let mut builder = crate::runtime::audit_log::AuditEvent::builder("config/resolve")
1459 .principal(actor.clone())
1460 .source(crate::runtime::audit_log::AuditAuthSource::Password)
1461 .resource(format!(
1462 "config:{}",
1463 config_target_resource(collection, key)
1464 ))
1465 .outcome(outcome)
1466 .correlation_id(request_id.clone())
1467 .fields([
1468 crate::runtime::audit_log::AuditFieldEscaper::field("actor", actor),
1469 crate::runtime::audit_log::AuditFieldEscaper::field("collection", collection),
1470 crate::runtime::audit_log::AuditFieldEscaper::field("key", key),
1471 crate::runtime::audit_log::AuditFieldEscaper::field(
1472 "target",
1473 config_target_resource(collection, key),
1474 ),
1475 crate::runtime::audit_log::AuditFieldEscaper::field("reason", reason),
1476 crate::runtime::audit_log::AuditFieldEscaper::field("request_id", request_id),
1477 crate::runtime::audit_log::AuditFieldEscaper::field(
1478 "connection_id",
1479 current_connection_id(),
1480 ),
1481 ]);
1482 if let Some(tenant) = current_tenant() {
1483 builder = builder.tenant(tenant);
1484 }
1485 if let Some(secret_ref) = secret_ref {
1486 builder = builder.fields([
1487 crate::runtime::audit_log::AuditFieldEscaper::field("resolved_store", "vault"),
1488 crate::runtime::audit_log::AuditFieldEscaper::field(
1489 "resolved_collection",
1490 secret_ref.collection.as_str(),
1491 ),
1492 crate::runtime::audit_log::AuditFieldEscaper::field(
1493 "resolved_key",
1494 secret_ref.key.as_str(),
1495 ),
1496 crate::runtime::audit_log::AuditFieldEscaper::field(
1497 "resolved_target",
1498 format!("{}.{}", secret_ref.collection, secret_ref.key),
1499 ),
1500 ]);
1501 }
1502 self.audit_log().record_event(builder.build());
1503 }
1504}
1505
1506fn value_looks_like_secret_ref(value: &Value) -> bool {
1511 let Value::Json(bytes) = value else {
1512 return false;
1513 };
1514 let Ok(json) = crate::json::from_slice::<crate::json::Value>(bytes) else {
1515 return false;
1516 };
1517 let Some(object) = json.as_object() else {
1518 return false;
1519 };
1520 object.get("type").and_then(|value| value.as_str()) == Some("secret_ref")
1521}
1522
1523fn secret_ref_chain_error(
1524 source_collection: &str,
1525 source_key: &str,
1526 target_collection: &str,
1527 target_key: &str,
1528) -> RedDBError {
1529 RedDBError::InvalidConfig(format!(
1530 "secret_ref chain rejected: config `{source_collection}.{source_key}` points at vault `{target_collection}.{target_key}` which is itself a secret_ref; depth-1 invariant requires the target to resolve to a non-secret_ref value"
1531 ))
1532}
1533
1534fn parse_config_secret_ref(value: &Value) -> RedDBResult<ConfigSecretRef> {
1535 let Value::Json(bytes) = value else {
1536 return Err(RedDBError::InvalidConfig(
1537 "CONFIG value is not a SecretRef".to_string(),
1538 ));
1539 };
1540 let json = crate::json::from_slice::<crate::json::Value>(bytes).map_err(|err| {
1541 RedDBError::InvalidConfig(format!("CONFIG SecretRef is malformed: {err}"))
1542 })?;
1543 let Some(object) = json.as_object() else {
1544 return Err(RedDBError::InvalidConfig(
1545 "CONFIG SecretRef must be an object".to_string(),
1546 ));
1547 };
1548 let get_str = |field: &str| -> RedDBResult<&str> {
1549 object
1550 .get(field)
1551 .and_then(|value| value.as_str())
1552 .ok_or_else(|| RedDBError::InvalidConfig(format!("CONFIG SecretRef missing {field}")))
1553 };
1554 if get_str("type")? != "secret_ref" {
1555 return Err(RedDBError::InvalidConfig(
1556 "CONFIG value is not a SecretRef".to_string(),
1557 ));
1558 }
1559 if get_str("store")? != "vault" {
1560 return Err(RedDBError::InvalidConfig(
1561 "CONFIG SecretRef store is unsupported".to_string(),
1562 ));
1563 }
1564 Ok(ConfigSecretRef {
1565 collection: get_str("collection")?.to_string(),
1566 key: get_str("key")?.to_string(),
1567 })
1568}
1569
1570fn config_target_resource(collection: &str, key: &str) -> String {
1571 if collection == "red.config" {
1572 format!("red.config/{}", key.to_ascii_lowercase())
1573 } else {
1574 format!("{collection}.{key}")
1575 }
1576}
1577
1578fn config_write_output(
1579 raw_query: &str,
1580 collection: &str,
1581 key: &str,
1582 version: i64,
1583 id: EntityId,
1584 value_type: Option<ConfigValueType>,
1585 schema_version: Option<i64>,
1586 tags: &[String],
1587 statement: &'static str,
1588 affected_rows: u64,
1589) -> RuntimeQueryResult {
1590 let mut result = UnifiedResult::with_columns(vec![
1591 "ok".into(),
1592 "collection".into(),
1593 "key".into(),
1594 "version".into(),
1595 "value_type".into(),
1596 "schema_version".into(),
1597 "tags".into(),
1598 "id".into(),
1599 ]);
1600 let mut record = UnifiedRecord::new();
1601 record.set("ok", Value::Boolean(true));
1602 record.set("collection", Value::text(collection.to_string()));
1603 record.set("key", Value::text(key.to_string()));
1604 record.set("version", Value::Integer(version));
1605 record.set("value_type", config_value_type_value(value_type));
1606 record.set(
1607 "schema_version",
1608 schema_version.map(Value::Integer).unwrap_or(Value::Null),
1609 );
1610 record.set("tags", config_tags_value(tags));
1611 record.set("id", Value::Integer(id.raw() as i64));
1612 result.push(record);
1613 RuntimeQueryResult {
1614 query: raw_query.to_string(),
1615 mode: reddb_rql::modes::QueryMode::Sql,
1616 statement,
1617 engine: "config",
1618 result,
1619 affected_rows,
1620 statement_type: if statement == "delete" {
1621 "delete"
1622 } else {
1623 "update"
1624 },
1625 bookmark: None,
1626 notice: None,
1627 }
1628}
1629
1630fn invalid_config_volatility(operation: &str) -> RedDBError {
1631 RedDBError::InvalidOperation(format!(
1632 "CONFIG does not support KV-only volatility operation {operation}"
1633 ))
1634}
1635
1636fn resolve_config_schema(
1637 latest: Option<&ConfigVersion>,
1638 requested_type: Option<ConfigValueType>,
1639) -> (Option<ConfigValueType>, Option<i64>) {
1640 let previous_type = latest.and_then(|version| version.value_type);
1641 let previous_schema_version = latest.and_then(|version| version.schema_version);
1642 match requested_type {
1643 Some(value_type) if Some(value_type) != previous_type => (
1644 Some(value_type),
1645 Some(previous_schema_version.unwrap_or(0) + 1),
1646 ),
1647 Some(value_type) => (Some(value_type), previous_schema_version.or(Some(1))),
1648 None => (previous_type, previous_schema_version),
1649 }
1650}
1651
1652fn validate_config_value_type(value: &Value, value_type: ConfigValueType) -> RedDBResult<()> {
1653 let valid = match value_type {
1654 ConfigValueType::Bool => matches!(value, Value::Boolean(_)),
1655 ConfigValueType::Int => matches!(
1656 value,
1657 Value::Integer(_) | Value::UnsignedInteger(_) | Value::BigInt(_)
1658 ),
1659 ConfigValueType::String => matches!(value, Value::Text(_)),
1660 ConfigValueType::Url => validate_config_url(value),
1661 ConfigValueType::Object => validate_config_json_shape(value, true),
1662 ConfigValueType::Array => {
1663 matches!(value, Value::Array(_) | Value::Vector(_))
1664 || validate_config_json_shape(value, false)
1665 }
1666 };
1667 if valid {
1668 Ok(())
1669 } else {
1670 Err(RedDBError::InvalidConfig(format!(
1671 "CONFIG value type mismatch: expected {}, got {}",
1672 value_type.as_str(),
1673 config_actual_value_type(value),
1674 )))
1675 }
1676}
1677
1678fn is_enforcement_mode_config(collection: &str, key: &str) -> bool {
1683 collection == "red.config" && key == "policy.enforcement_mode"
1684}
1685
1686fn validate_enforcement_mode_value(value: &Value) -> RedDBResult<()> {
1690 let text = match value {
1691 Value::Text(text) => text.as_ref(),
1692 _ => {
1693 return Err(RedDBError::InvalidConfig(format!(
1694 "config key `{}` must be a string ({} or {}); got {}",
1695 crate::auth::enforcement_mode::ENFORCEMENT_MODE_CONFIG_KEY,
1696 crate::auth::enforcement_mode::PolicyEnforcementMode::LegacyRbac.as_str(),
1697 crate::auth::enforcement_mode::PolicyEnforcementMode::PolicyOnly.as_str(),
1698 config_actual_value_type(value),
1699 )));
1700 }
1701 };
1702 if crate::auth::enforcement_mode::PolicyEnforcementMode::parse(text).is_some() {
1703 Ok(())
1704 } else {
1705 Err(RedDBError::InvalidConfig(format!(
1706 "config key `{}` accepts only `{}` or `{}`, got `{}`",
1707 crate::auth::enforcement_mode::ENFORCEMENT_MODE_CONFIG_KEY,
1708 crate::auth::enforcement_mode::PolicyEnforcementMode::LegacyRbac.as_str(),
1709 crate::auth::enforcement_mode::PolicyEnforcementMode::PolicyOnly.as_str(),
1710 text,
1711 )))
1712 }
1713}
1714
1715fn validate_config_url(value: &Value) -> bool {
1716 let url = match value {
1717 Value::Url(value) => value.as_str(),
1718 Value::Text(value) => value.as_ref(),
1719 _ => return false,
1720 };
1721 url.starts_with("http://") || url.starts_with("https://") || url.starts_with("ftp://")
1722}
1723
1724fn validate_config_json_shape(value: &Value, object: bool) -> bool {
1725 let Value::Json(bytes) = value else {
1726 return false;
1727 };
1728 let Ok(json) = crate::json::from_slice::<crate::json::Value>(bytes) else {
1729 return false;
1730 };
1731 matches!(
1732 (object, json),
1733 (true, crate::json::Value::Object(_)) | (false, crate::json::Value::Array(_))
1734 )
1735}
1736
1737fn config_actual_value_type(value: &Value) -> &'static str {
1738 match value {
1739 Value::Null => "null",
1740 Value::Boolean(_) => "bool",
1741 Value::Integer(_) | Value::UnsignedInteger(_) | Value::BigInt(_) => "int",
1742 Value::Text(_) => "string",
1743 Value::Url(_) => "url",
1744 Value::Json(bytes) => match crate::json::from_slice::<crate::json::Value>(bytes) {
1745 Ok(crate::json::Value::Object(_)) => "object",
1746 Ok(crate::json::Value::Array(_)) => "array",
1747 _ => "json",
1748 },
1749 Value::Array(_) | Value::Vector(_) => "array",
1750 _ => "other",
1751 }
1752}
1753
1754fn config_value_type_value(value_type: Option<ConfigValueType>) -> Value {
1755 value_type
1756 .map(|value_type| Value::text(value_type.as_str()))
1757 .unwrap_or(Value::Null)
1758}
1759
1760fn config_value_type_from_value(value: &Value) -> Option<ConfigValueType> {
1761 match value {
1762 Value::Text(value) => ConfigValueType::parse(value.as_ref()),
1763 _ => None,
1764 }
1765}
1766
1767fn config_tags_value(tags: &[String]) -> Value {
1768 if tags.is_empty() {
1769 return Value::Null;
1770 }
1771 Value::Array(tags.iter().map(|tag| Value::text(tag.clone())).collect())
1772}
1773
1774fn config_tags_from_value(value: Option<&Value>) -> Vec<String> {
1775 match value {
1776 Some(Value::Array(values)) => values
1777 .iter()
1778 .filter_map(|value| match value {
1779 Value::Text(tag) => Some(tag.to_string()),
1780 _ => None,
1781 })
1782 .collect(),
1783 Some(Value::Json(bytes)) => crate::json::from_slice::<crate::json::Value>(bytes)
1784 .ok()
1785 .and_then(|value| value.as_array().map(|values| values.to_vec()))
1786 .map(|values| {
1787 values
1788 .into_iter()
1789 .filter_map(|value| value.as_str().map(ToOwned::to_owned))
1790 .collect()
1791 })
1792 .unwrap_or_default(),
1793 _ => Vec::new(),
1794 }
1795}
1796
1797fn config_payload_sensitivity(
1798 resource_type: &str,
1799 field: &str,
1800 value: &Value,
1801) -> crate::runtime::control_events::Sensitivity {
1802 let payload = config_payload_bytes(value);
1803 if config_payload_raw_allowed(resource_type, field) {
1804 crate::runtime::control_events::Sensitivity::raw(
1805 String::from_utf8_lossy(&payload).into_owned(),
1806 )
1807 } else {
1808 crate::runtime::control_events::Sensitivity::hashed(&payload)
1809 }
1810}
1811
1812fn config_payload_bytes(value: &Value) -> Vec<u8> {
1813 let json = reddb_types::Value::to_json(value);
1814 crate::serde_json::to_vec(&json).unwrap_or_else(|_| value.to_string().into_bytes())
1815}
1816
1817fn config_payload_raw_allowed(resource_type: &str, field: &str) -> bool {
1818 const RAW_PAYLOAD_FIELDS: &[(&str, &str)] = &[("audit_surface", "payload")];
1819 RAW_PAYLOAD_FIELDS
1820 .iter()
1821 .any(|(allowed_type, allowed_field)| {
1822 *allowed_type == resource_type && *allowed_field == field
1823 })
1824}
1825
1826fn config_mutability_label(mutability: crate::auth::registry::Mutability) -> &'static str {
1827 match mutability {
1828 crate::auth::registry::Mutability::Immutable => "immutable",
1829 crate::auth::registry::Mutability::MutableViaGovernance => "mutable_via_governance",
1830 }
1831}
1832
1833fn current_unix_ms() -> u64 {
1834 std::time::SystemTime::now()
1835 .duration_since(std::time::UNIX_EPOCH)
1836 .unwrap_or_default()
1837 .as_millis() as u64
1838}