//! Process monitors.
//!
//! `monitor(target)` asks the runtime to deliver a single [`Down`] when
//! `target` terminates, and hands back a [`Monitor`] — the [`Receiver`] to read
//! it from, plus the identity (`id`, `target`) needed to take the registration
//! back down with [`demonitor`]. A monitor is:
//!
//!   - **unidirectional** — the watcher learns of the target's death, but the
//!     target learns nothing of the watcher, and the watcher is unaffected by
//!     the death beyond the notification (contrast a *link*, which propagates
//!     failure);
//!   - **one-shot** — exactly one `Down` is ever sent for a given monitor.
//!     The returned channel closes afterwards, so a second `recv()` yields
//!     `Err(RecvError)`.
//!
//! This generalizes the older single-`supervisor_channel` mechanism: a
//! supervisor is just a hard-wired monitor that the parent installs at spawn
//! time. Here any actor may monitor any pid, any number of times.
//!
//! ## Reasons
//!
//! [`DownReason`] is deliberately payload-free. A panicking actor's payload
//! has a single owner and is delivered to whoever `join()`s the actor (as
//! `JoinError`); a monitor only learns *that* it panicked, not the value.
//! Monitoring a pid that is already gone (reclaimed, or never alive) yields
//! [`DownReason::NoProc`] immediately, mirroring Erlang's `noproc`.
//!
//! ## Demonitoring
//!
//! Each `monitor()` registration is tagged with a process-unique [`MonitorId`].
//! [`demonitor`] removes the registration named by a [`Monitor`] from its
//! target's slot, returning `Some(id)` if a live registration was found or
//! `None` if it had already fired (or the target is gone). Dropping the
//! [`Monitor`] afterwards discards any `Down` that the target had *already*
//! queued — the equivalent of Erlang's `demonitor(Ref, [flush])`.
//!
//! ## Races
//!
//! Registration (below) and `finalize_actor` (in `runtime`) both run under the
//! shared-state mutex, so a target that is still alive when its monitor is
//! registered is guaranteed to deliver a real `Down`; there is no window in
//! which the death slips between the liveness check and the registration.
//! `demonitor` is protected by the generation half of the pid: if the target
//! has died and its slot index been recycled, `slot_mut(target)` fails the
//! generation check and `demonitor` is a clean no-op — it can never strip a
//! *different* actor's monitor that happens to share the slot index.

use crate::channel::{channel, Receiver, Sender};
use crate::pid::Pid;
use crate::scheduler::with_runtime;

/// Why a monitored actor went down.
///
/// `Copy` because it carries no payload — see the module docs for why the
/// panic payload is *not* included here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DownReason {
    /// The target returned normally.
    Exit,
    /// The target panicked. The payload is delivered to the actor's joiner,
    /// not to monitors.
    Panic,
    /// The target was cooperatively cancelled via `request_stop`.
    Stopped,
    /// The target was already gone (finished and reclaimed, or never alive)
    /// at the moment `monitor()` was called.
    NoProc,
}

/// A monitored actor's termination notice.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Down {
    /// The pid that was being monitored.
    pub pid: Pid,
    /// How it went down.
    pub reason: DownReason,
}

/// A process-unique identifier for one `monitor()` registration.
///
/// Opaque and `Copy`. Allocated from a monotonic counter in shared state, so
/// it is never reused for the lifetime of the runtime — distinct `monitor()`
/// calls on the same target get distinct ids, which is what lets [`demonitor`]
/// tear down exactly one of several monitors on a target.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct MonitorId(pub(crate) u64);

/// A live monitor: the receiving end of the one-shot [`Down`] channel, plus the
/// identity needed to [`demonitor`] it.
///
/// Read the notification from [`Monitor::rx`]. Not `Clone` (the receiver is a
/// single consumer). Dropping it closes the receiving end; if a `Down` was
/// already queued it is discarded with the channel.
pub struct Monitor {
    /// This registration's process-unique id.
    pub id: MonitorId,
    /// The pid being monitored.
    pub target: Pid,
    /// The one-shot channel the `Down` arrives on.
    pub rx: Receiver<Down>,
}

/// Monitor `target`. Returns a [`Monitor`] whose `rx` receives exactly one
/// [`Down`].
///
/// If `target` is still live, the `Down` arrives when it terminates. If
/// `target` is already gone, a [`DownReason::NoProc`] `Down` is queued
/// immediately so the caller's `rx.recv()` returns without parking.
pub fn monitor(target: Pid) -> Monitor {
    let (tx, rx) = channel::<Down>();

    // Register under the target's cold lock. `tx.clone()` takes the channel's
    // own lock — a Channel-class RawMutex, explicitly permitted *under* a Leaf
    // (cold) lock by the lock order (see raw_mutex.rs). We must still not
    // *send* under the lock, as `Sender::send` can unpark a parked receiver,
    // and there's no reason to nest that.
    let (id, registered) = with_runtime(|inner| {
        let id = inner.alloc_monitor_id();
        let registered = match inner.slot_at(target) {
            Some(slot) => {
                let mut cold = slot.cold.lock();
                if slot.is_live_for(target) {
                    cold.monitors.push((id, tx.clone()));
                    true
                } else {
                    false
                }
            }
            None => false,
        };
        (id, registered)
    });

    if !registered {
        let _ = tx.send(Down { pid: target, reason: DownReason::NoProc });
    }

    Monitor { id, target, rx }
}

/// Cancel the monitor `m`. Returns `Some(id)` if a live registration was found
/// on the target's slot and removed, or `None` if there was nothing to remove
/// — the target already fired its `Down` (the registration is drained on
/// finalize), was never alive (`NoProc`), or has been reclaimed.
///
/// This stops any *future* `Down`. To also discard a `Down` the target may have
/// *already* queued (the finalize-races-demonitor case), drop `m` afterwards;
/// dropping the [`Monitor`] closes its receiver and the queued notice goes with
/// it — the analogue of Erlang's `demonitor(Ref, [flush])`.
pub fn demonitor(m: &Monitor) -> Option<MonitorId> {
    // Remove the registration under the target's cold lock, but move the
    // `Sender` *out* and let it drop only after the lock is released:
    // dropping the last sender runs `Sender::drop`, which may unpark a parked
    // receiver — legal under a cold lock, but pointless to nest.
    let removed: Option<(MonitorId, Sender<Down>)> = with_runtime(|inner| {
        let slot = inner.slot_at(m.target)?;
        let mut cold = slot.cold.lock();
        if slot.generation() != m.target.generation() {
            return None; // slot reused; the Down already fired
        }
        let pos = cold.monitors.iter().position(|(mid, _)| *mid == m.id)?;
        Some(cold.monitors.remove(pos))
    });
    // `removed`'s sender drops here, outside the lock.
    removed.map(|(id, _sender)| id)
}