//! Dexter Athena gateway client: mirrors `build_athena_headers`, `athena_insert`, and
//! `_coordinate_gateway_backoff` from `scrape_global_registry.py` / `scrape_nl_comps.py`.
//! Uses `athena_rs` when default gateway headers suffice; raw `reqwest` PUT when optional
//! identity / strip-nulls headers are required.

use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use athena_rs::client::backend::BackendError;
use athena_rs::{build_gateway_endpoint, AthenaClient, GATEWAY_INSERT_PATH};
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde_json::{json, Value};
use tokio::sync::Mutex;

const GATEWAY_POLICY_PAUSE_SEC: u64 = 15;
const GATEWAY_COOLDOWN_SEC: u64 = 150;
const GATEWAY_MAX_CONCURRENT_INSERTS: usize = 5;

/// Insert outcome aligned with Python `AthenaInsertStatus` literals.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AthenaInsertStatus {
    Ok,
    Duplicate,
    Error,
    Exception,
}

impl AthenaInsertStatus {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Ok => "ok",
            Self::Duplicate => "duplicate",
            Self::Error => "error",
            Self::Exception => "exception",
        }
    }
}

/// Shared gateway policy backoff + insert concurrency cap during cool-down (Python globals).
pub struct GatewayInsertCoordinator {
    inner: Mutex<CoordinatorState>,
}

struct CoordinatorState {
    cooldown_until: Option<Instant>,
    insert_cap: usize,
}

impl Default for CoordinatorState {
    fn default() -> Self {
        Self {
            cooldown_until: None,
            insert_cap: GATEWAY_MAX_CONCURRENT_INSERTS,
        }
    }
}

impl GatewayInsertCoordinator {
    pub fn new() -> Self {
        Self {
            inner: Mutex::new(CoordinatorState::default()),
        }
    }

    /// `min(max_workers, cap)` while in cool-down, else `max_workers` (see Python batch insert).
    pub async fn insert_concurrency_for(&self, max_workers: usize) -> usize {
        let guard = self.inner.lock().await;
        let now = Instant::now();
        let in_cooldown = guard.cooldown_until.map(|t| now < t).unwrap_or(false);
        if in_cooldown {
            max_workers.min(guard.insert_cap).max(1)
        } else {
            max_workers
        }
    }

    pub async fn in_cooldown(&self) -> bool {
        let guard = self.inner.lock().await;
        guard
            .cooldown_until
            .map(|t| Instant::now() < t)
            .unwrap_or(false)
    }

    /// Under lock: if not already cooling, sleep 15s then set ~150s cool-down and cap inserts.
    /// If already in cool-down, return immediately (matches `_coordinate_gateway_backoff`).
    pub async fn coordinate_backoff_on_gateway_policy_error(&self, batch_max_workers: usize) {
        let guard = self.inner.lock().await;
        let now = Instant::now();
        if guard.cooldown_until.map(|t| now < t).unwrap_or(false) {
            return;
        }
        tracing::warn!(
            "Gateway API key policy unavailable; pausing {}s and throttling inserts for ~{}s.",
            GATEWAY_POLICY_PAUSE_SEC,
            GATEWAY_COOLDOWN_SEC
        );
        drop(guard);
        tokio::time::sleep(Duration::from_secs(GATEWAY_POLICY_PAUSE_SEC)).await;
        let mut guard = self.inner.lock().await;
        guard.cooldown_until = Some(Instant::now() + Duration::from_secs(GATEWAY_COOLDOWN_SEC));
        guard.insert_cap = batch_max_workers.min(GATEWAY_MAX_CONCURRENT_INSERTS).max(1);
    }
}

impl Default for GatewayInsertCoordinator {
    fn default() -> Self {
        Self::new()
    }
}

