//! Process-wide signal listeners (`addSignalListener`).
//!
//! Signals are process-wide but avril runs workers as threads with their own
//! isolates, so delivery fans out: one process-lifetime feed per signal
//! pushes each delivery into every live isolate that currently binds it.
//! Listeners therefore fire in the entry isolate, the REPL, and workers
//! alike, and need no pledge: observing a delivery grants no authority.
//! But only the entry isolate (and the REPL's main isolate) suppress the
//! default death: worker listeners observe every delivery but never count
//! toward suppression, so with no entry listener a lethal delivery still
//! kills the process and worker delivery is best-effort.
//!
//! Listener feeds build on tokio's signal handling, where every
//! registration shares one chaining handler instead of replacing the
//! last. Serve claims SIGINT for shutdown and the REPL claims it for eval
//! interrupt through the same registry, so those owners keep running
//! alongside listeners regardless of install order: a listened SIGINT
//! still shuts serve down and still interrupts REPL evaluation. The
//! supervisor (supervision.rs) checks the listener count and skips its
//! own shutdown-and-die path while listeners exist.
//!
//! tokio's registry never restores the previous disposition once installed,
//! so feeds are never dropped: when the last listener goes away the feed
//! emulates the recorded prior fate (re-raising under the default
//! disposition, or nothing when it was ignored or owned) instead of
//! swallowing the signal forever.

use std::collections::{HashMap, VecDeque};
use std::convert::Infallible;
use std::io;
use std::sync::{Arc, Mutex, OnceLock, Weak};

use avril_core::Op;
use tokio::runtime::Handle;
use tokio::sync::Notify;

#[cfg(any(target_os = "linux", target_os = "macos"))]
use super::registry::supervision::Tree;
use avril_host::io::response;
#[cfg(any(target_os = "linux", target_os = "macos"))]
use avril_subprocess::signals::{Prior, original_prior};

struct Globals {
    /// Entry-isolate bindings per signal. Worker bindings fan out
    /// deliveries but never suppress: the supervisor consults these
    /// before carrying out a default death.
    counts: Mutex<HashMap<&'static str, usize>>,
    /// Feeds already running; feeds are process-lifetime, never removed.
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    feeds: Mutex<HashMap<&'static str, ()>>,
    /// Every live per-isolate state, for fanning deliveries out.
    registries: Mutex<Vec<Weak<State>>>,
}

static GLOBALS: OnceLock<Globals> = OnceLock::new();

fn globals() -> &'static Globals {
    GLOBALS.get_or_init(|| Globals {
        counts: Mutex::new(HashMap::new()),
        #[cfg(any(target_os = "linux", target_os = "macos"))]
        feeds: Mutex::new(HashMap::new()),
        registries: Mutex::new(Vec::new()),
    })
}

/// Per-isolate listener state, owned by the process extension's registry.
pub(crate) struct State {
    /// Whether this isolate is a worker. Worker bindings receive fan-out
    /// deliveries but never count toward death suppression.
    worker: bool,
    #[cfg(any(target_os = "linux", target_os = "macos"))]
    runtime: Handle,
    bindings: Mutex<HashMap<&'static str, usize>>,
    pending: Mutex<VecDeque<&'static str>>,
    notify: Notify,
}

impl State {
    pub(crate) fn create(runtime: Handle, worker: bool) -> Arc<Self> {
        let _ = &runtime;
        let state = Arc::new(Self {
            worker,
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            runtime,
            bindings: Mutex::new(HashMap::new()),
            pending: Mutex::new(VecDeque::new()),
            notify: Notify::new(),
        });
        globals()
            .registries
            .lock()
            .unwrap()
            .push(Arc::downgrade(&state));
        state
    }

    pub(crate) fn bind(&self, name: &str) -> io::Result<()> {
        bind_impl(self, name)
    }

