pub mod error;
pub mod middleware;
pub mod routes;

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::Duration;

use axum::Router;
use axum::routing::{get, post, put};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio_rustls::server::TlsStream;

use std::path::PathBuf;

use crate::auth::Authenticator;
use crate::config::{DeployConfig, HttpConfig, ReloadState, SharedConfig};
use crate::error::Result;
use crate::namespace::NamespaceManager;
use crate::retention::GracePendingMap;
use crate::server::state::{ServerLifecycle, ServerState};
use crate::tls::{TlsProvider, TlsProviders};

/// Shared handle to the map of topics currently in the reap grace period.
///
/// Keyed by `(NamespaceId, TopicName)`, value is a [`GracePendingEntry`] with
/// the instant the topic entered grace plus the config values captured at that
/// time. Updated exclusively by `RetentionTask`.
///
/// [`GracePendingEntry`]: crate::retention::GracePendingEntry
pub type GracePendingHandle = Arc<tokio::sync::RwLock<GracePendingMap>>;

use routes::{
    admin_config, admin_drain, admin_drain_status, admin_pending_reap, admin_reload,
    admin_tls_reload, admin_tls_status, fetch, health_live, health_ready, list_topics, log_level,
    log_level_get, metrics, publish, require_admin_auth, require_initialized, topic_info,
};

/// Handle for dynamically reloading the log level filter.
pub type LogFilterReloadHandle =
    tracing_subscriber::reload::Handle<tracing_subscriber::EnvFilter, tracing_subscriber::Registry>;

/// Request to trigger a drain via the admin HTTP API.
pub struct DrainRequest {
    pub reason: Option<String>,
    pub timeout_secs: Option<u64>,
    pub response: tokio::sync::oneshot::Sender<std::result::Result<DrainResponse, String>>,
}

/// Response returned after a drain is initiated.
pub struct DrainResponse {
    pub active_connections: u64,
    pub drain_started_wall: u64,
    pub drain_deadline_wall: u64,
}

/// Shared state passed to all HTTP handlers via axum's State extractor.
pub struct AppState<M: NamespaceManager> {
    pub manager: Arc<M>,
    pub auth: Arc<dyn Authenticator>,
    pub message_id_counter: Arc<AtomicU64>,
    pub prom_handle: Option<metrics_exporter_prometheus::PrometheusHandle>,
    pub log_reload_handle: Arc<LogFilterReloadHandle>,
    pub lifecycle: Arc<ServerLifecycle>,
    pub admin_api_key: Option<String>,
    pub drain_trigger: Option<mpsc::Sender<DrainRequest>>,
    pub deploy_config: DeployConfig,
    pub shared_config: SharedConfig,
    /// Path to the config file that was loaded. `None` when running on compiled
    /// defaults (no file found). Used by the reload endpoint to know which file
    /// to re-read.
    pub config_path: Option<Arc<PathBuf>>,
    /// Shared reload state for the `POST /admin/reload` endpoint.
    /// `None` when no config file was loaded (compiled defaults — reload not possible).
    pub reload_state: Option<Arc<ReloadState>>,
    /// Topics currently in the reap grace period, shared with `RetentionTask`.
    /// Used by `GET /admin/topics/pending-reap`.
    pub grace_pending: GracePendingHandle,
}

// Manual Clone impl to avoid requiring M: Clone.
impl<M: NamespaceManager> Clone for AppState<M> {
    fn clone(&self) -> Self {
        Self {
            manager: Arc::clone(&self.manager),
            auth: Arc::clone(&self.auth),
            message_id_counter: Arc::clone(&self.message_id_counter),
            prom_handle: self.prom_handle.clone(),
            log_reload_handle: Arc::clone(&self.log_reload_handle),
            lifecycle: Arc::clone(&self.lifecycle),
            admin_api_key: self.admin_api_key.clone(),
            drain_trigger: self.drain_trigger.clone(),
            deploy_config: self.deploy_config.clone(),
            shared_config: Arc::clone(&self.shared_config),
            config_path: self.config_path.clone(),
            reload_state: self.reload_state.clone(),
            grace_pending: Arc::clone(&self.grace_pending),
        }
    }
}

