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

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

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

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,
    )
}

/// Replace all backtick-quoted occurrences of each table name with
/// `test_db.`-qualified versions.  Handles JOINs, subqueries, and UNIONs.
fn qualify(sql: &str, tables: &[&str]) -> String {
    let mut result = sql.to_string();
    for table in tables {
        result = result.replace(&format!("`{table}`"), &format!("test_db.`{table}`"));
    }
    result
}

/// Helper: create a simple MergeTree table with the given column definitions.
fn create_simple_table(client: &ClickhouseClient, table: &str, cols: Vec<(&str, &str)>) {
    let order_col = cols[0].0;
    let mut stmt = Table::create();
    stmt.table(Alias::new(table)).col(
        ColumnDef::new(Alias::new(cols[0].0))
            .custom(Alias::new(cols[0].1))
            .not_null(),
    );
    for (name, typ) in &cols[1..] {
        stmt.col(ColumnDef::new(Alias::new(*name)).custom(Alias::new(*typ)));
    }
    let sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(stmt)
            .engine("MergeTree")
            .order_by([Alias::new(order_col)]),
    );
    ensure_table(client, table, &sql);
}

// ════════════════════════════════════════════════════════════════════════
// Section 1 – Basic DML
// ════════════════════════════════════════════════════════════════════════

// ── 1. INSERT rows and verify count ────────────────────────────────────

#[test]
#[allow(clippy::approx_constant)]
fn test_insert_values() {
    let client = client();
    if !client.is_available() {
        return;
    }
    ensure_test_db(&client);

    let table = "test_dml_1";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("name", "String"), ("value", "Float64")],
    );

    let sql = qualify(
        &Query::insert()
            .into_table(Alias::new(table))
            .columns([Alias::new("id"), Alias::new("name"), Alias::new("value")])
            .values_panic([1.into(), "alice".into(), 3.14.into()])
            .values_panic([2.into(), "bob".into(), 2.71.into()])
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    client.execute(&sql).unwrap();

    let count = client
        .query(&format!("SELECT count() FROM test_db.{table}"))
        .unwrap();
    assert_eq!(count.trim(), "2");
}

// ── 2. INSERT + SELECT with WHERE ──────────────────────────────────────

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

    let table = "test_dml_2";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("name", "String"), ("score", "Int32")],
    );

    // Insert three rows
    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name, score) VALUES (1, 'alice', 90), (2, 'bob', 75), (3, 'carol', 90)"
        ))
        .unwrap();

    // Build SELECT with WHERE using the query builder
    let sql = qualify(
        &Query::select()
            .column(Alias::new("name"))
            .from(Alias::new(table))
            .and_where(Expr::col(Alias::new("score")).eq(90))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 2);
    assert!(rows.contains(&"alice"));
    assert!(rows.contains(&"carol"));
}

// ── 3. SELECT with LIMIT / OFFSET ─────────────────────────────────────

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

    let table = "test_dml_3";
    create_simple_table(&client, table, vec![("id", "Int32"), ("name", "String")]);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES (1,'a'), (2,'b'), (3,'c'), (4,'d'), (5,'e')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("name"))
            .from(Alias::new(table))
            .order_by(Alias::new("id"), Order::Asc)
            .limit(2)
            .offset(2)
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows, vec!["c", "d"]);
}

// ── 4. SELECT DISTINCT ────────────────────────────────────────────────

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

    let table = "test_dml_4";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("category", "String")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, category) VALUES \
             (1,'A'), (2,'B'), (3,'A'), (4,'C'), (5,'B')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .distinct()
            .column(Alias::new("category"))
            .from(Alias::new(table))
            .order_by(Alias::new("category"), Order::Asc)
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows, vec!["A", "B", "C"]);
}

// ── 5. SELECT JOIN ────────────────────────────────────────────────────

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

    let left = "test_dml_5a";
    let right = "test_dml_5b";

    create_simple_table(&client, left, vec![("id", "Int32"), ("name", "String")]);
    create_simple_table(
        &client,
        right,
        vec![("id", "Int32"), ("parent_id", "Int32"), ("value", "String")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{left} (id, name) VALUES (1,'alpha'), (2,'beta')"
        ))
        .unwrap();
    client
        .execute(&format!(
            "INSERT INTO test_db.{right} (id, parent_id, value) VALUES (10,1,'x'), (20,2,'y')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column((Alias::new(left), Alias::new("name")))
            .column((Alias::new(right), Alias::new("value")))
            .from(Alias::new(left))
            .left_join(
                Alias::new(right),
                Expr::col((Alias::new(left), Alias::new("id")))
                    .equals((Alias::new(right), Alias::new("parent_id"))),
            )
            .order_by((Alias::new(left), Alias::new("id")), Order::Asc)
            .to_string(ClickhouseQueryBuilder),
        &[left, right],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 2);
    assert!(rows[0].contains("alpha") && rows[0].contains("x"));
    assert!(rows[1].contains("beta") && rows[1].contains("y"));
}

