//! Integration tests for the `sqlshape` command.

use assert_cmd::Command;
use duckdb::Connection as DuckConnection;
use mysql::prelude::Queryable as _;
use predicates::prelude::*;
use rusqlite::Connection as SqliteConnection;
use serde_json::Value;
use tempfile::tempdir;

fn cargo_command() -> Command {
    Command::cargo_bin("sqlshape").expect("binary")
}

#[test]
fn help_includes_engines_diff_and_shared_flags() {
    let mut command = cargo_command();
    command
        .arg("--help")
        .assert()
        .success()
        .stdout(predicate::str::contains("sqlshape [OPTIONS] --url"))
        .stdout(predicate::str::contains("sqlshape [OPTIONS] diff"))
        .stdout(predicate::str::contains("--format <FORMAT>"))
        .stdout(predicate::str::contains("--engine <ENGINE>"))
        .stdout(predicate::str::contains("--before-url <CONNECTION>"))
        .stdout(predicate::str::contains("--include-system"))
        .stdout(predicate::str::contains("postgres"))
        .stdout(predicate::str::contains("duckdb"));
}

#[test]
fn ambiguous_path_requires_engine() {
    let mut command = cargo_command();
    command
        .args(["--url", r".\data\app.db"])
        .assert()
        .code(2)
        .stderr(predicate::str::contains("--engine"))
        .stderr(predicate::str::contains("ambiguous"));
}

#[test]
fn invalid_engine_errors_do_not_echo_password() {
    let mut command = cargo_command();
    command
        .args([
            "--engine",
            "oracle",
            "--url",
            "postgres://user:secret@localhost/db",
        ])
        .assert()
        .code(2)
        .stderr(predicate::str::contains("oracle"))
        .stderr(predicate::str::contains("secret").not());
}

#[test]
fn summarizes_sqlite_schema_as_json() {
    let temp = tempdir().expect("tempdir");
    let path = temp.path().join("sample.db");
    let connection = SqliteConnection::open(&path).expect("db");
    connection
        .execute_batch(
            "PRAGMA foreign_keys = ON;
             CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT NOT NULL, city TEXT DEFAULT 'Paris');
             CREATE TABLE orders(id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL,
                total REAL, FOREIGN KEY(user_id) REFERENCES users(id));
             CREATE INDEX idx_orders_user_id ON orders(user_id);",
        )
        .expect("schema");
    drop(connection);

    let mut command = cargo_command();
    command
        .args(["--engine", "sqlite", "--url"])
        .arg(&path)
        .arg("--json")
        .assert()
        .success()
        .stdout(predicate::str::contains("\"engine\":\"sqlite\""))
        .stdout(predicate::str::contains("\"source_redacted\""))
        .stdout(predicate::str::contains("\"schema\":\"main\""))
        .stdout(predicate::str::contains("\"name\":\"users\""))
        .stdout(predicate::str::contains("\"primary_key\":[\"id\"]"))
        .stdout(predicate::str::contains("\"foreign_keys\""))
        .stdout(predicate::str::contains("\"idx_orders_user_id\""));
}

#[test]
fn infers_sqlite_engine_from_url_scheme() {
    let temp = tempdir().expect("tempdir");
    let path = temp.path().join("scheme.db");
    let connection = SqliteConnection::open(&path).expect("db");
    connection
        .execute_batch("CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT NOT NULL);")
        .expect("schema");
    drop(connection);

    let url = sqlite_url(&path);
    let mut command = cargo_command();
    command
        .args(["--url", &url, "--json"])
        .assert()
        .success()
        .stdout(predicate::str::contains("\"engine\":\"sqlite\""))
        .stdout(predicate::str::contains("\"name\":\"users\""));
}

#[test]
fn supports_format_json_and_toon_flags() {
    let temp = tempdir().expect("tempdir");
    let path = temp.path().join("formats.db");
    let connection = SqliteConnection::open(&path).expect("db");
    connection
        .execute_batch("CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT NOT NULL);")
        .expect("schema");
    drop(connection);

    let mut json_command = cargo_command();
    json_command
        .args(["--engine", "sqlite", "--url"])
        .arg(&path)
        .args(["--format", "json"])
        .assert()
        .success()
        .stdout(predicate::str::contains("\"engine\":\"sqlite\""));

    let mut toon_command = cargo_command();
    toon_command
        .args(["--engine", "sqlite", "--url"])
        .arg(&path)
        .arg("--toon")
        .assert()
        .success()
        .stdout(predicate::str::contains("engine: sqlite"))
        .stdout(predicate::str::contains("tables:"));
}

