use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::{env, fs};
use toml::Value;

pub const CORE_EXTENSIONS_CSV: &str = include_str!("resources/core_extensions.csv");
pub const COMMUNITY_EXTENSIONS_CSV: &str = include_str!("resources/community_extensions.csv");

pub const DUCKDB_CORE_EXTENSIONS: [&str; 38] = [
    "autocomplete",
    "avro",
    "aws",
    "azure",
    "delta",
    "ducklake",
    "quack",
    "encodings",
    "excel",
    "fts",
    "httpfs",
    "http",
    "https",
    "s3",
    "iceberg",
    "icu",
    "inet",
    "jemalloc",
    "json",
    "lance",
    "motherduck",
    "md",
    "mysql",
    "mysql_scanner",
    "parquet",
    "postgres",
    "postgres_scanner",
    "spatial",
    "sqlite",
    "sqlite_scanner",
    "sqlite3",
    "tpcds",
    "tpch",
    "unity_catalog",
    "uc_catalog",
    "ui",
    "vortex",
    "vss",
];

#[cfg(unix)]
fn binary_name() -> &'static str {
    "duckdb"
}

#[cfg(windows)]
fn binary_name() -> &'static str {
    "duckdb.exe"
}

lazy_static! {
    static ref EMPTY_PROFILES: HashMap<String, Profile> = HashMap::new();
}
pub fn duckman_home_dir() -> PathBuf {
    dirs::home_dir()
        .expect("Could not find home directory")
        .join(".duckdb")
}

/// Top-level structure of ~/.duckdb/duckman.toml
#[derive(Debug, Serialize, Deserialize, Default)]
pub struct DuckmanConfig {
    #[serde(skip_serializing)]
    public_key: Option<String>,
    #[serde(default)]
    pub default: Option<String>,
    #[serde(default)]
    pub profile: Option<HashMap<String, Profile>>,
}

impl DuckmanConfig {
    pub fn get_private_key(&self) -> Option<String> {
        if let Some(public_key) = &self.public_key {
            if let Ok(private_key) =
                dotenvx_rs::dotenvx::get_private_key(&Some(public_key.to_string()), &None)
            {
                return Some(private_key);
            }
        }
        None
    }
}

