//! HTTP scrape API library; optional `scrape` module for Dexter Athena pipelines.

pub mod api;
pub mod audit;
pub mod cache;
pub mod config;
pub mod data;
pub mod http_profile;
pub mod monitoring;
pub mod persisted_config;
pub mod publications;
pub mod proxy;
pub mod request;
pub mod scrape;
pub mod screenshot;

use actix_cors::Cors;
use actix_web::middleware::from_fn;
use actix_web::{web, App, HttpServer};
use athena_rs::AthenaClient;
use colored::Colorize;
use moka::future::Cache;
use prometheus::Registry;
use redis::Client;
use std::collections::HashMap;
use std::env::var;
use std::sync::{Arc, RwLock};
use std::time::Duration;

use crate::audit::AuditBus;
use crate::cache::BloomFilterCache;
use crate::config::load_scraper_configs;
use crate::config::ScraperConfig;
use crate::http_profile::HttpClientProfile;
use crate::monitoring::metrics::{
    metrics_handler, prometheus_http_middleware, register_process_collector, AppMetrics,
};

pub struct AppState {
    pub athena: Option<Arc<AthenaClient>>,
    pub moka: Arc<Cache<String, String>>,
    pub prometheus_registry: Arc<Registry>,
    pub scraper_configs: Arc<HashMap<String, ScraperConfig>>,
    /// Enabled rows from Athena `scraper_overrides` (replaces YAML when present).
    pub scraper_overrides_cache: Arc<RwLock<HashMap<String, ScraperConfig>>>,
    pub metrics: Arc<AppMetrics>,
    pub bloom: Arc<BloomFilterCache>,
    pub redis_client: Option<Arc<Client>>,
    pub audit: Option<AuditBus>,
    pub http_profile_cache: Arc<RwLock<HashMap<String, HttpClientProfile>>>,
    pub screenshot: Option<Arc<crate::screenshot::ScreenshotService>>,
    pub publications: Option<Arc<crate::publications::PublicationS3Service>>,
}

impl AppState {
    /// DB override wins when enabled; otherwise YAML [`scraper_configs`](AppState::scraper_configs).
    pub fn resolve_scraper_config(&self, name: &str) -> Option<ScraperConfig> {
        if let Ok(g) = self.scraper_overrides_cache.read() {
            if let Some(c) = g.get(name) {
                return Some(c.clone());
            }
        }
        self.scraper_configs.get(name).cloned()
    }
}

