use std::collections::HashMap;
use std::sync::RwLock;

use anyhow::{anyhow, Result};
use athena_rs::client::AthenaClient;
use athena_rs::OrderDirection;
use serde_json::Value;

use crate::config::ScraperConfig;

const TABLE: &str = "scraper_overrides";

#[derive(Debug, Clone)]
pub struct ScraperOverrideRow {
    pub name: String,
    pub enabled: bool,
    pub config: ScraperConfig,
}

impl ScraperOverrideRow {
    pub fn from_athena_row(row: &Value) -> Result<Self> {
        let o = row.as_object().ok_or_else(|| anyhow!("scraper_overrides row not an object"))?;
        let gv = |k: &str| o.get(k).cloned();
        let name = gv("name")
            .and_then(|v| v.as_str().map(String::from))
            .ok_or_else(|| anyhow!("missing name"))?;
        let enabled = gv("enabled").and_then(|v| v.as_bool()).unwrap_or(true);
        let config_val = gv("config").ok_or_else(|| anyhow!("missing config"))?;
        if config_val.is_null() {
            return Err(anyhow!("config is null"));
        }
        let mut config: ScraperConfig = serde_json::from_value(config_val)
            .map_err(|e| anyhow!("config JSON is not a valid ScraperConfig: {}", e))?;
        config.name = name.clone();
        Ok(Self {
            name,
            enabled,
            config,
        })
    }

    pub fn to_insert_payload(&self, set_updated_at: bool) -> Value {
        let config_json = serde_json::to_value(&self.config).unwrap_or(Value::Null);
        let mut v = serde_json::json!({
            "name": self.name,
            "enabled": self.enabled,
            "config": config_json,
        });
        if set_updated_at {
            if let Value::Object(ref mut o) = v {
                o.insert(
                    "updated_at".into(),
                    Value::String(chrono::Utc::now().to_rfc3339()),
                );
            }
        }
        v
    }

    pub fn to_update_payload(&self) -> Value {
        self.to_insert_payload(true)
    }
}

pub async fn list_overrides(client: &AthenaClient) -> Result<Vec<ScraperOverrideRow>> {
    let res = client
        .fetch(TABLE)
        .where_neq("name", "")
        .order_by("name", OrderDirection::Asc)
        .limit(1000)
        .execute()
        .await
        .map_err(|e| anyhow!("fetch {}: {}", TABLE, e))?;
    let mut out = Vec::new();
    for row in &res.rows {
        out.push(ScraperOverrideRow::from_athena_row(row)?);
    }
    Ok(out)
}

pub async fn get_override_by_name(client: &AthenaClient, name: &str) -> Result<Option<ScraperOverrideRow>> {
    let res = client
        .fetch(TABLE)
        .where_eq("name", name)
        .limit(1)
        .execute()
        .await
        .map_err(|e| anyhow!("fetch scraper override: {}", e))?;
    let Some(row) = res.rows.first() else {
        return Ok(None);
    };
    Ok(Some(ScraperOverrideRow::from_athena_row(row)?))
}

pub async fn insert_override(client: &AthenaClient, row: &ScraperOverrideRow) -> Result<()> {
    if row.name.trim().is_empty() {
        return Err(anyhow!("name required"));
    }
    client
        .insert(TABLE)
        .payload(row.to_insert_payload(true))
        .execute()
        .await
        .map_err(|e| anyhow!("insert scraper override: {}", e))?;
    Ok(())
}

pub async fn update_override(client: &AthenaClient, name: &str, row: &ScraperOverrideRow) -> Result<()> {
    if row.name.trim().is_empty() {
        return Err(anyhow!("name required"));
    }
    client
        .update(TABLE)
        .where_eq("name", name)
        .payload(row.to_update_payload())
        .execute()
        .await
        .map_err(|e| anyhow!("update scraper override: {}", e))?;
    Ok(())
}

pub async fn delete_override(client: &AthenaClient, name: &str) -> Result<()> {
    client
        .delete(TABLE)
        .where_eq("name", name)
        .execute()
        .await
        .map_err(|e| anyhow!("delete scraper override: {}", e))?;
    Ok(())
}

/// Fetches enabled overrides from Athena and replaces `cache` contents. Returns count inserted.
pub async fn sync_scraper_overrides(
    client: &AthenaClient,
    cache: &RwLock<HashMap<String, ScraperConfig>>,
) -> Result<usize> {
    let rows = list_overrides(client).await?;
    let mut m: HashMap<String, ScraperConfig> = HashMap::new();
    for r in rows {
        if r.enabled {
            m.insert(r.name.clone(), r.config);
        }
    }
    let n = m.len();
    let mut cache_guard = cache.write().map_err(|_| anyhow!("scraper overrides cache poisoned"))?;
    *cache_guard = m;
    Ok(n)
}
