cipherstash_config/

canonical.rs

use std::collections::HashMap;
use std::str::FromStr;

use serde::{Deserialize, Serialize};

use crate::column::index::{K_DEFAULT, M_DEFAULT};
use crate::column::{ArrayIndexMode, Index, IndexType, SteVecMode, TokenFilter, Tokenizer};
use crate::errors::ConfigError;
use crate::{ColumnConfig, ColumnType};

/// The canonical plaintext type for a column in user-facing JSON config.
///
/// Maps to the internal [`ColumnType`] representation.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PlaintextType {
    BigInt,
    Boolean,
    Date,
    Decimal,
    #[serde(alias = "real", alias = "double")]
    Float,
    Int,
    #[serde(rename = "json", alias = "jsonb")]
    Json,
    SmallInt,
    #[default]
    Text,
    Timestamp,
}

impl std::fmt::Display for PlaintextType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BigInt => write!(f, "big_int"),
            Self::Boolean => write!(f, "boolean"),
            Self::Date => write!(f, "date"),
            Self::Decimal => write!(f, "decimal"),
            Self::Float => write!(f, "float"),
            Self::Int => write!(f, "int"),
            Self::Json => write!(f, "json"),
            Self::SmallInt => write!(f, "small_int"),
            Self::Text => write!(f, "text"),
            Self::Timestamp => write!(f, "timestamp"),
        }
    }
}

impl From<PlaintextType> for ColumnType {
    fn from(pt: PlaintextType) -> Self {
        match pt {
            PlaintextType::BigInt => ColumnType::BigInt,
            PlaintextType::Boolean => ColumnType::Boolean,
            PlaintextType::Date => ColumnType::Date,
            PlaintextType::Decimal => ColumnType::Decimal,
            PlaintextType::Float => ColumnType::Float,
            PlaintextType::Int => ColumnType::Int,
            PlaintextType::Json => ColumnType::Json,
            PlaintextType::SmallInt => ColumnType::SmallInt,
            PlaintextType::Text => ColumnType::Text,
            PlaintextType::Timestamp => ColumnType::Timestamp,
        }
    }
}

/// Stable identifier for a table/column pair, used as a key in config maps.
#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
pub struct Identifier {
    #[serde(rename = "t")]
    pub table: String,
    #[serde(rename = "c")]
    pub column: String,
}

impl Identifier {
    pub fn new(table: impl Into<String>, column: impl Into<String>) -> Self {
        Self {
            table: table.into(),
            column: column.into(),
        }
    }
}

impl std::fmt::Display for Identifier {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}.{}", self.table, self.column)
    }
}

/// Top-level canonical encryption configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CanonicalEncryptionConfig {
    #[serde(rename = "v")]
    pub version: u32,
    pub tables: Tables,
}

/// A mapping of table names to their column encryption configurations.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Tables(pub HashMap<String, Table>);