    /// The next delivered signal bound by this isolate, or `None` once it
    /// binds nothing: the JS pump loop ends, releasing the event loop.
    pub(crate) async fn poll(&self) -> Option<String> {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            if let Some(name) = self.pending.lock().unwrap().pop_front() {
                return Some(name.to_owned());
            }
            if self.bindings.lock().unwrap().values().sum::<usize>() == 0 {
                return None;
            }
            notified.await;
        }
    }

    pub(crate) fn unbind(&self, name: &str) {
        let Some(signal) = lookup(name) else {
            return;
        };
        let mut bindings = self.bindings.lock().unwrap();
        let Some(count) = bindings.get_mut(signal) else {
            return;
        };
        *count = count.saturating_sub(1);
        if *count == 0 {
            bindings.remove(signal);
        }
        drop(bindings);
        if !self.worker {
            let mut counts = globals().counts.lock().unwrap();
            if let Some(count) = counts.get_mut(signal) {
                *count = count.saturating_sub(1);
                if *count == 0 {
                    counts.remove(signal);
                }
            }
            drop(counts);
        }
        // Wake a polling pump so it observes the removed binding.
        self.notify.notify_waiters();
    }

    /// Drops every binding this isolate still holds: entry close and
    /// isolate teardown restore the supervisor's default fate. Worker
    /// bindings never counted, so they have nothing to release.
    pub(crate) fn shutdown(&self) {
        let bindings = std::mem::take(&mut *self.bindings.lock().unwrap());
        if !self.worker && !bindings.is_empty() {
            let mut counts = globals().counts.lock().unwrap();
            for (signal, held) in &bindings {
                if let Some(count) = counts.get_mut(signal) {
                    *count = count.saturating_sub(*held);
                    if *count == 0 {
                        counts.remove(signal);
                    }
                }
            }
        }
        globals()
            .registries
            .lock()
            .unwrap()
            .retain(|state| state.strong_count() != 0);
        self.pending.lock().unwrap().clear();
        self.notify.notify_waiters();
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    fn bound(&self, signal: &'static str) -> bool {
        self.bindings.lock().unwrap().contains_key(signal)
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    fn push(&self, signal: &'static str) {
        self.pending.lock().unwrap().push_back(signal);
        self.notify.notify_waiters();
    }
}

/// How many entry isolates currently bind this signal. Worker bindings
/// are excluded: only the entry (and REPL main) isolate suppress death.
pub(crate) fn listener_count(name: &str) -> usize {
    globals()
        .counts
        .lock()
        .unwrap()
        .get(name)
        .copied()
        .unwrap_or(0)
}

pub(crate) fn ops(state: &Arc<State>) -> Vec<Op> {
    let mut ops = Vec::new();
    ops.push(Op::new("op_signal_bind", {
        let state = state.clone();
        move |name: String| Ok::<_, Infallible>(response(state.bind(&name)))
    }));
    ops.push(Op::asynchronous("op_signal_poll", {
        let state = state.clone();
        move |(): ()| {
            let state = state.clone();
            async move {
                // A response tuple like every other op: result() unwraps it.
                Ok::<_, Infallible>(response(Ok::<_, io::Error>(state.poll().await)))
            }
        }
    }));
    ops.push(Op::new("op_signal_unbind", {
        let state = state.clone();
        move |name: String| {
            state.unbind(&name);
            Ok::<_, Infallible>(())
        }
    }));
    ops
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
struct Signal {
    name: &'static str,
    number: libc::c_int,
    kind: fn() -> tokio::signal::unix::SignalKind,
    /// Whether the default disposition terminates the process. SIGWINCH's is
    /// to ignore the delivery.
    kills: bool,
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
const LISTENABLE: [Signal; 6] = [
    Signal {
        name: "SIGINT",
        number: libc::SIGINT,
        kind: tokio::signal::unix::SignalKind::interrupt,
        kills: true,
    },
    Signal {
        name: "SIGTERM",
        number: libc::SIGTERM,
        kind: tokio::signal::unix::SignalKind::terminate,
        kills: true,
    },
    Signal {
        name: "SIGHUP",
        number: libc::SIGHUP,
        kind: tokio::signal::unix::SignalKind::hangup,
        kills: true,
    },
    Signal {
        name: "SIGUSR1",
        number: libc::SIGUSR1,
        kind: tokio::signal::unix::SignalKind::user_defined1,
        kills: true,
    },
    Signal {
        name: "SIGUSR2",
        number: libc::SIGUSR2,
        kind: tokio::signal::unix::SignalKind::user_defined2,
        kills: true,
    },
    Signal {
        name: "SIGWINCH",
        number: libc::SIGWINCH,
        kind: tokio::signal::unix::SignalKind::window_change,
        kills: false,
    },
];

#[cfg(any(target_os = "linux", target_os = "macos"))]
fn lookup(name: &str) -> Option<&'static str> {
    LISTENABLE
        .iter()
        .find(|signal| signal.name == name)
        .map(|signal| signal.name)
}

#[cfg(not(any(target_os = "linux", target_os = "macos")))]
fn lookup(_name: &str) -> Option<&'static str> {
    None
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
fn bind_impl(state: &State, name: &str) -> io::Result<()> {
    use super::sys::known_signal;

    let Some(signal) = LISTENABLE.iter().find(|signal| signal.name == name) else {
        if matches!(name, "SIGKILL" | "SIGSTOP") {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("{name} cannot be intercepted"),
            ));
        }
        if known_signal(name) {
            return Err(io::Error::new(
                io::ErrorKind::Unsupported,
                format!("{name} signal listeners are not supported"),
            ));
        }
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("Unknown signal {name}"),
        ));
    };
    // Install the process-wide feed before marking anything: a failed
    // install must not pin the signal. The single insert doubles as the
    // already-running check, so concurrent binds cannot start two feeds.
    let fresh = globals()
        .feeds
        .lock()
        .unwrap()
        .insert(signal.name, ())
        .is_none();
    if fresh {
        // An inherited SIG_IGN stays ignored for supervision, but an
        // explicit listener takes over the signal: that is the point of
        // listening.
        let _enter = state.runtime.enter();
        match tokio::signal::unix::signal((signal.kind)()) {
            Ok(stream) => {
                state
                    .runtime
                    .spawn(feed(signal.name, signal.number, stream));
            }
            Err(error) => {
                globals().feeds.lock().unwrap().remove(signal.name);
                return Err(error);
            }
        }
    }
    *state
        .bindings
        .lock()
        .unwrap()
        .entry(signal.name)
        .or_insert(0) += 1;
    if !state.worker {
        *globals()
            .counts
            .lock()
            .unwrap()
            .entry(signal.name)
            .or_insert(0) += 1;
    }
    Ok(())
}

