use actix_web::{HttpRequest, HttpResponse, http::StatusCode};
use serde_json::{Map, Value, json};
use sqlx::postgres::PgPool;
use sqlx::{Pool, Postgres};
use std::collections::HashMap;
use std::str::SplitN;
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::mpsc;
use tracing::{error, warn};
use uuid::Uuid;

use crate::AppState;
use crate::api::headers::x_athena_client::x_athena_client;
use crate::api::headers::{x_company_id, x_organization_id, x_user_id};
use crate::config_validation::runtime_env_settings;
use crate::data::api_keys::resolve_internal_api_key_id;
use crate::data::clients::touch_athena_client_last_seen;
use crate::drivers::postgresql::schema_cache::{
    get_public_table_column_types, postgres_column_descriptor_is_bigint,
    postgres_column_descriptor_is_timestamptz,
};
use crate::drivers::supabase::supabase;
use crate::utils::client_stats_batcher::ClientStatsBatcher;
use crate::utils::logging_task_limiter::spawn_best_effort_logging_task;

static LAST_REALTIME_REGISTRY_PURGE_TS: AtomicI64 = AtomicI64::new(0);

#[derive(Clone, Debug)]
struct GatewayLogBatcherConfig {
    channel_capacity: usize,
    flush_interval: Duration,
    flush_max_rows: usize,
}

impl Default for GatewayLogBatcherConfig {
    fn default() -> Self {
        Self {
            channel_capacity: 50_000,
            flush_interval: Duration::from_millis(100),
            flush_max_rows: 256,
        }
    }
}

enum GatewayLogBatchCommand {
    Request(RequestLogEntry),
    Operation(OperationLogEntry),
}

#[derive(Default)]
struct GatewayLogBatchState {
    request_rows: Vec<RequestLogEntry>,
    operation_rows: Vec<OperationLogEntry>,
}

impl GatewayLogBatchState {
    fn push(&mut self, command: GatewayLogBatchCommand) {
        match command {
            GatewayLogBatchCommand::Request(row) => self.request_rows.push(row),
            GatewayLogBatchCommand::Operation(row) => self.operation_rows.push(row),
        }
    }

    fn len(&self) -> usize {
        self.request_rows
            .len()
            .saturating_add(self.operation_rows.len())
    }

    fn take_request_rows(&mut self) -> Vec<RequestLogEntry> {
        std::mem::take(&mut self.request_rows)
    }

    fn take_operation_rows(&mut self) -> Vec<OperationLogEntry> {
        std::mem::take(&mut self.operation_rows)
    }

    fn is_empty(&self) -> bool {
        self.request_rows.is_empty() && self.operation_rows.is_empty()
    }
}

/// Batches gateway request/operation log writes into multi-row inserts.
pub struct GatewayLogBatcher {
    tx: mpsc::Sender<GatewayLogBatchCommand>,
}

impl GatewayLogBatcher {
    pub fn spawn(pool: PgPool, client_stats: Option<Arc<ClientStatsBatcher>>) -> Self {
        let config: GatewayLogBatcherConfig = GatewayLogBatcherConfig::default();
        let cap: usize = config.channel_capacity.max(1);
        let (tx, rx) = mpsc::channel(cap);
        tokio::spawn(run_gateway_log_batcher_worker(
            rx,
            pool,
            client_stats,
            config,
        ));
        Self { tx }
    }

    fn try_enqueue_request(&self, entry: RequestLogEntry) -> Result<(), RequestLogEntry> {
        self.try_send(GatewayLogBatchCommand::Request(entry))
            .map_err(|cmd| match cmd {
                GatewayLogBatchCommand::Request(entry) => entry,
                GatewayLogBatchCommand::Operation(_) => {
                    unreachable!("request enqueue returned operation command")
                }
            })
    }

    fn try_enqueue_operation(&self, entry: OperationLogEntry) -> Result<(), OperationLogEntry> {
        self.try_send(GatewayLogBatchCommand::Operation(entry))
            .map_err(|cmd| match cmd {
                GatewayLogBatchCommand::Operation(entry) => entry,
                GatewayLogBatchCommand::Request(_) => {
                    unreachable!("operation enqueue returned request command")
                }
            })
    }

    fn try_send(&self, command: GatewayLogBatchCommand) -> Result<(), GatewayLogBatchCommand> {
        match self.tx.try_send(command) {
            Ok(()) => Ok(()),
            Err(mpsc::error::TrySendError::Full(command)) => {
                warn!(
                    target: "athena_rs::gateway_log_batcher",
                    "gateway log batcher channel full; falling back to direct log write"
                );
                Err(command)
            }
            Err(mpsc::error::TrySendError::Closed(command)) => Err(command),
        }
    }
}

async fn run_gateway_log_batcher_worker(
    mut rx: mpsc::Receiver<GatewayLogBatchCommand>,
    pool: PgPool,
    client_stats: Option<Arc<ClientStatsBatcher>>,
    config: GatewayLogBatcherConfig,
) {
    let mut tick: tokio::time::Interval = tokio::time::interval(config.flush_interval);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut state: GatewayLogBatchState = GatewayLogBatchState::default();

    loop {
        tokio::select! {
            biased;
            cmd = rx.recv() => {
                match cmd {
                    Some(command) => {
                        state.push(command);
                        if state.len() >= config.flush_max_rows {
                            flush_gateway_log_batches(&pool, client_stats.as_ref(), &mut state).await;
                        }
                    }
                    None => {
                        flush_gateway_log_batches(&pool, client_stats.as_ref(), &mut state).await;
                        return;
                    }
                }
            }
            _ = tick.tick() => {
                flush_gateway_log_batches(&pool, client_stats.as_ref(), &mut state).await;
            }
        }
    }
}

async fn flush_gateway_log_batches(
    pool: &PgPool,
    client_stats: Option<&Arc<ClientStatsBatcher>>,
    state: &mut GatewayLogBatchState,
) {
    if state.is_empty() {
        return;
    }

    let request_rows = state.take_request_rows();
    if !request_rows.is_empty() {
        flush_request_log_batch(pool, client_stats, request_rows).await;
    }

    let operation_rows = state.take_operation_rows();
    if !operation_rows.is_empty() {
        flush_operation_log_batch(pool, client_stats, operation_rows).await;
    }
}

/// Structured metadata returned to callers so downstream handlers can correlate their work.
pub struct LoggedRequest {
    pub request_id: String,
    pub client_name: String,
    pub method: String,
    pub path: String,
    pub status_code: u16,
    pub time: i64,
}

#[derive(Clone, Debug, Default)]
pub struct RequestCompletionLogContext {
    pub status_code: Option<u16>,
    pub duration_ms: Option<u128>,
    pub cached: Option<bool>,
    pub cache_lookup_outcome: Option<String>,
    pub cache_source: Option<String>,
    pub operation: Option<String>,
    pub table_name: Option<String>,
}

#[derive(Clone, Debug, Default)]
pub struct RequestAuthLogContext {
    pub api_key_id: Option<String>,
    pub internal_api_key_id: Option<i64>,
    pub presented_api_key_public_id: Option<String>,
    pub presented_api_key_hash: Option<String>,
    pub presented_api_key_salt: Option<String>,
    pub api_key_authenticated: bool,
    pub api_key_authorized: bool,
    pub api_key_enforced: bool,
    pub api_key_auth_reason: Option<String>,
}

#[derive(Clone, Debug, Default)]
pub struct RouteRequestLogContext {
    pub request_id: Option<String>,
    pub source: Option<String>,
    pub outcome: Option<String>,
    pub method: Option<String>,
    pub path: Option<String>,
    pub host: Option<String>,
    pub wildcard_host_pattern: Option<String>,
    pub wildcard_prefix: Option<String>,
    pub route_key: Option<String>,
    pub public_op: Option<String>,
    pub resolved_client: Option<String>,
    pub status_code: Option<u16>,
    pub duration_ms: Option<u128>,
    pub error_message: Option<String>,
    pub metadata: Option<Value>,
}

