actor.rs

// Channel actor — the tokio task that owns ChannelState while Local.
//
// The actor has two modes:
//
//   Local  — owns Box<ChannelState> + Subsystems and runs its own 20 ms
//            ticker. Handles every command in-process.
//   Mixed  — state + subs have migrated into a MixGroup actor. The channel
//            actor becomes a forwarder: JS commands get relayed into the
//            mixer via `MixHandle::forward()`. Close / LeaveMix pull the
//            Member back out of the mixer and restore Local ownership.
//
// This is the "bite the bullet" design the MIGRATE.md note describes: the
// mixer owns the tick so all the cross-channel math (summed_minus, DTMF fan-
// out) runs in lockstep, free of the old deposit/emit race.

use std::collections::VecDeque;
use std::net::SocketAddr;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use tokio::net::UdpSocket;
use tokio::sync::mpsc;
use tokio::time::{interval, Interval, MissedTickBehavior};
use tokio_util::sync::CancellationToken;

use super::commands::{Command, Handle};
use super::dtmf::{DtmfReceiver, DtmfSender};
use super::mixer::{Member, MixHandle};
use super::player::Player;
use super::recorder::{FinishReason, Recorder, RecorderConfig, RecorderState};
use super::state::{ChannelState, CloseInfo};
use super::tick::{self, TickOutcome};

pub const TICK_MS: u64 = 20;
pub const DEFAULT_CMD_QUEUE_DEPTH: usize = 64;

/// RFC 3550 quality summary attached to the Close event. `in_*` is our
/// reception of the peer's stream (from `RxStats`); `out_*` is the peer's
/// reported reception of the stream we send (from its SR/RR, in `RemoteReport`).
#[derive(Debug, Clone, Default)]
pub struct RtcpSummary {
    /// Whether we have latched an inbound source (received at least one RTP
    /// packet). When false the `in_*` figures are zeroed, not meaningful — the
    /// mirror of `remote_valid` for the outbound direction.
    pub in_valid: bool,
    pub in_cumulative_lost: i32,
    pub in_fraction_lost: u8,
    pub in_jitter: u32,
    /// Whether the peer has sent us at least one report about our stream.
    pub remote_valid: bool,
    pub out_fraction_lost: u8,
    pub out_cumulative_lost: i32,
    pub out_jitter: u32,
    /// Round-trip time in ms, or `None` until the peer echoes one of our SRs.
    pub rtt_ms: Option<f64>,
}

/// Events the actor emits back to the outside world. The JS facade (Task #9)
/// will bridge these into a napi `ThreadsafeFunction`.
#[derive(Debug, Clone, Default)]
pub struct ChannelStats {
    pub in_count: u64,
    pub in_dropped: u64,
    pub in_skip: u64,
    pub out_count: u64,
    /// `None` until the channel has RTCP data (received RTP or a peer report).
    pub rtcp: Option<RtcpSummary>,
}

impl ChannelStats {
    /// Compute MOS from packet loss, using the ITU-T G.107 inspired formula
    /// from the C++ codebase (originally borrowed from FreeSWITCH).
    /// Returns 0.0 when no packets were received.
    pub fn mos(&self) -> f64 {
        if self.in_count == 0 {
            return 0.0;
        }
        // `saturating_sub`: `in_skip` now carries real RFC 3550 loss, which a
        // pathological stream could in principle drive above `in_count`.
        let received = self.in_count.saturating_sub(self.in_skip);
        let r = (received as f64 / self.in_count as f64) * 100.0;
        let r = r.clamp(0.0, 100.0);
        1.0 + (0.035 * r) + (0.000007 * r * (r - 60.0) * (100.0 - r))
    }
}

/// Snapshot the channel's stats for a Close event. Shared by the Local
/// (`actor.rs`) and Mixed (`mixer.rs`) close paths so the two never drift.
///
/// Routes the real RFC 3550 cumulative-loss figure into `in_skip` — the field
/// is otherwise never incremented, so `mos()` used to read near-perfect for
/// every call regardless of actual loss.
pub fn build_channel_stats(state: &ChannelState) -> ChannelStats {
    let in_count = state.in_count.load(Ordering::Relaxed);

    // `fraction_lost_session` is non-mutating: reading it here must not disturb
    // the per-interval counters that `rtcp_tx`'s periodic report blocks consume.
    let (in_lost, in_jitter, in_fraction, have_source) = {
        let rx = state.rx_stats.lock();
        (
            rx.cumulative_lost(),
            rx.jitter(),
            rx.fraction_lost_session(),
            rx.remote_ssrc.is_some(),
        )
    };
    let remote = *state.remote_report.lock();

    let rtcp = if have_source || remote.valid {
        Some(RtcpSummary {
            in_valid: have_source,
            // Zeroed until an inbound source is latched, so a send-only channel
            // doesn't report a phantom loss of 1 (`expected()` is 1 before the
            // first packet while `received` is still 0).
            in_cumulative_lost: if have_source { in_lost } else { 0 },
            in_fraction_lost: if have_source { in_fraction } else { 0 },
            in_jitter: if have_source { in_jitter } else { 0 },
            remote_valid: remote.valid,
            out_fraction_lost: remote.fraction_lost,
            out_cumulative_lost: remote.cumulative_lost,
            out_jitter: remote.jitter,
            rtt_ms: remote.rtt_ms,
        })
    } else {
        None
    };

    // Real inbound loss → in_skip (clamped: never negative, never exceeds the
    // number of packets we counted receiving). Only when a source is latched,
    // so the phantom pre-first-packet loss never leaks into MOS.
    let in_skip = if have_source {
        (in_lost.max(0) as u64).min(in_count)
    } else {
        0
    };

    ChannelStats {
        in_count,
        in_dropped: state.in_dropped + state.jitter.lock().dropped,
        in_skip,
        out_count: state.out_count,
        rtcp,
    }
}

