//! Give an actor a name so other actors can find it and message it.
//!
//! Without the registry, the only way to reach an actor is to already be
//! holding its [`Pid`], usually because you spawned it yourself or someone
//! passed it to you. That is fine for a worker you just created, but it does
//! not work for a well-known service that arbitrary parts of your program
//! need to find independently, like a logger, a config store, or a
//! connection pool. The registry solves this: an actor claims a name once,
//! and from then on any other actor can look that name up, or send to it
//! directly, without ever having been handed a `Pid`.
//!
//! ```
//! use smarm::{channel, register, run, send, spawn, unregister, whereis, Name};
//!
//! const COUNTER: Name<u64> = Name::new("counter");
//!
//! run(|| {
//!     let (ready_tx, ready_rx) = channel::<()>();
//!     let (tx, rx) = channel::<u64>();
//!
//!     let worker = spawn(move || {
//!         // Claim the name for this actor's inbox. Any actor holding
//!         // `COUNTER` can now reach this one by name.
//!         register(COUNTER, tx).unwrap();
//!         ready_tx.send(()).unwrap();
//!         assert_eq!(rx.recv().unwrap(), 42);
//!     });
//!
//!     ready_rx.recv().unwrap(); // wait for the worker to register
//!
//!     // Look the name up, or just send to it directly.
//!     assert_eq!(whereis("counter"), Some(worker.pid()));
//!     send(COUNTER, 42).unwrap();
//!
//!     worker.join().unwrap();
//!
//!     // The name dies with the actor: nobody holds it anymore.
//!     assert_eq!(whereis("counter"), None);
//! });
//! ```
//!
//! ## Names carry a message type
//!
//! A [`Name<M>`] is a plain string plus a type parameter `M`: the message
//! type that name expects to receive. [`Name::new`] is `const`, so the usual
//! pattern is a module-level constant like `COUNTER` above, shared by every
//! caller. The type parameter means a name is only ever sent the kind of
//! message it was declared for. If two different constants share the same
//! string but have different message types, they still address two
//! independent channels on the same actor: registering both just gives that
//! actor two ways to be reached, one per message type. This is how you give
//! one actor a "public" channel and a separate, differently-typed "admin"
//! channel under related names, without inventing an enum to merge them.
//!
//! ## One actor per name, looked up fresh every time
//!
//! A name always points at exactly one actor at a time (contrast a *process
//! group*, from the [`pg`](crate::pg) module, which is one name mapping to
//! many actors). Unlike a plain [`Pid`], which names one specific actor
//! forever and stops working the moment that actor dies, a name is
//! re-resolved on every [`send`]: if the actor holding it dies and a new one
//! registers under the same name, the next `send` reaches the new holder
//! automatically. Use a name for a long-lived service whose exact identity
//! you do not want to track by hand; use a `Pid` when you already have one
//! and want to talk to that exact actor.
//!
//! ## Registration ends when the actor does
//!
//! There is no separate step to clean up a name when its actor exits: dying
//! is enough. The next operation that touches a dead binding (a [`whereis`],
//! a [`send`], or another actor's [`register`] of the same name) notices the
//! actor is gone and clears the stale entry as a side effect, so the name
//! becomes free again. [`unregister`] is only for a live actor voluntarily
//! giving up a name it no longer wants; nothing has to call it on the way
//! out.
//!
//! ## Implementation notes
//!
//! These details matter if you are working on smarm itself; they are not
//! part of the public contract.
//!
//! Internally, each live actor that has published at least one channel owns
//! a `Mailbox`: its pid plus a set of typed channels, keyed by the message
//! type's `TypeId`. A stored channel is a `Box<dyn Any + Send>` that
//! is concretely a `Sender<M>`; resolving for `M` looks up that exact
//! `TypeId` and downcasts, so the downcast cannot fail on correct data (a
//! failure would be a bug in the registry itself, checked in debug builds).
//! Registering a name therefore means: find or create the actor's mailbox,
//! insert the channel under its type, and point the name at the actor's pid.
//!
//! There is no callback when an actor exits. Every operation that touches a
//! binding checks the target pid's liveness directly against the scheduler's
//! slot table (which also tracks a generation counter, so a dead actor's
//! reused slot index is never mistaken for the same actor). A binding to a
//! dead actor is treated as absent and dropped right there. This keeps the
//! registry decoupled from actor teardown, at the cost of a dead binding
//! lingering until something happens to look at it.
//!
//! The whole registry (both the name index and the per-actor mailboxes) sits
//! behind one lock, which is what lets a name-addressed [`send`] resolve and
//! clone the target's sender in a single critical section. The sender is
//! cloned while that lock is held, then the lock is released before the
//! actual send, since delivering a message can wake a parked receiver and
//! that wakeup work should not run while the registry is locked.

