evenframe_core 0.2.0

//! Schema comparison module - Database-agnostic core
//!
//! This module provides schema comparison functionality that works across
//! different database providers. Provider-specific implementations are in
//! submodules (surrealdb.rs, sql.rs).

pub mod filter;
#[cfg(feature = "surrealdb")]
pub mod import;
#[cfg(feature = "sql")]
pub mod sql;
#[cfg(feature = "surrealdb")]
pub mod surql;
pub mod types;

// Re-export commonly used types
pub use crate::schemasync::mockmake::MockGenerationConfig;
#[cfg(feature = "surrealdb")]
pub use import::SchemaImporter;
#[cfg(feature = "surrealdb")]
pub use surql::SurrealdbComparator;
pub use types::{
    AccessDefinition, AnalyzerDefinition, FieldDefinition, IndexDefinition, ObjectType,
    PermissionSet, SchemaDefinition, SchemaType, TableDefinition,
};

#[cfg(feature = "surrealdb")]
use crate::schemasync::config::{PerformanceConfig, SchemasyncMockGenConfig};
use crate::{
    EvenframeError, Result,
    schemasync::TableConfig,
    types::{FieldType, TaggedUnion, VariantData},
};
#[cfg(feature = "surrealdb")]
use ::surrealdb::Surreal;
#[cfg(feature = "surrealdb")]
use ::surrealdb::engine::remote::http::Client;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::collections::BTreeSet;

#[cfg(feature = "schemasync")]
use crate::schemasync::database::types::SchemaExport;

// ============================================================================
// SchemaComparator Trait - Database-agnostic interface
// ============================================================================

/// Trait for database schema comparison strategies
#[cfg(feature = "schemasync")]
#[async_trait::async_trait]
pub trait SchemaComparator: Send + Sync {
    /// Compare the current database schema with the expected schema from Rust structs
    async fn compare_schemas(
        &self,
        tables: &BTreeMap<String, TableConfig>,
        objects: &BTreeMap<String, crate::types::StructConfig>,
        enums: &BTreeMap<String, TaggedUnion>,
    ) -> Result<SchemaChanges>;

    /// Get the current schema from the database
    async fn get_current_schema(&self) -> Result<SchemaExport>;

    /// Check if this comparator supports embedded mode for comparison
    fn supports_embedded_comparison(&self) -> bool;
}

/// Factory function to create the appropriate comparator for a provider
#[cfg(feature = "schemasync")]
pub fn create_comparator<'a>(
    _provider: &'a dyn crate::schemasync::database::DatabaseProvider,
) -> Box<dyn SchemaComparator + 'a> {
    #[cfg(feature = "sql")]
    {
        Box::new(sql::SqlSchemaComparator::new(_provider))
    }
    #[cfg(not(feature = "sql"))]
    {
        panic!("No SQL feature enabled - use SurrealdbComparator directly for SurrealDB")
    }
}

// ============================================================================
// Comparator - Legacy struct for backward compatibility with static methods
// ============================================================================

/// Legacy Comparator struct - provides static comparison methods
/// For SurrealDB-specific comparator, use SurrealdbComparator
pub struct Comparator;

// ============================================================================
// Schema Change Types
// ============================================================================

pub use super::PreservationMode;

/// Types of changes that can occur in an access definition
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum AccessChangeType {
    JwtKeyChanged,
    IssuerKeyChanged,
    JwtUrlChanged,
    AuthenticateClauseChanged,
    DurationChanged,
    SigninChanged,
    SignupChanged,
    OtherChange(String),
}

impl AccessChangeType {
    /// Check if this change type is ignorable (e.g., rotating keys)
    pub fn is_ignorable(&self) -> bool {
        matches!(
            self,
            AccessChangeType::JwtKeyChanged | AccessChangeType::IssuerKeyChanged
        )
    }
}

impl std::fmt::Display for AccessChangeType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AccessChangeType::JwtKeyChanged => write!(f, "JWT key changed"),
            AccessChangeType::IssuerKeyChanged => write!(f, "Issuer key changed"),
            AccessChangeType::JwtUrlChanged => write!(f, "JWT URL changed"),
            AccessChangeType::AuthenticateClauseChanged => write!(f, "Authenticate clause changed"),
            AccessChangeType::DurationChanged => write!(f, "EvenframeDuration changed"),
            AccessChangeType::SigninChanged => write!(f, "Signin changed"),
            AccessChangeType::SignupChanged => write!(f, "Signup changed"),
            AccessChangeType::OtherChange(msg) => write!(f, "{}", msg),
        }
    }
}

