//! Unbounded MPSC channels.
//!
//! Inner state is `Arc<RawMutex<Inner<T>>>` so channels can be sent across OS
//! threads (required for the multi-scheduler runtime where a sender and
//! receiver may run on different scheduler threads simultaneously).
//!
//! ## Why `RawMutex` (Channel class), not `std::sync::Mutex`
//!
//! An actor holding a guard with preemption *enabled* can be timesliced
//! inside the critical section and resume on a different OS thread — the
//! pthread mutex would then be released from a thread that didn't lock it,
//! which is UB (Linux futexes happen to tolerate it, but it's not
//! guaranteed). `RawMutex` disables preemption for the guard's span and is
//! cross-thread-release sound by construction, closing the hole. It also
//! cannot poison. Channel locks form their own [`LockClass::Channel`]
//! (raw_mutex.rs): they may be taken under a cold (Leaf) lock — finalize and
//! `monitor()` clone senders that live in slots — but nothing may be locked
//! under them, which the debug build enforces. `recv_match` runs its user
//! predicate under this lock: keep it cheap, pure, and channel-free.
//!
//! Semantics:
//!   - Senders are clonable; the last sender drop closes the channel.
//!   - `Receiver::recv` on an empty open channel parks the receiver.
//!   - `Receiver::recv` on an empty closed channel returns `Err(RecvError)`.
//!   - `Sender::send` on an open channel always succeeds.
//!   - `Sender::send` on a closed channel (receiver dropped) returns
//!     `Err(SendError(value))`.
//!   - When a send pushes to a previously empty queue and a receiver is
//!     parked, the receiver is unparked.

use crate::pid::Pid;
use crate::raw_mutex::RawMutex;
use std::collections::VecDeque;
use std::sync::Arc;

pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
    let inner = Arc::new(RawMutex::new_channel(Inner {
        queue: VecDeque::new(),
        parked_receiver: None,
        senders: 1,
        receiver_alive: true,
    }));
    (Sender { inner: inner.clone() }, Receiver { inner })
}

struct Inner<T> {
    queue: VecDeque<T>,
    /// The parked receiver's `(pid, park-epoch)`. The epoch is the slot
    /// word's runtime-wide wait identity (see slot_state.rs): wakers call
    /// `unpark_at(pid, epoch)`, so an entry left over from an already-woken
    /// wait — a `select` loser arm, a satisfied `recv_timeout`'s timer — is
    /// inert: the wake fails the word's epoch CAS and no-ops. This replaces
    /// the old per-channel `cur_wait`/`next_wait_seq`/`timed_out` trio: wait
    /// identity now exists exactly once, in the slot word.
    parked_receiver: Option<(Pid, u32)>,
    senders: usize,
    receiver_alive: bool,
}

pub struct Sender<T> {
    inner: Arc<RawMutex<Inner<T>>>,
}

pub struct Receiver<T> {
    inner: Arc<RawMutex<Inner<T>>>,
}

#[derive(Debug, PartialEq, Eq)]
pub struct SendError<T>(pub T);

#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub struct RecvError;

impl std::fmt::Display for RecvError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "channel closed")
    }
}

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

/// Returned by [`Receiver::recv_timeout`].
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum RecvTimeoutError {
    /// The deadline passed with no message available.
    Timeout,
    /// All senders dropped with no message available — the bounded analogue
    /// of [`RecvError`].
    Disconnected,
}

impl std::fmt::Display for RecvTimeoutError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RecvTimeoutError::Timeout => write!(f, "recv timed out"),
            RecvTimeoutError::Disconnected => write!(f, "channel closed"),
        }
    }
}

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

impl<T> Clone for Sender<T> {
    fn clone(&self) -> Self {
        self.inner.lock().senders += 1;
        Sender { inner: self.inner.clone() }
    }
}

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        let unpark = {
            let mut g = self.inner.lock();
            g.senders -= 1;
            // Wake the parked receiver on the last sender drop regardless of
            // whether the queue is empty. A plain `recv` only ever parks on an
            // empty queue (so this is unchanged for it), but a selective
            // `recv_match` may be parked on a *non-empty* queue holding only
            // non-matching messages — it must wake to observe closure and
            // return Err rather than sleep forever.
            if g.senders == 0 {
                g.parked_receiver.take()
            } else {
                None
            }
        };
        if let Some((pid, epoch)) = unpark {
            crate::scheduler::unpark_at(pid, epoch);
        }
    }
}