use crate::channel::Sender;
use crate::pid::{Addressable, Name, Pid};
use crate::scheduler::{self_pid, with_runtime};
use std::any::{type_name, Any, TypeId};
use std::collections::HashMap;

/// Why a [`register`] call was rejected.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RegisterError {
    /// The name is bound to a different, still-live actor.
    NameTaken { holder: Pid },
    /// The caller is not a live actor (cannot happen for `self`, kept for
    /// symmetry / future explicit-pid registration).
    NoProc,
}

impl std::fmt::Display for RegisterError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RegisterError::NameTaken { holder } => {
                write!(f, "name is already registered to live actor {holder}")
            }
            RegisterError::NoProc => write!(f, "caller is not a live actor"),
        }
    }
}

impl std::error::Error for RegisterError {}

/// Why a send did not deliver. Every variant carries the undelivered message
/// back, mirroring [`crate::channel::SendError`], so a failed send never
/// silently drops what you tried to send.
///
/// `Debug` and `Display` are hand-written so neither requires `M: Debug`,
/// since the payload is handed back to you, not printed.
pub enum SendError<M> {
    /// No live actor is currently registered under this name. Returned only
    /// by name-addressed [`send`]; the pid-addressed counterpart of "nothing
    /// there" is [`SendError::Dead`].
    Unresolved(M),
    /// The actor this pid identifies has died, even if its slot has since
    /// been taken over by a different, live actor. A direct `Pid<A>` send
    /// never redirects to that new occupant; contrast name-addressed
    /// [`send`], which would reach it. Returned by the pid-addressed sends,
    /// [`send_to`] and [`send_dyn`].
    Dead(M),
    /// The actor is live but has not published a channel for this message
    /// type.
    NoChannel(M),
    /// The actor's channel for this message type is closed (its receiver has
    /// been dropped).
    Closed(M),
    /// No live member was available to deliver to: returned by
    /// [`dispatch`](crate::dispatch) when the target process group is empty
    /// or every member in it has died. The name-addressed counterpart of
    /// this case is [`SendError::Unresolved`].
    NoMember(M),
}

impl<M> SendError<M> {
    /// Recover the undelivered message.
    pub fn into_inner(self) -> M {
        match self {
            SendError::Unresolved(m)
            | SendError::Dead(m)
            | SendError::NoChannel(m)
            | SendError::Closed(m)
            | SendError::NoMember(m) => m,
        }
    }

    fn variant(&self) -> &'static str {
        match self {
            SendError::Unresolved(_) => "Unresolved",
            SendError::Dead(_) => "Dead",
            SendError::NoChannel(_) => "NoChannel",
            SendError::Closed(_) => "Closed",
            SendError::NoMember(_) => "NoMember",
        }
    }
}

impl<M> std::fmt::Debug for SendError<M> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "SendError::{}", self.variant())
    }
}

impl<M> std::fmt::Display for SendError<M> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SendError::Unresolved(_) => write!(f, "no live actor registered under that name"),
            SendError::Dead(_) => write!(f, "the addressed actor is no longer the live incarnation"),
            SendError::NoChannel(_) => write!(f, "actor has no channel for this message type"),
            SendError::Closed(_) => write!(f, "the actor's channel for this type is closed"),
            SendError::NoMember(_) => write!(f, "no live member in the process group"),
        }
    }
}

impl<M> std::error::Error for SendError<M> {}

/// A registry-stored channel, type-erased over its message type. The stored
/// object must serve two readers: `clone_sender` (downcast back to the concrete
/// `Sender<M>`) and the runtime introspection snapshot (queued length without
/// knowing `M`). A bare `Box<dyn Any>` gives the first but not the second, so
/// we erase behind this small trait instead.
trait ErasedSender: Send {
    fn as_any(&self) -> &dyn Any;
    fn queued_len(&self) -> usize;
}

