pub mod messages;
pub mod jitter;

use crate::audio::{AudioBackend, pipewire_impl::PipeWireBackend};
use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES};
use crate::network::{
    NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket,
    iroh_impl::IrohTransport,
    gossip::IrohGossipState,
};
use crate::core::messages::{CoreCommand, UiEvent};

use crate::config::NetworkMode;
use iroh::{Endpoint, EndpointId, RelayMode, endpoint::presets, protocol::Router};
use iroh_gossip::net::Gossip;
use tokio::sync::{mpsc, Mutex};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Duration;

pub struct CoreController {
    cmd_tx: mpsc::Sender<CoreCommand>,
}

impl CoreController {
    pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self {
        let (cmd_tx, cmd_rx) = mpsc::channel(100);

        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime");
            rt.block_on(async move {
                crate::log_msg("Starting core network loop in dedicated Tokio runtime");
                if let Err(e) = run_core_loop(cmd_rx, ui_tx).await {
                    crate::log_msg(&format!("App core loop failed: {:?}", e));
                }
            });
        });

        Self { cmd_tx }
    }

    pub fn send(&self, cmd: CoreCommand) -> Result<(), mpsc::error::TrySendError<CoreCommand>> {
        self.cmd_tx.try_send(cmd)
    }
}

/// How long a peer may stay "reconnecting" after a transient drop before we give
/// up and evict it. Comfortably past the QUIC idle timeout (~30s) so a genuine
/// reconnect has time to complete, but short enough that a crashed/departed peer
/// clears from the room promptly.
const RECONNECT_GRACE: Duration = Duration::from_secs(45);

/// Per-peer reconnect grace timers (see [`RECONNECT_GRACE`]). Shared between the
/// room-event task (which arms one on a transient drop and cancels it on a
/// gossip rejoin) and the conn-event task (which cancels it when the audio link
/// actually comes back).
type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHandle<()>>>>;

/// Peers we've completed at least one audio link with. Lets the conn-event task
/// tell a genuine reconnect (arm an eviction timer) from a first-ever dial (don't).
/// Scrubbed whenever a peer is evicted or leaves so a later rejoin starts clean.
type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;

/// Cancel and forget a peer's pending grace timer, if any. No-op if none is armed.
fn cancel_grace_timer(timers: &GraceTimers, peer_id: &EndpointId) {
    if let Some(handle) = timers.lock().unwrap().remove(peer_id) {
        handle.abort();
    }
}

/// Arm a per-peer reconnect grace timer that evicts the peer if its link hasn't
/// recovered within [`RECONNECT_GRACE`]. No-op if a timer is already pending for
/// the peer, so the earliest drop notice — whether the gossip `PeerConnectionLost`
/// or the transport `Connecting` — sets one hard deadline, rather than a flapping
/// link repeatedly resetting the clock and dodging eviction forever. On firing it
/// also scrubs the peer from `seen_connected` so a later rejoin isn't treated as a
/// reconnect on its initial dial.
fn arm_grace_timer(
    timers: &GraceTimers,
    seen_connected: &SeenConnected,
    transport: &Arc<IrohTransport>,
    jitter: &Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
    ui_tx: &mpsc::Sender<UiEvent>,
    grace: Duration,
    peer_id: EndpointId,
) {
    let mut timers_guard = timers.lock().unwrap();
    if timers_guard.contains_key(&peer_id) {
        return;
    }
    let transport_evict = transport.clone();
    let jitter_evict = jitter.clone();
    let ui_evict = ui_tx.clone();
    let timers_evict = timers.clone();
    let seen_evict = seen_connected.clone();
    let handle = tokio::spawn(async move {
        tokio::time::sleep(grace).await;
        crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
        transport_evict.disconnect_peer(peer_id).await;
        jitter_evict.lock().await.remove(&peer_id);
        // Scrub our internal state *before* announcing the eviction, so anything
        // that observes `PeerConnectionFailed` (or a rejoin racing it) sees a clean
        // slate — a later dial for this identity is then a fresh first-dial, not a
        // reconnect.
        timers_evict.lock().unwrap().remove(&peer_id);
        seen_evict.lock().unwrap().remove(&peer_id);
        let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
    });
    timers_guard.insert(peer_id, handle);
}

/// Scale a frame in place by a per-peer volume factor, saturating to the i16
/// range. A volume within `f32::EPSILON` of 1.0 is treated as unity and skipped,
/// matching the mixer hot path that avoids touching unmodified frames.
fn apply_volume(frame: &mut [i16], vol: f32) {
    if (vol - 1.0).abs() <= f32::EPSILON {
        return;
    }
    for sample in frame.iter_mut() {
        *sample = (*sample as f32 * vol).clamp(i16::MIN as f32, i16::MAX as f32) as i16;
    }
}