#[derive(Debug, Clone)]
pub enum Event {
    Close {
        reason: String,
        stats: ChannelStats,
    },
    Play {
        state: PlayState,
        reason: Option<String>,
    },
    Record {
        state: RecordState,
        reason: Option<String>,
        file: Option<String>,
        filesize: Option<u64>,
    },
    Telephone {
        digit: char,
    },
    Mix {
        state: String,
    },
}

#[derive(Debug, Clone)]
pub enum PlayState {
    Start,
    End,
}

#[derive(Debug, Clone)]
pub enum RecordState {
    Recording,
    Finished,
}

/// Abstract sink for events — a trait so tests can capture events into a
/// channel and the napi facade can forward into a ThreadsafeFunction.
pub trait EventSink: Send + Sync + 'static {
    fn post(&self, ev: Event);
}

pub struct SpawnConfig {
    pub id: u64,
    #[allow(dead_code)]
    pub bind_addr: SocketAddr,
    pub ssrc: u32,
    pub events: Arc<dyn EventSink>,
    /// `Some` when the port came from the managed pool — handed to
    /// ChannelState so the port is returned on actor exit.
    pub port_reservation: Option<crate::portpool::PortReservation>,
    /// Our own ICE password — used for STUN Binding Request integrity checks.
    pub local_icepwd: String,
}

#[cfg(test)]
pub async fn spawn(cfg: SpawnConfig) -> std::io::Result<Handle> {
    let rtp_sock = UdpSocket::bind(cfg.bind_addr).await?;
    let local_addr = rtp_sock.local_addr()?;
    let rtcp_port = local_addr
        .port()
        .checked_add(1)
        .unwrap_or(local_addr.port());
    let rtcp_sock = UdpSocket::bind(SocketAddr::new(local_addr.ip(), rtcp_port)).await?;
    spawn_with_sockets(cfg, rtp_sock, rtcp_sock, local_addr)
}

/// Sync variant: takes already-bound sockets. Used by the JS facade so
/// openchannel() can return a value (not a Promise) that index.js can read
/// `.local.port` on immediately.
pub fn spawn_with_sockets(
    cfg: SpawnConfig,
    rtp_sock: UdpSocket,
    rtcp_sock: UdpSocket,
    local_addr: SocketAddr,
) -> std::io::Result<Handle> {
    let mut state = ChannelState::new(cfg.id, local_addr, rtp_sock, rtcp_sock, cfg.ssrc);
    state.port_reservation = cfg.port_reservation;
    *state.local_icepwd.lock() = cfg.local_icepwd;

    let cancel = CancellationToken::new();

    // DTLS keying material is published here once the handshake completes so
    // the inbound RTCP readers can build their SRTCP decrypt context (Tier 2).
    // Both the dedicated P+1 loop and — under rtcp-mux — the RTP recv_loop
    // subscribe, so create the watch pair before spawning either.
    let (srtp_key_tx, srtp_key_rx) = tokio::sync::watch::channel(None);
    state.srtp_key_tx = Some(srtp_key_tx);

    // Spawn the recv_loop — reads the RTP socket continuously, classifies
    // STUN/DTLS/RTP and feeds jitter/DTLS-mpsc immediately. It also demuxes
    // rtcp-mux'd RTCP (RFC 5761) off the RTP port, hence the RTCP accounting
    // fields and the SRTCP key subscription.
    super::recv_loop::spawn(super::recv_loop::RecvLoopConfig {
        sock: state.rtp_sock.clone(),
        jitter: state.jitter.clone(),
        remote_addr: state.remote_addr.clone(),
        in_count: state.in_count.clone(),
        rx_stats: state.rx_stats.clone(),
        remote_report: state.remote_report.clone(),
        local_ssrc: state.ssrc,
        local_icepwd: state.local_icepwd.clone(),
        dtls_tx: state.dtls_inbound_tx.clone(),
        key_rx: srtp_key_rx.clone(),
        cancel: cancel.clone(),
    });

    // Spawn the inbound RTCP reader on the P+1 control socket. It shares the
    // recv_loop cancellation token, so the close path stops both at once.
    super::rtcp_loop::spawn(super::rtcp_loop::RtcpLoopConfig {
        sock: state.rtcp_sock.clone(),
        rx_stats: state.rx_stats.clone(),
        remote_report: state.remote_report.clone(),
        local_ssrc: state.ssrc,
        key_rx: srtp_key_rx,
        cancel: cancel.clone(),
    });

    state.recv_cancel = Some(cancel);

    let (tx, rx) = mpsc::channel::<Command>(DEFAULT_CMD_QUEUE_DEPTH);
    let events = cfg.events;
    let handle_cmd = tx.clone();
    tokio::spawn(async move { run(Box::new(state), rx, events, handle_cmd).await });
    Ok(Handle {
        id: cfg.id,
        cmd: tx,
    })
}

/// Upper bound on the pre-buffer queue — cap is drop-oldest. Sized to hold
/// 30 s at 16 kHz mono, which covers typical IVR prompt lengths (2–15 s) with
/// headroom and bounds memory at ~1 MB per pending recorder. C++ uses an
/// unbounded queue; a bound is deliberate here to protect a multi-channel
/// server from a stuck/runaway prompt.
pub const PREBUFFER_CAPACITY_SAMPLES: usize = 480_000;

/// A recorder configured by `playrecord` but not yet opened. The file isn't
/// created until activation (play-end or barge-in). Holds a cached
/// `file_str` so the activation event has the path after `cfg` is moved
/// into `Recorder::open`.
pub struct PendingRecorder {
    pub cfg: RecorderConfig,
    pub file_str: String,
}

