drizzle_migrations/sqlite/

statements.rs

//! `SQLite` SQL statement types and generation (v7 DDL format)
//!
//! This implements the full statement generation from drizzle-kit beta.
//! - `JsonStatement` enum represents migration operations
//! - Convertor functions convert statements to SQL strings

use crate::sqlite::ddl::{
    CheckConstraint, Column, ForeignKey, Index, PrimaryKey, Table, TableSql, UniqueConstraint, View,
};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// SQL statement breakpoint marker (used by drizzle-kit)
pub const BREAKPOINT: &str = "--> statement-breakpoint";

fn quote_ident(ident: &str) -> String {
    format!("`{}`", ident.replace('`', "``"))
}

// =============================================================================
// JSON Statement Types (matching statements.ts)
// =============================================================================

/// Full table information for create/recreate operations
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TableFull {
    pub name: String,
    pub columns: Vec<Column>,
    pub pk: Option<PrimaryKey>,
    pub fks: Vec<ForeignKey>,
    pub uniques: Vec<UniqueConstraint>,
    pub checks: Vec<CheckConstraint>,
    /// Whether the table has STRICT mode enabled
    #[serde(default)]
    pub strict: bool,
    /// Whether the table is WITHOUT ROWID
    #[serde(default)]
    pub without_rowid: bool,
}

impl TableFull {
    #[must_use]
    pub fn new(name: &str) -> Self {
        Self {
            name: name.to_string(),
            columns: Vec::new(),
            pk: None,
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: false,
            without_rowid: false,
        }
    }
}

fn table_full_to_table(table: &TableFull) -> Table {
    Table {
        name: table.name.clone().into(),
        strict: table.strict,
        without_rowid: table.without_rowid,
    }
}

fn create_table_sql(table: &TableFull) -> String {
    let ddl_table = table_full_to_table(table);
    TableSql::new(&ddl_table)
        .columns(&table.columns)
        .primary_key(table.pk.as_ref())
        .foreign_keys(&table.fks)
        .unique_constraints(&table.uniques)
        .check_constraints(&table.checks)
        .create_table_sql()
}

/// All possible JSON statement types
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum JsonStatement {
    CreateTable(CreateTableStatement),
    DropTable(DropTableStatement),
    RenameTable(RenameTableStatement),
    AddColumn(AddColumnStatement),
    DropColumn(DropColumnStatement),
    RenameColumn(RenameColumnStatement),
    RecreateColumn(RecreateColumnStatement),
    RecreateTable(RecreateTableStatement),
    CreateIndex(CreateIndexStatement),
    DropIndex(DropIndexStatement),
    CreateView(CreateViewStatement),
    DropView(DropViewStatement),
    RenameView(RenameViewStatement),
}