fn is_sensitive_field_name(name: &str) -> bool {
    let normalized: String = name.to_ascii_lowercase().replace('-', "").replace('_', "");
    [
        "authorization",
        "apikey",
        "xapikey",
        "xathenakey",
        "cookie",
        "setcookie",
        "token",
        "secret",
        "password",
        "passwd",
        "jwt",
    ]
    .iter()
    .any(|needle| normalized.contains(needle))
}

fn redact_query_string(query: &str) -> String {
    if query.is_empty() {
        return "NOT_FOUND".to_string();
    }

    query
        .split('&')
        .map(|part| {
            let mut split: SplitN<'_, char> = part.splitn(2, '=');
            let key: &str = split.next().unwrap_or_default();
            let value: Option<&str> = split.next();
            if is_sensitive_field_name(key) {
                if value.is_some() {
                    format!("{}=[REDACTED]", key)
                } else {
                    format!("{}=", key)
                }
            } else {
                part.to_string()
            }
        })
        .collect::<Vec<String>>()
        .join("&")
}

fn redact_json_value(value: Value) -> Value {
    match value {
        Value::Object(map) => {
            let mut redacted: Map<String, Value> = Map::new();
            for (key, nested_value) in map {
                if is_sensitive_field_name(&key) {
                    redacted.insert(key, Value::String("[REDACTED]".to_string()));
                } else {
                    redacted.insert(key, redact_json_value(nested_value));
                }
            }
            Value::Object(redacted)
        }
        Value::Array(list) => Value::Array(list.into_iter().map(redact_json_value).collect()),
        other => other,
    }
}

fn redact_optional_secret(value: Option<String>) -> Option<String> {
    value.map(|_| "[REDACTED]".to_string())
}

fn normalize_ip(x_real_ip: &str, req: &HttpRequest) -> String {
    if !x_real_ip.is_empty() && x_real_ip != "NOT_FOUND" {
        return x_real_ip.to_string();
    }

    if let Some(forwarded) = req
        .headers()
        .get("X-Forwarded-For")
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.split(',').next())
        .map(|v| v.trim())
        .filter(|v| !v.is_empty())
    {
        return forwarded.to_string();
    }

    req.peer_addr()
        .map(|addr| addr.ip().to_string())
        .unwrap_or_else(|| "NOT_FOUND".to_string())
}

fn build_athena_url(req: &HttpRequest, host: &str) -> String {
    if host.is_empty() || host == "NOT_FOUND" {
        return "NOT_FOUND".to_string();
    }

    let scheme = if let Some(proto) = req
        .headers()
        .get("X-Forwarded-Proto")
        .and_then(|v| v.to_str().ok())
        .filter(|v| !v.is_empty())
    {
        proto.to_string()
    } else {
        req.connection_info().scheme().to_string()
    };

    format!("{}://{}", scheme, host)
}

/// Log the incoming HTTP request to Supabase (legacy behavior) and Postgres when enabled.
pub fn log_request(
    req: HttpRequest,
    state: Option<&AppState>,
    request_id: Option<String>,
    auth_context: Option<&RequestAuthLogContext>,
) -> LoggedRequest {
    let client_name: String = x_athena_client(&req);
    let user_id: String = x_user_id::get_x_user_id(&req).unwrap_or_else(|| "NOT_FOUND".to_string());
    let company_id: String =
        x_company_id::get_x_company_id(&req).unwrap_or_else(|| "NOT_FOUND".to_string());
    let organization_id: String =
        x_organization_id::get_x_organization_id(&req).unwrap_or_else(|| "NOT_FOUND".to_string());

    let request_id: String = request_id.unwrap_or_else(|| Uuid::new_v4().to_string());
    let auth_context: RequestAuthLogContext = auth_context.cloned().unwrap_or_default();
    let body_json: Value = req
        .match_info()
        .get("body")
        .and_then(|body| serde_json::from_str(body).ok())
        .unwrap_or_else(|| json!({}));
    let body_json = redact_json_value(body_json);

    let query_string: String = redact_query_string(req.query_string());

    let headers: HashMap<String, String> = req
        .headers()
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();

    let redacted_headers: HashMap<String, String> = headers
        .iter()
        .map(|(key, value)| {
            if is_sensitive_field_name(key) {
                (key.clone(), "[REDACTED]".to_string())
            } else {
                (key.clone(), value.clone())
            }
        })
        .collect();

    let mut headers_map: serde_json::Map<String, Value> = serde_json::Map::new();
    for (key, value) in redacted_headers.iter() {
        headers_map.insert(key.clone(), Value::String(value.clone()));
    }
    let headers_json: Value = Value::Object(headers_map);

    let method: String = req.method().as_str().to_string();
    let path: String = req.path().to_string();
    let status_http_code: u16 = req
        .app_data::<HttpResponse>()
        .map(|resp| resp.status().as_u16())
        .unwrap_or(0);

    let x_real_ip: String = headers
        .get("X-Real-IP")
        .unwrap_or(&"NOT_FOUND".to_string())
        .to_string();
    let normalized_ip: String = normalize_ip(&x_real_ip, &req);

    let user_agent: String = req
        .headers()
        .get("User-Agent")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("NOT_FOUND")
        .to_string();

    if user_agent
        == "Better Stack Better Uptime Bot Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
    {
        return LoggedRequest {
            request_id,
            client_name,
            method,
            path,
            status_code: status_http_code,
            time: 0,
        };
    }

    let time: i64 = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_else(|_| std::time::Duration::new(0, 0))
        .as_secs() as i64;

    let host: String = req
        .headers()
        .get("Host")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("NOT_FOUND")
        .to_string();
    let athena_url = build_athena_url(&req, &host);

    let cache_control: String = req
        .headers()
        .get("Cache-Control")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("NOT_FOUND")
        .to_string();

    let cache_control_lc: String = cache_control.to_ascii_lowercase();
    let cached: bool = !cache_control_lc.contains("no-cache");

    let request_data: Value = json!({
        "object": "request",
        "request_id": request_id,
        "user_id": user_id,
        "company_id": company_id,
        "api_key_id": auth_context.api_key_id.clone(),
        "presented_api_key_public_id": auth_context.presented_api_key_public_id.clone(),
        "presented_api_key_hash": redact_optional_secret(auth_context.presented_api_key_hash.clone()),
        "presented_api_key_salt": redact_optional_secret(auth_context.presented_api_key_salt.clone()),
        "api_key_authenticated": auth_context.api_key_authenticated,
        "api_key_authorized": auth_context.api_key_authorized,
        "api_key_enforced": auth_context.api_key_enforced,
        "api_key_auth_reason": auth_context.api_key_auth_reason.clone(),
        "body": body_json.clone(),
        "query_string": query_string,
        "headers": headers_json.clone(),
        "method": method,
        "path": path,
        "http_code": status_http_code,
        "ipv4": normalized_ip,
        "user_agent": user_agent,
        "time": time,
        "host": host,
        "athena_url": athena_url,
        "cached": cached
    });

    if let Some(state) = state
        && let Some(logging_client) = state.logging_client_name.as_ref()
        && let Some(pool) = state.pg_registry.get_pool(logging_client)
    {
        let entry: RequestLogEntry = RequestLogEntry {
            request_id: request_id.clone(),
            client: client_name.clone(),
            method: method.clone(),
            path: path.clone(),
            query_string: query_string.clone(),
            status_code: status_http_code as i32,
            ipv4: normalized_ip.clone(),
            user_agent: user_agent.clone(),
            headers: headers_json.clone(),
            body: body_json.clone(),
            user_id: user_id.clone(),
            company_id: company_id.clone(),
            organization_id: organization_id.clone(),
            api_key_id: auth_context.api_key_id.clone(),
            internal_api_key_id: auth_context.internal_api_key_id,
            presented_api_key_public_id: auth_context.presented_api_key_public_id.clone(),
            presented_api_key_hash: redact_optional_secret(
                auth_context.presented_api_key_hash.clone(),
            ),
            presented_api_key_salt: redact_optional_secret(
                auth_context.presented_api_key_salt.clone(),
            ),
            api_key_authenticated: auth_context.api_key_authenticated,
            api_key_authorized: auth_context.api_key_authorized,
            api_key_enforced: auth_context.api_key_enforced,
            api_key_auth_reason: auth_context.api_key_auth_reason.clone(),
            host: host.clone(),
            athena_url: athena_url.clone(),
            cached,
            duration_ms: None,
            cache_lookup_outcome: None,
            cache_source: None,
            operation: None,
            table_name: None,
            time,
        };
        let mut fallback_entry: Option<RequestLogEntry> = None;
        if let Some(gateway_log_batcher) = state.gateway_log_batcher.as_ref() {
            if let Err(entry) = gateway_log_batcher.try_enqueue_request(entry) {
                fallback_entry = Some(entry);
            }
        } else {
            fallback_entry = Some(entry);
        }

        if let Some(entry) = fallback_entry {
            let pool_clone: Pool<Postgres> = pool.clone();
            let client_stats: Option<Arc<ClientStatsBatcher>> = state.client_stats_batcher.clone();
            let logging_task_limiter = state.logging_task_limiter.clone();
            spawn_best_effort_logging_task(
                logging_task_limiter,
                "gateway_request_log",
                async move {
                    if let Err(err) = insert_request_log(pool_clone, client_stats, entry).await {
                        error!(error = %err, "failed to write gateway request log");
                    }
                },
            );
        }
    }

    let request_data_clone: Value = request_data.clone();
    actix_web::rt::spawn(async move {
        if let Ok(client) = supabase().await {
            let _ = client
                .insert("event_log_api", request_data_clone.clone())
                .await;
        }
    });

    LoggedRequest {
        request_id,
        client_name,
        method,
        path,
        status_code: status_http_code,
        time,
    }
}