/// Run the dexter-sentinel HTTP server (Actix). Loads `.env` if present.
pub async fn run_server() -> std::io::Result<()> {
    dotenv::dotenv().ok();

    tracing_subscriber::fmt()
        .with_ansi(true)
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    print_banner();

    tracing::info!(
        "{} {}",
        "→".bright_cyan(),
        "Loading scraper configurations...".bright_white()
    );

    let scraper_configs: HashMap<String, config::ScraperConfig> =
        load_scraper_configs("config/scrapers").unwrap_or_else(|e| {
            tracing::warn!("Failed to load scraper configs: {}", e);
            HashMap::new()
        });

    tracing::info!(
        "{} Loaded {} scraper configuration(s)",
        "✓".bright_green(),
        scraper_configs.len()
    );

    let athena: Option<Arc<AthenaClient>> = data::Provider::new().await.map(|p| p.client);
    if athena.is_some() {
        tracing::info!("{} Athena DB connected", "✓".bright_green());
    } else {
        tracing::warn!(
            "{} Athena not configured (set ATHENA_URL and ATHENA_API_KEY)",
            "⚠".bright_yellow()
        );
    }

    let redis_client: Option<Arc<Client>> = var("REDIS_URL")
        .ok()
        .and_then(|url| Client::open(url).ok())
        .map(Arc::new);

    if redis_client.is_some() {
        tracing::info!("{} Redis connected", "✓".bright_green());
    } else {
        tracing::warn!(
            "{} Redis not configured (REDIS_URL missing)",
            "⚠".bright_yellow()
        );
    }

    let (worker_threads_override, effective_worker_threads) = resolve_worker_threads();
    let moka: Arc<Cache<String, String>> = Arc::new(Cache::new(10_000));
    let prometheus_registry: Arc<Registry> = Arc::new(Registry::new());
    register_process_collector(&prometheus_registry);
    let metrics: Arc<AppMetrics> = Arc::new(
        AppMetrics::new(&prometheus_registry, effective_worker_threads)
            .expect("Failed to init metrics"),
    );
    let bloom: Arc<BloomFilterCache> =
        Arc::new(BloomFilterCache::new(1_000_000, 0.01).expect("valid bloom filter params"));
    {
        let (bits, hashes) = bloom.bloom_prometheus_dimensions();
        metrics.set_bloom_filter_stats(bits, hashes);
    }
    crate::scrape::log::wire_pipeline_prometheus_counters(
        metrics.pipeline_rows_scraped_total.clone(),
        metrics.pipeline_rows_inserted_total.clone(),
    );

    let audit: Option<AuditBus> = athena.clone().map(|a| {
        crate::audit::spawn_audit_flusher(
            Arc::clone(&a),
            Some(metrics.clone()),
            8192,
            100,
            Duration::from_secs(2),
        )
    });

    let http_profile_cache: Arc<RwLock<HashMap<String, HttpClientProfile>>> =
        Arc::new(RwLock::new(HashMap::new()));
    if let Some(ref a) = athena {
        if let Ok(list) = data::http_profiles::list_profiles(a.as_ref()).await {
            let mut m = HashMap::new();
            for p in list {
                if p.enabled {
                    m.insert(p.name.clone(), p);
                }
            }
            if let Ok(mut g) = http_profile_cache.write() {
                *g = m;
            }
        }
    }

    let scraper_overrides_cache: Arc<RwLock<HashMap<String, ScraperConfig>>> =
        Arc::new(RwLock::new(HashMap::new()));
    if let Some(ref a) = athena {
        match data::scraper_overrides::sync_scraper_overrides(a.as_ref(), scraper_overrides_cache.as_ref())
            .await
        {
            Ok(n) => tracing::info!("Loaded {} scraper override(s) from Athena", n),
            Err(e) => tracing::warn!("Scraper overrides preload failed: {}", e),
        }
    }

    let screenshot = crate::screenshot::ScreenshotService::from_env().await.map(Arc::new);
    let publications = crate::publications::PublicationS3Service::from_env().await.map(Arc::new);

    let state: web::Data<AppState> = web::Data::new(AppState {
        athena,
        moka,
        prometheus_registry: prometheus_registry.clone(),
        scraper_configs: Arc::new(scraper_configs),
        scraper_overrides_cache,
        metrics: metrics.clone(),
        bloom,
        redis_client,
        audit,
        http_profile_cache,
        screenshot,
        publications,
    });

    let registry_data: web::Data<Arc<Registry>> = web::Data::new(prometheus_registry);
    let metrics_data: web::Data<Arc<AppMetrics>> = web::Data::new(metrics.clone());

    let host: String = var("HOST").unwrap_or_else(|_| "0.0.0.0".to_string());
    let port: u16 = var("PORT")
        .ok()
        .and_then(|p| p.parse().ok())
        .unwrap_or(8080);

    tracing::info!(
        "{} Server starting on {}:{}",
        "→".bright_cyan(),
        host.bright_white(),
        port.to_string().bright_yellow()
    );
    if let Some(n) = worker_threads_override {
        tracing::info!("{} Actix worker threads set to {}", "→".bright_cyan(), n);
    } else {
        tracing::info!(
            "{} Actix worker threads using default auto count (reported: {})",
            "→".bright_cyan(),
            effective_worker_threads
        );
    }

    let mut server = HttpServer::new(move || {
        let cors: Cors = Cors::permissive();
        App::new()
            .wrap(from_fn(prometheus_http_middleware))
            .wrap(cors)
            .app_data(state.clone())
            .app_data(registry_data.clone())
            .app_data(metrics_data.clone())
            .route("/", web::get().to(api::health::root_handler))
            .route("/health", web::get().to(api::health::health_handler))
            .route(
                "/metrics",
                web::get().to(metrics_handler),
            )
            .route("/scrape", web::post().to(api::scrape::scrape_post_handler))
            .route(
                "/scrape/{config_name}",
                web::get().to(api::scrape::scrape_get_handler),
            )
            .route(
                "/api/cache/registry/add",
                web::post().to(api::cache::registry::add_url_handler),
            )
            .route(
                "/api/cache/registry/check",
                web::get().to(api::cache::registry::check_url_handler),
            )
            .route(
                "/api/http-profiles",
                web::get().to(api::http_profiles::list_http_profiles),
            )
            .route(
                "/api/http-profiles",
                web::post().to(api::http_profiles::post_http_profile),
            )
            .route(
                "/api/http-profiles/{name}",
                web::get().to(api::http_profiles::get_http_profile),
            )
            .route(
                "/api/http-profiles/{name}",
                web::put().to(api::http_profiles::put_http_profile),
            )
            .route(
                "/api/http-profiles/{name}",
                web::delete().to(api::http_profiles::delete_http_profile),
            )
            .route(
                "/api/provider-requests",
                web::get().to(api::provider_requests::list_provider_requests),
            )
            .route(
                "/api/provider-requests",
                web::post().to(api::provider_requests::post_provider_request),
            )
            .route(
                "/api/provider-requests/{id}",
                web::get().to(api::provider_requests::get_provider_request),
            )
            .route(
                "/api/provider-requests/{id}",
                web::put().to(api::provider_requests::put_provider_request),
            )
            .route(
                "/api/provider-requests/{id}",
                web::delete().to(api::provider_requests::delete_provider_request),
            )
            .route(
                "/api/scraper-overrides/reload",
                web::post().to(api::scraper_overrides::reload_scraper_overrides),
            )
            .route(
                "/api/scraper-overrides",
                web::get().to(api::scraper_overrides::list_scraper_overrides),
            )
            .route(
                "/api/scraper-overrides",
                web::post().to(api::scraper_overrides::post_scraper_override),
            )
            .route(
                "/api/scraper-overrides/{name}",
                web::put().to(api::scraper_overrides::put_scraper_override),
            )
            .route(
                "/api/scraper-overrides/{name}",
                web::delete().to(api::scraper_overrides::delete_scraper_override),
            )
            .route("/api/screenshot", web::post().to(api::screenshot::post_screenshot))
            .route(
                "/api/publications/pwc-al/sync",
                web::post().to(api::publications::post_pwc_al_sync),
            )
    });
    if let Some(n) = worker_threads_override {
        server = server.workers(n);
    }
    server.bind((host.as_str(), port))?.run().await
}

fn print_banner() {
    let ver = env!("CARGO_PKG_VERSION");
    println!();
    println!(
        "  {}  {}",
        "dexter-sentinel".bright_cyan().bold(),
        format!("v{}", ver).bright_yellow()
    );
    println!(
        "  {}\n",
        "Config-driven HTTP scrape API · Actix Web · Prometheus".bright_white()
    );
}

fn resolve_worker_threads() -> (Option<usize>, usize) {
    let default_workers = std::thread::available_parallelism()
        .map(usize::from)
        .unwrap_or(1);
    match var("MAX_WORKER_THREADS") {
        Ok(raw) => {
            let trimmed = raw.trim();
            if trimmed.is_empty() {
                return (None, default_workers);
            }
            match trimmed.parse::<usize>() {
                Ok(n) if n > 0 => (Some(n), n),
                _ => {
                    tracing::warn!(
                        "Ignoring MAX_WORKER_THREADS='{}' (expected positive integer)",
                        raw
                    );
                    (None, default_workers)
                }
            }
        }
        Err(_) => (None, default_workers),
    }
}
