use std::collections::HashSet;
use std::sync::{Arc, Mutex};

use serde_json::{Map, Value};
use tokio::sync::Semaphore;

use crate::scrape::athena::{AthenaInsertStatus, DexterAthena};
use crate::scrape::log::rate_add_inserted;
use crate::scrape::mapping::{
    entity_type_from_dutch_rights_form, filter_ee_payload, filter_pl_company_payload,
    filter_ru_payload, map_global_row_to_cz_company, map_global_row_to_ee_company,
    map_global_row_to_pl_company, map_global_row_to_ru_company,
};

pub(crate) fn row_to_gateway_payload(row: &Map<String, Value>) -> Value {
    let mut m = Map::new();
    for (k, v) in row {
        if v.is_null() {
            continue;
        }
        if k == "registries_slug"
            || k == "registries_extra_registration_numbers"
            || k == "registries_pl_source_document"
        {
            continue;
        }
        m.insert(k.clone(), v.clone());
    }
    let rf = row.get("rights_form").and_then(|v| v.as_str());
    if let Some(et) = entity_type_from_dutch_rights_form(rf) {
        m.insert("entity_type".to_string(), Value::String(et.to_string()));
    }
    m.remove("id");
    m.remove("global_company_id");
    Value::Object(m)
}

fn ru_row_to_gateway_payload(row: &Map<String, Value>) -> Value {
    let base = filter_ru_payload(row);
    let mut m = Map::new();
    for (k, v) in base {
        if !v.is_null() {
            m.insert(k, v);
        }
    }
    m.remove("id");
    m.remove("created_at");
    Value::Object(m)
}

fn ee_row_to_gateway_payload(row: &Map<String, Value>) -> Value {
    let base = filter_ee_payload(row);
    let mut m = Map::new();
    for (k, v) in base {
        if !v.is_null() {
            m.insert(k, v);
        }
    }
    m.remove("id");
    m.remove("created_at");
    Value::Object(m)
}

fn plcz_row_to_gateway_payload(row: &Map<String, Value>) -> Value {
    let base = filter_pl_company_payload(row);
    let mut m = Map::new();
    for (k, v) in base {
        if !v.is_null() {
            m.insert(k, v);
        }
    }
    m.remove("id");
    m.remove("created_at");
    Value::Object(m)
}

pub struct AthenaDatabaseInserter {
    dexter: Arc<DexterAthena>,
    max_workers: usize,
    country_upper: String,
}

impl AthenaDatabaseInserter {
    pub fn new(dexter: Arc<DexterAthena>, max_workers: usize, country_upper: String) -> Self {
        Self {
            dexter,
            max_workers,
            country_upper,
        }
    }