impl<M: Send + 'static> ErasedSender for Sender<M> {
    fn as_any(&self) -> &dyn Any {
        self
    }
    fn queued_len(&self) -> usize {
        Sender::queued_len(self)
    }
}

/// One typed channel of an actor, type-erased. Concretely a `Sender<M>` filed
/// under `TypeId::of::<M>()`; `msg_type` is `type_name::<M>()`, kept for
/// observability tooling and as the debug cross-check on the downcast.
struct Channel {
    sender: Box<dyn ErasedSender>,
    msg_type: &'static str,
}

/// An actor's messageable surface: its identity plus every typed channel it has
/// published, keyed by message [`TypeId`]. Stored once per live actor; reached
/// by pid (directly) or by any name pointing at that pid.
struct Mailbox {
    pid: Pid,
    channels: HashMap<TypeId, Channel>,
}

impl Mailbox {
    fn new(pid: Pid) -> Self {
        Self { pid, channels: HashMap::new() }
    }

    /// Clone the `Sender<M>` for this actor, if it has one. Called **under the
    /// registry Leaf lock**: `Sender::clone` takes a Channel lock, which is
    /// legal under a Leaf (Leaf -> Channel).
    fn clone_sender<M: Send + 'static>(&self) -> Option<Sender<M>> {
        let ch = self.channels.get(&TypeId::of::<M>())?;
        let tx = match ch.sender.as_any().downcast_ref::<Sender<M>>() {
            Some(tx) => tx,
            None => panic!(
                "smarm: channel keyed by TypeId but downcast to its own type failed (core corrupt)"
            ),
        };
        debug_assert_eq!(ch.msg_type, type_name::<M>(), "msg_type / TypeId disagree");
        Some(tx.clone())
    }
}

/// Per-actor registry view handed to runtime introspection: registered names
/// and summed mailbox depth, tagged with the mailbox's `pid` so a stale
/// incarnation can be filtered against the slab. Covers only *published*
/// channels (`register` / `install` / `spawn_addr` / gen_server start); an
/// actor that holds only a private `channel()` receiver is invisible here and
/// reports depth 0.
pub(crate) struct MailboxInfo {
    pub(crate) pid: Pid,
    pub(crate) names: Vec<&'static str>,
    pub(crate) depth: u32,
}

/// The directory. Invariant (held under the registry lock): every value in
/// `by_name` is the full [`Pid`] (index *and* generation) of an actor that
/// published a [`Mailbox`] into `by_index` at registration time. Stale entries
/// (dead holders, including holders whose slot has since been re-tenanted by
/// a different actor) violate nothing: they are pruned on contact, and the
/// generation makes "dead" decidable even after slot reuse.
pub(crate) struct Registry {
    /// `pid.index() -> the actor's mailbox`. The handle store.
    by_index: HashMap<u32, Mailbox>,
    /// `name -> holder pid`. Several names may map to one actor. The full pid
    /// (not just the index) is load-bearing: an index alone cannot tell a dead
    /// holder from the live actor now tenanting its recycled slot. Comparing
    /// only the index would make such a name read as live-held (unresolvable
    /// and unregisterable at once) and could misdeliver to whatever new,
    /// same-typed actor now sits in that slot.
    by_name: HashMap<&'static str, Pid>,
}

impl Registry {
    pub(crate) fn new() -> Self {
        Self { by_index: HashMap::new(), by_name: HashMap::new() }
    }

    /// Drop a dead holder's artifacts: every name bound to it, and its
    /// mailbox, but only while the mailbox is still *its own*. A recycled
    /// slot's mailbox belongs to the live tenant (publish replaces it
    /// wholesale on pid mismatch) and is left untouched.
    fn prune_holder(&mut self, holder: Pid) {
        self.by_name.retain(|_, p| *p != holder);
        if self.by_index.get(&holder.index()).is_some_and(|mb| mb.pid == holder) {
            self.by_index.remove(&holder.index());
        }
    }

