use std::sync::Arc;

use super::cache::{write_global_registry_csv, PipelineCache};
use super::config::{GlobalRegistryPipelineConfig, RegistryEngine};
use super::db::AthenaDatabaseInserter;
use super::fetch::fetch_mapped_global_rows;
use super::query::{load_words_file, QueryGenerator};
use super::seed::SeedCsvAccess;
use crate::scrape::athena::DexterAthena;
use crate::scrape::log::rate_add_scraped;
use crate::scrape::mapping::{
    map_global_row_to_cz_company, map_global_row_to_ee_company, map_global_row_to_pl_company,
    map_global_row_to_ru_company,
};

const RANDOM_QUERY_MAX_LEN_CAP: usize = 64;

pub async fn run_randomized_registry_loop(
    http: reqwest::Client,
    dexter: Option<Arc<DexterAthena>>,
    cfg: GlobalRegistryPipelineConfig,
) -> anyhow::Result<()> {
    let mut rng = rand::thread_rng();
    let words = load_words_file(&cfg.words_file);
    if cfg.words_file.is_file() && words.is_empty() {
        tracing::warn!(
            "No words loaded from '{}' (empty or unreadable).",
            cfg.words_file.display()
        );
    }

    let seed_access = if cfg.randomize_no_seed_csv {
        None
    } else if let Some(ref p) = cfg.seed_csv_path {
        SeedCsvAccess::open(p)?
    } else {
        match SeedCsvAccess::open(&cfg.output_csv)? {
            Some(sa) => Some(sa),
            None => {
                let scripts_nl = std::path::PathBuf::from("scripts/nl_companies.csv");
                SeedCsvAccess::open(&scripts_nl)?
            }
        }
    };

    let text_seed = if cfg.query_seed.trim().is_empty() {
        "fi".to_string()
    } else {
        cfg.query_seed.clone()
    };

    let mut max_query_len = cfg.max_query_len;
    let min_query_len = cfg.min_query_len;
    if max_query_len < min_query_len {
        tracing::warn!(
            "--max-query-length ({}) < --min-query-length ({}); clamping max to min.",
            max_query_len,
            min_query_len
        );
        max_query_len = min_query_len;
    }
    let mut effective_max = max_query_len;
    let expand_cap = RANDOM_QUERY_MAX_LEN_CAP.max(max_query_len);

    let cache = PipelineCache::new(cfg.output_csv.clone(), &cfg.country_code);

    let inserter = if cfg.insert {
        dexter.clone().map(|d| {
            AthenaDatabaseInserter::new(d, cfg.max_workers, cfg.country_code.to_uppercase())
        })
    } else {
        None
    };

    let mut generator = QueryGenerator::new(words, seed_access.as_ref(), text_seed);
    let mut last_result_names: Vec<String> = vec![];

    loop {
        let (query, from_fallback) = loop {
            let (q, fb) =
                generator.next(&last_result_names, min_query_len, effective_max, &mut rng);
            if !fb {
                break (q, fb);
            }
            if effective_max >= expand_cap {
                break (q, fb);
            }
            let prev = effective_max;
            effective_max += 1;
            tracing::info!(
                "No novel text queries up to length {prev}; raising max query length to {effective_max}."
            );
        };

        generator.mark_seen(query.clone());
        tracing::info!(
            "Randomized query: '{}' (from_fallback={from_fallback}, {} unique queries)",
            query,
            generator.seen_count()
        );

        let (audit_bus, http_prof) = match &dexter {
            Some(d) => (d.audit().cloned(), d.http_audit_profile.clone()),
            None => (None, None),
        };
        let rows = fetch_mapped_global_rows(
            &http,
            cfg.engine,
            &query,
            &cfg.country_code,
            &cfg.kvk_api_url,
            &cfg.northdata_base_url,
            &cfg.egrul_base_url,
            cfg.registries_io_base_url.as_str(),
            cfg.registries_io_page_limit,
            cfg.northdata_cookie.as_deref(),
            cfg.vat_id_search_engine,
            audit_bus,
            http_prof,
        )
        .await;

        tracing::info!("Fetched {} mapped row(s) for query '{query}'.", rows.len());
        rate_add_scraped(rows.len() as u64);

        if rows.is_empty() {
            tracing::warn!("No rows to save.");
            continue;
        }

        let rows = Arc::new(rows);
        let csv_path = cache.output_csv_path();
        let rows_csv = Arc::clone(&rows);
        let csv_task = tokio::task::spawn_blocking(move || {
            write_global_registry_csv(&csv_path, rows_csv.as_slice(), true)
        });

        let ins_fut = async {
            if let Some(ins) = &inserter {
                ins.process_rows(
                    Arc::clone(&rows),
                    cache.hash_cache_ref(),
                    cache.ee_cache_ref(),
                    cache.ru_cache_ref(),
                    cache.pl_cache_ref(),
                    cache.cz_cache_ref(),
                )
                .await;
            }
        };

        let (csv_join, ()) = tokio::join!(csv_task, ins_fut);
        csv_join
            .map_err(|e| anyhow::anyhow!("csv writer task: {e}"))?
            .map_err(|e| anyhow::anyhow!("csv write: {e}"))?;

        for r in rows.iter() {
            if let Some(h) = r.get("hash").and_then(|v| v.as_str()) {
                if !h.is_empty() {
                    cache.insert_hash(h.to_string());
                }
            }
            if let Some(ee) = map_global_row_to_ee_company(r) {
                if let Some(rk) = ee.get("registrikood").and_then(|v| v.as_str()) {
                    if !rk.is_empty() {
                        cache.insert_ee(rk.to_string());
                    }
                }
            }
            if let Some(ru) = map_global_row_to_ru_company(r) {
                let d = ru
                    .get("ogrn")
                    .or_else(|| ru.get("inn"))
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty());
                if let Some(k) = d {
                    cache.insert_ru(k.to_string());
                }
            }
            if let Some(pl) = map_global_row_to_pl_company(r) {
                if let Some(rn) = pl
                    .get("registration_number")
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                {
                    cache.insert_pl(rn.to_string());
                }
            }
            if let Some(cz) = map_global_row_to_cz_company(r) {
                if let Some(rn) = cz
                    .get("registration_number")
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                {
                    cache.insert_cz(rn.to_string());
                }
            }
        }
        last_result_names = rows
            .iter()
            .filter_map(|r| {
                r.get("name")
                    .and_then(|v| v.as_str())
                    .map(|s| s.trim().to_string())
                    .filter(|s| !s.is_empty())
            })
            .take(120)
            .collect();
    }
}