/// A mapping of column names to their encryption configuration within a single table.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Table(pub HashMap<String, Column>);

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Column {
    #[serde(default, alias = "cast_as")]
    pub plaintext_type: PlaintextType,
    #[serde(default)]
    pub indexes: Indexes,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Indexes {
    pub ore: Option<OreIndexOpts>,
    pub ope: Option<OpeIndexOpts>,
    pub unique: Option<UniqueIndexOpts>,
    #[serde(rename = "match")]
    pub match_index: Option<MatchIndexOpts>,
    pub ste_vec: Option<SteVecIndexOpts>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OreIndexOpts {}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OpeIndexOpts {}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UniqueIndexOpts {
    #[serde(default)]
    pub token_filters: Vec<TokenFilter>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MatchIndexOpts {
    #[serde(default = "default_tokenizer")]
    pub tokenizer: Tokenizer,
    #[serde(default)]
    pub token_filters: Vec<TokenFilter>,
    #[serde(default = "default_k")]
    pub k: usize,
    #[serde(default = "default_m")]
    pub m: usize,
    #[serde(default)]
    pub include_original: bool,
}

impl Default for MatchIndexOpts {
    fn default() -> Self {
        Self {
            tokenizer: Tokenizer::Standard,
            token_filters: vec![],
            k: K_DEFAULT,
            m: M_DEFAULT,
            include_original: false,
        }
    }
}

fn default_tokenizer() -> Tokenizer {
    Tokenizer::Standard
}

fn default_k() -> usize {
    K_DEFAULT
}

fn default_m() -> usize {
    M_DEFAULT
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SteVecIndexOpts {
    pub prefix: String,
    #[serde(default)]
    pub term_filters: Vec<TokenFilter>,
    #[serde(default = "default_array_index_mode")]
    pub array_index_mode: ArrayIndexMode,
    #[serde(default)]
    pub mode: SteVecMode,
}

fn default_array_index_mode() -> ArrayIndexMode {
    ArrayIndexMode::ALL
}

impl FromStr for CanonicalEncryptionConfig {
    type Err = ConfigError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        serde_json::from_str(s).map_err(|e| ConfigError::ParseError(e.to_string()))
    }
}

impl CanonicalEncryptionConfig {
    pub fn into_config_map(self) -> Result<HashMap<Identifier, ColumnConfig>, ConfigError> {
        if self.version != 1 {
            return Err(ConfigError::UnsupportedVersion {
                version: self.version,
                expected: 1,
            });
        }

        let mut map = HashMap::new();

        for (table_name, table) in self.tables.0 {
            for (column_name, column) in table.0 {
                let identifier = Identifier::new(&table_name, &column_name);
                let config = column.into_column_config(&table_name, &column_name)?;
                map.insert(identifier, config);
            }
        }

        Ok(map)
    }
}

impl Column {
    fn into_column_config(
        self,
        table_name: &str,
        column_name: &str,
    ) -> Result<ColumnConfig, ConfigError> {
        let column_type: ColumnType = self.plaintext_type.into();

        if self.indexes.ste_vec.is_some() && self.plaintext_type != PlaintextType::Json {
            return Err(ConfigError::SteVecRequiresJson {
                table: table_name.to_owned(),
                column: column_name.to_owned(),
                found_plaintext_type: self.plaintext_type.to_string(),
            });
        }

        if self.indexes.match_index.is_some() && self.plaintext_type != PlaintextType::Text {
            return Err(ConfigError::MatchRequiresText {
                table: table_name.to_owned(),
                column: column_name.to_owned(),
                found_plaintext_type: self.plaintext_type.to_string(),
            });
        }

        let mut config = ColumnConfig::build(column_name).casts_as(column_type);

        if self.indexes.ore.is_some() {
            config = config.add_index(Index::new_ore());
        }

        if self.indexes.ope.is_some() {
            config = config.add_index(Index::new_ope());
        }

        if let Some(unique_opts) = self.indexes.unique {
            config = config.add_index(Index::new(IndexType::Unique {
                token_filters: unique_opts.token_filters,
            }));
        }

        if let Some(match_opts) = self.indexes.match_index {
            config = config.add_index(Index::new(IndexType::Match {
                tokenizer: match_opts.tokenizer,
                token_filters: match_opts.token_filters,
                k: match_opts.k,
                m: match_opts.m,
                include_original: match_opts.include_original,
            }));
        }

        if let Some(ste_vec_opts) = self.indexes.ste_vec {
            config = config.add_index(Index::new(IndexType::SteVec {
                prefix: ste_vec_opts.prefix,
                term_filters: ste_vec_opts.term_filters,
                array_index_mode: ste_vec_opts.array_index_mode,
                mode: ste_vec_opts.mode,
            }));
        }

        Ok(config)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn it_deserializes_all_plaintext_types() {
        let cases = vec![
            ("text", PlaintextType::Text),
            ("int", PlaintextType::Int),
            ("small_int", PlaintextType::SmallInt),
            ("big_int", PlaintextType::BigInt),
            ("float", PlaintextType::Float),
            ("boolean", PlaintextType::Boolean),
            ("date", PlaintextType::Date),
            ("json", PlaintextType::Json),
            ("decimal", PlaintextType::Decimal),
            ("timestamp", PlaintextType::Timestamp),
        ];

        for (input, expected) in cases {
            let result: PlaintextType = serde_json::from_value(json!(input)).unwrap();
            assert_eq!(result, expected, "Failed for input: {input}");
        }
    }

    #[test]
    fn it_defaults_to_text() {
        let pt: PlaintextType = Default::default();
        assert_eq!(pt, PlaintextType::Text);
    }

    #[test]
    fn it_accepts_jsonb_alias() {
        let result: PlaintextType = serde_json::from_value(json!("jsonb")).unwrap();
        assert_eq!(result, PlaintextType::Json);
    }

    #[test]
    fn it_accepts_real_alias() {
        let result: PlaintextType = serde_json::from_value(json!("real")).unwrap();
        assert_eq!(result, PlaintextType::Float);
    }

    #[test]
    fn it_accepts_double_alias() {
        let result: PlaintextType = serde_json::from_value(json!("double")).unwrap();
        assert_eq!(result, PlaintextType::Float);
    }

    #[test]
    fn it_converts_to_column_type() {
        assert_eq!(ColumnType::from(PlaintextType::Text), ColumnType::Text);
        assert_eq!(ColumnType::from(PlaintextType::Int), ColumnType::Int);
        assert_eq!(
            ColumnType::from(PlaintextType::SmallInt),
            ColumnType::SmallInt
        );
        assert_eq!(ColumnType::from(PlaintextType::BigInt), ColumnType::BigInt);
        assert_eq!(ColumnType::from(PlaintextType::Float), ColumnType::Float);
        assert_eq!(
            ColumnType::from(PlaintextType::Boolean),
            ColumnType::Boolean
        );
        assert_eq!(ColumnType::from(PlaintextType::Date), ColumnType::Date);
        assert_eq!(ColumnType::from(PlaintextType::Json), ColumnType::Json);
        assert_eq!(
            ColumnType::from(PlaintextType::Decimal),
            ColumnType::Decimal
        );
        assert_eq!(
            ColumnType::from(PlaintextType::Timestamp),
            ColumnType::Timestamp
        );
    }

    #[test]
    fn it_serializes_to_canonical_names() {
        assert_eq!(
            serde_json::to_value(PlaintextType::Text).unwrap(),
            json!("text")
        );
        assert_eq!(
            serde_json::to_value(PlaintextType::Json).unwrap(),
            json!("json")
        );
        assert_eq!(
            serde_json::to_value(PlaintextType::BigInt).unwrap(),
            json!("big_int")
        );
    }

    #[test]
    fn it_parses_minimal_config() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {}
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        assert_eq!(config.version, 1);
    }

    #[test]
    fn it_accepts_cast_as_field_name() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "cast_as": "int"
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let table = config.tables.0.get("users").unwrap();
        let column = table.0.get("email").unwrap();
        assert_eq!(column.plaintext_type, PlaintextType::Int);
    }

    #[test]
    fn it_accepts_plaintext_type_field_name() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "plaintext_type": "int"
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let table = config.tables.0.get("users").unwrap();
        let column = table.0.get("email").unwrap();
        assert_eq!(column.plaintext_type, PlaintextType::Int);
    }

    #[test]
    fn it_parses_ore_index() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "age": {
                        "plaintext_type": "int",
                        "indexes": { "ore": {} }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config.tables.0.get("users").unwrap().0.get("age").unwrap();
        assert!(col.indexes.ore.is_some());
    }

    #[test]
    fn it_parses_ope_index() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "age": {
                        "plaintext_type": "int",
                        "indexes": { "ope": {} }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config.tables.0.get("users").unwrap().0.get("age").unwrap();
        assert!(col.indexes.ope.is_some());
    }

    #[test]
    fn it_round_trips_ope_index_through_json() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "age": {
                        "plaintext_type": "int",
                        "indexes": { "ope": {} }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input.clone()).unwrap();
        let serialized = serde_json::to_value(&config).unwrap();
        let reparsed: CanonicalEncryptionConfig = serde_json::from_value(serialized).unwrap();

        let col = reparsed
            .tables
            .0
            .get("users")
            .unwrap()
            .0
            .get("age")
            .unwrap();
        assert!(col.indexes.ope.is_some());
        assert!(col.indexes.ore.is_none());
    }

    #[test]
    fn it_parses_match_index_with_defaults() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "name": {
                        "plaintext_type": "text",
                        "indexes": { "match": {} }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config.tables.0.get("users").unwrap().0.get("name").unwrap();
        let match_opts = col.indexes.match_index.as_ref().unwrap();
        assert_eq!(match_opts.tokenizer, Tokenizer::Standard);
        assert_eq!(match_opts.k, 6);
        assert_eq!(match_opts.m, 2048);
        assert!(!match_opts.include_original);
        assert!(match_opts.token_filters.is_empty());
    }

    #[test]
    fn it_parses_unique_index() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "plaintext_type": "text",
                        "indexes": {
                            "unique": {
                                "token_filters": [{ "kind": "downcase" }]
                            }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config
            .tables
            .0
            .get("users")
            .unwrap()
            .0
            .get("email")
            .unwrap();
        let unique_opts = col.indexes.unique.as_ref().unwrap();
        assert_eq!(unique_opts.token_filters.len(), 1);
    }

    #[test]
    fn it_parses_ste_vec_index() {
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "plaintext_type": "json",
                        "indexes": {
                            "ste_vec": {
                                "prefix": "event-data",
                                "array_index_mode": "all"
                            }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config
            .tables
            .0
            .get("events")
            .unwrap()
            .0
            .get("data")
            .unwrap();
        let ste_vec_opts = col.indexes.ste_vec.as_ref().unwrap();
        assert_eq!(ste_vec_opts.prefix, "event-data");
    }

    #[test]
    fn it_defaults_ste_vec_mode_to_compat() {
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "plaintext_type": "json",
                        "indexes": {
                            "ste_vec": { "prefix": "event-data" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config
            .tables
            .0
            .get("events")
            .unwrap()
            .0
            .get("data")
            .unwrap();
        let ste_vec_opts = col.indexes.ste_vec.as_ref().unwrap();
        assert_eq!(ste_vec_opts.mode, SteVecMode::Compat);
    }

    #[test]
    fn it_serializes_defaulted_ste_vec_mode_as_compat() {
        // A config parsed without an explicit mode re-serializes with the
        // defaulted mode pinned on the wire as "compat".
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "plaintext_type": "json",
                        "indexes": {
                            "ste_vec": { "prefix": "event-data" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config
            .tables
            .0
            .get("events")
            .unwrap()
            .0
            .get("data")
            .unwrap();
        let ste_vec_opts = col.indexes.ste_vec.as_ref().unwrap();
        let serialized = serde_json::to_value(ste_vec_opts).unwrap();
        assert_eq!(serialized["mode"], json!("compat"));
    }

    #[test]
    fn it_parses_ste_vec_mode_standard() {
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "plaintext_type": "json",
                        "indexes": {
                            "ste_vec": { "prefix": "event-data", "mode": "standard" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config
            .tables
            .0
            .get("events")
            .unwrap()
            .0
            .get("data")
            .unwrap();
        let ste_vec_opts = col.indexes.ste_vec.as_ref().unwrap();
        assert_eq!(ste_vec_opts.mode, SteVecMode::Standard);
    }

    #[test]
    fn it_propagates_ste_vec_mode_into_config_map() {
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "plaintext_type": "json",
                        "indexes": {
                            "ste_vec": { "prefix": "event-data", "mode": "standard" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();
        let col = map.get(&Identifier::new("events", "data")).unwrap();

        assert!(matches!(
            col.indexes[0].index_type,
            IndexType::SteVec {
                mode: SteVecMode::Standard,
                ..
            }
        ));
    }

    #[test]
    fn it_parses_empty_indexes() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "plaintext_type": "text",
                        "indexes": {}
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let col = config
            .tables
            .0
            .get("users")
            .unwrap()
            .0
            .get("email")
            .unwrap();
        assert!(col.indexes.ore.is_none());
        assert!(col.indexes.unique.is_none());
        assert!(col.indexes.match_index.is_none());
        assert!(col.indexes.ste_vec.is_none());
    }

    #[test]
    fn it_converts_to_config_map() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "plaintext_type": "text",
                        "indexes": {
                            "ore": {},
                            "unique": { "token_filters": [{ "kind": "downcase" }] }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();

        let id = Identifier::new("users", "email");
        let col = map.get(&id).unwrap();
        assert_eq!(col.cast_type, ColumnType::Text);
        assert_eq!(col.indexes.len(), 2);
    }

    #[test]
    fn it_defaults_empty_column_to_text() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {}
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();

        let id = Identifier::new("users", "email");
        let col = map.get(&id).unwrap();
        assert_eq!(col.cast_type, ColumnType::Text);
        assert!(col.indexes.is_empty());
    }

    #[test]
    fn it_rejects_ste_vec_on_non_json_column() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "plaintext_type": "text",
                        "indexes": {
                            "ste_vec": { "prefix": "test" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let result = config.into_config_map();
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("ste_vec"),
            "Error should mention ste_vec: {err}"
        );
        assert!(err.contains("json"), "Error should mention json: {err}");
    }

    #[test]
    fn it_allows_ste_vec_on_json_column() {
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "plaintext_type": "json",
                        "indexes": {
                            "ste_vec": { "prefix": "event-data" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();

        let id = Identifier::new("events", "data");
        let col = map.get(&id).unwrap();
        assert_eq!(col.cast_type, ColumnType::Json);
    }

    #[test]
    fn it_parses_from_json_string() {
        let json_str =
            r#"{"v":1,"tables":{"t":{"c":{"plaintext_type":"int","indexes":{"ore":{}}}}}}"#;
        let config: CanonicalEncryptionConfig = json_str.parse().unwrap();
        let map = config.into_config_map().unwrap();
        let col = map.get(&Identifier::new("t", "c")).unwrap();
        assert_eq!(col.cast_type, ColumnType::Int);
    }

    #[test]
    fn it_handles_backwards_compat_cast_as_jsonb() {
        let input = json!({
            "v": 1,
            "tables": {
                "events": {
                    "data": {
                        "cast_as": "jsonb",
                        "indexes": {
                            "ste_vec": { "prefix": "test" }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();
        let id = Identifier::new("events", "data");
        let col = map.get(&id).unwrap();
        assert_eq!(col.cast_type, ColumnType::Json);
    }

    #[test]
    fn it_produces_correct_index_types_for_multi_index_column() {
        let input = json!({
            "v": 1,
            "tables": {
                "encrypted": {
                    "encrypted_text": {
                        "cast_as": "text",
                        "indexes": {
                            "unique": {},
                            "match": {},
                            "ore": {}
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();

        let id = Identifier::new("encrypted", "encrypted_text");
        let col = map.get(&id).unwrap();

        assert_eq!(col.cast_type, ColumnType::Text);
        assert_eq!(col.name, "encrypted_text");
        assert_eq!(col.indexes.len(), 3);

        let index_types: Vec<_> = col.indexes.iter().map(|i| &i.index_type).collect();
        assert!(index_types.contains(&&IndexType::Ore));
        assert!(index_types
            .iter()
            .any(|t| matches!(t, IndexType::Unique { .. })));
        assert!(index_types
            .iter()
            .any(|t| matches!(t, IndexType::Match { .. })));
    }

    #[test]
    fn it_maps_all_cast_as_values_to_correct_column_types() {
        let cases = vec![
            ("text", ColumnType::Text),
            ("int", ColumnType::Int),
            ("small_int", ColumnType::SmallInt),
            ("big_int", ColumnType::BigInt),
            ("boolean", ColumnType::Boolean),
            ("date", ColumnType::Date),
            ("float", ColumnType::Float),
            ("decimal", ColumnType::Decimal),
            ("timestamp", ColumnType::Timestamp),
            // Legacy aliases
            ("double", ColumnType::Float),
            ("real", ColumnType::Float),
            ("jsonb", ColumnType::Json),
            ("json", ColumnType::Json),
        ];

        for (cast_as, expected_type) in cases {
            let input = json!({
                "v": 1,
                "tables": {
                    "t": {
                        "c": { "cast_as": cast_as }
                    }
                }
            });

            let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
            let map = config.into_config_map().unwrap();
            let col = map.get(&Identifier::new("t", "c")).unwrap();
            assert_eq!(
                col.cast_type, expected_type,
                "Failed for cast_as: {cast_as}"
            );
        }
    }

    #[test]
    fn it_preserves_match_index_defaults_in_config_map() {
        let input = json!({
            "v": 1,
            "tables": {
                "t": {
                    "c": {
                        "cast_as": "text",
                        "indexes": { "match": {} }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();
        let col = map.get(&Identifier::new("t", "c")).unwrap();

        assert_eq!(col.indexes.len(), 1);
        assert_eq!(
            col.indexes[0].index_type,
            IndexType::Match {
                tokenizer: Tokenizer::Standard,
                token_filters: vec![],
                k: 6,
                m: 2048,
                include_original: false,
            }
        );
    }

    /// This fixture represents the JSON stored in `eql_v2_configuration.data`
    /// after running the proxy integration test schema setup.
    /// If this test breaks, existing production configs will fail to load.
    #[test]
    fn it_parses_real_eql_integration_test_config() {
        let input = json!({
            "v": 1,
            "tables": {
                "encrypted": {
                    "encrypted_text": {
                        "cast_as": "text",
                        "indexes": {
                            "unique": {},
                            "match": {},
                            "ore": {}
                        }
                    },
                    "encrypted_bool": {
                        "cast_as": "boolean",
                        "indexes": {
                            "unique": {},
                            "ore": {}
                        }
                    },
                    "encrypted_int2": {
                        "cast_as": "small_int",
                        "indexes": {
                            "unique": {},
                            "ore": {}
                        }
                    },
                    "encrypted_int4": {
                        "cast_as": "int",
                        "indexes": {
                            "unique": {},
                            "ore": {}
                        }
                    },
                    "encrypted_int8": {
                        "cast_as": "big_int",
                        "indexes": {
                            "unique": {},
                            "ore": {}
                        }
                    },
                    "encrypted_float8": {
                        "cast_as": "double",
                        "indexes": {
                            "unique": {},
                            "ore": {}
                        }
                    },
                    "encrypted_date": {
                        "cast_as": "date",
                        "indexes": {
                            "unique": {},
                            "ore": {}
                        }
                    },
                    "encrypted_jsonb": {
                        "cast_as": "jsonb",
                        "indexes": {
                            "ste_vec": {
                                "prefix": "encrypted/encrypted_jsonb"
                            }
                        }
                    },
                    "encrypted_jsonb_filtered": {
                        "cast_as": "jsonb",
                        "indexes": {
                            "ste_vec": {
                                "prefix": "encrypted/encrypted_jsonb_filtered",
                                "term_filters": [{ "kind": "downcase" }]
                            }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();

        // Verify all 9 columns are present
        assert_eq!(map.len(), 9);

        // Spot check key columns
        let text_col = map
            .get(&Identifier::new("encrypted", "encrypted_text"))
            .unwrap();
        assert_eq!(text_col.cast_type, ColumnType::Text);
        assert_eq!(text_col.indexes.len(), 3);

        let bool_col = map
            .get(&Identifier::new("encrypted", "encrypted_bool"))
            .unwrap();
        assert_eq!(bool_col.cast_type, ColumnType::Boolean);
        assert_eq!(bool_col.indexes.len(), 2);

        let float_col = map
            .get(&Identifier::new("encrypted", "encrypted_float8"))
            .unwrap();
        assert_eq!(float_col.cast_type, ColumnType::Float); // "double" maps to Float

        let jsonb_col = map
            .get(&Identifier::new("encrypted", "encrypted_jsonb"))
            .unwrap();
        assert_eq!(jsonb_col.cast_type, ColumnType::Json); // "jsonb" maps to Json
        assert_eq!(jsonb_col.indexes.len(), 1);
        assert!(matches!(
            jsonb_col.indexes[0].index_type,
            IndexType::SteVec { ref prefix, .. } if prefix == "encrypted/encrypted_jsonb"
        ));

        let filtered_col = map
            .get(&Identifier::new("encrypted", "encrypted_jsonb_filtered"))
            .unwrap();
        assert!(matches!(
            &filtered_col.indexes[0].index_type,
            IndexType::SteVec { term_filters, .. } if term_filters.len() == 1
        ));
    }

    #[test]
    fn it_rejects_unsupported_version() {
        let input = json!({
            "v": 2,
            "tables": {
                "users": {
                    "email": {}
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let result = config.into_config_map();
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("unsupported config version"), "Error: {err}");
    }

    #[test]
    fn it_rejects_match_index_on_non_text_column() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "age": {
                        "plaintext_type": "int",
                        "indexes": { "match": {} }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let result = config.into_config_map();
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("match"), "Error should mention match: {err}");
        assert!(err.contains("text"), "Error should mention text: {err}");
    }

    #[test]
    fn it_displays_identifier() {
        let id = Identifier::new("users", "email");
        assert_eq!(id.to_string(), "users.email");
    }

    #[test]
    fn it_silently_ignores_dropped_legacy_fields() {
        let input = json!({
            "v": 1,
            "tables": {
                "users": {
                    "email": {
                        "cast_as": "text",
                        "mode": "encrypted",
                        "in_place": true,
                        "indexes": {
                            "unique": {
                                "token_filters": [{ "kind": "downcase" }],
                                "mode": "encrypted",
                                "in_place": false
                            }
                        }
                    }
                }
            }
        });

        let config: CanonicalEncryptionConfig = serde_json::from_value(input).unwrap();
        let map = config.into_config_map().unwrap();
        let col = map.get(&Identifier::new("users", "email")).unwrap();
        assert_eq!(col.cast_type, ColumnType::Text);
        assert_eq!(col.indexes.len(), 1);
    }
}