/// Represents changes between two schemas
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaChanges {
    pub new_tables: Vec<String>,
    pub removed_tables: Vec<String>,
    pub modified_tables: Vec<TableChanges>,
    pub new_accesses: Vec<String>,
    pub removed_accesses: Vec<String>,
    pub modified_accesses: Vec<AccessChange>,
    #[serde(default)]
    pub new_analyzers: Vec<String>,
    #[serde(default)]
    pub removed_analyzers: Vec<String>,
    #[serde(default)]
    pub modified_analyzers: Vec<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AccessChange {
    pub access_name: String,
    pub changes: Vec<AccessChangeType>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableChanges {
    pub table_name: String,
    pub new_fields: Vec<String>,
    pub removed_fields: Vec<String>,
    pub modified_fields: Vec<FieldChange>,
    pub permission_changed: bool,
    pub schema_type_changed: bool,
    pub new_events: Vec<String>,
    pub removed_events: Vec<String>,
    pub new_indexes: Vec<IndexDefinition>,
    pub removed_indexes: Vec<IndexDefinition>,
    /// Same-name indexes whose definition (kind, parameters, comment) changed.
    #[serde(default)]
    pub modified_indexes: Vec<IndexDefinition>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum ChangeType {
    Added,
    Removed,
    Modified,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FieldChange {
    pub field_name: String,
    pub old_type: String,
    pub new_type: String,
    pub change_type: ChangeType,
    pub required_changed: bool,
    pub default_changed: bool,
}

impl SchemaChanges {
    /// Check if a specific field is unchanged
    pub fn is_field_unchanged(&self, table: &str, field: &str) -> bool {
        // If table is new or removed, field is not unchanged
        if self.new_tables.contains(&table.to_string())
            || self.removed_tables.contains(&table.to_string())
        {
            return false;
        }

        // Check modified tables
        for table_change in &self.modified_tables {
            if table_change.table_name == table {
                // If field is new or removed, it's not unchanged
                if table_change.new_fields.contains(&field.to_string())
                    || table_change.removed_fields.contains(&field.to_string())
                {
                    return false;
                }

                // If field is modified, it's not unchanged
                for field_change in &table_change.modified_fields {
                    if field_change.field_name == field {
                        return false;
                    }
                }

                // Field exists in table and is not in any change list
                return true;
            }
        }

        // Table not in modified list, so if it exists in old schema, field is unchanged
        true
    }

    /// Get all fields that need new data generation
    pub fn get_fields_needing_generation(&self, table: &str) -> Vec<String> {
        let mut fields = Vec::new();

        // If table is new, all fields need generation
        if self.new_tables.contains(&table.to_string()) {
            return vec!["*".to_string()]; // Special marker for all fields
        }

        // Find table in modified tables
        for table_change in &self.modified_tables {
            if table_change.table_name == table {
                // Add all new fields
                fields.extend(table_change.new_fields.clone());

                // Optionally add modified fields based on configuration
                // For now, we'll regenerate modified fields
                for field_change in &table_change.modified_fields {
                    fields.push(field_change.field_name.clone());
                }
            }
        }

        fields
    }

    /// Create a summary of changes
    pub fn summary(&self) -> String {
        let mut summary = Vec::new();

        if !self.new_tables.is_empty() {
            summary.push(format!("New tables: {}", self.new_tables.join(", ")));
        }

        if !self.removed_tables.is_empty() {
            summary.push(format!(
                "Removed tables: {}",
                self.removed_tables.join(", ")
            ));
        }

        if !self.modified_tables.is_empty() {
            summary.push(format!(
                "Modified tables: {}",
                self.modified_tables
                    .iter()
                    .map(|t| t.table_name.clone())
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }

        for table in &self.modified_tables {
            if !table.new_events.is_empty() {
                summary.push(format!(
                    "New events on {}: {}",
                    table.table_name,
                    table.new_events.join(", ")
                ));
            }
            if !table.removed_events.is_empty() {
                summary.push(format!(
                    "Removed events on {}: {}",
                    table.table_name,
                    table.removed_events.join(", ")
                ));
            }
            if !table.new_indexes.is_empty() {
                let names: Vec<&str> = table.new_indexes.iter().map(|i| i.name.as_str()).collect();
                summary.push(format!(
                    "New indexes on {}: {}",
                    table.table_name,
                    names.join(", ")
                ));
            }
            if !table.removed_indexes.is_empty() {
                let names: Vec<&str> = table
                    .removed_indexes
                    .iter()
                    .map(|i| i.name.as_str())
                    .collect();
                summary.push(format!(
                    "Removed indexes on {}: {}",
                    table.table_name,
                    names.join(", ")
                ));
            }
            if !table.modified_indexes.is_empty() {
                let names: Vec<&str> = table
                    .modified_indexes
                    .iter()
                    .map(|i| i.name.as_str())
                    .collect();
                summary.push(format!(
                    "Modified indexes on {}: {}",
                    table.table_name,
                    names.join(", ")
                ));
            }
        }

        for (label, names) in [
            ("New analyzers", &self.new_analyzers),
            ("Removed analyzers", &self.removed_analyzers),
            ("Modified analyzers", &self.modified_analyzers),
        ] {
            if !names.is_empty() {
                summary.push(format!("{label}: {}", names.join(", ")));
            }
        }

        if !self.new_accesses.is_empty() {
            summary.push(format!("New accesses: {}", self.new_accesses.join(", ")));
        }

        if !self.removed_accesses.is_empty() {
            summary.push(format!(
                "Removed accesses: {}",
                self.removed_accesses.join(", ")
            ));
        }

        if !self.modified_accesses.is_empty() {
            summary.push(format!(
                "Modified accesses: {}",
                self.modified_accesses
                    .iter()
                    .map(|a| a.access_name.clone())
                    .collect::<Vec<_>>()
                    .join(", ")
            ));
        }

        if summary.is_empty() {
            "No changes detected".to_string()
        } else {
            summary.join("\n")
        }
    }
}

// ============================================================================
// Comparator Static Methods - Database-agnostic comparison logic
// ============================================================================

impl Comparator {
    /// Compare two schemas and return the differences
    pub fn compare(old: &SchemaDefinition, new: &SchemaDefinition) -> Result<SchemaChanges> {
        tracing::debug!("Starting detailed schema comparison");

        let mut changes = SchemaChanges {
            new_tables: Vec::new(),
            removed_tables: Vec::new(),
            modified_tables: Vec::new(),
            new_accesses: Vec::new(),
            removed_accesses: Vec::new(),
            modified_accesses: Vec::new(),
            new_analyzers: Vec::new(),
            removed_analyzers: Vec::new(),
            modified_analyzers: Vec::new(),
        };

        // Get all table names from both schemas
        let old_tables: BTreeSet<String> = old.tables.keys().cloned().collect();
        let new_tables: BTreeSet<String> = new.tables.keys().cloned().collect();

        tracing::trace!(
            old_table_count = old_tables.len(),
            new_table_count = new_tables.len(),
            "Comparing table sets"
        );

        // Find new tables
        for table in new_tables.difference(&old_tables) {
            tracing::trace!(table = %table, "Found new table");
            changes.new_tables.push(table.clone());
        }

        // Find removed tables
        for table in old_tables.difference(&new_tables) {
            tracing::trace!(table = %table, "Found removed table");
            changes.removed_tables.push(table.clone());
        }

        // Find modified tables
        for table in old_tables.intersection(&new_tables) {
            tracing::trace!(table = %table, "Comparing table");
            let (old_table, new_table) = match (old.tables.get(table), new.tables.get(table)) {
                (Some(o), Some(n)) => (o, n),
                _ => {
                    let msg =
                        format!("Table '{table}' present in key set but missing in schema maps");
                    tracing::error!(table = %table, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };
            if let Some(table_changes) = Self::compare_tables(table, old_table, new_table)? {
                tracing::trace!(
                    table = %table,
                    new_fields = table_changes.new_fields.len(),
                    removed_fields = table_changes.removed_fields.len(),
                    modified_fields = table_changes.modified_fields.len(),
                    "Table has changes"
                );
                changes.modified_tables.push(table_changes);
            }
        }

        // Also compare edges
        let old_edges: BTreeSet<String> = old.edges.keys().cloned().collect();
        let new_edges: BTreeSet<String> = new.edges.keys().cloned().collect();

        for edge in new_edges.difference(&old_edges) {
            changes.new_tables.push(edge.clone());
        }

        for edge in old_edges.difference(&new_edges) {
            changes.removed_tables.push(edge.clone());
        }

        for edge in old_edges.intersection(&new_edges) {
            let (old_edge, new_edge) = match (old.edges.get(edge), new.edges.get(edge)) {
                (Some(o), Some(n)) => (o, n),
                _ => {
                    let msg =
                        format!("Edge '{edge}' present in key set but missing in schema maps");
                    tracing::error!(edge = %edge, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };
            if let Some(edge_changes) = Self::compare_tables(edge, old_edge, new_edge)? {
                changes.modified_tables.push(edge_changes);
            }
        }

        // Compare accesses
        let old_access_names: BTreeSet<String> =
            old.accesses.iter().map(|a| a.name.clone()).collect();
        let new_access_names: BTreeSet<String> =
            new.accesses.iter().map(|a| a.name.clone()).collect();

        // Find new accesses
        for access_name in new_access_names.difference(&old_access_names) {
            changes.new_accesses.push(access_name.clone());
        }

        // Find removed accesses
        for access_name in old_access_names.difference(&new_access_names) {
            changes.removed_accesses.push(access_name.clone());
        }

        // Find modified accesses
        for access_name in old_access_names.intersection(&new_access_names) {
            let old_access = match old.accesses.iter().find(|a| &a.name == access_name) {
                Some(a) => a,
                None => {
                    let msg = format!(
                        "Access '{access_name}' present in key set but missing in old schema list"
                    );
                    tracing::error!(access = %access_name, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };
            let new_access = match new.accesses.iter().find(|a| &a.name == access_name) {
                Some(a) => a,
                None => {
                    let msg = format!(
                        "Access '{access_name}' present in key set but missing in new schema list"
                    );
                    tracing::error!(access = %access_name, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };

            if let Some(access_change) = Self::compare_accesses(old_access, new_access) {
                changes.modified_accesses.push(access_change);
            }
        }

        // Compare analyzers by name and normalized statement
        let old_analyzers: BTreeMap<&str, &AnalyzerDefinition> =
            old.analyzers.iter().map(|a| (a.name.as_str(), a)).collect();
        let new_analyzer_names: BTreeSet<&str> =
            new.analyzers.iter().map(|a| a.name.as_str()).collect();

        for analyzer in &new.analyzers {
            match old_analyzers.get(analyzer.name.as_str()) {
                None => changes.new_analyzers.push(analyzer.name.clone()),
                Some(old_analyzer) if old_analyzer.statement != analyzer.statement => {
                    changes.modified_analyzers.push(analyzer.name.clone());
                }
                Some(_) => {}
            }
        }

        for analyzer in &old.analyzers {
            if !new_analyzer_names.contains(analyzer.name.as_str()) {
                changes.removed_analyzers.push(analyzer.name.clone());
            }
        }

        Self::flag_indexes_affected_by_analyzer_changes(old, new, &mut changes);

        tracing::debug!(
            new_tables = changes.new_tables.len(),
            removed_tables = changes.removed_tables.len(),
            modified_tables = changes.modified_tables.len(),
            new_accesses = changes.new_accesses.len(),
            removed_accesses = changes.removed_accesses.len(),
            modified_accesses = changes.modified_accesses.len(),
            new_analyzers = changes.new_analyzers.len(),
            removed_analyzers = changes.removed_analyzers.len(),
            modified_analyzers = changes.modified_analyzers.len(),
            "Schema comparison complete"
        );

        Ok(changes)
    }

    /// Flag FULLTEXT indexes that must be rebuilt because of analyzer changes.
    ///
    /// - A FULLTEXT index keeps the terms produced by the analyzer it was built
    ///   with. `DEFINE ANALYZER OVERWRITE` does not reindex, and until the index
    ///   is re-defined its searches silently stop matching. Indexes using a
    ///   modified analyzer are therefore marked modified, which makes
    ///   `define_tables` re-apply them.
    /// - SurrealDB refuses `REMOVE ANALYZER` while an index references it. An
    ///   index moving off a removed analyzer is therefore also listed as
    ///   removed, so it is dropped before the analyzer and recreated with its
    ///   new definition afterwards.
    fn flag_indexes_affected_by_analyzer_changes(
        old: &SchemaDefinition,
        new: &SchemaDefinition,
        changes: &mut SchemaChanges,
    ) {
        if changes.modified_analyzers.is_empty() && changes.removed_analyzers.is_empty() {
            return;
        }

        for (table_name, new_table) in new.tables.iter().chain(new.edges.iter()) {
            let Some(old_table) = old.tables.get(table_name).or(old.edges.get(table_name)) else {
                // New tables are defined in full anyway
                continue;
            };

            for index in &new_table.indexes {
                let Some(old_index) = old_table.indexes.iter().find(|i| i.name == index.name)
                else {
                    continue;
                };

                let uses_modified_analyzer = fulltext_analyzer(&index.definition)
                    .is_some_and(|a| changes.modified_analyzers.iter().any(|m| m == a));
                let leaves_removed_analyzer = fulltext_analyzer(&old_index.definition)
                    .is_some_and(|a| changes.removed_analyzers.iter().any(|r| r == a));
                if !uses_modified_analyzer && !leaves_removed_analyzer {
                    continue;
                }

                let position = changes
                    .modified_tables
                    .iter()
                    .position(|t| &t.table_name == table_name);
                let table_changes = match position {
                    Some(position) => &mut changes.modified_tables[position],
                    None => {
                        changes.modified_tables.push(TableChanges {
                            table_name: table_name.clone(),
                            new_fields: Vec::new(),
                            removed_fields: Vec::new(),
                            modified_fields: Vec::new(),
                            permission_changed: false,
                            schema_type_changed: false,
                            new_events: Vec::new(),
                            removed_events: Vec::new(),
                            new_indexes: Vec::new(),
                            removed_indexes: Vec::new(),
                            modified_indexes: Vec::new(),
                        });
                        changes.modified_tables.last_mut().unwrap()
                    }
                };

                if leaves_removed_analyzer
                    && !table_changes
                        .removed_indexes
                        .iter()
                        .any(|i| i.name == old_index.name)
                {
                    table_changes.removed_indexes.push(old_index.clone());
                }
                if !table_changes
                    .modified_indexes
                    .iter()
                    .any(|i| i.name == index.name)
                {
                    table_changes.modified_indexes.push(index.clone());
                }
            }
        }
    }

    /// Compare two table definitions
    fn compare_tables(
        table_name: &str,
        old_table: &TableDefinition,
        new_table: &TableDefinition,
    ) -> Result<Option<TableChanges>> {
        let mut table_changes = TableChanges {
            table_name: table_name.to_string(),
            new_fields: Vec::new(),
            removed_fields: Vec::new(),
            modified_fields: Vec::new(),
            permission_changed: false,
            schema_type_changed: false,
            new_events: Vec::new(),
            removed_events: Vec::new(),
            new_indexes: Vec::new(),
            removed_indexes: Vec::new(),
            modified_indexes: Vec::new(),
        };

        // Check schema type change
        if old_table.schema_type != new_table.schema_type {
            table_changes.schema_type_changed = true;
        }

        // Check permission changes
        if old_table.permissions != new_table.permissions {
            table_changes.permission_changed = true;
        }

        // Compare regular fields
        let old_fields: BTreeSet<String> = old_table.fields.keys().cloned().collect();
        let new_fields: BTreeSet<String> = new_table.fields.keys().cloned().collect();

        // Find new fields
        for field in new_fields.difference(&old_fields) {
            table_changes.new_fields.push(field.clone());
        }

        // Find removed fields
        for field in old_fields.difference(&new_fields) {
            table_changes.removed_fields.push(field.clone());
        }

        // Find modified fields
        for field in old_fields.intersection(&new_fields) {
            let old_field = match old_table.fields.get(field) {
                Some(f) => f,
                None => {
                    let msg = format!(
                        "Field '{field}' present in old/new intersection but missing in old_table '{}'",
                        table_name
                    );
                    tracing::error!(table = %table_name, field = %field, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };
            let new_field = match new_table.fields.get(field) {
                Some(f) => f,
                None => {
                    let msg = format!(
                        "Field '{field}' present in old/new intersection but missing in new_table '{}'",
                        table_name
                    );
                    tracing::error!(table = %table_name, field = %field, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };

            // If the field types differ, record a modification without deep recursion
            if old_field.field_type != new_field.field_type {
                // Mark entire field as modified; avoid recursive descent to prevent stack overflows
                table_changes.modified_fields.push(FieldChange {
                    field_name: field.to_string(),
                    old_type: old_field.field_type.to_string(),
                    new_type: new_field.field_type.to_string(),
                    change_type: ChangeType::Modified,
                    required_changed: false,
                    default_changed: false,
                });
            } else {
                // Types are the same, check for other changes (required, default)
                if let Some(field_change) = Self::compare_fields(field, old_field, new_field) {
                    table_changes.modified_fields.push(field_change);
                }
            }
        }

        // Compare array wildcard fields
        let old_wildcard_fields: BTreeSet<String> =
            old_table.array_wildcard_fields.keys().cloned().collect();
        let new_wildcard_fields: BTreeSet<String> =
            new_table.array_wildcard_fields.keys().cloned().collect();

        // Find new wildcard fields (these represent new fields)
        for field in new_wildcard_fields.difference(&old_wildcard_fields) {
            table_changes.new_fields.push(format!("{}[*]", field));
        }

        // Find removed wildcard fields
        for field in old_wildcard_fields.difference(&new_wildcard_fields) {
            table_changes.removed_fields.push(format!("{}[*]", field));
        }

        // Find modified wildcard fields
        for field in old_wildcard_fields.intersection(&new_wildcard_fields) {
            let old_wild = match old_table.array_wildcard_fields.get(field) {
                Some(f) => f,
                None => {
                    let msg = format!(
                        "Wildcard field '{field}[*]' present in intersection but missing in old_table '{}'",
                        table_name
                    );
                    tracing::error!(table = %table_name, field = %field, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };
            let new_wild = match new_table.array_wildcard_fields.get(field) {
                Some(f) => f,
                None => {
                    let msg = format!(
                        "Wildcard field '{field}[*]' present in intersection but missing in new_table '{}'",
                        table_name
                    );
                    tracing::error!(table = %table_name, field = %field, "{}", msg);
                    return Err(EvenframeError::comparison(msg));
                }
            };
            if let Some(field_change) =
                Self::compare_fields(&format!("{}[*]", field), old_wild, new_wild)
            {
                table_changes.modified_fields.push(field_change);
            }
        }

        // Compare events by *normalized* statement contents. SurrealDB's
        // schema export round-trips `DEFINE EVENT` with `ON <table>`,
        // single-quoted strings, no `OVERWRITE`, and no inner trailing
        // semicolons; emit-side statements typically include
        // `OVERWRITE`, `ON TABLE <table>`, double-quoted strings, and
        // trailing semicolons. A byte-exact compare classifies every
        // semantically-identical event as both removed and added,
        // which produces correct but noisy REMOVE+DEFINE churn each
        // sync. Normalizing both sides makes equality reflect actual
        // intent and keeps idempotent syncs idempotent.
        let mut old_normalized: BTreeMap<String, String> = BTreeMap::new();
        for stmt in &old_table.events {
            old_normalized.insert(normalize_event_statement(stmt), stmt.clone());
        }
        let mut new_normalized: BTreeMap<String, String> = BTreeMap::new();
        for stmt in &new_table.events {
            new_normalized.insert(normalize_event_statement(stmt), stmt.clone());
        }

        for (key, stmt) in &new_normalized {
            if !old_normalized.contains_key(key) {
                table_changes.new_events.push(stmt.clone());
            }
        }

        for (key, stmt) in &old_normalized {
            if !new_normalized.contains_key(key) {
                table_changes.removed_events.push(stmt.clone());
            }
        }

        // Compare indexes by name. An index present in the DB but absent from
        // code is an orphan and must be dropped. A same-name index whose
        // columns or definition (kind, parameters, comment) differ is
        // modified; it must be flagged so the table is revisited and its
        // DEFINE INDEX OVERWRITE statements are re-applied.
        let old_indexes: BTreeMap<&str, &IndexDefinition> = old_table
            .indexes
            .iter()
            .map(|i| (i.name.as_str(), i))
            .collect();
        let new_index_names: BTreeSet<&str> =
            new_table.indexes.iter().map(|i| i.name.as_str()).collect();

        for index in &new_table.indexes {
            match old_indexes.get(index.name.as_str()) {
                None => table_changes.new_indexes.push(index.clone()),
                Some(old_index)
                    if old_index.columns != index.columns
                        || old_index.definition != index.definition =>
                {
                    table_changes.modified_indexes.push(index.clone());
                }
                Some(_) => {}
            }
        }

        for index in &old_table.indexes {
            if !new_index_names.contains(index.name.as_str()) {
                table_changes.removed_indexes.push(index.clone());
            }
        }

        // Return None if no changes detected
        if table_changes.new_fields.is_empty()
            && table_changes.removed_fields.is_empty()
            && table_changes.modified_fields.is_empty()
            && !table_changes.permission_changed
            && !table_changes.schema_type_changed
            && table_changes.new_events.is_empty()
            && table_changes.removed_events.is_empty()
            && table_changes.new_indexes.is_empty()
            && table_changes.removed_indexes.is_empty()
            && table_changes.modified_indexes.is_empty()
        {
            Ok(None)
        } else {
            Ok(Some(table_changes))
        }
    }

    /// Compare two field definitions
    fn compare_fields(
        field_name: &str,
        old_field: &FieldDefinition,
        new_field: &FieldDefinition,
    ) -> Option<FieldChange> {
        let mut changed = false;

        // Check for basic changes first
        let mut basic_change = FieldChange {
            field_name: field_name.to_string(),
            old_type: old_field.field_type.to_string(),
            new_type: new_field.field_type.to_string(),
            change_type: ChangeType::Modified,
            required_changed: false,
            default_changed: false,
        };

        // Check required change
        if old_field.required != new_field.required {
            basic_change.required_changed = true;
            changed = true;
        }

        // Check default value change
        if old_field.default_value != new_field.default_value {
            basic_change.default_changed = true;
            changed = true;
        }

        // Check type change
        if old_field.field_type != new_field.field_type {
            changed = true;
        }

        // Check assertion change (validator-derived or manual ASSERT clauses).
        // Normalized so cosmetic reformatting by SurrealDB doesn't cause churn.
        if normalize_assert(&old_field.assertions) != normalize_assert(&new_field.assertions) {
            changed = true;
        }

        if changed { Some(basic_change) } else { None }
    }

    /// Deep comparison of object types to find granular changes
    pub fn compare_object_types(
        prefix: &str,
        old_type: &ObjectType,
        new_type: &ObjectType,
    ) -> Vec<FieldChange> {
        let mut changes = Vec::new();

        match (old_type, new_type) {
            // Both are objects - compare fields
            (ObjectType::Object(old_fields), ObjectType::Object(new_fields)) => {
                let old_keys: BTreeSet<String> = old_fields.keys().cloned().collect();
                let new_keys: BTreeSet<String> = new_fields.keys().cloned().collect();

                // Find added fields
                for key in new_keys.difference(&old_keys) {
                    let field_path = if prefix.is_empty() {
                        key.clone()
                    } else {
                        format!("{}.{}", prefix, key)
                    };

                    changes.push(FieldChange {
                        field_name: field_path,
                        old_type: String::new(),
                        new_type: new_fields[key].to_string(),
                        change_type: ChangeType::Added,
                        required_changed: false,
                        default_changed: false,
                    });
                }

                // Find removed fields
                for key in old_keys.difference(&new_keys) {
                    let field_path = if prefix.is_empty() {
                        key.clone()
                    } else {
                        format!("{}.{}", prefix, key)
                    };

                    changes.push(FieldChange {
                        field_name: field_path,
                        old_type: old_fields[key].to_string(),
                        new_type: String::new(),
                        change_type: ChangeType::Removed,
                        required_changed: false,
                        default_changed: false,
                    });
                }

                // Compare common fields
                for key in old_keys.intersection(&new_keys) {
                    let field_path = if prefix.is_empty() {
                        key.clone()
                    } else {
                        format!("{}.{}", prefix, key)
                    };

                    let old_field_type = &old_fields[key];
                    let new_field_type = &new_fields[key];

                    if old_field_type != new_field_type {
                        // Recursively compare nested types
                        let nested_changes =
                            Self::compare_object_types(&field_path, old_field_type, new_field_type);
                        if nested_changes.is_empty() {
                            // If no nested changes, the types themselves are different
                            changes.push(FieldChange {
                                field_name: field_path,
                                old_type: old_field_type.to_string(),
                                new_type: new_field_type.to_string(),
                                change_type: ChangeType::Modified,
                                required_changed: false,
                                default_changed: false,
                            });
                        } else {
                            changes.extend(nested_changes);
                        }
                    }
                }
            }
            // Nullable types - unwrap and compare
            (ObjectType::Nullable(old_inner), ObjectType::Nullable(new_inner)) => {
                changes.extend(Self::compare_object_types(prefix, old_inner, new_inner));
            }
            // Different types entirely
            _ => {
                if old_type != new_type {
                    changes.push(FieldChange {
                        field_name: prefix.to_string(),
                        old_type: old_type.to_string(),
                        new_type: new_type.to_string(),
                        change_type: ChangeType::Modified,
                        required_changed: false,
                        default_changed: false,
                    });
                }
            }
        }

        changes
    }

    /// Compare two access definitions
    fn compare_accesses(
        old_access: &AccessDefinition,
        new_access: &AccessDefinition,
    ) -> Option<AccessChange> {
        let mut changes = Vec::new();

        // Compare access type
        if old_access.access_type != new_access.access_type {
            changes.push(AccessChangeType::OtherChange(format!(
                "Access type changed from {:?} to {:?}",
                old_access.access_type, new_access.access_type
            )));
        }

        // Compare database level
        if old_access.database_level != new_access.database_level {
            let old_level = if old_access.database_level {
                "DATABASE"
            } else {
                "NAMESPACE"
            };
            let new_level = if new_access.database_level {
                "DATABASE"
            } else {
                "NAMESPACE"
            };
            changes.push(AccessChangeType::OtherChange(format!(
                "Access level changed from {} to {}",
                old_level, new_level
            )));
        }

        // Compare signup query
        if old_access.signup_query != new_access.signup_query {
            changes.push(AccessChangeType::SignupChanged);
        }

        // Compare signin query
        if old_access.signin_query != new_access.signin_query {
            changes.push(AccessChangeType::SigninChanged);
        }

        // Compare JWT configuration
        if old_access.jwt_algorithm != new_access.jwt_algorithm {
            changes.push(AccessChangeType::OtherChange(format!(
                "JWT algorithm changed from {:?} to {:?}",
                old_access.jwt_algorithm, new_access.jwt_algorithm
            )));
        }

        if old_access.jwt_key != new_access.jwt_key {
            changes.push(AccessChangeType::JwtKeyChanged);
        }

        if old_access.jwt_url != new_access.jwt_url {
            changes.push(AccessChangeType::JwtUrlChanged);
        }

        if old_access.issuer_key != new_access.issuer_key {
            changes.push(AccessChangeType::IssuerKeyChanged);
        }

        // Compare authenticate clause
        if old_access.authenticate != new_access.authenticate {
            changes.push(AccessChangeType::AuthenticateClauseChanged);
        }

        // Compare durations
        if old_access.duration_for_token != new_access.duration_for_token {
            changes.push(AccessChangeType::DurationChanged);
        }

        if old_access.duration_for_session != new_access.duration_for_session {
            changes.push(AccessChangeType::DurationChanged);
        }

        // Compare bearer configuration
        if old_access.bearer_for != new_access.bearer_for {
            changes.push(AccessChangeType::OtherChange(format!(
                "Bearer FOR changed from {:?} to {:?}",
                old_access.bearer_for, new_access.bearer_for
            )));
        }

        if changes.is_empty() {
            None
        } else {
            Some(AccessChange {
                access_name: old_access.name.clone(),
                changes,
            })
        }
    }
}

// ============================================================================
// Helper Functions
// ============================================================================

/// Canonicalize a `DEFINE EVENT` statement so two semantically equivalent
/// statements compare equal regardless of cosmetic differences.
///
/// Cosmetic differences observed in practice:
/// - SurrealDB's schema export drops `OVERWRITE` / `IF NOT EXISTS`
///   modifiers and the optional `TABLE` keyword in `ON TABLE <name>`,
///   while emit-side typically includes them.
/// - SurrealDB's export round-trips quoted strings using `'...'`, while
///   emit-side often uses `"..."` (both are equivalent in SurrealQL).
/// - Emit-side adds inner `;` between statements in the `THEN { ... }`
///   block; export-side strips them.
/// - Whitespace, including line breaks, varies freely.
///
/// The walk preserves string-literal contents verbatim (no quote
/// flipping inside a literal), then normalizes the surface syntax of
/// the statement after the walk.
fn normalize_event_statement(stmt: &str) -> String {
    let trimmed = stmt.trim();
    let bytes = trimmed.as_bytes();
    let mut out = String::with_capacity(trimmed.len());
    let mut quote: Option<u8> = None;
    let mut i = 0;
    while i < bytes.len() {
        let c = bytes[i];
        match quote {
            Some(q) => {
                // Inside a string literal — pass through, but emit a
                // canonical `'` for the closing quote so `"foo"` and
                // `'foo'` normalize to the same form.
                if c == q {
                    out.push('\'');
                    quote = None;
                } else {
                    out.push(c as char);
                }
                i += 1;
            }
            None => {
                if c == b'"' || c == b'\'' {
                    quote = Some(c);
                    out.push('\'');
                    i += 1;
                } else if c == b';' {
                    // Drop semicolons outside string literals — they're
                    // always optional terminators in SurrealQL events.
                    i += 1;
                } else if c.is_ascii_whitespace() {
                    if !out.is_empty() && !out.ends_with(' ') {
                        out.push(' ');
                    }
                    i += 1;
                } else {
                    out.push(c as char);
                    i += 1;
                }
            }
        }
    }

    // Canonicalize the leading modifiers and the `ON [TABLE]` clause.
    let mut canonical = out.trim().to_string();
    canonical = canonical.replacen("DEFINE EVENT OVERWRITE ", "DEFINE EVENT ", 1);
    canonical = canonical.replacen("DEFINE EVENT IF NOT EXISTS ", "DEFINE EVENT ", 1);
    canonical = canonical.replace(" ON TABLE ", " ON ");
    canonical
}

/// Normalize an ASSERT expression for churn-free comparison between the
/// locally-generated schema and the form SurrealDB stores and returns.
///
/// SurrealDB rewrites assertions on store: it drops redundant parentheses
/// (e.g. `$value = NULL OR (string::len($value) <= 5)` becomes
/// `$value = NULL OR string::len($value) <= 5`) and canonicalizes string
/// literals to single quotes (`"@"` becomes `'@'`). To avoid re-emitting an
/// unchanged field every sync, this neutralizes those differences: it joins
/// multiple clauses with ` AND `, drops `(`/`)`, maps `"`/`` ` `` to `'`,
/// collapses whitespace runs to a single space, and trims a trailing `;`.
/// Empty input normalizes to the empty string, so unasserted fields never
/// report a change.
fn normalize_assert(assertions: &[String]) -> String {
    let joined = assertions.join(" AND ");
    let mut out = String::with_capacity(joined.len());
    let mut prev_space = false;
    for c in joined.chars() {
        match c {
            '(' | ')' => {}
            '"' | '`' => {
                out.push('\'');
                prev_space = false;
            }
            c if c.is_whitespace() => {
                if !out.is_empty() && !prev_space {
                    out.push(' ');
                    prev_space = true;
                }
            }
            c => {
                out.push(c);
                prev_space = false;
            }
        }
    }
    out.trim().trim_end_matches(';').trim().to_string()
}

#[cfg(test)]
mod normalize_assert_tests {
    use super::normalize_assert;

    #[test]
    fn paren_and_quote_insensitive() {
        // Desired (as emitted) vs stored (as SurrealDB returns it) must match.
        let desired = vec!["$value = NULL OR (string::len($value) <= 5)".to_string()];
        let stored = vec!["$value = NULL OR string::len($value) <= 5".to_string()];
        assert_eq!(normalize_assert(&desired), normalize_assert(&stored));

        let desired = vec!["string::starts_with($value, \"@\")".to_string()];
        let stored = vec!["string::starts_with($value, '@')".to_string()];
        assert_eq!(normalize_assert(&desired), normalize_assert(&stored));
    }

    #[test]
    fn empty_is_empty_and_whitespace_collapses() {
        assert_eq!(normalize_assert(&[]), "");
        assert_eq!(
            normalize_assert(&["  string::len($value)   >  0 ;".to_string()]),
            "string::len$value > 0"
        );
    }

    #[test]
    fn genuinely_different_asserts_differ() {
        let a = vec!["string::len($value) <= 5".to_string()];
        let b = vec!["string::len($value) <= 6".to_string()];
        assert_ne!(normalize_assert(&a), normalize_assert(&b));
    }
}

#[cfg(test)]
mod normalize_event_statement_tests {
    use super::normalize_event_statement;

    #[test]
    fn export_and_emit_forms_match() {
        let exported = "DEFINE EVENT track_activity ON user_settings WHEN $event = 'CREATE' OR $event = 'UPDATE' THEN { IF $auth != NONE { fn::track_activity($before, $after, $event) } }";
        let emitted = "DEFINE EVENT OVERWRITE track_activity ON TABLE user_settings WHEN $event = \"CREATE\" OR $event = \"UPDATE\" THEN { IF $auth != NONE { fn::track_activity($before, $after, $event); }; };";
        assert_eq!(
            normalize_event_statement(exported),
            normalize_event_statement(emitted)
        );
    }

    #[test]
    fn whitespace_collapses() {
        let a = "DEFINE EVENT foo  ON   bar  WHEN $event = 'CREATE'\n  THEN  {  }";
        let b = "DEFINE EVENT foo ON bar WHEN $event = 'CREATE' THEN { }";
        assert_eq!(normalize_event_statement(a), normalize_event_statement(b));
    }

    #[test]
    fn string_contents_preserved() {
        // Whitespace and `;` inside a string literal must stay verbatim.
        let stmt = "DEFINE EVENT foo ON bar WHEN $name = 'hello;  world' THEN { }";
        assert_eq!(
            normalize_event_statement(stmt),
            "DEFINE EVENT foo ON bar WHEN $name = 'hello;  world' THEN { }"
        );
    }

    #[test]
    fn distinct_statements_stay_distinct() {
        let a = "DEFINE EVENT foo ON bar WHEN $event = 'CREATE' THEN { }";
        let b = "DEFINE EVENT foo ON bar WHEN $event = 'UPDATE' THEN { }";
        assert_ne!(normalize_event_statement(a), normalize_event_statement(b));
    }
}

/// Helper function to collect object type names referenced in a field type
pub fn collect_referenced_objects(
    field_type: &FieldType,
    objects_to_process: &mut Vec<String>,
    enums: &BTreeMap<String, TaggedUnion>,
) {
    match field_type {
        FieldType::Other(type_name) => {
            // Check if this is an enum
            if let Some(enum_def) = enums.get(type_name) {
                // For enums, we need to collect all variant data types
                for variant in &enum_def.variants {
                    if let Some(variant_data) = &variant.data {
                        match variant_data {
                            VariantData::InlineStruct(enum_struct) => {
                                objects_to_process.push(enum_struct.struct_name.clone())
                            }
                            VariantData::DataStructureRef(referenced_field_type) => {
                                if let FieldType::Other(data) = referenced_field_type {
                                    objects_to_process.push(data.clone());
                                }
                            }
                        }
                    }
                }
            } else {
                // Not an enum, just a regular object/struct
                objects_to_process.push(type_name.clone());
            }
        }
        FieldType::Option(inner) | FieldType::Vec(inner) | FieldType::RecordLink(inner) => {
            collect_referenced_objects(inner, objects_to_process, enums);
        }
        FieldType::Tuple(types) => {
            for t in types {
                collect_referenced_objects(t, objects_to_process, enums);
            }
        }
        FieldType::Struct(fields) => {
            for (_, field_type) in fields {
                collect_referenced_objects(field_type, objects_to_process, enums);
            }
        }
        FieldType::HashMap(key_type, value_type) | FieldType::BTreeMap(key_type, value_type) => {
            collect_referenced_objects(key_type, objects_to_process, enums);
            collect_referenced_objects(value_type, objects_to_process, enums);
        }
        _ => {}
    }
}

// ============================================================================
// Merger - Legacy struct for backward compatibility
// ============================================================================

/// Main entry point for Schemasync Merge functionality
#[cfg(feature = "surrealdb")]
pub struct Merger<'a> {
    pub client: &'a Surreal<Client>,
    pub default_mock_gen_config: SchemasyncMockGenConfig,
    pub performance: PerformanceConfig,
}

#[cfg(feature = "surrealdb")]
impl<'a> Merger<'a> {
    /// Create a new Merger instance
    pub async fn new(
        client: &'a Surreal<Client>,
        default_mock_gen_config: SchemasyncMockGenConfig,
        performance: PerformanceConfig,
    ) -> Result<Self> {
        Ok(Self {
            client,
            default_mock_gen_config,
            performance,
        })
    }

    /// Import schema from production database
    pub async fn import_schema_from_db(&self) -> Result<SchemaDefinition> {
        tracing::debug!("Importing schema from production database");
        let importer = SchemaImporter::new(self.client);
        let schema = importer.import_schema_only().await?;
        tracing::debug!(
            tables = schema.tables.len(),
            edges = schema.edges.len(),
            accesses = schema.accesses.len(),
            "Schema imported"
        );
        Ok(schema)
    }

    /// Generate schema from Rust structs
    pub fn generate_schema_from_structs(
        &self,
        tables: &BTreeMap<String, TableConfig>,
    ) -> Result<SchemaDefinition> {
        tracing::debug!(
            table_count = tables.len(),
            "Generating schema from Rust structs"
        );
        let schema = SchemaDefinition::from_table_configs(
            tables,
            self.default_mock_gen_config.scripting_asserts,
        )?;
        tracing::debug!(
            tables = schema.tables.len(),
            edges = schema.edges.len(),
            "Schema generated from structs"
        );
        Ok(schema)
    }

    pub fn compare_schemas(
        &self,
        old: &SchemaDefinition,
        new: &SchemaDefinition,
    ) -> Result<SchemaChanges> {
        tracing::debug!("Comparing schemas using legacy method");
        Comparator::compare(old, new)
    }

    /// Export mock data to file
    pub async fn export_mock_data(&self, _file_path: &str) -> Result<()> {
        // Implementation will use the generated statements
        // and write them to the specified file
        todo!("Implement export_mock_data")
    }

    /// Generate preserved data for a specific table
    pub async fn generate_preserved_data(
        &self,
        table_name: &str,
        table_config: &TableConfig,
        mock_config: MockGenerationConfig,
        existing_records: Vec<serde_json::Value>,
        target_count: usize,
        schema_changes: Option<&SchemaChanges>,
    ) -> Vec<serde_json::Value> {
        use serde_json::Value;

        // Determine how many records to preserve vs generate
        let existing_count = existing_records.len();
        let mut result = Vec::new();

        match mock_config.preservation_mode {
            PreservationMode::None => {
                // No preservation - generate all new data
                result = self.generate_new_records(table_name, table_config, target_count);
            }
            PreservationMode::Smart => {
                // Smart preservation - keep unchanged fields, regenerate specified fields
                if existing_count > 0 {
                    // Determine which fields need regeneration
                    let mut fields_to_regenerate = mock_config.regenerate_fields.clone();

                    // If schema changes are provided, add fields that changed
                    if let Some(changes) = schema_changes {
                        // Get fields that need regeneration based on schema changes
                        let schema_fields_needing_generation =
                            changes.get_fields_needing_generation(table_name);

                        // If all fields need generation (new table), regenerate everything
                        if schema_fields_needing_generation.contains(&"*".to_string()) {
                            // Generate all new records for new tables
                            result =
                                self.generate_new_records(table_name, table_config, target_count);
                            return result;
                        }

                        // Add schema-detected fields to the regeneration list
                        for field in schema_fields_needing_generation {
                            if !fields_to_regenerate.contains(&field) {
                                fields_to_regenerate.push(field);
                            }
                        }
                    }

                    for mut record in existing_records {
                        // Regenerate specified fields
                        if let Value::Object(ref mut map) = record {
                            // First, add any new fields that don't exist in the record
                            for field in &table_config.struct_config.fields {
                                if !map.contains_key(&field.field_name) {
                                    // This is a new field, generate value
                                    let new_value = Self::generate_field_value(field, table_config);
                                    map.insert(field.field_name.clone(), new_value);
                                }
                            }

                            // Then, regenerate fields that need it
                            for field_name in &fields_to_regenerate {
                                if let Some(field) = table_config
                                    .struct_config
                                    .fields
                                    .iter()
                                    .find(|f| &f.field_name == field_name)
                                {
                                    // Generate new value for this field
                                    let new_value = Self::generate_field_value(field, table_config);
                                    map.insert(field_name.clone(), new_value);
                                }
                            }
                        }

                        result.push(record);
                    }

                    // Generate additional records if needed
                    if target_count > existing_count {
                        let additional = self.generate_new_records(
                            table_name,
                            table_config,
                            target_count - existing_count,
                        );
                        result.extend(additional);
                    }
                } else {
                    // No existing data or preservation disabled
                    result = self.generate_new_records(table_name, table_config, target_count);
                }
            }
            PreservationMode::Full => {
                // Full preservation - keep all data, only add new fields
                if existing_count > 0 {
                    // Check if target count is less than existing count
                    if target_count < existing_count {
                        eprintln!(
                            "\n  WARNING: Full preservation mode with data reduction detected!"
                        );
                        eprintln!(
                            "   Table '{}' has {} existing records but target count is set to {}",
                            table_name, existing_count, target_count
                        );
                        eprintln!(
                            "   This will DELETE {} records!",
                            existing_count - target_count
                        );
                        eprintln!("\n   Options:");
                        eprintln!(
                            "   1. Change the target count (n) to {} or higher to preserve all records",
                            existing_count
                        );
                        eprintln!("   2. Use Smart preservation mode instead of Full");
                        eprintln!(
                            "   3. Set preservation_mode to None if you want to regenerate all data"
                        );
                        eprintln!(
                            "\n   In a production environment, this would require user confirmation."
                        );
                        eprintln!(
                            "   For now, proceeding with target count of {} records.\n",
                            target_count
                        );

                        for mut record in existing_records {
                            // Only add new fields that don't exist
                            if let Value::Object(ref mut map) = record {
                                for field in &table_config.struct_config.fields {
                                    if !map.contains_key(&field.field_name) {
                                        // This is a new field, generate value
                                        let new_value =
                                            Self::generate_field_value(field, table_config);
                                        map.insert(field.field_name.clone(), new_value);
                                    }
                                }
                            }

                            result.push(record);
                        }
                    } else {
                        // Normal case: preserve all existing records
                        for mut record in existing_records {
                            // Only add new fields that don't exist
                            if let Value::Object(ref mut map) = record {
                                for field in &table_config.struct_config.fields {
                                    if !map.contains_key(&field.field_name) {
                                        // This is a new field, generate value
                                        let new_value =
                                            Self::generate_field_value(field, table_config);
                                        map.insert(field.field_name.clone(), new_value);
                                    }
                                }
                            }

                            result.push(record);
                        }

                        // Generate additional records if needed
                        if target_count > existing_count {
                            let additional = self.generate_new_records(
                                table_name,
                                table_config,
                                target_count - existing_count,
                            );
                            result.extend(additional);
                        }
                    }
                } else {
                    result = self.generate_new_records(table_name, table_config, target_count);
                }
            }
        }

        result
    }

    /// Generate new records for a table
    fn generate_new_records(
        &self,
        _table_name: &str,
        table_config: &TableConfig,
        count: usize,
    ) -> Vec<serde_json::Value> {
        use serde_json::Value;

        let mut records = Vec::new();

        for _ in 0..count {
            let mut record = serde_json::Map::new();

            // Generate values for each field
            for field in &table_config.struct_config.fields {
                let value = Self::generate_field_value(field, table_config);
                record.insert(field.field_name.clone(), value);
            }

            records.push(Value::Object(record));
        }

        records
    }

    /// Generate a value for a specific field
    fn generate_field_value(
        field: &crate::types::StructField,
        _table_config: &TableConfig,
    ) -> serde_json::Value {
        use crate::types::FieldType;
        use serde_json::json;

        // Use format if available
        if let Some(format) = &field.format {
            let value = format.generate_formatted_value();

            // Check if the format generates numeric values
            match format {
                crate::schemasync::mockmake::format::Format::Percentage
                | crate::schemasync::mockmake::format::Format::Latitude
                | crate::schemasync::mockmake::format::Format::Longitude
                | crate::schemasync::mockmake::format::Format::CurrencyAmount => {
                    // Try to parse as number
                    if let Ok(num) = value.parse::<f64>() {
                        return json!(num);
                    }
                }
                _ => {}
            }

            return json!(value);
        }

        // Generate based on field type
        match &field.field_type {
            FieldType::String => json!(crate::schemasync::Mockmaker::random_string(8)),
            FieldType::Bool => json!(rand::random::<bool>()),
            FieldType::U8
            | FieldType::U16
            | FieldType::U32
            | FieldType::U64
            | FieldType::U128
            | FieldType::Usize => json!(rand::random::<u32>() % 100),
            FieldType::I8
            | FieldType::I16
            | FieldType::I32
            | FieldType::I64
            | FieldType::I128
            | FieldType::Isize => json!(rand::random::<i32>() % 100),
            FieldType::F32 | FieldType::F64 => json!(rand::random::<f64>() * 100.0),
            FieldType::Option(inner) => {
                if rand::random::<bool>() {
                    let inner_field = crate::types::StructField {
                        field_name: field.field_name.clone(),
                        field_type: *inner.clone(),
                        format: field.format.clone(),
                        ..Default::default()
                    };
                    Self::generate_field_value(&inner_field, _table_config)
                } else {
                    json!(null)
                }
            }
            FieldType::Vec(_) => json!([]),
            FieldType::Other(type_name) => {
                // Handle common types
                if type_name.contains("DateTime") {
                    json!(chrono::Utc::now().to_rfc3339())
                } else {
                    json!(format!("{}:1", type_name.to_lowercase()))
                }
            }
            _ => json!(null),
        }
    }
}

/// The analyzer named in a `FULLTEXT ANALYZER <name> ...` index definition.
fn fulltext_analyzer(definition: &str) -> Option<&str> {
    let mut rest = definition.trim_start();
    for keyword in ["FULLTEXT", "ANALYZER"] {
        let head = rest.get(..keyword.len())?;
        if !head.eq_ignore_ascii_case(keyword)
            || !rest[keyword.len()..].starts_with(char::is_whitespace)
        {
            return None;
        }
        rest = rest[keyword.len()..].trim_start();
    }
    if let Some(quoted) = rest.strip_prefix('`') {
        return quoted.split('`').next();
    }
    if let Some(quoted) = rest.strip_prefix('⟨') {
        return quoted.split('⟩').next();
    }
    rest.split_whitespace().next()
}

#[cfg(test)]
mod index_diff_tests {
    use super::*;
    use crate::schemasync::compare::types::{IndexDefinition, SchemaType};

    fn table_with_indexes(name: &str, indexes: Vec<IndexDefinition>) -> TableDefinition {
        TableDefinition {
            name: name.to_string(),
            schema_type: SchemaType::Schemafull,
            fields: BTreeMap::new(),
            array_wildcard_fields: BTreeMap::new(),
            permissions: None,
            indexes,
            events: Vec::new(),
        }
    }

    fn idx(name: &str, columns: &[&str], unique: bool) -> IndexDefinition {
        IndexDefinition {
            name: name.to_string(),
            columns: columns.iter().map(|c| c.to_string()).collect(),
            unique,
            definition: if unique { "UNIQUE" } else { "" }.to_string(),
        }
    }

    fn idx_with_definition(name: &str, columns: &[&str], definition: &str) -> IndexDefinition {
        IndexDefinition {
            name: name.to_string(),
            columns: columns.iter().map(|c| c.to_string()).collect(),
            unique: false,
            definition: definition.to_string(),
        }
    }

    fn schema_with(table_name: &str, table: TableDefinition) -> SchemaDefinition {
        let mut tables = BTreeMap::new();
        tables.insert(table_name.to_string(), table);
        SchemaDefinition {
            tables,
            edges: BTreeMap::new(),
            accesses: Vec::new(),
            analyzers: Vec::new(),
        }
    }

    fn analyzer(name: &str, statement: &str) -> AnalyzerDefinition {
        AnalyzerDefinition {
            name: name.to_string(),
            statement: statement.to_string(),
        }
    }

    #[test]
    fn compare_flags_index_with_changed_definition() {
        let old = schema_with(
            "post",
            table_with_indexes(
                "post",
                vec![idx_with_definition(
                    "post_search",
                    &["body"],
                    "FULLTEXT ANALYZER en BM25(1.2,0.75)",
                )],
            ),
        );
        let new = schema_with(
            "post",
            table_with_indexes(
                "post",
                vec![idx_with_definition(
                    "post_search",
                    &["body"],
                    "FULLTEXT ANALYZER en BM25(1.2,0.75) HIGHLIGHTS",
                )],
            ),
        );

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert_eq!(changes.modified_tables.len(), 1);
        let tc = &changes.modified_tables[0];
        assert!(tc.new_indexes.is_empty());
        assert!(tc.removed_indexes.is_empty());
        assert_eq!(tc.modified_indexes.len(), 1);
        assert_eq!(tc.modified_indexes[0].name, "post_search");
    }

    #[test]
    fn compare_flags_index_with_changed_columns() {
        let old = schema_with(
            "post",
            table_with_indexes("post", vec![idx("idx_post_x", &["a"], false)]),
        );
        let new = schema_with(
            "post",
            table_with_indexes("post", vec![idx("idx_post_x", &["a", "b"], false)]),
        );

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert_eq!(changes.modified_tables[0].modified_indexes.len(), 1);
    }

    #[test]
    fn compare_diffs_analyzers() {
        let table = table_with_indexes("post", vec![]);
        let mut old = schema_with("post", table.clone());
        old.analyzers = vec![
            analyzer("same", "DEFINE ANALYZER same TOKENIZERS blank"),
            analyzer("changed", "DEFINE ANALYZER changed TOKENIZERS blank"),
            analyzer("gone", "DEFINE ANALYZER gone TOKENIZERS blank"),
        ];
        let mut new = schema_with("post", table);
        new.analyzers = vec![
            analyzer("same", "DEFINE ANALYZER same TOKENIZERS blank"),
            analyzer(
                "changed",
                "DEFINE ANALYZER changed TOKENIZERS blank FILTERS lowercase",
            ),
            analyzer("added", "DEFINE ANALYZER added TOKENIZERS class"),
        ];

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert!(changes.modified_tables.is_empty());
        assert_eq!(changes.new_analyzers, vec!["added".to_string()]);
        assert_eq!(changes.removed_analyzers, vec!["gone".to_string()]);
        assert_eq!(changes.modified_analyzers, vec!["changed".to_string()]);
    }

    #[test]
    fn modified_analyzer_flags_dependent_fulltext_index() {
        let indexes = vec![
            idx_with_definition(
                "post_search",
                &["body"],
                "FULLTEXT ANALYZER en BM25(1.2,0.75)",
            ),
            idx_with_definition(
                "post_other",
                &["title"],
                "FULLTEXT ANALYZER other BM25(1.2,0.75)",
            ),
            idx("idx_post_slug", &["slug"], true),
        ];
        let mut old = schema_with("post", table_with_indexes("post", indexes.clone()));
        old.analyzers = vec![
            analyzer("en", "DEFINE ANALYZER en TOKENIZERS blank"),
            analyzer("other", "DEFINE ANALYZER other TOKENIZERS blank"),
        ];
        let mut new = schema_with("post", table_with_indexes("post", indexes));
        new.analyzers = vec![
            analyzer(
                "en",
                "DEFINE ANALYZER en TOKENIZERS blank FILTERS snowball(english)",
            ),
            analyzer("other", "DEFINE ANALYZER other TOKENIZERS blank"),
        ];

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert_eq!(changes.modified_analyzers, vec!["en".to_string()]);
        assert_eq!(changes.modified_tables.len(), 1);
        let tc = &changes.modified_tables[0];
        assert_eq!(tc.table_name, "post");
        let names: Vec<&str> = tc
            .modified_indexes
            .iter()
            .map(|i| i.name.as_str())
            .collect();
        assert_eq!(names, vec!["post_search"]);
        assert!(tc.removed_indexes.is_empty());
    }

    #[test]
    fn index_leaving_removed_analyzer_is_dropped_before_redefinition() {
        let mut old = schema_with(
            "post",
            table_with_indexes(
                "post",
                vec![idx_with_definition(
                    "post_search",
                    &["body"],
                    "FULLTEXT ANALYZER legacy BM25(1.2,0.75)",
                )],
            ),
        );
        old.analyzers = vec![analyzer(
            "legacy",
            "DEFINE ANALYZER legacy TOKENIZERS blank",
        )];
        let mut new = schema_with(
            "post",
            table_with_indexes(
                "post",
                vec![idx_with_definition(
                    "post_search",
                    &["body"],
                    "FULLTEXT ANALYZER en BM25(1.2,0.75)",
                )],
            ),
        );
        new.analyzers = vec![analyzer("en", "DEFINE ANALYZER en TOKENIZERS class")];

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert_eq!(changes.removed_analyzers, vec!["legacy".to_string()]);
        let tc = &changes.modified_tables[0];
        assert_eq!(tc.removed_indexes.len(), 1);
        assert_eq!(tc.removed_indexes[0].name, "post_search");
        assert_eq!(tc.modified_indexes.len(), 1, "flagged once, not duplicated");
        assert_eq!(
            tc.modified_indexes[0].definition,
            "FULLTEXT ANALYZER en BM25(1.2,0.75)"
        );
    }

    #[test]
    fn extracts_fulltext_analyzer_name() {
        assert_eq!(
            fulltext_analyzer("FULLTEXT ANALYZER en BM25(1.2,0.75) HIGHLIGHTS"),
            Some("en")
        );
        assert_eq!(
            fulltext_analyzer("FULLTEXT ANALYZER `my az` BM25"),
            Some("my az")
        );
        assert_eq!(fulltext_analyzer("fulltext analyzer en"), Some("en"));
        assert_eq!(fulltext_analyzer("FULLTEXTANALYZER en"), None);
        assert_eq!(fulltext_analyzer("FULLTEXT ANALYZER ⟨az⟩"), Some("az"));
        assert_eq!(fulltext_analyzer("UNIQUE"), None);
        assert_eq!(fulltext_analyzer(""), None);
    }

    #[test]
    fn compare_flags_orphan_index_for_removal() {
        let old = schema_with(
            "reaction",
            table_with_indexes(
                "reaction",
                vec![
                    idx("idx_reaction_user_message", &["user", "message"], true),
                    idx("idx_reaction_created_at", &["created_at"], false),
                ],
            ),
        );
        let new = schema_with(
            "reaction",
            table_with_indexes(
                "reaction",
                vec![idx("idx_reaction_user_message", &["user", "message"], true)],
            ),
        );

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert_eq!(changes.modified_tables.len(), 1);
        let tc = &changes.modified_tables[0];
        assert_eq!(tc.table_name, "reaction");
        assert!(tc.new_indexes.is_empty());
        assert_eq!(tc.removed_indexes.len(), 1);
        assert_eq!(tc.removed_indexes[0].name, "idx_reaction_created_at");
    }

    #[test]
    fn compare_flags_added_index() {
        let old = schema_with("reaction", table_with_indexes("reaction", vec![]));
        let new = schema_with(
            "reaction",
            table_with_indexes(
                "reaction",
                vec![idx("idx_reaction_created_at", &["created_at"], false)],
            ),
        );

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert_eq!(changes.modified_tables.len(), 1);
        let tc = &changes.modified_tables[0];
        assert_eq!(tc.new_indexes.len(), 1);
        assert_eq!(tc.new_indexes[0].name, "idx_reaction_created_at");
        assert!(tc.removed_indexes.is_empty());
    }

    #[test]
    fn compare_returns_no_change_when_indexes_identical() {
        let indexes = vec![idx("idx_reaction_user_message", &["user", "message"], true)];
        let old = schema_with("reaction", table_with_indexes("reaction", indexes.clone()));
        let new = schema_with("reaction", table_with_indexes("reaction", indexes));

        let changes = Comparator::compare(&old, &new).expect("compare");
        assert!(changes.modified_tables.is_empty());
    }
}