//! Process groups — a `name → multiset<Member>` map (RFC 012).
//!
//! Sits parallel to [`registry`](crate::registry), not on top of it. The
//! registry is a *bimap*: at most one pid per name. A group is the opposite —
//! many pids per name, the same pid in many groups — so it cannot be a
//! generalization of the bimap; it is its own keyspace sharing only the
//! liveness signal source (monitors) and the lock *class*.
//!
//! ## Cleanup is eager and monitor-driven (unlike the registry)
//!
//! The registry prunes a stale binding lazily, on contact, because it only
//! ever resolves one binding at a time. A group is *iterated* — `members` fans
//! out to every member — so it must not carry dead members across a broadcast.
//! Each [`join`] installs a `monitor(pid)` and keeps the resulting [`Monitor`]
//! *alongside the group entry*; the actor's one-shot `Down` lands in that
//! monitor's channel when it dies. Every group operation [`reap`s][reap] the
//! group it touches first — draining each membership's monitor with a
//! non-blocking `try_recv` — and on the first sign of death sweeps that pid out
//! of *every* group via the predicate primitive. So the set is self-pruning on
//! contact rather than filtered on read; the read path additionally applies a
//! generation-checked liveness backstop so a member already dead but not yet
//! reaped is never *returned*, even though eviction remains the monitor's job.
//!
//! [reap]: ProcessGroups::reap_group
//!
//! ## Eviction is one dumb primitive
//!
//! [`ProcessGroups::remove_where`] removes every member matching a predicate
//! from every group. The primitive never knows *why* a member leaves — that is
//! the caller's concern. Its first caller is the monitor death hook (this RFC,
//! via `reap_group`); the later `evict_incarnation(node, inc)` sweep (RFC 010)
//! reuses the same predicate path, which is the whole reason to shape it as a
//! predicate.
//!
//! ## Identity is cluster-shaped from the first commit
//!
//! A [`Member`] is not a bare [`Pid`]: it is `(NodeId, Incarnation, Pid)`, a
//! field-for-field image of a modern BEAM pid (`NEW_PID_EXT`) so the eventual
//! ETF codec (RFC 011) is a near-identity mapping. In this RFC `NodeId` and
//! `Incarnation` are runtime-init constants — one node, one fixed incarnation
//! — threaded through storage and eviction anyway so the public surface never
//! has to change to acquire them once clustering (RFC 010) supplies real
//! values. No cluster types leak out of the public functions: callers pass and
//! receive [`Pid`]; the node/incarnation are filled from runtime identity.
//!
//! ## Locking
//!
//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry`. The
//! group lock is never held with another Leaf (it never touches the registry
//! or a slot's cold lock). The two places that *do* need another lock are kept
//! off the group-lock path:
//!
//!   - `monitor()` / `demonitor()` acquire the target's cold lock (Leaf) and
//!     so run *before* / *after* the group lock, never under it.
//!   - draining a monitor with `try_recv` takes the channel's Channel-class
//!     lock — permitted *under* a Leaf by the lock order (`raw_mutex.rs`), and
//!     a channel critical section only does the lock-free unpark protocol, so
//!     no Leaf is ever nested under it.
//!
//! Evicted and rejected [`Monitor`]s are dropped only *after* the group lock is
//! released, so a receiver-drop never runs a wakeup under the lock (same
//! discipline as `demonitor`).

use crate::monitor::{demonitor, monitor, Monitor};
use crate::pid::Pid;
use crate::scheduler::with_runtime;
use std::collections::HashMap;

/// A cluster node handle. A `u32` integer handle, *not* an interned atom — the
/// single deliberate divergence from the BEAM wire shape (RFC 011 names it).
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
pub struct NodeId(u32);

impl NodeId {
    #[inline]
    pub const fn new(v: u32) -> Self {
        Self(v)
    }
    #[inline]
    pub const fn get(self) -> u32 {
        self.0
    }
}

impl From<u32> for NodeId {
    #[inline]
    fn from(v: u32) -> Self {
        Self(v)
    }
}

/// A node's incarnation epoch — the BEAM `Creation` field adopted verbatim: it
/// separates a crashed node from its restart. Fixed for the life of a run
/// until clustering supplies a real one.
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
pub struct Incarnation(u32);

impl Incarnation {
    #[inline]
    pub const fn new(v: u32) -> Self {
        Self(v)
    }
    #[inline]
    pub const fn get(self) -> u32 {
        self.0
    }
}

