use super::ClickhouseClient;
use sea_query::*;
use sea_query_clickhouse::*;

fn client() -> ClickhouseClient {
    ClickhouseClient::new()
}

fn ensure_table(client: &ClickhouseClient, table: &str, create_sql: &str) {
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
    client.execute(create_sql).unwrap();
}

/// Build CREATE TABLE SQL with the `test_db.` database prefix.
fn create_db_sql(table: &str, builder: ClickhouseTableCreate) -> String {
    let sql = builder.build();
    sql.replacen(
        &format!("CREATE TABLE `{table}`"),
        &format!("CREATE TABLE test_db.`{table}`"),
        1,
    )
}

/// Prefix the first backtick-quoted occurrence of `table` with `test_db.`.
fn with_db(sql: String, table: &str) -> String {
    sql.replacen(&format!("`{table}`"), &format!("test_db.`{table}`"), 1)
}

fn ensure_test_db(client: &ClickhouseClient) {
    let _ = client.execute("CREATE DATABASE IF NOT EXISTS test_db");
}

// ── Test 1: Basic MergeTree table ──────────────────────────────────────

#[test]
fn test_create_basic_table() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_1";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("aspect").double().not_null())
                .col(ColumnDef::new("image").string())
                .col(ColumnDef::new("created_at").date_time())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 2: All integer types ──────────────────────────────────────────

#[test]
fn test_create_all_integer_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_2";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("c_i8").tiny_integer())
                .col(ColumnDef::new("c_i16").small_integer())
                .col(ColumnDef::new("c_i32").integer())
                .col(ColumnDef::new("c_i64").big_integer())
                .col(ColumnDef::new("c_i128").int128())
                .col(ColumnDef::new("c_i256").int256())
                .col(ColumnDef::new("c_u8").tiny_unsigned())
                .col(ColumnDef::new("c_u16").small_unsigned())
                .col(ColumnDef::new("c_u32").unsigned())
                .col(ColumnDef::new("c_u64").big_unsigned())
                .col(ColumnDef::new("c_u128").uint128())
                .col(ColumnDef::new("c_u256").uint256())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["c_i32"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 3: All float types ────────────────────────────────────────────

#[test]
fn test_create_all_float_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_3";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_f32").float32())
                .col(ColumnDef::new("c_f64").float64())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 4: Decimal types ─────────────────────────────────────────────

#[test]
fn test_create_decimal_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_4";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_d32").decimal32(2))
                .col(ColumnDef::new("c_d64").decimal64(4))
                .col(ColumnDef::new("c_d128").decimal128(6))
                .col(ColumnDef::new("c_d256").decimal256(8))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 5: Date types ────────────────────────────────────────────────

#[test]
fn test_create_date_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_5";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_date").date())
                .col(ColumnDef::new("c_date32").date32())
                .col(ColumnDef::new("c_dt").date_time())
                .col(ColumnDef::new("c_dt_utc").date_time_tz("UTC"))
                .col(ColumnDef::new("c_dt64").date_time64(3))
                .col(ColumnDef::new("c_dt64_utc").date_time64_tz(6, "UTC"))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 6: String types ──────────────────────────────────────────────

#[test]
fn test_create_string_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_6";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_str").string())
                .col(ColumnDef::new("c_fixed").fixed_string(10))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 7: Special types (UUID, IPv4, IPv6, Bool, Map, Tuple) ───────

#[test]
fn test_create_special_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_7";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_uuid").uuid())
                .col(ColumnDef::new("c_ipv4").ipv4())
                .col(ColumnDef::new("c_ipv6").ipv6())
                .col(ColumnDef::new("c_bool").boolean())
                .col(ColumnDef::new("c_map").map("String", "String"))
                .col(ColumnDef::new("c_point").tuple(&["Float64", "Float64"]))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 8: Enum types ────────────────────────────────────────────────

#[test]
fn test_create_enum_types() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_8";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_enum8").enum8(&[("active", 1), ("inactive", 2)]))
                .col(ColumnDef::new("c_enum16").enum16(&[("red", 1), ("green", 2), ("blue", 3)]))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 9: Nullable types ────────────────────────────────────────────

#[test]
fn test_create_nullable_type() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_9";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_opt_str").clickhouse_nullable("String"))
                .col(ColumnDef::new("c_opt_u32").clickhouse_nullable("UInt32"))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 10: LowCardinality ───────────────────────────────────────────

#[test]
fn test_create_low_cardinality() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_10";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_status").low_cardinality("String"))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 11: AggregateFunction with AggregatingMergeTree ──────────────

#[test]
fn test_create_aggregate_function() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_11";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("c_sum").simple_aggregate_function("sum", "UInt64"))
                .to_owned(),
        )
        .engine("AggregatingMergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 12: CREATE TABLE IF NOT EXISTS ───────────────────────────────

#[test]
fn test_create_if_not_exists() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_12";
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));

    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .if_not_exists()
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("name").string())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );

    // First create — table does not exist
    client.execute(&sql).unwrap();

    // Second create with IF NOT EXISTS — should succeed without error
    client.execute(&sql).unwrap();

    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}

