//! Versioned, manifest-driven schema migrations.
//!
//! The manifest (emitted by the `#[entity]` macro) is the desired schema. We
//! keep the *last-migrated* state in a committed `.snapshot.json` and diff the
//! current manifest against it to **generate** timestamped `.sql` files; a
//! **runner** then applies any not-yet-recorded files and logs them in a
//! `_roundabout_migrations` table.
//!
//! Migrations are **additive-only by default** — `CREATE TABLE` / `ADD COLUMN`
//! always run, while drops and type changes are written as commented-out SQL
//! (and warned about) unless `ROUND_ABOUT_DESTRUCTIVE=1` is set. FK constraints
//! are not emitted (the macro models a relation as a plain `<name>_id` column).

use crate::db::postgres::column_ddl;
use deadpool_postgres::Pool;
use roundabout_orm_schema::EntityDef;
use std::collections::hash_map::DefaultHasher;
use std::collections::HashSet;
use std::error::Error;
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};

type DynErr = Box<dyn Error + Send + Sync>;

const SNAPSHOT_FILE: &str = ".snapshot.json";

/// Diff the manifest against the last-migrated snapshot; if they differ, write a
/// new migration file and advance the snapshot. Returns the file path written.
pub fn generate(manifest: &[EntityDef], dir: impl AsRef<Path>) -> Option<PathBuf> {
    let dir = dir.as_ref();
    if let Err(e) = std::fs::create_dir_all(dir) {
        eprintln!("[roundabout_orm] cannot create migrations dir {}: {e}", dir.display());
        return None;
    }

    let snapshot = load_snapshot(dir);
    let destructive = std::env::var("ROUND_ABOUT_DESTRUCTIVE").is_ok();
    let mut stmts: Vec<String> = Vec::new();

    for ent in manifest {
        match snapshot.iter().find(|e| e.table == ent.table) {
            // Brand-new entity → create the table.
            None => {
                let cols: Vec<String> = ent.fields.iter().map(column_ddl).collect();
                stmts.push(format!(
                    "CREATE TABLE IF NOT EXISTS {} ({});",
                    ent.table,
                    cols.join(", ")
                ));
            }
            // Existing entity → reconcile columns.
            Some(prev) => {
                for f in &ent.fields {
                    match prev.fields.iter().find(|pf| pf.name == f.name) {
                        None => stmts.push(format!(
                            "ALTER TABLE {} ADD COLUMN IF NOT EXISTS {};",
                            ent.table,
                            column_ddl(f)
                        )),
                        Some(pf) if pf.sql_type != f.sql_type => stmts.push(gate(
                            destructive,
                            format!(
                                "ALTER TABLE {} ALTER COLUMN {} TYPE {} USING {}::{};",
                                ent.table, f.name, f.sql_type, f.name, f.sql_type
                            ),
                            &format!("change type of {}.{} to {}", ent.table, f.name, f.sql_type),
                        )),
                        Some(_) => {}
                    }
                }
                for pf in &prev.fields {
                    if !ent.fields.iter().any(|f| f.name == pf.name) {
                        stmts.push(gate(
                            destructive,
                            format!("ALTER TABLE {} DROP COLUMN IF EXISTS {};", ent.table, pf.name),
                            &format!("drop column {}.{}", ent.table, pf.name),
                        ));
                    }
                }
            }
        }
    }

    // Tables present in the snapshot but no longer in the manifest.
    for prev in &snapshot {
        if !manifest.iter().any(|e| e.table == prev.table) {
            stmts.push(gate(
                destructive,
                format!("DROP TABLE IF EXISTS {};", prev.table),
                &format!("drop table {}", prev.table),
            ));
        }
    }

    if stmts.is_empty() {
        return None; // schema already matches the snapshot
    }

    let millis = now_millis();
    let path = dir.join(format!("{millis:013}_auto.sql"));
    let body = format!(
        "-- roundabout_orm auto-migration\n-- generated at epoch-millis {millis}\n\n{}\n",
        stmts.join("\n")
    );
    if let Err(e) = std::fs::write(&path, body) {
        eprintln!("[roundabout_orm] cannot write migration {}: {e}", path.display());
        return None;
    }
    // Advance the snapshot to the manifest so we don't regenerate next run.
    write_snapshot(dir, manifest);
    println!("[roundabout_orm] generated migration {}", path.display());
    Some(path)
}

