use std::fmt::Write;

use sea_query::*;

use crate::ClickhouseQueryBuilder;

/// ClickHouse table engine specification.
#[derive(Debug, Clone, PartialEq)]
pub struct ClickhouseTableEngine {
    pub name: String,
    pub params: Vec<String>,
}

/// ClickHouse-specific table creation options appended after the base DDL.
#[derive(Debug, Clone, Default)]
pub struct ClickhouseTableOptions {
    pub engine: Option<ClickhouseTableEngine>,
    pub order_by: Vec<DynIden>,
    pub partition_by: Option<Expr>,
    pub primary_key: Vec<DynIden>,
    pub sample_by: Option<Expr>,
    pub ttl: Option<String>,
    pub settings: Vec<(String, String)>,
}

/// Wrapper around [`TableCreateStatement`] that appends ClickHouse-specific
/// clauses (ENGINE, ORDER BY, PARTITION BY, etc.) after the base DDL.
#[derive(Debug, Clone)]
pub struct ClickhouseTableCreate {
    inner: TableCreateStatement,
    options: ClickhouseTableOptions,
}

impl ClickhouseTableCreate {
    pub fn new(stmt: TableCreateStatement) -> Self {
        Self {
            inner: stmt,
            options: ClickhouseTableOptions::default(),
        }
    }

    pub fn engine(mut self, name: &str) -> Self {
        self.options.engine = Some(ClickhouseTableEngine {
            name: name.to_string(),
            params: Vec::new(),
        });
        self
    }

    pub fn engine_with_params(mut self, name: &str, params: Vec<String>) -> Self {
        self.options.engine = Some(ClickhouseTableEngine {
            name: name.to_string(),
            params,
        });
        self
    }

    pub fn order_by<I, T>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = T>,
        T: IntoIden,
    {
        self.options.order_by = columns.into_iter().map(|c| c.into_iden()).collect();
        self
    }

    pub fn partition_by_expr(mut self, expr: Expr) -> Self {
        self.options.partition_by = Some(expr);
        self
    }

    pub fn primary_key<I, T>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = T>,
        T: IntoIden,
    {
        self.options.primary_key = columns.into_iter().map(|c| c.into_iden()).collect();
        self
    }

    pub fn sample_by_expr(mut self, expr: Expr) -> Self {
        self.options.sample_by = Some(expr);
        self
    }

    pub fn ttl(mut self, ttl: &str) -> Self {
        self.options.ttl = Some(ttl.to_string());
        self
    }

    pub fn table_setting(mut self, key: &str, value: &str) -> Self {
        self.options
            .settings
            .push((key.to_string(), value.to_string()));
        self
    }

    /// Build the CREATE TABLE string.
    pub fn build(self) -> String {
        let base_sql = self.inner.build(ClickhouseQueryBuilder);
        let mut result = normalize_create_table_parens(&base_sql);
        self.append_options(&mut result);
        result
    }

    fn append_options(&self, sql: &mut String) {
        let opts = &self.options;

        if let Some(engine) = &opts.engine {
            sql.push_str(" ENGINE = ");
            sql.push_str(&engine.name);
            sql.push('(');
            for (i, param) in engine.params.iter().enumerate() {
                if i > 0 {
                    sql.push_str(", ");
                }
                sql.push_str(param);
            }
            sql.push(')');
        }

        if !opts.order_by.is_empty() {
            sql.push_str(" ORDER BY (");
            for (i, col) in opts.order_by.iter().enumerate() {
                if i > 0 {
                    sql.push_str(", ");
                }
                let q = ClickhouseQueryBuilder.quote();
                write!(sql, "{}{}{}", { q.left() }, col, { q.right() }).unwrap();
            }
            sql.push(')');
        }

        if let Some(partition_by) = &opts.partition_by {
            sql.push_str(" PARTITION BY ");
            let expr_sql = render_expr(partition_by);
            sql.push_str(&expr_sql);
        }

        if !opts.primary_key.is_empty() {
            sql.push_str(" PRIMARY KEY (");
            for (i, col) in opts.primary_key.iter().enumerate() {
                if i > 0 {
                    sql.push_str(", ");
                }
                let q = ClickhouseQueryBuilder.quote();
                write!(sql, "{}{}{}", { q.left() }, col, { q.right() }).unwrap();
            }
            sql.push(')');
        }

        if let Some(sample_by) = &opts.sample_by {
            sql.push_str(" SAMPLE BY ");
            let expr_sql = render_expr(sample_by);
            sql.push_str(&expr_sql);
        }

        if let Some(ttl) = &opts.ttl {
            sql.push_str(" TTL ");
            sql.push_str(ttl);
        }

        if !opts.settings.is_empty() {
            sql.push_str(" SETTINGS ");
            for (i, (key, value)) in opts.settings.iter().enumerate() {
                if i > 0 {
                    sql.push_str(", ");
                }
                write!(sql, "{key} = {value}").unwrap();
            }
        }
    }
}

/// The default `prepare_table_create_statement` formats columns as `( col1, col2 )`
/// with spaces inside parentheses. ClickHouse convention is `(col1, col2)`.
impl std::fmt::Display for ClickhouseTableCreate {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let base_sql = self.inner.build(ClickhouseQueryBuilder);
        let mut result = normalize_create_table_parens(&base_sql);
        self.append_options(&mut result);
        f.write_str(&result)
    }
}

fn normalize_create_table_parens(sql: &str) -> String {
    let s = sql.replacen(" ( ", " (", 1);
    if s.ends_with(" )") {
        format!("{})", &s[..s.len() - 2])
    } else {
        s
    }
}

fn render_expr(expr: &Expr) -> String {
    let mut sql = String::new();
    ClickhouseQueryBuilder.prepare_expr(expr, &mut sql);
    sql
}
