//! Process links + `trap_exit`.
//!
//! A *link* is bidirectional and persistent, in contrast to a [`monitor`]
//! (unidirectional, one-shot): once two actors are linked, the *abnormal*
//! death of either propagates to the other. "Abnormal" means a panic or a
//! cooperative [`request_stop`] — a normal return never propagates.
//!
//! What propagation *does* depends on whether the surviving peer traps exits:
//!
//!   - **not trapping** (the default): the peer is cooperatively
//!     [`request_stop`]'d, so a crash fate-shares across the whole link set —
//!     Erlang's "let it crash". Because the peer's own links are walked when
//!     *it* finalizes, the stop cascades transitively.
//!   - **trapping** (called [`trap_exit`]): the peer instead receives an
//!     [`ExitSignal`] *message* on its trap inbox and keeps running, turning
//!     peer failures into ordinary inbox events (Erlang's
//!     `process_flag(trap_exit, true)`).
//!
//! Linking an already-dead pid is not a silent no-op: it behaves exactly like
//! an immediate abnormal death of the target with reason [`DownReason::NoProc`]
//! — a trapping caller gets a message, a non-trapping caller is stopped.
//!
//! ## Payload
//!
//! [`ExitSignal`] reuses [`DownReason`] and, like a monitor, does **not**
//! carry the panic payload: a panic's payload has a single owner and is
//! delivered to whoever `join()`s the actor. A trap inbox only learns *that*
//! (and *how*) a peer died.
//!
//! ## Inbox
//!
//! The trap inbox is a dedicated channel, distinct from the monitor [`Down`]
//! channel: monitors are documented as one-shot/unidirectional, whereas a trap
//! inbox receives one [`ExitSignal`] per linked peer death over the actor's
//! lifetime. [`trap_exit`] enables trapping and hands back the inbox in one
//! call; calling it again installs a fresh inbox (the previous one closes).
//!
//! ## Races
//!
//! [`link`] registration and `finalize_actor` (in `runtime`) both serialize on
//! the shared mutex, so a target that is live when the link is registered is
//! guaranteed to propagate its eventual death; there is no window in which the
//! death slips between the liveness check and the registration. As elsewhere,
//! we never `send` or `request_stop` while holding the shared lock — those
//! re-enter the runtime — so the act is always performed after the lock is
//! released.
//!
//! [`monitor`]: crate::monitor::monitor
//! [`Down`]: crate::monitor::Down
//! [`request_stop`]: crate::scheduler::request_stop

use crate::channel::{channel, Receiver};
use crate::monitor::DownReason;
use crate::pid::Pid;
use crate::scheduler::{request_stop, self_pid, with_runtime};

/// A linked peer's death, delivered to a trapping actor's inbox.
///
/// `Copy` because it carries no payload — see the module docs for why the
/// panic payload is *not* included.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ExitSignal {
    /// The linked pid that died (or the dead pid that was linked).
    pub from: Pid,
    /// How it went down. Only ever `Panic`, `Stopped`, or `NoProc` — a normal
    /// `Exit` does not propagate, so it never appears here.
    pub reason: DownReason,
}

/// Trap exits on the current actor and return its exit inbox.
///
/// After this call, the abnormal death of a linked peer arrives here as an
/// [`ExitSignal`] message instead of cooperatively stopping this actor.
/// Calling it again installs a fresh inbox; the previously returned receiver
/// then closes.
pub fn trap_exit() -> Receiver<ExitSignal> {
    let (tx, rx) = channel::<ExitSignal>();
    let me = self_pid();
    with_runtime(|inner| {
        if let Some(slot) = inner.slot_at(me) {
            let mut cold = slot.cold.lock();
            // Own slot: generation is necessarily current (we're running).
            if let Some(actor) = cold.actor.as_mut() {
                actor.trap = Some(tx);
            }
        }
    });
    rx
}

/// Bidirectionally link the current actor to `target`.
///
/// If `target` is live, the link is recorded on both actors and either's
/// abnormal death will propagate to the other. If `target` is already gone,
/// this delivers an immediate [`DownReason::NoProc`] exit signal to the caller
/// (a message if trapping, otherwise a cooperative stop). Linking yourself, or
/// re-linking an existing peer, is a no-op.
pub fn link(target: Pid) {
    let me = self_pid();
    if target == me {
        return;
    }

    // Cold locks are leaves: never hold two at once. The link is recorded one
    // side at a time, TARGET FIRST — that ordering is what makes the race
    // window sound:
    //
    // - Once `me` is in `target.links`, the target's death always reaches us
    //   (its finalize cascade walks that list). So after step 1 succeeds, the
    //   link semantics are already live.
    // - If the target dies between step 1 and step 2, its cascade removes
    //   `target` from OUR links (a no-op, we haven't added it yet) and
    //   delivers the exit signal — correct, the link was established. Our
    //   subsequent step-2 insert leaves a stale `target` entry in `me.links`;
    //   stale entries are benign by construction (every cascade walk
    //   re-verifies the peer's word; `unlink` removes them like any other).
    //
    // The reverse order would be unsound: target dying in the window would
    // walk its links WITHOUT us — a silently dead link that we believe is live.
    let registered_on_target = with_runtime(|inner| match inner.slot_at(target) {
        Some(slot) => {
            let mut cold = slot.cold.lock();
            if slot.is_live_for(target) && cold.actor.is_some() {
                if !cold.links.contains(&me) {
                    cold.links.push(me);
                }
                true
            } else {
                false
            }
        }
        None => false,
    });

    if registered_on_target {
        with_runtime(|inner| {
            let slot = inner.slot_at(me).expect("link: own slot vanished");
            let mut cold = slot.cold.lock();
            if !cold.links.contains(&target) {
                cold.links.push(target);
            }
        });
        return;
    }

    // Target already gone: deliver NoProc to ourselves — as a message if
    // trapping, otherwise as a cooperative stop.
    let my_trap = with_runtime(|inner| {
        inner.slot_at(me).and_then(|slot| {
            let cold = slot.cold.lock();
            cold.actor.as_ref().and_then(|a| a.trap.clone())
        })
    });
    match my_trap {
        Some(tx) => {
            let _ = tx.send(ExitSignal { from: target, reason: DownReason::NoProc });
        }
        None => request_stop(me),
    }
}

/// Remove the link between the current actor and `target`, in both directions.
///
/// After this, neither actor's death propagates to the other. A no-op if the
/// two were not linked.
pub fn unlink(target: Pid) {
    let me = self_pid();
    if target == me {
        return;
    }
    with_runtime(|inner| {
        // One cold lock at a time (leaf rule). Order is immaterial here:
        // a half-removed link is just a stale entry on one side, and stale
        // entries are benign (re-verified on every cascade walk).
        if let Some(slot) = inner.slot_at(me) {
            let mut cold = slot.cold.lock();
            cold.links.retain(|p| *p != target);
        }
        if let Some(slot) = inner.slot_at(target) {
            let mut cold = slot.cold.lock();
            if slot.generation() == target.generation() {
                cold.links.retain(|p| *p != me);
            }
        }
    });
}