    /// Runtime introspection input: per-slot-index registry view, giving the
    /// actor's registered names (inverted from `by_name`) and its mailbox
    /// depth (queued messages summed across every published typed channel).
    /// Carries each mailbox's full `pid` so the caller can discard a stale
    /// incarnation's entry against the slab's live generation. Names are
    /// matched to mailboxes by *full pid*, so a stale name (dead holder)
    /// still annotates the corpse's own mailbox if that survives, but never a
    /// recycled slot's new tenant; names that attach to no mailbox are
    /// dropped, since that violates no invariant and they get pruned on next
    /// contact.
    pub(crate) fn introspect_map(&self) -> HashMap<u32, MailboxInfo> {
        let mut names: HashMap<Pid, Vec<&'static str>> = HashMap::new();
        for (&name, &pid) in &self.by_name {
            names.entry(pid).or_default().push(name);
        }
        let mut out: HashMap<u32, MailboxInfo> = HashMap::with_capacity(self.by_index.len());
        for (&idx, mb) in &self.by_index {
            let depth: usize = mb.channels.values().map(|c| c.sender.queued_len()).sum();
            out.insert(
                idx,
                MailboxInfo {
                    pid: mb.pid,
                    names: names.remove(&mb.pid).unwrap_or_default(),
                    depth: depth.min(u32::MAX as usize) as u32,
                },
            );
        }
        out
    }

    /// Single-actor form of [`introspect_map`](Self::introspect_map): the
    /// registry view for one slot index, or `None` if no mailbox is published
    /// there. Used by the runtime's per-actor introspection so its cost stays
    /// proportional to the one actor rather than locking every channel in the
    /// runtime.
    pub(crate) fn introspect_one(&self, idx: u32) -> Option<MailboxInfo> {
        let mb = self.by_index.get(&idx)?;
        let depth: usize = mb.channels.values().map(|c| c.sender.queued_len()).sum();
        let names = self
            .by_name
            .iter()
            .filter_map(|(&n, &p)| (p == mb.pid).then_some(n))
            .collect();
        Some(MailboxInfo { pid: mb.pid, names, depth: depth.min(u32::MAX as usize) as u32 })
    }
}

/// Is `pid` a live actor right now? Atomic slot-word read; no lock.
fn live(inner: &crate::runtime::RuntimeInner, pid: Pid) -> bool {
    inner.slot_at(pid).is_some_and(|s| s.is_live_for(pid))
}

/// Give the current actor's channel a name, so other actors can find and
/// message it by that name instead of needing its [`Pid`].
///
/// Calling this again with the same `(name, type)` from the same actor is
/// harmless. Registering a *second* message type under the same (or a
/// different) name from the same actor just adds another typed channel to
/// that actor's mailbox; it does not replace the first.
///
/// Fails with [`RegisterError::NameTaken`] if the name is currently held by a
/// *different* live actor. A name held by an actor that has since died is not
/// considered taken: it is quietly reclaimed and handed to you. Panics if
/// called outside [`run`](crate::run).
pub fn register<M: Send + 'static>(name: Name<M>, tx: Sender<M>) -> Result<(), RegisterError> {
    register_with(self_pid(), name.as_str(), tx)
}

/// Bind `name` to `pid`'s mailbox and publish `tx` under `M`'s [`TypeId`], for
/// an explicit (already-live) actor rather than `self`. The shared core of
/// [`register`] (which passes `self_pid()`) and the parent-side server-name
/// bind in `gen_server`, which names a freshly spawned server before its body
/// has run, so the name resolves the instant `start()` returns. Same collision
/// rules and lock discipline as `register`.
pub(crate) fn register_with<M: Send + 'static>(
    me: Pid,
    key: &'static str,
    tx: Sender<M>,
) -> Result<(), RegisterError> {
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        if !live(inner, me) {
            return Err(RegisterError::NoProc);
        }
        if let Some(&holder) = reg.by_name.get(key) {
            if holder == me {
                // Same actor: just add the channel below.
            } else if live(inner, holder) {
                return Err(RegisterError::NameTaken { holder });
            } else {
                // Dead holder: free the name (and its other stale artifacts).
                // Liveness is judged against the *stored* pid, generation
                // included, so a recycled slot's live tenant no longer makes a
                // dead name read as taken.
                reg.prune_holder(holder);
            }
        }
        // Publish (or extend) the mailbox with this channel, then bind the name.
        publish_channel::<M>(&mut reg, me, tx);
        reg.by_name.insert(key, me);
        Ok(())
    })
}