// ── 6. GROUP BY + HAVING ──────────────────────────────────────────────

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

    let table = "test_dml_6";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("category", "String"), ("amount", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, category, amount) VALUES \
             (1,'A',10), (2,'A',20), (3,'B',5), (4,'B',15), (5,'C',2)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("category"))
            .expr(Expr::col(Alias::new("amount")).sum())
            .from(Alias::new(table))
            .group_by_col(Alias::new("category"))
            .and_having(Expr::col(Alias::new("category")).is_not_null())
            .order_by(Alias::new("category"), Order::Asc)
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 3);
    assert!(rows[0].starts_with("A\t")); // sum = 30
    assert!(rows[1].starts_with("B\t")); // sum = 20
    assert!(rows[2].starts_with("C\t")); // sum = 2
}

// ── 7. ORDER BY ASC / DESC ────────────────────────────────────────────

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

    let table = "test_dml_7";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("name", "String"), ("score", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name, score) VALUES \
             (1,'alice',90), (2,'bob',75), (3,'carol',85)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("name"))
            .from(Alias::new(table))
            .order_by(Alias::new("score"), Order::Desc)
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows, vec!["alice", "carol", "bob"]);
}

// ── 8. Subquery in FROM ───────────────────────────────────────────────

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

    let table = "test_dml_8";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("name", "String"), ("score", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name, score) VALUES \
             (1,'alice',90), (2,'bob',75), (3,'carol',85)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("name"))
            .column(Alias::new("score"))
            .from_subquery(
                Query::select()
                    .columns([Alias::new("name"), Alias::new("score")])
                    .from(Alias::new(table))
                    .and_where(Expr::col(Alias::new("score")).gt(80))
                    .to_owned(),
                Alias::new("subq"),
            )
            .order_by(Alias::new("score"), Order::Desc)
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 2);
    assert!(rows[0].starts_with("alice\t"));
    assert!(rows[1].starts_with("carol\t"));
}

// ── 9. UNION ALL ──────────────────────────────────────────────────────

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

    let table = "test_dml_9";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("category", "String"), ("value", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, category, value) VALUES \
             (1,'A',10), (2,'A',20), (3,'B',30), (4,'B',40)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("value"))
            .from(Alias::new(table))
            .and_where(Expr::col(Alias::new("category")).eq("A"))
            .union(
                UnionType::All,
                Query::select()
                    .column(Alias::new("value"))
                    .from(Alias::new(table))
                    .and_where(Expr::col(Alias::new("category")).eq("B"))
                    .to_owned(),
            )
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 4);
}

// ── 10. UPDATE (ALTER TABLE … UPDATE syntax) ──────────────────────────

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

    let table = "test_dml_10";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("name", "String"), ("value", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name, value) VALUES (1,'before',10), (2,'keep',20)"
        ))
        .unwrap();

    // The builder produces: ALTER TABLE `table` UPDATE `name`='after', `value`=99 WHERE `id`=1
    let sql = qualify(
        &Query::update()
            .table(Alias::new(table))
            .values([
                (Alias::new("name"), "after".into()),
                (Alias::new("value"), 99.into()),
            ])
            .and_where(Expr::col(Alias::new("id")).eq(1))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    // Make the mutation synchronous so verification is reliable
    let sql = format!("{sql} SETTINGS mutations_sync = 1");
    client.execute(&sql).unwrap();

    // Verify only the targeted row changed
    let result = client
        .query(&format!(
            "SELECT name, value FROM test_db.{table} WHERE id = 1"
        ))
        .unwrap();
    assert!(result.contains("after"));
    assert!(result.contains("99"));

    let other = client
        .query(&format!(
            "SELECT name, value FROM test_db.{table} WHERE id = 2"
        ))
        .unwrap();
    assert!(other.contains("keep"));
    assert!(other.contains("20"));
}