/// Logs gateway operations (insert/fetch/update/delete/query/ping) to Postgres when configured.
pub fn log_operation_event(
    state: Option<&AppState>,
    request: &LoggedRequest,
    operation: &str,
    table_name: Option<&str>,
    duration_ms: u128,
    status: StatusCode,
    details: Option<Value>,
) {
    if let Some(state) = state
        && let Some(logging_client) = state.logging_client_name.as_ref()
        && let Some(pool) = state.pg_registry.get_pool(logging_client)
    {
        let details_value: Value = details.unwrap_or_else(|| json!({}));

        // Determine if this is an error based on status code
        let is_error: bool = status.is_client_error() || status.is_server_error();

        // Extract message from details JSON if present
        let message: Option<String> = details_value
            .get("message")
            .and_then(|v| v.as_str())
            .or_else(|| details_value.get("error").and_then(|v| v.as_str()))
            .or_else(|| details_value.get("error_code").and_then(|v| v.as_str()))
            .map(|s| s.to_string());

        // Extract cache_key from details JSON if present
        let cache_key: Option<String> = details_value
            .get("cache_key")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        let cache_lookup_outcome: Option<String> = details_value
            .get("cache_lookup_outcome")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        let cache_source: Option<String> = details_value
            .get("cache_source")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        let cached: Option<bool> = details_value.get("cached").and_then(|v| v.as_bool());

        state.metrics_state.record_gateway_operation(
            &request.client_name,
            table_name,
            operation,
            request.method.as_str(),
            request.path.as_str(),
            status.as_u16(),
            cache_lookup_outcome.as_deref(),
            cache_source.as_deref(),
            duration_ms as f64 / 1000.0,
        );

        let entry: OperationLogEntry = OperationLogEntry {
            request_id: request.request_id.clone(),
            operation: operation.to_string(),
            table_name: table_name.map(|value| value.to_string()),
            client: request.client_name.clone(),
            method: request.method.clone(),
            path: request.path.clone(),
            status_code: status.as_u16() as i32,
            duration_ms: duration_ms as i64,
            details: details_value,
            time: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_else(|_| std::time::Duration::new(0, 0))
                .as_secs() as i64,
            error: is_error,
            message,
            cache_key,
            cache_lookup_outcome,
            cache_source,
            cached,
        };
        let mut fallback_entry: Option<OperationLogEntry> = None;
        if let Some(gateway_log_batcher) = state.gateway_log_batcher.as_ref() {
            if let Err(entry) = gateway_log_batcher.try_enqueue_operation(entry) {
                fallback_entry = Some(entry);
            }
        } else {
            fallback_entry = Some(entry);
        }

        if let Some(entry) = fallback_entry {
            let pool_clone: sqlx::Pool<sqlx::Postgres> = pool.clone();
            let operation_name: String = operation.to_string();
            let client_stats: Option<Arc<ClientStatsBatcher>> = state.client_stats_batcher.clone();
            let logging_task_limiter = state.logging_task_limiter.clone();
            spawn_best_effort_logging_task(
                logging_task_limiter,
                "gateway_operation_log",
                async move {
                    if let Err(err) = insert_operation_log(pool_clone, client_stats, entry).await {
                        error!(error = %err, operation = operation_name, "failed to write gateway operation log");
                    }
                },
            );
        }
    }
}

pub fn finalize_request_log(
    state: Option<&AppState>,
    request_id: &str,
    completion: RequestCompletionLogContext,
) {
    if let Some(state) = state
        && let Some(logging_client) = state.logging_client_name.as_ref()
        && let Some(pool) = state.pg_registry.get_pool(logging_client)
    {
        let pool_clone: Pool<Postgres> = pool.clone();
        let request_id = request_id.to_string();
        let logging_task_limiter = state.logging_task_limiter.clone();
        spawn_best_effort_logging_task(
            logging_task_limiter,
            "gateway_request_finalize",
            async move {
                if let Err(err) =
                    update_request_log_completion(pool_clone, &request_id, &completion).await
                {
                    error!(
                        error = %err,
                        request_id = %request_id,
                        "failed to finalize gateway request log"
                    );
                }
            },
        );
    }
}

pub fn log_route_request_event(state: Option<&AppState>, context: RouteRequestLogContext) {
    if let Some(state) = state
        && let Some(logging_client) = state.logging_client_name.as_ref()
        && let Some(pool) = state.pg_registry.get_pool(logging_client)
    {
        let duration_ms: Option<i64> = context
            .duration_ms
            .map(|value| value.min(i64::MAX as u128) as i64);
        let entry = RouteRequestLogEntry {
            request_id: context
                .request_id
                .unwrap_or_else(|| Uuid::new_v4().to_string()),
            source: context.source.unwrap_or_else(|| "unknown".to_string()),
            outcome: context.outcome.unwrap_or_else(|| "unknown".to_string()),
            method: context.method,
            path: context.path,
            host: context.host,
            wildcard_host_pattern: context.wildcard_host_pattern,
            wildcard_prefix: context.wildcard_prefix,
            route_key: context.route_key,
            public_op: context.public_op,
            resolved_client: context.resolved_client,
            status_code: context.status_code.map(|value| value as i32),
            duration_ms,
            error_message: context.error_message,
            metadata: context.metadata.unwrap_or_else(|| json!({})),
            time: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_else(|_| std::time::Duration::new(0, 0))
                .as_secs() as i64,
        };

        let pool_clone: Pool<Postgres> = pool.clone();
        let source_for_log = entry.source.clone();
        let outcome_for_log = entry.outcome.clone();
        let logging_task_limiter = state.logging_task_limiter.clone();
        spawn_best_effort_logging_task(
            logging_task_limiter,
            "gateway_route_request_log",
            async move {
                if let Err(err) = insert_route_request_log(pool_clone, entry).await {
                    error!(
                        error = %err,
                        source = %source_for_log,
                        outcome = %outcome_for_log,
                        "failed to write gateway route request log"
                    );
                }
            },
        );
    }
}

