use bollard::container::{Config, CreateContainerOptions, ListContainersOptions, StartContainerOptions};
use bollard::exec::{CreateExecOptions, StartExecResults};
use bollard::image::CreateImageOptions;
use bollard::models::{HostConfig, PortBinding, Mount, MountTypeEnum, ContainerSummary};
use bollard::Docker;
use futures_util::StreamExt;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;

const PG_CONTAINER_NAME: &str = "aws-local-postgres";
const PG_VOLUME_NAME: &str = "aws-local-pgdata";
const PG_SUPERUSER: &str = "aws_local_admin";
const PG_SUPERUSER_PASSWORD: &str = "aws_local_admin_password";

const MSSQL_CONTAINER_NAME: &str = "aws-local-mssql";
const MSSQL_VOLUME_NAME: &str = "aws-local-mssqldata";
const MSSQL_SA_PASSWORD: &str = "Str0ngP@ssw0rd!";

const NETWORK_NAME: &str = "localdev";

/// Database engine variant that determines which container and SQL dialect to use.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DbEngine {
    Postgres { major: String },
    Mssql { year: String },
}

/// Map an RDS `Engine` + `EngineVersion` pair to a `DbEngine`.
///
/// - `"postgres"` / `"aurora-postgresql"` → `DbEngine::Postgres`
/// - `"sqlserver-ee"`, `"sqlserver-se"`, `"sqlserver-ex"`, `"sqlserver-web"` → `DbEngine::Mssql`
pub fn engine_from_params(engine: &str, version: &str) -> DbEngine {
    if engine.starts_with("sqlserver") {
        DbEngine::Mssql { year: parse_mssql_year(version) }
    } else {
        DbEngine::Postgres { major: parse_pg_major(version) }
    }
}

#[derive(Clone)]
pub struct DockerManager {
    client: Docker,
    postgres_port: Arc<Mutex<Option<u16>>>,
    mssql_port: Arc<Mutex<Option<u16>>>,
}

impl DockerManager {
    pub fn new() -> Result<Self, bollard::errors::Error> {
        let client = Docker::connect_with_local_defaults()?;
        Ok(Self {
            client,
            postgres_port: Arc::new(Mutex::new(None)),
            mssql_port: Arc::new(Mutex::new(None)),
        })
    }

    // ── public high-level API ───────────────────────────────────────

    /// Ensure the container for the given engine is running and return
    /// the host port that maps to it.
    pub async fn ensure_container(&self, engine: &DbEngine) -> Result<u16, String> {
        match engine {
            DbEngine::Postgres { major } => self.ensure_postgres_inner(major).await,
            DbEngine::Mssql { year } => self.ensure_mssql_inner(year).await,
        }
    }

    /// Keep the old name around so the integration test still compiles.
    pub async fn ensure_postgres(&self, engine_version: Option<&str>) -> Result<u16, String> {
        let major = parse_pg_major(engine_version.unwrap_or("16"));
        self.ensure_postgres_inner(&major).await
    }

    /// Create a database + user using the correct dialect for `engine`.
    pub async fn create_database_for_engine(
        &self,
        engine: &DbEngine,
        port: u16,
        db_name: &str,
        username: &str,
        password: &str,
    ) -> Result<(), String> {
        match engine {
            DbEngine::Postgres { .. } => {
                self.create_postgres_database(port, db_name, username, password).await
            }
            DbEngine::Mssql { .. } => {
                self.create_mssql_database(db_name, username, password).await
            }
        }
    }

    /// Drop a database + user using the correct dialect for `engine`.
    pub async fn drop_database_for_engine(
        &self,
        engine: &DbEngine,
        port: u16,
        db_name: &str,
        username: &str,
    ) -> Result<(), String> {
        match engine {
            DbEngine::Postgres { .. } => {
                self.drop_postgres_database(port, db_name, username).await
            }
            DbEngine::Mssql { .. } => {
                self.drop_mssql_database(db_name, username).await
            }
        }
    }