/// Normalized RMS level of a frame in `[0.0, 1.0]` (32768 = full scale), for the
/// UI level meter. An empty frame reads as 0.0.
fn frame_level(frame: &[i16]) -> f32 {
    let sum_sq: f32 = frame.iter().map(|&x| (x as f32).powi(2)).sum();
    let rms = (sum_sq / frame.len().max(1) as f32).sqrt();
    (rms / 32768.0).clamp(0.0, 1.0)
}

/// Peak-hold every this many captured samples (~100ms @ 48kHz) before emitting a
/// [`UiEvent::MicLevel`], so the meter doesn't flood the UI runtime at frame rate.
const MIC_LEVEL_REPORT_SAMPLES: usize = 4800;

/// A standalone, capture-only mic monitor for gate calibration outside a call.
/// Owns the worker thread that reads raw PCM and reports its level; the PipeWire
/// capture stream itself lives in the shared backend. Tear down by stopping the
/// backend's capture (which closes the channel) and joining this thread.
struct MicMonitor {
    thread: std::thread::JoinHandle<()>,
}

/// Drains a capture channel, reporting the raw (un-gated) mic level to the UI.
/// Returns when the channel closes (i.e. the backend's capture stream stopped).
fn run_mic_monitor(rx: std::sync::mpsc::Receiver<Vec<i16>>, ui_tx: mpsc::Sender<UiEvent>) {
    let mut peak = 0.0f32;
    let mut acc = 0usize;
    while let Ok(pcm) = rx.recv() {
        peak = peak.max(frame_level(&pcm));
        acc += pcm.len();
        if acc >= MIC_LEVEL_REPORT_SAMPLES {
            // Drop on a full channel — a stale meter reading is harmless.
            let _ = ui_tx.try_send(UiEvent::MicLevel(peak));
            peak = 0.0;
            acc = 0;
        }
    }
    // Channel closed: the monitor was stopped. Snap the meter back to zero.
    let _ = ui_tx.try_send(UiEvent::MicLevel(0.0));
}

/// Stops a standalone mic monitor if one is running. MUST NOT be called while a
/// room session is active — `backend.stop()` would also tear down the call's
/// capture/playback. Monitor and session are mutually exclusive by construction.
fn stop_mic_monitor(backend: &PipeWireBackend, monitor: Option<MicMonitor>) {
    if let Some(m) = monitor {
        let _ = backend.stop();
        let _ = m.thread.join();
    }
}

/// Sum per-peer frames sample-by-sample into one `frame_len`-sample output,
/// saturating each summed sample to the i16 range so a loud mix clips rather than
/// wrapping. Peers shorter than `frame_len` contribute 0 past their end; an empty
/// peer set yields a silent frame.
fn mix_frames(peer_frames: &[Vec<i16>], frame_len: usize) -> Vec<i16> {
    let mut mixed = vec![0i16; frame_len];
    if peer_frames.is_empty() {
        return mixed;
    }
    for (i, out) in mixed.iter_mut().enumerate() {
        let sum: i32 = peer_frames
            .iter()
            .map(|f| f.get(i).copied().unwrap_or(0) as i32)
            .sum();
        *out = sum.clamp(i16::MIN as i32, i16::MAX as i32) as i16;
    }
    mixed
}

/// Handles the transport's per-peer link-state stream (`ConnEvent`): arms/cancels
/// reconnect grace timers, tracks which peers we've linked with, and forwards
/// link state to the UI. Pulled out of the conn-event task as a unit so the
/// reconnect-eviction behavior can be tested without standing up a full session.
/// Exposed (with a `grace` override) for that purpose; not part of the public API.
pub struct ConnEventHandler {
    ui_tx: mpsc::Sender<UiEvent>,
    grace_timers: GraceTimers,
    seen_connected: SeenConnected,
    transport: Arc<IrohTransport>,
    jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
    grace: Duration,
}

impl ConnEventHandler {
    pub fn new(
        ui_tx: mpsc::Sender<UiEvent>,
        grace_timers: GraceTimers,
        seen_connected: SeenConnected,
        transport: Arc<IrohTransport>,
        jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
    ) -> Self {
        Self {
            ui_tx,
            grace_timers,
            seen_connected,
            transport,
            jitter,
            grace: RECONNECT_GRACE,
        }
    }

    /// Override the eviction grace window. For tests that can't wait 45s.
    pub fn with_grace(mut self, grace: Duration) -> Self {
        self.grace = grace;
        self
    }

