data_connector/
schema.rs

1//! Schema configuration for storage backends.
2//!
3//! Provides YAML-driven customization of table names and column names.
4//! When no schema config is provided, all defaults match the current
5//! hardcoded behavior — zero behavioral change.
6
7use std::collections::{HashMap, HashSet};
8
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11
12// ────────────────────────────────────────────────────────────────────────────
13// Types
14// ────────────────────────────────────────────────────────────────────────────
15
16/// Top-level schema configuration. Drives all SQL generation and key naming.
17///
18/// Every field has a default matching current hardcoded behavior, so omitting
19/// the entire `schema:` section in YAML produces identical queries to today.
20#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
21#[serde(default)]
22pub struct SchemaConfig {
23    /// Schema owner / key prefix (e.g. `"ADMIN"` for Oracle, `"myapp"` for Redis).
24    /// The dot in `ADMIN."TABLE"` is generated by `qualified_table()`, not stored here.
25    #[serde(skip_serializing_if = "Option::is_none")]
26    pub owner: Option<String>,
27
28    /// Starting schema version. Set this when your database already has
29    /// migrations applied (e.g. `version: 3` skips migrations 1–3).
30    /// `None` means start from 0 (apply all migrations).
31    #[serde(default, skip_serializing_if = "Option::is_none")]
32    pub version: Option<u32>,
33
34    /// Whether to run schema migrations automatically on startup.
35    /// Defaults to `false` (safe by default). When `false` and pending
36    /// migrations are detected, startup fails with the exact SQL statements
37    /// needed so you can review and apply them manually.
38    /// Set to `true` to opt in to automatic migration.
39    #[serde(default = "default_auto_migrate")]
40    pub auto_migrate: bool,
41
42    pub conversations: TableConfig,
43    pub responses: TableConfig,
44    pub conversation_items: TableConfig,
45    pub conversation_item_links: TableConfig,
46    pub conversation_memories: TableConfig,
47}
48
49/// Per-table schema configuration.
50#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
51#[serde(default)]
52pub struct TableConfig {
53    /// Physical table name (or Redis key component).
54    pub table: String,
55
56    /// Column name overrides: `logical_name -> db_column_name`.
57    /// Fields not listed here use their logical name unchanged.
58    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
59    pub columns: HashMap<String, String>,
60
61    /// Extra columns to include in DDL, INSERT, and SELECT statements.
62    /// Populated by hooks at runtime; schema config declares the column
63    /// definitions so backends know the SQL type and default value.
64    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
65    pub extra_columns: HashMap<String, ColumnDef>,
66
67    /// Logical column names to omit from DDL, INSERT, and SELECT.
68    /// Use when the physical schema lacks a standard column
69    /// (e.g. a team's RESPONSES table has no `safety_identifier`).
70    #[serde(default, skip_serializing_if = "HashSet::is_empty")]
71    pub skip_columns: HashSet<String>,
72}
73
74/// Column definition for extra (user-defined) columns declared in schema config.
75#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
76pub struct ColumnDef {
77    /// SQL type string, e.g. `"VARCHAR2(256)"` or `"TEXT"`.
78    pub sql_type: String,
79
80    /// Default value used on INSERT when no hook provides a value.
81    /// `None` means NULL.
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub default_value: Option<Value>,
84}
85
86// ────────────────────────────────────────────────────────────────────────────
87// Defaults
88// ────────────────────────────────────────────────────────────────────────────
89
90fn default_auto_migrate() -> bool {
91    std::env::var("DB_AUTO_MIGRATE")
92        .ok()
93        .map(|v| v.eq_ignore_ascii_case("true") || v == "1")
94        .unwrap_or(false)
95}
96
97impl Default for SchemaConfig {
98    fn default() -> Self {
99        Self {
100            owner: None,
101            version: None,
102            auto_migrate: default_auto_migrate(),
103            conversations: TableConfig::with_table("conversations"),
104            responses: TableConfig::with_table("responses"),
105            conversation_items: TableConfig::with_table("conversation_items"),
106            conversation_item_links: TableConfig::with_table("conversation_item_links"),
107            conversation_memories: TableConfig::with_table("conversation_memories"),
108        }
109    }
110}
111
112// ────────────────────────────────────────────────────────────────────────────
113// TableConfig methods
114// ────────────────────────────────────────────────────────────────────────────
115
116impl TableConfig {
117    /// Create a `TableConfig` with the given table name and no column overrides.
118    pub fn with_table(name: &str) -> Self {
119        Self {
120            table: name.to_string(),
121            ..Default::default()
122        }
123    }
124
125    /// Resolve a column name.
126    ///
127    /// Returns the remapped name if an override is configured, or the
128    /// logical field name unchanged otherwise.
129    pub fn col<'a>(&'a self, field: &'a str) -> &'a str {
130        self.columns.get(field).map(String::as_str).unwrap_or(field)
131    }
132
133    /// Returns `true` if `field` is in the `skip_columns` set.
134    ///
135    /// Skipped columns are omitted from DDL, INSERT, and SELECT statements.
136    pub fn is_skipped(&self, field: &str) -> bool {
137        self.skip_columns.contains(field)
138    }
139
140    /// Fully qualified table name, e.g. `ADMIN."MY_TABLE"` with an owner
141    /// or just `my_table` without.
142    ///
143    /// The table name is quoted to preserve case. For Oracle, call
144    /// `SchemaConfig::uppercase_for_oracle()` first so the quoted names match
145    /// Oracle's uppercase catalog entries.
146    pub fn qualified_table(&self, owner: Option<&str>) -> String {
147        match owner {
148            Some(o) => format!("{o}.\"{}\"", self.table),
149            None => self.table.clone(),
150        }
151    }
152}
153
154// ────────────────────────────────────────────────────────────────────────────
155// Validation
156// ────────────────────────────────────────────────────────────────────────────
157
158impl SchemaConfig {
159    /// Uppercase all table names and column override values in-place.
160    ///
161    /// Oracle folds unquoted identifiers to uppercase, so existing tables and
162    /// columns are stored as `CONVERSATIONS`, `CONV_ID`, etc.
163    /// `qualified_table()` quotes identifiers (`OWNER."table"`), making them
164    /// case-sensitive.  Calling this in `OracleStore::new()` ensures the
165    /// quoted names match the actual uppercase catalog entries Oracle created.
166    pub fn uppercase_for_oracle(&mut self) {
167        for tc in [
168            &mut self.conversations,
169            &mut self.responses,
170            &mut self.conversation_items,
171            &mut self.conversation_item_links,
172            &mut self.conversation_memories,
173        ] {
174            tc.table.make_ascii_uppercase();
175            for val in tc.columns.values_mut() {
176                val.make_ascii_uppercase();
177            }
178            // Uppercase extra column keys (the physical column names in Oracle).
179            // We must collect keys first to avoid borrowing conflicts.
180            let keys: Vec<String> = tc.extra_columns.keys().cloned().collect();
181            for key in keys {
182                if let Some(def) = tc.extra_columns.remove(&key) {
183                    tc.extra_columns.insert(key.to_ascii_uppercase(), def);
184                }
185            }
186            // NOTE: skip_columns are NOT uppercased. They are logical field
187            // names used in is_skipped() checks throughout the backends, and
188            // callers always pass lowercase literals (e.g. "safety_identifier").
189            // Uppercasing them would break the case-sensitive HashSet lookup.
190        }
191    }
192
193    /// Validate the entire schema config at startup. Rejects invalid identifiers.
194    pub fn validate(&self) -> Result<(), String> {
195        // Validate owner
196        if let Some(ref owner) = self.owner {
197            validate_identifier(owner).map_err(|e| format!("owner: {e}"))?;
198        }
199
200        // Validate each table config
201        Self::validate_table("conversations", &self.conversations)?;
202        Self::validate_table("responses", &self.responses)?;
203        Self::validate_table("conversation_items", &self.conversation_items)?;
204        Self::validate_table("conversation_item_links", &self.conversation_item_links)?;
205        Self::validate_table("conversation_memories", &self.conversation_memories)?;
206
207        Ok(())
208    }
209
210    fn validate_table(label: &str, tc: &TableConfig) -> Result<(), String> {
211        validate_identifier(&tc.table).map_err(|e| {
212            if tc.table.is_empty() {
213                format!("{label}.table: table name is required (got empty string — did you omit the 'table' key in your config?)")
214            } else {
215                format!("{label}.table: {e}")
216            }
217        })?;
218
219        for (logical, physical) in &tc.columns {
220            validate_identifier(logical)
221                .map_err(|e| format!("{label}.columns key '{logical}': {e}"))?;
222            validate_identifier(physical)
223                .map_err(|e| format!("{label}.columns value '{physical}': {e}"))?;
224        }
225
226        for (name, def) in &tc.extra_columns {
227            validate_identifier(name)
228                .map_err(|e| format!("{label}.extra_columns key '{name}': {e}"))?;
229            if def.sql_type.is_empty() {
230                return Err(format!(
231                    "{label}.extra_columns['{name}']: sql_type must not be empty"
232                ));
233            }
234            validate_sql_type(&def.sql_type)
235                .map_err(|e| format!("{label}.extra_columns['{name}'].sql_type: {e}"))?;
236        }
237
238        // Reject extra_columns that shadow core column names (case-insensitive,
239        // since Oracle normalizes to uppercase)
240        let core = core_columns_for(label);
241        for name in tc.extra_columns.keys() {
242            let upper = name.to_ascii_uppercase();
243            if core.iter().any(|c| c.to_ascii_uppercase() == upper) {
244                return Err(format!(
245                    "{label}.extra_columns: '{name}' shadows a core column name"
246                ));
247            }
248        }
249
250        // Detect case-insensitive collisions between extra_columns keys
251        // (e.g. "tenant_id" and "TENANT_ID" would collide after uppercase_for_oracle)
252        let mut folded: HashSet<String> = HashSet::new();
253        for name in tc.extra_columns.keys() {
254            let upper = name.to_ascii_uppercase();
255            if !folded.insert(upper) {
256                return Err(format!(
257                    "{label}.extra_columns: case-insensitive collision on '{name}' \
258                     (Oracle normalizes identifiers to uppercase)"
259                ));
260            }
261        }
262
263        for name in &tc.skip_columns {
264            validate_identifier(name).map_err(|e| format!("{label}.skip_columns '{name}': {e}"))?;
265            if primary_key_columns_for(label).contains(&name.as_str()) {
266                return Err(format!(
267                    "{label}.skip_columns: cannot skip '{name}' — it is part of the primary key"
268                ));
269            }
270            if !core.contains(&name.as_str()) {
271                return Err(format!(
272                    "{label}.skip_columns: '{name}' is not a recognized column \
273                     (known: {core:?})"
274                ));
275            }
276        }
277
278        Ok(())
279    }
280}
281
282fn primary_key_columns_for(label: &str) -> &'static [&'static str] {
283    match label {
284        "conversation_memories" => &["memory_id"],
285        _ => &["id"],
286    }
287}
288
289/// Core logical column names for each table, used by validation to reject
290/// `extra_columns` that would shadow built-in fields.
291fn core_columns_for(label: &str) -> &'static [&'static str] {
292    match label {
293        "conversations" => &["id", "created_at", "metadata"],
294        "responses" => &[
295            "id",
296            "conversation_id",
297            "previous_response_id",
298            "input",
299            "created_at",
300            "safety_identifier",
301            "model",
302            "raw_response",
303        ],
304        "conversation_items" => &[
305            "id",
306            "response_id",
307            "item_type",
308            "role",
309            "content",
310            "status",
311            "created_at",
312        ],
313        "conversation_item_links" => &["conversation_id", "item_id", "added_at"],
314        "conversation_memories" => &[
315            "memory_id",
316            "conversation_id",
317            "conversation_version",
318            "response_id",
319            "memory_type",
320            "status",
321            "attempt",
322            "owner_id",
323            "next_run_at",
324            "lease_until",
325            "content",
326            "memory_config",
327            "scope_id",
328            "error_msg",
329            "created_at",
330            "updated_at",
331        ],
332        _ => &[],
333    }
334}
335
336/// Maximum identifier length. Oracle caps at 30 (pre-12.2) or 128,
337/// Postgres at 63. We use 128 as a generous upper bound.
338const MAX_IDENTIFIER_LEN: usize = 128;
339
340/// Reject identifiers that are empty, too long, or contain characters outside `[a-zA-Z0-9_]`.
341fn validate_identifier(name: &str) -> Result<(), String> {
342    if name.is_empty() {
343        return Err("identifier must not be empty".to_string());
344    }
345    if name.len() > MAX_IDENTIFIER_LEN {
346        return Err(format!(
347            "identifier '{name}' exceeds maximum length of {MAX_IDENTIFIER_LEN} characters"
348        ));
349    }
350    if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
351        return Err(format!(
352            "invalid identifier '{name}' — only ASCII alphanumeric and underscores allowed"
353        ));
354    }
355    Ok(())
356}
357
358/// Maximum length for a `sql_type` string in a `ColumnDef`.
359const MAX_SQL_TYPE_LEN: usize = 64;
360
361/// Validate a `sql_type` string for use in DDL.
362///
363/// Accepts only characters needed for standard SQL type declarations:
364/// alphanumeric, underscore, space, parentheses, comma, period.
365/// Examples: `VARCHAR(128)`, `TIMESTAMP WITH TIME ZONE`, `NUMBER(10,2)`.
366fn validate_sql_type(sql_type: &str) -> Result<(), String> {
367    if sql_type.trim().is_empty() {
368        return Err("sql_type must not be whitespace-only".to_string());
369    }
370    if sql_type.len() > MAX_SQL_TYPE_LEN {
371        return Err(format!(
372            "sql_type '{sql_type}' exceeds maximum length of {MAX_SQL_TYPE_LEN} characters"
373        ));
374    }
375    if !sql_type
376        .chars()
377        .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | ' ' | '(' | ')' | ',' | '.'))
378    {
379        return Err(format!(
380            "invalid sql_type '{sql_type}' — only ASCII alphanumeric, underscore, space, \
381             parentheses, comma, and period are allowed"
382        ));
383    }
384    Ok(())
385}
386
387// ────────────────────────────────────────────────────────────────────────────
388// Tests
389// ────────────────────────────────────────────────────────────────────────────
390
391#[cfg(test)]
392mod tests {
393    use super::*;
394
395    // ── Default config ────────────────────────────────────────────────────
396
397    #[test]
398    fn default_config_matches_hardcoded_names() {
399        let cfg = SchemaConfig::default();
400        assert_eq!(cfg.conversations.table, "conversations");
401        assert_eq!(cfg.responses.table, "responses");
402        assert_eq!(cfg.conversation_items.table, "conversation_items");
403        assert_eq!(cfg.conversation_item_links.table, "conversation_item_links");
404        assert_eq!(cfg.conversation_memories.table, "conversation_memories");
405        assert!(cfg.owner.is_none());
406    }
407
408    #[test]
409    fn default_config_validates_successfully() {
410        SchemaConfig::default()
411            .validate()
412            .expect("default config should be valid");
413    }
414
415    // ── col() ─────────────────────────────────────────────────────────────
416
417    #[test]
418    fn col_returns_field_name_when_no_override() {
419        let tc = TableConfig::with_table("t");
420        assert_eq!(tc.col("id"), "id");
421        assert_eq!(tc.col("created_at"), "created_at");
422    }
423
424    #[test]
425    fn col_returns_override_when_configured() {
426        let mut tc = TableConfig::with_table("t");
427        tc.columns
428            .insert("id".to_string(), "CONVERSATION_ID".to_string());
429        assert_eq!(tc.col("id"), "CONVERSATION_ID");
430        // Non-overridden field still returns itself
431        assert_eq!(tc.col("created_at"), "created_at");
432    }
433
434    // ── qualified_table() ─────────────────────────────────────────────────
435
436    #[test]
437    fn qualified_table_without_owner() {
438        let tc = TableConfig::with_table("conversations");
439        assert_eq!(tc.qualified_table(None), "conversations");
440    }
441
442    #[test]
443    fn qualified_table_with_owner() {
444        let tc = TableConfig::with_table("CONVERSATIONS");
445        assert_eq!(tc.qualified_table(Some("ADMIN")), "ADMIN.\"CONVERSATIONS\"");
446    }
447
448    // ── uppercase_for_oracle() ──────────────────────────────────────────
449
450    #[test]
451    fn uppercase_for_oracle_converts_defaults() {
452        let mut cfg = SchemaConfig::default();
453        cfg.uppercase_for_oracle();
454        assert_eq!(cfg.conversations.table, "CONVERSATIONS");
455        assert_eq!(cfg.responses.table, "RESPONSES");
456        assert_eq!(cfg.conversation_items.table, "CONVERSATION_ITEMS");
457        assert_eq!(cfg.conversation_item_links.table, "CONVERSATION_ITEM_LINKS");
458        assert_eq!(cfg.conversation_memories.table, "CONVERSATION_MEMORIES");
459        cfg.validate().expect("uppercased config should be valid");
460    }
461
462    #[test]
463    fn uppercase_for_oracle_converts_custom_table_and_columns() {
464        let mut cfg = SchemaConfig::default();
465        cfg.conversations.table = "my_convos".to_string();
466        cfg.conversations
467            .columns
468            .insert("id".to_string(), "conv_id".to_string());
469        cfg.uppercase_for_oracle();
470        assert_eq!(cfg.conversations.table, "MY_CONVOS");
471        assert_eq!(cfg.conversations.col("id"), "CONV_ID");
472    }
473
474    // ── validate() ────────────────────────────────────────────────────────
475
476    #[test]
477    fn validate_accepts_valid_identifiers() {
478        let mut cfg = SchemaConfig {
479            owner: Some("ADMIN_01".to_string()),
480            ..Default::default()
481        };
482        cfg.conversations.table = "MY_CONVERSATIONS".to_string();
483        cfg.conversations
484            .columns
485            .insert("id".to_string(), "CONV_ID".to_string());
486        cfg.validate().expect("should be valid");
487    }
488
489    #[test]
490    fn validate_rejects_empty_table_name_with_helpful_message() {
491        let mut cfg = SchemaConfig::default();
492        cfg.conversations.table = String::new();
493        let err = cfg.validate().unwrap_err();
494        assert!(
495            err.contains("conversations.table") && err.contains("table name is required"),
496            "unexpected: {err}"
497        );
498    }
499
500    #[test]
501    fn validate_rejects_overly_long_identifier() {
502        let mut cfg = SchemaConfig::default();
503        cfg.conversations.table = "a".repeat(129);
504        let err = cfg.validate().unwrap_err();
505        assert!(err.contains("exceeds maximum length"), "unexpected: {err}");
506    }
507
508    #[test]
509    fn validate_rejects_special_characters() {
510        let mut cfg = SchemaConfig::default();
511        cfg.conversations.table = "table;DROP".to_string();
512        let err = cfg.validate().unwrap_err();
513        assert!(
514            err.contains("conversations.table") && err.contains("invalid identifier"),
515            "unexpected: {err}"
516        );
517    }
518
519    #[test]
520    fn validate_rejects_dots_in_owner() {
521        let cfg = SchemaConfig {
522            owner: Some("ADMIN.SCHEMA".to_string()),
523            ..Default::default()
524        };
525        let err = cfg.validate().unwrap_err();
526        assert!(
527            err.contains("owner") && err.contains("invalid identifier"),
528            "unexpected: {err}"
529        );
530    }
531
532    #[test]
533    fn validate_rejects_invalid_column_override_key() {
534        let mut cfg = SchemaConfig::default();
535        cfg.responses
536            .columns
537            .insert("bad;col".to_string(), "GOOD_COL".to_string());
538        let err = cfg.validate().unwrap_err();
539        assert!(
540            err.contains("columns key") && err.contains("invalid identifier"),
541            "unexpected: {err}"
542        );
543    }
544
545    #[test]
546    fn validate_rejects_invalid_column_override_value() {
547        let mut cfg = SchemaConfig::default();
548        cfg.responses
549            .columns
550            .insert("id".to_string(), "bad column!".to_string());
551        let err = cfg.validate().unwrap_err();
552        assert!(
553            err.contains("columns value") && err.contains("invalid identifier"),
554            "unexpected: {err}"
555        );
556    }
557
558    // ── Serde roundtrip ───────────────────────────────────────────────────
559
560    #[test]
561    fn serde_roundtrip_default() {
562        let cfg = SchemaConfig::default();
563        let json = serde_json::to_string(&cfg).expect("serialize");
564        let restored: SchemaConfig = serde_json::from_str(&json).expect("deserialize");
565        assert_eq!(cfg, restored);
566    }
567
568    #[test]
569    fn serde_roundtrip_custom() {
570        let mut cfg = SchemaConfig {
571            owner: Some("ADMIN".to_string()),
572            ..Default::default()
573        };
574        cfg.conversations.table = "CONVERSATIONS".to_string();
575        cfg.conversations
576            .columns
577            .insert("id".to_string(), "CONVERSATION_ID".to_string());
578
579        let json = serde_json::to_string(&cfg).expect("serialize");
580        let restored: SchemaConfig = serde_json::from_str(&json).expect("deserialize");
581        assert_eq!(cfg, restored);
582    }
583
584    #[test]
585    fn serde_deserialize_empty_object_uses_defaults() {
586        let cfg: SchemaConfig = serde_json::from_str("{}").expect("deserialize empty");
587        assert_eq!(cfg, SchemaConfig::default());
588    }
589
590    // ── version / auto_migrate serde ────────────────────────────────────
591
592    #[test]
593    fn serde_roundtrip_with_version_and_auto_migrate() {
594        let cfg = SchemaConfig {
595            version: Some(3),
596            auto_migrate: false,
597            ..Default::default()
598        };
599        let json = serde_json::to_string(&cfg).expect("serialize");
600        let restored: SchemaConfig = serde_json::from_str(&json).expect("deserialize");
601        assert_eq!(restored.version, Some(3));
602        assert!(!restored.auto_migrate);
603    }
604
605    #[test]
606    fn serde_defaults_version_none_and_auto_migrate_false() {
607        let cfg: SchemaConfig = serde_json::from_str("{}").expect("deserialize empty");
608        assert_eq!(cfg.version, None);
609        assert!(!cfg.auto_migrate);
610    }
611
612    #[test]
613    fn serde_version_none_is_omitted_from_json() {
614        let cfg = SchemaConfig::default();
615        let json = serde_json::to_string(&cfg).expect("serialize");
616        assert!(
617            !json.contains("version"),
618            "version:None should be skipped: {json}"
619        );
620    }
621
622    // ── is_skipped() ─────────────────────────────────────────────────────
623
624    #[test]
625    fn is_skipped_returns_false_by_default() {
626        let tc = TableConfig::with_table("t");
627        assert!(!tc.is_skipped("id"));
628        assert!(!tc.is_skipped("safety_identifier"));
629    }
630
631    #[test]
632    fn is_skipped_returns_true_for_configured_fields() {
633        let mut tc = TableConfig::with_table("t");
634        tc.skip_columns.insert("safety_identifier".to_string());
635        tc.skip_columns.insert("raw_response".to_string());
636        assert!(tc.is_skipped("safety_identifier"));
637        assert!(tc.is_skipped("raw_response"));
638        assert!(!tc.is_skipped("id"));
639    }
640
641    // ── extra_columns ────────────────────────────────────────────────────
642
643    #[test]
644    fn extra_columns_default_is_empty() {
645        let tc = TableConfig::with_table("t");
646        assert!(tc.extra_columns.is_empty());
647    }
648
649    #[test]
650    fn validate_accepts_valid_extra_columns() {
651        let mut cfg = SchemaConfig::default();
652        cfg.conversations.extra_columns.insert(
653            "EXPIRES_AT".to_string(),
654            ColumnDef {
655                sql_type: "TIMESTAMP".to_string(),
656                default_value: None,
657            },
658        );
659        cfg.validate().expect("should be valid");
660    }
661
662    #[test]
663    fn validate_rejects_invalid_extra_column_name() {
664        let mut cfg = SchemaConfig::default();
665        cfg.conversations.extra_columns.insert(
666            "bad;col".to_string(),
667            ColumnDef {
668                sql_type: "TEXT".to_string(),
669                default_value: None,
670            },
671        );
672        let err = cfg.validate().unwrap_err();
673        assert!(
674            err.contains("extra_columns key") && err.contains("invalid identifier"),
675            "unexpected: {err}"
676        );
677    }
678
679    #[test]
680    fn validate_rejects_empty_sql_type() {
681        let mut cfg = SchemaConfig::default();
682        cfg.conversations.extra_columns.insert(
683            "MY_COL".to_string(),
684            ColumnDef {
685                sql_type: String::new(),
686                default_value: None,
687            },
688        );
689        let err = cfg.validate().unwrap_err();
690        assert!(
691            err.contains("sql_type must not be empty"),
692            "unexpected: {err}"
693        );
694    }
695
696    #[test]
697    fn validate_rejects_invalid_skip_column_name() {
698        let mut cfg = SchemaConfig::default();
699        cfg.responses.skip_columns.insert("bad;col".to_string());
700        let err = cfg.validate().unwrap_err();
701        assert!(
702            err.contains("skip_columns") && err.contains("invalid identifier"),
703            "unexpected: {err}"
704        );
705    }
706
707    #[test]
708    fn validate_accepts_valid_skip_columns() {
709        let mut cfg = SchemaConfig::default();
710        cfg.responses
711            .skip_columns
712            .insert("safety_identifier".to_string());
713        cfg.validate().expect("should be valid");
714    }
715
716    #[test]
717    fn validate_rejects_skip_id() {
718        let mut cfg = SchemaConfig::default();
719        cfg.responses.skip_columns.insert("id".to_string());
720        let err = cfg.validate().unwrap_err();
721        assert!(
722            err.contains("cannot skip 'id'") && err.contains("primary key"),
723            "unexpected: {err}"
724        );
725    }
726
727    #[test]
728    fn validate_rejects_skip_conversation_memory_id() {
729        let mut cfg = SchemaConfig::default();
730        cfg.conversation_memories
731            .skip_columns
732            .insert("memory_id".to_string());
733        let err = cfg.validate().unwrap_err();
734        assert!(
735            err.contains("cannot skip 'memory_id'") && err.contains("primary key"),
736            "unexpected: {err}"
737        );
738    }
739
740    // ── validate_sql_type ─────────────────────────────────────────────────
741
742    #[test]
743    fn validate_accepts_valid_sql_types() {
744        let mut cfg = SchemaConfig::default();
745        for sql_type in [
746            "VARCHAR(128)",
747            "TEXT",
748            "TIMESTAMP",
749            "TIMESTAMP WITH TIME ZONE",
750            "NUMBER(10,2)",
751            "VARCHAR2(256)",
752            "BIGINT",
753        ] {
754            cfg.conversations.extra_columns.insert(
755                "TEST_COL".to_string(),
756                ColumnDef {
757                    sql_type: sql_type.to_string(),
758                    default_value: None,
759                },
760            );
761            cfg.validate()
762                .unwrap_or_else(|e| panic!("sql_type '{sql_type}' should be valid: {e}"));
763        }
764    }
765
766    #[test]
767    fn validate_rejects_sql_injection_in_sql_type() {
768        let mut cfg = SchemaConfig::default();
769        cfg.conversations.extra_columns.insert(
770            "MY_COL".to_string(),
771            ColumnDef {
772                sql_type: "TEXT); DROP TABLE responses; --".to_string(),
773                default_value: None,
774            },
775        );
776        let err = cfg.validate().unwrap_err();
777        assert!(
778            err.contains("sql_type") && err.contains("invalid"),
779            "unexpected: {err}"
780        );
781    }
782
783    #[test]
784    fn validate_rejects_overly_long_sql_type() {
785        let mut cfg = SchemaConfig::default();
786        cfg.conversations.extra_columns.insert(
787            "MY_COL".to_string(),
788            ColumnDef {
789                sql_type: "A".repeat(65),
790                default_value: None,
791            },
792        );
793        let err = cfg.validate().unwrap_err();
794        assert!(
795            err.contains("sql_type") && err.contains("exceeds maximum length"),
796            "unexpected: {err}"
797        );
798    }
799
800    // ── uppercase_for_oracle with extra_columns ──────────────────────────
801
802    #[test]
803    fn uppercase_for_oracle_converts_extra_column_keys() {
804        let mut cfg = SchemaConfig::default();
805        cfg.conversations.extra_columns.insert(
806            "expires_at".to_string(),
807            ColumnDef {
808                sql_type: "TIMESTAMP".to_string(),
809                default_value: None,
810            },
811        );
812        cfg.uppercase_for_oracle();
813        assert!(cfg.conversations.extra_columns.contains_key("EXPIRES_AT"));
814        assert!(!cfg.conversations.extra_columns.contains_key("expires_at"));
815    }
816
817    #[test]
818    fn uppercase_for_oracle_preserves_skip_column_names_lowercase() {
819        let mut cfg = SchemaConfig::default();
820        cfg.responses
821            .skip_columns
822            .insert("safety_identifier".to_string());
823        cfg.uppercase_for_oracle();
824        // skip_columns must stay lowercase — backends call is_skipped("safety_identifier")
825        assert!(cfg.responses.skip_columns.contains("safety_identifier"));
826        assert!(!cfg.responses.skip_columns.contains("SAFETY_IDENTIFIER"));
827    }
828
829    // ── Serde roundtrip with extra_columns and skip_columns ──────────────
830
831    #[test]
832    fn serde_roundtrip_with_extra_and_skip() {
833        let mut cfg = SchemaConfig::default();
834        cfg.conversations.extra_columns.insert(
835            "EXPIRES_AT".to_string(),
836            ColumnDef {
837                sql_type: "TIMESTAMP".to_string(),
838                default_value: Some(Value::String("2099-01-01".to_string())),
839            },
840        );
841        cfg.responses
842            .skip_columns
843            .insert("safety_identifier".to_string());
844
845        let json = serde_json::to_string(&cfg).expect("serialize");
846        let restored: SchemaConfig = serde_json::from_str(&json).expect("deserialize");
847        assert_eq!(cfg, restored);
848    }
849
850    #[test]
851    fn validate_rejects_extra_column_shadowing_core_column() {
852        let mut cfg = SchemaConfig::default();
853        cfg.responses.extra_columns.insert(
854            "CREATED_AT".to_string(),
855            ColumnDef {
856                sql_type: "TIMESTAMP".to_string(),
857                default_value: None,
858            },
859        );
860        let err = cfg.validate().unwrap_err();
861        assert!(err.contains("shadows a core column"), "unexpected: {err}");
862
863        // Also check a conversation core column
864        let mut cfg2 = SchemaConfig::default();
865        cfg2.conversations.extra_columns.insert(
866            "metadata".to_string(),
867            ColumnDef {
868                sql_type: "TEXT".to_string(),
869                default_value: None,
870            },
871        );
872        let err2 = cfg2.validate().unwrap_err();
873        assert!(err2.contains("shadows a core column"), "unexpected: {err2}");
874    }
875
876    #[test]
877    fn validate_rejects_unknown_skip_column() {
878        let mut cfg = SchemaConfig::default();
879        cfg.responses
880            .skip_columns
881            .insert("safty_identifier".to_string()); // typo
882        let err = cfg.validate().unwrap_err();
883        assert!(err.contains("not a recognized column"), "unexpected: {err}");
884    }
885
886    #[test]
887    fn validate_rejects_case_colliding_extra_columns() {
888        let mut cfg = SchemaConfig::default();
889        cfg.conversations.extra_columns.insert(
890            "tenant_id".to_string(),
891            ColumnDef {
892                sql_type: "TEXT".to_string(),
893                default_value: None,
894            },
895        );
896        cfg.conversations.extra_columns.insert(
897            "TENANT_ID".to_string(),
898            ColumnDef {
899                sql_type: "TEXT".to_string(),
900                default_value: None,
901            },
902        );
903        let err = cfg.validate().unwrap_err();
904        assert!(
905            err.contains("case-insensitive collision"),
906            "unexpected: {err}"
907        );
908    }
909
910    #[test]
911    fn validate_accepts_conversation_memories_extra_columns() {
912        let mut cfg = SchemaConfig::default();
913        cfg.conversation_memories.extra_columns.insert(
914            "tenant_id".to_string(),
915            ColumnDef {
916                sql_type: "VARCHAR2(128)".to_string(),
917                default_value: None,
918            },
919        );
920        cfg.validate().expect("schema should validate");
921    }
922
923    #[test]
924    fn validate_rejects_conversation_memories_core_column_shadowing() {
925        let mut cfg = SchemaConfig::default();
926        cfg.conversation_memories.extra_columns.insert(
927            "memory_id".to_string(),
928            ColumnDef {
929                sql_type: "VARCHAR2(128)".to_string(),
930                default_value: None,
931            },
932        );
933        let err = cfg.validate().expect_err("shadowing should fail");
934        assert!(err.contains("conversation_memories.extra_columns"));
935        assert!(err.contains("shadows a core column name"));
936    }
937}
data_connector/schema.rs

data_connector/
schema.rs