struct RequestLogEntry {
    request_id: String,
    client: String,
    method: String,
    path: String,
    query_string: String,
    status_code: i32,
    ipv4: String,
    user_agent: String,
    headers: Value,
    body: Value,
    user_id: String,
    company_id: String,
    organization_id: String,
    api_key_id: Option<String>,
    internal_api_key_id: Option<i64>,
    presented_api_key_public_id: Option<String>,
    presented_api_key_hash: Option<String>,
    presented_api_key_salt: Option<String>,
    api_key_authenticated: bool,
    api_key_authorized: bool,
    api_key_enforced: bool,
    api_key_auth_reason: Option<String>,
    host: String,
    athena_url: String,
    cached: bool,
    duration_ms: Option<i64>,
    cache_lookup_outcome: Option<String>,
    cache_source: Option<String>,
    operation: Option<String>,
    table_name: Option<String>,
    time: i64,
}

struct RealtimeConnectionEntry {
    request_id: String,
    client_name: String,
    ipv4: String,
    athena_url: String,
    api_key_public_id: Option<String>,
    api_key_external_id: Option<String>,
    api_key_internal_id: Option<i64>,
    api_key_authenticated: bool,
    api_key_authorized: bool,
    method: String,
    path: String,
    user_agent: String,
    status_code: i32,
}

struct RouteRequestLogEntry {
    request_id: String,
    source: String,
    outcome: String,
    method: Option<String>,
    path: Option<String>,
    host: Option<String>,
    wildcard_host_pattern: Option<String>,
    wildcard_prefix: Option<String>,
    route_key: Option<String>,
    public_op: Option<String>,
    resolved_client: Option<String>,
    status_code: Option<i32>,
    duration_ms: Option<i64>,
    error_message: Option<String>,
    metadata: Value,
    time: i64,
}

async fn insert_request_log(
    pool: PgPool,
    client_stats: Option<Arc<ClientStatsBatcher>>,
    entry: RequestLogEntry,
) -> Result<(), sqlx::Error> {
    let client_name = entry.client.clone();
    let status_code = entry.status_code;
    let cached = entry.cached;
    let realtime_entry = RealtimeConnectionEntry {
        request_id: entry.request_id.clone(),
        client_name: entry.client.clone(),
        ipv4: entry.ipv4.clone(),
        athena_url: entry.athena_url.clone(),
        api_key_public_id: entry.presented_api_key_public_id.clone(),
        api_key_external_id: entry.api_key_id.clone(),
        api_key_internal_id: entry.internal_api_key_id,
        api_key_authenticated: entry.api_key_authenticated,
        api_key_authorized: entry.api_key_authorized,
        method: entry.method.clone(),
        path: entry.path.clone(),
        user_agent: entry.user_agent.clone(),
        status_code: entry.status_code,
    };

    let column_types = get_public_table_column_types(&pool, "gateway_request_log").await?;
    let mut query_builder =
        sqlx::QueryBuilder::<Postgres>::new("INSERT INTO gateway_request_log (");
    let mut first = true;
    let mut push_column = |query: &mut sqlx::QueryBuilder<'_, Postgres>, column: &str| {
        if !first {
            query.push(", ");
        }
        query.push(column);
        first = false;
    };

    push_column(&mut query_builder, "request_id");
    push_column(&mut query_builder, "client");
    push_column(&mut query_builder, "method");
    push_column(&mut query_builder, "path");
    push_column(&mut query_builder, "query_string");
    push_column(&mut query_builder, "status_code");
    push_column(&mut query_builder, "ipv4");
    push_column(&mut query_builder, "user_agent");
    push_column(&mut query_builder, "headers");
    push_column(&mut query_builder, "body");
    push_column(&mut query_builder, "user_id");
    push_column(&mut query_builder, "company_id");
    push_column(&mut query_builder, "organization_id");
    let api_key_id_type = column_types.get("api_key_id").cloned();
    let has_api_key_id = api_key_id_type.is_some();
    if has_api_key_id {
        push_column(&mut query_builder, "api_key_id");
    }
    let has_presented_public_id = column_types.contains_key("presented_api_key_public_id");
    if has_presented_public_id {
        push_column(&mut query_builder, "presented_api_key_public_id");
    }
    let has_presented_hash = column_types.contains_key("presented_api_key_hash");
    if has_presented_hash {
        push_column(&mut query_builder, "presented_api_key_hash");
    }
    let has_presented_salt = column_types.contains_key("presented_api_key_salt");
    if has_presented_salt {
        push_column(&mut query_builder, "presented_api_key_salt");
    }
    let has_authenticated = column_types.contains_key("api_key_authenticated");
    if has_authenticated {
        push_column(&mut query_builder, "api_key_authenticated");
    }
    let has_authorized = column_types.contains_key("api_key_authorized");
    if has_authorized {
        push_column(&mut query_builder, "api_key_authorized");
    }
    let has_enforced = column_types.contains_key("api_key_enforced");
    if has_enforced {
        push_column(&mut query_builder, "api_key_enforced");
    }
    let has_auth_reason = column_types.contains_key("api_key_auth_reason");
    if has_auth_reason {
        push_column(&mut query_builder, "api_key_auth_reason");
    }
    push_column(&mut query_builder, "host");
    push_column(&mut query_builder, "cached");
    let has_duration_ms = column_types.contains_key("duration_ms");
    if has_duration_ms {
        push_column(&mut query_builder, "duration_ms");
    }
    let has_cache_lookup_outcome = column_types.contains_key("cache_lookup_outcome");
    if has_cache_lookup_outcome {
        push_column(&mut query_builder, "cache_lookup_outcome");
    }
    let has_cache_source = column_types.contains_key("cache_source");
    if has_cache_source {
        push_column(&mut query_builder, "cache_source");
    }
    let has_operation = column_types.contains_key("operation");
    if has_operation {
        push_column(&mut query_builder, "operation");
    }
    let has_table_name = column_types.contains_key("table_name");
    if has_table_name {
        push_column(&mut query_builder, "table_name");
    }
    let has_time = column_types.contains_key("time");
    if has_time {
        push_column(&mut query_builder, "time");
    }

    query_builder.push(") VALUES (");
    let mut separated = query_builder.separated(", ");
    separated.push_bind(entry.request_id);
    separated.push_bind(entry.client);
    separated.push_bind(entry.method);
    separated.push_bind(entry.path);
    separated.push_bind(entry.query_string);
    separated.push_bind(entry.status_code);
    separated.push_bind(entry.ipv4);
    separated.push_bind(entry.user_agent);
    separated.push_bind(entry.headers);
    separated.push_bind(entry.body);
    separated.push_bind(entry.user_id);
    separated.push_bind(entry.company_id);
    separated.push_bind(entry.organization_id);
    if let Some(descriptor) = api_key_id_type.as_deref() {
        if postgres_column_descriptor_is_bigint(descriptor) {
            let internal_id = match entry.internal_api_key_id {
                Some(value) => Some(value),
                None => match entry.api_key_id.as_deref() {
                    Some(external_id) => resolve_internal_api_key_id(&pool, external_id).await?,
                    None => None,
                },
            };
            separated.push_bind(internal_id);
        } else {
            separated.push_bind(entry.api_key_id);
        }
    }
    if has_presented_public_id {
        separated.push_bind(entry.presented_api_key_public_id);
    }
    if has_presented_hash {
        separated.push_bind(entry.presented_api_key_hash);
    }
    if has_presented_salt {
        separated.push_bind(entry.presented_api_key_salt);
    }
    if has_authenticated {
        separated.push_bind(entry.api_key_authenticated);
    }
    if has_authorized {
        separated.push_bind(entry.api_key_authorized);
    }
    if has_enforced {
        separated.push_bind(entry.api_key_enforced);
    }
    if has_auth_reason {
        separated.push_bind(entry.api_key_auth_reason);
    }
    separated.push_bind(entry.host);
    separated.push_bind(entry.cached);
    if has_duration_ms {
        separated.push_bind(entry.duration_ms);
    }
    if has_cache_lookup_outcome {
        separated.push_bind(entry.cache_lookup_outcome);
    }
    if has_cache_source {
        separated.push_bind(entry.cache_source);
    }
    if has_operation {
        separated.push_bind(entry.operation);
    }
    if has_table_name {
        separated.push_bind(entry.table_name);
    }
    if has_time {
        match column_types.get("time").map(String::as_str) {
            Some(descriptor) if postgres_column_descriptor_is_timestamptz(descriptor) => {
                separated.push_bind(chrono::DateTime::<chrono::Utc>::from_timestamp(
                    entry.time, 0,
                ));
            }
            _ => {
                separated.push_bind(entry.time);
            }
        };
    }
    separated.push_unseparated(")");

    query_builder.build().execute(&pool).await?;

    upsert_realtime_connection_registry(&pool, realtime_entry).await?;
    maybe_purge_inactive_realtime_registry(&pool).await?;
    if let Some(ref batcher) = client_stats {
        batcher.try_enqueue_request_stats(&client_name, status_code, cached);
        batcher.try_enqueue_last_seen(&client_name);
    } else {
        upsert_client_request_statistics(&pool, &client_name, status_code, cached).await?;
        let _ = touch_athena_client_last_seen(&pool, &client_name).await;
    }
    Ok(())
}