/// Insert or extend the current actor's mailbox with one typed channel, filed
/// under its message [`TypeId`]. Shared by [`register`] (which then binds a
/// name) and [`install`] (which does not). A leftover mailbox at this slot
/// index from a dead prior incarnation (pid mismatch) is replaced wholesale.
/// Caller holds the registry lock and has established that `me` is live.
fn publish_channel<M: Send + 'static>(reg: &mut Registry, me: Pid, tx: Sender<M>) {
    let mb = reg.by_index.entry(me.index()).or_insert_with(|| Mailbox::new(me));
    if mb.pid != me {
        *mb = Mailbox::new(me);
    }
    mb.channels.insert(
        TypeId::of::<M>(),
        Channel { sender: Box::new(tx), msg_type: type_name::<M>() },
    );
}

/// Publish the current actor's `Sender<A::Msg>` into its mailbox **without**
/// binding a name, and hand back the typed [`Pid<A>`] that addresses this
/// actor directly.
///
/// This is for an actor that wants to be reachable directly by its pid,
/// rather than only through a re-resolving [`Name`]: call this once with your
/// inbox sender, then hand the returned `Pid<A>` to whoever should be able to
/// message you. Unlike [`register`] there is no name to collide on, and the
/// current actor is always live while inside `run()`, so this cannot fail.
/// Panics if called outside [`run`](crate::run).
pub fn install<A: Addressable>(tx: Sender<A::Msg>) -> Pid<A> {
    let me = self_pid();
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        debug_assert!(live(inner, me), "self_pid() is a live actor inside run()");
        publish_channel::<A::Msg>(&mut reg, me, tx);
    });
    // `me` is this actor; re-type the identity as `Pid<A>` (the channel for
    // `A::Msg` was just published, so the typed address is now messageable).
    Pid::from_raw(me.raw())
}

/// Publish `tx` into `pid`'s mailbox under `M`'s [`TypeId`], for an explicit
/// (freshly minted, already-live) actor rather than `self`. The parent-side
/// half of [`spawn_addr`](crate::spawn_addr): the spawner makes the inbox and
/// publishes the sender here *before* handing back the `Pid<A>`, so an
/// immediate `send_to` on the returned pid always resolves. The address is
/// live the instant the caller holds it, with no dependence on the spawned
/// actor's body having run yet.
///
/// Caller guarantees `pid` is the just-installed actor (queued, this exact
/// incarnation); `publish_channel` replaces any stale leftover at the slot.
pub(crate) fn install_for<M: Send + 'static>(pid: Pid, tx: Sender<M>) {
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        debug_assert!(live(inner, pid), "install_for: pid must be a freshly spawned, live actor");
        publish_channel::<M>(&mut reg, pid, tx);
    });
}

/// Look up which actor currently holds `name`, if any. Returns `None` if the
/// name is unbound, or if it was bound to an actor that has since died (the
/// stale binding is cleared as a side effect of this call).
pub fn whereis(name: &str) -> Option<Pid> {
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        let pid = *reg.by_name.get(name)?;
        if live(inner, pid) {
            Some(pid)
        } else {
            // Generation-checked against the stored holder: a recycled slot's
            // live tenant reads dead here, and the stale name heals.
            reg.prune_holder(pid);
            None
        }
    })
}

/// What a name is bound to, three-valued (bridge soak signature 4).
///
/// [`Live`](NameResolution::Live) is [`whereis`]'s `Some`.
/// [`Corpse`](NameResolution::Corpse) carries the *stored* holder pid of a
/// dead-but-unpruned binding — a state Erlang cannot represent (its name
/// death unregisters atomically; smarm's prune is lazy), captured here before
/// the prune that `whereis` performs discards it, so the caller can consult
/// [`terminal_reason`](crate::monitor::terminal_reason) for the tenancy's
/// real down reason. [`Unbound`](NameResolution::Unbound) matches Erlang's
/// unregistered name.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NameResolution {
    /// The stored holder is live (generation-checked); the binding stands.
    Live(Pid),
    /// The stored holder is dead. The binding was pruned on the way out —
    /// the name heals exactly as `whereis` heals it; only the evidence is
    /// returned instead of discarded. A second resolve is `Unbound`.
    Corpse(Pid),
    /// No binding stored (never registered, or already pruned by any reader).
    Unbound,
}

