//! [`MigrationRunner`] — execute migrations from one or more sources.

use crate::{
    migration::{Migration, RawMigration},
    schema::SchemaExecutor,
    source::MigrationSource,
};

// ── RawMigrationAdapter ───────────────────────────────────────────────────────

/// Bridge: wraps a [`RawMigration`] (sync SQL-string) into the async
/// [`Migration`] interface so the runner can treat both types uniformly.
struct RawMigrationAdapter(Box<dyn RawMigration>);

#[async_trait::async_trait]
impl Migration for RawMigrationAdapter {
    fn name(&self) -> &str {
        self.0.name()
    }

    async fn up(&self, schema: &SchemaExecutor) -> anyhow::Result<()> {
        schema.raw_execute(&self.0.up()).await
    }

    async fn down(&self, schema: &SchemaExecutor) -> anyhow::Result<()> {
        schema.raw_execute(&self.0.down()).await
    }
}

// ── MigrationRunner ───────────────────────────────────────────────────────────

/// Runs migrations from multiple sources against a PostgreSQL database.
///
/// Sources are applied in registration order.  A `_migrations` tracking table
/// records which migrations have already run so reruns are safe.
///
/// # Example
///
/// ```rust,ignore
/// use rok_orm_migrate::{MigrationRunner, FileSource, Migration, SchemaExecutor};
/// use async_trait::async_trait;
///
/// pub struct CreateUsersTable;
///
/// #[async_trait]
/// impl Migration for CreateUsersTable {
///     fn name(&self) -> &str { "2026_05_18_000001_create_users_table" }
///
///     async fn up(&self, schema: &SchemaExecutor) -> anyhow::Result<()> {
///         schema.create("users", |t| {
///             t.id();
///             t.string("email").not_null().unique();
///             t.string("name").not_null();
///             t.timestamps();
///         }).await
///     }
///
///     async fn down(&self, schema: &SchemaExecutor) -> anyhow::Result<()> {
///         schema.drop_table_if_exists("users").await
///     }
/// }
///
/// MigrationRunner::new(pool)
///     .source(rok_auth::migrations())          // embedded SQL migrations
///     .source(FileSource::new("./migrations")) // SQL file migrations
///     .migration(CreateUsersTable)             // schema-builder migration
///     .run()
///     .await?;
/// ```
pub struct MigrationRunner {
    pool: sqlx::PgPool,
    migrations: Vec<Box<dyn Migration>>,
}

impl MigrationRunner {
    pub fn new(pool: sqlx::PgPool) -> Self {
        Self {
            pool,
            migrations: Vec::new(),
        }
    }

    /// Register a [`MigrationSource`] (embedded SQL or file-based SQL).
    ///
    /// Each raw migration from the source is wrapped in [`RawMigrationAdapter`]
    /// so it can run alongside schema-builder migrations.
    pub fn source(mut self, s: impl MigrationSource + 'static) -> Self {
        for m in s.migrations() {
            self.migrations.push(Box::new(RawMigrationAdapter(m)));
        }
        self
    }

    /// Register a single schema-builder [`Migration`] inline.
    ///
    /// ```rust,ignore
    /// runner.migration(CreateUsersTable)
    /// ```
    pub fn migration(mut self, m: impl Migration + 'static) -> Self {
        self.migrations.push(Box::new(m));
        self
    }

    /// Register multiple schema-builder [`Migration`]s from a `Vec` or iterator.
    ///
    /// Useful when collecting migrations from a `database/migrations/mod.rs`:
    ///
    /// ```rust,ignore
    /// runner.migrations(vec![
    ///     Box::new(CreateUsersTable),
    ///     Box::new(CreatePostsTable),
    /// ])
    /// ```
    pub fn migrations(mut self, ms: impl IntoIterator<Item = Box<dyn Migration>>) -> Self {
        self.migrations.extend(ms);
        self
    }

    /// Register a raw-SQL [`RawMigration`] inline (backward-compatible).
    ///
    /// Prefer `.migration()` with the schema-builder trait for new code.
    pub fn raw_migration(mut self, m: impl RawMigration + 'static) -> Self {
        self.migrations
            .push(Box::new(RawMigrationAdapter(Box::new(m))));
        self
    }

    // ── private helpers ───────────────────────────────────────────────────────

    async fn ensure_table(&self) -> Result<(), sqlx::Error> {
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS _migrations (
                id     BIGSERIAL    PRIMARY KEY,
                name   TEXT         NOT NULL UNIQUE,
                batch  INTEGER      NOT NULL DEFAULT 1,
                run_at TIMESTAMPTZ  NOT NULL DEFAULT NOW()
            )",
        )
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn applied(&self) -> Result<Vec<String>, sqlx::Error> {
        let rows: Vec<(String,)> = sqlx::query_as("SELECT name FROM _migrations ORDER BY id")
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(|(n,)| n).collect())
    }

    async fn current_batch(&self) -> Result<i32, sqlx::Error> {
        let row: Option<(i32,)> = sqlx::query_as("SELECT COALESCE(MAX(batch), 0) FROM _migrations")
            .fetch_optional(&self.pool)
            .await?;
        Ok(row.map(|(b,)| b).unwrap_or(0))
    }

    // ── public API ────────────────────────────────────────────────────────────

    /// Apply all pending migrations.
    pub async fn run(&self) -> anyhow::Result<()> {
        self.ensure_table().await?;
        let schema = SchemaExecutor::new(self.pool.clone());
        let applied = self.applied().await?;
        let batch = self.current_batch().await? + 1;

        for m in &self.migrations {
            let name = m.name().to_string();
            if applied.contains(&name) {
                continue;
            }
            m.up(&schema).await?;
            sqlx::query("INSERT INTO _migrations (name, batch) VALUES ($1, $2)")
                .bind(&name)
                .bind(batch)
                .execute(&self.pool)
                .await?;
            println!("Migrated:  {name}");
        }
        Ok(())
    }

    /// Rollback the last batch of migrations.
    pub async fn rollback(&self) -> anyhow::Result<()> {
        self.ensure_table().await?;
        let schema = SchemaExecutor::new(self.pool.clone());
        let batch = self.current_batch().await?;
        if batch == 0 {
            println!("Nothing to rollback.");
            return Ok(());
        }

        let names: Vec<(String,)> =
            sqlx::query_as("SELECT name FROM _migrations WHERE batch = $1 ORDER BY id DESC")
                .bind(batch)
                .fetch_all(&self.pool)
                .await?;

        for (name,) in names {
            if let Some(m) = self.migrations.iter().find(|m| m.name() == name) {
                m.down(&schema).await?;
                sqlx::query("DELETE FROM _migrations WHERE name = $1")
                    .bind(&name)
                    .execute(&self.pool)
                    .await?;
                println!("Rolled back: {name}");
            }
        }
        Ok(())
    }

    /// Print the status of every migration (Applied / Pending).
    pub async fn status(&self) -> anyhow::Result<()> {
        self.ensure_table().await?;
        let applied = self.applied().await?;
        for m in &self.migrations {
            let status = if applied.contains(&m.name().to_string()) {
                "Applied "
            } else {
                "Pending "
            };
            println!("[{status}] {}", m.name());
        }
        Ok(())
    }
}