pub async fn run_single_registry_pass(
    http: &reqwest::Client,
    dexter: Option<Arc<DexterAthena>>,
    cfg: &GlobalRegistryPipelineConfig,
    query: &str,
    append_csv: bool,
) -> anyhow::Result<()> {
    let q = query.trim();
    if q.is_empty() {
        anyhow::bail!("empty query");
    }
    let (audit_bus, http_prof) = match &dexter {
        Some(d) => (d.audit().cloned(), d.http_audit_profile.clone()),
        None => (None, None),
    };
    let rows = fetch_mapped_global_rows(
        http,
        cfg.engine,
        q,
        &cfg.country_code,
        &cfg.kvk_api_url,
        &cfg.northdata_base_url,
        &cfg.egrul_base_url,
        cfg.registries_io_base_url.as_str(),
        cfg.registries_io_page_limit,
        cfg.northdata_cookie.as_deref(),
        cfg.vat_id_search_engine,
        audit_bus,
        http_prof,
    )
    .await;

    tracing::info!("Mapped {} row(s) for single query.", rows.len());
    rate_add_scraped(rows.len() as u64);

    let cache = PipelineCache::new(cfg.output_csv.clone(), &cfg.country_code);

    let inserter = if cfg.insert {
        dexter.map(|d| {
            AthenaDatabaseInserter::new(d, cfg.max_workers, cfg.country_code.to_uppercase())
        })
    } else {
        None
    };

    if rows.is_empty() {
        tracing::warn!("No rows produced.");
        return Ok(());
    }

    let rows = Arc::new(rows);
    let csv_path = cache.output_csv_path();
    let rows_csv = Arc::clone(&rows);
    let append = append_csv;
    let csv_task = tokio::task::spawn_blocking(move || {
        write_global_registry_csv(&csv_path, rows_csv.as_slice(), append)
    });

    let ins_fut = async {
        if let Some(ins) = &inserter {
            ins.process_rows(
                Arc::clone(&rows),
                cache.hash_cache_ref(),
                cache.ee_cache_ref(),
                cache.ru_cache_ref(),
                cache.pl_cache_ref(),
                cache.cz_cache_ref(),
            )
            .await;
        }
    };

    let (csv_join, ()) = tokio::join!(csv_task, ins_fut);
    csv_join
        .map_err(|e| anyhow::anyhow!("csv writer task: {e}"))?
        .map_err(|e| anyhow::anyhow!("csv write: {e}"))?;

    for r in rows.iter() {
        if let Some(h) = r.get("hash").and_then(|v| v.as_str()) {
            if !h.is_empty() {
                cache.insert_hash(h.to_string());
            }
        }
        if let Some(ee) = map_global_row_to_ee_company(r) {
            if let Some(rk) = ee.get("registrikood").and_then(|v| v.as_str()) {
                if !rk.is_empty() {
                    cache.insert_ee(rk.to_string());
                }
            }
        }
        if let Some(ru) = map_global_row_to_ru_company(r) {
            let d = ru
                .get("ogrn")
                .or_else(|| ru.get("inn"))
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty());
            if let Some(k) = d {
                cache.insert_ru(k.to_string());
            }
        }
        if let Some(pl) = map_global_row_to_pl_company(r) {
            if let Some(rn) = pl
                .get("registration_number")
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty())
            {
                cache.insert_pl(rn.to_string());
            }
        }
        if let Some(cz) = map_global_row_to_cz_company(r) {
            if let Some(rn) = cz
                .get("registration_number")
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty())
            {
                cache.insert_cz(rn.to_string());
            }
        }
    }
    Ok(())
}

/// KVK-only randomized loop using a shared `DexterAthena` HTTP client (legacy signature).
pub async fn run_randomized_kvk_loop(
    dexter: Arc<DexterAthena>,
    mut cfg: GlobalRegistryPipelineConfig,
) -> anyhow::Result<()> {
    cfg.engine = RegistryEngine::Kvk;
    let http = dexter.http().clone();
    run_randomized_registry_loop(http, Some(dexter), cfg).await
}