// ── 11. DELETE (ALTER TABLE … DELETE WHERE syntax) ────────────────────

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

    let table = "test_dml_11";
    create_simple_table(&client, table, vec![("id", "Int32"), ("name", "String")]);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES (1,'delete_me'), (2,'keep_me')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::delete()
            .from_table(Alias::new(table))
            .and_where(Expr::col(Alias::new("id")).eq(1))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let sql = format!("{sql} SETTINGS mutations_sync = 1");
    client.execute(&sql).unwrap();

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

    let remaining = client
        .query(&format!("SELECT name FROM test_db.{table}"))
        .unwrap();
    assert_eq!(remaining.trim(), "keep_me");
}

// ════════════════════════════════════════════════════════════════════════
// Section 2 – ClickHouse SELECT Extensions (via ClickhouseSelect)
// ════════════════════════════════════════════════════════════════════════

// ── 12. FINAL modifier ────────────────────────────────────────────────

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

    let table = "test_dml_12";
    // FINAL requires a ReplacingMergeTree (or similar) engine
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
    let create_sql = create_db_sql(
        table,
        ClickhouseTableCreate::new(
            Table::create()
                .table(Alias::new(table))
                .col(ColumnDef::new(Alias::new("id")).integer().not_null())
                .col(ColumnDef::new(Alias::new("name")).string())
                .col(ColumnDef::new(Alias::new("ver")).integer())
                .to_owned(),
        )
        .engine_with_params("ReplacingMergeTree", vec!["ver".to_string()])
        .order_by([Alias::new("id")]),
    );
    client.execute(&create_sql).unwrap();

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name, ver) VALUES (1,'alpha',1), (1,'alpha2',2), (2,'beta',1)"
        ))
        .unwrap();

    let raw = ClickhouseSelect::new(
        Query::select()
            .column(Alias::new("id"))
            .column(Alias::new("name"))
            .from(Alias::new(table))
            .to_owned(),
    )
    .r#final()
    .to_string();

    let sql = qualify(&raw, &[table]);
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    // FINAL collapses the two rows with id=1 into the latest version (ver=2),
    // so we expect 2 rows: one for id=1 and one for id=2
    assert_eq!(rows.len(), 2);
    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}

// ── 13. PREWHERE ──────────────────────────────────────────────────────

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

    let table = "test_dml_13";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("status", "Int32"), ("name", "String")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, status, name) VALUES \
             (1,1,'active'), (2,0,'inactive'), (3,1,'active2')"
        ))
        .unwrap();

    let raw = ClickhouseSelect::new(
        Query::select()
            .column(Alias::new("name"))
            .from(Alias::new(table))
            .and_where(Expr::col(Alias::new("name")).like("active%"))
            .to_owned(),
    )
    .prewhere(Expr::col(Alias::new("status")).eq(1))
    .to_string();

    let sql = qualify(&raw, &[table]);
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 2);
    assert!(rows.contains(&"active"));
    assert!(rows.contains(&"active2"));
}

// ── 14. SAMPLE ────────────────────────────────────────────────────────

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

    let table = "test_dml_14";
    // SAMPLE requires a table with SAMPLE BY clause.
    // The sample expression must be present in the primary key / order by key.
    // We use intHash32(id) as the sample key and include it in ORDER BY.
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
    let create_sql = format!(
        "CREATE TABLE test_db.{table} (\
           id Int32, \
           name String\
         ) ENGINE = MergeTree \
         ORDER BY (intHash32(id), id) \
         SAMPLE BY intHash32(id)"
    );
    client.execute(&create_sql).unwrap();

    // Insert enough rows so SAMPLE 1.0 can return something
    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES \
             (1,'a'), (2,'b'), (3,'c'), (4,'d'), (5,'e'), \
             (6,'f'), (7,'g'), (8,'h'), (9,'i'), (10,'j')"
        ))
        .unwrap();

    let raw = ClickhouseSelect::new(
        Query::select()
            .column(Alias::new("id"))
            .from(Alias::new(table))
            .to_owned(),
    )
    .sample(1.0) // use 1.0 so we always get all rows (deterministic)
    .to_string();

    let sql = qualify(&raw, &[table]);
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 10);
    // Cleanup
    let _ = client.execute(&format!("DROP TABLE IF EXISTS test_db.{table}"));
}

