use std::fmt::Write;

use sea_query::*;

use crate::ClickhouseQueryBuilder;

/// ClickHouse SAMPLE clause configuration.
#[derive(Debug, Clone, PartialEq)]
pub struct ClickhouseSample {
    pub coefficient: f64,
    pub offset: Option<f64>,
}

/// Wrapper around [`SelectStatement`] that adds ClickHouse-specific clauses:
/// PREWHERE, FINAL, SAMPLE, and SETTINGS.
///
/// Build standard query parts with sea-query's `Query::select()`, then wrap
/// with `ClickhouseSelect::new()` to add ClickHouse features.
#[derive(Debug, Clone)]
pub struct ClickhouseSelect {
    inner: SelectStatement,
    prewhere: Option<Condition>,
    is_final: bool,
    sample: Option<ClickhouseSample>,
    settings: Vec<(String, String)>,
}

impl ClickhouseSelect {
    pub fn new(stmt: SelectStatement) -> Self {
        Self {
            inner: stmt,
            prewhere: None,
            is_final: false,
            sample: None,
            settings: Vec::new(),
        }
    }

    /// Add a PREWHERE condition. PREWHERE is a ClickHouse-specific optimization
    /// that filters data before reading it from disk.
    pub fn prewhere<C: IntoCondition>(mut self, condition: C) -> Self {
        let cond = condition.into_condition();
        self.prewhere = Some(match self.prewhere.take() {
            Some(existing) => existing.add(cond),
            None => Cond::all().add(cond),
        });
        self
    }

    /// Add the FINAL modifier. Forces ClickHouse to fully merge data parts
    /// before returning results.
    pub fn r#final(mut self) -> Self {
        self.is_final = true;
        self
    }

    /// Add a SAMPLE clause for approximate query processing.
    pub fn sample(mut self, coefficient: f64) -> Self {
        self.sample = Some(ClickhouseSample {
            coefficient,
            offset: None,
        });
        self
    }

    /// Add a SAMPLE clause with an OFFSET.
    pub fn sample_with_offset(mut self, coefficient: f64, offset: f64) -> Self {
        self.sample = Some(ClickhouseSample {
            coefficient,
            offset: Some(offset),
        });
        self
    }

    /// Add a query-level SETTINGS key-value pair.
    pub fn setting(mut self, key: &str, value: &str) -> Self {
        self.settings.push((key.to_string(), value.to_string()));
        self
    }

    fn has_ch_extensions(&self) -> bool {
        self.is_final
            || self.sample.is_some()
            || self.prewhere.is_some()
            || !self.settings.is_empty()
    }

    /// Build the ClickHouse query, returning the SQL string and bound values.
    pub fn build(self) -> (String, Values) {
        if !self.has_ch_extensions() {
            return self.inner.build(ClickhouseQueryBuilder);
        }

        let (placeholder, numbered) = ClickhouseQueryBuilder.placeholder();
        let mut sql_writer = SqlWriterValues::new(placeholder, numbered);
        self.inner
            .build_collect_into(ClickhouseQueryBuilder, &mut sql_writer);
        let (base_sql, base_values) = sql_writer.into_parts();

        let mut prewhere_sql = String::new();
        let mut prewhere_values = Values(Vec::new());
        if let Some(ref cond) = self.prewhere {
            let (placeholder, numbered) = ClickhouseQueryBuilder.placeholder();
            let mut pw_writer = SqlWriterValues::new(placeholder, numbered);
            let holder = ConditionHolder::new_with_condition(cond.clone());
            ClickhouseQueryBuilder.prepare_condition(&holder, "PREWHERE", &mut pw_writer);
            let (ps, pv) = pw_writer.into_parts();
            prewhere_sql = ps;
            prewhere_values = pv;
        }

        self.inject_parameterized(&base_sql, base_values, &prewhere_sql, prewhere_values)
    }

    fn inject_into_inlined(&self, base_sql: &str, prewhere_str: &str) -> String {
        let mut result = String::with_capacity(base_sql.len() + 128);

        let from_clause_end = find_from_clause_end(base_sql);

        let prewhere_insert_pos = find_prewhere_insert_pos(base_sql);

        result.push_str(&base_sql[..from_clause_end]);

        if self.is_final {
            result.push_str(" FINAL");
        }
        if let Some(ref sample) = self.sample {
            write!(result, " SAMPLE {}", sample.coefficient).unwrap();
            if let Some(offset) = sample.offset {
                write!(result, " OFFSET {offset}").unwrap();
            }
        }

        result.push_str(&base_sql[from_clause_end..prewhere_insert_pos]);

        if !prewhere_str.is_empty() {
            result.push(' ');
            result.push_str(prewhere_str);
        }

        result.push_str(&base_sql[prewhere_insert_pos..]);

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

        result
    }

    fn inject_parameterized(
        &self,
        base_sql: &str,
        base_values: Values,
        prewhere_sql: &str,
        prewhere_values: Values,
    ) -> (String, Values) {
        let mut result = String::with_capacity(base_sql.len() + 128);

        let from_clause_end = find_from_clause_end(base_sql);
        let prewhere_insert_pos = find_prewhere_insert_pos(base_sql);

        result.push_str(&base_sql[..from_clause_end]);

        if self.is_final {
            result.push_str(" FINAL");
        }
        if let Some(ref sample) = self.sample {
            write!(result, " SAMPLE {}", sample.coefficient).unwrap();
            if let Some(offset) = sample.offset {
                write!(result, " OFFSET {offset}").unwrap();
            }
        }

        result.push_str(&base_sql[from_clause_end..prewhere_insert_pos]);

        let params_before_prewhere = count_placeholders(&result);
        result.push_str(prewhere_sql);
        result.push_str(&base_sql[prewhere_insert_pos..]);

        let mut values: Vec<Value> = base_values.into_iter().collect();
        let prewhere_vals: Vec<Value> = prewhere_values.into_iter().collect();

        for (i, v) in prewhere_vals.into_iter().enumerate() {
            values.insert(params_before_prewhere + i, v);
        }

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

        (result, Values(values))
    }
}

