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

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

}