//! WebSocket server for CDC Athena.
//!
//! - **URL:** Use `ws://host:port/ws` when the server is plain HTTP; use `wss://` only when
//!   the server is behind TLS (otherwise you get SSL WRONG_VERSION_NUMBER).
//! - **Header:** Send `X-Athena-Client: <client_name>` on the upgrade request; without it the
//!   server returns 400. Events are filtered by matching this value to `EventMessage::organization_id`.
//!
//! ## Messages
//!
//! **Client → Server (you send):**
//! - **Travel (replay):** `{"type":"travel","organization_id":"<id>","since_seq":123,"since_ts_ms":0,"limit":1000}`  
//!   All fields except `type` are optional. Omit `organization_id` to replay all subscribed channels.
//!
//! **Server → Client (you receive):**
//! - On connect: `{"status":"subscribed","client_id":"...","message":"..."}`
//! - CDC events: `{"organization_id":"...","data":{"seq":1,"ts_ms":...,"payload":{...}}}`
//! - After travel: same event shape for each replayed record, then `{"status":"travel_complete","organization_id":...,"since_seq":...,"since_ts_ms":...,"limit":...}`

mod connection;
pub mod events;
mod routes;
mod state;

use std::collections::HashSet;
use std::error::Error as StdError;
use std::io::{Error, ErrorKind};
use std::net::SocketAddr;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::Arc;

use axum::Router;
use axum::routing::{get, post};
use tokio::net::TcpListener;
use tokio::sync::{Mutex, broadcast};
use tracing::info;

use events::set_broadcast_tx;
use state::AppState;

/// Default capacity of the broadcast channel (number of in-flight messages).
const BROADCAST_CAPACITY: usize = 1000;
/// Config key that opt-ins to legacy Axum wildcard syntax conversion.
const CDC_ALLOW_LEGACY_AXUM_WILDCARDS_CONFIG_KEY: &str = "cdc.allow_legacy_axum_wildcards";

fn panic_payload_to_string(payload: Box<dyn std::any::Any + Send>) -> String {
    if let Some(message) = payload.downcast_ref::<&str>() {
        return (*message).to_string();
    }
    if let Some(message) = payload.downcast_ref::<String>() {
        return message.clone();
    }
    "unknown panic payload".to_string()
}

fn is_axum_legacy_wildcard_panic(message: &str) -> bool {
    message.contains("Path segments must not start with `*`.")
}

fn normalize_legacy_wildcard_path(
    path: &str,
    allow_legacy_wildcards: bool,
) -> Result<String, Error> {
    let has_legacy_wildcard: bool = path
        .split('/')
        .any(|segment| !segment.is_empty() && segment.starts_with('*'));
    if !has_legacy_wildcard {
        return Ok(path.to_string());
    }

    if !allow_legacy_wildcards {
        return Err(Error::new(
            ErrorKind::InvalidInput,
            format!(
                "CDC route '{}' uses legacy wildcard syntax (`/*name`). Use `{{*name}}` syntax or set {}=true in config.yaml to enable automatic legacy conversion.",
                path, CDC_ALLOW_LEGACY_AXUM_WILDCARDS_CONFIG_KEY
            ),
        ));
    }

    let normalized = path
        .split('/')
        .map(|segment| {
            if segment.is_empty() || !segment.starts_with('*') {
                segment.to_string()
            } else {
                let wildcard_name = segment.strip_prefix('*').unwrap_or_default().trim();
                let capture_name = if wildcard_name.is_empty() {
                    "wildcard"
                } else {
                    wildcard_name
                };
                format!("{{*{}}}", capture_name)
            }
        })
        .collect::<Vec<String>>()
        .join("/");

    info!(
        "Normalized legacy CDC route path '{}' -> '{}' due to {}=true",
        path, normalized, CDC_ALLOW_LEGACY_AXUM_WILDCARDS_CONFIG_KEY
    );
    Ok(normalized)
}

