//! Stopping without cutting anyone off (oxygenhub#233). On SIGTERM (a
//! `systemctl restart`) or Ctrl-C the hub stops taking connections, ends
//! its event streams (browsers reconnect by themselves), and gives the
//! requests in flight up to `DRAIN` to finish. Given a listening socket by
//! systemd (socket activation, docs/deploy.md), it serves that rather than
//! binding: the socket outlives the restart, so a request that arrives
//! between the old process and the new one waits in its queue instead of
//! being refused.

use std::sync::OnceLock;
use tokio::sync::watch;

/// How long the requests in flight get to finish.
pub const DRAIN: std::time::Duration = std::time::Duration::from_secs(30);

/// How long, after the drain, work still running on the runtime's
/// blocking threads (a slow request's repository work, a bucket call) may
/// hold the process. Dropping the runtime would wait for it however long
/// it takes: on 2026-10-17 a restart stood a minute past "stopped", until
/// systemd killed it.
pub const EXIT_GRACE: std::time::Duration = std::time::Duration::from_secs(1);

/// End the runtime once `serve` has returned, cutting off what still runs.
pub fn finish(rt: tokio::runtime::Runtime) {
    rt.shutdown_timeout(EXIT_GRACE);
}

fn sender() -> &'static watch::Sender<bool> {
    static STOP: OnceLock<watch::Sender<bool>> = OnceLock::new();
    STOP.get_or_init(|| watch::channel(false).0)
}

/// Begin stopping.
pub fn begin() {
    sender().send_replace(true);
}

pub fn is_stopping() -> bool {
    *sender().borrow()
}

/// Done when the hub begins stopping: for event streams to end on, and for
/// the servers' graceful shutdown.
pub async fn stopping() {
    let mut rx = sender().subscribe();
    let _ = rx.wait_for(|s| *s).await;
}

/// Wait for SIGTERM or Ctrl-C, then begin stopping. Installed before the
/// servers start, so a signal never kills the process outright.
pub fn on_signal() -> anyhow::Result<impl std::future::Future<Output = ()>> {
    #[cfg(unix)]
    let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
    Ok(async move {
        #[cfg(unix)]
        tokio::select! {
            _ = term.recv() => tracing::info!("SIGTERM: stopping (requests in flight may take {} s)", DRAIN.as_secs()),
            _ = tokio::signal::ctrl_c() => tracing::info!("stopping"),
        }
        #[cfg(not(unix))]
        {
            let _ = tokio::signal::ctrl_c().await;
            tracing::info!("stopping");
        }
        begin();
    })
}

/// The listening socket systemd passed (LISTEN_FDS, first descriptor 3),
/// when it passed one to this process.
pub fn systemd_listener() -> Option<std::net::TcpListener> {
    #[cfg(unix)]
    {
        use std::os::fd::FromRawFd;
        let pid: u32 = std::env::var("LISTEN_PID").ok()?.parse().ok()?;
        let fds: u32 = std::env::var("LISTEN_FDS").ok()?.parse().ok()?;
        if pid != std::process::id() || fds < 1 {
            return None;
        }
        // SAFETY: systemd hands this process descriptor 3 as an open,
        // listening socket (sd_listen_fds(3)); nothing else owns it.
        let listener = unsafe { std::net::TcpListener::from_raw_fd(3) };
        listener.set_nonblocking(true).ok()?;
        Some(listener)
    }
    #[cfg(not(unix))]
    None
}

#[cfg(test)]
mod tests {
    /// Blocking work still running when the hub stops does not hold the
    /// process: the runtime ends within its grace.
    #[test]
    fn finishing_does_not_wait_for_blocking_work() {
        let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
        rt.spawn_blocking(|| std::thread::sleep(std::time::Duration::from_secs(30)));
        let started = std::time::Instant::now();
        super::finish(rt);
        assert!(started.elapsed() < super::EXIT_GRACE * 3, "{:?}", started.elapsed());
    }
}