async fn insert_route_request_log(
    pool: PgPool,
    entry: RouteRequestLogEntry,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO gateway_route_request_log (
            request_id,
            source,
            outcome,
            method,
            path,
            host,
            wildcard_host_pattern,
            wildcard_prefix,
            route_key,
            public_op,
            resolved_client,
            status_code,
            duration_ms,
            error_message,
            metadata,
            time
        )
        VALUES (
            $1, $2, $3, $4, $5, $6, $7, $8,
            $9, $10, $11, $12, $13, $14, $15, $16
        )
        "#,
    )
    .bind(entry.request_id)
    .bind(entry.source)
    .bind(entry.outcome)
    .bind(entry.method)
    .bind(entry.path)
    .bind(entry.host)
    .bind(entry.wildcard_host_pattern)
    .bind(entry.wildcard_prefix)
    .bind(entry.route_key)
    .bind(entry.public_op)
    .bind(entry.resolved_client)
    .bind(entry.status_code)
    .bind(entry.duration_ms)
    .bind(entry.error_message)
    .bind(entry.metadata)
    .bind(entry.time)
    .execute(&pool)
    .await?;

    Ok(())
}

async fn upsert_realtime_connection_registry(
    pool: &PgPool,
    entry: RealtimeConnectionEntry,
) -> Result<(), sqlx::Error> {
    let key_api = entry
        .api_key_public_id
        .as_deref()
        .or(entry.api_key_external_id.as_deref())
        .unwrap_or("ANON");

    let registry_key = format!(
        "{}|{}|{}|{}",
        entry.client_name, entry.ipv4, entry.athena_url, key_api
    );

    sqlx::query(
        r#"
        INSERT INTO gateway_connection_registry_realtime (
            registry_key,
            client_name,
            ipv4,
            athena_url,
            api_key_public_id,
            api_key_external_id,
            api_key_internal_id,
            api_key_authenticated,
            api_key_authorized,
            last_method,
            last_path,
            user_agent,
            last_request_id,
            last_status_code,
            total_hits,
            first_seen_at,
            last_seen_at,
            updated_at
        )
        VALUES (
            $1,
            $2,
            $3,
            $4,
            $5,
            $6,
            $7,
            $8,
            $9,
            $10,
            $11,
            $12,
            $13,
            $14,
            1,
            now(),
            now(),
            now()
        )
        ON CONFLICT (registry_key) DO UPDATE
        SET api_key_public_id = EXCLUDED.api_key_public_id,
            api_key_external_id = EXCLUDED.api_key_external_id,
            api_key_internal_id = EXCLUDED.api_key_internal_id,
            api_key_authenticated = EXCLUDED.api_key_authenticated,
            api_key_authorized = EXCLUDED.api_key_authorized,
            last_method = EXCLUDED.last_method,
            last_path = EXCLUDED.last_path,
            user_agent = EXCLUDED.user_agent,
            last_request_id = EXCLUDED.last_request_id,
            last_status_code = EXCLUDED.last_status_code,
            total_hits = gateway_connection_registry_realtime.total_hits + 1,
            last_seen_at = now(),
            updated_at = now()
        "#,
    )
    .bind(registry_key)
    .bind(entry.client_name)
    .bind(entry.ipv4)
    .bind(entry.athena_url)
    .bind(entry.api_key_public_id)
    .bind(entry.api_key_external_id)
    .bind(entry.api_key_internal_id)
    .bind(entry.api_key_authenticated)
    .bind(entry.api_key_authorized)
    .bind(entry.method)
    .bind(entry.path)
    .bind(entry.user_agent)
    .bind(entry.request_id)
    .bind(entry.status_code)
    .execute(pool)
    .await?;

    Ok(())
}

async fn maybe_purge_inactive_realtime_registry(pool: &PgPool) -> Result<(), sqlx::Error> {
    let now_ts = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_else(|_| std::time::Duration::new(0, 0))
        .as_secs() as i64;

    let runtime_env = runtime_env_settings();
    let run_every_secs = runtime_env.realtime_registry_purge_interval_secs as i64;
    let inactive_after_secs = runtime_env.realtime_registry_inactive_after_secs;
    let previous = LAST_REALTIME_REGISTRY_PURGE_TS.load(Ordering::Relaxed);
    if now_ts - previous < run_every_secs {
        return Ok(());
    }

    if LAST_REALTIME_REGISTRY_PURGE_TS
        .compare_exchange(previous, now_ts, Ordering::SeqCst, Ordering::Relaxed)
        .is_err()
    {
        return Ok(());
    }

    sqlx::query(
        r#"
        DELETE FROM gateway_connection_registry_realtime
        WHERE last_seen_at < now() - ($1::bigint * interval '1 second')
        "#,
    )
    .bind(inactive_after_secs as i64)
    .execute(pool)
    .await?;

    Ok(())
}

struct OperationLogEntry {
    request_id: String,
    operation: String,
    table_name: Option<String>,
    client: String,
    method: String,
    path: String,
    status_code: i32,
    duration_ms: i64,
    details: Value,
    time: i64,
    error: bool,
    message: Option<String>,
    cache_key: Option<String>,
    cache_lookup_outcome: Option<String>,
    cache_source: Option<String>,
    cached: Option<bool>,
}

