use clickhouse::Row;
use serde::{Deserialize, Serialize};

use crate::connections::Connection;

use super::table::Ch;

#[derive(Serialize, Deserialize, Row)]
pub struct Database {
    pub name: String,
}

pub async fn all_databases(ch: Ch) -> Vec<Database> {
    let Ok(mut cursor) = ch
        .query("SELECT name FROM system.databases")
        .fetch::<Database>()
    else {
        log::error!("Failed to fetch databases");
        return Vec::new();
    };
    let mut databases = Vec::new();
    while let Ok(Some(row)) = cursor.next().await {
        databases.push(row);
    }
    databases
}

#[derive(Serialize, Deserialize, Row)]
pub struct Table {
    pub name: String,
}

/// Metadata about a single column as returned by `system.columns`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ColumnInfo {
    pub name: String,
    /// The raw ClickHouse type string, e.g. `"UInt64"`, `"String"`, `"DateTime64(3)"`.
    pub column_type: String,
}

/// Return the column names and types for `database.table` in definition order.
pub async fn describe_table(conn: &Connection, database: &str, table: &str) -> Vec<ColumnInfo> {
    let query = format!(
        "SELECT name, type FROM system.columns \
         WHERE database = '{}' AND table = '{}' \
         ORDER BY position \
         FORMAT TSVWithNames",
        database.replace('\'', "''"),
        table.replace('\'', "''")
    );

    let client = reqwest::Client::new();
    let mut request = client
        .post(conn.url())
        .query(&[("user", conn.user.as_str())])
        .body(query);
    if !conn.password.is_empty() {
        request = request.query(&[("password", conn.password.as_str())]);
    }

    let Ok(response) = request.send().await else {
        log::error!("describe_table: request failed for {database}.{table}");
        return Vec::new();
    };
    if !response.status().is_success() {
        log::error!("describe_table: bad status for {database}.{table}");
        return Vec::new();
    }
    let Ok(text) = response.text().await else {
        return Vec::new();
    };

    let mut lines = text.lines();
    let _ = lines.next(); // skip TSV header row

    lines
        .filter_map(|line| {
            let mut parts = line.splitn(2, '\t');
            let name = parts.next()?.to_string();
            let column_type = parts.next()?.to_string();
            Some(ColumnInfo { name, column_type })
        })
        .collect()
}

pub async fn all_tables(ch: Ch, database: &str) -> Vec<Table> {
    let Ok(mut cursor) = ch
        .query("SELECT name FROM system.tables WHERE database = ?")
        .bind(database)
        .fetch::<Table>()
    else {
        log::error!("Failed to fetch tables for database: {}", database);
        return Vec::new();
    };
    let mut tables = Vec::new();
    while let Ok(Some(row)) = cursor.next().await {
        tables.push(row);
    }
    tables
}