// ── Test 13: ReplacingMergeTree with engine params ────────────────────

#[test]
fn test_create_with_engine_params() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_13";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("aspect").double().not_null())
                .col(ColumnDef::new("ver").integer())
                .to_owned(),
        )
        .engine_with_params("ReplacingMergeTree", vec!["ver".to_string()])
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 14: PARTITION BY ─────────────────────────────────────────────

#[test]
fn test_create_with_partition_by() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_14";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("aspect").double().not_null())
                .col(ColumnDef::new("created_at").date_time())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"])
        .partition_by_expr(Expr::cust("toYYYYMM(created_at)")),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 15: PRIMARY KEY (different from ORDER BY) ────────────────────

#[test]
fn test_create_with_primary_key() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_15";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("aspect").double().not_null())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id", "aspect"])
        .primary_key(["id"]),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 16: TTL ──────────────────────────────────────────────────────

#[test]
fn test_create_with_ttl() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_16";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("created_at").date_time())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"])
        .ttl("created_at + INTERVAL 1 MONTH"),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 17: SETTINGS ─────────────────────────────────────────────────

#[test]
fn test_create_with_settings() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_17";
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"])
        .table_setting("index_granularity", "8192"),
    );
    ensure_table(&client, table, &sql);
}

// ── Test 18: DROP TABLE ───────────────────────────────────────────────

#[test]
fn test_drop_table() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_18";

    // Create the table first
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &create_sql);

    // Drop the table using the library builder
    let sql = with_db(
        Table::drop()
            .table(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        table,
    );
    client.execute(&sql).unwrap();
}

// ── Test 19: DROP TABLE IF EXISTS ─────────────────────────────────────

#[test]
fn test_drop_table_if_exists() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);

    // Drop a table that does not exist — should succeed without error
    let sql = with_db(
        Table::drop()
            .table(Alias::new("nonexistent_table"))
            .if_exists()
            .to_string(ClickhouseQueryBuilder),
        "nonexistent_table",
    );
    client.execute(&sql).unwrap();
}

// ── Test 20: ALTER TABLE ADD COLUMN ───────────────────────────────────

#[test]
fn test_alter_add_column() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_19";

    // Create a basic table without the new column
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("name").string())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &create_sql);

    // ALTER TABLE ADD COLUMN
    let sql = with_db(
        Table::alter()
            .table(Alias::new(table))
            .add_column(ColumnDef::new("new_col").custom(Alias::new("Int32")))
            .to_string(ClickhouseQueryBuilder),
        table,
    );
    client.execute(&sql).unwrap();

    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}

// ── Test 21: ALTER TABLE DROP COLUMN ──────────────────────────────────

#[test]
fn test_alter_drop_column() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_20";

    // Create a table with a column to drop
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("name").string())
                .col(ColumnDef::new("to_drop").string())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &create_sql);

    // ALTER TABLE DROP COLUMN
    let sql = with_db(
        Table::alter()
            .table(Alias::new(table))
            .drop_column("to_drop")
            .to_string(ClickhouseQueryBuilder),
        table,
    );
    client.execute(&sql).unwrap();

    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}

// ── Test 22: ALTER TABLE RENAME COLUMN ────────────────────────────────

#[test]
fn test_alter_rename_column() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_21";

    // Create a table with the column to rename
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("original_col").string())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &create_sql);

    // ALTER TABLE RENAME COLUMN
    let sql = with_db(
        Table::alter()
            .table(Alias::new(table))
            .rename_column("original_col", "renamed_col")
            .to_string(ClickhouseQueryBuilder),
        table,
    );
    client.execute(&sql).unwrap();

    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}

// ── Test 23: RENAME TABLE ─────────────────────────────────────────────

#[test]
fn test_rename_table() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_22";
    let renamed = "test_ddl_22_renamed";

    // Create the original table
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("name").string())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &create_sql);

    // RENAME TABLE — use raw SQL to qualify both source and target
    let sql = format!("RENAME TABLE test_db.`{table}` TO test_db.`{renamed}`");
    client.execute(&sql).unwrap();

    // Cleanup: drop the renamed table
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{renamed}"));
}

// ── Test 24: TRUNCATE TABLE ───────────────────────────────────────────

#[test]
fn test_truncate_table() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);
    let table = "test_ddl_23";

    // Create the table
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new("id").integer().not_null())
                .col(ColumnDef::new("name").string())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &create_sql);

    // Insert a row
    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES (1, 'test')"
        ))
        .unwrap();

    // Verify row exists
    let count = client
        .query(&format!("SELECT count() FROM test_db.{table}"))
        .unwrap();
    assert_eq!(count.trim(), "1");

    // TRUNCATE TABLE using the library builder
    let sql = with_db(
        Table::truncate()
            .table(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        table,
    );
    client.execute(&sql).unwrap();

    // Verify table is empty
    let count = client
        .query(&format!("SELECT count() FROM test_db.{table}"))
        .unwrap();
    assert_eq!(count.trim(), "0");

    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}