impl<T> Drop for Receiver<T> {
    fn drop(&mut self) {
        self.inner.lock().receiver_alive = false;
    }
}

impl<T> Sender<T> {
    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
        let unpark = {
            let mut g = self.inner.lock();
            if !g.receiver_alive {
                return Err(SendError(value));
            }
            g.queue.push_back(value);
            g.parked_receiver.take()
        };
        if let Some((pid, epoch)) = unpark {
            crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: Some(pid) });
            crate::scheduler::unpark_at(pid, epoch);
        } else {
            crate::te!(crate::trace::Event::Send { sender: crate::actor::current_pid().unwrap_or(crate::pid::Pid::new(u32::MAX, u32::MAX)), receiver: None });
        }
        Ok(())
    }
}

impl<T> Receiver<T> {
    pub fn recv(&self) -> Result<T, RecvError> {
        loop {
            {
                let mut g = self.inner.lock();
                if let Some(v) = g.queue.pop_front() {
                    return Ok(v);
                }
                if g.senders == 0 {
                    return Err(RecvError);
                }
                let me = crate::actor::current_pid()
                    .expect("recv() called outside an actor");
                debug_assert!(
                    g.parked_receiver.is_none_or(|(p, _)| p == me),
                    "channel has more than one receiver"
                );
                // begin_wait is lock-free — legal under the Channel lock;
                // registering in the same critical section makes the epoch
                // atomic with the senders' view of the registration.
                g.parked_receiver = Some((me, crate::scheduler::begin_wait()));
                crate::te!(crate::trace::Event::RecvPark(me));
            }
            // Release the lock before parking — the unparker will need it.
            crate::scheduler::park_current();
            // Woken up — record it before looping to check the queue.
            crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
        }
    }

    /// Bounded receive: like [`recv`](Self::recv), but gives up once
    /// `timeout` has elapsed, returning [`RecvTimeoutError::Timeout`].
    ///
    /// Built on the same timer machinery as `Mutex::lock_timeout`: the wait
    /// registers a `WaitTimeout` entry stamped with the wait's park-epoch;
    /// on expiry the channel (as the
    /// [`TimerTarget`](crate::timer::TimerTarget)) checks whether *this*
    /// wait is still parked and, only then, cancels it. A wake that races
    /// the deadline resolves message-first: if a message is available when
    /// the receiver runs, it is delivered even if the timer had already
    /// fired. A satisfied or abandoned wait leaves its timer entry to expire
    /// as a no-op (registration gone; epoch consumed), per the
    /// no-cancellation convention in `timer.rs`.
    ///
    /// The wake is classified from state alone — wakes are precise (the only
    /// stamped wakers of this wait are a send, the last-sender drop, and the
    /// timer; a stop wake unwinds out of `park_current` and never reaches
    /// the classification), so: message queued → `Ok`; `senders == 0` →
    /// `Disconnected`; neither → it was the timer → `Timeout`.
    ///
    /// `Duration::ZERO` is a valid timeout: it parks until the immediately-
    /// due timer is drained, then reports `Timeout` unless a message was
    /// already queued.
    pub fn recv_timeout(&self, timeout: std::time::Duration) -> Result<T, RecvTimeoutError>
    where
        T: Send + 'static,
    {
        let me = crate::actor::current_pid()
            .expect("recv_timeout() called outside an actor");

        // Fast path + wait registration, one critical section.
        let epoch;
        {
            let mut g = self.inner.lock();
            if let Some(v) = g.queue.pop_front() {
                return Ok(v);
            }
            if g.senders == 0 {
                return Err(RecvTimeoutError::Disconnected);
            }
            debug_assert!(
                g.parked_receiver.is_none_or(|(p, _)| p == me),
                "channel has more than one receiver"
            );
            epoch = crate::scheduler::begin_wait();
            g.parked_receiver = Some((me, epoch));
            crate::te!(crate::trace::Event::RecvPark(me));
        }

        // Arm the timer after releasing the channel lock (insert takes the
        // timers lock; never nest under a Channel lock). A send or even the
        // timer itself may unpark us before we park — the RunningNotified
        // protocol makes the park below return immediately in that case.
        let deadline = crate::timer::deadline_from_now(timeout);
        let target: std::sync::Arc<dyn crate::timer::TimerTarget> = self.inner.clone();
        crate::scheduler::insert_wait_timer(deadline, me, target, epoch);

        crate::scheduler::park_current();
        crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
        let mut g = self.inner.lock();
        if let Some(v) = g.queue.pop_front() {
            return Ok(v);
        }
        if g.senders == 0 {
            return Err(RecvTimeoutError::Disconnected);
        }
        Err(RecvTimeoutError::Timeout)
    }