/// Per-channel subsystems the tick pipeline hands off to. Keeping them in a
/// sibling struct (not in ChannelState) avoids mixing pipeline-owned and
/// control-owned state — the mixer will move ChannelState but leaves these
/// with the channel actor, since they're driven by commands more than ticks.
#[derive(Default)]
pub struct Subsystems {
    pub player: Option<Player>,
    /// Multiple simultaneous recorders — one ongoing + one power-gated is the
    /// primary use case (see test/interface/projectrtprecord.js "dual
    /// recording"). Keyed informally by `recorder.file()`; same path replaces
    /// the existing recorder, different path coexists.
    pub recorders: Vec<Recorder>,
    /// Barge-in detector for `playrecord`. When the player is playing and
    /// inbound RMS crosses `power_threshold`, the player is interrupted and
    /// a `play/end reason=interrupted` event fires. Cleared when the player
    /// ends (naturally or via interrupt).
    pub bargein: Option<BargeInState>,
    /// Recorder queued by `playrecord` but not yet opened. Activated on
    /// player-end or on barge-in interrupt (matches C++ `pendingrecorder`).
    pub pending_recorder: Option<PendingRecorder>,
    /// Inbound samples captured while `pending_recorder` is set — drained
    /// into the recorder on activation via `Recorder::write_raw` so they are
    /// written regardless of the recorder's start-above-power gate.
    pub prebuffer: VecDeque<i16>,
    /// JS-initiated dtmf via `channel.dtmf(...)` — targets state.remote_addr.
    pub dtmf_send: DtmfSender,
    /// Mix-relay dtmf — when a digit is detected on a peer's inbound while
    /// mixed, a full RFC 2833 burst is enqueued here and emitted to the
    /// local remote on the next tick. Matches the C++ mux DTMF behaviour.
    pub dtmf_relay: DtmfSender,
    pub dtmf_recv: DtmfReceiver,
    /// Live readers — each maps to one JS `createReadStream` consumer.
    /// Fed post-decode at the same point as recorders; see audio_reader.rs
    /// for the drop policy.
    pub readers: Vec<super::audio_reader::AudioReader>,
    /// Active writer, if any. Only one outbound source at a time —
    /// `createWriteStream` supersedes any running `play`, and vice
    /// versa. See audio_writer.rs for the drain / end semantics.
    pub writer: Option<super::audio_writer::AudioWriter>,
}

pub struct BargeInState {
    pub power_threshold: i32,
    pub power_ma: crate::firfilter::MaFilter,
}

/// Activate a `pending_recorder`: open the recorder, flush the pre-buffer
/// via `write_raw` (bypassing the start-above-power gate so barge-in speech
/// is captured), replace any existing recorder at the same path, and queue
/// a `record event:recording` event when not gated.
///
/// Events are pushed into `pending_events` rather than posted directly so
/// the tick-pipeline callers (which drain `state.pending_events` after each
/// tick) can order them alongside the `play/end` that precedes activation.
///
/// No-op (returns false) if no `pending_recorder` is set. Matches C++
/// `activateplayrecordrecorder` (projectrtpchannel.cpp:849-878).
pub(super) async fn activate_pending_recorder(
    subs: &mut Subsystems,
    pending_events: &mut Vec<Event>,
) -> bool {
    let pending = match subs.pending_recorder.take() {
        Some(p) => p,
        None => return false,
    };
    let PendingRecorder { cfg, file_str } = pending;
    let is_gated = cfg.start_above_power.is_some();
    match Recorder::open(cfg).await {
        Ok(mut rec) => {
            let drained: Vec<i16> = subs.prebuffer.drain(..).collect();
            if !drained.is_empty() {
                let frame = if rec.num_channels() == 2 {
                    let mut inter = Vec::with_capacity(drained.len() * 2);
                    for s in &drained {
                        inter.push(*s);
                        inter.push(*s);
                    }
                    inter
                } else {
                    drained
                };
                let _ = rec.write_raw(&frame).await;
            }
            if let Some(idx) = subs.recorders.iter().position(|r| r.file() == rec.file()) {
                let mut old = subs.recorders.remove(idx);
                old.close(FinishReason::ChannelClosed);
            }
            subs.recorders.push(rec);
            if !is_gated {
                pending_events.push(Event::Record {
                    state: RecordState::Recording,
                    reason: None,
                    file: Some(file_str),
                    filesize: None,
                });
            }
            true
        }
        Err(e) => {
            subs.prebuffer.clear();
            pending_events.push(Event::Record {
                state: RecordState::Finished,
                reason: Some(format!("open-failed: {e}")),
                file: Some(file_str),
                filesize: None,
            });
            false
        }
    }
}

/// Top-level actor ownership. `Local` runs the per-channel ticker; `Mixed`
/// delegates ticks + most commands to the mix actor.
//
// `Local` is the common (non-mixed) case and carries the full `Subsystems`
// inline; `Mixed` is small. Boxing `Local` to equalize the variants would
// add a heap indirection on the hot per-channel path and ripple through the
// mix-migration `Member` type, so we keep it inline deliberately.
#[allow(clippy::large_enum_variant)]
enum Mode {
    Local {
        state: Box<ChannelState>,
        subs: Subsystems,
    },
    Mixed {
        mix: MixHandle,
    },
}