impl From<u32> for Incarnation {
    #[inline]
    fn from(v: u32) -> Self {
        Self(v)
    }
}

/// The fixed single-node identity used until clustering (RFC 010) supplies
/// real values. Carried like `wake_slot` so the public API never has to change
/// to acquire it.
pub const DEFAULT_NODE_ID: NodeId = NodeId(0);
/// The fixed incarnation for the single-node default. Non-zero so it never
/// collides with a BEAM "any creation" wildcard at interop time.
pub const DEFAULT_INCARNATION: Incarnation = Incarnation(1);

/// A group member's full identity: `(node, incarnation, pid)`.
///
/// Deliberately a field-for-field image of a modern BEAM pid (`NEW_PID_EXT`).
/// In memory it is a plain struct — no wire packing; the packed representation
/// belongs to the `RemoteRef` boundary (RFC 010/011), not here.
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
pub struct Member {
    /// Which node the pid lives on. `DEFAULT_NODE_ID` while single-node.
    pub node: NodeId,
    /// The node's incarnation epoch at the time of joining.
    pub incarnation: Incarnation,
    /// Pure local slot identity — unchanged; cluster identity is layered
    /// *around* it here rather than overloading `Pid::generation`.
    pub pid: Pid,
}

/// One membership: a [`Member`] and the [`Monitor`] that watches its liveness.
/// The monitor lives *alongside* the group entry (RFC 012) so a group is
/// self-contained: draining the membership tells us whether the member is
/// still alive, and dropping the membership drops its monitor.
struct Membership {
    member: Member,
    monitor: Monitor,
}

/// The store: `name → multiset<Member>`. Within a single group a `Member`
/// appears at most once (`join` is idempotent); the *multiset* framing is for
/// cluster-readiness — the same pid is freely a member of many groups, and the
/// width admits multiples in general. Held under one Leaf-class `RawMutex`.
pub(crate) struct ProcessGroups {
    groups: HashMap<String, Vec<Membership>>,
}

impl ProcessGroups {
    pub(crate) fn new() -> Self {
        Self { groups: HashMap::new() }
    }

    /// Insert `ms` into `group`. Idempotent on the *member*: if the member is
    /// already present the new membership is handed back (`Some`) so the caller
    /// can tear its now-redundant monitor down outside the lock; `None` means
    /// it was inserted.
    fn join(&mut self, group: &str, ms: Membership) -> Option<Membership> {
        let v = self.groups.entry(group.to_owned()).or_default();
        if v.iter().any(|e| e.member == ms.member) {
            return Some(ms);
        }
        v.push(ms);
        None
    }

    /// Remove `member`'s membership from `group`, returning it (so the caller
    /// can `demonitor` it outside the lock). An emptied group is pruned.
    fn leave(&mut self, group: &str, member: Member) -> Option<Membership> {
        let v = self.groups.get_mut(group)?;
        let pos = v.iter().position(|e| e.member == member)?;
        let removed = v.remove(pos);
        if v.is_empty() {
            self.groups.remove(group);
        }
        Some(removed)
    }