    /// Selective receive: remove and return the first queued message for which
    /// `pred` holds, leaving the rest in arrival order. If no queued message
    /// matches, parks and re-scans on every send (a selective receiver may park
    /// on a *non-empty* queue). Returns `Err(RecvError)` only once the channel
    /// is closed and no queued message matches.
    ///
    /// `pred` is run while the channel lock is held: keep it cheap and pure,
    /// and do not call back into this channel from inside it. It is modelled as
    /// `Fn` (not `FnMut`) deliberately — it is re-run from scratch on every
    /// scan, so a stateful predicate would observe surprising re-counting.
    pub fn recv_match<F>(&self, pred: F) -> Result<T, RecvError>
    where
        F: Fn(&T) -> bool,
    {
        loop {
            {
                let mut g = self.inner.lock();
                if let Some(i) = g.queue.iter().position(|v| pred(v)) {
                    // position() found it, so remove() returns Some.
                    return Ok(g.queue.remove(i).unwrap());
                }
                if g.senders == 0 {
                    // Closed and nothing queued can ever match.
                    return Err(RecvError);
                }
                let me = crate::actor::current_pid()
                    .expect("recv_match() called outside an actor");
                debug_assert!(
                    g.parked_receiver.is_none_or(|(p, _)| p == me),
                    "channel has more than one receiver"
                );
                g.parked_receiver = Some((me, crate::scheduler::begin_wait()));
                crate::te!(crate::trace::Event::RecvPark(me));
            }
            // Release the lock before parking — the unparker will need it.
            crate::scheduler::park_current();
            crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap()));
        }
    }

    /// Non-blocking selective receive. `Ok(Some(v))` if a queued message
    /// matched `pred` (removed, rest left in order), `Ok(None)` if the channel
    /// is open but nothing matched, `Err(RecvError)` if closed and nothing
    /// matched. Same predicate contract as [`recv_match`](Self::recv_match).
    pub fn try_recv_match<F>(&self, pred: F) -> Result<Option<T>, RecvError>
    where
        F: Fn(&T) -> bool,
    {
        let mut g = self.inner.lock();
        if let Some(i) = g.queue.iter().position(|v| pred(v)) {
            return Ok(Some(g.queue.remove(i).unwrap()));
        }
        if g.senders == 0 {
            return Err(RecvError);
        }
        Ok(None)
    }

    /// Non-blocking. `Ok(Some(v))` if a message was available, `Ok(None)` if
    /// the channel is empty but open, `Err(RecvError)` if closed and drained.
    pub fn try_recv(&self) -> Result<Option<T>, RecvError> {
        let mut g = self.inner.lock();
        if let Some(v) = g.queue.pop_front() {
            return Ok(Some(v));
        }
        if g.senders == 0 {
            return Err(RecvError);
        }
        Ok(None)
    }
}

// ---------------------------------------------------------------------------
// TimerTarget — the expiry half of recv_timeout
// ---------------------------------------------------------------------------