    /// Legacy create_database kept for backward compat (Postgres only).
    pub async fn create_database(
        &self,
        port: u16,
        db_name: &str,
        username: &str,
        password: &str,
    ) -> Result<(), String> {
        self.create_postgres_database(port, db_name, username, password).await
    }

    /// Legacy drop_database kept for backward compat (Postgres only).
    pub async fn drop_database(
        &self,
        port: u16,
        db_name: &str,
        username: &str,
    ) -> Result<(), String> {
        self.drop_postgres_database(port, db_name, username).await
    }

    // ── Postgres internals ──────────────────────────────────────────

    async fn ensure_postgres_inner(&self, major: &str) -> Result<u16, String> {
        {
            let cached = self.postgres_port.lock().await;
            if let Some(p) = *cached {
                return Ok(p);
            }
        }

        let mut port_lock = self.postgres_port.lock().await;
        // Double-check after acquiring lock
        if let Some(p) = *port_lock {
            return Ok(p);
        }

        let p = self.start_or_create_container(
            PG_CONTAINER_NAME,
            &format!("postgres:{major}"),
            5432,
            "/var/lib/postgresql/data",
            PG_VOLUME_NAME,
            vec![
                format!("POSTGRES_USER={PG_SUPERUSER}"),
                format!("POSTGRES_PASSWORD={PG_SUPERUSER_PASSWORD}"),
            ],
            "Postgres",
        ).await?;

        *port_lock = Some(p);
        Ok(p)
    }

    /// Get a tokio-postgres connection to the superuser database.
    pub async fn connect_admin(&self, port: u16) -> Result<tokio_postgres::Client, String> {
        let conn_str = format!(
            "host=127.0.0.1 port={port} user={PG_SUPERUSER} password={PG_SUPERUSER_PASSWORD} dbname=postgres"
        );
        let (client, connection) = tokio_postgres::connect(&conn_str, tokio_postgres::NoTls)
            .await
            .map_err(|e| format!("Postgres connect error: {e}"))?;

        tokio::spawn(async move {
            if let Err(e) = connection.await {
                tracing::error!("Postgres connection error: {e}");
            }
        });

        Ok(client)
    }

    async fn create_postgres_database(
        &self,
        port: u16,
        db_name: &str,
        username: &str,
        password: &str,
    ) -> Result<(), String> {
        let client = self.connect_admin(port).await?;

        let create_user = format!(
            "CREATE USER \"{}\" WITH PASSWORD '{}'",
            escape_ident(username),
            escape_literal(password),
        );
        if let Err(e) = client.execute(&create_user, &[]).await {
            let msg = e.to_string();
            if !msg.contains("already exists") {
                return Err(format!("CREATE USER failed: {e}"));
            }
        }

        let create_db = format!(
            "CREATE DATABASE \"{}\" OWNER \"{}\"",
            escape_ident(db_name),
            escape_ident(username),
        );
        if let Err(e) = client.execute(&create_db, &[]).await {
            let msg = e.to_string();
            if !msg.contains("already exists") {
                return Err(format!("CREATE DATABASE failed: {e}"));
            }
        }

        Ok(())
    }

    async fn drop_postgres_database(
        &self,
        port: u16,
        db_name: &str,
        username: &str,
    ) -> Result<(), String> {
        let client = self.connect_admin(port).await?;

        let terminate = format!(
            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{}'",
            escape_literal(db_name),
        );
        let _ = client.execute(&terminate, &[]).await;

        let drop_db = format!("DROP DATABASE IF EXISTS \"{}\"", escape_ident(db_name));
        client.execute(&drop_db, &[]).await
            .map_err(|e| format!("DROP DATABASE failed: {e}"))?;

        let drop_user = format!("DROP USER IF EXISTS \"{}\"", escape_ident(username));
        let _ = client.execute(&drop_user, &[]).await;

        Ok(())
    }

