use std::collections::{HashMap, HashSet};

use super::{OnConflict, OnConflictAction, SqlBuilder};
use crate::types::DriverType;

/// MySQL-specific SQL builder.
pub struct MysqlSqlBuilder;

impl SqlBuilder for MysqlSqlBuilder {
    fn driver_type(&self) -> DriverType {
        DriverType::Mysql
    }

    fn supports_returning(&self) -> bool {
        false
    }

    fn rewrite_placeholders(&self, sql: &str) -> String {
        sql.to_string()
    }

    fn build_insert(
        &self,
        table: &str,
        columns: &[&str],
        row_count: usize,
        on_conflict: Option<&OnConflict>,
        returning_columns: Option<&[&str]>,
        _column_types: Option<&[&str]>,
    ) -> (String, usize) {
        // MySQL does not support RETURNING natively, but the driver simulates it
        // by appending RETURNING and issuing a follow-up SELECT when needed.
        let cols = columns.join(", ");
        let total_params = columns.len() * row_count;

        let row_placeholder = format!("({})", vec!["?"; columns.len()].join(", "));
        let value_rows: Vec<&str> = (0..row_count).map(|_| row_placeholder.as_str()).collect();

        let sql = if let Some(oc) = on_conflict {
            match &oc.action {
                OnConflictAction::DoNothing => {
                    // MySQL: INSERT IGNORE INTO ... (ignores all unique constraint violations)
                    format!(
                        "INSERT IGNORE INTO {} ({}) VALUES {}",
                        table,
                        cols,
                        value_rows.join(", ")
                    )
                }
                OnConflictAction::DoUpdate(update_cols) => {
                    // MySQL: ON DUPLICATE KEY UPDATE col = VALUES(col)
                    // Fallback: when update_cols is empty, use all inserted columns (matches PG driver)
                    let cols_to_update: Vec<&str> = if update_cols.is_empty() {
                        columns.to_vec()
                    } else {
                        update_cols.to_vec()
                    };
                    let update_clause: Vec<String> = cols_to_update
                        .iter()
                        .map(|c| format!("{} = VALUES({})", c, c))
                        .collect();
                    format!(
                        "INSERT INTO {} ({}) VALUES {} ON DUPLICATE KEY UPDATE {}",
                        table,
                        cols,
                        value_rows.join(", "),
                        update_clause.join(", ")
                    )
                }
            }
        } else {
            format!(
                "INSERT INTO {} ({}) VALUES {}",
                table,
                cols,
                value_rows.join(", ")
            )
        };

        // MySQL doesn't support RETURNING natively, but the driver simulates it
        let mut sql = sql;
        if let Some(ret_cols) = returning_columns {
            sql.push_str(&format!(" RETURNING {}", ret_cols.join(", ")));
        }

        (sql, total_params)
    }

    fn build_update_many(
        &self,
        table: &str,
        key_columns: &[&str],
        value_columns: &[&str],
        row_count: usize,
        returning_columns: Option<&[&str]>,
        _column_types: Option<&[&str]>,
        skip_map: Option<&HashMap<usize, HashSet<String>>>,
    ) -> (String, usize) {
        // MySQL: UPDATE table AS t
        //        JOIN (VALUES ROW(?, ?, ?), ROW(?, ?, ?)) AS v(col1, col2, col3)
        //        ON t.key = v.key
        //        SET t.col = IF(v._skip_col, t.col, v.col)
        let all_columns: Vec<&str> = key_columns
            .iter()
            .chain(value_columns.iter())
            .copied()
            .collect();

        // Collect skip columns needed
        let skip_columns_needed: Vec<String> = skip_map
            .map(|m| {
                let mut set = HashSet::new();
                for skip_set in m.values() {
                    for col in skip_set {
                        set.insert(col.clone());
                    }
                }
                let mut v: Vec<String> = set.into_iter().collect();
                v.sort();
                v
            })
            .unwrap_or_default();

        let all_value_cols: Vec<String> = all_columns
            .iter()
            .map(|s| (*s).to_string())
            .chain(skip_columns_needed.iter().map(|c| format!("_skip_{}", c)))
            .collect();

        let params_per_row = all_columns.len() + skip_columns_needed.len();
        let total_params = params_per_row * row_count;

        let row_placeholder = format!("ROW({})", vec!["?"; params_per_row].join(", "));
        let value_rows: Vec<&str> = (0..row_count).map(|_| row_placeholder.as_str()).collect();

        let set_clause: Vec<String> = value_columns
            .iter()
            .map(|col| {
                if skip_columns_needed.contains(&(*col).to_string()) {
                    format!("t.{} = IF(v._skip_{}, t.{}, v.{})", col, col, col, col)
                } else {
                    format!("t.{} = v.{}", col, col)
                }
            })
            .collect();

        let on_clause: Vec<String> = key_columns
            .iter()
            .map(|col| format!("t.{} = v.{}", col, col))
            .collect();

        let mut sql = format!(
            "UPDATE {} AS t JOIN (VALUES {}) AS v({}) ON {} SET {}",
            table,
            value_rows.join(", "),
            all_value_cols.join(", "),
            on_clause.join(" AND "),
            set_clause.join(", "),
        );

        // MySQL doesn't support RETURNING natively, but the driver simulates it
        if let Some(ret_cols) = returning_columns {
            let prefixed: Vec<String> = ret_cols.iter().map(|c| format!("t.{}", c)).collect();
            sql.push_str(&format!(" RETURNING {}", prefixed.join(", ")));
        }

        (sql, total_params)
    }

