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, map_global_row_to_ee_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;
        }
        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 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)
}

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 => {
                tracing::info!(
                    "Inserted: ({}) {} [ee_companies]",
                    rk,
                    ee_row.get("name").and_then(|v| v.as_str()).unwrap_or("")
                );
                ee_cache.lock().unwrap().insert(rk);
                rate_add_inserted(1);
            }
            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 => {
                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);
                rate_add_inserted(1);
            }
            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 process_one_row_inserts(
        &self,
        row: &Map<String, Value>,
        hash_cache: &Mutex<HashSet<String>>,
        ee_cache: Option<&Mutex<HashSet<String>>>,
    ) {
        self.insert_one_global(row, hash_cache).await;
        if self.country_upper == "EE" {
            if let Some(ec) = ee_cache {
                if let Some(ee) = map_global_row_to_ee_company(row) {
                    self.insert_one_ee(&ee, ec).await;
                }
            }
        }
    }

    /// Insert `global_companies_registry` rows and, when `country_code == EE`, matching `ee_companies`.
    pub async fn process_rows(
        &self,
        rows: Vec<Map<String, Value>>,
        hash_cache: Arc<Mutex<HashSet<String>>>,
        ee_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 {
            let hc = Arc::clone(&hash_cache);
            let ec = ee_cache.clone();
            let permit = match sem.clone().acquire_owned().await {
                Ok(p) => p,
                Err(_) => continue,
            };

            // To make self passed cleanly to the spawn block, we will clone the components of self.
            // Oh right, self.dexter is Arc, but self is a reference. Let's make self Arced or clone values.
            let dexter_clone = self.dexter.clone();
            let mw = self.max_workers;
            let cc = self.country_upper.clone();

            // Wait, we need an inserter inside the thread!
            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());
                inserter.process_one_row_inserts(&row, &hc, ee_mutex).await;
            }));
        }
        for h in handles {
            let _ = h.await;
        }
    }
}