async fn run(
    initial_state: Box<ChannelState>,
    mut cmds: mpsc::Receiver<Command>,
    events: Arc<dyn EventSink>,
    self_cmd: mpsc::Sender<Command>,
) {
    let id = initial_state.id;
    let mut mode = Mode::Local {
        state: initial_state,
        subs: Subsystems::default(),
    };
    let mut ticker = interval(Duration::from_millis(TICK_MS));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);

    let mut closing: Option<String> = None;

    while closing.is_none() {
        match &mut mode {
            Mode::Local { state, subs } => {
                match step_local(state, subs, &mut cmds, &mut ticker, &events).await {
                    LocalStep::Continue => {}
                    LocalStep::Close(reason) => closing = Some(reason),
                    LocalStep::EnterMix { mix, ack } => {
                        // Migrate into the mixer. Need to move out of `mode`
                        // which requires taking ownership — swap to a temp
                        // Local with a dummy then rebuild the real Mixed.
                        let (state_out, mut subs_out) = take_local(&mut mode);
                        // Clear jitter so a remixed channel that restarts
                        // its SN counter doesn't get rejected as out-of-window.
                        state_out.jitter.lock().clear();
                        // Stop any active player — matches C++ mix2 behavior.
                        // Without this, a ringing player on an existing channel
                        // leaks into the mix output (e.g. blind transfer: b
                        // hears ringing while c rings, then c answers and the
                        // mix is established — the ringing must stop).
                        if subs_out.player.is_some() {
                            subs_out.player = None;
                            subs_out.bargein = None;
                            events.post(Event::Play {
                                state: PlayState::End,
                                reason: Some("mix".into()),
                            });
                        }
                        // Any pending playrecord recorder: no Recording was
                        // emitted yet, so nothing to finish — just drop.
                        subs_out.pending_recorder = None;
                        subs_out.prebuffer.clear();
                        match mix
                            .add(Box::new(Member::new(state_out, subs_out, events.clone())))
                            .await
                        {
                            Ok(()) => {
                                events.post(Event::Mix {
                                    state: "start".to_string(),
                                });
                                mode = Mode::Mixed { mix };
                                let _ = ack.send(());
                            }
                            Err(()) => {
                                // Mixer is gone (shouldn't happen in normal
                                // flow). Fall back to Local mode — state was
                                // consumed into the `Member` that got
                                // dropped. Repopulate a zero-state placeholder
                                // and close so the channel doesn't hang.
                                closing = Some("mix-add-failed".to_string());
                            }
                        }
                    }
                }
            }
            Mode::Mixed { mix } => {
                match cmds.recv().await {
                    None => closing = Some("handle-dropped".to_string()),
                    Some(Command::Close { reason }) => {
                        if let Some(member) = mix.remove(id).await {
                            restore_local(&mut mode, member);
                            events.post(Event::Mix {
                                state: "finished".to_string(),
                            });
                        }
                        closing = Some(reason);
                    }
                    Some(Command::LeaveMix { ack }) => {
                        if let Some(member) = mix.remove(id).await {
                            restore_local(&mut mode, member);
                            events.post(Event::Mix {
                                state: "finished".to_string(),
                            });
                        }
                        let _ = ack.send(());
                    }
                    Some(Command::EnterMix { mix: new_mix, ack }) => {
                        // Already in a mix. If it's the same one (facade
                        // re-asserts on same-group mix()), post a fresh
                        // `mix/start` event so JS sees one per mix() call.
                        // Different-group migration isn't supported yet.
                        if new_mix.id == mix.id {
                            events.post(Event::Mix {
                                state: "start".to_string(),
                            });
                        }
                        let _ = ack.send(());
                    }
                    Some(cmd) => {
                        // All other commands get forwarded to the mixer for
                        // application against this member's state/subs.
                        mix.forward(id, cmd).await;
                    }
                }
            }
        }
    }

    let reason = closing.unwrap_or_default();

    // If we arrived here via a path that left us mixed (shouldn't happen
    // since close/leavemix above pull out first, but be defensive), still
    // try to reclaim state for the stats payload. Use a timeout so a dead
    // mixer doesn't strand the actor.
    if let Mode::Mixed { mix } = &mode {
        if let Ok(Some(member)) =
            tokio::time::timeout(Duration::from_millis(100), mix.remove(id)).await
        {
            restore_local(&mut mode, member);
            events.post(Event::Mix {
                state: "finished".to_string(),
            });
        }
    }

    let (state, subs) = match &mut mode {
        Mode::Local { state, subs } => (state, subs),
        Mode::Mixed { .. } => {
            // Last-resort: the defensive `mix.remove(id)` above timed out or
            // returned None, so we never transitioned back to Local. Emit a
            // minimal Close event without stats — but unregister from the
            // registry FIRST (same ordering rule as the Local path below).
            // Without this, a mixed channel whose mixer was contended at
            // teardown stays in CHANNEL_REGISTRY forever and leaks
            // `stats.channel.current` (rtp_open_count) even though the client
            // saw a clean Close.
            super::facade::unregister_channel(id);
            events.post(Event::Close {
                reason,
                stats: ChannelStats::default(),
            });
            drop(self_cmd);
            return;
        }
    };

    // Player: if still active on close, emit `play/end reason=channelclosed`.
    if subs.player.is_some() {
        subs.player = None;
        subs.bargein = None;
        events.post(Event::Play {
            state: PlayState::End,
            reason: Some("channelclosed".into()),
        });
    }
    for mut rec in subs.recorders.drain(..) {
        let file_str = rec.file().to_string_lossy().into_owned();
        let size = rec.file_size();
        rec.close(FinishReason::ChannelClosed);
        events.post(Event::Record {
            state: RecordState::Finished,
            reason: Some("channelclosed".into()),
            file: Some(file_str),
            filesize: Some(size),
        });
    }
    // Drop readers — their mpsc senders close, forwarder tasks exit, JS
    // `Readable.push(null)` fires `end`. No event emitted here: the
    // JS-side Readable already signals `end`/`close` to userland.
    subs.readers.clear();
    // Writer: same pattern, opposite direction. Dropping it closes the
    // Receiver; the JS side's next `push_writer_bytes` will see the
    // channel is gone and surface an error on the Writable.
    subs.writer = None;
    // A pending_recorder was accepted by `playrecord` but never activated —
    // the file was never opened, so there are no bytes to report. Still emit
    // a Finished event so every `record` start has a matching finish on the
    // wire (babble-sip tolerates missing, but the protocol rule is paired).
    if let Some(pending) = subs.pending_recorder.take() {
        events.post(Event::Record {
            state: RecordState::Finished,
            reason: Some("channelclosed".into()),
            file: Some(pending.file_str),
            filesize: Some(0),
        });
    }
    subs.prebuffer.clear();
    // RTCP BYE (Tier 2): tell the peer the stream is ending now, before we
    // cancel the loops and drop the sockets. Best-effort; encrypted as SRTCP
    // on a secure channel.
    super::rtcp_tx::send_bye(state).await;
    // Abort any in-flight DTLS handshake task. Without this, a handshake
    // that hasn't completed (peer disappeared, no response, etc.) outlives
    // the channel and busy-spins the runtime: the task's mpsc senders are
    // about to drop, and webrtc-dtls's handshake driver re-polls recv on
    // every Err. Abort first, then drop the inbound sender — that order
    // guarantees the spawned task is gone before any remaining `recv()`
    // future would see `None`.
    if let Some(abort) = state.dtls_handshake_abort.take() {
        abort.abort();
    }
    *state.dtls_inbound_tx.lock() = None;
    // Cancel the recv_loop before collecting stats.
    if let Some(cancel) = state.recv_cancel.take() {
        cancel.cancel();
    }
    let stats = build_channel_stats(state);
    state.close_info = Some(CloseInfo {
        reason: reason.clone(),
    });
    // Order matters: `unregister_channel` decrements the registry BEFORE
    // the Close event is posted. Otherwise a JS `afterEach` that asserts
    // `stats.channel.current === 0` races the TSFN queue — the callback
    // can fire while the registry still holds this channel.
    super::facade::unregister_channel(id);
    events.post(Event::Close { reason, stats });
    drop(self_cmd);
}