// ── 15. SETTINGS ──────────────────────────────────────────────────────

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

    let table = "test_dml_15";
    create_simple_table(&client, table, vec![("id", "Int32"), ("name", "String")]);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES (1,'x'), (2,'y')"
        ))
        .unwrap();

    let raw = ClickhouseSelect::new(
        Query::select()
            .column(Alias::new("id"))
            .from(Alias::new(table))
            .to_owned(),
    )
    .setting("max_threads", "2")
    .to_string();

    let sql = qualify(&raw, &[table]);
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows.len(), 2);
}

// ════════════════════════════════════════════════════════════════════════
// Section 3 – ClickHouse Function Tests
// ════════════════════════════════════════════════════════════════════════

// ── 16. toString ──────────────────────────────────────────────────────

#[test]
fn test_func_to_string() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::to_string(Expr::val(42)))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "42");
}

// ── 17. toUInt32 ──────────────────────────────────────────────────────

#[test]
fn test_func_to_uint32() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::to_uint32(Expr::val("123")))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "123");
}

// ── 18. now() ─────────────────────────────────────────────────────────

#[test]
fn test_func_now() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::now())
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    // Should return a non-empty datetime string, e.g. "2024-06-15 12:34:56"
    assert!(!result.trim().is_empty());
}

// ── 19. today() ───────────────────────────────────────────────────────

#[test]
fn test_func_today() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::today())
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    // Should return a date like "2024-06-15"
    assert!(!result.trim().is_empty());
}

// ── 20. length ────────────────────────────────────────────────────────

#[test]
fn test_func_length() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::length(Expr::val("hello")))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "5");
}

// ── 21. toYear ────────────────────────────────────────────────────────

#[test]
fn test_func_to_year() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::to_year(ChFunc::to_date(Expr::val("2024-06-15"))))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "2024");
}

// ── 22. groupArray ────────────────────────────────────────────────────

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

    let table = "test_dml_fn_1";
    create_simple_table(&client, table, vec![("id", "Int32"), ("name", "String")]);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES (1,'a'), (2,'b'), (3,'c')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .expr(ChFunc::group_array(Expr::col(Alias::new("name"))))
            .from(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    // Should contain all three values in an array representation
    let trimmed = result.trim();
    assert!(trimmed.contains('a'));
    assert!(trimmed.contains('b'));
    assert!(trimmed.contains('c'));
}

// ── 23. uniq ──────────────────────────────────────────────────────────

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

    let table = "test_dml_fn_2";
    create_simple_table(&client, table, vec![("id", "Int32"), ("name", "String")]);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, name) VALUES \
             (1,'a'), (2,'b'), (3,'a'), (4,'c'), (5,'b')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .expr(ChFunc::uniq(Expr::col(Alias::new("name"))))
            .from(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "3"); // a, b, c
}

// ── 24. sumIf ─────────────────────────────────────────────────────────

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

    let table = "test_dml_fn_3";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("category", "String"), ("amount", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, category, amount) VALUES \
             (1,'A',10), (2,'A',20), (3,'B',30), (4,'B',40)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .expr(ChFunc::sum_if(
                Expr::col(Alias::new("amount")),
                Expr::col(Alias::new("category")).eq("A"),
            ))
            .from(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "30"); // 10 + 20
}

// ── 25. countIf ───────────────────────────────────────────────────────

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

    let table = "test_dml_fn_4";
    create_simple_table(
        &client,
        table,
        vec![("id", "Int32"), ("category", "String"), ("amount", "Int32")],
    );

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, category, amount) VALUES \
             (1,'A',10), (2,'B',20), (3,'A',30), (4,'C',5)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .expr(ChFunc::count_if(Expr::col(Alias::new("category")).eq("A")))
            .from(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "2");
}

// ── 26. if() ──────────────────────────────────────────────────────────

#[test]
fn test_func_if() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::r#if(
            Expr::val(1).gt(Expr::val(0)),
            Expr::val("yes"),
            Expr::val("no"),
        ))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "yes");
}

// ── 27. isNull ────────────────────────────────────────────────────────

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

    let table = "test_dml_fn_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("val").clickhouse_nullable("String"))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, val) VALUES (1,'hello'), (2,NULL)"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("id"))
            .from(Alias::new(table))
            .and_where(Expr::expr(ChFunc::is_null(Expr::col(Alias::new("val")))).eq(1))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "2");
}