/// TLS-wrapping listener for HTTPS.
///
/// Accepts TCP connections and performs TLS handshakes concurrently.
/// Multiple handshakes proceed in parallel via a `JoinSet` — a slow or
/// malicious client performing a partial handshake does not block other
/// connections. `accept()` returns the first successfully completed
/// handshake while continuing to accept new TCP connections.
struct TlsListener {
    inner: TcpListener,
    acceptor: Arc<TlsProvider>,
    handshake_timeout: Duration,
    /// In-flight TLS handshakes. Each task resolves to `Some(stream, addr)`
    /// on success or `None` on failure/timeout.
    pending: tokio::task::JoinSet<Option<(TlsStream<TcpStream>, SocketAddr)>>,
}

impl axum::serve::Listener for TlsListener {
    type Io = TlsStream<TcpStream>;
    type Addr = SocketAddr;

    async fn accept(&mut self) -> (Self::Io, Self::Addr) {
        loop {
            tokio::select! {
                biased;

                // Drain completed handshakes first (prevents unbounded accumulation).
                Some(result) = self.pending.join_next(), if !self.pending.is_empty() => {
                    match result {
                        Ok(Some(conn)) => return conn,
                        Ok(None) => {} // handshake failed/timed out — already logged
                        Err(e) => {
                            tracing::debug!(error = %e, "HTTPS TLS handshake task panicked");
                        }
                    }
                }

                // Accept new TCP connections and spawn concurrent handshake tasks.
                result = self.inner.accept() => {
                    match result {
                        Ok((stream, addr)) => {
                            let acceptor = Arc::clone(&self.acceptor);
                            let timeout = self.handshake_timeout;
                            self.pending.spawn(async move {
                                let start = std::time::Instant::now();
                                match tokio::time::timeout(timeout, acceptor.accept(stream)).await {
                                    Ok(Ok(tls_stream)) => {
                                        let elapsed = start.elapsed();
                                        metrics::counter!("fafb_tls_handshake_total", "status" => "success")
                                            .increment(1);
                                        metrics::histogram!("fafb_tls_handshake_duration_seconds")
                                            .record(elapsed.as_secs_f64());
                                        Some((tls_stream, addr))
                                    }
                                    Ok(Err(e)) => {
                                        tracing::debug!(peer = %addr, error = %e, "HTTPS TLS handshake failed");
                                        metrics::counter!("fafb_tls_handshake_total", "status" => "error")
                                            .increment(1);
                                        metrics::histogram!("fafb_tls_handshake_duration_seconds")
                                            .record(start.elapsed().as_secs_f64());
                                        None
                                    }
                                    Err(_) => {
                                        tracing::debug!(
                                            peer = %addr,
                                            timeout_secs = timeout.as_secs(),
                                            "HTTPS TLS handshake timed out"
                                        );
                                        metrics::counter!("fafb_tls_handshake_total", "status" => "timeout")
                                            .increment(1);
                                        metrics::counter!("fafb_tls_handshake_timeout_total").increment(1);
                                        None
                                    }
                                }
                            });
                        }
                        Err(e) => {
                            tracing::debug!(error = %e, "HTTPS TCP accept failed");
                        }
                    }
                }
            }
        }
    }

    fn local_addr(&self) -> std::io::Result<Self::Addr> {
        self.inner.local_addr()
    }
}

/// HTTP server for publish and query endpoints.
///
/// Mirrors the TCP server's constructor pattern: generic over the namespace
/// manager, holds shared auth and config, supports graceful shutdown.
pub struct HttpServer<M: NamespaceManager> {
    config: HttpConfig,
    state: AppState<M>,
    tls_provider: Option<Arc<TlsProvider>>,
    tls_providers: TlsProviders,
}