/// Replaces `mode` temporarily with a placeholder Mixed so we can move the
/// owned state + subs out. Caller must immediately rebuild `mode` on the
/// success path. On failure the placeholder Mixed is left in place and the
/// actor proceeds to close.
fn take_local(mode: &mut Mode) -> (Box<ChannelState>, Subsystems) {
    // Swap out with a temporary Mixed placeholder — we'll overwrite mode
    // right after. The placeholder MixHandle is never used because the
    // caller transitions mode before any further command handling runs.
    let placeholder = Mode::Mixed {
        mix: MixHandle {
            id: 0,
            cmd: mpsc::channel(1).0,
        },
    };
    let taken = std::mem::replace(mode, placeholder);
    match taken {
        Mode::Local { state, subs } => (state, subs),
        Mode::Mixed { .. } => unreachable!("take_local called from non-Local mode"),
    }
}

fn restore_local(mode: &mut Mode, member: Box<Member>) {
    let Member { state, subs, .. } = *member;
    *mode = Mode::Local { state, subs };
}

enum LocalStep {
    Continue,
    Close(String),
    EnterMix {
        mix: MixHandle,
        ack: tokio::sync::oneshot::Sender<()>,
    },
}

async fn step_local(
    state: &mut ChannelState,
    subs: &mut Subsystems,
    cmds: &mut mpsc::Receiver<Command>,
    ticker: &mut Interval,
    events: &Arc<dyn EventSink>,
) -> LocalStep {
    tokio::select! {
        biased;

        cmd = cmds.recv() => {
            match cmd {
                None => LocalStep::Close("handle-dropped".to_string()),
                Some(cmd) => match handle_command_local(state, subs, cmd, events).await {
                    LocalOutcome::Continue => LocalStep::Continue,
                    LocalOutcome::Close(r) => LocalStep::Close(r),
                    LocalOutcome::EnterMix { mix, ack } => LocalStep::EnterMix { mix, ack },
                }
            }
        }

        _ = ticker.tick() => {
            let outcome = tick::run(state, subs).await;
            for ev in state.pending_events.drain(..) {
                events.post(ev);
            }
            if outcome == TickOutcome::Stop {
                LocalStep::Close("idle-timeout".to_string())
            } else {
                LocalStep::Continue
            }
        }
    }
}

enum LocalOutcome {
    Continue,
    Close(String),
    EnterMix {
        mix: MixHandle,
        ack: tokio::sync::oneshot::Sender<()>,
    },
}