    /// The one dumb eviction primitive: drop every member matching `pred` from
    /// every group, pruning emptied groups, and return the evicted memberships'
    /// monitors for the caller to drop outside the lock. The primitive does not
    /// know *why* a member leaves. Callers: the death hook (`reap_group`) and,
    /// later, `evict_incarnation` (RFC 010), over the same path. Insertion
    /// order within a group is preserved (`members` / `pick` are order-stable).
    fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) -> Vec<Monitor> {
        let mut evicted = Vec::new();
        self.groups.retain(|_, v| {
            let mut i = 0;
            while i < v.len() {
                if pred(&v[i].member) {
                    evicted.push(v.remove(i).monitor);
                } else {
                    i += 1;
                }
            }
            !v.is_empty()
        });
        evicted
    }

    /// Drain-on-contact death hook. Drains every membership monitor in `group`
    /// with a non-blocking `try_recv`: a delivered `Down` (any reason) or a
    /// closed channel means that member is dead. On the first death detected,
    /// sweep *all* of the dead pids out of *every* group via [`remove_where`]
    /// — a death is removed from each group it joined, not just the one being
    /// touched. Returns the evicted monitors to drop outside the lock.
    fn reap_group(&mut self, group: &str) -> Vec<Monitor> {
        let dead: Vec<Pid> = {
            let Some(v) = self.groups.get(group) else {
                return Vec::new();
            };
            v.iter()
                .filter_map(|e| match e.monitor.rx.try_recv() {
                    // A Down arrived, or the channel closed and drained: dead.
                    Ok(Some(_)) | Err(_) => Some(e.member.pid),
                    // Empty but open — the sender still lives in the slot: alive.
                    Ok(None) => None,
                })
                .collect()
        };
        if dead.is_empty() {
            return Vec::new();
        }
        self.remove_where(|m| dead.contains(&m.pid))
    }

    /// Raw enumeration of a group's members — no liveness filtering. Used by
    /// tests to assert storage state independently of the read-path backstop.
    #[cfg(test)]
    fn members_of(&self, group: &str) -> Vec<Member> {
        self.groups
            .get(group)
            .map(|v| v.iter().map(|e| e.member).collect())
            .unwrap_or_default()
    }

    /// Live members of `group`, in insertion order. The `is_live` oracle is the
    /// read-path backstop (Phase 3): a member whose slot is already dead is
    /// dropped from the *result* even if its `Down` has not been drained yet.
    /// Backstop only — the entry stays in storage; eviction is the monitor's
    /// job (`reap_group`).
    fn members_where(&self, group: &str, mut is_live: impl FnMut(Pid) -> bool) -> Vec<Pid> {
        self.groups
            .get(group)
            .map(|v| v.iter().map(|e| e.member.pid).filter(|&p| is_live(p)).collect())
            .unwrap_or_default()
    }

    /// The first live member of `group` in insertion order — stateless
    /// first-live `pick`, with the same read-path backstop as `members_where`.
    fn first_member_where(&self, group: &str, mut is_live: impl FnMut(Pid) -> bool) -> Option<Pid> {
        self.groups.get(group)?.iter().map(|e| e.member.pid).find(|&p| is_live(p))
    }
}

/// Build the full member identity for `pid` from runtime identity.
fn member_for(inner: &crate::runtime::RuntimeInner, pid: Pid) -> Member {
    Member { node: inner.node_id, incarnation: inner.incarnation, pid }
}

/// Is `pid` a live actor right now? Generation-checked atomic slot-word read,
/// no lock — identical to the registry's guard. The read-path backstop: a
/// generation is never reused, so a dead member is detectable independently of
/// whether its monitor `Down` has been drained yet.
fn live(inner: &crate::runtime::RuntimeInner, pid: Pid) -> bool {
    inner.slot_at(pid).is_some_and(|s| s.is_live_for(pid))
}

/// Add `pid` to `group`. The same pid may join many groups; within one group a
/// pid is a member at most once (idempotent). Returns `true` if this call newly
/// added the membership, `false` if it was already a member.
///
/// Installs a `monitor(pid)` whose one-shot `Down` drives eviction: the
/// registration races `finalize_actor` under the slot's cold lock exactly as
/// every other monitor does, so no death slips between the join and the
/// registration. A redundant (idempotent) join tears its extra monitor back
/// down.
///
/// Panics if called outside `Runtime::run()`.
pub fn join(group: impl Into<String>, pid: Pid) -> bool {
    let group = group.into();
    // Install the monitor BEFORE taking the group lock: monitor() acquires the
    // target's cold lock (Leaf), and two Leaf locks are never held at once.
    let mon = monitor(pid);

    let (rejected, reaped) = with_runtime(|inner| {
        let ms = Membership { member: member_for(inner, pid), monitor: mon };
        let mut pg = inner.process_groups.lock();
        let reaped = pg.reap_group(&group);
        let rejected = pg.join(&group, ms);
        (rejected, reaped)
    });

    // Outside the group lock: drop the reaped (dead) monitors, and if this join
    // was redundant, demonitor + drop the extra monitor we just installed.
    drop(reaped);
    match rejected {
        Some(dup) => {
            demonitor(&dup.monitor);
            false
        }
        None => true,
    }
}

/// Drop `pid`'s membership of `group`. Returns whether a membership was
/// removed. The membership's monitor is demonitored and dropped.
///
/// Panics if called outside `Runtime::run()`.
pub fn leave(group: &str, pid: Pid) -> bool {
    let (removed, reaped) = with_runtime(|inner| {
        let member = member_for(inner, pid);
        let mut pg = inner.process_groups.lock();
        let reaped = pg.reap_group(group);
        let removed = pg.leave(group, member);
        (removed, reaped)
    });

    drop(reaped);
    match removed {
        Some(ms) => {
            demonitor(&ms.monitor);
            true
        }
        None => false,
    }
}

