use std::io;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::Duration;

use tokio::runtime::Handle;
use tokio_util::sync::CancellationToken;

pub struct Tree {
    processes: Mutex<Vec<Weak<super::Process>>>,
    sync_interrupt: AtomicU64,
    installed: AtomicBool,
    /// Whether the install owns SIGINT (run mode). Serve and the REPL
    /// claim SIGINT through the tree instead; listener feeds consult
    /// holds() to tell a supervisor-owned death from one they must
    /// emulate themselves.
    interrupt: AtomicBool,
    /// Claimed by the subsystem that owns SIGINT shutdown or interrupt
    /// handling (serve's CLI listener, the REPL). Read by the supervisor
    /// loop when it holds SIGINT; otherwise served by claim_interrupt's
    /// dedicated feed. Either way the owner shares one chaining registry
    /// with listener feeds, so install order never matters.
    #[cfg_attr(windows, allow(dead_code))]
    interrupt_owner: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
    /// Guards claim_interrupt's dedicated feed against double installs.
    #[cfg(unix)]
    interrupt_claimed: AtomicBool,
    /// Blocks new spawns once the process is dying, so respawn loops cannot
    /// relaunch children during the shutdown grace period.
    closed: AtomicBool,
    /// Claimed by the subsystem that owns SIGTERM shutdown (serve's CLI
    /// listener). The tree is the sole TERM listener, so claiming and
    /// dispatch serialize on this slot: a delivery either runs the owner or
    /// predates it, never both and never neither.
    #[cfg_attr(windows, allow(dead_code))]
    term_owner: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
    pub(super) stopped: CancellationToken,
}
impl Tree {
    pub fn global() -> Arc<Self> {
        static TREE: OnceLock<Arc<Tree>> = OnceLock::new();
        TREE.get_or_init(|| {
            Arc::new(Self {
                processes: Mutex::new(Vec::new()),
                sync_interrupt: AtomicU64::new(0),
                installed: AtomicBool::new(false),
                interrupt: AtomicBool::new(false),
                interrupt_owner: Mutex::new(None),
                #[cfg(unix)]
                interrupt_claimed: AtomicBool::new(false),
                closed: AtomicBool::new(false),
                term_owner: Mutex::new(None),
                stopped: CancellationToken::new(),
            })
        })
        .clone()
    }
    pub(super) fn register(&self, process: &Arc<super::Process>) {
        let mut processes = self.processes.lock().unwrap();
        processes.retain(|process| process.strong_count() != 0);
        processes.push(Arc::downgrade(process));
    }
    pub fn force_live_processes(&self) {
        self.closed.store(true, Ordering::Release);
        for process in self.live() {
            process.request("terminated", None, true);
        }
    }
    pub(super) fn sync_generation(&self) -> u64 {
        self.sync_interrupt.load(Ordering::SeqCst)
    }
    #[cfg(unix)]
    pub fn claim_term(
        self: &Arc<Self>,
        runtime: &Handle,
        owner: Arc<dyn Fn() + Send + Sync>,
    ) -> io::Result<()> {
        *self.term_owner.lock().unwrap() = Some(owner);
        self.install(runtime, false)
    }
    /// Hands SIGINT deliveries to a subsystem owner through the shared
    /// chaining registry, alongside any listener feeds. Modes whose
    /// supervisor loop does not hold SIGINT get a dedicated feed; run
    /// mode's loop routes to the owner itself.
    #[cfg(unix)]
    pub fn claim_interrupt(
        self: &Arc<Self>,
        runtime: &Handle,
        owner: Arc<dyn Fn() + Send + Sync>,
    ) -> io::Result<()> {
        *self.interrupt_owner.lock().unwrap() = Some(owner.clone());
        if self.interrupt.load(Ordering::Acquire)
            || self.interrupt_claimed.swap(true, Ordering::AcqRel)
        {
            return Ok(());
        }
        let _enter = runtime.enter();
        let mut stream =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) {
                Ok(stream) => stream,
                Err(error) => {
                    *self.interrupt_owner.lock().unwrap() = None;
                    self.interrupt_claimed.store(false, Ordering::Release);
                    return Err(error);
                }
            };
        runtime.spawn(async move {
            while stream.recv().await.is_some() {
                owner();
            }
        });
        Ok(())
    }
    pub(super) fn is_closed(&self) -> bool {
        self.closed.load(Ordering::Acquire)
    }
    /// Whether the supervisor itself holds this signal's handler, and so
    /// owns its default fate. Listener feeds emulate the prior disposition
    /// only for signals nobody holds.
    #[cfg(unix)]
    pub(crate) fn holds(&self, signal: libc::c_int) -> bool {
        if !self.installed.load(Ordering::Acquire) {
            return false;
        }
        signal == libc::SIGTERM
            || (signal == libc::SIGINT
                && (self.interrupt.load(Ordering::Acquire)
                    || self.interrupt_owner.lock().unwrap().is_some()))
    }
    /// Asks every live process to shut down with its configured signal and
    /// grace period. One shot: already-finished processes ignore the request,
    /// and later spawns start fresh.
    pub(crate) fn stop_live_processes(&self) {
        for process in self.live() {
            process.request("terminated", None, false);
        }
    }
    /// Stops only processes blocked in a synchronous wait, leaving other
    /// session work alone. The REPL uses this so Ctrl-C interrupts runSync
    /// without cancelling background operations.
    pub fn stop_sync_processes(&self) {
        // Pair the generation with runSync's waiter publication. SeqCst
        // ensures an interrupt racing spawn sees the waiter, or the waiter
        // sees the interrupt, rather than both missing each other.
        self.sync_interrupt.fetch_add(1, Ordering::SeqCst);
        for process in self.live() {
            if !process.worker && process.sync_wait.load(Ordering::SeqCst) {
                process.request("terminated", None, false);
            }
        }
    }
    fn live(&self) -> Vec<Arc<super::Process>> {
        self.processes
            .lock()
            .unwrap()
            .iter()
            .filter_map(Weak::upgrade)
            .collect()
    }
    pub(super) fn install(
        self: &Arc<Self>,
        runtime: &Handle,
        handle_interrupt: bool,
    ) -> io::Result<()> {
        if self.installed.swap(true, Ordering::AcqRel) {
            return Ok(());
        }
        self.interrupt.store(handle_interrupt, Ordering::Release);
        let _enter = runtime.enter();
        let mut signals = match Signals::listen(handle_interrupt) {
            Ok(Some(signals)) => signals,
            Ok(None) => return Ok(()),
            Err(error) => {
                self.installed.store(false, Ordering::Release);
                return Err(error);
            }
        };
        let tree = self.clone();
        runtime.spawn(async move {
            loop {
                let delivery = signals.recv().await;
                // A claimed interrupt belongs to its owner (serve
                // shutdown, REPL eval interrupt); listeners still fire
                // through their own feeds.
                #[cfg(unix)]
                if matches!(delivery, Delivery::Interrupt) {
                    let owner = tree.interrupt_owner.lock().unwrap().clone();
                    if let Some(owner) = owner {
                        owner();
                        continue;
                    }
                }
                // Claimed TERM belongs to its owner, including repeats.
                // In-flight children keep running during serve's drain.
                #[cfg(unix)]
                if matches!(delivery, Delivery::Terminate) {
                    let owner = tree.term_owner.lock().unwrap().clone();
                    if let Some(owner) = owner {
                        owner();
                        continue;
                    }
                }
                // Listened signals belong to their JS listeners: skip the
                // shutdown and death below, so delivery does not kill the
                // process while listeners exist. Serve's claimed TERM still
                // runs its owner above.
                if super::super::signals::listener_count(delivery.name()) > 0 {
                    continue;
                }
                tree.closed.store(true, Ordering::Release);
                tree.stop_live_processes();
                // Bounded shutdown, abortable by an impatient repeat.
                let live = tree.live();
                let pending =
                    futures::future::join_all(live.iter().map(|process| process.wait(true, None)));
                tokio::select! {
                    _ = tokio::time::timeout(Duration::from_secs(2), pending) => {},
                    _ = signals.recv() => {},
                }
                // Grace-resisting descendants cannot outlive the owner.
                for process in tree.live() {
                    process.kill_scope();
                }
                tree.stopped.cancel();
                delivery.die();
                return;
            }
        });
        Ok(())
    }
}