impl std::fmt::Display for ClickhouseSelect {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if !self.has_ch_extensions() {
            return f.write_str(&self.inner.to_string(ClickhouseQueryBuilder));
        }

        let base_sql = self.inner.to_string(ClickhouseQueryBuilder);

        let prewhere_str = if let Some(ref cond) = self.prewhere {
            build_condition_inlined(cond, "PREWHERE")
        } else {
            String::new()
        };

        f.write_str(&self.inject_into_inlined(&base_sql, &prewhere_str))
    }
}

fn build_condition_inlined(cond: &Condition, keyword: &str) -> String {
    let helper = Query::select()
        .expr(Expr::val(1))
        .cond_where(cond.clone())
        .to_string(ClickhouseQueryBuilder);
    if let Some(where_pos) = find_keyword_pos(&helper, " WHERE ") {
        let condition_sql = &helper[where_pos + 7..];
        format!("{keyword} {condition_sql}")
    } else {
        String::new()
    }
}

fn find_from_clause_end(sql: &str) -> usize {
    let clause_keywords = [
        " LEFT JOIN ",
        " RIGHT JOIN ",
        " INNER JOIN ",
        " CROSS JOIN ",
        " FULL JOIN ",
        " JOIN ",
        " PREWHERE ",
        " WHERE ",
        " GROUP BY ",
        " HAVING ",
        " ORDER BY ",
        " LIMIT ",
        " UNION ",
    ];

    let from_pos = find_keyword_pos(sql, " FROM ");
    if let Some(fp) = from_pos {
        let after_from = fp + 6;
        let mut earliest = sql.len();
        for kw in &clause_keywords {
            if let Some(pos) = find_keyword_pos(&sql[after_from..], kw) {
                earliest = earliest.min(after_from + pos);
            }
        }
        earliest
    } else {
        sql.len()
    }
}

fn find_prewhere_insert_pos(sql: &str) -> usize {
    let where_pos = find_keyword_pos(sql, " WHERE ");
    if let Some(wp) = where_pos {
        return wp;
    }
    let fallback_keywords = [" GROUP BY ", " HAVING ", " ORDER BY ", " LIMIT ", " UNION "];
    for kw in &fallback_keywords {
        if let Some(pos) = find_keyword_pos(sql, kw) {
            return pos;
        }
    }
    sql.len()
}

fn find_keyword_pos(sql: &str, keyword: &str) -> Option<usize> {
    let mut search_from = 0;
    while let Some(pos) = sql[search_from..].find(keyword) {
        let abs_pos = search_from + pos;
        if !is_inside_string_literal(sql, abs_pos) {
            return Some(abs_pos);
        }
        search_from = abs_pos + 1;
    }
    None
}

fn is_inside_string_literal(sql: &str, pos: usize) -> bool {
    let mut in_single = false;
    let mut escaped = false;
    for (i, c) in sql.char_indices() {
        if i >= pos {
            break;
        }
        if escaped {
            escaped = false;
            continue;
        }
        if c == '\\' {
            escaped = true;
            continue;
        }
        if c == '\'' {
            in_single = !in_single;
        }
    }
    in_single
}

fn count_placeholders(sql: &str) -> usize {
    let mut count = 0;
    let mut in_single = false;
    let mut escaped = false;
    for c in sql.chars() {
        if escaped {
            escaped = false;
            continue;
        }
        if c == '\\' {
            escaped = true;
            continue;
        }
        if c == '\'' {
            in_single = !in_single;
            continue;
        }
        if !in_single && c == '?' {
            count += 1;
        }
    }
    count
}