#[test]
fn summarizes_duckdb_schema_as_json() {
    let temp = tempdir().expect("tempdir");
    let path = temp.path().join("warehouse.duckdb");
    let connection = DuckConnection::open(&path).expect("db");
    connection
        .execute_batch(
            "CREATE TABLE users(id INTEGER PRIMARY KEY, name VARCHAR NOT NULL);
             CREATE VIEW active_users AS SELECT id, name FROM users;",
        )
        .expect("schema");
    drop(connection);

    let mut command = cargo_command();
    command
        .args(["--engine", "duckdb", "--url"])
        .arg(&path)
        .arg("--json")
        .assert()
        .success()
        .stdout(predicate::str::contains("\"engine\":\"duckdb\""))
        .stdout(predicate::str::contains("\"schema\":\"main\""))
        .stdout(predicate::str::contains("\"name\":\"users\""))
        .stdout(predicate::str::contains("\"kind\":\"table\""))
        .stdout(predicate::str::contains("\"name\":\"active_users\""))
        .stdout(predicate::str::contains("\"kind\":\"view\""));
}

#[test]
fn diff_reports_table_and_column_changes() {
    let temp = tempdir().expect("tempdir");
    let before_path = temp.path().join("before.db");
    let after_path = temp.path().join("after.db");

    let before = SqliteConnection::open(&before_path).expect("before db");
    before
        .execute_batch("CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT NOT NULL);")
        .expect("before schema");
    drop(before);

    let after = SqliteConnection::open(&after_path).expect("after db");
    after
        .execute_batch(
            "CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT NOT NULL, city TEXT);
             CREATE TABLE events(id INTEGER PRIMARY KEY, kind TEXT);",
        )
        .expect("after schema");
    drop(after);

    let mut command = cargo_command();
    command
        .args(["diff", "--before-engine", "sqlite", "--before-url"])
        .arg(&before_path)
        .args(["--after-engine", "sqlite", "--after-url"])
        .arg(&after_path)
        .arg("--json")
        .assert()
        .success()
        .stdout(predicate::str::contains("\"added_tables\""))
        .stdout(predicate::str::contains("\"name\":\"events\""))
        .stdout(predicate::str::contains("\"changed_tables\""))
        .stdout(predicate::str::contains("\"added_columns\""))
        .stdout(predicate::str::contains("\"city\""));
}

#[test]
fn live_postgres_smoke_when_configured() {
    let Some(database_url) = env_url("SQLSHAPE_POSTGRES_URL") else {
        return;
    };
    let schema = format!("sqlshape_smoke_{}", std::process::id());
    let tls = postgres_native_tls::MakeTlsConnector::new(
        native_tls::TlsConnector::builder()
            .danger_accept_invalid_certs(true)
            .build()
            .expect("tls connector"),
    );
    let mut client = postgres::Client::connect(&database_url, tls).expect("postgres connection");
    client
        .batch_execute(&format!(
            "DROP SCHEMA IF EXISTS {schema} CASCADE;
             CREATE SCHEMA {schema};
             CREATE TABLE {schema}.users(id INTEGER PRIMARY KEY, name TEXT NOT NULL);
             CREATE TABLE {schema}.orders(
                id INTEGER PRIMARY KEY,
                user_id INTEGER NOT NULL REFERENCES {schema}.users(id)
             );
             CREATE INDEX idx_orders_user_id ON {schema}.orders(user_id);"
        ))
        .expect("postgres fixture");
    drop(client);

    let report = run_sqlshape_json(&[
        "sqlshape",
        "--engine",
        "postgres",
        "--url",
        &database_url,
        "--schema",
        &schema,
        "--json",
    ]);
    assert_report_has_table(&report, "users");
    assert_report_has_table(&report, "orders");
    assert_report_has_index(&report, "orders", "idx_orders_user_id");
    assert_report_has_foreign_key(&report, "orders");

    let tls = postgres_native_tls::MakeTlsConnector::new(
        native_tls::TlsConnector::builder()
            .danger_accept_invalid_certs(true)
            .build()
            .expect("tls connector"),
    );
    let mut client = postgres::Client::connect(&database_url, tls).expect("postgres cleanup");
    client
        .batch_execute(&format!("DROP SCHEMA IF EXISTS {schema} CASCADE;"))
        .expect("postgres cleanup");
}