/// Resolve `name` like [`whereis`], but keep the corpse: the dead-holder arm
/// returns the stored pid it pruned instead of a bare `None`. Same lock
/// discipline and pruning behavior as `whereis`; same `Runtime::run()`
/// context contract.
pub fn resolve_name(name: &str) -> NameResolution {
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        let Some(&pid) = reg.by_name.get(name) else {
            return NameResolution::Unbound;
        };
        if live(inner, pid) {
            NameResolution::Live(pid)
        } else {
            reg.prune_holder(pid);
            NameResolution::Corpse(pid)
        }
    })
}

/// Like [`whereis`], but returns a *typed* [`Pid<A>`] instead of a bare
/// [`Pid`], so a follow-up [`send_to`] is compile-checked instead of needing
/// the untyped [`send_dyn`] escape hatch. `None` if the name is unbound or its
/// holder has died.
///
/// The type `A` is not checked against what the name's holder actually
/// published: if you pick the wrong `A`, this still succeeds, but the next
/// send against the returned pid degrades to [`SendError::NoChannel`] rather
/// than reaching the wrong actor or the wrong channel.
///
/// Panics if called outside [`run`](crate::run).
pub fn lookup_as<A: Addressable>(name: &str) -> Option<Pid<A>> {
    whereis(name).map(crate::pid::assert_type::<A>)
}

/// Resolve `name` to its actor's pid and a cloned `Sender<M>`, all under one
/// lock acquisition. The crate-internal building block for `gen_server`'s
/// by-name addressing: a named server publishes its inbox as a
/// `Sender<Envelope<G>>` (via [`register_with`]), and the server's `call` /
/// `cast` / `whereis_server` recover that exact typed sender here to rebuild a
/// `GenServerRef<G>`. `None` if unbound, dead (pruned on the way out), or
/// holding no `M` channel.
pub(crate) fn resolve_named_sender<M: Send + 'static>(name: &str) -> Option<(Pid, Sender<M>)> {
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        let pid = *reg.by_name.get(name)?;
        if !live(inner, pid) {
            // Stored-pid liveness, generation included: a name whose holder
            // died is pruned (heals) even if the slot has a new tenant.
            // Otherwise the tenant's mailbox would make the name unresolvable
            // without pruning, wedging it for the tenant's lifetime.
            reg.prune_holder(pid);
            return None;
        }
        // A live holder's mailbox is its own (publish replaces wholesale on
        // pid mismatch, and one live actor per slot), so index lookup is safe.
        let tx = reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>)?;
        Some((pid, tx))
    })
}

/// Give up a name. Returns the actor it pointed at, if that actor was still
/// live. Only the *name* is freed; the actor's mailbox (and any other names
/// bound to it) are unaffected. A binding to an already-dead actor reports
/// `None`, since there was nothing live to release.
pub fn unregister(name: &str) -> Option<Pid> {
    with_runtime(|inner| {
        let mut reg = inner.registry.lock();
        let pid = reg.by_name.remove(name)?;
        if live(inner, pid) { Some(pid) } else { None }
    })
}

/// Look `name` up and deliver `msg` to whichever actor currently holds it.
/// This is the point of naming an actor: a name you can send a message to
/// directly, without a separate lookup step.
///
/// On failure the message comes back to you, wrapped in the [`SendError`]
/// variant that explains why: [`SendError::Unresolved`] if no live actor
/// currently holds the name, [`SendError::NoChannel`] if the actor that holds
/// it never published a channel for `M`, or [`SendError::Closed`] if it
/// published one but has since dropped the receiving end. Panics if called
/// outside [`run`](crate::run).
pub fn send<M: Send + 'static>(name: Name<M>, msg: M) -> Result<(), SendError<M>> {
    let key = name.as_str();
    with_runtime(|inner| {
        // Resolve + clone the sender under the registry lock, then drop the
        // lock before sending (a send can unpark a receiver).
        let tx = {
            let mut reg = inner.registry.lock();
            let pid = match reg.by_name.get(key) {
                Some(&p) => p,
                None => return Err(SendError::Unresolved(msg)),
            };
            if !live(inner, pid) {
                // Stored-pid liveness (generation included), so a recycled
                // slot's new live tenant is never mistaken for the name's
                // original (now-dead) holder.
                reg.prune_holder(pid);
                return Err(SendError::Unresolved(msg));
            }
            match reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>) {
                Some(tx) => tx,
                None => return Err(SendError::NoChannel(msg)),
            }
        };
        tx.send(msg).map_err(|crate::channel::SendError(m)| SendError::Closed(m))
    })
}