#[derive(Clone, Copy)]
enum Delivery {
    Interrupt,
    #[cfg_attr(windows, allow(dead_code))]
    Terminate,
}

impl Delivery {
    fn name(self) -> &'static str {
        match self {
            Self::Interrupt => "SIGINT",
            Self::Terminate => "SIGTERM",
        }
    }
}

struct Signals {
    interrupt: Option<Stream>,
    terminate: Option<Stream>,
}
impl Signals {
    async fn recv(&mut self) -> Delivery {
        async fn next(stream: &mut Option<Stream>) {
            if let Some(stream) = stream
                && stream.recv().await.is_some()
            {
                return;
            }
            std::future::pending().await
        }
        tokio::select! {
            _ = next(&mut self.interrupt) => Delivery::Interrupt,
            _ = next(&mut self.terminate) => Delivery::Terminate,
        }
    }
}

#[cfg(unix)]
type Stream = tokio::signal::unix::Signal;
#[cfg(unix)]
impl Signals {
    fn listen(handle_interrupt: bool) -> io::Result<Option<Self>> {
        use tokio::signal::unix::{SignalKind, signal};
        // An inherited SIG_IGN stays ignored: taking over the signal would
        // change the fate of a process that was launched to survive it.
        // Consult the startup snapshot, never the live disposition, which
        // earlier listener feeds may already have taken over.
        let interrupt =
            if handle_interrupt && !avril_subprocess::signals::originally_ignored(libc::SIGINT) {
                Some(signal(SignalKind::interrupt())?)
            } else {
                None
            };
        let terminate = if avril_subprocess::signals::originally_ignored(libc::SIGTERM) {
            None
        } else {
            Some(signal(SignalKind::terminate())?)
        };
        Ok((interrupt.is_some() || terminate.is_some()).then_some(Self { interrupt, terminate }))
    }
}
#[cfg(unix)]
impl Delivery {
    fn die(self) {
        let number = match self {
            Self::Interrupt => libc::SIGINT,
            Self::Terminate => libc::SIGTERM,
        };
        unsafe {
            libc::signal(number, libc::SIG_DFL);
            libc::raise(number);
        }
    }
}

// Windows has no termination signal to supervise: console close, logoff and
// shutdown end the process, and kill-on-close jobs take the children along.
#[cfg(windows)]
type Stream = tokio::signal::windows::CtrlC;
#[cfg(windows)]
impl Signals {
    fn listen(handle_interrupt: bool) -> io::Result<Option<Self>> {
        if !handle_interrupt {
            return Ok(None);
        }
        Ok(Some(Self {
            interrupt: Some(tokio::signal::windows::ctrl_c()?),
            terminate: None,
        }))
    }
}
#[cfg(windows)]
impl Delivery {
    fn die(self) {
        // The default console control handler exits with STATUS_CONTROL_C_EXIT.
        std::process::exit(0xC000013Au32 as i32);
    }
}