async fn handle_command_local(
    state: &mut ChannelState,
    subs: &mut Subsystems,
    cmd: Command,
    events: &Arc<dyn EventSink>,
) -> LocalOutcome {
    match cmd {
        Command::Close { reason } => LocalOutcome::Close(reason),

        Command::EnterMix { mix, ack } => LocalOutcome::EnterMix { mix, ack },

        Command::LeaveMix { ack } => {
            // Not in a mix — ack immediately. Matches C++ `unmix()` returning
            // true even when the channel wasn't mixed.
            let _ = ack.send(());
            LocalOutcome::Continue
        }

        Command::Direction(d) => {
            state.direction = d;
            LocalOutcome::Continue
        }

        Command::Echo { enabled } => {
            state.echo = enabled;
            LocalOutcome::Continue
        }

        Command::Remote { cfg, ack } => {
            state.set_remote_addr(cfg.addr);
            state.ticks_without_rtp = 0;
            state.remote_pt = cfg.payload_type;
            state.rtcpmux = cfg.rtcpmux;
            state.codecx.set_negotiated_pt(cfg.payload_type);
            if let Some(pt) = cfg.rfc2833_payload_type {
                state.rfc2833_pt = pt;
            }
            if let Some(pt) = cfg.ilbc_payload_type {
                state.codecx.set_local_ilbc_pt(pt);
            }
            if let Some(pwd) = &cfg.icepwd {
                state.remote_icepwd = pwd.clone();
            }

            // DTLS: if the remote config includes DTLS setup, start the handshake.
            // The cert is the process-lifetime one whose fingerprint is what
            // the peer was promised via SDP — see `dtls::get_certificate`.
            if let Some(ref dtls) = cfg.dtls {
                // If a previous Remote with DTLS already started a handshake,
                // abort it before kicking off a new one — otherwise the old
                // task would race against the new one for the same socket.
                if let Some(abort) = state.dtls_handshake_abort.take() {
                    abort.abort();
                }
                let (dtls_tx, dtls_rx) = mpsc::channel::<Vec<u8>>(64);
                *state.dtls_inbound_tx.lock() = Some(dtls_tx);
                let h = super::dtls_session::spawn_handshake(
                    dtls.setup,
                    state.local_addr,
                    state.rtp_sock.clone(),
                    dtls_rx,
                    crate::dtls::get_certificate(),
                    state.remote_addr.clone(),
                );
                state.dtls_result_rx = Some(h.result_rx);
                state.dtls_handshake_abort = Some(h.abort);
            }

            state.remote = Some(cfg);
            state.remote_confirmed = true;
            let _ = ack.send(());
            LocalOutcome::Continue
        }

        Command::Play { cfg, ack } => {
            if subs.player.is_some() {
                subs.player = None;
                subs.bargein = None;
                events.post(Event::Play {
                    state: PlayState::End,
                    reason: Some("replaced".into()),
                });
            }
            // A bare `play` supersedes any `playrecord` pending recorder.
            // No Recording event was emitted for it yet (activation hasn't
            // fired), so no matching Finished is owed.
            subs.pending_recorder = None;
            subs.prebuffer.clear();
            subs.player = Some(Player::new(cfg));
            events.post(Event::Play {
                state: PlayState::Start,
                reason: Some("new".into()),
            });
            let _ = ack.send(());
            LocalOutcome::Continue
        }

        Command::Record { cfg, ack } => {
            // Record at the channel's native rate (16k for G.722, else 8k).
            let mut cfg = cfg;
            cfg.sample_rate = state.codecx.native_samplerate();
            // A bare `record` at a different path leaves a pending recorder
            // dangling — clear it so the play-end activation doesn't later
            // open it alongside this fresh one.
            subs.pending_recorder = None;
            subs.prebuffer.clear();
            let file_str = cfg.file.to_string_lossy().into_owned();
            let is_gated = cfg.start_above_power.is_some();
            // If an existing recorder at the same path is paused, resume it
            // instead of replacing it — preserves the WAV so both segments
            // end up in one file.
            if let Some(rec) = subs
                .recorders
                .iter_mut()
                .find(|r| r.file() == cfg.file && r.state() == RecorderState::Paused)
            {
                rec.resume();
                events.post(Event::Record {
                    state: RecordState::Recording,
                    reason: None,
                    file: Some(file_str),
                    filesize: None,
                });
            } else {
                match Recorder::open(cfg.clone()).await {
                    Ok(rec) => {
                        if let Some(idx) =
                            subs.recorders.iter().position(|r| r.file() == rec.file())
                        {
                            let mut old = subs.recorders.remove(idx);
                            let size = old.file_size();
                            old.close(FinishReason::ChannelClosed);
                            events.post(Event::Record {
                                state: RecordState::Finished,
                                reason: Some("channelclosed".into()),
                                file: Some(file_str.clone()),
                                filesize: Some(size),
                            });
                        }
                        subs.recorders.push(rec);
                        if !is_gated {
                            events.post(Event::Record {
                                state: RecordState::Recording,
                                reason: None,
                                file: Some(file_str),
                                filesize: None,
                            });
                        }
                    }
                    Err(e) => {
                        events.post(Event::Record {
                            state: RecordState::Finished,
                            reason: Some(format!("open-failed: {e}")),
                            file: Some(file_str),
                            filesize: None,
                        });
                    }
                }
            }
            let _ = ack.send(());
            LocalOutcome::Continue
        }

        Command::CreateReadStream { id, cfg, sender } => {
            subs.readers
                .push(super::audio_reader::AudioReader::new(id, cfg, sender));
            LocalOutcome::Continue
        }

        Command::DestroyReadStream { id } => {
            subs.readers.retain(|r| r.id() != id);
            LocalOutcome::Continue
        }

        Command::CreateWriteStream { id, cfg, receiver } => {
            // A writer supersedes any active player — they share the
            // outbound-source slot. Emit `play/end reason=replaced` to
            // match what `Command::Play` does when it replaces another
            // player.
            if subs.player.is_some() {
                subs.player = None;
                subs.bargein = None;
                events.post(Event::Play {
                    state: PlayState::End,
                    reason: Some("replaced".into()),
                });
            }
            subs.pending_recorder = None;
            subs.prebuffer.clear();
            subs.writer = Some(super::audio_writer::AudioWriter::new(id, cfg, receiver));
            LocalOutcome::Continue
        }

        Command::DestroyWriteStream { id } => {
            if let Some(w) = subs.writer.as_ref() {
                if w.id() == id {
                    subs.writer = None;
                }
            }
            LocalOutcome::Continue
        }

        Command::RecordFinish { file } => {
            if let Some(idx) = subs.recorders.iter().position(|r| r.file() == file) {
                let mut rec = subs.recorders.remove(idx);
                let file_str = rec.file().to_string_lossy().into_owned();
                let size = rec.file_size();
                rec.close(FinishReason::Requested);
                events.post(Event::Record {
                    state: RecordState::Finished,
                    reason: Some("requested".into()),
                    file: Some(file_str),
                    filesize: Some(size),
                });
            }
            LocalOutcome::Continue
        }

        Command::RecordSetPaused { file, paused } => {
            if let Some(rec) = subs.recorders.iter_mut().find(|r| r.file() == file) {
                if paused {
                    rec.pause();
                } else {
                    rec.resume();
                }
            }
            LocalOutcome::Continue
        }

        Command::PlayRecord { cfg, ack } => {
            if subs.player.is_some() {
                subs.player = None;
                subs.bargein = None;
                events.post(Event::Play {
                    state: PlayState::End,
                    reason: Some("replaced".into()),
                });
            }
            // Supersede any previously-queued pending recorder — its file was
            // never opened, so no Record event is owed (matches C++: no file
            // created, no emit). The pre-buffer is cleared so the new
            // recording starts fresh when play ends.
            subs.pending_recorder = None;
            subs.prebuffer.clear();

            subs.player = Some(Player::new(cfg.player));
            events.post(Event::Play {
                state: PlayState::Start,
                reason: Some("new".into()),
            });

            if cfg.interrupt {
                if let Some(threshold) = cfg.bargein_power {
                    let mut ma = crate::firfilter::MaFilter::new();
                    if let Some(n) = cfg.bargein_packets {
                        ma.reset(n as usize);
                    }
                    subs.bargein = Some(BargeInState {
                        power_threshold: threshold,
                        power_ma: ma,
                    });
                }
            }

            // Queue the recorder — actual `Recorder::open` is deferred until
            // play-end (or barge-in). The pre-buffer accumulates inbound
            // samples while the player runs; on activation it is flushed
            // via `write_raw`. C++ `pendingrecorder` equivalent.
            // Record at the channel's native rate (16k for G.722, else 8k).
            let mut recorder_cfg = cfg.recorder;
            recorder_cfg.sample_rate = state.codecx.native_samplerate();
            let file_str = recorder_cfg.file.to_string_lossy().into_owned();
            subs.pending_recorder = Some(PendingRecorder {
                cfg: recorder_cfg,
                file_str,
            });

            let _ = ack.send(());
            LocalOutcome::Continue
        }

        Command::Dtmf { digits } => {
            subs.dtmf_send.enqueue(&digits);
            LocalOutcome::Continue
        }
    }
}