/// Resolve a *raw* pid to its mailbox and deliver `msg` on the channel for `M`,
/// with **no redirect**. The stored mailbox must be this exact incarnation
/// (generation included) and still live; otherwise the actor this pid named
/// is gone and the result is [`SendError::Dead`], even when the slot now
/// holds a different, live actor (which is left untouched). Shared by
/// [`send_to`] (typed, `M = A::Msg`, channel guaranteed on an installed
/// actor) and [`send_dyn`] (explicit `M`, where `NoChannel` is a real
/// outcome).
fn send_to_pid<M: Send + 'static>(
    inner: &crate::runtime::RuntimeInner,
    pid: Pid,
    msg: M,
) -> Result<(), SendError<M>> {
    // Resolve + clone the sender under the registry lock, then drop the lock
    // before sending (a send can unpark a receiver), same order as `send`.
    let tx = {
        let mut reg = inner.registry.lock();
        match reg.by_index.get(&pid.index()).map(|m| m.pid) {
            // Exact incarnation, still alive: its `M` channel, or NoChannel.
            Some(stored) if stored == pid && live(inner, pid) => {
                match reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>) {
                    Some(tx) => tx,
                    None => return Err(SendError::NoChannel(msg)),
                }
            }
            // Our incarnation's mailbox, but the actor has died: prune + Dead.
            Some(stored) if stored == pid => {
                reg.prune_holder(pid);
                return Err(SendError::Dead(msg));
            }
            // A different incarnation (or nothing) occupies the slot: the actor
            // this pid named is gone. Do not disturb any newer occupant.
            _ => return Err(SendError::Dead(msg)),
        }
    };
    tx.send(msg).map_err(|crate::channel::SendError(m)| SendError::Closed(m))
}

/// Deliver `msg` directly to the exact actor identified by `pid`. Unlike
/// name-addressed [`send`], there is **no redirect**: if that specific actor
/// has died, the message comes back as [`SendError::Dead`], even if its slot
/// has since been taken over by a different, live actor. Use this when you
/// already hold a `Pid<A>` and want to talk to that one actor specifically;
/// use [`send`] with a [`Name`] when you want whichever actor currently holds
/// a name.
///
/// The message type is the actor's `A::Msg`, so on a live actor that has
/// installed its inbox (via [`install`] or [`register`]) the channel is
/// always present; [`SendError::NoChannel`] therefore means the actor is live
/// but never published a `Pid<A>`-reachable inbox. Panics if called outside
/// [`run`](crate::run).
pub fn send_to<A: Addressable>(pid: Pid<A>, msg: A::Msg) -> Result<(), SendError<A::Msg>> {
    with_runtime(|inner| send_to_pid::<A::Msg>(inner, pid.erase(), msg))
}

/// The escape hatch for sending to a bare, untyped [`Pid`] when the typed
/// [`send_to`] is unavailable, for example a pid recovered from a [`Down`]
/// notification or a group's `members()` list, where you no longer know the
/// actor's message type at compile time.
///
/// Because the message type is not checked at compile time here, this is the
/// one send that can genuinely be live-but-wrong: the actor may be alive yet
/// expose no channel for `M`, in which case you get [`SendError::NoChannel`]
/// back instead of a misdelivery. Liveness and redirect behavior are
/// otherwise identical to [`send_to`]: identity-bound, no redirect,
/// [`SendError::Dead`] once the addressed incarnation is gone. Prefer
/// `send_to` with a typed `Pid<A>` whenever you have one; reach for this only
/// when you don't. Panics if called outside [`run`](crate::run).
///
/// [`Down`]: crate::Down
pub fn send_dyn<M: Send + 'static>(pid: Pid, msg: M) -> Result<(), SendError<M>> {
    with_runtime(|inner| send_to_pid::<M>(inner, pid, msg))
}