/// Fan-out read: every live member of `group`.
///
/// The touched group is reaped first, so dead members are evicted before the
/// read. The generation-checked liveness read is a belt-and-braces backstop:
/// even in the finalize window where a member is already dead but its `Down`
/// has not yet landed, it is dropped from the result.
///
/// Panics if called outside `Runtime::run()`.
pub fn members(group: &str) -> Vec<Pid> {
    let (pids, reaped) = with_runtime(|inner| {
        let mut pg = inner.process_groups.lock();
        let reaped = pg.reap_group(group);
        let pids = pg.members_where(group, |pid| live(inner, pid));
        (pids, reaped)
    });
    drop(reaped);
    pids
}

/// Pool/discovery read: one live member of `group`, or `None` if empty.
/// Stateless first-live selection: the touched group is reaped first, and the
/// generation-checked liveness backstop skips any member already dead but not
/// yet reaped. Smarter routing is an explicitly later, clustered concern per
/// RFC 010.
///
/// Panics if called outside `Runtime::run()`.
pub fn pick(group: &str) -> Option<Pid> {
    let (picked, reaped) = with_runtime(|inner| {
        let mut pg = inner.process_groups.lock();
        let reaped = pg.reap_group(group);
        let picked = pg.first_member_where(group, |pid| live(inner, pid));
        (picked, reaped)
    });
    drop(reaped);
    picked
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::channel::{channel, Sender};
    use crate::monitor::{Down, DownReason, MonitorId};

    fn member(index: u32, generation: u32) -> Member {
        Member {
            node: DEFAULT_NODE_ID,
            incarnation: DEFAULT_INCARNATION,
            pid: Pid::new(index, generation),
        }
    }

    /// A synthetic membership with a real (but slot-less) monitor channel. The
    /// returned `Sender` stands in for the slot's `Down` sender: hold it to
    /// keep the member "alive" (`try_recv` → `Ok(None)`), `send` a `Down` to
    /// simulate death, or `drop` it to simulate a drained/closed channel.
    fn synth(index: u32, generation: u32) -> (Membership, Sender<Down>) {
        let pid = Pid::new(index, generation);
        let (tx, rx) = channel::<Down>();
        let ms = Membership {
            member: member(index, generation),
            monitor: Monitor { id: MonitorId(0), target: pid, rx },
        };
        (ms, tx)
    }

    #[test]
    fn join_is_idempotent_within_a_group() {
        let mut pg = ProcessGroups::new();
        let (a, _ta) = synth(1, 0);
        let (b, _tb) = synth(1, 0);
        assert!(pg.join("workers", a).is_none(), "first join inserts");
        assert!(pg.join("workers", b).is_some(), "second identical join is handed back");
        assert_eq!(pg.members_of("workers"), vec![member(1, 0)]);
    }

    #[test]
    fn same_pid_in_many_groups_is_independent() {
        let mut pg = ProcessGroups::new();
        let (a, _ta) = synth(1, 0);
        let (b, _tb) = synth(1, 0);
        let (c, _tc) = synth(2, 0);
        pg.join("a", a);
        pg.join("b", b);
        pg.join("b", c);
        assert_eq!(pg.members_of("a"), vec![member(1, 0)]);
        assert_eq!(pg.members_of("b"), vec![member(1, 0), member(2, 0)]);
    }

    #[test]
    fn distinct_generations_are_distinct_members() {
        // ABA guard: same slot index, different generation = different actor.
        let mut pg = ProcessGroups::new();
        let (a, _ta) = synth(1, 0);
        let (b, _tb) = synth(1, 1);
        assert!(pg.join("g", a).is_none());
        assert!(pg.join("g", b).is_none(), "different generation is a distinct member");
        assert_eq!(pg.members_of("g"), vec![member(1, 0), member(1, 1)]);
    }

    #[test]
    fn leave_removes_one_membership_and_prunes_empty_groups() {
        let mut pg = ProcessGroups::new();
        let (a, _ta) = synth(1, 0);
        let (b, _tb) = synth(2, 0);
        pg.join("g", a);
        pg.join("g", b);
        assert!(pg.leave("g", member(1, 0)).is_some());
        assert_eq!(pg.members_of("g"), vec![member(2, 0)]);
        assert!(pg.leave("g", member(1, 0)).is_none(), "second leave finds nothing");
        assert!(pg.leave("g", member(2, 0)).is_some());
        assert!(pg.members_of("g").is_empty(), "group is now empty");
        assert!(pg.leave("never", member(9, 0)).is_none(), "leaving an unknown group is a no-op");
    }

    #[test]
    fn remove_where_sweeps_every_group() {
        let mut pg = ProcessGroups::new();
        for (g, (m, _t)) in [("a", synth(1, 0)), ("a", synth(2, 0)), ("b", synth(1, 0)), ("c", synth(3, 0))] {
            pg.join(g, m);
        }
        // Death of pid index 1 (any generation) evicts it everywhere.
        let evicted = pg.remove_where(|mem| mem.pid.index() == 1);
        assert_eq!(evicted.len(), 2, "pid 1 was in a and b");
        assert_eq!(pg.members_of("a"), vec![member(2, 0)]);
        assert!(pg.members_of("b").is_empty(), "b held only pid 1; pruned");
        assert_eq!(pg.members_of("c"), vec![member(3, 0)]);
    }

    #[test]
    fn remove_where_can_match_an_incarnation_sweep() {
        // Shape check for the later evict_incarnation(node, inc) caller.
        let mut pg = ProcessGroups::new();
        let pid = Pid::new(1, 0);
        let (tx, rx) = channel::<Down>();
        let dead = Membership {
            member: Member { node: DEFAULT_NODE_ID, incarnation: Incarnation::new(7), pid },
            monitor: Monitor { id: MonitorId(0), target: pid, rx },
        };
        let _keep = tx;
        let (live, _tl) = synth(2, 0);
        pg.join("g", dead);
        pg.join("g", live);
        let evicted = pg.remove_where(|mem| mem.incarnation == Incarnation::new(7));
        assert_eq!(evicted.len(), 1);
        assert_eq!(pg.members_of("g"), vec![member(2, 0)]);
    }

    #[test]
    fn reap_keeps_live_members() {
        let mut pg = ProcessGroups::new();
        let (a, _ta) = synth(1, 0); // sender held: member stays alive
        pg.join("a", a);
        assert!(pg.reap_group("a").is_empty(), "no deaths");
        assert_eq!(pg.members_of("a"), vec![member(1, 0)]);
    }

    #[test]
    fn reap_evicts_a_dead_member_and_sweeps_all_its_groups() {
        let mut pg = ProcessGroups::new();
        let (a1, ta1) = synth(1, 0); // pid 1 in group a
        let (a2, _ta2) = synth(2, 0); // pid 2 in group a (stays alive)
        let (b1, _tb1) = synth(1, 0); // pid 1 in group b
        pg.join("a", a1);
        pg.join("a", a2);
        pg.join("b", b1);
        // pid 1 dies: its group-a monitor receives a Down. Its group-b monitor
        // has not — reap must still sweep pid 1 out of b by the pid predicate.
        ta1.send(Down { pid: Pid::new(1, 0), reason: DownReason::Exit }).unwrap();
        let evicted = pg.reap_group("a");
        assert_eq!(evicted.len(), 2, "pid 1's memberships in both a and b are evicted");
        assert_eq!(pg.members_of("a"), vec![member(2, 0)]);
        assert!(pg.members_of("b").is_empty(), "swept from b too; pruned");
    }

    #[test]
    fn reap_treats_a_closed_channel_as_dead() {
        let mut pg = ProcessGroups::new();
        let (a, ta) = synth(1, 0);
        pg.join("a", a);
        drop(ta); // sender gone, queue empty → try_recv = Err(RecvError) = dead
        let evicted = pg.reap_group("a");
        assert_eq!(evicted.len(), 1);
        assert!(pg.members_of("a").is_empty());
    }

    #[test]
    fn read_backstop_hides_a_member_the_monitor_has_not_yet_reaped() {
        let mut pg = ProcessGroups::new();
        // Both senders held: reap_group would see Ok(None) and evict neither.
        let (a, _ta) = synth(1, 0);
        let (b, _tb) = synth(2, 0);
        pg.join("g", a);
        pg.join("g", b);

        // The slot-word oracle already reports pid 1 dead (finalize window),
        // ahead of any Down delivery.
        let dead = Pid::new(1, 0);
        let oracle = |pid: Pid| pid != dead;

        assert_eq!(pg.members_where("g", oracle), vec![Pid::new(2, 0)], "dead pid filtered from read");
        assert_eq!(pg.first_member_where("g", oracle), Some(Pid::new(2, 0)), "pick skips the dead first member");

        // Backstop does not evict — that stays the monitor's job; raw storage
        // still holds both until reap runs.
        assert_eq!(pg.members_of("g"), vec![member(1, 0), member(2, 0)]);
    }
}