//! 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());

    }

}