//! HTTP and WebSocket upgrade handlers for the CDC server.

use super::connection::handle_socket;
use super::events::EventMessage;
use super::state::AppState;
use axum::{
    Json,
    extract::{State, ws::WebSocketUpgrade},
    http::{HeaderMap, StatusCode},
    response::IntoResponse,
};
use serde_json::json;
use std::collections::HashSet;
use std::env;
use tokio::sync::MutexGuard;
use tracing::warn;

const DEFAULT_WILDCARD_HOST_PATTERN: &str = "*.v3.athena-cluster.com";
const ENV_WILDCARD_HOST_PATTERN: &str = "ATHENA_WILDCARD_HOST_PATTERN";
const ENV_WILDCARD_HOST_ROUTING_ENABLED: &str = "ATHENA_WILDCARD_HOST_ROUTING_ENABLED";

fn wildcard_host_pattern() -> String {
    env::var(ENV_WILDCARD_HOST_PATTERN)
        .map(|value| value.trim().to_string())
        .ok()
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| DEFAULT_WILDCARD_HOST_PATTERN.to_string())
}

fn wildcard_host_routing_enabled() -> bool {
    env::var(ENV_WILDCARD_HOST_ROUTING_ENABLED)
        .ok()
        .map(|value| value.trim().to_ascii_lowercase())
        .map(|value| !matches!(value.as_str(), "0" | "false" | "no" | "off"))
        .unwrap_or(true)
}

fn header_value_trimmed(headers: &HeaderMap, name: &str) -> Option<String> {
    headers
        .get(name)
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
}

fn first_forwarded_host(headers: &HeaderMap) -> Option<String> {
    header_value_trimmed(headers, "x-forwarded-host")
        .and_then(|value| value.split(',').next().map(str::trim).map(str::to_string))
        .filter(|value| !value.is_empty())
}

fn forwarded_header_host(headers: &HeaderMap) -> Option<String> {
    let forwarded = header_value_trimmed(headers, "forwarded")?;
    forwarded
        .split(',')
        .flat_map(|entry| entry.split(';'))
        .filter_map(|part| part.trim().split_once('='))
        .find_map(|(key, value)| {
            if !key.trim().eq_ignore_ascii_case("host") {
                return None;
            }
            let host = value.trim().trim_matches('"').trim();
            if host.is_empty() {
                None
            } else {
                Some(host.to_string())
            }
        })
}

fn direct_database_header_present(headers: &HeaderMap) -> bool {
    ["x-jdbc-url", "x-athena-jdbc-url", "x-pg-uri"]
        .into_iter()
        .any(|name| header_value_trimmed(headers, name).is_some())
}

fn extract_prefix_from_host(host_header: &str, wildcard_pattern: &str) -> Option<String> {
    let host = normalize_host(host_header)?;
    let suffix = wildcard_pattern
        .trim()
        .trim_end_matches('.')
        .to_ascii_lowercase()
        .strip_prefix("*.")?
        .trim()
        .to_string();
    if suffix.is_empty() || suffix.contains('*') {
        return None;
    }

    let dotted_suffix = format!(".{suffix}");
    if !host.ends_with(&dotted_suffix) {
        return None;
    }

    let prefix = &host[..host.len().saturating_sub(dotted_suffix.len())];
    if prefix.is_empty()
        || prefix.contains('.')
        || !prefix
            .chars()
            .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '-' || ch == '_')
    {
        return None;
    }

    Some(prefix.to_string())
}

fn normalize_host(host_header: &str) -> Option<String> {
    let raw = host_header
        .trim()
        .trim_end_matches('.')
        .to_ascii_lowercase();
    if raw.is_empty() || raw.starts_with('[') {
        return None;
    }

    let without_port = raw.split(':').next().unwrap_or_default().trim();
    if without_port.is_empty() {
        None
    } else {
        Some(without_port.to_string())
    }
}

fn infer_client_from_wildcard_host(headers: &HeaderMap) -> Option<String> {
    if !wildcard_host_routing_enabled()
        || header_value_trimmed(headers, "x-athena-client").is_some()
        || direct_database_header_present(headers)
    {
        return None;
    }

    let pattern = wildcard_host_pattern();
    [
        header_value_trimmed(headers, "host"),
        first_forwarded_host(headers),
        forwarded_header_host(headers),
    ]
    .into_iter()
    .flatten()
    .find_map(|host| extract_prefix_from_host(&host, &pattern))
}

fn effective_client_id(headers: &HeaderMap) -> String {
    header_value_trimmed(headers, "x-athena-client")
        .or_else(|| infer_client_from_wildcard_host(headers))
        .unwrap_or_default()
}

/// WebSocket upgrade handler.
///
/// Requires `X-Athena-Client` or an equivalent tenant wildcard host.
/// Returns 400 with a JSON body if neither client selector is present.
pub async fn ws_handler(
    ws: WebSocketUpgrade,
    headers: HeaderMap,
    State(state): State<AppState>,
) -> impl IntoResponse {
    let client_id = effective_client_id(&headers);

    if client_id.is_empty() {
        warn!(
            target: "athena::cdc::websocket",
            wildcard_host_pattern = %wildcard_host_pattern(),
            "websocket upgrade rejected: missing X-Athena-Client header or tenant wildcard host"
        );
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({
                "error": "Missing client selector",
                "message": "Pass X-Athena-Client or connect through the tenant wildcard hostname to subscribe to CDC events",
                "hint": format!(
                    "Add request header X-Athena-Client: <your_client_name>, or use host <your_client_name>.{}",
                    wildcard_host_pattern()
                        .trim()
                        .trim_start_matches("*.")
                )
            })),
        )
            .into_response();
    }

    ws.on_upgrade(move |socket| handle_socket(socket, state, client_id))
}

/// POST /events — publish an event.
///
/// Returns 200 with `status: "voided"` if no WebSocket subscriber for `organization_id`;
/// otherwise broadcasts and returns 200 with `status: "delivered"`.
pub async fn publish_event(
    State(state): State<AppState>,
    Json(event): Json<EventMessage>,
) -> impl IntoResponse {
    let organization_id: String = event.organization_id.clone();

    let has_subscribers: bool = {
        let subscribers: MutexGuard<'_, HashSet<String>> = state.active_subscribers.lock().await;
        subscribers.contains(&organization_id)
    };

    if !has_subscribers {
        return (
            StatusCode::OK,
            Json(json!({
                "message": "there was no subscriber to this organization_id channel so the message has been voided",
                "organization_id": organization_id,
                "status": "voided",
                "success": true
            })),
        )
            .into_response();
    }

    let event_json: String = match serde_json::to_string(&event) {
        Ok(s) => s,
        Err(e) => {
            warn!("Failed to serialize event for broadcast: {}", e);
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                "Failed to serialize event",
            )
                .into_response();
        }
    };
    if let Err(err) = state.tx.send(event_json) {
        warn!("Failed to broadcast event: {}", err);
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            "Failed to broadcast event",
        )
            .into_response();
    }

    (
        StatusCode::OK,
        Json(json!({
            "status": "delivered",
            "success": true,
            "organization_id": organization_id
        })),
    )
        .into_response()
}

/// GET /status — server health; returns JSON with `status: "ok"`.
pub async fn status() -> impl IntoResponse {
    (
        StatusCode::OK,
        Json(json!({
            "status": "ok",
            "success": true,
            "server": "dms-server-api"
        })),
    )
        .into_response()
}