async fn insert_operation_log(
    pool: PgPool,
    client_stats: Option<Arc<ClientStatsBatcher>>,
    entry: OperationLogEntry,
) -> Result<(), sqlx::Error> {
    let client_name = entry.client.clone();
    let table_name = entry.table_name.clone();
    let operation = entry.operation.clone();
    let is_error = entry.error || entry.status_code >= 400;

    let column_types = get_public_table_column_types(&pool, "gateway_operation_log").await?;
    let mut query_builder =
        sqlx::QueryBuilder::<Postgres>::new("INSERT INTO gateway_operation_log (");
    let mut first = true;
    let mut push_column = |query: &mut sqlx::QueryBuilder<'_, Postgres>, column: &str| {
        if !first {
            query.push(", ");
        }
        query.push(column);
        first = false;
    };

    push_column(&mut query_builder, "request_id");
    push_column(&mut query_builder, "operation");
    push_column(&mut query_builder, "table_name");
    push_column(&mut query_builder, "client");
    push_column(&mut query_builder, "method");
    push_column(&mut query_builder, "path");
    push_column(&mut query_builder, "status_code");
    push_column(&mut query_builder, "duration_ms");
    push_column(&mut query_builder, "details");
    push_column(&mut query_builder, "time");
    let has_error = column_types.contains_key("error");
    if has_error {
        push_column(&mut query_builder, "error");
    }
    let has_message = column_types.contains_key("message");
    if has_message {
        push_column(&mut query_builder, "message");
    }
    let has_cache_key = column_types.contains_key("cache_key");
    if has_cache_key {
        push_column(&mut query_builder, "cache_key");
    }
    let has_cache_lookup_outcome = column_types.contains_key("cache_lookup_outcome");
    if has_cache_lookup_outcome {
        push_column(&mut query_builder, "cache_lookup_outcome");
    }
    let has_cache_source = column_types.contains_key("cache_source");
    if has_cache_source {
        push_column(&mut query_builder, "cache_source");
    }
    let has_cached = column_types.contains_key("cached");
    if has_cached {
        push_column(&mut query_builder, "cached");
    }

    query_builder.push(") VALUES (");
    let mut separated = query_builder.separated(", ");
    separated.push_bind(entry.request_id);
    separated.push_bind(entry.operation);
    separated.push_bind(entry.table_name);
    separated.push_bind(entry.client);
    separated.push_bind(entry.method);
    separated.push_bind(entry.path);
    separated.push_bind(entry.status_code);
    separated.push_bind(entry.duration_ms);
    separated.push_bind(entry.details);
    separated.push_bind(entry.time);
    if has_error {
        separated.push_bind(entry.error);
    }
    if has_message {
        separated.push_bind(entry.message);
    }
    if has_cache_key {
        separated.push_bind(entry.cache_key);
    }
    if has_cache_lookup_outcome {
        separated.push_bind(entry.cache_lookup_outcome);
    }
    if has_cache_source {
        separated.push_bind(entry.cache_source);
    }
    if has_cached {
        separated.push_bind(entry.cached);
    }
    separated.push_unseparated(")");

    query_builder.build().execute(&pool).await?;

    if let Some(ref batcher) = client_stats {
        batcher.try_enqueue_operation_stats(&client_name);
        if let Some(table_name) = table_name.as_deref() {
            batcher.try_enqueue_table_stats(&client_name, table_name, &operation, is_error);
        }
        batcher.try_enqueue_last_seen(&client_name);
    } else {
        upsert_client_operation_statistics(&pool, &client_name).await?;
        if let Some(table_name) = table_name.as_deref() {
            upsert_client_table_statistics(&pool, &client_name, table_name, &operation, is_error)
                .await?;
        }
        let _ = touch_athena_client_last_seen(&pool, &client_name).await;
    }
    Ok(())
}

async fn flush_request_log_batch(
    pool: &PgPool,
    client_stats: Option<&Arc<ClientStatsBatcher>>,
    request_rows: Vec<RequestLogEntry>,
) {
    if request_rows.is_empty() {
        return;
    }

    if let Err(err) = insert_request_logs_bulk(pool, &request_rows).await {
        warn!(
            target: "athena_rs::gateway_log_batcher",
            error = %err,
            rows = request_rows.len(),
            "gateway request log bulk insert failed; falling back to per-row writes"
        );
        for entry in request_rows {
            if let Err(entry_err) =
                insert_request_log(pool.clone(), client_stats.cloned(), entry).await
            {
                error!(error = %entry_err, "failed to write gateway request log during fallback");
            }
        }
        return;
    }

    for entry in request_rows {
        let realtime_entry = RealtimeConnectionEntry {
            request_id: entry.request_id.clone(),
            client_name: entry.client.clone(),
            ipv4: entry.ipv4.clone(),
            athena_url: entry.athena_url.clone(),
            api_key_public_id: entry.presented_api_key_public_id.clone(),
            api_key_external_id: entry.api_key_id.clone(),
            api_key_internal_id: entry.internal_api_key_id,
            api_key_authenticated: entry.api_key_authenticated,
            api_key_authorized: entry.api_key_authorized,
            method: entry.method.clone(),
            path: entry.path.clone(),
            user_agent: entry.user_agent.clone(),
            status_code: entry.status_code,
        };

        if let Err(err) = upsert_realtime_connection_registry(pool, realtime_entry).await {
            error!(
                error = %err,
                request_id = %entry.request_id,
                "failed to upsert realtime connection registry"
            );
        }

        if let Some(batcher) = client_stats {
            batcher.try_enqueue_request_stats(&entry.client, entry.status_code, entry.cached);
            batcher.try_enqueue_last_seen(&entry.client);
        } else {
            if let Err(err) = upsert_client_request_statistics(
                pool,
                &entry.client,
                entry.status_code,
                entry.cached,
            )
            .await
            {
                error!(
                    error = %err,
                    client = %entry.client,
                    "failed to upsert client request statistics"
                );
            }
            if let Err(err) = touch_athena_client_last_seen(pool, &entry.client).await {
                error!(
                    error = %err,
                    client = %entry.client,
                    "failed to touch athena client last_seen"
                );
            }
        }
    }

    if let Err(err) = maybe_purge_inactive_realtime_registry(pool).await {
        error!(error = %err, "failed to purge inactive realtime registry entries");
    }
}

async fn flush_operation_log_batch(
    pool: &PgPool,
    client_stats: Option<&Arc<ClientStatsBatcher>>,
    operation_rows: Vec<OperationLogEntry>,
) {
    if operation_rows.is_empty() {
        return;
    }

    if let Err(err) = insert_operation_logs_bulk(pool, &operation_rows).await {
        warn!(
            target: "athena_rs::gateway_log_batcher",
            error = %err,
            rows = operation_rows.len(),
            "gateway operation log bulk insert failed; falling back to per-row writes"
        );
        for entry in operation_rows {
            if let Err(entry_err) =
                insert_operation_log(pool.clone(), client_stats.cloned(), entry).await
            {
                error!(
                    error = %entry_err,
                    "failed to write gateway operation log during fallback"
                );
            }
        }
        return;
    }

    for entry in operation_rows {
        let is_error = entry.error || entry.status_code >= 400;

        if let Some(batcher) = client_stats {
            batcher.try_enqueue_operation_stats(&entry.client);
            if let Some(table_name) = entry.table_name.as_deref() {
                batcher.try_enqueue_table_stats(
                    &entry.client,
                    table_name,
                    &entry.operation,
                    is_error,
                );
            }
            batcher.try_enqueue_last_seen(&entry.client);
        } else {
            if let Err(err) = upsert_client_operation_statistics(pool, &entry.client).await {
                error!(
                    error = %err,
                    client = %entry.client,
                    "failed to upsert client operation statistics"
                );
            }
            if let Some(table_name) = entry.table_name.as_deref()
                && let Err(err) = upsert_client_table_statistics(
                    pool,
                    &entry.client,
                    table_name,
                    &entry.operation,
                    is_error,
                )
                .await
            {
                error!(
                    error = %err,
                    client = %entry.client,
                    table = %table_name,
                    operation = %entry.operation,
                    "failed to upsert client table statistics"
                );
            }
            if let Err(err) = touch_athena_client_last_seen(pool, &entry.client).await {
                error!(
                    error = %err,
                    client = %entry.client,
                    "failed to touch athena client last_seen"
                );
            }
        }
    }
}

