use std::sync::Arc;

use super::cache::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_ee_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 {
        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 rows = fetch_mapped_global_rows(
            &http,
            cfg.engine,
            &query,
            &cfg.country_code,
            &cfg.kvk_api_url,
            &cfg.northdata_base_url,
            cfg.northdata_cookie.as_deref(),
            cfg.vat_id_search_engine,
        )
        .await;

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

        if let Some(ins) = &inserter {
            ins.process_rows(rows.clone(), cache.hash_cache_ref(), cache.ee_cache_ref())
                .await;
        }

        if !rows.is_empty() {
            cache.save_rows_to_csv(&rows, true)?;
            for r in &rows {
                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());
                        }
                    }
                }
            }
            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();
        } else {
            tracing::warn!("No rows to save.");
        }
    }
}

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 rows = fetch_mapped_global_rows(
        http,
        cfg.engine,
        q,
        &cfg.country_code,
        &cfg.kvk_api_url,
        &cfg.northdata_base_url,
        cfg.northdata_cookie.as_deref(),
        cfg.vat_id_search_engine,
    )
    .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 let Some(ins) = &inserter {
        ins.process_rows(rows.clone(), cache.hash_cache_ref(), cache.ee_cache_ref())
            .await;
    }

    if !rows.is_empty() {
        cache.save_rows_to_csv(&rows, append_csv)?;
        for r in &rows {
            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());
                    }
                }
            }
        }
    } else {
        tracing::warn!("No rows produced.");
    }
    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
}