/// Starts the DMS WebSocket server.
///
/// Binds to `0.0.0.0:port`, exposes `/ws` (WebSocket), `/events` (POST), and `/status` (GET).
/// Clients must pass the `X-Athena-Client` header when connecting to subscribe
/// to events for that client. The header value is matched against `EventMessage::organization_id`.
///
/// # Errors
///
/// Returns an error if binding or serving the TCP listener fails.
pub async fn websocket_server(
    port: u16,
    allow_legacy_axum_wildcards: bool,
) -> Result<(), Box<dyn StdError>> {
    let capacity: usize = BROADCAST_CAPACITY;
    let (tx, _) = broadcast::channel::<String>(capacity);
    set_broadcast_tx(tx.clone());
    info!("DMS Broadcast channel created with capacity: {}", capacity);

    let active_subscribers: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
    let app_state: AppState = AppState {
        tx: tx.clone(),
        active_subscribers,
    };

    if allow_legacy_axum_wildcards {
        info!(
            "{}=true enabled; converting legacy CDC wildcard route syntax to Axum v0.8-compatible patterns",
            CDC_ALLOW_LEGACY_AXUM_WILDCARDS_CONFIG_KEY
        );
    }

    let ws_route = normalize_legacy_wildcard_path("/ws", allow_legacy_axum_wildcards)?;
    let events_route = normalize_legacy_wildcard_path("/events", allow_legacy_axum_wildcards)?;
    let status_route = normalize_legacy_wildcard_path("/status", allow_legacy_axum_wildcards)?;

    let app: Router = match catch_unwind(AssertUnwindSafe(|| {
        Router::new()
            .route(ws_route.as_str(), get(routes::ws_handler))
            .route(events_route.as_str(), post(routes::publish_event))
            .route(status_route.as_str(), get(routes::status))
            .with_state(app_state)
    })) {
        Ok(router) => router,
        Err(payload) => {
            let panic_message: String = panic_payload_to_string(payload);
            let message = if is_axum_legacy_wildcard_panic(&panic_message) {
                format!(
                    "CDC router path registration failed due to Axum wildcard validation. \
Use `{{*wildcard}}` syntax for wildcard captures, or set {}=true \
in config.yaml to opt in to legacy route syntax compatibility.",
                    CDC_ALLOW_LEGACY_AXUM_WILDCARDS_CONFIG_KEY
                )
            } else {
                format!("CDC router initialization failed: {}", panic_message)
            };
            return Err(Box::new(Error::new(ErrorKind::InvalidInput, message)));
        }
    };

    let addr: SocketAddr = SocketAddr::from(([0, 0, 0, 0], port));
    info!("Server running at http://{}", addr);
    info!("WebSocket endpoint available at ws://{}/ws", addr);

    let listener: TcpListener = TcpListener::bind(&addr).await?;
    axum::serve(listener, app).await?;
    info!("Server started successfully");

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::normalize_legacy_wildcard_path;

    #[test]
    fn normalize_legacy_wildcard_path_returns_original_for_non_legacy_paths() {
        let path = normalize_legacy_wildcard_path("/ws", false).expect("path should pass");
        assert_eq!(path, "/ws");
    }

    #[test]
    fn normalize_legacy_wildcard_path_rejects_legacy_path_without_opt_in() {
        let err = normalize_legacy_wildcard_path("/events/*rest", false)
            .expect_err("legacy wildcard path should be rejected without opt-in");
        assert!(
            err.to_string().contains("cdc.allow_legacy_axum_wildcards"),
            "error should describe how to enable legacy compatibility"
        );
    }

    #[test]
    fn normalize_legacy_wildcard_path_converts_legacy_path_with_opt_in() {
        let path = normalize_legacy_wildcard_path("/events/*rest", true)
            .expect("legacy wildcard path should be converted when opted in");
        assert_eq!(path, "/events/{*rest}");
    }
}