// ── 28. has (array) ──────────────────────────────────────────────────

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

    let table = "test_dml_fn_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("tags").custom(Alias::new("Array(String)")))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, tags) VALUES (1, ['a','b','c']), (2, ['x','y'])"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("id"))
            .from(Alias::new(table))
            .and_where(Expr::expr(ChFunc::has(Expr::col(Alias::new("tags")), Expr::val("a"))).eq(1))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "1");
}

// ── 29. replaceAll ────────────────────────────────────────────────────

#[test]
fn test_func_replace_all() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::replace_all(
            Expr::val("hello world"),
            Expr::val("world"),
            Expr::val("ClickHouse"),
        ))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "hello ClickHouse");
}

// ── 30. sipHash64 ────────────────────────────────────────────────────

#[test]
fn test_func_sip_hash64() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::sip_hash64(Expr::val("test")))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    // Should return a non-empty hash string
    assert!(!result.trim().is_empty());
    // Should be a valid UInt64 (all digits)
    assert!(result.trim().chars().all(|c| c.is_ascii_digit()));
}

// ── 31. generateUUIDv4 ───────────────────────────────────────────────

#[test]
fn test_func_generate_uuid_v4() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::generate_uuid_v4())
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    let uuid = result.trim();
    // UUID v4 format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
    assert_eq!(uuid.len(), 36);
    assert!(uuid.chars().filter(|c| *c == '-').count() == 4);
}

// ── 32. Nested functions ──────────────────────────────────────────────

#[test]
fn test_func_nested() {
    let client = client();
    if !client.is_available() {
        return;
    }
    let sql = Query::select()
        .expr(ChFunc::to_start_of_month(ChFunc::to_date(Expr::val(
            "2024-06-15",
        ))))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "2024-06-01");
}

// ── 33. Function in WHERE clause ──────────────────────────────────────

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

    let table = "test_dml_fn_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("created_at").date())
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, created_at) VALUES \
             (1,'2024-01-15'), (2,'2024-06-20'), (3,'2023-11-30')"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .column(Alias::new("id"))
            .from(Alias::new(table))
            .and_where(Expr::expr(ChFunc::to_year(Expr::col(Alias::new("created_at")))).eq(2024))
            .order_by(Alias::new("id"), Order::Asc)
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows, vec!["1", "2"]);
}

// ── 34. JSON operations ───────────────────────────────────────────────

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

    // JSONExtractString
    let sql = Query::select()
        .expr(ChFunc::json_extract_string([
            Expr::val("{\"name\":\"ClickHouse\",\"version\":\"24.0\"}"),
            Expr::val("name"),
        ]))
        .to_string(ClickhouseQueryBuilder);
    let result = client.query(&sql).unwrap();
    assert_eq!(result.trim(), "ClickHouse");

    // JSON -> operator (ChBinOper::GetJsonField) via a table with a string column
    ensure_test_db(&client);
    let table = "test_dml_fn_json";
    create_simple_table(&client, table, vec![("id", "Int32"), ("data", "String")]);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, data) VALUES (1, '{{\"key\": \"value\"}}')"
        ))
        .unwrap();

    // Use JSONExtractString function instead of -> operator which is restricted
    // in ClickHouse 26.x when used with string columns
    let sql = qualify(
        &Query::select()
            .expr(ChFunc::json_extract_string([
                Expr::col(Alias::new("data")),
                Expr::val("key"),
            ]))
            .from(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    assert!(result.trim().contains("value"));
}

// ── 35. arrayJoin ─────────────────────────────────────────────────────

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

    let table = "test_dml_fn_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("tags").custom(Alias::new("Array(String)")))
                .to_owned(),
        )
        .engine("MergeTree")
        .order_by(["id"]),
    );
    ensure_table(&client, table, &sql);

    client
        .execute(&format!(
            "INSERT INTO test_db.{table} (id, tags) VALUES (1, ['x','y','z'])"
        ))
        .unwrap();

    let sql = qualify(
        &Query::select()
            .expr(ChFunc::array_join(Expr::col(Alias::new("tags"))))
            .from(Alias::new(table))
            .to_string(ClickhouseQueryBuilder),
        &[table],
    );
    let result = client.query(&sql).unwrap();
    let rows: Vec<&str> = result.trim().lines().collect();
    assert_eq!(rows, vec!["x", "y", "z"]);
}