    pub async fn handle(&self, event: ConnEvent) {
        match event {
            ConnEvent::Connecting(id) => {
                // A reconnect (we've linked with this peer before): arm an eviction
                // timer so a peer that never comes back is cleared even when gossip
                // doesn't re-report the drop — the transport reliably re-emits this
                // on every outage, gossip's NeighborDown does not. A first-ever dial
                // (not yet in seen_connected) gets no timer; Connected cancels it on
                // recovery.
                if self.seen_connected.lock().unwrap().contains(&id) {
                    arm_grace_timer(
                        &self.grace_timers,
                        &self.seen_connected,
                        &self.transport,
                        &self.jitter,
                        &self.ui_tx,
                        self.grace,
                        id,
                    );
                }
                let _ = self.ui_tx.send(UiEvent::PeerConnecting { id }).await;
            }
            ConnEvent::Connected(id) => {
                // The audio link came back — the peer recovered within the grace
                // window, so cancel its eviction.
                cancel_grace_timer(&self.grace_timers, &id);
                self.seen_connected.lock().unwrap().insert(id);
                let _ = self.ui_tx.send(UiEvent::PeerConnected { id }).await;
            }
            ConnEvent::Left(id) => {
                // The peer closed its link gracefully (intentional leave) — evict
                // immediately, like a PeerLeft, instead of leaving it "reconnecting"
                // until the grace timer or the slow gossip Leave.
                cancel_grace_timer(&self.grace_timers, &id);
                self.seen_connected.lock().unwrap().remove(&id);
                self.transport.disconnect_peer(id).await;
                self.jitter.lock().await.remove(&id);
                let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
            }
        }
    }
}

struct ActiveSession {
    endpoint: Endpoint,
    router: Router,
    room_state: Arc<IrohGossipState>,
    capture_thread: std::thread::JoinHandle<()>,
    datagram_task: tokio::task::JoinHandle<()>,
    mixer_task: tokio::task::JoinHandle<()>,
    event_task: tokio::task::JoinHandle<()>,
    conn_event_task: tokio::task::JoinHandle<()>,
    grace_timers: GraceTimers,
    transport: Arc<IrohTransport>,
    /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop.
    echo_cancel: Option<crate::audio::echo_cancel::EchoCancelGuard>,
}

impl ActiveSession {
    async fn shutdown(self, audio_backend: Arc<PipeWireBackend>) {
        crate::log_msg("ActiveSession::shutdown started");
        self.datagram_task.abort();
        self.mixer_task.abort();
        self.event_task.abort();
        self.conn_event_task.abort();
        // Abort any pending reconnect grace timers so they can't fire a stray
        // eviction (or touch a torn-down transport) after the session is gone.
        for (_, handle) in self.grace_timers.lock().unwrap().drain() {
            handle.abort();
        }
        crate::log_msg("Aborted tasks");

        let audio_backend_clone = audio_backend.clone();
        let _ = tokio::task::spawn_blocking(move || {
            crate::log_msg("Stopping audio backend...");
            let _ = audio_backend_clone.stop();
            crate::log_msg("Audio backend stopped");
        }).await;

        // Unload the echo-cancel module now that the audio streams releasing its
        // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.)
        drop(self.echo_cancel);

        crate::log_msg("Leaving room...");
        let _ = self.room_state.leave().await;
        // Close peer links with the graceful goodbye code so remotes evict us
        // promptly (not after the reconnect grace / slow gossip Leave), and stop
        // our supervisors so none redial the about-to-close endpoint.
        self.transport.leave().await;
        crate::log_msg("Room left");

        crate::log_msg("Shutting down router...");
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), self.router.shutdown()).await;
        crate::log_msg("Router shut down");

        crate::log_msg("Joining capture thread...");
        let _ = self.capture_thread.join();
        crate::log_msg("ActiveSession::shutdown complete");
    }
}

