use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use efdn_telem::canonical::service::ServiceInfo;
use efdn_telem::metrics::MetricsHandle;
use tokio::sync::{Mutex as AsyncMutex, Notify, Semaphore, watch};
use tokio::task::JoinHandle;
use tokio_util::task::TaskTracker;

use crate::config::{GatewayBind, GatewayConfig, GatewayMetricsRoute, GatewayRequestTimeout};
use crate::error::GatewayError;
use crate::health::HealthHandle;
use crate::metrics;
use crate::service::GatewayServiceInfo;

// Enough for scrape bursts while bounding idle socket abuse
const MAX_CONNECTIONS: usize = 1024;

#[path = "routes.rs"]
mod routes;
#[path = "server.rs"]
mod server;

pub use server::start;

/// Active telemetry gateway handle
#[derive(Clone, Debug)]
#[must_use = "keep GatewayHandle alive until service shutdown"]
pub struct GatewayHandle {
    inner: Arc<GatewayInner>,
}

#[derive(Debug)]
struct GatewayInner {
    service: ServiceInfo,
    input: GatewayServiceInfo,
    health: HealthHandle,
    metrics: Mutex<Option<MetricsHandle>>,
    metrics_route: GatewayMetricsRoute,
    request_timeout: GatewayRequestTimeout,
    local_addr: Option<GatewayBind>,
    shutdown: Option<watch::Sender<bool>>,
    task: Mutex<Option<JoinHandle<()>>>,
    connections: TaskTracker,
    connection_limit: Arc<Semaphore>,
    shutdown_once: AtomicBool,
    shutdown_done: Notify,
    shutdown_result: AsyncMutex<Option<Result<(), GatewayError>>>,
    connection_errors: Mutex<Vec<String>>,
}

impl GatewayHandle {
    fn new(
        config: &GatewayConfig,
        service: ServiceInfo,
        input: GatewayServiceInfo,
        shutdown: Option<watch::Sender<bool>>,
        task: Option<JoinHandle<()>>,
        local_addr: Option<GatewayBind>,
    ) -> Self {
        Self {
            inner: Arc::new(GatewayInner {
                service,
                input,
                health: HealthHandle::new(),
                metrics: Mutex::new(None),
                metrics_route: config.metrics,
                request_timeout: config.request_timeout,
                local_addr,
                shutdown,
                task: Mutex::new(task),
                connections: TaskTracker::new(),
                connection_limit: Arc::new(Semaphore::new(MAX_CONNECTIONS)),
                shutdown_once: AtomicBool::new(false),
                shutdown_done: Notify::new(),
                shutdown_result: AsyncMutex::new(None),
                connection_errors: Mutex::new(Vec::new()),
            }),
        }
    }

    /// Returns the service identity used by metrics and canonical records.
    pub fn service(&self) -> ServiceInfo {
        self.inner.service.clone()
    }

    /// Returns the gateway health handle.
    pub fn health(&self) -> HealthHandle {
        self.inner.health.clone()
    }

    /// Attaches a metrics registry to the `/metrics` route.
    ///
    /// # Errors
    /// [`GatewayError`] if metric descriptions cannot be registered.
    pub fn attach_metrics(&self, metrics: MetricsHandle) -> Result<(), GatewayError> {
        crate::metrics::describe();
        *self
            .inner
            .metrics
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(metrics);
        Ok(())
    }

    /// Returns the bound listener address when the gateway is enabled.
    pub fn local_addr(&self) -> Option<GatewayBind> {
        self.inner.local_addr
    }

    /// Stops the gateway listener and active connection tasks.
    ///
    /// # Errors
    /// [`GatewayError`] if the listener task or a connection task failed.
    pub async fn shutdown(&self) -> Result<(), GatewayError> {
        if !self.inner.shutdown_once.swap(true, Ordering::AcqRel) {
            self.spawn_shutdown();
        }
        self.await_shutdown_result().await
    }

    fn inner(&self) -> Arc<GatewayInner> {
        Arc::clone(&self.inner)
    }

    fn set_task(&self, task: JoinHandle<()>) {
        *self
            .inner
            .task
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(task);
    }

    fn spawn_shutdown(&self) {
        let handle = self.clone();
        tokio::spawn(async move {
            let result = handle.run_shutdown().await;
            *handle.inner.shutdown_result.lock().await = Some(result);
            handle.inner.shutdown_done.notify_waiters();
        });
    }

    async fn await_shutdown_result(&self) -> Result<(), GatewayError> {
        loop {
            let notified = self.inner.shutdown_done.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            if let Some(result) = self.inner.shutdown_result.lock().await.clone() {
                return result;
            }
            notified.await;
        }
    }

    async fn run_shutdown(&self) -> Result<(), GatewayError> {
        if let Some(shutdown) = &self.inner.shutdown {
            let _ = shutdown.send(true);
        }
        let listener = self.await_task().await;
        self.inner.connections.close();
        self.inner.connections.wait().await;
        combine_gateway_results(listener, self.connection_task_result())
    }

    async fn await_task(&self) -> Result<(), GatewayError> {
        let Some(task) = self
            .inner
            .task
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .take()
        else {
            return Ok(());
        };
        task.await
            .map_err(|error| GatewayError::ServerTask(error.to_string()))
    }

    fn connection_task_result(&self) -> Result<(), GatewayError> {
        let mut errors = self
            .inner
            .connection_errors
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if errors.is_empty() {
            return Ok(());
        }
        Err(GatewayError::ServerTask(
            std::mem::take(&mut *errors).join("\n"),
        ))
    }
}

fn combine_gateway_results(
    listener: Result<(), GatewayError>,
    connections: Result<(), GatewayError>,
) -> Result<(), GatewayError> {
    let failures = [listener, connections]
        .into_iter()
        .filter_map(Result::err)
        .map(|error| error.to_string())
        .collect::<Vec<_>>();
    if failures.is_empty() {
        return Ok(());
    }
    Err(GatewayError::ServerTask(failures.join("\n")))
}

impl GatewayInner {
    fn metrics_handle(&self) -> Option<MetricsHandle> {
        self.metrics
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    fn metrics_attached(&self) -> bool {
        self.metrics
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .is_some()
    }

    fn metrics_body(&self) -> Result<Option<String>, GatewayError> {
        let Some(handle) = self.metrics_handle() else {
            return Ok(None);
        };
        metrics::render(&handle, &self.health).map(Some)
    }
}