/// Same header set as Python `build_athena_headers`.
pub fn build_athena_headers(
    athena_key: &str,
    athena_client: &str,
    user_id: Option<&str>,
    company_id: Option<&str>,
    organization_id: Option<&str>,
    strip_nulls: Option<bool>,
) -> HeaderMap {
    let mut m = HeaderMap::new();
    m.insert(
        reqwest::header::CONTENT_TYPE,
        HeaderValue::from_static("application/json"),
    );
    if let Ok(v) = HeaderValue::from_str(athena_client) {
        m.insert(HeaderName::from_static("x-athena-client"), v);
    }
    if !athena_key.is_empty() {
        if let Ok(v) = HeaderValue::from_str(athena_key) {
            m.insert(HeaderName::from_static("apikey"), v.clone());
            m.insert(HeaderName::from_static("x-api-key"), v);
        }
    }
    if let Some(uid) = user_id {
        if let Ok(v) = HeaderValue::from_str(uid) {
            m.insert(HeaderName::from_static("x-user-id"), v);
        }
    }
    if let Some(cid) = company_id {
        if let Ok(v) = HeaderValue::from_str(cid) {
            m.insert(HeaderName::from_static("x-company-id"), v);
        }
    }
    if let Some(oid) = organization_id {
        if let Ok(v) = HeaderValue::from_str(oid) {
            m.insert(HeaderName::from_static("x-organization-id"), v);
        }
    }
    if let Some(strip) = strip_nulls {
        m.insert(
            HeaderName::from_static("x-strip-nulls"),
            HeaderValue::from_static(if strip { "true" } else { "false" }),
        );
    }
    m
}

fn jitter_200_500_ms() -> Duration {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.subsec_nanos() as u64)
        .unwrap_or(0);
    let extra = nanos % 301;
    Duration::from_millis(200 + extra)
}

fn is_gateway_api_key_policy_error(detail: &Value) -> bool {
    if let Some(msg) = detail.get("message").and_then(|v| v.as_str()) {
        if msg.contains("API key policy unavailable") {
            return true;
        }
    }
    let blob = match detail {
        Value::String(s) => s.as_str(),
        _ => return detail.to_string().contains("Gateway API key enforcement"),
    };
    blob.contains("Gateway API key enforcement") || blob.contains("API key policy unavailable")
}

fn sdk_err_to_value(e: BackendError) -> Value {
    Value::String(e.to_string())
}

fn is_duplicate_status_409(text: &str) -> bool {
    text.contains("status=409") || text.contains("status = 409") || text.contains("(status=409,")
}

/// Dexter-facing Athena client (`ATHENA_*` env), separate from dexter-sentinel server config.
pub struct DexterAthena {
    client: AthenaClient,
    base_url: String,
    http: reqwest::Client,
    api_key: String,
    client_name: String,
    user_id: Option<String>,
    company_id: Option<String>,
    organization_id: Option<String>,
    strip_nulls: Option<bool>,
    coordinator: Arc<GatewayInsertCoordinator>,
}

impl DexterAthena {
    /// Explicit gateway URL, API key, and client name (e.g. `dexter-scrape` CLI flags).
    pub async fn from_cli(
        base_url: &str,
        api_key: &str,
        client_name: &str,
        user_id: Option<String>,
        company_id: Option<String>,
        organization_id: Option<String>,
        strip_nulls: Option<bool>,
    ) -> anyhow::Result<Self> {
        let base_trim = base_url.trim_end_matches('/').to_string();

        let client = AthenaClient::new(&base_trim, api_key, client_name).await?;

        Ok(Self {
            client,
            base_url: base_trim,
            http: reqwest::Client::builder()
                .timeout(Duration::from_secs(30))
                .build()?,
            api_key: api_key.to_string(),
            client_name: client_name.to_string(),
            user_id,
            company_id,
            organization_id,
            strip_nulls,
            coordinator: Arc::new(GatewayInsertCoordinator::new()),
        })
    }

