//! Batched async audit: `http_request_log` + `db_operation_log` via Athena (non-blocking enqueue).

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

use athena_rs::client::AthenaClient;
use serde_json::{json, Value};
use tokio::sync::mpsc;

use crate::monitoring::metrics::AppMetrics;

pub const MAX_AUDIT_URL_CHARS: usize = 8192;

#[derive(Debug, Clone)]
pub struct HttpAuditMeta {
    pub source: &'static str,
    pub profile_name: Option<String>,
    pub config_name: Option<String>,
    pub extra: Option<Value>,
}

#[derive(Debug)]
enum AuditEvent {
    Http {
        meta: HttpAuditMeta,
        method: String,
        url: String,
        status_code: Option<u16>,
        duration_ms: u64,
        error: Option<String>,
        response_body_bytes: Option<i64>,
    },
    Db {
        table_name: String,
        status: String,
        detail: Option<String>,
        payload_keys: Value,
    },
}

#[derive(Clone)]
pub struct AuditBus {
    tx: mpsc::Sender<AuditEvent>,
    metrics: Option<Arc<AppMetrics>>,
    capacity: usize,
}

impl AuditBus {
    pub fn try_log_http(
        &self,
        meta: HttpAuditMeta,
        method: &str,
        url: &str,
        status_code: Option<u16>,
        duration_ms: u64,
        error: Option<String>,
        response_body_bytes: Option<i64>,
    ) {
        let mut url_s = url.to_string();
        if url_s.len() > MAX_AUDIT_URL_CHARS {
            url_s.truncate(MAX_AUDIT_URL_CHARS);
        }
        let ev = AuditEvent::Http {
            meta,
            method: method.to_string(),
            url: url_s,
            status_code,
            duration_ms,
            error,
            response_body_bytes,
        };
        match self.tx.try_send(ev) {
            Ok(()) => {}
            Err(mpsc::error::TrySendError::Full(_)) | Err(mpsc::error::TrySendError::Closed(_)) => {
                if let Some(m) = &self.metrics {
                    m.audit_log_dropped_total.inc();
                }
                tracing::warn!("audit queue full or closed (capacity {}), dropped HTTP log", self.capacity);
            }
        }
    }

    pub fn try_log_db(
        &self,
        table_name: &str,
        status: &str,
        detail: Option<String>,
        payload_keys: Value,
    ) {
        let ev = AuditEvent::Db {
            table_name: table_name.to_string(),
            status: status.to_string(),
            detail,
            payload_keys,
        };
        match self.tx.try_send(ev) {
            Ok(()) => {}
            Err(mpsc::error::TrySendError::Full(_)) | Err(mpsc::error::TrySendError::Closed(_)) => {
                if let Some(m) = &self.metrics {
                    m.audit_log_dropped_total.inc();
                }
                tracing::warn!("audit queue full or closed, dropped DB log");
            }
        }
    }
}

/// Wrap a `reqwest` send future with HTTP audit (optional).
pub async fn logged_send<F>(
    audit: Option<&AuditBus>,
    send_fut: F,
    method: &'static str,
    url: &str,
    meta: HttpAuditMeta,
) -> reqwest::Result<reqwest::Response>
where
    F: std::future::Future<Output = reqwest::Result<reqwest::Response>>,
{
    let start = Instant::now();
    let res = send_fut.await;
    let ms = start.elapsed().as_millis() as u64;
    if let Some(bus) = audit {
        match &res {
            Ok(r) => {
                let code = Some(r.status().as_u16());
                let clen = r.content_length().map(|x| x as i64);
                bus.try_log_http(meta, method, url, code, ms, None, clen);
            }
            Err(e) => {
                bus.try_log_http(
                    meta,
                    method,
                    url,
                    None,
                    ms,
                    Some(e.to_string()),
                    None,
                );
            }
        }
    }
    res
}

pub fn spawn_audit_flusher(
    athena: Arc<AthenaClient>,
    metrics: Option<Arc<AppMetrics>>,
    capacity: usize,
    batch_max: usize,
    flush_interval: Duration,
) -> AuditBus {
    let (tx, mut rx) = mpsc::channel::<AuditEvent>(capacity);
    let bus = AuditBus {
        tx: tx.clone(),
        metrics: metrics.clone(),
        capacity,
    };

    tokio::spawn(async move {
        let mut buf: Vec<AuditEvent> = Vec::with_capacity(batch_max);
        let mut interval = tokio::time::interval(flush_interval);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        let _ = interval.tick().await;

        async fn flush_batch(
            batch: &mut Vec<AuditEvent>,
            athena: &AthenaClient,
            metrics: &Option<Arc<AppMetrics>>,
        ) {
            if batch.is_empty() {
                return;
            }
            for ev in batch.drain(..) {
                match ev {
                    AuditEvent::Http {
                        meta,
                        method,
                        url,
                        status_code,
                        duration_ms,
                        error,
                        response_body_bytes,
                    } => {
                        let payload = json!({
                            "source": meta.source,
                            "method": method,
                            "url": url,
                            "status_code": status_code.map(|x| x as i64),
                            "duration_ms": duration_ms as i64,
                            "error": error,
                            "profile_name": meta.profile_name,
                            "config_name": meta.config_name,
                            "response_body_bytes": response_body_bytes,
                            "extra": meta.extra,
                        });
                        if athena
                            .insert("http_request_log")
                            .payload(payload)
                            .execute()
                            .await
                            .is_ok()
                        {
                            if let Some(m) = metrics {
                                m.audit_log_flushed_total.inc();
                            }
                        }
                    }
                    AuditEvent::Db {
                        table_name,
                        status,
                        detail,
                        payload_keys,
                    } => {
                        // Wrap keys in an object: bare JSON arrays are bound as text[] by the gateway,
                        // but db_operation_log.payload_keys is jsonb.
                        let payload = json!({
                            "table_name": table_name,
                            "status": status,
                            "detail": detail,
                            "payload_keys": json!({ "keys": payload_keys }),
                        });
                        if athena
                            .insert("db_operation_log")
                            .payload(payload)
                            .execute()
                            .await
                            .is_ok()
                        {
                            if let Some(m) = metrics {
                                m.audit_log_flushed_total.inc();
                            }
                        }
                    }
                }
            }
        }

        loop {
            tokio::select! {
                maybe = rx.recv() => {
                    match maybe {
                        Some(ev) => {
                            buf.push(ev);
                            if buf.len() >= batch_max {
                                flush_batch(&mut buf, &athena, &metrics).await;
                            }
                        }
                        None => {
                            flush_batch(&mut buf, &athena, &metrics).await;
                            break;
                        }
                    }
                }
                _ = interval.tick() => {
                    flush_batch(&mut buf, &athena, &metrics).await;
                }
            }
        }
    });

    bus
}