impl<T: Send + 'static> crate::timer::TimerTarget for RawMutex<Inner<T>> {
    fn on_timeout(&self, pid: Pid, epoch: u32) {
        // Cancel the wait only if THIS wait (epoch match) is still
        // registered. If a sender already took `parked_receiver`, the
        // receiver is waking with a message — message wins, the timer
        // no-ops. If a later wait by the same receiver is registered, the
        // epoch mismatches — stale entry, no-op. (The unpark_at would fail
        // its word CAS in either case anyway; checking under the lock keeps
        // the registration bookkeeping exact.)
        let unpark = {
            let mut g = self.lock();
            if g.parked_receiver == Some((pid, epoch)) {
                g.parked_receiver = None;
                true
            } else {
                false
            }
        };
        // Unpark outside the channel lock — it may take the run-queue lock;
        // legal under a Channel lock, but pointless to nest.
        if unpark {
            crate::scheduler::unpark_at(pid, epoch);
        }
    }
}

// ---------------------------------------------------------------------------
// select — ready-index wait over multiple receivers
// ---------------------------------------------------------------------------

mod sealed {
    pub trait Sealed {}
}
impl<T> sealed::Sealed for Receiver<T> {}

/// An arm of a [`select`]. Implemented by [`Receiver`]; sealed, because the
/// registration contract below is part of the runtime's wake protocol.
///
/// Contract (all under the arm's own lock): `sel_register` checks-or-
/// registers atomically — if the arm is ready it does NOT register and
/// returns `false`; otherwise it publishes `(pid, epoch)` where its wakers
/// will find it. "Ready" means a receive would not park: a message is
/// queued, or the arm is closed.
pub trait Selectable: sealed::Sealed {
    #[doc(hidden)]
    fn sel_register(&self, pid: Pid, epoch: u32) -> bool;
    #[doc(hidden)]
    fn sel_ready(&self) -> bool;
}

impl<T> Selectable for Receiver<T> {
    fn sel_register(&self, pid: Pid, epoch: u32) -> bool {
        let mut g = self.inner.lock();
        if !g.queue.is_empty() || g.senders == 0 {
            return false;
        }
        debug_assert!(
            g.parked_receiver.is_none_or(|(p, _)| p == pid),
            "channel has more than one receiver"
        );
        g.parked_receiver = Some((pid, epoch));
        true
    }

    fn sel_ready(&self) -> bool {
        let g = self.inner.lock();
        !g.queue.is_empty() || g.senders == 0
    }
}

/// Park on every arm at once; return the index of the first ready one.
///
/// "Ready" means a receive on that arm would not park: a message is queued,
/// or the arm is **closed** (so the caller's `try_recv` observes the
/// disconnect — a dead arm is an event, not a hang). The caller consumes the
/// arm itself, typically via [`Receiver::try_recv`]; single-receiver
/// channels guarantee nothing can steal the message in between.
///
/// A closed arm stays ready *forever*: once its disconnect has been
/// observed, drop it from the arm set — under priority order it would
/// otherwise win every subsequent call and starve every higher-indexed arm.
///
/// Arms are scanned **in order**: index 0 is the highest priority, both on
/// the immediate-ready path and after a wake. This is a documented
/// guarantee (compose like BEAM receive clauses: put control channels
/// first), not an accident — and therefore there is NO fairness promise; a
/// saturated arm 0 starves arm 1 by design.
///
/// One actor may select on a channel and later `recv` on it (or select on
/// overlapping sets) freely. What stays illegal is what was always illegal:
/// two *different* actors receiving on one channel.
///
/// Built on the consuming-wake protocol (see slot_state.rs): all arms are
/// registered under one wait epoch; the winning wake consumes it, so losing
/// arms' registrations are inert and need no cancellation pass — they
/// self-clean at their wakers' failed CAS, or get overwritten by this
/// receiver's next wait on that channel.
///
/// Panics if `arms` is empty, or when called outside an actor.
pub fn select(arms: &[&dyn Selectable]) -> usize {
    assert!(!arms.is_empty(), "select() on an empty arm list");
    let me = crate::actor::current_pid().expect("select() called outside an actor");
    loop {
        let epoch = crate::scheduler::begin_wait();
        if let Some(i) = register_arms(me, epoch, arms) {
            return i;
        }

        crate::scheduler::park_current();
        // Woken precisely: an arm's send (message) or last-sender drop
        // (closure) consumed our epoch, and both leave their arm ready —
        // return the first one, in priority order (which may be a
        // different, higher-priority arm than the one that woke us; its
        // message stays queued and re-reports ready on the next call).
        for (i, arm) in arms.iter().enumerate() {
            if arm.sel_ready() {
                return i;
            }
        }
        // Unreachable by protocol (a stop wake unwinds out of
        // park_current). Defensive: re-open the wait and re-register —
        // stale own-registrations are overwritten.
    }
}