    // ── MSSQL internals ─────────────────────────────────────────────

    async fn ensure_mssql_inner(&self, year: &str) -> Result<u16, String> {
        {
            let cached = self.mssql_port.lock().await;
            if let Some(p) = *cached {
                return Ok(p);
            }
        }

        let mut port_lock = self.mssql_port.lock().await;
        // Double-check after acquiring lock
        if let Some(p) = *port_lock {
            return Ok(p);
        }

        let p = self.start_or_create_container(
            MSSQL_CONTAINER_NAME,
            &format!("mcr.microsoft.com/mssql/server:{year}-latest"),
            1433,
            "/var/opt/mssql",
            MSSQL_VOLUME_NAME,
            vec![
                "ACCEPT_EULA=Y".to_string(),
                format!("MSSQL_SA_PASSWORD={MSSQL_SA_PASSWORD}"),
                "MSSQL_PID=Developer".to_string(),
            ],
            "MSSQL",
        ).await?;

        *port_lock = Some(p);
        Ok(p)
    }

    async fn create_mssql_database(
        &self,
        db_name: &str,
        username: &str,
        password: &str,
    ) -> Result<(), String> {
        let sql = format!(
            "CREATE DATABASE [{db_name}]; \
             CREATE LOGIN [{username}] WITH PASSWORD='{password}'; \
             USE [{db_name}]; \
             CREATE USER [{username}] FOR LOGIN [{username}]; \
             ALTER ROLE db_owner ADD MEMBER [{username}];",
            db_name = escape_mssql_bracket(db_name),
            username = escape_mssql_bracket(username),
            password = escape_literal(password),
        );

        self.exec_sqlcmd(&sql).await
    }

    async fn drop_mssql_database(
        &self,
        db_name: &str,
        username: &str,
    ) -> Result<(), String> {
        let sql = format!(
            "USE [master]; \
             ALTER DATABASE [{db_name}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; \
             DROP DATABASE [{db_name}]; \
             DROP LOGIN [{username}];",
            db_name = escape_mssql_bracket(db_name),
            username = escape_mssql_bracket(username),
        );

        self.exec_sqlcmd(&sql).await
    }

    /// Run a T-SQL statement inside the MSSQL container via `sqlcmd`.
    async fn exec_sqlcmd(&self, sql: &str) -> Result<(), String> {
        let cmd_args = vec![
            "/opt/mssql-tools18/bin/sqlcmd".to_string(),
            "-S".to_string(), "localhost".to_string(),
            "-U".to_string(), "sa".to_string(),
            "-P".to_string(), MSSQL_SA_PASSWORD.to_string(),
            "-C".to_string(),
            "-Q".to_string(), sql.to_string(),
        ];

        let exec = self.client
            .create_exec(
                MSSQL_CONTAINER_NAME,
                CreateExecOptions {
                    cmd: Some(cmd_args.iter().map(|s| s.as_str()).collect()),
                    attach_stdout: Some(true),
                    attach_stderr: Some(true),
                    ..Default::default()
                },
            )
            .await
            .map_err(|e| format!("docker exec create error: {e}"))?;

        let output = self.client
            .start_exec(&exec.id, None)
            .await
            .map_err(|e| format!("docker exec start error: {e}"))?;

        if let StartExecResults::Attached { mut output, .. } = output {
            let mut stdout = String::new();
            while let Some(chunk) = output.next().await {
                match chunk {
                    Ok(msg) => stdout.push_str(&msg.to_string()),
                    Err(e) => tracing::warn!("sqlcmd output read error: {e}"),
                }
            }
            // sqlcmd returns "Msg " for errors
            if stdout.contains("Msg ") && !stdout.contains("already exists") {
                return Err(format!("sqlcmd error: {stdout}"));
            }
        }

        Ok(())
    }

    // ── shared container lifecycle ──────────────────────────────────