    async fn insert_one_ee(&self, ee_row: &Map<String, Value>, ee_cache: &Mutex<HashSet<String>>) {
        let Some(rk) = ee_row
            .get("registrikood")
            .and_then(|v| v.as_str())
            .map(str::to_string)
        else {
            return;
        };
        if rk.is_empty() {
            return;
        }
        {
            let g = ee_cache.lock().unwrap();
            if g.contains(&rk) {
                return;
            }
        }
        let payload = ee_row_to_gateway_payload(ee_row);
        let (status, detail) = self
            .dexter
            .insert_row_with_retry_resilient("ee_companies", payload, self.max_workers)
            .await;
        match status {
            AthenaInsertStatus::Ok => {
                rate_add_inserted(1);
                tracing::info!(
                    "Inserted: ({}) {} [ee_companies]",
                    rk,
                    ee_row.get("name").and_then(|v| v.as_str()).unwrap_or("")
                );
                ee_cache.lock().unwrap().insert(rk);
            }
            AthenaInsertStatus::Duplicate => {
                ee_cache.lock().unwrap().insert(rk);
            }
            AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
                tracing::error!(
                    "Failed ee_companies {}: {:?}",
                    ee_row.get("name").and_then(|v| v.as_str()).unwrap_or(""),
                    detail
                );
            }
        }
    }

    async fn insert_one_global(
        &self,
        row: &Map<String, Value>,
        hash_cache: &Mutex<HashSet<String>>,
    ) {
        let Some(hkey) = row.get("hash").and_then(|v| v.as_str()).map(str::to_string) else {
            return;
        };
        if hkey.is_empty() {
            return;
        }
        {
            let g = hash_cache.lock().unwrap();
            if g.contains(&hkey) {
                return;
            }
        }
        let payload = row_to_gateway_payload(row);
        let (status, detail) = self
            .dexter
            .insert_row_with_retry_resilient("global_companies_registry", payload, self.max_workers)
            .await;
        match status {
            AthenaInsertStatus::Ok => {
                rate_add_inserted(1);
                tracing::info!(
                    "Inserted: ({}) {} [global_companies_registry]",
                    row.get("company_number")
                        .map(|v| v.to_string())
                        .unwrap_or_default(),
                    row.get("name").and_then(|v| v.as_str()).unwrap_or("")
                );
                hash_cache.lock().unwrap().insert(hkey);
            }
            AthenaInsertStatus::Duplicate => {
                hash_cache.lock().unwrap().insert(hkey);
            }
            AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
                tracing::error!(
                    "Failed insert {} ({}): {:?}",
                    row.get("name").and_then(|v| v.as_str()).unwrap_or(""),
                    row.get("company_number")
                        .map(|v| v.to_string())
                        .unwrap_or_default(),
                    detail
                );
            }
        }
    }

    async fn insert_one_ru(
        &self,
        ru_row: &Map<String, Value>,
        ru_cache: &Mutex<HashSet<String>>,
    ) {
        let dedupe = ru_row
            .get("ogrn")
            .or_else(|| ru_row.get("inn"))
            .and_then(|v| v.as_str())
            .map(str::to_string)
            .filter(|s| !s.is_empty());
        let Some(dkey) = dedupe else {
            return;
        };
        {
            let g = ru_cache.lock().unwrap();
            if g.contains(&dkey) {
                return;
            }
        }
        let payload = ru_row_to_gateway_payload(ru_row);
        let (status, detail) = self
            .dexter
            .insert_row_with_retry_resilient("ru_companies", payload, self.max_workers)
            .await;
        match status {
            AthenaInsertStatus::Ok => {
                rate_add_inserted(1);
                tracing::info!(
                    "Inserted: ({}) {} [ru_companies]",
                    dkey,
                    ru_row.get("name").and_then(|v| v.as_str()).unwrap_or("")
                );
                ru_cache.lock().unwrap().insert(dkey);
            }
            AthenaInsertStatus::Duplicate => {
                ru_cache.lock().unwrap().insert(dkey);
            }
            AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
                tracing::error!(
                    "Failed ru_companies {}: {:?}",
                    ru_row.get("name").and_then(|v| v.as_str()).unwrap_or(""),
                    detail
                );
            }
        }
    }

    async fn insert_one_pl(
        &self,
        pl_row: &Map<String, Value>,
        pl_cache: &Mutex<HashSet<String>>,
    ) {
        let Some(rn) = pl_row
            .get("registration_number")
            .and_then(|v| v.as_str())
            .map(str::to_string)
        else {
            return;
        };
        if rn.is_empty() {
            return;
        }
        {
            let g = pl_cache.lock().unwrap();
            if g.contains(&rn) {
                return;
            }
        }
        let payload = plcz_row_to_gateway_payload(pl_row);
        let (status, detail) = self
            .dexter
            .insert_row_with_retry_resilient("pl_companies", payload, self.max_workers)
            .await;
        match status {
            AthenaInsertStatus::Ok => {
                rate_add_inserted(1);
                tracing::info!(
                    "Inserted: ({}) {} [pl_companies]",
                    rn,
                    pl_row.get("name").and_then(|v| v.as_str()).unwrap_or("")
                );
                pl_cache.lock().unwrap().insert(rn);
            }
            AthenaInsertStatus::Duplicate => {
                pl_cache.lock().unwrap().insert(rn);
            }
            AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
                tracing::error!(
                    "Failed pl_companies {}: {:?}",
                    pl_row.get("name").and_then(|v| v.as_str()).unwrap_or(""),
                    detail
                );
            }
        }
    }

    async fn insert_one_cz(
        &self,
        cz_row: &Map<String, Value>,
        cz_cache: &Mutex<HashSet<String>>,
    ) {
        let Some(rn) = cz_row
            .get("registration_number")
            .and_then(|v| v.as_str())
            .map(str::to_string)
        else {
            return;
        };
        if rn.is_empty() {
            return;
        }
        {
            let g = cz_cache.lock().unwrap();
            if g.contains(&rn) {
                return;
            }
        }
        let payload = plcz_row_to_gateway_payload(cz_row);
        let (status, detail) = self
            .dexter
            .insert_row_with_retry_resilient("cz_companies", payload, self.max_workers)
            .await;
        match status {
            AthenaInsertStatus::Ok => {
                rate_add_inserted(1);
                tracing::info!(
                    "Inserted: ({}) {} [cz_companies]",
                    rn,
                    cz_row.get("name").and_then(|v| v.as_str()).unwrap_or("")
                );
                cz_cache.lock().unwrap().insert(rn);
            }
            AthenaInsertStatus::Duplicate => {
                cz_cache.lock().unwrap().insert(rn);
            }
            AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
                tracing::error!(
                    "Failed cz_companies {}: {:?}",
                    cz_row.get("name").and_then(|v| v.as_str()).unwrap_or(""),
                    detail
                );
            }
        }
    }

    async fn process_one_row_inserts(
        &self,
        row: &Map<String, Value>,
        hash_cache: &Mutex<HashSet<String>>,
        ee_cache: Option<&Mutex<HashSet<String>>>,
        ru_cache: Option<&Mutex<HashSet<String>>>,
        pl_cache: Option<&Mutex<HashSet<String>>>,
        cz_cache: Option<&Mutex<HashSet<String>>>,
    ) {
        match self.country_upper.as_str() {
            "EE" => {
                if let (Some(ec), Some(ee)) = (ee_cache, map_global_row_to_ee_company(row)) {
                    tokio::join!(
                        self.insert_one_global(row, hash_cache),
                        self.insert_one_ee(&ee, ec)
                    );
                } else {
                    self.insert_one_global(row, hash_cache).await;
                }
            }
            "RU" => {
                if let (Some(rc), Some(ru)) = (ru_cache, map_global_row_to_ru_company(row)) {
                    tokio::join!(
                        self.insert_one_global(row, hash_cache),
                        self.insert_one_ru(&ru, rc)
                    );
                } else {
                    self.insert_one_global(row, hash_cache).await;
                }
            }
            "PL" => {
                if let (Some(pc), Some(pl)) = (pl_cache, map_global_row_to_pl_company(row)) {
                    tokio::join!(
                        self.insert_one_global(row, hash_cache),
                        self.insert_one_pl(&pl, pc)
                    );
                } else {
                    self.insert_one_global(row, hash_cache).await;
                }
            }
            "CZ" => {
                if let (Some(zc), Some(cz)) = (cz_cache, map_global_row_to_cz_company(row)) {
                    tokio::join!(
                        self.insert_one_global(row, hash_cache),
                        self.insert_one_cz(&cz, zc)
                    );
                } else {
                    self.insert_one_global(row, hash_cache).await;
                }
            }
            _ => {
                self.insert_one_global(row, hash_cache).await;
            }
        }
    }

    /// Insert `global_companies_registry` and country tables (`ee_companies`, `ru_companies`, `pl_companies`, `cz_companies`) when applicable.
    pub async fn process_rows(
        &self,
        rows: Arc<Vec<Map<String, Value>>>,
        hash_cache: Arc<Mutex<HashSet<String>>>,
        ee_cache: Option<Arc<Mutex<HashSet<String>>>>,
        ru_cache: Option<Arc<Mutex<HashSet<String>>>>,
        pl_cache: Option<Arc<Mutex<HashSet<String>>>>,
        cz_cache: Option<Arc<Mutex<HashSet<String>>>>,
    ) {
        if rows.is_empty() {
            tracing::warn!("No rows to insert.");
            return;
        }
        let n = self
            .dexter
            .coordinator()
            .insert_concurrency_for(self.max_workers)
            .await;
        if n < self.max_workers {
            tracing::info!(
                "Insert concurrency capped at {} (--max-workers {}) during gateway cool-down.",
                n,
                self.max_workers
            );
        }
        let sem = Arc::new(Semaphore::new(n));
        let mut handles = vec![];
        for row in rows.iter().cloned() {
            let hc = Arc::clone(&hash_cache);
            let ec = ee_cache.clone();
            let rc = ru_cache.clone();
            let pc = pl_cache.clone();
            let zc = cz_cache.clone();
            let permit = match sem.clone().acquire_owned().await {
                Ok(p) => p,
                Err(_) => continue,
            };

            let dexter_clone = self.dexter.clone();
            let mw = self.max_workers;
            let cc = self.country_upper.clone();

            let inserter = Self::new(dexter_clone, mw, cc);

            handles.push(tokio::spawn(async move {
                let _p = permit;
                let ee_mutex = ec.as_ref().map(|a| a.as_ref());
                let ru_mutex = rc.as_ref().map(|a| a.as_ref());
                let pl_mutex = pc.as_ref().map(|a| a.as_ref());
                let cz_mutex = zc.as_ref().map(|a| a.as_ref());
                inserter
                    .process_one_row_inserts(&row, &hc, ee_mutex, ru_mutex, pl_mutex, cz_mutex)
                    .await;
            }));
        }
        for h in handles {
            let _ = h.await;
        }
    }
}