async fn insert_request_logs_bulk(
    pool: &PgPool,
    entries: &[RequestLogEntry],
) -> Result<(), sqlx::Error> {
    if entries.is_empty() {
        return Ok(());
    }

    let column_types = get_public_table_column_types(pool, "gateway_request_log").await?;
    let mut query_builder =
        sqlx::QueryBuilder::<Postgres>::new("INSERT INTO gateway_request_log (");
    let mut first = true;
    let mut push_column = |query: &mut sqlx::QueryBuilder<'_, Postgres>, column: &str| {
        if !first {
            query.push(", ");
        }
        query.push(column);
        first = false;
    };

    push_column(&mut query_builder, "request_id");
    push_column(&mut query_builder, "client");
    push_column(&mut query_builder, "method");
    push_column(&mut query_builder, "path");
    push_column(&mut query_builder, "query_string");
    push_column(&mut query_builder, "status_code");
    push_column(&mut query_builder, "ipv4");
    push_column(&mut query_builder, "user_agent");
    push_column(&mut query_builder, "headers");
    push_column(&mut query_builder, "body");
    push_column(&mut query_builder, "user_id");
    push_column(&mut query_builder, "company_id");
    push_column(&mut query_builder, "organization_id");
    let api_key_id_type = column_types.get("api_key_id").cloned();
    let has_api_key_id = api_key_id_type.is_some();
    if has_api_key_id {
        push_column(&mut query_builder, "api_key_id");
    }
    let has_presented_public_id = column_types.contains_key("presented_api_key_public_id");
    if has_presented_public_id {
        push_column(&mut query_builder, "presented_api_key_public_id");
    }
    let has_presented_hash = column_types.contains_key("presented_api_key_hash");
    if has_presented_hash {
        push_column(&mut query_builder, "presented_api_key_hash");
    }
    let has_presented_salt = column_types.contains_key("presented_api_key_salt");
    if has_presented_salt {
        push_column(&mut query_builder, "presented_api_key_salt");
    }
    let has_authenticated = column_types.contains_key("api_key_authenticated");
    if has_authenticated {
        push_column(&mut query_builder, "api_key_authenticated");
    }
    let has_authorized = column_types.contains_key("api_key_authorized");
    if has_authorized {
        push_column(&mut query_builder, "api_key_authorized");
    }
    let has_enforced = column_types.contains_key("api_key_enforced");
    if has_enforced {
        push_column(&mut query_builder, "api_key_enforced");
    }
    let has_auth_reason = column_types.contains_key("api_key_auth_reason");
    if has_auth_reason {
        push_column(&mut query_builder, "api_key_auth_reason");
    }
    push_column(&mut query_builder, "host");
    push_column(&mut query_builder, "cached");
    let has_duration_ms = column_types.contains_key("duration_ms");
    if has_duration_ms {
        push_column(&mut query_builder, "duration_ms");
    }
    let has_cache_lookup_outcome = column_types.contains_key("cache_lookup_outcome");
    if has_cache_lookup_outcome {
        push_column(&mut query_builder, "cache_lookup_outcome");
    }
    let has_cache_source = column_types.contains_key("cache_source");
    if has_cache_source {
        push_column(&mut query_builder, "cache_source");
    }
    let has_operation = column_types.contains_key("operation");
    if has_operation {
        push_column(&mut query_builder, "operation");
    }
    let has_table_name = column_types.contains_key("table_name");
    if has_table_name {
        push_column(&mut query_builder, "table_name");
    }
    let has_time = column_types.contains_key("time");
    if has_time {
        push_column(&mut query_builder, "time");
    }

    let mut resolved_internal_api_key_ids: Vec<Option<i64>> = Vec::new();
    if let Some(descriptor) = api_key_id_type.as_deref()
        && postgres_column_descriptor_is_bigint(descriptor)
    {
        resolved_internal_api_key_ids = Vec::with_capacity(entries.len());
        for entry in entries {
            let internal_id = match entry.internal_api_key_id {
                Some(value) => Some(value),
                None => match entry.api_key_id.as_deref() {
                    Some(external_id) => resolve_internal_api_key_id(pool, external_id).await?,
                    None => None,
                },
            };
            resolved_internal_api_key_ids.push(internal_id);
        }
    }

    query_builder.push(") ");
    query_builder.push_values(entries.iter().enumerate(), |mut row, (idx, entry)| {
        row.push_bind(&entry.request_id);
        row.push_bind(&entry.client);
        row.push_bind(&entry.method);
        row.push_bind(&entry.path);
        row.push_bind(&entry.query_string);
        row.push_bind(entry.status_code);
        row.push_bind(&entry.ipv4);
        row.push_bind(&entry.user_agent);
        row.push_bind(entry.headers.clone());
        row.push_bind(entry.body.clone());
        row.push_bind(&entry.user_id);
        row.push_bind(&entry.company_id);
        row.push_bind(&entry.organization_id);
        if let Some(descriptor) = api_key_id_type.as_deref() {
            if postgres_column_descriptor_is_bigint(descriptor) {
                row.push_bind(resolved_internal_api_key_ids[idx]);
            } else {
                row.push_bind(entry.api_key_id.clone());
            }
        }
        if has_presented_public_id {
            row.push_bind(entry.presented_api_key_public_id.clone());
        }
        if has_presented_hash {
            row.push_bind(entry.presented_api_key_hash.clone());
        }
        if has_presented_salt {
            row.push_bind(entry.presented_api_key_salt.clone());
        }
        if has_authenticated {
            row.push_bind(entry.api_key_authenticated);
        }
        if has_authorized {
            row.push_bind(entry.api_key_authorized);
        }
        if has_enforced {
            row.push_bind(entry.api_key_enforced);
        }
        if has_auth_reason {
            row.push_bind(entry.api_key_auth_reason.clone());
        }
        row.push_bind(&entry.host);
        row.push_bind(entry.cached);
        if has_duration_ms {
            row.push_bind(entry.duration_ms);
        }
        if has_cache_lookup_outcome {
            row.push_bind(entry.cache_lookup_outcome.clone());
        }
        if has_cache_source {
            row.push_bind(entry.cache_source.clone());
        }
        if has_operation {
            row.push_bind(entry.operation.clone());
        }
        if has_table_name {
            row.push_bind(entry.table_name.clone());
        }
        if has_time {
            match column_types.get("time").map(String::as_str) {
                Some(descriptor) if postgres_column_descriptor_is_timestamptz(descriptor) => {
                    row.push_bind(chrono::DateTime::<chrono::Utc>::from_timestamp(
                        entry.time, 0,
                    ));
                }
                _ => {
                    row.push_bind(entry.time);
                }
            }
        }
    });

    query_builder.build().execute(pool).await?;
    Ok(())
}