#[test]
fn live_mysql_smoke_when_configured() {
    let Some(database_url) = env_url("SQLSHAPE_MYSQL_URL") else {
        return;
    };
    let suffix = std::process::id();
    let users = format!("sqlshape_users_{suffix}");
    let orders = format!("sqlshape_orders_{suffix}");
    let index = format!("idx_sqlshape_orders_user_id_{suffix}");
    let opts = mysql::Opts::from_url(&database_url).expect("mysql url");
    let pool = mysql::Pool::new(opts).expect("mysql pool");
    let mut connection = pool.get_conn().expect("mysql connection");
    connection
        .query_drop(format!("DROP TABLE IF EXISTS `{orders}`"))
        .expect("drop orders");
    connection
        .query_drop(format!("DROP TABLE IF EXISTS `{users}`"))
        .expect("drop users");
    connection
        .query_drop(format!(
            "CREATE TABLE `{users}`(id INT NOT NULL PRIMARY KEY, name VARCHAR(64) NOT NULL) ENGINE=InnoDB"
        ))
        .expect("create users");
    connection
        .query_drop(format!(
            "CREATE TABLE `{orders}`(
                id INT NOT NULL PRIMARY KEY,
                user_id INT NOT NULL,
                CONSTRAINT fk_{orders}_users FOREIGN KEY (user_id) REFERENCES `{users}`(id)
             ) ENGINE=InnoDB"
        ))
        .expect("create orders");
    connection
        .query_drop(format!("CREATE INDEX `{index}` ON `{orders}`(user_id)"))
        .expect("create index");
    drop(connection);

    let report = run_sqlshape_json(&[
        "sqlshape",
        "--engine",
        "mysql",
        "--url",
        &database_url,
        "--table",
        &users,
        "--table",
        &orders,
        "--json",
    ]);
    assert_report_has_table(&report, &users);
    assert_report_has_table(&report, &orders);
    assert_report_has_index(&report, &orders, &index);
    assert_report_has_foreign_key(&report, &orders);

    let mut connection = pool.get_conn().expect("mysql cleanup connection");
    connection
        .query_drop(format!("DROP TABLE IF EXISTS `{orders}`"))
        .expect("drop orders");
    connection
        .query_drop(format!("DROP TABLE IF EXISTS `{users}`"))
        .expect("drop users");
}

#[test]
fn live_sqlserver_smoke_when_configured() {
    let Some(database_url) = env_url("SQLSHAPE_MSSQL_URL") else {
        return;
    };
    let schema = format!("sqlshape_smoke_{}", std::process::id());
    let create_sql = format!(
        "IF SCHEMA_ID(N'{schema}') IS NOT NULL EXEC(N'DROP SCHEMA {schema}');
         EXEC(N'CREATE SCHEMA {schema}');
         CREATE TABLE {schema}.users(id INT NOT NULL PRIMARY KEY, name NVARCHAR(64) NOT NULL);
         CREATE TABLE {schema}.orders(
            id INT NOT NULL PRIMARY KEY,
            user_id INT NOT NULL,
            CONSTRAINT fk_{schema}_orders_users FOREIGN KEY(user_id) REFERENCES {schema}.users(id)
         );
         CREATE INDEX idx_{schema}_orders_user_id ON {schema}.orders(user_id);"
    );
    sqlserver_batch(
        &database_url,
        &format!(
            "DROP TABLE IF EXISTS {schema}.orders; DROP TABLE IF EXISTS {schema}.users; IF SCHEMA_ID(N'{schema}') IS NOT NULL EXEC(N'DROP SCHEMA {schema}'); {create_sql}"
        ),
    );

    let report = run_sqlshape_json(&[
        "sqlshape",
        "--engine",
        "sqlserver",
        "--url",
        &database_url,
        "--schema",
        &schema,
        "--json",
    ]);
    assert_report_has_table(&report, "users");
    assert_report_has_table(&report, "orders");
    assert_report_has_index(&report, "orders", &format!("idx_{schema}_orders_user_id"));
    assert_report_has_foreign_key(&report, "orders");

    sqlserver_batch(
        &database_url,
        &format!(
            "DROP TABLE IF EXISTS {schema}.orders;
             DROP TABLE IF EXISTS {schema}.users;
             IF SCHEMA_ID(N'{schema}') IS NOT NULL EXEC(N'DROP SCHEMA {schema}');"
        ),
    );
}