    /// `ATHENA_API_KEY` required. Defaults: `ATHENA_URL` = cluster, `ATHENA_CLIENT` = dexter.
    pub async fn from_env() -> anyhow::Result<Self> {
        let api_key = std::env::var("ATHENA_API_KEY")
            .map_err(|_| anyhow::anyhow!("ATHENA_API_KEY is required for Dexter scrape inserts"))?;
        let base_url = std::env::var("ATHENA_URL")
            .unwrap_or_else(|_| "https://cluster.athena-db.com".to_string());
        let client_name = std::env::var("ATHENA_CLIENT").unwrap_or_else(|_| "dexter".to_string());
        let user_id = std::env::var("ATHENA_USER_ID")
            .ok()
            .filter(|s| !s.is_empty());
        let company_id = std::env::var("ATHENA_COMPANY_ID")
            .ok()
            .filter(|s| !s.is_empty());
        let organization_id = std::env::var("ATHENA_ORGANIZATION_ID")
            .ok()
            .filter(|s| !s.is_empty());
        let strip_nulls = std::env::var("ATHENA_STRIP_NULLS").ok().and_then(|s| {
            match s.to_lowercase().as_str() {
                "true" | "1" | "yes" => Some(true),
                "false" | "0" | "no" => Some(false),
                _ => None,
            }
        });
        Self::from_cli(
            &base_url,
            &api_key,
            &client_name,
            user_id,
            company_id,
            organization_id,
            strip_nulls,
        )
        .await
    }

    pub fn coordinator(&self) -> Arc<GatewayInsertCoordinator> {
        Arc::clone(&self.coordinator)
    }

    /// Shared HTTP client (timeouts, TLS) for Northdata/KVK/Bizzy fetches.
    pub fn http(&self) -> &reqwest::Client {
        &self.http
    }

    fn needs_raw_insert(&self) -> bool {
        self.user_id.is_some()
            || self.company_id.is_some()
            || self.organization_id.is_some()
            || self.strip_nulls.is_some()
    }

    async fn insert_via_sdk(&self, table_name: &str, payload: Value) -> Result<(), BackendError> {
        self.client
            .insert(table_name)
            .payload(payload)
            .execute()
            .await?;
        Ok(())
    }

    async fn insert_via_put(
        &self,
        table_name: &str,
        payload: Value,
    ) -> Result<(), (reqwest::StatusCode, Value)> {
        let headers = build_athena_headers(
            &self.api_key,
            &self.client_name,
            self.user_id.as_deref(),
            self.company_id.as_deref(),
            self.organization_id.as_deref(),
            self.strip_nulls,
        );
        let url = build_gateway_endpoint(&self.base_url, GATEWAY_INSERT_PATH);
        let body = json!({
            "table_name": table_name,
            "insert_body": payload,
        });
        let mut req = self.http.put(url).json(&body);
        for (k, v) in headers.iter() {
            req = req.header(k, v);
        }
        let resp = req.send().await.map_err(|e| {
            (
                reqwest::StatusCode::REQUEST_TIMEOUT,
                Value::String(e.to_string()),
            )
        })?;
        let status = resp.status();
        let text = resp.text().await.unwrap_or_default();
        if status.is_success() {
            return Ok(());
        }
        if status.as_u16() == 409 {
            return Err((status, Value::String(text)));
        }
        let detail = serde_json::from_str(&text).unwrap_or(Value::String(text));
        Err((status, detail))
    }

    async fn insert_once_inner(
        &self,
        table_name: &str,
        payload: Value,
    ) -> (AthenaInsertStatus, Value) {
        if self.needs_raw_insert() {
            match self.insert_via_put(table_name, payload).await {
                Ok(()) => (AthenaInsertStatus::Ok, Value::Null),
                Err((status, detail)) => {
                    if status.as_u16() == 409 {
                        (AthenaInsertStatus::Duplicate, detail)
                    } else {
                        (AthenaInsertStatus::Error, detail)
                    }
                }
            }
        } else {
            match self.insert_via_sdk(table_name, payload).await {
                Ok(()) => (AthenaInsertStatus::Ok, Value::Null),
                Err(e) => {
                    let s = e.to_string();
                    if is_duplicate_status_409(&s) {
                        return (AthenaInsertStatus::Duplicate, Value::String(s.clone()));
                    }
                    (AthenaInsertStatus::Error, sdk_err_to_value(e))
                }
            }
        }
    }