#[cfg(test)]
mod tests {
    use super::super::commands::Direction;
    use super::*;
    use std::sync::Mutex;
    use tokio::sync::mpsc as tmpsc;

    struct TestSink {
        tx: tmpsc::UnboundedSender<Event>,
    }
    impl EventSink for TestSink {
        fn post(&self, ev: Event) {
            let _ = self.tx.send(ev);
        }
    }

    struct CountingSink {
        count: Mutex<Vec<Event>>,
    }
    impl EventSink for CountingSink {
        fn post(&self, ev: Event) {
            self.count.lock().unwrap().push(ev);
        }
    }

    #[tokio::test]
    async fn close_command_exits_loop_and_posts_close_event() {
        let (tx, mut rx) = tmpsc::unbounded_channel();
        let sink = Arc::new(TestSink { tx });

        let handle = spawn(SpawnConfig {
            id: 42,
            bind_addr: "127.0.0.1:0".parse().unwrap(),
            ssrc: 1,
            events: sink,
            port_reservation: None,
            local_icepwd: String::new(),
        })
        .await
        .unwrap();

        handle.close("test").await;

        let ev = tokio::time::timeout(Duration::from_millis(500), rx.recv())
            .await
            .expect("no close event")
            .expect("stream ended");
        match ev {
            Event::Close { reason, .. } => assert_eq!(reason, "test"),
            other => panic!("unexpected event: {:?}", other),
        }
    }

    #[tokio::test]
    async fn direction_command_mutates_state_via_tick_observable() {
        let sink = Arc::new(CountingSink {
            count: Mutex::new(Vec::new()),
        });
        let handle = spawn(SpawnConfig {
            id: 1,
            bind_addr: "127.0.0.1:0".parse().unwrap(),
            ssrc: 1,
            events: sink.clone(),
            port_reservation: None,
            local_icepwd: String::new(),
        })
        .await
        .unwrap();

        handle
            .direction(Direction {
                send: false,
                recv: false,
            })
            .await;
        handle.echo(true).await;
        handle.dtmf("1#").await;

        tokio::time::sleep(Duration::from_millis(80)).await;

        handle.close("ok").await;
        tokio::time::sleep(Duration::from_millis(80)).await;

        let events = sink.count.lock().unwrap();
        assert!(events.iter().any(|e| matches!(e, Event::Close { .. })));
    }

    #[tokio::test]
    async fn play_command_emits_play_start_event_and_acks() {
        let (tx, mut rx) = tmpsc::unbounded_channel();
        let sink = Arc::new(TestSink { tx });

        let handle = spawn(SpawnConfig {
            id: 1,
            bind_addr: "127.0.0.1:0".parse().unwrap(),
            ssrc: 1,
            events: sink,
            port_reservation: None,
            local_icepwd: String::new(),
        })
        .await
        .unwrap();

        handle
            .play(crate::channel::player::SoundSoupSpec {
                files: vec![],
                overall_loops: None,
                interrupt: false,
            })
            .await
            .unwrap();

        let mut saw_play = false;
        while let Ok(Some(ev)) = tokio::time::timeout(Duration::from_millis(100), rx.recv()).await {
            if matches!(
                ev,
                Event::Play {
                    state: PlayState::Start,
                    ..
                }
            ) {
                saw_play = true;
                break;
            }
        }
        assert!(saw_play, "expected Play::Start event");

        handle.close("done").await;
    }

    fn test_recorder_cfg(name: &str) -> RecorderConfig {
        let p = std::env::temp_dir().join(name);
        let _ = std::fs::remove_file(&p);
        RecorderConfig {
            file: p,
            num_channels: 1,
            sample_rate: 8000,
            max_duration_ms: None,
            start_above_power: None,
            finish_below_power: None,
            max_since_start_power: None,
            min_duration_ms: None,
            power_averaging_packets: None,
        }
    }

    #[tokio::test]
    async fn activate_pending_recorder_noop_when_none() {
        let mut events: Vec<Event> = Vec::new();
        let mut subs = Subsystems::default();
        let activated = activate_pending_recorder(&mut subs, &mut events).await;
        assert!(!activated);
        assert!(subs.recorders.is_empty());
        assert!(events.is_empty());
    }