impl JsonStatement {
    /// Get the type name of this statement
    #[must_use]
    pub const fn type_name(&self) -> &'static str {
        match self {
            Self::CreateTable(_) => "create_table",
            Self::DropTable(_) => "drop_table",
            Self::RenameTable(_) => "rename_table",
            Self::AddColumn(_) => "add_column",
            Self::DropColumn(_) => "drop_column",
            Self::RenameColumn(_) => "rename_column",
            Self::RecreateColumn(_) => "recreate_column",
            Self::RecreateTable(_) => "recreate_table",
            Self::CreateIndex(_) => "create_index",
            Self::DropIndex(_) => "drop_index",
            Self::CreateView(_) => "create_view",
            Self::DropView(_) => "drop_view",
            Self::RenameView(_) => "rename_view",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateTableStatement {
    pub table: TableFull,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DropTableStatement {
    pub table_name: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RenameTableStatement {
    pub from: String,
    pub to: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AddColumnStatement {
    pub column: Column,
    pub fk: Option<ForeignKey>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DropColumnStatement {
    pub column: Column,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RenameColumnStatement {
    pub table: String,
    pub from: String,
    pub to: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RecreateColumnStatement {
    pub column: Column,
    pub fk: Option<ForeignKey>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RecreateTableStatement {
    pub from: TableFull,
    pub to: TableFull,
    #[serde(default)]
    pub data: Option<RebuildTableData>,
}

/// Validated data movement rendered as part of one table rebuild.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct RebuildTableData {
    pub copies: BTreeMap<String, RebuildCopyExpression>,
    pub validations: Vec<RebuildDataValidation>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "camelCase", deny_unknown_fields)]
pub enum RebuildCopyExpression {
    HexTextToBlob {
        source: String,
    },
    IntegerMap {
        source: String,
        cases: Vec<(i64, i64)>,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "camelCase", deny_unknown_fields)]
pub enum RebuildDataValidation {
    HexText { column: String, bytes: usize },
    JsonValid { column: String },
    IntegerSet { column: String, allowed: Vec<i64> },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateIndexStatement {
    pub index: Index,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DropIndexStatement {
    pub index: Index,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateViewStatement {
    pub view: View,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DropViewStatement {
    pub view: View,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RenameViewStatement {
    pub from: View,
    pub to: View,
}

// =============================================================================
// Convertor - Statement to SQL (matching convertor.ts)
// =============================================================================

/// Convert a JSON statement to SQL string(s)
#[must_use]
pub fn convert_statement(statement: &JsonStatement) -> Vec<String> {
    match statement {
        JsonStatement::CreateTable(st) => vec![convert_create_table(st)],
        JsonStatement::DropTable(st) => vec![convert_drop_table(st)],
        JsonStatement::RenameTable(st) => vec![convert_rename_table(st)],
        JsonStatement::AddColumn(st) => vec![convert_add_column(st)],
        JsonStatement::DropColumn(st) => vec![convert_drop_column(st)],
        JsonStatement::RenameColumn(st) => vec![convert_rename_column(st)],
        JsonStatement::RecreateColumn(st) => convert_recreate_column(st),
        JsonStatement::RecreateTable(st) => convert_recreate_table(st),
        JsonStatement::CreateIndex(st) => vec![convert_create_index(st)],
        JsonStatement::DropIndex(st) => vec![convert_drop_index(st)],
        JsonStatement::CreateView(st) => vec![convert_create_view(st)],
        JsonStatement::DropView(st) => vec![convert_drop_view(st)],
        JsonStatement::RenameView(st) => convert_rename_view(st),
    }
}

/// Convert multiple statements to SQL with optional breakpoints
pub fn statements_to_sql(statements: &[JsonStatement], breakpoints: bool) -> String {
    let sql_statements: Vec<String> = statements.iter().flat_map(convert_statement).collect();

    if breakpoints {
        sql_statements.join(&format!("\n{BREAKPOINT}\n"))
    } else {
        sql_statements.join("\n")
    }
}

/// A grouped statement with its corresponding SQL
#[derive(Debug, Clone)]
pub struct GroupedStatement {
    /// The JSON statement
    pub json_statement: JsonStatement,
    /// The generated SQL statements
    pub sql_statements: Vec<String>,
}

/// Result of converting JSON statements to SQL
#[derive(Debug, Clone)]
pub struct ConversionResult {
    /// All SQL statements flattened
    pub sql_statements: Vec<String>,
    /// Statements grouped with their JSON source
    pub grouped_statements: Vec<GroupedStatement>,
}

/// Convert JSON statements to SQL with grouping information
#[must_use]
pub fn from_json(statements: Vec<JsonStatement>) -> ConversionResult {
    let grouped: Vec<GroupedStatement> = statements
        .into_iter()
        .map(|statement| {
            let sql_statements = convert_statement(&statement);
            GroupedStatement {
                json_statement: statement,
                sql_statements,
            }
        })
        .collect();

    let sql_statements: Vec<String> = grouped
        .iter()
        .flat_map(|g| g.sql_statements.clone())
        .collect();

    ConversionResult {
        sql_statements,
        grouped_statements: grouped,
    }
}

// =============================================================================
// Individual Convertors
// =============================================================================

fn convert_create_table(st: &CreateTableStatement) -> String {
    create_table_sql(&st.table)
}

fn convert_drop_table(st: &DropTableStatement) -> String {
    format!("DROP TABLE {};", quote_ident(&st.table_name))
}

fn convert_rename_table(st: &RenameTableStatement) -> String {
    format!(
        "ALTER TABLE {} RENAME TO {};",
        quote_ident(&st.from),
        quote_ident(&st.to)
    )
}

fn convert_add_column(st: &AddColumnStatement) -> String {
    let column = &st.column;
    let column_def = column.to_column_sql(false, false);

    let reference = st
        .fk
        .as_ref()
        .map(|fk| {
            let to_cols = fk
                .columns_to
                .iter()
                .map(|c| quote_ident(c))
                .collect::<Vec<_>>()
                .join(",");
            if fk.name_explicit {
                format!(
                    " CONSTRAINT {} REFERENCES {}({})",
                    quote_ident(&fk.name),
                    quote_ident(&fk.table_to),
                    to_cols
                )
            } else {
                format!(" REFERENCES {}({})", quote_ident(&fk.table_to), to_cols)
            }
        })
        .unwrap_or_default();

    format!(
        "ALTER TABLE {} ADD {}{};",
        quote_ident(&column.table),
        column_def,
        reference
    )
}

fn convert_drop_column(st: &DropColumnStatement) -> String {
    format!(
        "ALTER TABLE {} DROP COLUMN {};",
        quote_ident(&st.column.table),
        quote_ident(&st.column.name)
    )
}

fn convert_rename_column(st: &RenameColumnStatement) -> String {
    format!(
        "ALTER TABLE {} RENAME COLUMN {} TO {};",
        quote_ident(&st.table),
        quote_ident(&st.from),
        quote_ident(&st.to)
    )
}

fn convert_recreate_column(st: &RecreateColumnStatement) -> Vec<String> {
    // Drop and re-add the column
    let drop = format!(
        "ALTER TABLE {} DROP COLUMN {};",
        quote_ident(&st.column.table),
        quote_ident(&st.column.name)
    );
    let add = convert_add_column(&AddColumnStatement {
        column: st.column.clone(),
        fk: st.fk.clone(),
    });
    vec![drop, add]
}

fn convert_recreate_table(st: &RecreateTableStatement) -> Vec<String> {
    let name = &st.to.name;
    let new_table_name = format!("__new_{name}");

    let data = st.data.as_ref();
    let copied_columns = st
        .to
        .columns
        .iter()
        .filter(|target| {
            target.generated.is_none()
                && (data.is_some_and(|data| data.copies.contains_key(target.name.as_ref()))
                    || st
                        .from
                        .columns
                        .iter()
                        .any(|source| source.name == target.name && source.generated.is_none()))
        })
        .collect::<Vec<_>>();
    let cols_str = copied_columns
        .iter()
        .map(|column| quote_ident(&column.name))
        .collect::<Vec<_>>()
        .join(", ");
    let select_str = copied_columns
        .iter()
        .map(|column| {
            data.and_then(|data| data.copies.get(column.name.as_ref()))
                .map_or_else(|| quote_ident(&column.name), render_copy_expression)
        })
        .collect::<Vec<_>>()
        .join(", ");

    let mut statements = Vec::new();

    if let Some(data) = data
        && !data.validations.is_empty()
    {
        let guard = format!("__drizzle_rebuild_guard_{name}");
        statements.push(format!(
            "CREATE TEMP TABLE {} (`valid` INTEGER NOT NULL CHECK (`valid` = 1)) STRICT;",
            quote_ident(&guard)
        ));
        for validation in &data.validations {
            statements.push(format!(
                "INSERT INTO {}(`valid`) SELECT 0 FROM {} WHERE {} LIMIT 1;",
                quote_ident(&guard),
                quote_ident(name),
                render_invalid_predicate(validation)
            ));
        }
        statements.push(format!("DROP TABLE {};", quote_ident(&guard)));
    }

    // 1. Disable foreign keys
    statements.push("PRAGMA foreign_keys=OFF;".to_string());

    // 2. Create new table with temp name
    let mut tmp_table = st.to.clone();
    tmp_table.name.clone_from(&new_table_name);
    // Update check constraint table references
    for check in &mut tmp_table.checks {
        check.table = new_table_name.clone().into();
    }
    statements.push(convert_create_table(&CreateTableStatement {
        table: tmp_table,
    }));

    // 3. Copy data. If the new table has NOT NULL columns without a default
    // that the old table can't provide, the copy will fail on non-empty
    // tables — keep generating (matching drizzle-kit) but flag it.
    let missing_not_null: Vec<&str> = st
        .to
        .columns
        .iter()
        .filter(|col| {
            col.not_null
                && col.default.is_none()
                && col.generated.is_none()
                && !st.from.columns.iter().any(|from| from.name == col.name)
        })
        .map(|col| col.name.as_ref())
        .collect();
    let warning = if missing_not_null.is_empty() {
        String::new()
    } else {
        format!(
            "-- WARNING: new NOT NULL column(s) without a default: {} — the copy below fails if `{name}` has rows\n",
            missing_not_null.join(", ")
        )
    };
    statements.push(format!(
        "{warning}INSERT INTO {}({cols_str}) SELECT {select_str} FROM {};",
        quote_ident(&new_table_name),
        quote_ident(name)
    ));

    // 4. Drop old table
    statements.push(format!("DROP TABLE {};", quote_ident(name)));

    // 5. Rename new table
    statements.push(format!(
        "ALTER TABLE {} RENAME TO {};",
        quote_ident(&new_table_name),
        quote_ident(name)
    ));

    // 6. Re-enable foreign keys
    statements.push("PRAGMA foreign_keys=ON;".to_string());

    statements
}

fn render_copy_expression(expression: &RebuildCopyExpression) -> String {
    match expression {
        RebuildCopyExpression::HexTextToBlob { source } => {
            format!("unhex({})", quote_ident(source))
        }
        RebuildCopyExpression::IntegerMap { source, cases } => {
            let cases = cases
                .iter()
                .map(|(from, to)| format!(" WHEN {from} THEN {to}"))
                .collect::<String>();
            format!("CASE {}{cases} ELSE NULL END", quote_ident(source))
        }
    }
}

fn render_invalid_predicate(validation: &RebuildDataValidation) -> String {
    match validation {
        RebuildDataValidation::HexText { column, bytes } => {
            let column = quote_ident(column);
            let chars = bytes.saturating_mul(2);
            format!(
                "{column} IS NOT NULL AND (typeof({column}) <> 'text' OR length({column}) <> {chars} OR coalesce(length(unhex({column})), -1) <> {bytes})"
            )
        }
        RebuildDataValidation::JsonValid { column } => {
            let column = quote_ident(column);
            format!(
                "{column} IS NOT NULL AND (typeof({column}) <> 'text' OR json_valid({column}) <> 1)"
            )
        }
        RebuildDataValidation::IntegerSet { column, allowed } => {
            let column = quote_ident(column);
            let allowed = allowed
                .iter()
                .map(i64::to_string)
                .collect::<Vec<_>>()
                .join(", ");
            format!(
                "{column} IS NOT NULL AND (typeof({column}) <> 'integer' OR {column} NOT IN ({allowed}))"
            )
        }
    }
}

fn convert_create_index(st: &CreateIndexStatement) -> String {
    st.index.create_index_sql()
}

fn convert_drop_index(st: &DropIndexStatement) -> String {
    format!("DROP INDEX IF EXISTS {};", quote_ident(&st.index.name))
}

fn convert_create_view(st: &CreateViewStatement) -> String {
    st.view.create_view_sql()
}

fn convert_drop_view(st: &DropViewStatement) -> String {
    format!("DROP VIEW {};", quote_ident(&st.view.name))
}

fn convert_rename_view(st: &RenameViewStatement) -> Vec<String> {
    // SQLite doesn't support RENAME VIEW, so we drop and recreate.
    // Two separate statements: executors run one statement per string.
    vec![
        format!("DROP VIEW IF EXISTS {};", quote_ident(&st.from.name)),
        st.to.create_view_sql(),
    ]
}

// =============================================================================
// Statement Preparation Helpers
// =============================================================================

/// Prepare add column statements with FK associations
#[must_use]
pub fn prepare_add_columns(columns: &[Column], fks: &[ForeignKey]) -> Vec<AddColumnStatement> {
    columns
        .iter()
        .map(|col| {
            let fk = fks
                .iter()
                .find(|fk| {
                    fk.columns.len() == 1 && fk.columns[0] == col.name && fk.table == col.table
                })
                .cloned();
            AddColumnStatement {
                column: col.clone(),
                fk,
            }
        })
        .collect()
}

// =============================================================================
// Topological Sorting for Table Dependencies
// =============================================================================

use crate::sqlite::SchemaDiff;
use crate::sqlite::collection::{DiffType, EntityDiff};
use crate::traits::EntityKind;
use std::collections::{HashMap, HashSet};

/// Result of topological sorting with circular dependency detection
pub struct TopologicalSortResult<'a> {
    /// Sorted tables
    pub tables: Vec<&'a EntityDiff>,
    /// Whether circular dependencies were detected
    pub has_circular_deps: bool,
}

/// Topological sort tables for CREATE: referenced tables come first
fn topological_sort_tables_for_create<'a>(
    tables: &[&'a EntityDiff],
    diff: &SchemaDiff,
) -> TopologicalSortResult<'a> {
    if tables.len() <= 1 {
        return TopologicalSortResult {
            tables: tables.to_vec(),
            has_circular_deps: false,
        };
    }

    // Build a map of table name -> entity diff
    let mut table_map: HashMap<String, &EntityDiff> = HashMap::new();
    for t in tables {
        if let Some(name) = t.name.split(':').next_back() {
            table_map.insert(name.to_string(), *t);
        }
    }

    // Build dependency graph: table -> tables it depends on (via FKs)
    let mut dependencies: HashMap<String, HashSet<String>> = HashMap::new();
    for table_name in table_map.keys() {
        dependencies.insert(table_name.clone(), HashSet::new());
    }

    // Find FK dependencies
    for fk_diff in diff.by_kind(EntityKind::ForeignKey) {
        if fk_diff.diff_type == DiffType::Create
            && let Some(crate::sqlite::ddl::SqliteEntity::ForeignKey(fk)) = fk_diff.right.as_ref()
        {
            let from_table = fk.table.to_string();
            let to_table = fk.table_to.to_string();
            // from_table depends on to_table (to_table must be created first)
            if table_map.contains_key(&from_table)
                && table_map.contains_key(&to_table)
                && let Some(deps) = dependencies.get_mut(&from_table)
            {
                deps.insert(to_table);
            }
        }
    }

    // Tables with no dependencies come first, then tables that depend on them, etc.
    // Ready-sets and the circular-dependency fallback are sorted so the emitted
    // SQL order is deterministic.
    let mut result = Vec::new();
    let mut remaining: HashSet<String> = table_map.keys().cloned().collect();
    let mut satisfied: HashSet<String> = HashSet::new();
    let mut has_circular_deps = false;

    while !remaining.is_empty() {
        // Find tables whose dependencies are all satisfied
        let mut ready: Vec<String> = remaining
            .iter()
            .filter(|t| {
                dependencies
                    .get(*t)
                    .is_none_or(|deps| deps.iter().all(|d| satisfied.contains(d)))
            })
            .cloned()
            .collect();
        ready.sort_unstable();

        if ready.is_empty() {
            // Circular dependency detected - add remaining in sorted order
            has_circular_deps = true;
            let mut leftover: Vec<&String> = remaining.iter().collect();
            leftover.sort_unstable();
            for t in leftover {
                if let Some(entity) = table_map.get(t) {
                    result.push(*entity);
                }
            }
            break;
        }

        for t in ready {
            remaining.remove(&t);
            satisfied.insert(t.clone());
            if let Some(entity) = table_map.get(&t) {
                result.push(*entity);
            }
        }
    }

    TopologicalSortResult {
        tables: result,
        has_circular_deps,
    }
}

/// Topological sort tables for DROP: tables with FKs come first (reverse of create)
fn topological_sort_tables_for_drop<'a>(
    tables: &[&'a EntityDiff],
    diff: &SchemaDiff,
) -> TopologicalSortResult<'a> {
    // For drops, reverse the create order: tables that reference others drop first
    let create_result = topological_sort_tables_for_create(tables, diff);
    TopologicalSortResult {
        tables: create_result.tables.into_iter().rev().collect(),
        has_circular_deps: create_result.has_circular_deps,
    }
}

fn append_drop_table_stmts(statements: &mut Vec<String>, diff: &SchemaDiff) {
    let dropped_tables = diff.dropped_tables();
    let drop_result = topological_sort_tables_for_drop(&dropped_tables, diff);

    // If there are circular dependencies in drops, wrap with PRAGMA
    if drop_result.has_circular_deps && !drop_result.tables.is_empty() {
        statements.push("PRAGMA foreign_keys=OFF;".to_string());
    }

    for entity_diff in &drop_result.tables {
        if let Some(name) = entity_diff.name.split(':').next_back() {
            statements.push(convert_drop_table(&DropTableStatement {
                table_name: name.to_string(),
            }));
        }
    }

    if drop_result.has_circular_deps && !drop_result.tables.is_empty() {
        statements.push("PRAGMA foreign_keys=ON;".to_string());
    }
}

fn build_table_full(diff: &SchemaDiff, table: &crate::sqlite::ddl::Table) -> TableFull {
    let columns_for_table: Vec<Column> = diff
        .by_kind(EntityKind::Column)
        .into_iter()
        .filter(|d| d.diff_type == DiffType::Create)
        .filter_map(|d| d.right.as_ref())
        .filter_map(|e| match e {
            crate::sqlite::ddl::SqliteEntity::Column(c) if c.table == table.name => Some(c.clone()),
            _ => None,
        })
        .collect();

    let pk = diff
        .by_kind(EntityKind::PrimaryKey)
        .into_iter()
        .filter(|d| d.diff_type == DiffType::Create)
        .filter_map(|d| d.right.as_ref())
        .find_map(|e| match e {
            crate::sqlite::ddl::SqliteEntity::PrimaryKey(p) if p.table == table.name => {
                Some(p.clone())
            }
            _ => None,
        });

    let fks: Vec<ForeignKey> = diff
        .by_kind(EntityKind::ForeignKey)
        .into_iter()
        .filter(|d| d.diff_type == DiffType::Create)
        .filter_map(|d| d.right.as_ref())
        .filter_map(|e| match e {
            crate::sqlite::ddl::SqliteEntity::ForeignKey(fk) if fk.table == table.name => {
                Some(fk.clone())
            }
            _ => None,
        })
        .collect();

    let uniques: Vec<UniqueConstraint> = diff
        .by_kind(EntityKind::UniqueConstraint)
        .into_iter()
        .filter(|d| d.diff_type == DiffType::Create)
        .filter_map(|d| d.right.as_ref())
        .filter_map(|e| match e {
            crate::sqlite::ddl::SqliteEntity::UniqueConstraint(u) if u.table == table.name => {
                Some(u.clone())
            }
            _ => None,
        })
        .collect();

    let checks: Vec<CheckConstraint> = diff
        .by_kind(EntityKind::CheckConstraint)
        .into_iter()
        .filter(|d| d.diff_type == DiffType::Create)
        .filter_map(|d| d.right.as_ref())
        .filter_map(|e| match e {
            crate::sqlite::ddl::SqliteEntity::CheckConstraint(c) if c.table == table.name => {
                Some(c.clone())
            }
            _ => None,
        })
        .collect();

    TableFull {
        name: table.name.to_string(),
        columns: columns_for_table,
        pk,
        fks,
        uniques,
        checks,
        strict: table.strict,
        without_rowid: table.without_rowid,
    }
}

fn append_create_table_stmts(statements: &mut Vec<String>, diff: &SchemaDiff) {
    let created_tables = diff.created_tables();
    let create_result = topological_sort_tables_for_create(&created_tables, diff);

    // If there are circular dependencies, wrap creates with PRAGMA to allow out-of-order creation
    if create_result.has_circular_deps && !create_result.tables.is_empty() {
        statements.push("PRAGMA foreign_keys=OFF;".to_string());
    }

    for entity_diff in &create_result.tables {
        if let Some(crate::sqlite::ddl::SqliteEntity::Table(table)) = entity_diff.right.as_ref() {
            let table_full = build_table_full(diff, table);
            statements.push(convert_create_table(&CreateTableStatement {
                table: table_full,
            }));
        }
    }

    // Re-enable foreign keys if we disabled them for circular dependencies
    if create_result.has_circular_deps && !create_result.tables.is_empty() {
        statements.push("PRAGMA foreign_keys=ON;".to_string());
    }
}

fn append_add_column_stmts(statements: &mut Vec<String>, diff: &SchemaDiff) {
    for entity_diff in diff.by_kind(EntityKind::Column) {
        if entity_diff.diff_type == DiffType::Create
            && let Some(crate::sqlite::ddl::SqliteEntity::Column(col)) = entity_diff.right.as_ref()
        {
            let table_was_created = diff.created_tables().iter().any(|t| t.name == col.table);

            if !table_was_created {
                statements.push(convert_add_column(&AddColumnStatement {
                    column: col.clone(),
                    fk: None,
                }));
            }
        }
    }
}

fn append_index_stmts(statements: &mut Vec<String>, diff: &SchemaDiff) {
    for entity_diff in diff.by_kind(EntityKind::Index) {
        match entity_diff.diff_type {
            DiffType::Drop => {
                if let Some(crate::sqlite::ddl::SqliteEntity::Index(idx)) =
                    entity_diff.left.as_ref()
                {
                    statements.push(convert_drop_index(&DropIndexStatement {
                        index: idx.clone(),
                    }));
                }
            }
            DiffType::Create => {
                if let Some(crate::sqlite::ddl::SqliteEntity::Index(idx)) =
                    entity_diff.right.as_ref()
                {
                    statements.push(convert_create_index(&CreateIndexStatement {
                        index: idx.clone(),
                    }));
                }
            }
            DiffType::Alter => {
                // For index alter: drop old, create new
                if let (
                    Some(crate::sqlite::ddl::SqliteEntity::Index(old)),
                    Some(crate::sqlite::ddl::SqliteEntity::Index(new)),
                ) = (entity_diff.left.as_ref(), entity_diff.right.as_ref())
                {
                    statements.push(convert_drop_index(&DropIndexStatement {
                        index: old.clone(),
                    }));
                    statements.push(convert_create_index(&CreateIndexStatement {
                        index: new.clone(),
                    }));
                }
            }
        }
    }
}

// =============================================================================
// SQLite SQL Generator
// =============================================================================

/// `SQLite` SQL generator for migration diffs
///
/// Generates SQL statements from schema diffs with proper ordering:
/// 1. Table drops (reverse dependency order)
/// 2. Table creates (dependency order - referenced tables first)
/// 3. Column additions for existing tables
/// 4. Index operations
pub struct Generator {
    /// Whether to include statement breakpoints
    pub breakpoints: bool,
}

impl Default for Generator {
    fn default() -> Self {
        Self::new()
    }
}

impl Generator {
    #[must_use]
    pub const fn new() -> Self {
        Self { breakpoints: true }
    }

    #[must_use]
    pub const fn with_breakpoints(mut self, breakpoints: bool) -> Self {
        self.breakpoints = breakpoints;
        self
    }

    /// Generate SQL from a schema diff
    #[must_use]
    pub fn generate_migration(&self, diff: &SchemaDiff) -> Vec<String> {
        let mut statements = Vec::new();

        append_drop_table_stmts(&mut statements, diff);
        append_create_table_stmts(&mut statements, diff);
        append_add_column_stmts(&mut statements, diff);
        append_index_stmts(&mut statements, diff);

        statements
    }

    /// Generate SQL from migration statements
    #[must_use]
    pub fn statements_to_sql(&self, statements: &[String]) -> String {
        if self.breakpoints {
            statements.join(&format!("\n{BREAKPOINT}\n"))
        } else {
            statements.join("\n")
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sqlite::ddl::{Column, IndexColumn};

    #[test]
    fn test_create_table_simple() {
        let table = TableFull {
            name: "users".to_string(),
            columns: vec![
                Column::new("users", "id", "integer").not_null(),
                Column::new("users", "name", "text").not_null(),
            ],
            pk: None,
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: false,
            without_rowid: false,
        };

        let sql = convert_create_table(&CreateTableStatement { table });
        assert_eq!(
            sql,
            "CREATE TABLE `users` (\n\t`id` INTEGER NOT NULL,\n\t`name` TEXT NOT NULL\n);"
        );
    }

    #[test]
    fn test_create_table_with_pk() {
        let table = TableFull {
            name: "users".to_string(),
            columns: vec![Column::new("users", "id", "integer")],
            pk: Some(PrimaryKey::from_strings(
                "users".to_string(),
                "users_pk".to_string(),
                vec!["id".to_string()],
            )),
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: false,
            without_rowid: false,
        };

        let sql = convert_create_table(&CreateTableStatement { table });
        assert_eq!(
            sql,
            "CREATE TABLE `users` (\n\t`id` INTEGER PRIMARY KEY\n);"
        );
    }

    #[test]
    fn test_drop_table() {
        let sql = convert_drop_table(&DropTableStatement {
            table_name: "users".to_string(),
        });
        assert_eq!(sql, "DROP TABLE `users`;");
    }

    #[test]
    fn test_add_column() {
        let sql = convert_add_column(&AddColumnStatement {
            column: Column::new("users", "email", "text").not_null(),
            fk: None,
        });
        assert_eq!(sql, "ALTER TABLE `users` ADD `email` TEXT NOT NULL;");
    }

    #[test]
    fn test_create_index() {
        let index = Index::new(
            "users",
            "idx_users_email",
            vec![IndexColumn {
                value: "email".into(),
                is_expression: false,
            }],
        )
        .unique();

        let sql = convert_create_index(&CreateIndexStatement { index });
        assert_eq!(
            sql,
            "CREATE UNIQUE INDEX `idx_users_email` ON `users`(`email`);"
        );
    }

    #[test]
    fn test_rename_view_returns_two_statements() {
        let mut from = View::new("old_view");
        from.definition = Some("SELECT 1".into());
        let mut to = View::new("new_view");
        to.definition = Some("SELECT 1".into());

        let statements =
            convert_statement(&JsonStatement::RenameView(RenameViewStatement { from, to }));
        assert_eq!(
            statements,
            vec![
                "DROP VIEW IF EXISTS `old_view`;".to_string(),
                "CREATE VIEW `new_view` AS SELECT 1;".to_string(),
            ]
        );
    }

    #[test]
    fn test_recreate_warns_on_new_not_null_column_without_default() {
        let from = TableFull {
            name: "users".to_string(),
            columns: vec![Column::new("users", "id", "integer").not_null()],
            pk: None,
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: false,
            without_rowid: false,
        };
        let to = TableFull {
            name: "users".to_string(),
            columns: vec![
                Column::new("users", "id", "integer").not_null(),
                Column::new("users", "email", "text").not_null(),
            ],
            pk: None,
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: false,
            without_rowid: false,
        };

        let statements = convert_recreate_table(&RecreateTableStatement {
            from,
            to,
            data: None,
        });
        let insert = statements
            .iter()
            .find(|s| s.contains("INSERT INTO"))
            .expect("insert statement");
        assert!(
            insert.starts_with("-- WARNING:") && insert.contains("email"),
            "expected warning comment ahead of the INSERT, got: {insert}"
        );
    }

    #[test]
    fn test_create_table_strict() {
        let table = TableFull {
            name: "data".to_string(),
            columns: vec![
                Column::new("data", "id", "integer").not_null(),
                Column::new("data", "value", "text").not_null(),
            ],
            pk: None,
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: true,
            without_rowid: false,
        };

        let sql = convert_create_table(&CreateTableStatement { table });
        assert_eq!(
            sql,
            "CREATE TABLE `data` (\n\t`id` INTEGER NOT NULL,\n\t`value` TEXT NOT NULL\n) STRICT;"
        );
    }

    #[test]
    fn test_create_table_without_rowid() {
        let table = TableFull {
            name: "kv".to_string(),
            columns: vec![
                Column::new("kv", "key", "text").not_null(),
                Column::new("kv", "value", "blob"),
            ],
            pk: Some(PrimaryKey::from_strings(
                "kv".to_string(),
                "kv_pk".to_string(),
                vec!["key".to_string()],
            )),
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: false,
            without_rowid: true,
        };

        let sql = convert_create_table(&CreateTableStatement { table });
        assert_eq!(
            sql,
            "CREATE TABLE `kv` (\n\t`key` TEXT PRIMARY KEY NOT NULL,\n\t`value` BLOB\n) WITHOUT ROWID;"
        );
    }

    #[test]
    fn test_create_table_strict_without_rowid() {
        let table = TableFull {
            name: "cache".to_string(),
            columns: vec![
                Column::new("cache", "key", "text").not_null(),
                Column::new("cache", "data", "blob"),
            ],
            pk: Some(PrimaryKey::from_strings(
                "cache".to_string(),
                "cache_pk".to_string(),
                vec!["key".to_string()],
            )),
            fks: Vec::new(),
            uniques: Vec::new(),
            checks: Vec::new(),
            strict: true,
            without_rowid: true,
        };

        let sql = convert_create_table(&CreateTableStatement { table });
        assert_eq!(
            sql,
            "CREATE TABLE `cache` (\n\t`key` TEXT PRIMARY KEY NOT NULL,\n\t`data` BLOB\n) WITHOUT ROWID, STRICT;"
        );
    }
}