    /// Check if a container exists and start it, or create a fresh one.
    /// Returns the host port mapped to `container_port`.
    async fn start_or_create_container(
        &self,
        container_name: &str,
        image: &str,
        container_port: u16,
        data_path: &str,
        volume_name: &str,
        env_vars: Vec<String>,
        label: &str,
    ) -> Result<u16, String> {
        let port_key = format!("{container_port}/tcp");

        // Check if container already exists
        let filters = HashMap::from([
            ("name".to_string(), vec![container_name.to_string()]),
        ]);
        let containers = self.client
            .list_containers(Some(ListContainersOptions {
                all: true,
                filters,
                ..Default::default()
            }))
            .await
            .map_err(|e| format!("Docker list error: {e}"))?;

        if let Some(container) = containers.first() {
            let state = container.state.as_deref().unwrap_or("");
            if state == "running" {
                let p = extract_host_port(container, container_port)?;
                return Ok(p);
            }
            self.client
                .start_container(container_name, None::<StartContainerOptions<String>>)
                .await
                .map_err(|e| format!("Docker start error: {e}"))?;
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            let p = self.inspect_port(container_name, &port_key).await?;
            return Ok(p);
        }

        // Container doesn't exist — pull image, create, and start
        tracing::info!(image = %image, "pulling {label} image");

        let mut pull = self.client.create_image(
            Some(CreateImageOptions { from_image: image, ..Default::default() }),
            None,
            None,
        );
        while let Some(result) = pull.next().await {
            if let Err(e) = result {
                tracing::warn!("pull warning: {e}");
            }
        }

        let host_config = HostConfig {
            port_bindings: Some(HashMap::from([(
                port_key.clone(),
                Some(vec![PortBinding {
                    host_ip: Some("0.0.0.0".to_string()),
                    host_port: Some("0".to_string()), // dynamic port
                }]),
            )])),
            mounts: Some(vec![Mount {
                target: Some(data_path.to_string()),
                source: Some(volume_name.to_string()),
                typ: Some(MountTypeEnum::VOLUME),
                ..Default::default()
            }]),
            network_mode: Some(NETWORK_NAME.to_string()),
            ..Default::default()
        };

        let env_refs: Vec<&str> = env_vars.iter().map(|s| s.as_str()).collect();

        let config = Config {
            image: Some(image),
            env: Some(env_refs),
            host_config: Some(host_config),
            exposed_ports: Some(HashMap::from([(port_key.as_str(), HashMap::new())])),
            ..Default::default()
        };

        self.client
            .create_container(
                Some(CreateContainerOptions { name: container_name, ..Default::default() }),
                config,
            )
            .await
            .map_err(|e| format!("Docker create error: {e}"))?;

        self.client
            .start_container(container_name, None::<StartContainerOptions<String>>)
            .await
            .map_err(|e| format!("Docker start error: {e}"))?;

        tracing::info!("waiting for {label} to be ready");
        tokio::time::sleep(std::time::Duration::from_secs(3)).await;

        let p = self.inspect_port(container_name, &port_key).await?;
        tracing::info!(port = p, "{label} container ready");
        Ok(p)
    }

    /// Inspect a running container and extract the host port for a given
    /// container port key (e.g. `"5432/tcp"`).
    async fn inspect_port(&self, container_name: &str, port_key: &str) -> Result<u16, String> {
        let info = self.client
            .inspect_container(container_name, None)
            .await
            .map_err(|e| format!("Docker inspect error: {e}"))?;

        let bindings = info
            .network_settings
            .and_then(|ns| ns.ports)
            .ok_or("no port bindings found")?;

        let port_vec = bindings
            .get(port_key)
            .and_then(|v| v.as_ref())
            .ok_or_else(|| format!("{port_key} not bound"))?;

        port_vec
            .first()
            .and_then(|pb| pb.host_port.as_ref())
            .and_then(|p| p.parse::<u16>().ok())
            .ok_or_else(|| "could not parse host port".to_string())
    }
}