async fn insert_operation_logs_bulk(
    pool: &PgPool,
    entries: &[OperationLogEntry],
) -> Result<(), sqlx::Error> {
    if entries.is_empty() {
        return Ok(());
    }

    let column_types = get_public_table_column_types(pool, "gateway_operation_log").await?;
    let mut query_builder =
        sqlx::QueryBuilder::<Postgres>::new("INSERT INTO gateway_operation_log (");
    let mut first = true;
    let mut push_column = |query: &mut sqlx::QueryBuilder<'_, Postgres>, column: &str| {
        if !first {
            query.push(", ");
        }
        query.push(column);
        first = false;
    };

    push_column(&mut query_builder, "request_id");
    push_column(&mut query_builder, "operation");
    push_column(&mut query_builder, "table_name");
    push_column(&mut query_builder, "client");
    push_column(&mut query_builder, "method");
    push_column(&mut query_builder, "path");
    push_column(&mut query_builder, "status_code");
    push_column(&mut query_builder, "duration_ms");
    push_column(&mut query_builder, "details");
    push_column(&mut query_builder, "time");
    let has_error = column_types.contains_key("error");
    if has_error {
        push_column(&mut query_builder, "error");
    }
    let has_message = column_types.contains_key("message");
    if has_message {
        push_column(&mut query_builder, "message");
    }
    let has_cache_key = column_types.contains_key("cache_key");
    if has_cache_key {
        push_column(&mut query_builder, "cache_key");
    }
    let has_cache_lookup_outcome = column_types.contains_key("cache_lookup_outcome");
    if has_cache_lookup_outcome {
        push_column(&mut query_builder, "cache_lookup_outcome");
    }
    let has_cache_source = column_types.contains_key("cache_source");
    if has_cache_source {
        push_column(&mut query_builder, "cache_source");
    }
    let has_cached = column_types.contains_key("cached");
    if has_cached {
        push_column(&mut query_builder, "cached");
    }

    query_builder.push(") ");
    query_builder.push_values(entries.iter(), |mut row, entry| {
        row.push_bind(&entry.request_id);
        row.push_bind(&entry.operation);
        row.push_bind(entry.table_name.clone());
        row.push_bind(&entry.client);
        row.push_bind(&entry.method);
        row.push_bind(&entry.path);
        row.push_bind(entry.status_code);
        row.push_bind(entry.duration_ms);
        row.push_bind(entry.details.clone());
        row.push_bind(entry.time);
        if has_error {
            row.push_bind(entry.error);
        }
        if has_message {
            row.push_bind(entry.message.clone());
        }
        if has_cache_key {
            row.push_bind(entry.cache_key.clone());
        }
        if has_cache_lookup_outcome {
            row.push_bind(entry.cache_lookup_outcome.clone());
        }
        if has_cache_source {
            row.push_bind(entry.cache_source.clone());
        }
        if has_cached {
            row.push_bind(entry.cached);
        }
    });

    query_builder.build().execute(pool).await?;
    Ok(())
}

async fn update_request_log_completion(
    pool: PgPool,
    request_id: &str,
    completion: &RequestCompletionLogContext,
) -> Result<(), sqlx::Error> {
    let duration_ms: Option<i64> = completion
        .duration_ms
        .map(|value| value.min(i64::MAX as u128) as i64);
    let status_code = completion.status_code.map(|value| value as i32);

    let column_types = get_public_table_column_types(&pool, "gateway_request_log").await?;
    let mut query_builder = sqlx::QueryBuilder::<Postgres>::new("UPDATE gateway_request_log SET ");
    let mut first = true;

    if column_types.contains_key("status_code") {
        query_builder.push("status_code = COALESCE(");
        query_builder.push_bind(status_code);
        query_builder.push(", status_code)");
        first = false;
    }
    if column_types.contains_key("duration_ms") {
        if !first {
            query_builder.push(", ");
        }
        query_builder.push("duration_ms = COALESCE(");
        query_builder.push_bind(duration_ms);
        query_builder.push(", duration_ms)");
        first = false;
    }
    if column_types.contains_key("cached") {
        if !first {
            query_builder.push(", ");
        }
        query_builder.push("cached = COALESCE(");
        query_builder.push_bind(completion.cached);
        query_builder.push(", cached)");
        first = false;
    }
    if column_types.contains_key("cache_lookup_outcome") {
        if !first {
            query_builder.push(", ");
        }
        query_builder.push("cache_lookup_outcome = COALESCE(");
        query_builder.push_bind(completion.cache_lookup_outcome.clone());
        query_builder.push(", cache_lookup_outcome)");
        first = false;
    }
    if column_types.contains_key("cache_source") {
        if !first {
            query_builder.push(", ");
        }
        query_builder.push("cache_source = COALESCE(");
        query_builder.push_bind(completion.cache_source.clone());
        query_builder.push(", cache_source)");
        first = false;
    }
    if column_types.contains_key("operation") {
        if !first {
            query_builder.push(", ");
        }
        query_builder.push("operation = COALESCE(");
        query_builder.push_bind(completion.operation.clone());
        query_builder.push(", operation)");
        first = false;
    }
    if column_types.contains_key("table_name") {
        if !first {
            query_builder.push(", ");
        }
        query_builder.push("table_name = COALESCE(");
        query_builder.push_bind(completion.table_name.clone());
        query_builder.push(", table_name)");
        first = false;
    }

    if first {
        return Ok(());
    }

    query_builder.push(" WHERE request_id = ");
    query_builder.push_bind(request_id);

    query_builder.build().execute(&pool).await?;

    Ok(())
}

async fn upsert_client_request_statistics(
    pool: &PgPool,
    client_name: &str,
    status_code: i32,
    cached: bool,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO client_statistics (
            client_name,
            total_requests,
            successful_requests,
            failed_requests,
            total_cached_requests,
            total_operations,
            last_request_at
        )
        VALUES (
            $1,
            1,
            CASE WHEN $2 >= 200 AND $2 < 400 THEN 1 ELSE 0 END,
            CASE WHEN $2 >= 400 THEN 1 ELSE 0 END,
            CASE WHEN $3 THEN 1 ELSE 0 END,
            0,
            now()
        )
        ON CONFLICT (client_name) DO UPDATE
        SET total_requests = client_statistics.total_requests + 1,
            successful_requests = client_statistics.successful_requests
                + CASE WHEN EXCLUDED.successful_requests > 0 THEN 1 ELSE 0 END,
            failed_requests = client_statistics.failed_requests
                + CASE WHEN EXCLUDED.failed_requests > 0 THEN 1 ELSE 0 END,
            total_cached_requests = client_statistics.total_cached_requests
                + CASE WHEN EXCLUDED.total_cached_requests > 0 THEN 1 ELSE 0 END,
            last_request_at = now(),
            updated_at = now()
        "#,
    )
    .bind(client_name)
    .bind(status_code)
    .bind(cached)
    .execute(pool)
    .await?;

    Ok(())
}

async fn upsert_client_operation_statistics(
    pool: &PgPool,
    client_name: &str,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO client_statistics (
            client_name,
            total_requests,
            successful_requests,
            failed_requests,
            total_cached_requests,
            total_operations,
            last_operation_at
        )
        VALUES ($1, 0, 0, 0, 0, 1, now())
        ON CONFLICT (client_name) DO UPDATE
        SET total_operations = client_statistics.total_operations + 1,
            last_operation_at = now(),
            updated_at = now()
        "#,
    )
    .bind(client_name)
    .execute(pool)
    .await?;

    Ok(())
}

async fn upsert_client_table_statistics(
    pool: &PgPool,
    client_name: &str,
    table_name: &str,
    operation: &str,
    is_error: bool,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO client_table_statistics (
            client_name,
            table_name,
            operation,
            total_operations,
            error_operations,
            last_operation_at
        )
        VALUES ($1, $2, $3, 1, CASE WHEN $4 THEN 1 ELSE 0 END, now())
        ON CONFLICT (client_name, table_name, operation) DO UPDATE
        SET total_operations = client_table_statistics.total_operations + 1,
            error_operations = client_table_statistics.error_operations
                + CASE WHEN EXCLUDED.error_operations > 0 THEN 1 ELSE 0 END,
            last_operation_at = now(),
            updated_at = now()
        "#,
    )
    .bind(client_name)
    .bind(table_name)
    .bind(operation)
    .bind(is_error)
    .execute(pool)
    .await?;

    Ok(())
}