    #[tokio::test]
    async fn activate_pending_recorder_flushes_prebuffer_and_emits_recording() {
        let cfg = test_recorder_cfg("actor_activate_flush.wav");
        let path = cfg.file.clone();
        let file_str = path.to_string_lossy().into_owned();

        let mut events: Vec<Event> = Vec::new();
        let mut subs = Subsystems {
            pending_recorder: Some(PendingRecorder {
                cfg,
                file_str: file_str.clone(),
            }),
            ..Default::default()
        };
        subs.prebuffer.extend(vec![500i16; 800]);

        let activated = activate_pending_recorder(&mut subs, &mut events).await;
        assert!(activated);
        assert!(subs.pending_recorder.is_none());
        assert!(subs.prebuffer.is_empty());
        assert_eq!(subs.recorders.len(), 1);

        // Recording event fires (recorder is not gated).
        assert!(
            events.iter().any(|e| matches!(e,
                Event::Record { state: RecordState::Recording, file: Some(f), .. } if f == &file_str
            )),
            "expected Record::Recording event, got {:?}",
            events
        );

        // Close and verify the file has the 800 pre-buffer samples.
        let mut rec = subs.recorders.pop().unwrap();
        rec.close(FinishReason::Completed);
        let info = crate::soundfile::js_info(file_str.clone()).unwrap();
        assert_eq!(info.subchunksize, 1600);
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn activate_pending_recorder_stereo_interleaves_prebuffer() {
        let mut cfg = test_recorder_cfg("actor_activate_stereo.wav");
        cfg.num_channels = 2;
        let path = cfg.file.clone();
        let file_str = path.to_string_lossy().into_owned();

        let mut events: Vec<Event> = Vec::new();
        let mut subs = Subsystems {
            pending_recorder: Some(PendingRecorder {
                cfg,
                file_str: file_str.clone(),
            }),
            ..Default::default()
        };
        subs.prebuffer.extend(vec![100i16; 400]);

        assert!(activate_pending_recorder(&mut subs, &mut events).await);
        let mut rec = subs.recorders.pop().unwrap();
        rec.close(FinishReason::Completed);

        let info = crate::soundfile::js_info(file_str).unwrap();
        // 400 mono samples → 800 stereo samples × 2 bytes = 1600 bytes PCM.
        assert_eq!(info.subchunksize, 1600);
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn activate_pending_recorder_gated_suppresses_recording_event() {
        let mut cfg = test_recorder_cfg("actor_activate_gated.wav");
        cfg.start_above_power = Some(50);
        let path = cfg.file.clone();
        let file_str = path.to_string_lossy().into_owned();

        let mut events: Vec<Event> = Vec::new();
        let mut subs = Subsystems {
            pending_recorder: Some(PendingRecorder {
                cfg,
                file_str: file_str.clone(),
            }),
            ..Default::default()
        };

        assert!(activate_pending_recorder(&mut subs, &mut events).await);
        // Gated recorders don't emit Recording until the gate opens; the
        // helper itself should not queue one at activation time.
        assert!(!events.iter().any(|e| matches!(
            e,
            Event::Record {
                state: RecordState::Recording,
                ..
            }
        )));
        let mut rec = subs.recorders.pop().unwrap();
        rec.close(FinishReason::Completed);
        let _ = std::fs::remove_file(&path);
    }

    /// Regression test for the "maxduration timeout during prompt" bug:
    /// a `playrecord` must not open its recorder at command-accept time.
    /// Previously `Recorder::open` ran inside the PlayRecord handler, so a
    /// 10 s maxduration started counting immediately — while the prompt
    /// was still playing — and fired `finished.timeout` before the caller
    /// could speak. The fix defers `Recorder::open` to play-end via
    /// `pending_recorder`.
    ///
    /// Here we issue a playrecord with an empty-soup player (so it finishes
    /// on the first tick read) and a short maxduration. The expected event
    /// order is: play/start → play/end → record/recording. No
    /// record/finished.timeout should fire before activation.
    #[tokio::test]
    async fn playrecord_defers_recorder_open_until_play_end() {
        let (tx, mut rx) = tmpsc::unbounded_channel();
        let sink = Arc::new(TestSink { tx });

        let handle = spawn(SpawnConfig {
            id: 7,
            bind_addr: "127.0.0.1:0".parse().unwrap(),
            ssrc: 1,
            events: sink,
            port_reservation: None,
            local_icepwd: String::new(),
        })
        .await
        .unwrap();

        let rec_cfg = test_recorder_cfg("actor_playrecord_defer.wav");
        let rec_path = rec_cfg.file.clone();
        let rec_path_str = rec_path.to_string_lossy().into_owned();

        let cfg = crate::channel::commands::PlayRecordConfig {
            player: crate::channel::player::SoundSoupSpec {
                files: vec![],
                overall_loops: None,
                interrupt: false,
            },
            recorder: RecorderConfig {
                max_duration_ms: Some(100),
                ..rec_cfg
            },
            interrupt: false,
            bargein_power: None,
            bargein_packets: None,
        };
        handle.play_record(cfg).await.unwrap();

        // Collect events for ~250 ms — enough for at least one tick + the
        // empty player to finish + activation to run.
        let mut observed: Vec<Event> = Vec::new();
        let deadline = tokio::time::Instant::now() + Duration::from_millis(250);
        while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, rx.recv()).await {
            observed.push(ev);
        }

        let play_start_idx = observed
            .iter()
            .position(|e| {
                matches!(
                    e,
                    Event::Play {
                        state: PlayState::Start,
                        ..
                    }
                )
            })
            .expect("expected Play::Start");
        let play_end_idx = observed
            .iter()
            .position(|e| {
                matches!(
                    e,
                    Event::Play {
                        state: PlayState::End,
                        ..
                    }
                )
            })
            .expect("expected Play::End after empty soup");
        let rec_rec_idx = observed
            .iter()
            .position(|e| {
                matches!(e, Event::Record { state: RecordState::Recording, file: Some(f), .. }
                      if f == &rec_path_str)
            })
            .expect("expected Record::Recording after play ends");

        assert!(
            play_start_idx < play_end_idx,
            "Play::Start must precede Play::End: {:?}",
            observed
        );
        assert!(
            play_end_idx < rec_rec_idx,
            "Play::End must precede Record::Recording (activation order): {:?}",
            observed
        );

        // Crucially: no timeout finish event should appear before the
        // Recording event. (The regression was maxduration firing during
        // the play phase, which would emit Record::Finished first.)
        let early_finish = observed[..rec_rec_idx].iter().any(|e| {
            matches!(
                e,
                Event::Record {
                    state: RecordState::Finished,
                    ..
                }
            )
        });
        assert!(
            !early_finish,
            "no Record::Finished should fire before Record::Recording: {:?}",
            observed
        );

        handle.close("done").await;
        let _ = std::fs::remove_file(&rec_path);
    }
}