fn extract_host_port(container: &ContainerSummary, container_port: u16) -> Result<u16, String> {
    container
        .ports
        .as_ref()
        .and_then(|ports| {
            ports.iter().find(|p| p.private_port == container_port)
                .and_then(|p| p.public_port)
        })
        .map(|p| p as u16)
        .ok_or_else(|| format!("container has no mapped port for {container_port}"))
}

fn parse_pg_major(engine_version: &str) -> String {
    engine_version.split('.').next().unwrap_or("16").to_string()
}

/// Map an MSSQL `EngineVersion` to the Docker image year tag.
///
/// AWS RDS uses versions like `15.00.4123.1.v1` (SQL Server 2019) and
/// `16.00.4175.1.v1` (SQL Server 2022).  We only care about the major
/// number: 15 → 2019, 16 → 2022.
fn parse_mssql_year(engine_version: &str) -> String {
    let major = engine_version.split('.').next().unwrap_or("");
    match major {
        "15" => "2019".to_string(),
        "16" => "2022".to_string(),
        _ => "2022".to_string(),
    }
}

fn escape_ident(s: &str) -> String {
    s.replace('"', "\"\"")
}

fn escape_literal(s: &str) -> String {
    s.replace('\'', "''")
}

/// Escape a value for use inside MSSQL bracket-quoted identifiers.
/// The only character that needs escaping inside `[…]` is `]`, which
/// is doubled to `]]`.
fn escape_mssql_bracket(s: &str) -> String {
    s.replace(']', "]]")
}

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

    #[test]
    fn parse_pg_major_extracts_major() {
        assert_eq!(parse_pg_major("15.4"), "15");
        assert_eq!(parse_pg_major("16"), "16");
    }

    #[test]
    fn parse_mssql_year_maps_known_versions() {
        assert_eq!(parse_mssql_year("15.00.4123.1.v1"), "2019");
        assert_eq!(parse_mssql_year("15.0"), "2019");
        assert_eq!(parse_mssql_year("16.00.4175.1.v1"), "2022");
        assert_eq!(parse_mssql_year("16.0"), "2022");
    }

    #[test]
    fn parse_mssql_year_defaults_to_2022_for_unknown() {
        assert_eq!(parse_mssql_year("14.0"), "2022");
        assert_eq!(parse_mssql_year(""), "2022");
    }

    #[test]
    fn engine_from_params_postgres() {
        assert_eq!(
            engine_from_params("postgres", "15.4"),
            DbEngine::Postgres { major: "15".to_string() },
        );
        assert_eq!(
            engine_from_params("aurora-postgresql", "16"),
            DbEngine::Postgres { major: "16".to_string() },
        );
    }

    #[test]
    fn engine_from_params_mssql() {
        assert_eq!(
            engine_from_params("sqlserver-ee", "16.0"),
            DbEngine::Mssql { year: "2022".to_string() },
        );
        assert_eq!(
            engine_from_params("sqlserver-se", "15.0"),
            DbEngine::Mssql { year: "2019".to_string() },
        );
        assert_eq!(
            engine_from_params("sqlserver-ex", "15.00.4123.1.v1"),
            DbEngine::Mssql { year: "2019".to_string() },
        );
        assert_eq!(
            engine_from_params("sqlserver-web", ""),
            DbEngine::Mssql { year: "2022".to_string() },
        );
    }

    #[test]
    fn escape_ident_handles_quotes() {
        assert_eq!(escape_ident("my\"db"), "my\"\"db");
        assert_eq!(escape_ident("normal"), "normal");
    }

    #[test]
    fn escape_literal_handles_quotes() {
        assert_eq!(escape_literal("pass'word"), "pass''word");
    }

    #[test]
    fn escape_mssql_bracket_handles_closing_bracket() {
        assert_eq!(escape_mssql_bracket("my]db"), "my]]db");
        assert_eq!(escape_mssql_bracket("normal"), "normal");
    }
}