/// Apply every `*.sql` migration not yet recorded, in filename order, each in
/// its own transaction. Returns how many were applied.
pub async fn run_pending(pool: &Pool, dir: impl AsRef<Path>) -> Result<usize, DynErr> {
    let dir = dir.as_ref();
    let mut client = pool.get().await?;
    client
        .batch_execute(
            "CREATE TABLE IF NOT EXISTS _roundabout_migrations (
                id          BIGSERIAL PRIMARY KEY,
                filename    TEXT UNIQUE NOT NULL,
                checksum    TEXT NOT NULL,
                applied_at  TIMESTAMPTZ NOT NULL DEFAULT now()
            );",
        )
        .await?;

    let applied: HashSet<String> = client
        .query("SELECT filename FROM _roundabout_migrations", &[])
        .await?
        .iter()
        .map(|r| r.get::<_, String>(0))
        .collect();

    let mut files: Vec<PathBuf> = match std::fs::read_dir(dir) {
        Ok(rd) => rd
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().and_then(|x| x.to_str()) == Some("sql"))
            .collect(),
        Err(_) => Vec::new(),
    };
    files.sort();

    let mut count = 0;
    for path in files {
        let fname = path.file_name().unwrap().to_string_lossy().to_string();
        if applied.contains(&fname) {
            continue;
        }
        let sql = std::fs::read_to_string(&path)?;
        let checksum = checksum(&sql);
        // A comment-only file (all destructive ops skipped) is a no-op we still record.
        let has_sql = sql
            .lines()
            .any(|l| !l.trim().is_empty() && !l.trim_start().starts_with("--"));

        let tx = client.transaction().await?;
        if has_sql {
            tx.batch_execute(&sql).await?;
        }
        tx.execute(
            "INSERT INTO _roundabout_migrations (filename, checksum) VALUES ($1, $2)",
            &[&fname, &checksum],
        )
        .await?;
        tx.commit().await?;
        println!("[roundabout_orm] applied migration {fname}");
        count += 1;
    }
    Ok(count)
}

/// Wrap a destructive statement: run it live when enabled, otherwise turn it
/// into a documenting comment and warn.
fn gate(destructive: bool, sql: String, desc: &str) -> String {
    if destructive {
        sql
    } else {
        eprintln!(
            "[roundabout_orm] WARNING: skipped destructive migration ({desc}); \
             set ROUND_ABOUT_DESTRUCTIVE=1 to apply"
        );
        format!("-- [skipped] {desc} (set ROUND_ABOUT_DESTRUCTIVE=1 to apply): {sql}")
    }
}

fn load_snapshot(dir: &Path) -> Vec<EntityDef> {
    std::fs::read_to_string(dir.join(SNAPSHOT_FILE))
        .ok()
        .and_then(|s| serde_json::from_str(&s).ok())
        .unwrap_or_default()
}

fn write_snapshot(dir: &Path, manifest: &[EntityDef]) {
    match serde_json::to_string_pretty(manifest) {
        Ok(json) => {
            if let Err(e) = std::fs::write(dir.join(SNAPSHOT_FILE), json) {
                eprintln!("[roundabout_orm] cannot write snapshot: {e}");
            }
        }
        Err(e) => eprintln!("[roundabout_orm] cannot serialize snapshot: {e}"),
    }
}

fn checksum(s: &str) -> String {
    let mut h = DefaultHasher::new();
    s.hash(&mut h);
    format!("{:016x}", h.finish())
}

fn now_millis() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use roundabout_orm_schema::FieldDef;

    fn field(name: &str, sql: &str, pk: bool) -> FieldDef {
        FieldDef {
            name: name.into(),
            sql_type: sql.into(),
            rust_type: "x".into(),
            read: true,
            update: true,
            ws: false,
            primary_key: pk,
            default_expr: None,
        }
    }

    fn products(fields: Vec<FieldDef>) -> Vec<EntityDef> {
        vec![EntityDef {
            name: "Product".into(),
            table: "products".into(),
            fields,
            relations: vec![],
        }]
    }

    /// A unique temp dir per test (nanosecond clock keeps parallel tests apart).
    fn tmp_dir() -> PathBuf {
        let nanos = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
        let d = std::env::temp_dir().join(format!("ra_mig_test_{nanos}"));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    #[test]
    fn first_generate_creates_table_then_idempotent() {
        let dir = tmp_dir();
        let m = products(vec![field("id", "BIGINT", true), field("name", "TEXT", false)]);
        let path = generate(&m, &dir).expect("first run should generate a migration");
        let sql = std::fs::read_to_string(&path).unwrap();
        assert!(sql.contains("CREATE TABLE IF NOT EXISTS products"));
        // No schema change → no new migration.
        assert!(generate(&m, &dir).is_none());
    }

    #[test]
    fn added_column_is_additive() {
        let dir = tmp_dir();
        generate(&products(vec![field("id", "BIGINT", true)]), &dir).unwrap();
        let path = generate(
            &products(vec![field("id", "BIGINT", true), field("name", "TEXT", false)]),
            &dir,
        )
        .unwrap();
        let sql = std::fs::read_to_string(&path).unwrap();
        assert!(sql.contains("ALTER TABLE products ADD COLUMN IF NOT EXISTS name TEXT"));
    }

    /// Default (no `ROUND_ABOUT_DESTRUCTIVE`) must never emit a live DROP.
    #[test]
    fn dropped_column_is_skipped_by_default() {
        let dir = tmp_dir();
        generate(
            &products(vec![field("id", "BIGINT", true), field("legacy", "TEXT", false)]),
            &dir,
        )
        .unwrap();
        let path = generate(&products(vec![field("id", "BIGINT", true)]), &dir).unwrap();
        let sql = std::fs::read_to_string(&path).unwrap();
        assert!(sql.contains("-- [skipped]"));
        assert!(sql.contains("drop column products.legacy"));
        // The DROP appears only inside the comment, never as a live statement.
        let live_drop = sql
            .lines()
            .any(|l| !l.trim_start().starts_with("--") && l.contains("DROP COLUMN"));
        assert!(!live_drop, "destructive DROP must not run by default");
    }
}