fn env_url(name: &str) -> Option<String> {
    std::env::var(name)
        .ok()
        .filter(|value| !value.trim().is_empty())
}

fn sqlite_url(path: &std::path::Path) -> String {
    format!("sqlite:///{}", path.to_string_lossy().replace('\\', "/"))
}

fn run_sqlshape_json(args: &[&str]) -> Value {
    let source_url = args
        .windows(2)
        .find_map(|pair| (pair[0] == "--url").then_some(pair[1]))
        .unwrap_or("");
    let output = std::process::Command::new(env!("CARGO_BIN_EXE_sqlshape"))
        .args(&args[1..])
        .output()
        .expect("run sqlshape");
    let stdout = String::from_utf8_lossy(&output.stdout);
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        output.status.success(),
        "sqlshape failed with {}:\n{}",
        output.status,
        sanitize_output(&format!("{stdout}\n{stderr}"), source_url)
    );
    serde_json::from_str(&stdout).unwrap_or_else(|error| {
        panic!(
            "invalid JSON: {error}\n{}",
            sanitize_output(&stdout, source_url)
        )
    })
}

fn sanitize_output(text: &str, source_url: &str) -> String {
    let mut redacted = text.replace(source_url, "<redacted-url>");
    if let Ok(url) = url::Url::parse(source_url) {
        if let Some(password) = url.password() {
            redacted = redacted.replace(password, "***");
        }
    }
    redacted
}

fn assert_report_has_table(report: &Value, name: &str) {
    assert!(
        report["tables"]
            .as_array()
            .expect("tables array")
            .iter()
            .any(|table| table["name"] == name),
        "expected table {name} in {report}"
    );
}

fn assert_report_has_index(report: &Value, table_name: &str, index_name: &str) {
    let table = report["tables"]
        .as_array()
        .expect("tables array")
        .iter()
        .find(|table| table["name"] == table_name)
        .expect("table");
    assert!(
        table["indexes"]
            .as_array()
            .expect("indexes array")
            .iter()
            .any(|index| index["name"] == index_name),
        "expected index {index_name} on {table_name} in {report}"
    );
}

fn assert_report_has_foreign_key(report: &Value, table_name: &str) {
    let table = report["tables"]
        .as_array()
        .expect("tables array")
        .iter()
        .find(|table| table["name"] == table_name)
        .expect("table");
    assert!(
        !table["foreign_keys"]
            .as_array()
            .expect("foreign keys array")
            .is_empty(),
        "expected foreign key on {table_name} in {report}"
    );
}

fn sqlserver_batch(database_url: &str, sql: &str) {
    let runtime = tokio::runtime::Runtime::new().expect("sqlserver runtime");
    runtime
        .block_on(async {
            use tokio_util::compat::TokioAsyncWriteCompatExt as _;

            let ado = sqlserver_ado_string(database_url);
            let config = tiberius::Config::from_ado_string(&ado)?;
            let tcp = tokio::net::TcpStream::connect(config.get_addr()).await?;
            tcp.set_nodelay(true)?;
            let mut client = tiberius::Client::connect(config, tcp.compat_write()).await?;
            client.simple_query(sql).await?.into_results().await?;
            Ok::<_, Box<dyn std::error::Error>>(())
        })
        .unwrap_or_else(|error| panic!("sqlserver fixture failed: {error}"));
}

fn sqlserver_ado_string(raw: &str) -> String {
    let url = url::Url::parse(raw).expect("sqlserver url");
    let host = url.host_str().expect("sqlserver host");
    let server = url.port().map_or_else(
        || format!("tcp:{host}"),
        |port| format!("tcp:{host},{port}"),
    );
    let mut parts = vec![format!("server={server}")];
    if !url.username().is_empty() {
        parts.push(format!("User ID={}", url.username()));
    }
    if let Some(password) = url.password() {
        parts.push(format!("Password={password}"));
    }
    let database = url.path().trim_start_matches('/');
    if !database.is_empty() {
        parts.push(format!("Database={database}"));
    }
    let trust_cert = url
        .query_pairs()
        .any(|(key, value)| key.eq_ignore_ascii_case("trust_cert") && value == "true");
    parts.push(format!("TrustServerCertificate={trust_cert}"));
    parts.join(";")
}