/// The registration pass shared by [`select`] and [`select_timeout`]:
/// check-or-register each arm, in priority order, each atomically under its
/// own lock. Cross-arm atomicity is unnecessary: an arm becoming ready
/// right after its registration wakes the caller through the protocol (the
/// prep-to-park window is closed by RunningNotified).
///
/// `Some(i)` = arm `i` was ready, the pass stopped, and the wait has been
/// RETIRED (no park may follow): earlier arms hold live-epoch
/// registrations, so the epoch is bumped, a landed notification eaten, and
/// a pending stop re-observed — without which a stale arm wake could fault
/// a later one-shot park. `None` = every arm registered; the caller parks.
fn register_arms(me: Pid, epoch: u32, arms: &[&dyn Selectable]) -> Option<usize> {
    for (i, arm) in arms.iter().enumerate() {
        if !arm.sel_register(me, epoch) {
            crate::scheduler::retire_wait();
            return Some(i);
        }
    }
    None
}

/// The [`select_timeout`] timer target: stateless, because precise wakes
/// make classification a pure function of channel state. The entry is
/// stamped with the select's epoch; if an arm already won, this unpark dies
/// at the word's epoch CAS (the no-cancellation convention in `timer.rs`).
struct SelectTimeout;
impl crate::timer::TimerTarget for SelectTimeout {
    fn on_timeout(&self, pid: Pid, epoch: u32) {
        crate::scheduler::unpark_at(pid, epoch);
    }
}

/// [`select`] with a deadline: returns `Some(index)` like `select`, or
/// `None` once `timeout` elapses with no arm ready.
///
/// All of `select`'s semantics carry over (priority order, closed arms
/// permanently ready, no fairness promise). The timeout is one more stamped
/// waker on the same wait epoch — nothing is registered in any arm for it,
/// so there is nothing to cancel or leak: an arm winning leaves the timer
/// entry to expire as a stale-epoch no-op; the timer winning leaves the
/// arms' registrations to self-clean exactly as a `select` loser's would.
///
/// The wake is classified from state alone (wakes are precise): some arm
/// ready → `Some` of the first, in priority order; none ready → the timer
/// was the only remaining stamped waker → `None`. A message that races the
/// deadline resolves message-first, as `recv_timeout` does.
///
/// `Duration::ZERO` is a valid timeout: it parks until the immediately-due
/// timer is drained, then reports `None` unless an arm was already ready.
///
/// Panics if `arms` is empty, or when called outside an actor.
pub fn select_timeout(
    arms: &[&dyn Selectable],
    timeout: std::time::Duration,
) -> Option<usize> {
    assert!(!arms.is_empty(), "select_timeout() on an empty arm list");
    let me = crate::actor::current_pid().expect("select_timeout() called outside an actor");
    let epoch = crate::scheduler::begin_wait();
    if let Some(i) = register_arms(me, epoch, arms) {
        return Some(i); // ready now: the timer was never armed
    }

    // Arm the timer after the registration pass, outside every Channel
    // lock (insert takes the timers lock).
    let deadline = crate::timer::deadline_from_now(timeout);
    let target: std::sync::Arc<dyn crate::timer::TimerTarget> = std::sync::Arc::new(SelectTimeout);
    crate::scheduler::insert_wait_timer(deadline, me, target, epoch);

    crate::scheduler::park_current();
    // Woken precisely: an arm (ready below) or the timer (nothing ready).
    arms.iter().position(|arm| arm.sel_ready())
}