impl<M: NamespaceManager + 'static> HttpServer<M> {
    /// Create a new HTTP server from pre-built shared state.
    pub fn new(
        config: HttpConfig,
        state: AppState<M>,
        tls_provider: Option<Arc<TlsProvider>>,
    ) -> Self {
        Self {
            config,
            state,
            tls_provider,
            tls_providers: TlsProviders::default(),
        }
    }

    /// Set the labeled TLS providers for admin cert reload/status endpoints.
    pub fn with_tls_providers(mut self, providers: TlsProviders) -> Self {
        self.tls_providers = providers;
        self
    }

    /// Build the axum router with all routes.
    fn router(&self) -> Router {
        let lifecycle = Arc::clone(&self.state.lifecycle);
        let api_routes = Router::new()
            .route("/metrics", get(metrics::<M>))
            .route("/{namespace}/{topic}/publish", post(publish::<M>))
            .route("/{namespace}/{topic}/fetch", get(fetch::<M>))
            .route("/{namespace}/{topic}/info", get(topic_info::<M>))
            .route("/{namespace}/topics", get(list_topics::<M>))
            .route(
                "/admin/log-level",
                put(log_level::<M>).get(log_level_get::<M>),
            )
            .route_layer(axum::middleware::from_fn_with_state(
                lifecycle,
                require_initialized,
            ));

        let admin_routes = Router::new()
            .route("/drain", post(admin_drain::<M>))
            .route("/drain/status", get(admin_drain_status::<M>))
            .route("/reload", post(admin_reload::<M>))
            .route("/config", get(admin_config::<M>))
            .route("/tls/reload", post(admin_tls_reload::<M>))
            .route("/tls/status", get(admin_tls_status::<M>))
            .route("/topics/pending-reap", get(admin_pending_reap::<M>))
            .layer(axum::Extension(self.tls_providers.clone()))
            .route_layer(axum::middleware::from_fn_with_state(
                self.state.clone(),
                require_admin_auth::<M>,
            ));

        Router::new()
            .route("/health", get(health_ready::<M>))
            .route("/health/live", get(health_live))
            .route("/health/ready", get(health_ready::<M>))
            .nest("/admin", admin_routes)
            .nest("/v1", api_routes)
            .with_state(self.state.clone())
    }

    /// Bind to the configured address/port and serve.
    pub async fn run(&self) -> Result<()> {
        let addr = format!("{}:{}", self.config.bind_address, self.config.port);
        let listener = TcpListener::bind(&addr).await?;
        tracing::info!(address = %addr, tls = self.tls_provider.is_some(), "HTTP server listening");
        if self.tls_provider.is_some() {
            tracing::info!("TLS enabled for HTTP listener");
        } else {
            tracing::warn!(
                "TLS is disabled for HTTP — connections are unencrypted. \
                 Set [tls] enabled=true with cert_file/key_file for production use."
            );
        }
        self.serve(listener).await
    }

    /// Serve with a pre-bound listener (useful for tests binding to port 0).
    pub async fn run_with_listener(&self, listener: TcpListener) -> Result<()> {
        tracing::info!(
            address = %listener.local_addr().map(|a| a.to_string()).unwrap_or_default(),
            tls = self.tls_provider.is_some(),
            "HTTP server listening"
        );
        self.serve(listener).await
    }

    async fn serve(&self, listener: TcpListener) -> Result<()> {
        let router = self.router();

        match &self.tls_provider {
            Some(tls) => {
                let tls_listener = TlsListener {
                    inner: listener,
                    acceptor: Arc::clone(tls),
                    handshake_timeout: tls.handshake_timeout(),
                    pending: tokio::task::JoinSet::new(),
                };
                let mut lifecycle_rx = self.state.lifecycle.subscribe();
                axum::serve(tls_listener, router)
                    .with_graceful_shutdown(async move {
                        loop {
                            if lifecycle_rx.changed().await.is_err() {
                                break;
                            }
                            if *lifecycle_rx.borrow() >= ServerState::ShuttingDown {
                                break;
                            }
                        }
                    })
                    .await?;
            }
            None => {
                let mut lifecycle_rx = self.state.lifecycle.subscribe();
                axum::serve(listener, router)
                    .with_graceful_shutdown(async move {
                        loop {
                            if lifecycle_rx.changed().await.is_err() {
                                break;
                            }
                            if *lifecycle_rx.borrow() >= ServerState::ShuttingDown {
                                break;
                            }
                        }
                    })
                    .await?;
            }
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::namespace::DefaultNamespaceManager;
    use crate::queue::InMemoryQueueEngine;
    use crate::storage::mock::MockStorage;

    #[test]
    fn server_is_send() {
        fn assert_send<T: Send>() {}
        assert_send::<HttpServer<DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>>>>();
    }
}