#[cfg(not(any(target_os = "linux", target_os = "macos")))]
fn bind_impl(_state: &State, _name: &str) -> io::Result<()> {
    Err(io::Error::new(
        io::ErrorKind::Unsupported,
        "Signal listeners are not supported on this platform",
    ))
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
async fn feed(name: &'static str, number: libc::c_int, mut stream: tokio::signal::unix::Signal) {
    let kills = LISTENABLE
        .iter()
        .find(|signal| signal.name == name)
        .is_some_and(|signal| signal.kills);
    loop {
        if stream.recv().await.is_none() {
            return;
        }
        // Fan out first: worker listeners observe every delivery even
        // when nobody suppresses, and delivery racing a real death is
        // best-effort.
        let targets = {
            let mut registries = globals().registries.lock().unwrap();
            registries.retain(|state| state.strong_count() != 0);
            registries
                .iter()
                .filter_map(Weak::upgrade)
                .filter(|state| state.bound(name))
                .collect::<Vec<_>>()
        };
        for target in targets {
            target.push(name);
        }
        if listener_count(name) == 0
            && kills
            && original_prior(number) == Prior::Default
            && !Tree::global().holds(number)
        {
            // Nobody suppresses: emulate the startup fate, since the
            // installed handler would otherwise swallow the delivery. The
            // supervisor owns its signals' deaths itself.
            die(number);
        }
    }
}

#[cfg(any(target_os = "linux", target_os = "macos"))]
fn die(number: libc::c_int) {
    unsafe {
        libc::signal(number, libc::SIG_DFL);
        libc::raise(number);
    }
}