#[derive(Debug, Serialize, Deserialize, Default)]
pub struct Profile {
    #[serde(skip_serializing)]
    pub scope: Option<String>,
    pub description: Option<String>,
    pub duckdb_version: Option<String>,
    pub init_sql: Option<String>,
    pub parquet_key: Option<String>,
    pub settings: Option<toml::Table>,
    pub quack_server: Option<QuackServer>,
    #[serde(default)]
    pub extensions: Vec<String>,
    #[serde(default)]
    pub environment: HashMap<String, String>,
    #[serde(default)]
    pub secret: HashMap<String, toml::Table>,
    #[serde(default)]
    pub bucket: HashMap<String, toml::Table>,
    #[serde(default)]
    pub attached: HashMap<String, AttachedDb>,
    #[serde(default)]
    pub ducklake: HashMap<String, DuckLake>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct Secret {
    pub name: Option<String>,
    #[serde(rename = "type")]
    pub secret_type: String,
    pub provider: Option<String>,
    pub key_id: Option<String>,
    pub secret: Option<String>,
    pub region: Option<String>,
    pub endpoint: Option<String>,
    pub account_name: Option<String>,
    pub connection_string: Option<String>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct AttachedDb {
    #[serde(rename = "type")]
    pub db_type: Option<String>,
    #[serde(rename = "path")]
    pub db_path: String,
    pub sql: Option<String>,
    pub options: Option<toml::Table>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct QuackServer {
    pub uri: Option<String>,
    pub bind_address: Option<String>,
    pub port: Option<u16>,
    pub token: Option<String>,
    pub allow_other_hostname: Option<bool>,
    pub disable_ssl: Option<bool>,
    pub whoami: Option<QuackServerWhoami>,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct QuackServerWhoami {
    pub name: Option<String>,
    pub provider: Option<String>,
    pub hostname: Option<String>,
    pub region: Option<String>,
    pub meta: Option<String>,
}

impl QuackServerWhoami {
    pub fn get_sql(&self) -> String {
        let mut sql = "call quack_identify(".to_owned();
        let mut pairs: Vec<String> = vec![];
        if let Some(name) = &self.name {
            pairs.push(format!("name = '{}'", name));
        }
        if let Some(provider) = &self.provider {
            pairs.push(format!("provider = '{}'", provider));
        }
        if let Some(hostname) = &self.hostname {
            pairs.push(format!("hostname = '{}'", hostname));
        }
        if let Some(region) = &self.region {
            pairs.push(format!("region = '{}'", region));
        }
        if let Some(meta) = &self.meta {
            pairs.push(format!("meta = '{}'", meta));
        }
        sql.push_str(&pairs.join(", "));
        sql.push_str(")");
        sql
    }
}

impl QuackServer {
    pub fn get_uri(&self) -> String {
        if let Some(uri) = &self.uri {
            return uri.to_string();
        }
        if let Some(bind_addr) = &self.bind_address {
            let port = self.port.unwrap_or(9494);
            return format!("quack://{bind_addr}:{port}");
        }
        if let Some(port) = &self.port {
            let bind_addr = &self.bind_address.clone().unwrap_or("localhost".to_string());
            return format!("quack://{bind_addr}:{port}");
        }
        if let Some(allow_other_hostname) = self.allow_other_hostname {
            if allow_other_hostname {
                return "quack:0.0.0.0:9494".to_owned();
            }
        }
        "quack:localhost".to_owned()
    }
    pub fn sql_to_start_server(&self) -> String {
        let quack_uri = self.get_uri();
        let mut sql = format!("call quack_serve('{}'", quack_uri);
        if let Some(token) = &self.token {
            sql.push_str(&format!(", token = '{}'", token));
        }
        if let Some(allow_other_hostname) = &self.allow_other_hostname {
            sql.push_str(&format!(
                ", allow_other_hostname => {}",
                allow_other_hostname
            ));
            if let Some(disable_ssl) = &self.disable_ssl {
                sql.push_str(&format!(", disable_ssl => {}", disable_ssl));
            }
        } else if quack_uri.contains("0.0.0.0") {
            sql.push_str(", allow_other_hostname => true");
        }
        sql.push(')');
        sql
    }
}

impl AttachedDb {
    pub fn attach_options(&self, private_key: &Option<String>) -> String {
        if self.db_type.is_none() && self.options.is_none() {
            return "".to_owned();
        }
        let mut options_text = " (".to_owned();
        if let Some(db_type) = &self.db_type {
            options_text.push_str(&format!("type {},", db_type));
        }
        if let Some(options) = &self.options {
            for (name, value) in options.iter() {
                let mut text_value = convert_toml_value_to_sql_value(private_key, value);
                if name.eq_ignore_ascii_case("secret") {
                    text_value = text_value.trim_matches('\'').to_string();
                }
                options_text.push_str(&format!(" {} {},", name, text_value));
            }
        }
        options_text.remove(options_text.len() - 1);
        options_text.push(')');
        return options_text;
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub struct DuckLake {
    pub catalog_endpoint: String,
    pub data_path: String,
    pub sql: Option<String>,
}

impl DuckmanConfig {
    pub fn home_dir() -> PathBuf {
        duckman_home_dir()
    }

    pub fn versions_dir() -> PathBuf {
        Self::home_dir().join("versions")
    }

    pub fn snippets_dir() -> PathBuf {
        duckman_home_dir().join("snippets")
    }

    pub fn version_dir(version: &str) -> PathBuf {
        Self::versions_dir().join(normalize_duckdb_version(version))
    }

    pub fn version_binary(version: &str) -> PathBuf {
        Self::version_dir(version).join(binary_name())
    }

    pub fn extension_path(duckdb_version: &str, ext_name: &str) -> PathBuf {
        DuckmanConfig::home_dir()
            .join("extensions")
            .join(duckdb_version)
            .join(duckdb_platform_id())
            .join(format!("{}.duckdb_extension", ext_name))
    }

    pub fn config_file() -> PathBuf {
        Self::home_dir().join("duckman.toml")
    }

    pub fn installed_versions() -> Vec<String> {
        let versions_dir = Self::versions_dir();
        if !versions_dir.exists() {
            return Vec::new();
        }
        let mut versions: Vec<String> = fs::read_dir(&versions_dir)
            .map(|rd| {
                rd.filter_map(|e| {
                    let entry = e.ok()?;
                    if entry.path().is_dir() {
                        Some(entry.file_name().to_string_lossy().to_string())
                    } else {
                        None
                    }
                })
                .collect()
            })
            .unwrap_or_default();
        versions.sort();
        versions
    }

    pub fn is_duckdb_installed(duckdb_version: &str) -> bool {
        Self::version_binary(duckdb_version).exists()
    }

    pub fn is_ext_installed(duckdb_version: &str, ext_name: &str) -> bool {
        let buf = Self::extension_path(duckdb_version, ext_name);
        buf.exists()
    }

    pub fn load_global() -> anyhow::Result<Self> {
        Ok(Self::load_from(Self::config_file())?)
    }

    pub fn load() -> anyhow::Result<Self> {
        let mut global_config = Self::load_from(Self::config_file())?;
        // stamp global scope on all profiles from the home config
        if let Some(profiles) = global_config.profile.as_mut() {
            for profile in profiles.values_mut() {
                profile.scope = Some("global".to_string());
            }
        }
        let local_file = Path::new("duckman.toml");
        if local_file.exists() {
            let mut local_config = Self::load_from(local_file)?;
            // stamp local scope on all profiles from the local config
            if let Some(profiles) = local_config.profile.as_mut() {
                for profile in profiles.values_mut() {
                    profile.scope = Some("local".to_string());
                }
            }
            // local default takes priority
            if local_config.default.is_some() {
                global_config.default = local_config.default;
            }
            // merge profiles: local entries override global ones
            if let Some(local_profiles) = local_config.profile {
                let merged = global_config.profile.get_or_insert_with(HashMap::new);
                for (name, profile) in local_profiles {
                    merged.insert(name, profile);
                }
            }
        }
        Ok(global_config)
    }

    pub fn load_from<P: AsRef<Path>>(config_file: P) -> anyhow::Result<Self> {
        if config_file.as_ref().exists() {
            let content = fs::read_to_string(&config_file)?;
            let mut config: DuckmanConfig = toml::from_str(&content)?;
            if let Some(offset) = content.find("dotenv.public.key") {
                let remain = &content[offset + 17..];
                if let Some(offset2) = remain.find('\n') {
                    let pub_key = &remain[..offset2].replace(&['=', ':', '"', '\'', ' '], "");
                    config.public_key = Some(pub_key.to_owned());
                }
            }
            Ok(config)
        } else {
            Ok(DuckmanConfig {
                ..Default::default()
            })
        }
    }

    pub fn get_duckdb_version(&self, profile_name: &Option<String>) -> Option<String> {
        if let Ok(version) = env::var("DUCKDB_VERSION") {
            return Some(version);
        }
        if let Some(profile_name) = profile_name {
            if let Some(profiles) = &self.profile {
                if let Some(selected_profile) = profiles.get(profile_name) {
                    if let Some(profile_duckdb_version) = &selected_profile.duckdb_version {
                        Some(profile_duckdb_version.to_string());
                    }
                }
            }
        }
        if self.default.is_some() {
            return Some(self.default.clone().unwrap());
        }
        let versions = Self::installed_versions();
        if let Some(version) = versions.into_iter().next() {
            return Some(version);
        }
        None
    }

    pub fn save(&self) -> anyhow::Result<()> {
        use toml_edit::{DocumentMut, value};
        let home = Self::home_dir();
        fs::create_dir_all(&home)?;
        // use toml_edit to keep comments, reserved format
        let content = if let Ok(toml_content) = fs::read_to_string(Self::config_file()) {
            let mut doc: DocumentMut = toml_content.parse().expect("invalid toml");
            if let Some(default_version) = &self.default {
                doc["default"] = value(default_version.clone());
            }
            doc.to_string()
        } else {
            toml::to_string_pretty(self)?
        };
        fs::write(Self::config_file(), content)?;
        Ok(())
    }

    pub fn set_default(&mut self, version: &str) {
        let duckdb_version = normalize_duckdb_version(version);
        self.default = Some(duckdb_version);
    }

    pub fn get_profiles(&self) -> &HashMap<String, Profile> {
        if self.profile.is_some() {
            return self.profile.as_ref().unwrap();
        }
        &EMPTY_PROFILES
    }
}

pub fn convert_secret_to_sql(
    private_key: &Option<String>,
    name: &str,
    secret_value: &toml::Table,
) -> String {
    if let Some(sql) = secret_value.get("sql") {
        // replace \n with space and convert to one-line string
        return sql.as_str().unwrap().trim().replace('\n', " ");
    }
    let mut sql = format!("CREATE OR REPLACE SECRET {} (", name);
    for (key, value) in secret_value {
        let mut sql_value = convert_toml_value_to_sql_value(private_key, value);
        if key.eq_ignore_ascii_case("type") {
            sql_value = sql_value.replace(&['"', '\''], "");
        }
        sql.push_str(&format!(" {} {},", key, sql_value));
    }
    sql.remove(sql.len() - 1);
    //sql = sql[0..sql.len() - 2].to_string();
    sql.push_str(");");
    sql
}

pub fn convert_bucket_to_sql(
    private_key: &Option<String>,
    name: &str,
    bucket: &toml::Table,
) -> String {
    if let Some(sql) = bucket.get("sql") {
        // replace \n with space and convert to one-line string
        return sql.as_str().unwrap().trim().replace('\n', " ");
    }
    let mut sql = format!("CREATE OR REPLACE SECRET {} (", name);
    for (key, value) in bucket {
        let mut value = convert_toml_value_to_sql_value(private_key, value);
        if key.eq_ignore_ascii_case("type") || key.eq_ignore_ascii_case("provider") {
            value = value.replace(&['"', '\''], "").to_string();
        }
        sql.push_str(&format!(" {} {},", key, value));
    }
    sql.remove(sql.len() - 1);
    //sql = sql[0..sql.len() - 2].to_string();
    sql.push_str(");");
    sql
}

pub fn convert_attached_db_to_sql(
    private_key: &Option<String>,
    name: &str,
    db: &AttachedDb,
) -> String {
    if let Some(sql) = db.sql.as_ref() {
        // replace \n with space and convert to one-line string
        return sql.trim().replace('\n', " ");
    }
    // such as motherduck, `md:xxx`
    format!(
        "ATTACH '{}' AS {}{};",
        db.db_path,
        name,
        db.attach_options(private_key)
    )
}

pub fn convert_ducklake_to_sql(name: &str, db: &DuckLake) -> String {
    if let Some(sql) = db.sql.as_ref() {
        // replace \n with space and convert to one-line string
        return sql.trim().replace('\n', " ");
    }
    format!(
        "ATTACH  '{}'  as {} (DATA_PATH '{}');",
        db.catalog_endpoint, name, db.data_path
    )
}

pub fn convert_toml_value_to_sql_value(private_key: &Option<String>, value: &Value) -> String {
    match value {
        Value::String(s) => {
            format!("'{}\'", decrypt_value(private_key, s.as_str()))
        }
        Value::Integer(i) => i.to_string(),
        Value::Float(f) => f.to_string(),
        Value::Boolean(b) => b.to_string(),
        Value::Datetime(d) => d.to_string(),
        Value::Array(a) => a
            .iter()
            .map(|v| convert_toml_value_to_sql_value(private_key, v))
            .collect::<Vec<String>>()
            .join(", "),
        Value::Table(t) => {
            let pairs = t
                .iter()
                .map(|(k, v)| {
                    format!(
                        "'{}': {}",
                        k,
                        convert_toml_value_to_sql_value(private_key, v)
                    )
                })
                .collect::<Vec<String>>()
                .join(", ");
            format!("MAP {{{}}}", pairs)
        }
    }
}

pub fn decrypt_value(private_key: &Option<String>, value: &str) -> String {
    if value.starts_with("encrypted:") {
        if let Some(pri_key) = private_key {
            if let Ok(plain_value) = dotenvx_rs::decrypt_dotenvx_item(&pri_key, value) {
                return plain_value;
            }
        }
    }
    value.to_string()
}

pub fn inject_profile(
    duckdb_version: &str,
    profile: &Profile,
    args: &mut Vec<String>,
    new_env: &mut HashMap<String, String>,
    private_key: &Option<String>,
) {
    // Suppression for `--cmd` sentences
    args.push("--cmd".to_string());
    if cfg!(target_os = "windows") {
        args.push(".output NUL".to_string());
    } else {
        args.push(".output /dev/null".to_string());
    }
    // load or install extensions
    for ext_name in profile.extensions.iter() {
        if !DuckmanConfig::is_ext_installed(duckdb_version, ext_name) {
            let sql = if DUCKDB_CORE_EXTENSIONS.contains(&ext_name.as_ref()) {
                format!("install {};", ext_name)
            } else {
                format!("install {} from community;", ext_name)
            };
            args.push("-cmd".to_owned());
            args.push(sql);
        } else {
            args.push("-cmd".to_owned());
            args.push(format!("load {};", ext_name));
        }
    }
    // environment variable
    if !profile.environment.is_empty() {
        new_env.extend(
            profile
                .environment
                .clone()
                .iter()
                .map(|(k, v)| (k.to_uppercase(), decrypt_value(private_key, v))),
        );
    }
    // settings
    if let Some(settings) = &profile.settings {
        for (key, value) in settings {
            let sql_value = convert_toml_value_to_sql_value(private_key, value);
            let sql = format!("SET {} = {};", key, sql_value);
            args.push("-cmd".to_owned());
            args.push(sql);
        }
    }
    // secrets
    for (name, value) in profile.secret.iter() {
        let sql = convert_secret_to_sql(private_key, name, value);
        args.push("-cmd".to_owned());
        args.push(sql);
    }
    // buckets
    for (name, value) in profile.bucket.iter() {
        let sql = convert_bucket_to_sql(private_key, name, value);
        args.push("-cmd".to_owned());
        args.push(sql);
    }
    // parquet key
    if let Some(parquet_key) = &profile.parquet_key {
        let parquet_key = decrypt_value(private_key, parquet_key);
        let sql = format!("PRAGMA add_parquet_key('key256','{}');", parquet_key);
        args.push("-cmd".to_owned());
        args.push(sql);
    }
    // attached databases
    for (name, attached_db) in profile.attached.iter() {
        let sql = convert_attached_db_to_sql(private_key, name, attached_db);
        args.push("-cmd".to_owned());
        args.push(sql);
    }
    // ducklake
    for (name, ducklake) in profile.ducklake.iter() {
        let sql = convert_ducklake_to_sql(name, ducklake);
        args.push("-cmd".to_owned());
        args.push(sql);
    }
    // quack server
    if let Some(quack_server) = &profile.quack_server {
        args.push("-cmd".to_owned());
        args.push("load quack".to_owned());
        args.push("-cmd".to_owned());
        args.push(quack_server.sql_to_start_server());
        if let Some(whoami) = &quack_server.whoami {
            args.push("-cmd".to_owned());
            args.push(whoami.get_sql());
        }
    }
    // init sql
    if let Some(init_sql) = &profile.init_sql {
        let sql = init_sql.replace("\n", " ").trim().to_string();
        if !sql.is_empty() {
            args.push("-cmd".to_owned());
            args.push(sql);
        }
    }
    // Suppression for `--cmd` sentences
    args.push("--cmd".to_string());
    args.push(".output".to_string());
}

pub fn normalize_duckdb_version(version: &str) -> String {
    if version.starts_with('v') {
        version.to_string()
    } else {
        format!("v{}", version)
    }
}

fn duckdb_platform_id() -> &'static str {
    if cfg!(target_os = "macos") && cfg!(target_arch = "aarch64") {
        "osx_arm64"
    } else if cfg!(target_os = "macos") {
        "osx_amd64"
    } else if cfg!(target_os = "linux") && cfg!(target_arch = "aarch64") {
        "linux_arm64"
    } else if cfg!(target_os = "linux") {
        "linux_amd64"
    } else if cfg!(target_os = "windows") {
        "windows_amd64"
    } else {
        "linux_amd64"
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use testresult::TestResult;

    #[test]
    fn test_load_from() -> TestResult {
        let config = DuckmanConfig::load_from("../duckman.toml")?;
        println!("{:?}", config);
        println!("{:#?}", config.get_private_key());
        Ok(())
    }

    #[test]
    fn test_load_profiles() -> TestResult {
        let config = DuckmanConfig::load_from("../duckman.toml")?;
        for entry in config.get_profiles() {
            println!("{}", entry.0);
            println!("{:?}", entry.1);
        }
        Ok(())
    }

    #[test]
    fn test_secrets() -> TestResult {
        let config = DuckmanConfig::load_from("../duckman.toml")?;
        let default_profile = config.get_profiles().get("default").unwrap();
        for (key, value) in default_profile.secret.iter() {
            println!("{}", key);
            println!("sql: {:?}", convert_secret_to_sql(&None, "hello", value));
        }
        Ok(())
    }

    #[test]
    fn test_buckets() -> TestResult {
        let config = DuckmanConfig::load_from("../duckman.toml")?;
        let default_profile = config.get_profiles().get("default").unwrap();
        for (key, value) in default_profile.bucket.iter() {
            println!("{}", key);
            println!("sql: {:?}", convert_bucket_to_sql(&None, key, value));
        }
        Ok(())
    }

    #[test]
    fn test_convert_secret_to_sql() -> TestResult {
        Ok(())
    }

    #[test]
    fn test_ext_installed() -> TestResult {
        let installed = DuckmanConfig::is_ext_installed("v1.5.2", "shellfs");
        println!("installed: {}", installed);
        Ok(())
    }
}