    /// Single insert with cool-down jitter (Python `athena_insert`).
    pub async fn insert_row(
        &self,
        table_name: &str,
        payload: Value,
    ) -> (AthenaInsertStatus, Value) {
        if self.coordinator.in_cooldown().await {
            tokio::time::sleep(jitter_200_500_ms()).await;
        }
        self.insert_once_inner(table_name, payload).await
    }

    /// Like `_athena_insert_with_gateway_retry`: one coordinated backoff + retry on policy errors.
    pub async fn insert_row_with_retry(
        &self,
        table_name: &str,
        payload: Value,
        batch_max_workers: usize,
    ) -> (AthenaInsertStatus, Value) {
        let (st, detail) = self.insert_row(table_name, payload.clone()).await;
        if st == AthenaInsertStatus::Error && is_gateway_api_key_policy_error(&detail) {
            self.coordinator
                .coordinate_backoff_on_gateway_policy_error(batch_max_workers)
                .await;
            return self.insert_row(table_name, payload).await;
        }
        (st, detail)
    }
}

impl DexterAthena {
    /// Network failure mapping to `exception` (Python bare `except` path for `aiohttp`).
    pub async fn insert_row_resilient(
        &self,
        table_name: &str,
        payload: Value,
    ) -> (AthenaInsertStatus, Value) {
        if self.coordinator.in_cooldown().await {
            tokio::time::sleep(jitter_200_500_ms()).await;
        }
        if self.needs_raw_insert() {
            let headers = build_athena_headers(
                &self.api_key,
                &self.client_name,
                self.user_id.as_deref(),
                self.company_id.as_deref(),
                self.organization_id.as_deref(),
                self.strip_nulls,
            );
            let url = build_gateway_endpoint(&self.base_url, GATEWAY_INSERT_PATH);
            let body = json!({
                "table_name": table_name,
                "insert_body": payload,
            });
            let mut req = self.http.put(url).json(&body);
            for (k, v) in headers.iter() {
                req = req.header(k, v);
            }
            let resp = match req.send().await {
                Ok(r) => r,
                Err(e) => {
                    return (AthenaInsertStatus::Exception, Value::String(e.to_string()));
                }
            };
            let status = resp.status();
            let text = match resp.text().await {
                Ok(t) => t,
                Err(e) => {
                    return (AthenaInsertStatus::Exception, Value::String(e.to_string()));
                }
            };
            if status.is_success() {
                return (AthenaInsertStatus::Ok, Value::Null);
            }
            if status.as_u16() == 409 {
                return (AthenaInsertStatus::Duplicate, Value::String(text));
            }
            let detail = serde_json::from_str(&text).unwrap_or(Value::String(text));
            return (AthenaInsertStatus::Error, detail);
        }

        match self.insert_via_sdk(table_name, payload).await {
            Ok(()) => (AthenaInsertStatus::Ok, Value::Null),
            Err(e) => {
                let s = e.to_string();
                if is_duplicate_status_409(&s) {
                    (AthenaInsertStatus::Duplicate, Value::String(s))
                } else {
                    (AthenaInsertStatus::Error, sdk_err_to_value(e))
                }
            }
        }
    }

    pub async fn insert_row_with_retry_resilient(
        &self,
        table_name: &str,
        payload: Value,
        batch_max_workers: usize,
    ) -> (AthenaInsertStatus, Value) {
        let (st, detail) = self.insert_row_resilient(table_name, payload.clone()).await;
        if st == AthenaInsertStatus::Error && is_gateway_api_key_policy_error(&detail) {
            self.coordinator
                .coordinate_backoff_on_gateway_policy_error(batch_max_workers)
                .await;
            return self.insert_row_resilient(table_name, payload).await;
        }
        (st, detail)
    }
}