async fn run_core_loop(
    mut cmd_rx: mpsc::Receiver<CoreCommand>,
    ui_tx: mpsc::Sender<UiEvent>,
) -> Result<(), anyhow::Error> {
    let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
    let secret_key = iroh::SecretKey::generate();

    let audio_backend = Arc::new(PipeWireBackend::new());

    let is_muted = Arc::new(AtomicBool::new(false));
    let is_deafened = Arc::new(AtomicBool::new(false));
    let ptt_mode = Arc::new(AtomicBool::new(false));
    let ptt_active = Arc::new(AtomicBool::new(false));
    let noise_gate_threshold = Arc::new(std::sync::atomic::AtomicU32::new(0.01f32.to_bits()));
    let peer_volumes = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
    let mut current_name = "Anonymous".to_string();
    let mut network_mode = NetworkMode::default();

    let mut active_session: Option<ActiveSession> = None;
    // Standalone capture-only mic meter, live only when no session exists.
    let mut mic_monitor: Option<MicMonitor> = None;

    while let Some(cmd) = cmd_rx.recv().await {
        match cmd {
            CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation } => {
                current_name = name.clone();

                // Clean up any existing session
                if let Some(session) = active_session.take() {
                    crate::log_msg("Shutting down existing active session");
                    session.shutdown(audio_backend.clone()).await;
                }

                // Release a standalone mic monitor if running — it shares the
                // backend's single capture stream, so it must stop before the
                // call claims it. (Safe here: any session was just shut down.)
                stop_mic_monitor(&audio_backend, mic_monitor.take());

                // Build the endpoint per the configured relay/discovery posture.
                // All postures keep the in-memory address lookup (fed by tickets
                // and gossip); they differ in whether n0's relay and DNS presence
                // beacon are used. `Minimal` sets only the mandatory crypto
                // provider and deliberately omits the n0 DNS publish/resolve.
                let bind_result = match network_mode {
                    NetworkMode::N0Full => {
                        Endpoint::builder(presets::N0)
                            .secret_key(secret_key.clone())
                            .address_lookup(memory_lookup.clone())
                            .bind()
                            .await
                    }
                    NetworkMode::RelayNoDiscovery => {
                        Endpoint::builder(presets::Minimal)
                            .secret_key(secret_key.clone())
                            .relay_mode(RelayMode::Default)
                            .address_lookup(memory_lookup.clone())
                            .bind()
                            .await
                    }
                    NetworkMode::DirectOnly => {
                        Endpoint::builder(presets::Minimal)
                            .secret_key(secret_key.clone())
                            .relay_mode(RelayMode::Disabled)
                            .address_lookup(memory_lookup.clone())
                            .bind()
                            .await
                    }
                };
                let endpoint = match bind_result {
                    Ok(ep) => ep,
                    Err(e) => {
                        let _ = ui_tx.send(UiEvent::Error(format!("Failed to bind endpoint: {}", e))).await;
                        continue;
                    }
                };
                endpoint.online().await;

                // Determine target ticket
                let ticket_str = if ticket.trim().is_empty() || ticket == "create" {
                    let topic_id: [u8; 32] = rand::random();
                    let host_addr = endpoint.addr();
                    crate::log_msg(&format!("Creating room. host_addr={:?}, topic_id={:?}", host_addr, topic_id));
                    let ticket = PeerSpeakTicket { host_addr, topic_id };
                    ticket.to_string()
                } else {
                    let ticket_str = ticket.trim().to_string();
                    crate::log_msg(&format!("Joining room with existing ticket={}", ticket_str));
                    ticket_str
                };

                // Initialize Gossip and Transport
                let gossip = Gossip::builder().spawn(endpoint.clone());
                let (transport, audio_proto) = IrohTransport::new(endpoint.clone());
                let transport = Arc::new(transport);

                // Start Router
                let router = iroh::protocol::Router::builder(endpoint.clone())
                    .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
                    .accept(b"peerspeak-audio", audio_proto)
                    .spawn();

                let room_state = Arc::new(IrohGossipState::new(
                    endpoint.clone(),
                    gossip.clone(),
                    memory_lookup.clone(),
                ));

                let self_state = PeerState {
                    name: current_name.clone(),
                    is_muted: is_muted.load(Ordering::Relaxed),
                    addr: endpoint.addr(),
                };

                crate::log_msg(&format!("Attempting room_state.join with self_state={:?}", self_state));
                if let Err(e) = room_state.join(&ticket_str, self_state.clone()).await {
                    crate::log_msg(&format!("Error room_state.join failed: {:?}", e));
                    let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
                    let _ = router.shutdown().await;
                    continue;
                }
                crate::log_msg("Joined room successfully via room_state");

                // Setup raw audio channels
                let (capture_tx, capture_rx) = std::sync::mpsc::channel();
                let (playback_tx, playback_rx) = std::sync::mpsc::channel();

                // Echo cancellation: if enabled, load PipeWire's echo-cancel module
                // bound to the chosen real devices and route capture/playback
                // through its virtual nodes (the sink doubles as the AEC reference).
                // The guard unloads the module on drop — including the early-return
                // paths below, since it's a local until moved into the session. On
                // any failure, warn and fall back to the direct devices.
                let mut echo_cancel_guard = None;
                let (capture_target, playback_target) = if echo_cancellation {
                    match crate::audio::echo_cancel::enable(
                        input_device.as_deref(),
                        output_device.as_deref(),
                    ) {
                        Ok(guard) => {
                            echo_cancel_guard = Some(guard);
                            crate::log_msg("Echo cancellation enabled");
                            (
                                Some(crate::audio::echo_cancel::EC_SOURCE.to_string()),
                                Some(crate::audio::echo_cancel::EC_SINK.to_string()),
                            )
                        }
                        Err(e) => {
                            crate::log_msg(&format!(
                                "Echo cancellation unavailable, using direct devices: {e}"
                            ));
                            let _ = ui_tx
                                .send(UiEvent::Error(format!("Echo cancellation unavailable: {e}")))
                                .await;
                            (input_device.clone(), output_device.clone())
                        }
                    }
                } else {
                    (input_device.clone(), output_device.clone())
                };

                if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) {
                    let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
                    let _ = room_state.leave().await;
                    let _ = router.shutdown().await;
                    continue;
                }

                // Shared gauge: PipeWire publishes the playback ring's live depth
                // here (drain side + fill side); the mixer reads it to pace
                // production to the hardware clock instead of a fixed timer.
                let ring_fill = Arc::new(AtomicUsize::new(0));
                if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) {
                    let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await;
                    let _ = audio_backend.stop();
                    let _ = room_state.leave().await;
                    let _ = router.shutdown().await;
                    continue;
                }

                let jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>> = Arc::new(Mutex::new(HashMap::new()));

                // 1. Capture & encoding thread
                let is_muted_clone = is_muted.clone();
                let ptt_mode_clone = ptt_mode.clone();
                let ptt_active_clone = ptt_active.clone();
                let noise_gate_threshold_clone = noise_gate_threshold.clone();
                let transport_clone = transport.clone();
                let ui_tx_capture = ui_tx.clone();

                let capture_thread = std::thread::spawn(move || {
                    use opus::{Channels, Application};
                    let mut encoder = match OpusEncoder::new(48000, Channels::Mono, Application::Voip) {
                        Ok(enc) => enc,
                        Err(e) => {
                            crate::log_msg(&format!("Capture thread error: {:?}", e));
                            return;
                        }
                    };
                    // Per-sender packet sequence number, prepended to every frame so
                    // receivers can reorder and conceal loss. Wraps after ~years.
                    let mut seq: u32 = 0;
                    // Smooth noise gate (hysteresis + attack/release + hangover),
                    // carrying envelope state across frames. The live slider value
                    // is read per frame so changes apply immediately.
                    let mut gate = crate::audio::gate::NoiseGate::new(48000);
                    // Peak-held raw mic level for the settings meter, reported
                    // pre-gate/pre-mute so calibration reflects the true input.
                    let mut mic_peak = 0.0f32;
                    let mut mic_acc = 0usize;

                    while let Ok(mut pcm) = capture_rx.recv() {
                        mic_peak = mic_peak.max(frame_level(&pcm));
                        mic_acc += pcm.len();
                        if mic_acc >= MIC_LEVEL_REPORT_SAMPLES {
                            let _ = ui_tx_capture.try_send(UiEvent::MicLevel(mic_peak));
                            mic_peak = 0.0;
                            mic_acc = 0;
                        }

                        if is_muted_clone.load(Ordering::Relaxed) {
                            continue;
                        }
                        if ptt_mode_clone.load(Ordering::Relaxed) && !ptt_active_clone.load(Ordering::Relaxed) {
                            continue;
                        }

                        let ng_bits = noise_gate_threshold_clone.load(Ordering::Relaxed);
                        let ng_thresh = f32::from_bits(ng_bits);
                        // Apply the gate in place; skip transmitting a fully-closed
                        // frame so we don't send pure silence (the receiver's jitter
                        // buffer conceals the gap).
                        if !gate.process(&mut pcm, ng_thresh) {
                            continue;
                        }

                        if let Ok(encoded) = encoder.encode(&pcm) {
                            // Frame on the wire: [seq: u32 LE][opus payload].
                            let mut packet = Vec::with_capacity(4 + encoded.len());
                            packet.extend_from_slice(&seq.to_le_bytes());
                            packet.extend_from_slice(&encoded);
                            seq = seq.wrapping_add(1);
                            transport_clone.broadcast(bytes::Bytes::from(packet));
                        }
                    }
                });

                // 2. Receiver task: parse the sequence header and hand each packet
                //    to that peer's jitter buffer. Decoding happens later, on the
                //    playout side, so loss can be concealed at the right moment.
                let transport_recv = transport.clone();
                let jitter_recv = jitter.clone();
                let datagram_task = tokio::spawn(async move {
                    let mut datagram_rx = match transport_recv.receive_datagrams().await {
                        Ok(rx) => rx,
                        Err(e) => {
                            crate::log_msg(&format!("Receiver task error: {:?}", e));
                            return;
                        }
                    };

                    while let Some((from_peer, bytes)) = datagram_rx.recv().await {
                        if bytes.len() < 4 {
                            continue; // malformed: missing sequence header
                        }
                        let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
                        let payload = bytes[4..].to_vec();

                        let mut guard = jitter_recv.lock().await;
                        let buffer = match guard.entry(from_peer) {
                            std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
                            std::collections::hash_map::Entry::Vacant(entry) => {
                                match JitterBuffer::new() {
                                    Ok(jb) => entry.insert(jb),
                                    Err(e) => {
                                        crate::log_msg(&format!("Failed to init jitter buffer for {:?}: {:?}", from_peer, e));
                                        continue;
                                    }
                                }
                            }
                        };
                        buffer.insert(seq, payload);
                    }
                });

                // 3. Mixing & level extraction loop task. Production is paced by
                //    the playback ring's fill level (the PipeWire hardware clock),
                //    NOT a fixed software timer: we produce a 20ms frame only when
                //    the ring is below its target depth, so the long-run mix rate
                //    auto-matches the device drain rate and the producer/consumer
                //    beat (which otherwise churns ~20% of audio) disappears. Each
                //    produced frame pulls one concealed frame per peer from its
                //    jitter buffer, applies per-peer volume, and sums.
                let jitter_mixer = jitter.clone();
                let is_deafened_clone = is_deafened.clone();
                let peer_volumes_mixer = peer_volumes.clone();
                let ui_tx_mixer = ui_tx.clone();
                let ring_fill_mixer = ring_fill.clone();
                let mixer_task = tokio::spawn(async move {
                    // When the ring is at/above target we have nothing to do; nap
                    // briefly and re-check. Short enough (relative to the ~60ms
                    // target and ~21ms device quantum) that we always refill well
                    // before the ring can run dry.
                    const IDLE_NAP: Duration = Duration::from_millis(2);

                    // Pushing a level event per frame floods the UI runtime at
                    // ~50/sec. We peak-hold per-peer levels across this many
                    // produced frames and emit once per window (~10/sec) —
                    // peak-hold so a brief transient still lights the indicator.
                    const LEVEL_EMIT_FRAMES: u32 = 5;
                    let mut level_peaks: HashMap<EndpointId, f32> = HashMap::new();
                    let mut frames_since_emit: u32 = 0;

                    loop {
                        // Pace to the hardware clock: only produce while the ring
                        // is draining below target. Otherwise yield and re-check.
                        if ring_fill_mixer.load(Ordering::Relaxed) >= crate::audio::PLAYBACK_TARGET_SAMPLES {
                            tokio::time::sleep(IDLE_NAP).await;
                            continue;
                        }

                        let current_volumes = peer_volumes_mixer.lock().await.clone();
                        let mut peer_frames = Vec::new();

                        {
                            let mut guard = jitter_mixer.lock().await;
                            for (&peer_id, buffer) in guard.iter_mut() {
                                // `None` means idle/buffering: contribute nothing,
                                // but keep a (zero) entry so the UI sees it idle.
                                let Some(mut frame) = buffer.pop_frame() else {
                                    level_peaks.entry(peer_id).or_insert(0.0);
                                    continue;
                                };

                                let vol = current_volumes.get(&peer_id).copied().unwrap_or(1.0);
                                apply_volume(&mut frame, vol);

                                let peak = level_peaks.entry(peer_id).or_insert(0.0);
                                *peak = peak.max(frame_level(&frame));

                                peer_frames.push(frame);
                            }
                        }

                        let mixed = mix_frames(&peer_frames, FRAME_SAMPLES);

                        let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) {
                            vec![0i16; FRAME_SAMPLES]
                        } else {
                            mixed
                        };

                        if playback_tx.send(frame_to_send).is_err() {
                            break;
                        }

                        // Emit coalesced peaks once per window, then reset.
                        frames_since_emit += 1;
                        if frames_since_emit >= LEVEL_EMIT_FRAMES {
                            let levels: Vec<(EndpointId, f32)> = level_peaks.drain().collect();
                            let _ = ui_tx_mixer.send(UiEvent::AudioLevels(levels)).await;
                            frames_since_emit = 0;
                        }
                    }
                });

                // 4. Room event subscriber task
                let mut room_events = match room_state.subscribe_events().await {
                    Ok(rx) => rx,
                    Err(e) => {
                        let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe events: {}", e))).await;
                        continue;
                    }
                };
                let ui_tx_events = ui_tx.clone();
                let jitter_events = jitter.clone();
                let transport_events = transport.clone();
                let grace_timers: GraceTimers = Arc::new(std::sync::Mutex::new(HashMap::new()));
                let grace_timers_events = grace_timers.clone();
                let seen_connected: SeenConnected = Arc::new(std::sync::Mutex::new(HashSet::new()));
                let seen_connected_events = seen_connected.clone();
                let event_task = tokio::spawn(async move {
                    while let Some(event) = room_events.recv().await {
                        match event {
                            RoomEvent::PeerJoined(peer_id, state) => {
                                // A (re)join means the peer is back — cancel any
                                // pending reconnect grace timer before re-adding it.
                                cancel_grace_timer(&grace_timers_events, &peer_id);
                                // Establish the audio connection as soon as the peer
                                // is known (the transport dedupes the full-mesh race).
                                // Hand over the full address so reconnects can dial
                                // it directly rather than via the gossip lookup.
                                transport_events.connect_peer(state.addr.clone()).await;
                                let _ = ui_tx_events.send(UiEvent::PeerJoined { id: peer_id, state }).await;
                            }
                            RoomEvent::PeerLeft(peer_id) => {
                                // Graceful leave — evict immediately.
                                cancel_grace_timer(&grace_timers_events, &peer_id);
                                seen_connected_events.lock().unwrap().remove(&peer_id);
                                transport_events.disconnect_peer(peer_id).await;
                                jitter_events.lock().await.remove(&peer_id);
                                let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
                            }
                            RoomEvent::PeerUpdated(peer_id, state) => {
                                // A re-announce means the peer is alive — cancel any
                                // pending grace timer. It may also carry a fresh
                                // address (peer back on a new network); refresh the
                                // retained dial target so a later reconnect re-reaches
                                // it. Idempotent: an ordinary mute/unmute update just
                                // re-records the same address.
                                cancel_grace_timer(&grace_timers_events, &peer_id);
                                transport_events.connect_peer(state.addr.clone()).await;
                                let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
                            }
                            RoomEvent::PeerConnectionLost(peer_id) => {
                                // Transient drop: do NOT tear down the peer. Its audio
                                // supervisor stays alive and keeps redialing the
                                // retained address, so show "reconnecting" and arm a
                                // grace timer that evicts the peer only if the link
                                // hasn't recovered within RECONNECT_GRACE. A gossip
                                // rejoin (PeerJoined/PeerUpdated) or a transport
                                // reconnect (ConnEvent::Connected) cancels it first.
                                let _ = ui_tx_events.send(UiEvent::PeerConnecting { id: peer_id }).await;
                                arm_grace_timer(
                                    &grace_timers_events,
                                    &seen_connected_events,
                                    &transport_events,
                                    &jitter_events,
                                    &ui_tx_events,
                                    RECONNECT_GRACE,
                                    peer_id,
                                );
                            }
                        }
                    }
                });

                // 5. Connection-state forwarder: turns transport link state into
                //    per-peer UI indicators (connecting / reconnecting vs. live).
                let mut conn_events = match transport.subscribe_conn_events().await {
                    Ok(rx) => rx,
                    Err(e) => {
                        let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe conn events: {}", e))).await;
                        continue;
                    }
                };
                let conn_handler = ConnEventHandler::new(
                    ui_tx.clone(),
                    grace_timers.clone(),
                    seen_connected.clone(),
                    transport.clone(),
                    jitter.clone(),
                );
                let conn_event_task = tokio::spawn(async move {
                    while let Some(event) = conn_events.recv().await {
                        conn_handler.handle(event).await;
                    }
                });

                let session = ActiveSession {
                    endpoint: endpoint.clone(),
                    router,
                    room_state: room_state.clone(),
                    capture_thread,
                    datagram_task,
                    mixer_task,
                    event_task,
                    conn_event_task,
                    grace_timers,
                    transport: transport.clone(),
                    echo_cancel: echo_cancel_guard,
                };

                let self_id = endpoint.id().to_string();
                let _ = ui_tx.send(UiEvent::RoomJoined { ticket: ticket_str, self_id }).await;
                active_session = Some(session);
            }

            CoreCommand::Leave => {
                if let Some(session) = active_session.take() {
                    session.shutdown(audio_backend.clone()).await;
                    let _ = ui_tx.send(UiEvent::RoomLeft).await;
                }
            }

            CoreCommand::ToggleMute => {
                let current = is_muted.load(Ordering::Relaxed);
                let new_state = !current;
                is_muted.store(new_state, Ordering::Relaxed);

                if let Some(session) = &active_session {
                    let self_state = PeerState {
                        name: current_name.clone(),
                        is_muted: new_state,
                        addr: session.endpoint.addr(),
                    };
                    let _ = session.room_state.update_self_state(self_state).await;
                }
            }

            CoreCommand::ToggleDeafen => {
                let current = is_deafened.load(Ordering::Relaxed);
                is_deafened.store(!current, Ordering::Relaxed);
            }

            CoreCommand::SetPttMode(enabled) => {
                ptt_mode.store(enabled, Ordering::Relaxed);
            }

            CoreCommand::SetPttActive(active) => {
                ptt_active.store(active, Ordering::Relaxed);
            }

            CoreCommand::SetPeerVolume(peer_id, vol) => {
                let mut guard = peer_volumes.lock().await;
                guard.insert(peer_id, vol);
            }

            CoreCommand::SetNoiseGateThreshold(threshold) => {
                noise_gate_threshold.store(threshold.to_bits(), Ordering::Relaxed);
            }

            CoreCommand::SetMicMonitor { enabled, input_device } => {
                // During a call the in-call capture thread already reports the
                // mic level, and it owns the backend's capture stream — leave it be.
                if active_session.is_some() {
                    continue;
                }
                if enabled {
                    if mic_monitor.is_none() {
                        let (tx, rx) = std::sync::mpsc::channel();
                        match audio_backend.start_capture(tx, input_device) {
                            Ok(()) => {
                                let ui = ui_tx.clone();
                                let thread = std::thread::spawn(move || run_mic_monitor(rx, ui));
                                mic_monitor = Some(MicMonitor { thread });
                            }
                            Err(e) => {
                                let _ = ui_tx
                                    .send(UiEvent::Error(format!("Mic test unavailable: {e}")))
                                    .await;
                            }
                        }
                    }
                } else {
                    stop_mic_monitor(&audio_backend, mic_monitor.take());
                }
            }

            CoreCommand::SetNetworkMode(mode) => {
                network_mode = mode;
            }
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{apply_volume, frame_level, mix_frames};

    #[test]
    fn mix_of_no_peers_is_silence() {
        let mixed = mix_frames(&[], 4);
        assert_eq!(mixed, vec![0i16; 4]);
    }

    #[test]
    fn single_peer_passes_through_unchanged() {
        let frame = vec![100, -200, 300, -400];
        let mixed = mix_frames(std::slice::from_ref(&frame), 4);
        assert_eq!(mixed, frame);
    }

    #[test]
    fn two_peers_sum_sample_by_sample() {
        let a = vec![100, -200, 300, -400];
        let b = vec![50, 200, -100, 400];
        let mixed = mix_frames(&[a, b], 4);
        assert_eq!(mixed, vec![150, 0, 200, 0]);
    }

    #[test]
    fn loud_positive_mix_saturates_not_wraps() {
        // Two near-full-scale positive frames sum to ~2x i16::MAX. A plain i16
        // cast would wrap to a large negative value; the mixer must clamp to MAX.
        let a = vec![30_000; 4];
        let b = vec![30_000; 4];
        let mixed = mix_frames(&[a, b], 4);
        assert_eq!(mixed, vec![i16::MAX; 4]);
    }

    #[test]
    fn loud_negative_mix_saturates_to_min() {
        let a = vec![i16::MIN; 4];
        let b = vec![i16::MIN; 4];
        let mixed = mix_frames(&[a, b], 4);
        assert_eq!(mixed, vec![i16::MIN; 4]);
    }

    #[test]
    fn shorter_peer_frame_contributes_zero_past_its_end() {
        let full = vec![100, 100, 100, 100];
        let short = vec![10, 20]; // only first two samples
        let mixed = mix_frames(&[full, short], 4);
        assert_eq!(mixed, vec![110, 120, 100, 100]);
    }

    #[test]
    fn volume_unity_is_a_noop() {
        let mut frame = vec![100, -200, 300, -400];
        apply_volume(&mut frame, 1.0);
        assert_eq!(frame, vec![100, -200, 300, -400]);
    }

    #[test]
    fn volume_zero_mutes() {
        let mut frame = vec![100, -200, 300, -400];
        apply_volume(&mut frame, 0.0);
        assert_eq!(frame, vec![0, 0, 0, 0]);
    }

    #[test]
    fn volume_half_scales_samples() {
        let mut frame = vec![100, -200, 300, -400];
        apply_volume(&mut frame, 0.5);
        // 100*0.5=50, -200*0.5=-100, 300*0.5=150, -400*0.5=-200 (exact in f32 here)
        assert_eq!(frame, vec![50, -100, 150, -200]);
    }

    #[test]
    fn volume_boost_saturates_not_wraps() {
        // 20000 * 4.0 = 80000, well past i16::MAX — must clamp, not wrap.
        let mut frame = vec![20_000, -20_000, 20_000, -20_000];
        apply_volume(&mut frame, 4.0);
        assert_eq!(frame, vec![i16::MAX, i16::MIN, i16::MAX, i16::MIN]);
    }

    #[test]
    fn frame_level_of_silence_is_zero() {
        assert_eq!(frame_level(&[0, 0, 0, 0]), 0.0);
        assert_eq!(frame_level(&[]), 0.0);
    }

    #[test]
    fn frame_level_of_full_scale_is_about_one() {
        let full = vec![i16::MAX; 64];
        let level = frame_level(&full);
        assert!(level > 0.99 && level <= 1.0, "full-scale level was {level}");
    }

    #[test]
    fn volume_truncates_toward_zero() {
        let mut frame = vec![3, -3, 5, -5];
        apply_volume(&mut frame, 0.5);
        assert_eq!(frame, vec![1, -1, 2, -2]);
    }

    #[test]
    fn volume_amplifies_without_saturating() {
        let mut frame = vec![1000, -1000];
        apply_volume(&mut frame, 2.0);
        assert_eq!(frame, vec![2000, -2000]);
    }

    #[test]
    fn three_peers_sum_without_saturation() {
        let a = vec![10, 20];
        let b = vec![3, 4];
        let c = vec![100, -50];
        let mixed = mix_frames(&[a, b, c], 2);
        assert_eq!(mixed, vec![113, -26]);
    }

    #[test]
    fn mix_zero_pads_output_longer_than_peer_frames() {
        let a = vec![100, 200];
        let mixed = mix_frames(&[a], 4);
        assert_eq!(mixed, vec![100, 200, 0, 0]);
    }

    #[test]
    fn frame_level_mid_range() {
        let frame = vec![16384; 64];
        let level = frame_level(&frame);
        assert!((level - 0.5).abs() < 1e-3, "mid-range level was {level}");
    }
}