    fn build_find_by_pkeys(
        &self,
        table: &str,
        pkey_columns: &[&str],
        pkey_count: usize,
        select_column: &str,
        _column_types: Option<&[&str]>,
    ) -> (String, usize) {
        if pkey_columns.len() == 1 {
            // Single PK: WHERE id IN (?, ?, ...)
            let placeholders = vec!["?"; pkey_count].join(", ");
            let sql = format!(
                "SELECT {} FROM {} WHERE {} IN ({})",
                select_column, table, pkey_columns[0], placeholders
            );
            return (sql, pkey_count);
        }

        // Composite PK: JOIN (VALUES ROW(...), ROW(...)) AS v(...) ON ...
        let total_params = pkey_columns.len() * pkey_count;
        let row_placeholder = format!("ROW({})", vec!["?"; pkey_columns.len()].join(", "));
        let value_rows: Vec<&str> = (0..pkey_count).map(|_| row_placeholder.as_str()).collect();

        let on_clause: Vec<String> = pkey_columns
            .iter()
            .map(|col| format!("t.{} = v.{}", col, col))
            .collect();

        let select_col = if select_column == "*" {
            "t.*"
        } else {
            select_column
        };

        let sql = format!(
            "SELECT {} FROM {} AS t JOIN (VALUES {}) AS v({}) ON {}",
            select_col,
            table,
            value_rows.join(", "),
            pkey_columns.join(", "),
            on_clause.join(" AND "),
        );

        (sql, total_params)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // ========================================
    // INSERT tests
    // ========================================

    #[test]
    fn test_build_insert_single() {
        let builder = MysqlSqlBuilder;
        let (sql, params) = builder.build_insert("users", &["name", "email"], 1, None, None, None);
        assert_eq!(sql, "INSERT INTO users (name, email) VALUES (?, ?)");
        assert_eq!(params, 2);
    }

    #[test]
    fn test_build_insert_multi() {
        let builder = MysqlSqlBuilder;
        let (sql, params) = builder.build_insert("users", &["name", "email"], 2, None, None, None);
        assert_eq!(sql, "INSERT INTO users (name, email) VALUES (?, ?), (?, ?)");
        assert_eq!(params, 4);
    }

    #[test]
    fn test_build_insert_ignore() {
        // TS: INSERT IGNORE INTO users (name, email) VALUES (?, ?)
        let builder = MysqlSqlBuilder;
        let oc = OnConflict {
            columns: &["email"],
            action: OnConflictAction::DoNothing,
        };
        let (sql, params) =
            builder.build_insert("users", &["name", "email"], 1, Some(&oc), None, None);
        assert_eq!(sql, "INSERT IGNORE INTO users (name, email) VALUES (?, ?)");
        assert_eq!(params, 2);
    }

    #[test]
    fn test_build_insert_on_duplicate_key_update() {
        // TS: INSERT INTO users (...) VALUES (...) ON DUPLICATE KEY UPDATE name = VALUES(name)
        let builder = MysqlSqlBuilder;
        let oc = OnConflict {
            columns: &["email"],
            action: OnConflictAction::DoUpdate(&["name"]),
        };
        let (sql, params) =
            builder.build_insert("users", &["name", "email"], 1, Some(&oc), None, None);
        assert_eq!(
            sql,
            "INSERT INTO users (name, email) VALUES (?, ?) ON DUPLICATE KEY UPDATE name = VALUES(name)"
        );
        assert_eq!(params, 2);
    }

    #[test]
    fn test_build_insert_on_duplicate_key_update_multi() {
        let builder = MysqlSqlBuilder;
        let oc = OnConflict {
            columns: &["email"],
            action: OnConflictAction::DoUpdate(&["name", "email"]),
        };
        let (sql, params) =
            builder.build_insert("users", &["name", "email"], 2, Some(&oc), None, None);
        assert_eq!(
            sql,
            "INSERT INTO users (name, email) VALUES (?, ?), (?, ?) ON DUPLICATE KEY UPDATE name = VALUES(name), email = VALUES(email)"
        );
        assert_eq!(params, 4);
    }

    #[test]
    fn test_build_insert_on_duplicate_key_update_empty_cols_fallback() {
        // DoUpdate(&[]) should fall back to all inserted columns (matches PG driver)
        let builder = MysqlSqlBuilder;
        let oc = OnConflict {
            columns: &["email"],
            action: OnConflictAction::DoUpdate(&[]),
        };
        let (sql, params) =
            builder.build_insert("users", &["name", "email"], 1, Some(&oc), None, None);
        assert_eq!(
            sql,
            "INSERT INTO users (name, email) VALUES (?, ?) ON DUPLICATE KEY UPDATE name = VALUES(name), email = VALUES(email)"
        );
        assert_eq!(params, 2);
    }

    #[test]
    fn test_build_insert_with_returning() {
        // MySQL: appends RETURNING for driver-layer simulation (TS behavior)
        let builder = MysqlSqlBuilder;
        let (sql, _) = builder.build_insert("users", &["name"], 1, None, Some(&["id"]), None);
        assert!(sql.contains("RETURNING"));
        assert!(sql.ends_with("RETURNING id"));
    }

    // ========================================
    // UPDATE MANY tests
    // ========================================

    #[test]
    fn test_build_update_many() {
        // TS: UPDATE users AS t
        //     JOIN (VALUES ROW(?, ?, ?), ROW(?, ?, ?)) AS v(id, name, email)
        //     ON t.id = v.id SET t.name = v.name, t.email = v.email
        let builder = MysqlSqlBuilder;
        let (sql, params) =
            builder.build_update_many("users", &["id"], &["name", "email"], 2, None, None, None);
        assert_eq!(
            sql,
            "UPDATE users AS t JOIN (VALUES ROW(?, ?, ?), ROW(?, ?, ?)) AS v(id, name, email) ON t.id = v.id SET t.name = v.name, t.email = v.email"
        );
        assert_eq!(params, 6);
    }

    #[test]
    fn test_build_update_many_composite_key() {
        let builder = MysqlSqlBuilder;
        let (sql, params) = builder.build_update_many(
            "user_roles",
            &["user_id", "role_id"],
            &["level"],
            2,
            None,
            None,
            None,
        );
        assert_eq!(
            sql,
            "UPDATE user_roles AS t JOIN (VALUES ROW(?, ?, ?), ROW(?, ?, ?)) AS v(user_id, role_id, level) ON t.user_id = v.user_id AND t.role_id = v.role_id SET t.level = v.level"
        );
        assert_eq!(params, 6);
    }

    #[test]
    fn test_build_update_many_single_row() {
        let builder = MysqlSqlBuilder;
        let (sql, params) =
            builder.build_update_many("users", &["id"], &["name"], 1, None, None, None);
        assert_eq!(
            sql,
            "UPDATE users AS t JOIN (VALUES ROW(?, ?)) AS v(id, name) ON t.id = v.id SET t.name = v.name"
        );
        assert_eq!(params, 2);
    }

    // ========================================
    // FIND BY PKEYS tests
    // ========================================

    #[test]
    fn test_build_find_by_pkeys_single() {
        // TS: SELECT * FROM users WHERE id IN (?, ?, ?)
        let builder = MysqlSqlBuilder;
        let (sql, params) = builder.build_find_by_pkeys("users", &["id"], 3, "*", None);
        assert_eq!(sql, "SELECT * FROM users WHERE id IN (?, ?, ?)");
        assert_eq!(params, 3);
    }

    #[test]
    fn test_build_find_by_pkeys_composite() {
        // TS: SELECT t.* FROM users AS t
        //     JOIN (VALUES ROW(?, ?), ROW(?, ?)) AS v(org_id, user_id)
        //     ON t.org_id = v.org_id AND t.user_id = v.user_id
        let builder = MysqlSqlBuilder;
        let (sql, params) =
            builder.build_find_by_pkeys("user_roles", &["user_id", "role_id"], 2, "*", None);
        assert_eq!(
            sql,
            "SELECT t.* FROM user_roles AS t JOIN (VALUES ROW(?, ?), ROW(?, ?)) AS v(user_id, role_id) ON t.user_id = v.user_id AND t.role_id = v.role_id"
        );
        assert_eq!(params, 4);
    }

    #[test]
    fn test_build_find_by_pkeys_with_select_column() {
        let builder = MysqlSqlBuilder;
        let (sql, params) = builder.build_find_by_pkeys("users", &["id"], 2, "id, name", None);
        assert_eq!(sql, "SELECT id, name FROM users WHERE id IN (?, ?)");
        assert_eq!(params, 2);
    }
}
