//! Native WebRTC voice transport, signaling, peer protocol, and virtual audio.

#![allow(clippy::cast_possible_truncation)] // Audio normalization clamps before conversion.
#![allow(clippy::cast_precision_loss)] // Sample/rate conversion intentionally uses floating point.
#![allow(clippy::cast_sign_loss)] // Resampler positions are constructed from non-negative indices.
#![allow(clippy::missing_errors_doc)] // The error enum documents the shared failure contract.
#![allow(clippy::too_many_arguments)] // The RTC loop receives independent owned I/O resources.
#![allow(clippy::too_many_lines)] // Linear loops preserve the WebRTC mutation/drain ordering.

use libremetaverse_opus::{Channels, Decoder, Encoder};
use libremetaverse_types::UUID;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
#[cfg(feature = "real-audio")]
use std::collections::VecDeque;
use std::collections::{HashMap, HashSet};
use std::fmt;
use std::future::Future;
use std::io::Cursor;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::Path;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use str0m::change::{SdpAnswer, SdpOffer};
use str0m::channel::ChannelId;
use str0m::format::Codec;
use str0m::media::{Direction, MediaKind, MediaTime, Mid};
use str0m::net::{Protocol, Receive};
use str0m::{Candidate, Event, IceConnectionState, Input, Output, Rtc, RtcConfig};
use tokio::net::UdpSocket;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;

const DATA_CHANNEL_LABEL: &str = "SLData";
const SAMPLE_RATE: u32 = 48_000;
const FRAME_SAMPLES: usize = 960;
const FRAME_DURATION: Duration = Duration::from_millis(20);
const MAX_SIGNALING_BYTES: usize = 1024 * 1024;
const MAX_DATA_BYTES: usize = 64 * 1024;

/// Errors produced by the native WebRTC adapter.
#[derive(Debug)]
#[non_exhaustive]
pub enum WebRtcError {
    InvalidInput(&'static str),
    Io(std::io::Error),
    Signaling(String),
    Protocol(&'static str),
    Rtc(String),
    Audio(String),
    Timeout(&'static str),
    Closed,
}

impl fmt::Display for WebRtcError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidInput(message) | Self::Protocol(message) => formatter.write_str(message),
            Self::Io(error) => error.fmt(formatter),
            Self::Signaling(message) => write!(formatter, "voice signaling failed: {message}"),
            Self::Rtc(message) => write!(formatter, "WebRTC transport failed: {message}"),
            Self::Audio(message) => write!(formatter, "voice audio failed: {message}"),
            Self::Timeout(action) => write!(formatter, "{action} timed out"),
            Self::Closed => formatter.write_str("WebRTC voice session is closed"),
        }
    }
}

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

impl From<std::io::Error> for WebRtcError {
    fn from(error: std::io::Error) -> Self {
        Self::Io(error)
    }
}

/// A credential that is deliberately opaque and redacted from diagnostics.
#[derive(Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct VoiceSecret(String);

impl VoiceSecret {
    pub fn new(value: impl Into<String>) -> Result<Self, WebRtcError> {
        let value = value.into();
        if value.len() > 16 * 1024 {
            return Err(WebRtcError::InvalidInput("voice credential is too large"));
        }
        Ok(Self(value))
    }

    #[must_use]
    pub fn expose(&self) -> &str {
        &self.0
    }
}

impl fmt::Debug for VoiceSecret {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("VoiceSecret(<redacted>)")
    }
}

/// Initial `ProvisionVoiceAccountRequest` body.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ProvisionRequest {
    pub jsep: Jsep,
    pub channel_type: String,
    pub voice_server_type: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parcel_local_id: Option<i32>,
}

/// SDP envelope used by the grid voice service.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Jsep {
    #[serde(rename = "type")]
    pub kind: String,
    pub sdp: String,
}

/// Provisioning response with credentials protected from `Debug` output.
#[derive(Clone)]
pub struct ProvisionResponse {
    pub answer_sdp: String,
    pub viewer_session: UUID,
    pub channel: Option<String>,
    pub credentials: Option<VoiceSecret>,
}

impl fmt::Debug for ProvisionResponse {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ProvisionResponse")
            .field("answer_sdp", &"<redacted-sdp>")
            .field("viewer_session", &self.viewer_session)
            .field("channel", &self.channel.as_ref().map(|_| "<redacted>"))
            .field("credentials", &self.credentials)
            .finish()
    }
}

/// Candidate-completion body for `VoiceSignalingRequest`.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SignalingCompleteRequest {
    pub voice_server_type: String,
    pub viewer_session: String,
    pub candidate: SignalingComplete,
}

#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
pub struct SignalingComplete {
    pub completed: bool,
}

/// Logout body sent to `ProvisionVoiceAccountRequest` during teardown.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct CloseRequest {
    pub logout: bool,
    pub voice_server_type: String,
    pub viewer_session: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub channel: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub credentials: Option<VoiceSecret>,
}

type SignalFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, WebRtcError>> + Send + 'a>>;

/// Transport-neutral grid capability boundary. Implementations may use LLSD XML,
/// JSON fixtures, or another native Rust HTTP client without exposing credentials.
pub trait VoiceSignaling: Send + Sync {
    fn provision(&self, request: ProvisionRequest) -> SignalFuture<'_, ProvisionResponse>;
    fn complete(&self, request: SignalingCompleteRequest) -> SignalFuture<'_, ()>;
    fn close(&self, request: CloseRequest) -> SignalFuture<'_, ()>;
}

/// Selectable audio endpoint. Virtual endpoints are always available in CI.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct AudioDeviceInfo {
    pub id: String,
    pub name: String,
    pub input: bool,
    pub is_default: bool,
    pub is_virtual: bool,
}

/// Cross-platform device catalog. Real devices are compiled only with
/// `--features real-audio`; virtual input/output remain deterministic otherwise.
pub fn audio_devices() -> Result<Vec<AudioDeviceInfo>, WebRtcError> {
    #[allow(unused_mut)]
    let mut devices = vec![
        AudioDeviceInfo {
            id: "virtual:microphone".into(),
            name: "Virtual microphone (48 kHz mono)".into(),
            input: true,
            is_default: true,
            is_virtual: true,
        },
        AudioDeviceInfo {
            id: "virtual:speaker".into(),
            name: "Virtual speaker (decoded PCM sink)".into(),
            input: false,
            is_default: true,
            is_virtual: true,
        },
    ];
    #[cfg(feature = "real-audio")]
    let _ = enumerate_real_devices(&mut devices);
    Ok(devices)
}

#[cfg(feature = "real-audio")]
fn enumerate_real_devices(devices: &mut Vec<AudioDeviceInfo>) -> Result<(), WebRtcError> {
    use cpal::traits::{DeviceTrait, HostTrait};
    let host = cpal::default_host();
    let default_input = host
        .default_input_device()
        .and_then(|device| device.id().ok());
    let default_output = host
        .default_output_device()
        .and_then(|device| device.id().ok());
    for device in host
        .input_devices()
        .map_err(|error| WebRtcError::Audio(error.to_string()))?
    {
        let name = device.to_string();
        let id = device
            .id()
            .map_err(|error| WebRtcError::Audio(error.to_string()))?;
        devices.push(AudioDeviceInfo {
            id: format!("cpal:input:{id}"),
            is_default: default_input.as_ref() == Some(&id),
            name,
            input: true,
            is_virtual: false,
        });
    }
    for device in host
        .output_devices()
        .map_err(|error| WebRtcError::Audio(error.to_string()))?
    {
        let name = device.to_string();
        let id = device
            .id()
            .map_err(|error| WebRtcError::Audio(error.to_string()))?;
        devices.push(AudioDeviceInfo {
            id: format!("cpal:output:{id}"),
            is_default: default_output.as_ref() == Some(&id),
            name,
            input: false,
            is_virtual: false,
        });
    }
    Ok(())
}

/// One 20 ms, 48 kHz mono Opus packet.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OpusFrame(pub Vec<u8>);

/// Encodes PCM into WebRTC Opus frames using native libopus.
pub fn encode_pcm_48k_mono(samples: &[i16]) -> Result<Vec<OpusFrame>, WebRtcError> {
    let mut encoder = Encoder::voip(SAMPLE_RATE, Channels::Mono)
        .map_err(|error| WebRtcError::Audio(error.to_string()))?;
    let mut padded = samples.to_vec();
    let remainder = padded.len() % FRAME_SAMPLES;
    if remainder != 0 {
        padded.resize(padded.len() + FRAME_SAMPLES - remainder, 0);
    }
    if padded.is_empty() {
        padded.resize(FRAME_SAMPLES, 0);
    }
    let mut frames = Vec::with_capacity(padded.len() / FRAME_SAMPLES);
    for pcm in padded.chunks_exact(FRAME_SAMPLES) {
        let mut packet = vec![0_u8; 4_000];
        let count = encoder
            .encode(pcm, &mut packet)
            .map_err(|error| WebRtcError::Audio(error.to_string()))?;
        packet.truncate(count);
        frames.push(OpusFrame(packet));
    }
    Ok(frames)
}

/// Loads integer or float PCM WAV data, downmixes and linearly resamples it to
/// 48 kHz mono, then encodes it into paced Opus frames.
pub fn encode_wav(path: &Path) -> Result<Vec<OpusFrame>, WebRtcError> {
    let bytes = std::fs::read(path)?;
    encode_wav_bytes(&bytes)
}

pub fn encode_wav_bytes(bytes: &[u8]) -> Result<Vec<OpusFrame>, WebRtcError> {
    if bytes.len() > 128 * 1024 * 1024 {
        return Err(WebRtcError::InvalidInput("WAV file exceeds 128 MiB"));
    }
    let mut reader = hound::WavReader::new(Cursor::new(bytes))
        .map_err(|error| WebRtcError::Audio(error.to_string()))?;
    let spec = reader.spec();
    if spec.channels == 0 || spec.sample_rate == 0 || spec.channels > 32 {
        return Err(WebRtcError::InvalidInput(
            "unsupported WAV channel/rate metadata",
        ));
    }
    let raw: Vec<f32> = match spec.sample_format {
        hound::SampleFormat::Float => reader
            .samples::<f32>()
            .map(|sample| sample.map_err(|error| WebRtcError::Audio(error.to_string())))
            .collect::<Result<_, _>>()?,
        hound::SampleFormat::Int => {
            let scale = 2_f32.powi(i32::from(spec.bits_per_sample.saturating_sub(1)));
            reader
                .samples::<i32>()
                .map(|sample| {
                    sample
                        .map(|value| value as f32 / scale)
                        .map_err(|error| WebRtcError::Audio(error.to_string()))
                })
                .collect::<Result<_, _>>()?
        }
    };
    let channels = usize::from(spec.channels);
    let mono: Vec<f32> = raw
        .chunks(channels)
        .map(|frame| frame.iter().copied().sum::<f32>() / frame.len() as f32)
        .collect();
    if mono.is_empty() {
        return Err(WebRtcError::InvalidInput(
            "WAV file contains no audio samples",
        ));
    }
    let output_len = mono
        .len()
        .saturating_mul(SAMPLE_RATE as usize)
        .div_ceil(spec.sample_rate as usize);
    let mut pcm = Vec::with_capacity(output_len);
    for output_index in 0..output_len {
        let position = output_index as f64 * f64::from(spec.sample_rate) / f64::from(SAMPLE_RATE);
        let left = position.floor() as usize;
        let fraction = (position - left as f64) as f32;
        let a = mono[left.min(mono.len() - 1)];
        let b = mono[(left + 1).min(mono.len() - 1)];
        let value = a + (b - a) * fraction;
        pcm.push((value.clamp(-1.0, 1.0) * f32::from(i16::MAX)).round() as i16);
    }
    encode_pcm_48k_mono(&pcm)
}

/// Peer audio state delivered over `SLData`.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct PeerAudioState {
    pub power: Option<i32>,
    pub voice_active: Option<bool>,
    pub moderator_muted: Option<bool>,
    pub joined_primary: Option<bool>,
    pub left: bool,
    pub ssrc: Option<u32>,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct Int3 {
    pub x: i32,
    pub y: i32,
    pub z: i32,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct Int4 {
    pub x: i32,
    pub y: i32,
    pub z: i32,
    pub w: i32,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct AvatarPosition {
    pub sender_position: Option<Int3>,
    pub sender_heading: Option<Int4>,
    pub listener_position: Option<Int3>,
    pub listener_heading: Option<Int4>,
}

/// Observable session events. None contains capability URLs or credentials.
#[derive(Clone, Debug, PartialEq)]
pub enum VoiceEvent {
    IceState(IceConnectionState),
    Connected,
    DataChannelReady,
    PeerAudio(UUID, PeerAudioState),
    PeerPosition(UUID, AvatarPosition),
    MuteMap(HashMap<UUID, bool>),
    GainMap(HashMap<UUID, i32>),
    AudioFrame { samples: usize, peak: i16 },
    Diagnostic(String),
    Closed,
}

#[derive(Clone, Debug, Default)]
pub struct SessionSnapshot {
    pub connected: bool,
    pub data_channel_ready: bool,
    pub known_peers: HashSet<UUID>,
    pub received_audio_frames: u64,
    pub decoded_samples: u64,
    pub sent_audio_frames: u64,
    pub selected_input: String,
    pub selected_output: String,
    pub active_tasks: usize,
}

/// Session construction parameters.
#[derive(Clone, Debug)]
pub struct VoiceSessionConfig {
    pub bind_ip: IpAddr,
    pub parcel_local_id: Option<i32>,
    pub timeout: Duration,
    pub input_device: String,
    pub output_device: String,
}

impl Default for VoiceSessionConfig {
    fn default() -> Self {
        Self {
            bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
            parcel_local_id: None,
            timeout: Duration::from_secs(45),
            input_device: "virtual:microphone".into(),
            output_device: "virtual:speaker".into(),
        }
    }
}

enum Command {
    Data(String, oneshot::Sender<Result<bool, WebRtcError>>),
    Play(
        Vec<OpusFrame>,
        bool,
        oneshot::Sender<Result<(), WebRtcError>>,
    ),
    Stop(oneshot::Sender<()>),
    Shutdown(oneshot::Sender<()>),
}

struct Playback {
    frames: Vec<OpusFrame>,
    index: usize,
    looping: bool,
    next_frame: Instant,
    media_elapsed: Duration,
}

/// Owned native WebRTC voice session.
pub struct WebRtcVoiceSession {
    command: mpsc::Sender<Command>,
    events: Option<mpsc::Receiver<VoiceEvent>>,
    snapshot: Arc<Mutex<SessionSnapshot>>,
    signaling: Arc<dyn VoiceSignaling>,
    viewer_session: UUID,
    channel: Option<String>,
    credentials: Option<VoiceSecret>,
    task: Option<JoinHandle<()>>,
    timeout: Duration,
    local_sdp: String,
    remote_sdp: String,
    #[cfg(feature = "real-audio")]
    hardware_audio: Option<RealAudioBridge>,
}

impl WebRtcVoiceSession {
    /// Creates an SDP offer with Opus audio and ordered `SLData`, provisions it,
    /// applies the answer, and starts the single peer/audio run loop.
    pub async fn connect(
        signaling: Arc<dyn VoiceSignaling>,
        config: VoiceSessionConfig,
    ) -> Result<Self, WebRtcError> {
        validate_device(&config.input_device, true)?;
        validate_device(&config.output_device, false)?;
        let socket = UdpSocket::bind(SocketAddr::new(config.bind_ip, 0)).await?;
        let local_addr = socket.local_addr()?;
        if local_addr.ip().is_unspecified() {
            return Err(WebRtcError::InvalidInput(
                "bind_ip must be a concrete interface address for ICE",
            ));
        }
        let now = Instant::now();
        let mut rtc = RtcConfig::new().build(now);
        let _ = rtc.add_local_candidate(
            Candidate::host(local_addr, "udp")
                .map_err(|error| WebRtcError::Rtc(error.to_string()))?,
        );
        let mut change = rtc.sdp_api();
        let audio_mid = change.add_media(MediaKind::Audio, Direction::SendRecv, None, None, None);
        let channel_id = change.add_channel(DATA_CHANNEL_LABEL.into());
        let (offer, pending) = change.apply().ok_or(WebRtcError::Protocol(
            "SDP changes did not produce an offer",
        ))?;
        let local_sdp = offer.to_sdp_string();
        let request = ProvisionRequest {
            jsep: Jsep {
                kind: "offer".into(),
                sdp: local_sdp.clone(),
            },
            channel_type: "local".into(),
            voice_server_type: "webrtc".into(),
            parcel_local_id: config.parcel_local_id,
        };
        let mut response = tokio::time::timeout(config.timeout, signaling.provision(request))
            .await
            .map_err(|_| WebRtcError::Timeout("voice provisioning"))??;
        if response.answer_sdp.len() > MAX_SIGNALING_BYTES {
            abandon_provisioned(&signaling, &mut response, config.timeout).await;
            return Err(WebRtcError::Protocol("SDP answer exceeds 1 MiB"));
        }
        let remote_sdp = sanitize_remote_sdp(&response.answer_sdp);
        let answer = match SdpAnswer::from_sdp_string(&remote_sdp) {
            Ok(answer) => answer,
            Err(error) => {
                abandon_provisioned(&signaling, &mut response, config.timeout).await;
                return Err(WebRtcError::Rtc(error.to_string()));
            }
        };
        if let Err(error) = rtc.sdp_api().accept_answer(pending, answer) {
            abandon_provisioned(&signaling, &mut response, config.timeout).await;
            return Err(WebRtcError::Rtc(error.to_string()));
        }
        let complete = SignalingCompleteRequest {
            voice_server_type: "webrtc".into(),
            viewer_session: response.viewer_session.to_string(),
            candidate: SignalingComplete { completed: true },
        };
        let complete_result = tokio::time::timeout(config.timeout, signaling.complete(complete))
            .await
            .map_err(|_| WebRtcError::Timeout("ICE signaling completion"))
            .and_then(|result| result);
        if let Err(error) = complete_result {
            abandon_provisioned(&signaling, &mut response, config.timeout).await;
            return Err(error);
        }
        let (command_tx, command_rx) = mpsc::channel(64);
        let (event_tx, event_rx) = mpsc::channel(256);
        let (capture_tx, capture_rx) = mpsc::channel(32);
        #[cfg(feature = "real-audio")]
        let hardware_audio = match RealAudioBridge::open(
            &config.input_device,
            &config.output_device,
            capture_tx,
            &event_tx,
        ) {
            Ok(audio) => audio,
            Err(error) => {
                abandon_provisioned(&signaling, &mut response, config.timeout).await;
                return Err(error);
            }
        };
        #[cfg(feature = "real-audio")]
        let hardware_output = hardware_audio.output.clone();
        #[cfg(not(feature = "real-audio"))]
        let hardware_output = ();
        #[cfg(not(feature = "real-audio"))]
        drop(capture_tx);
        let snapshot = Arc::new(Mutex::new(SessionSnapshot {
            selected_input: config.input_device,
            selected_output: config.output_device,
            active_tasks: 1,
            ..SessionSnapshot::default()
        }));
        let run_snapshot = Arc::clone(&snapshot);
        let task = tokio::spawn(async move {
            Box::pin(run_client(
                rtc,
                socket,
                audio_mid,
                channel_id,
                command_rx,
                capture_rx,
                event_tx,
                run_snapshot,
                hardware_output,
            ))
            .await;
        });
        Ok(Self {
            command: command_tx,
            events: Some(event_rx),
            snapshot,
            signaling,
            viewer_session: response.viewer_session,
            channel: response.channel,
            credentials: response.credentials,
            task: Some(task),
            timeout: config.timeout,
            local_sdp,
            remote_sdp,
            #[cfg(feature = "real-audio")]
            hardware_audio: Some(hardware_audio),
        })
    }

    pub async fn next_event(&mut self) -> Option<VoiceEvent> {
        self.events.as_mut()?.recv().await
    }

    /// Transfers the event stream to a facade task while retaining the session's
    /// command and teardown handles.
    pub fn take_events(&mut self) -> Option<mpsc::Receiver<VoiceEvent>> {
        self.events.take()
    }

    #[must_use]
    pub fn snapshot(&self) -> SessionSnapshot {
        lock(&self.snapshot).clone()
    }

    pub async fn send_data(&self, message: impl Into<String>) -> Result<bool, WebRtcError> {
        let message = message.into();
        if message.len() > MAX_DATA_BYTES {
            return Err(WebRtcError::InvalidInput(
                "data-channel message exceeds 64 KiB",
            ));
        }
        let (reply_tx, reply_rx) = oneshot::channel();
        self.command
            .send(Command::Data(message, reply_tx))
            .await
            .map_err(|_| WebRtcError::Closed)?;
        reply_rx.await.map_err(|_| WebRtcError::Closed)?
    }

    /// Queues a bounded data-channel message without blocking the caller.
    pub fn try_send_data(&self, message: impl Into<String>) -> Result<bool, WebRtcError> {
        let message = message.into();
        if message.len() > MAX_DATA_BYTES {
            return Err(WebRtcError::InvalidInput(
                "data-channel message exceeds 64 KiB",
            ));
        }
        let (reply_tx, _reply_rx) = oneshot::channel();
        match self.command.try_send(Command::Data(message, reply_tx)) {
            Ok(()) => Ok(true),
            Err(mpsc::error::TrySendError::Full(_)) => Ok(false),
            Err(mpsc::error::TrySendError::Closed(_)) => Err(WebRtcError::Closed),
        }
    }

    #[must_use]
    pub fn viewer_session(&self) -> UUID {
        self.viewer_session
    }

    #[must_use]
    pub fn channel(&self) -> Option<String> {
        self.channel.clone()
    }

    #[must_use]
    pub fn credentials(&self) -> Option<String> {
        self.credentials
            .as_ref()
            .map(|credentials| credentials.expose().to_owned())
    }

    #[must_use]
    pub fn local_sdp(&self) -> String {
        self.local_sdp.clone()
    }

    #[must_use]
    pub fn remote_sdp(&self) -> String {
        self.remote_sdp.clone()
    }

    pub async fn set_peer_mute(&self, peer: UUID, mute: bool) -> Result<bool, WebRtcError> {
        self.send_data(json!({"m": {peer.to_string(): mute}}).to_string())
            .await
    }

    pub async fn set_peer_gain(&self, peer: UUID, gain: i32) -> Result<bool, WebRtcError> {
        self.send_data(json!({"ug": {peer.to_string(): gain.clamp(0, 220)}}).to_string())
            .await
    }

    pub async fn send_position(
        &self,
        avatar: [f64; 3],
        avatar_heading: [f64; 4],
        listener: [f64; 3],
        listener_heading: [f64; 4],
    ) -> Result<bool, WebRtcError> {
        let scale3 = |v: [f64; 3]| json!({"x": (v[0] * 100.0).round() as i32, "y": (v[1] * 100.0).round() as i32, "z": (v[2] * 100.0).round() as i32});
        let scale4 = |v: [f64; 4]| json!({"x": (v[0] * 100.0).round() as i32, "y": (v[1] * 100.0).round() as i32, "z": (v[2] * 100.0).round() as i32, "w": (v[3] * 100.0).round() as i32});
        self.send_data(
            json!({"sp": scale3(avatar), "sh": scale4(avatar_heading), "lp": scale3(listener), "lh": scale4(listener_heading)}).to_string(),
        )
        .await
    }

    pub async fn play_frames(
        &self,
        frames: Vec<OpusFrame>,
        looping: bool,
    ) -> Result<(), WebRtcError> {
        if frames.is_empty()
            || frames
                .iter()
                .any(|frame| frame.0.is_empty() || frame.0.len() > 4_000)
        {
            return Err(WebRtcError::InvalidInput("invalid Opus frame sequence"));
        }
        let (reply_tx, reply_rx) = oneshot::channel();
        self.command
            .send(Command::Play(frames, looping, reply_tx))
            .await
            .map_err(|_| WebRtcError::Closed)?;
        reply_rx.await.map_err(|_| WebRtcError::Closed)?
    }

    pub async fn play_wav(&self, path: &Path, looping: bool) -> Result<(), WebRtcError> {
        self.play_frames(encode_wav(path)?, looping).await
    }

    pub async fn stop_wav(&self) -> Result<(), WebRtcError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.command
            .send(Command::Stop(reply_tx))
            .await
            .map_err(|_| WebRtcError::Closed)?;
        reply_rx.await.map_err(|_| WebRtcError::Closed)
    }

    /// Sends leave/logout, drains the peer run loop, and joins its only task.
    pub async fn shutdown(&mut self) -> Result<(), WebRtcError> {
        let _ = self.send_data("{\"l\":true}").await;
        let close = CloseRequest {
            logout: true,
            voice_server_type: "webrtc".into(),
            viewer_session: self.viewer_session.to_string(),
            channel: self.channel.take(),
            credentials: self.credentials.take(),
        };
        let signal_result =
            match tokio::time::timeout(self.timeout, self.signaling.close(close)).await {
                Ok(result) => result,
                Err(_) => Err(WebRtcError::Timeout("voice signaling teardown")),
            };
        let (reply_tx, reply_rx) = oneshot::channel();
        let _ = self.command.send(Command::Shutdown(reply_tx)).await;
        let _ = tokio::time::timeout(self.timeout, reply_rx).await;
        let task_result = if let Some(mut task) = self.task.take() {
            if let Ok(result) = tokio::time::timeout(self.timeout, &mut task).await {
                result.map_err(|error| WebRtcError::Rtc(error.to_string()))
            } else {
                task.abort();
                let _ = task.await;
                Err(WebRtcError::Timeout("WebRTC peer teardown"))
            }
        } else {
            Ok(())
        };
        #[cfg(feature = "real-audio")]
        self.hardware_audio.take();
        signal_result.and(task_result)
    }
}

async fn abandon_provisioned(
    signaling: &Arc<dyn VoiceSignaling>,
    response: &mut ProvisionResponse,
    timeout: Duration,
) {
    let close = CloseRequest {
        logout: true,
        voice_server_type: "webrtc".into(),
        viewer_session: response.viewer_session.to_string(),
        channel: response.channel.take(),
        credentials: response.credentials.take(),
    };
    let _ = tokio::time::timeout(timeout, signaling.close(close)).await;
}

impl Drop for WebRtcVoiceSession {
    fn drop(&mut self) {
        if let Some(task) = self.task.take() {
            let close = CloseRequest {
                logout: true,
                voice_server_type: "webrtc".into(),
                viewer_session: self.viewer_session.to_string(),
                channel: self.channel.take(),
                credentials: self.credentials.take(),
            };
            let signaling = Arc::clone(&self.signaling);
            let command = self.command.clone();
            let timeout = self.timeout;
            if let Ok(runtime) = tokio::runtime::Handle::try_current() {
                runtime.spawn(async move {
                    let (reply_tx, reply_rx) = oneshot::channel();
                    let _ = command.send(Command::Shutdown(reply_tx)).await;
                    let _ = tokio::time::timeout(timeout, reply_rx).await;
                    let _ = tokio::time::timeout(timeout, signaling.close(close)).await;
                    let mut task = task;
                    if tokio::time::timeout(timeout, &mut task).await.is_err() {
                        task.abort();
                        let _ = task.await;
                    }
                });
            } else {
                task.abort();
                lock(&self.snapshot).active_tasks = 0;
            }
        }
    }
}

fn validate_device(id: &str, input: bool) -> Result<(), WebRtcError> {
    let devices = audio_devices()?;
    if devices
        .iter()
        .any(|device| device.id == id && device.input == input)
    {
        Ok(())
    } else {
        Err(WebRtcError::InvalidInput(
            "selected audio device is unavailable",
        ))
    }
}

async fn run_client(
    mut rtc: Rtc,
    socket: UdpSocket,
    audio_mid: Mid,
    channel_id: ChannelId,
    mut commands: mpsc::Receiver<Command>,
    mut capture: mpsc::Receiver<Vec<i16>>,
    events: mpsc::Sender<VoiceEvent>,
    snapshot: Arc<Mutex<SessionSnapshot>>,
    hardware_output: HardwareOutput,
) {
    let mut buffer = vec![0_u8; 65_536];
    let mut playback: Option<Playback> = None;
    let mut decoder = Decoder::new(SAMPLE_RATE, Channels::Mono).ok();
    let mut encoder = Encoder::voip(SAMPLE_RATE, Channels::Mono).ok();
    let mut captured_elapsed = Duration::ZERO;
    let mut running = true;
    while running {
        let deadline = match drain_client(
            &mut rtc,
            &socket,
            channel_id,
            &events,
            &snapshot,
            decoder.as_mut(),
            &hardware_output,
        )
        .await
        {
            Ok(value) => value,
            Err(error) => {
                let _ = events.send(VoiceEvent::Diagnostic(error.to_string())).await;
                break;
            }
        };
        let audio_deadline = playback.as_ref().map(|value| value.next_frame);
        let wake = audio_deadline.map_or(deadline, |value| value.min(deadline));
        tokio::select! {
            command = commands.recv() => match command {
                Some(Command::Data(message, reply)) => {
                    let result = rtc.channel(channel_id)
                        .ok_or(WebRtcError::Closed)
                        .and_then(|mut channel| channel.write(false, message.as_bytes()).map_err(|error| WebRtcError::Rtc(error.to_string())));
                    let _ = reply.send(result);
                }
                Some(Command::Play(frames, looping, reply)) => {
                    playback = Some(Playback { frames, index: 0, looping, next_frame: Instant::now(), media_elapsed: Duration::ZERO });
                    let _ = reply.send(Ok(()));
                }
                Some(Command::Stop(reply)) => {
                    playback = None;
                    let _ = reply.send(());
                }
                Some(Command::Shutdown(reply)) => {
                    playback = None;
                    rtc.disconnect();
                    let _ = drain_client(&mut rtc, &socket, channel_id, &events, &snapshot, decoder.as_mut(), &hardware_output).await;
                    let _ = reply.send(());
                    running = false;
                }
                None => running = false,
            },
            captured = capture.recv() => {
                if let (Some(pcm), Some(codec)) = (captured, encoder.as_mut())
                    && pcm.len() == FRAME_SAMPLES
                {
                        let mut packet = vec![0_u8; 4_000];
                        if let Ok(count) = codec.encode(&pcm, &mut packet) {
                            packet.truncate(count);
                            if write_opus(&mut rtc, audio_mid, &OpusFrame(packet), captured_elapsed).is_ok() {
                                lock(&snapshot).sent_audio_frames += 1;
                            }
                            captured_elapsed += FRAME_DURATION;
                        }
                }
            },
            receive = socket.recv_from(&mut buffer) => match receive {
                Ok((count, source)) => {
                    let Ok(destination) = socket.local_addr() else { break };
                    let Ok(contents) = buffer[..count].try_into() else { continue };
                    let input = Input::Receive(Instant::now(), Receive { proto: Protocol::Udp, source, destination, contents });
                    if let Err(error) = rtc.handle_input(input) {
                        let _ = events.send(VoiceEvent::Diagnostic(error.to_string())).await;
                        break;
                    }
                }
                Err(error) => {
                    let _ = events.send(VoiceEvent::Diagnostic(error.to_string())).await;
                    break;
                }
            },
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(wake)) => {
                let now = Instant::now();
                if audio_deadline.is_some_and(|value| value <= now) {
                    if let Some(active) = playback.as_mut()
                        && let Some(frame) = active.frames.get(active.index)
                    {
                            let result = write_opus(&mut rtc, audio_mid, frame, active.media_elapsed);
                            if result.is_ok() {
                                lock(&snapshot).sent_audio_frames += 1;
                            }
                            active.index += 1;
                            active.media_elapsed += FRAME_DURATION;
                            active.next_frame += FRAME_DURATION;
                            if active.index == active.frames.len() {
                                if active.looping { active.index = 0; } else { playback = None; }
                            }
                    }
                } else if let Err(error) = rtc.handle_input(Input::Timeout(now)) {
                    let _ = events.send(VoiceEvent::Diagnostic(error.to_string())).await;
                    break;
                }
            }
        }
    }
    {
        let mut state = lock(&snapshot);
        state.connected = false;
        state.data_channel_ready = false;
        state.active_tasks = 0;
    }
    let _ = events.send(VoiceEvent::Closed).await;
}

fn write_opus(
    rtc: &mut Rtc,
    mid: Mid,
    frame: &OpusFrame,
    elapsed: Duration,
) -> Result<(), WebRtcError> {
    let writer = rtc.writer(mid).ok_or(WebRtcError::Closed)?;
    let pt = writer
        .payload_params()
        .find(|params| params.spec().codec == Codec::Opus)
        .map(str0m::format::PayloadParams::pt)
        .ok_or(WebRtcError::Protocol("Opus was not negotiated"))?;
    writer
        .audio_level(-30, true)
        .write(
            pt,
            Instant::now(),
            MediaTime::from(elapsed),
            frame.0.clone(),
        )
        .map_err(|error| WebRtcError::Rtc(error.to_string()))
}

async fn drain_client(
    rtc: &mut Rtc,
    socket: &UdpSocket,
    channel_id: ChannelId,
    events: &mpsc::Sender<VoiceEvent>,
    snapshot: &Arc<Mutex<SessionSnapshot>>,
    decoder: Option<&mut Decoder>,
    hardware_output: &HardwareOutput,
) -> Result<Instant, WebRtcError> {
    #[cfg(not(feature = "real-audio"))]
    let () = hardware_output;
    let mut decoder = decoder;
    loop {
        match rtc
            .poll_output()
            .map_err(|error| WebRtcError::Rtc(error.to_string()))?
        {
            Output::Timeout(deadline) => return Ok(deadline),
            Output::Transmit(transmit) => {
                socket
                    .send_to(&transmit.contents, transmit.destination)
                    .await?;
            }
            Output::Event(event) => match event {
                Event::Connected => {
                    lock(snapshot).connected = true;
                    let _ = events.send(VoiceEvent::Connected).await;
                }
                Event::IceConnectionStateChange(state) => {
                    lock(snapshot).connected = matches!(
                        state,
                        IceConnectionState::Connected | IceConnectionState::Completed
                    );
                    let _ = events.send(VoiceEvent::IceState(state)).await;
                }
                Event::ChannelOpen(id, label)
                    if id == channel_id && label == DATA_CHANNEL_LABEL =>
                {
                    lock(snapshot).data_channel_ready = true;
                    if let Some(mut channel) = rtc.channel(id) {
                        channel
                            .write(false, b"{\"j\":{\"p\":true}}")
                            .map_err(|error| WebRtcError::Rtc(error.to_string()))?;
                    }
                    let _ = events.send(VoiceEvent::DataChannelReady).await;
                }
                Event::ChannelData(data) if !data.binary => {
                    if data.data.len() <= MAX_DATA_BYTES {
                        let text = std::str::from_utf8(&data.data).map_err(|_| {
                            WebRtcError::Protocol("data channel text was not UTF-8")
                        })?;
                        let replies = process_peer_message(text, snapshot, events).await?;
                        for reply in replies {
                            if let Some(mut channel) = rtc.channel(channel_id) {
                                let _ = channel.write(false, reply.as_bytes());
                            }
                        }
                    }
                }
                Event::MediaData(media) if media.params.spec().codec == Codec::Opus => {
                    if let Some(codec) = decoder.as_deref_mut() {
                        let mut pcm = vec![0_i16; FRAME_SAMPLES * 6];
                        if let Ok(count) =
                            codec.decode(Some(media.data.as_ref()), pcm.as_mut_slice(), false)
                        {
                            pcm.truncate(count);
                            let peak = i16::try_from(
                                pcm.iter()
                                    .map(|value| value.unsigned_abs())
                                    .max()
                                    .unwrap_or(0)
                                    .min(i16::MAX as u16),
                            )
                            .unwrap_or(i16::MAX);
                            {
                                let mut state = lock(snapshot);
                                state.received_audio_frames += 1;
                                state.decoded_samples += count as u64;
                            }
                            let _ = events
                                .send(VoiceEvent::AudioFrame {
                                    samples: count,
                                    peak,
                                })
                                .await;
                            #[cfg(feature = "real-audio")]
                            if let Some(output) = hardware_output {
                                let mut output = lock(output);
                                let remaining = 96_000_usize.saturating_sub(output.samples.len());
                                output.samples.extend(pcm.into_iter().take(remaining));
                            }
                        }
                    }
                }
                Event::ChannelClose(id) if id == channel_id => {
                    lock(snapshot).data_channel_ready = false;
                }
                Event::Closed => return Err(WebRtcError::Closed),
                _ => {}
            },
        }
    }
}

async fn process_peer_message(
    message: &str,
    snapshot: &Arc<Mutex<SessionSnapshot>>,
    events: &mpsc::Sender<VoiceEvent>,
) -> Result<Vec<String>, WebRtcError> {
    let root: Value = serde_json::from_str(message)
        .map_err(|_| WebRtcError::Protocol("malformed SLData JSON"))?;
    let object = root
        .as_object()
        .ok_or(WebRtcError::Protocol("SLData JSON must be an object"))?;
    let mut replies = Vec::new();
    if object.get("ping").and_then(Value::as_bool) == Some(true) {
        replies.push("{\"pong\":true}".into());
    }
    if let Some(map) = object.get("m").and_then(Value::as_object) {
        let values = parse_bool_map(map);
        if !values.is_empty() {
            let _ = events.send(VoiceEvent::MuteMap(values)).await;
        }
    }
    if let Some(map) = object.get("ug").and_then(Value::as_object) {
        let values = parse_int_map(map);
        if !values.is_empty() {
            let _ = events.send(VoiceEvent::GainMap(values)).await;
        }
    }
    let all_peers = !object.is_empty() && object.keys().all(|key| parse_uuid(key).is_some());
    if all_peers {
        for (key, value) in object {
            let Some(peer) = parse_uuid(key) else {
                continue;
            };
            let Some(map) = value.as_object() else {
                lock(snapshot).known_peers.remove(&peer);
                continue;
            };
            let state = PeerAudioState {
                power: map.get("p").and_then(as_i32),
                voice_active: map.get("V").or_else(|| map.get("v")).and_then(as_bool),
                moderator_muted: map.get("m").and_then(as_bool),
                joined_primary: map
                    .get("j")
                    .and_then(Value::as_object)
                    .and_then(|join| join.get("p"))
                    .and_then(as_bool),
                left: map.get("l").and_then(as_bool).unwrap_or(false),
                ssrc: map
                    .get("s")
                    .or_else(|| map.get("ssrc"))
                    .and_then(Value::as_u64)
                    .and_then(|value| u32::try_from(value).ok()),
            };
            if state.left {
                lock(snapshot).known_peers.remove(&peer);
            } else {
                lock(snapshot).known_peers.insert(peer);
            }
            let _ = events.send(VoiceEvent::PeerAudio(peer, state)).await;
            if let Some(position) = parse_position(map) {
                let _ = events.send(VoiceEvent::PeerPosition(peer, position)).await;
            }
        }
    } else if let Some(join) = object.get("j").and_then(Value::as_object)
        && let Some(peer) = join.get("id").and_then(Value::as_str).and_then(parse_uuid)
    {
        lock(snapshot).known_peers.insert(peer);
    }
    Ok(replies)
}

fn parse_uuid(value: &str) -> Option<UUID> {
    UUID::new_with_string(value.to_owned()).ok()
}

fn parse_bool_map(map: &Map<String, Value>) -> HashMap<UUID, bool> {
    map.iter()
        .filter_map(|(key, value)| Some((parse_uuid(key)?, as_bool(value)?)))
        .collect()
}

fn parse_int_map(map: &Map<String, Value>) -> HashMap<UUID, i32> {
    map.iter()
        .filter_map(|(key, value)| Some((parse_uuid(key)?, as_i32(value)?)))
        .collect()
}

fn as_bool(value: &Value) -> Option<bool> {
    value
        .as_bool()
        .or_else(|| value.as_i64().map(|number| number != 0))
        .or_else(|| value.as_str().and_then(|text| text.parse().ok()))
}

fn as_i32(value: &Value) -> Option<i32> {
    value
        .as_i64()
        .and_then(|number| i32::try_from(number).ok())
        .or_else(|| value.as_str().and_then(|text| text.parse().ok()))
}

fn parse_position(map: &Map<String, Value>) -> Option<AvatarPosition> {
    let result = AvatarPosition {
        sender_position: map.get("sp").and_then(parse_int3),
        sender_heading: map.get("sh").and_then(parse_int4),
        listener_position: map.get("lp").and_then(parse_int3),
        listener_heading: map.get("lh").and_then(parse_int4),
    };
    (result.sender_position.is_some()
        || result.sender_heading.is_some()
        || result.listener_position.is_some()
        || result.listener_heading.is_some())
    .then_some(result)
}

fn parse_int3(value: &Value) -> Option<Int3> {
    let map = value.as_object()?;
    Some(Int3 {
        x: as_i32(map.get("x")?)?,
        y: as_i32(map.get("y")?)?,
        z: as_i32(map.get("z")?)?,
    })
}

fn parse_int4(value: &Value) -> Option<Int4> {
    let map = value.as_object()?;
    Some(Int4 {
        x: as_i32(map.get("x")?)?,
        y: as_i32(map.get("y")?)?,
        z: as_i32(map.get("z")?)?,
        w: as_i32(map.get("w")?)?,
    })
}

fn sanitize_remote_sdp(sdp: &str) -> String {
    sdp.lines()
        .filter(|line| {
            if !line.starts_with("a=candidate:") {
                return true;
            }
            let fields: Vec<_> = line.split_whitespace().collect();
            fields.get(5).is_none_or(|port| *port != "0")
        })
        .collect::<Vec<_>>()
        .join("\r\n")
        + "\r\n"
}

fn lock<T>(value: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    value
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

#[cfg(feature = "real-audio")]
struct RealAudioBridge {
    _input: Option<cpal::Stream>,
    _output: Option<cpal::Stream>,
    output: Option<Arc<Mutex<RealOutputState>>>,
}

#[cfg(feature = "real-audio")]
struct RealOutputState {
    samples: VecDeque<i16>,
    current: i16,
    phase: u64,
    output_rate: u32,
    channels: usize,
}

#[cfg(feature = "real-audio")]
type HardwareOutput = Option<Arc<Mutex<RealOutputState>>>;
#[cfg(not(feature = "real-audio"))]
type HardwareOutput = ();

#[cfg(feature = "real-audio")]
impl RealAudioBridge {
    fn open(
        input_id: &str,
        output_id: &str,
        capture: mpsc::Sender<Vec<i16>>,
        events: &mpsc::Sender<VoiceEvent>,
    ) -> Result<Self, WebRtcError> {
        use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
        let host = cpal::default_host();
        let input = if input_id == "virtual:microphone" {
            None
        } else {
            let id = input_id
                .strip_prefix("cpal:input:")
                .ok_or(WebRtcError::InvalidInput("invalid CPAL input device ID"))?;
            let device = host
                .input_devices()
                .map_err(|error| WebRtcError::Audio(error.to_string()))?
                .find(|device| {
                    device
                        .id()
                        .ok()
                        .as_ref()
                        .is_some_and(|value| value.to_string() == id)
                })
                .ok_or(WebRtcError::InvalidInput(
                    "selected CPAL input device disappeared",
                ))?;
            let config = device
                .default_input_config()
                .map_err(|error| WebRtcError::Audio(error.to_string()))?;
            let channels = usize::from(config.channels());
            let input_rate = config.sample_rate();
            let stream_config = config.config();
            let stream = match config.sample_format() {
                cpal::SampleFormat::I16 => build_input::<i16>(
                    &device,
                    &stream_config,
                    channels,
                    input_rate,
                    capture,
                    events.clone(),
                )?,
                cpal::SampleFormat::F32 => build_input::<f32>(
                    &device,
                    &stream_config,
                    channels,
                    input_rate,
                    capture,
                    events.clone(),
                )?,
                cpal::SampleFormat::U16 => build_input::<u16>(
                    &device,
                    &stream_config,
                    channels,
                    input_rate,
                    capture,
                    events.clone(),
                )?,
                format => {
                    return Err(WebRtcError::Audio(format!(
                        "unsupported capture sample format {format}"
                    )));
                }
            };
            stream
                .play()
                .map_err(|error| WebRtcError::Audio(error.to_string()))?;
            Some(stream)
        };

        let (output, output_state) = if output_id == "virtual:speaker" {
            (None, None)
        } else {
            let id = output_id
                .strip_prefix("cpal:output:")
                .ok_or(WebRtcError::InvalidInput("invalid CPAL output device ID"))?;
            let device = host
                .output_devices()
                .map_err(|error| WebRtcError::Audio(error.to_string()))?
                .find(|device| {
                    device
                        .id()
                        .ok()
                        .as_ref()
                        .is_some_and(|value| value.to_string() == id)
                })
                .ok_or(WebRtcError::InvalidInput(
                    "selected CPAL output device disappeared",
                ))?;
            let config = device
                .default_output_config()
                .map_err(|error| WebRtcError::Audio(error.to_string()))?;
            let state = Arc::new(Mutex::new(RealOutputState {
                samples: VecDeque::with_capacity(96_000),
                current: 0,
                phase: 0,
                output_rate: config.sample_rate(),
                channels: usize::from(config.channels()),
            }));
            let stream_config = config.config();
            let stream = match config.sample_format() {
                cpal::SampleFormat::I16 => build_output::<i16>(
                    &device,
                    &stream_config,
                    Arc::clone(&state),
                    events.clone(),
                )?,
                cpal::SampleFormat::F32 => build_output::<f32>(
                    &device,
                    &stream_config,
                    Arc::clone(&state),
                    events.clone(),
                )?,
                cpal::SampleFormat::U16 => build_output::<u16>(
                    &device,
                    &stream_config,
                    Arc::clone(&state),
                    events.clone(),
                )?,
                format => {
                    return Err(WebRtcError::Audio(format!(
                        "unsupported playback sample format {format}"
                    )));
                }
            };
            stream
                .play()
                .map_err(|error| WebRtcError::Audio(error.to_string()))?;
            (Some(stream), Some(state))
        };
        Ok(Self {
            _input: input,
            _output: output,
            output: output_state,
        })
    }
}

#[cfg(feature = "real-audio")]
fn build_input<T>(
    device: &cpal::Device,
    config: &cpal::StreamConfig,
    channels: usize,
    input_rate: u32,
    sender: mpsc::Sender<Vec<i16>>,
    events: mpsc::Sender<VoiceEvent>,
) -> Result<cpal::Stream, WebRtcError>
where
    T: cpal::SizedSample,
    f32: cpal::FromSample<T>,
{
    use cpal::Sample as _;
    use cpal::traits::DeviceTrait;
    let mut phase = 0_u64;
    let mut frame = Vec::with_capacity(FRAME_SAMPLES);
    device
        .build_input_stream(
            *config,
            move |data: &[T], _| {
                for input_frame in data.chunks(channels) {
                    let mono = input_frame
                        .iter()
                        .copied()
                        .map(f32::from_sample)
                        .sum::<f32>()
                        / input_frame.len().max(1) as f32;
                    phase += u64::from(SAMPLE_RATE);
                    while phase >= u64::from(input_rate) {
                        phase -= u64::from(input_rate);
                        frame.push(i16::from_sample(mono.clamp(-1.0, 1.0)));
                        if frame.len() == FRAME_SAMPLES {
                            let _ = sender.try_send(std::mem::replace(
                                &mut frame,
                                Vec::with_capacity(FRAME_SAMPLES),
                            ));
                        }
                    }
                }
            },
            move |error| {
                let _ = events.try_send(VoiceEvent::Diagnostic(format!(
                    "capture stream failed: {error}"
                )));
            },
            None,
        )
        .map_err(|error| WebRtcError::Audio(error.to_string()))
}

#[cfg(feature = "real-audio")]
fn build_output<T>(
    device: &cpal::Device,
    config: &cpal::StreamConfig,
    state: Arc<Mutex<RealOutputState>>,
    events: mpsc::Sender<VoiceEvent>,
) -> Result<cpal::Stream, WebRtcError>
where
    T: cpal::SizedSample + cpal::FromSample<i16>,
{
    use cpal::traits::DeviceTrait;
    device
        .build_output_stream(
            *config,
            move |data: &mut [T], _| {
                let mut state = lock(&state);
                for output_frame in data.chunks_mut(state.channels) {
                    let sample = T::from_sample(state.current);
                    for output in output_frame {
                        *output = sample;
                    }
                    state.phase += u64::from(SAMPLE_RATE);
                    while state.phase >= u64::from(state.output_rate) {
                        state.phase -= u64::from(state.output_rate);
                        state.current = state.samples.pop_front().unwrap_or(0);
                    }
                }
            },
            move |error| {
                let _ = events.try_send(VoiceEvent::Diagnostic(format!(
                    "playback stream failed: {error}"
                )));
            },
            None,
        )
        .map_err(|error| WebRtcError::Audio(error.to_string()))
}

/// Deterministic native WebRTC answerer used by CI and offline diagnostics.
pub struct LoopbackSignaling {
    state: tokio::sync::Mutex<LoopbackState>,
    provisioning: AtomicBool,
    active_tasks: Arc<AtomicUsize>,
    received_messages: Arc<Mutex<Vec<String>>>,
    peer_id: UUID,
}

struct LoopbackState {
    shutdown: Option<oneshot::Sender<()>>,
    task: Option<JoinHandle<()>>,
    completed: bool,
    closed: bool,
}

impl LoopbackSignaling {
    #[must_use]
    pub fn new(peer_id: UUID) -> Arc<Self> {
        Arc::new(Self {
            state: tokio::sync::Mutex::new(LoopbackState {
                shutdown: None,
                task: None,
                completed: false,
                closed: false,
            }),
            provisioning: AtomicBool::new(false),
            active_tasks: Arc::new(AtomicUsize::new(0)),
            received_messages: Arc::new(Mutex::new(Vec::new())),
            peer_id,
        })
    }

    #[must_use]
    pub fn active_tasks(&self) -> usize {
        self.active_tasks.load(Ordering::Acquire)
    }

    #[must_use]
    pub fn received_messages(&self) -> Vec<String> {
        lock(&self.received_messages).clone()
    }

    pub async fn wait_closed(&self, timeout: Duration) -> Result<(), WebRtcError> {
        tokio::time::timeout(timeout, async {
            loop {
                if self.active_tasks() == 0 {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .map_err(|_| WebRtcError::Timeout("fake signaling teardown"))
    }
}

impl VoiceSignaling for LoopbackSignaling {
    fn provision(&self, request: ProvisionRequest) -> SignalFuture<'_, ProvisionResponse> {
        Box::pin(async move {
            struct ProvisionGuard<'a>(&'a AtomicBool);
            impl Drop for ProvisionGuard<'_> {
                fn drop(&mut self) {
                    self.0.store(false, Ordering::Release);
                }
            }

            if self
                .provisioning
                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
                .is_err()
            {
                return Err(WebRtcError::Signaling("duplicate provision request".into()));
            }
            let _provisioning = ProvisionGuard(&self.provisioning);
            if request.jsep.kind != "offer"
                || request.channel_type != "local"
                || request.voice_server_type != "webrtc"
            {
                return Err(WebRtcError::Protocol("invalid WebRTC provision request"));
            }
            let offer = SdpOffer::from_sdp_string(&request.jsep.sdp)
                .map_err(|error| WebRtcError::Rtc(error.to_string()))?;
            let socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await?;
            let local_addr = socket.local_addr()?;
            let mut rtc = RtcConfig::new().build(Instant::now());
            let _ = rtc.add_local_candidate(
                Candidate::host(local_addr, "udp")
                    .map_err(|error| WebRtcError::Rtc(error.to_string()))?,
            );
            let answer = rtc
                .sdp_api()
                .accept_offer(offer)
                .map_err(|error| WebRtcError::Rtc(error.to_string()))?;
            let (shutdown_tx, shutdown_rx) = oneshot::channel();
            let active = Arc::clone(&self.active_tasks);
            let messages = Arc::clone(&self.received_messages);
            let peer_id = self.peer_id;
            let viewer_session = UUID::random()
                .map_err(|_| WebRtcError::Signaling("could not create session id".into()))?;
            let credentials = VoiceSecret::new("offline-secret")?;
            let mut state = self.state.lock().await;
            if state.task.is_some() {
                return Err(WebRtcError::Signaling("duplicate provision request".into()));
            }
            active.fetch_add(1, Ordering::AcqRel);
            let task = tokio::spawn(async move {
                Box::pin(run_loopback_peer(
                    rtc,
                    socket,
                    shutdown_rx,
                    messages,
                    peer_id,
                    active,
                ))
                .await;
            });
            state.shutdown = Some(shutdown_tx);
            state.task = Some(task);
            Ok(ProvisionResponse {
                answer_sdp: answer.to_sdp_string(),
                viewer_session,
                channel: Some("offline-local".into()),
                credentials: Some(credentials),
            })
        })
    }

    fn complete(&self, request: SignalingCompleteRequest) -> SignalFuture<'_, ()> {
        Box::pin(async move {
            if request.voice_server_type != "webrtc"
                || !request.candidate.completed
                || parse_uuid(&request.viewer_session).is_none()
            {
                return Err(WebRtcError::Protocol("invalid ICE completion request"));
            }
            self.state.lock().await.completed = true;
            Ok(())
        })
    }

    fn close(&self, request: CloseRequest) -> SignalFuture<'_, ()> {
        Box::pin(async move {
            if !request.logout
                || request.voice_server_type != "webrtc"
                || parse_uuid(&request.viewer_session).is_none()
            {
                return Err(WebRtcError::Protocol("invalid voice close request"));
            }
            let (shutdown, task) = {
                let mut state = self.state.lock().await;
                if !state.completed {
                    return Err(WebRtcError::Protocol(
                        "voice session closed before signaling completion",
                    ));
                }
                state.closed = true;
                (state.shutdown.take(), state.task.take())
            };
            if let Some(shutdown) = shutdown {
                let _ = shutdown.send(());
            }
            if let Some(task) = task {
                let _ = task.await;
            }
            Ok(())
        })
    }
}

async fn run_loopback_peer(
    mut rtc: Rtc,
    socket: UdpSocket,
    mut shutdown: oneshot::Receiver<()>,
    messages: Arc<Mutex<Vec<String>>>,
    peer_id: UUID,
    active: Arc<AtomicUsize>,
) {
    let mut buffer = vec![0_u8; 65_536];
    let mut running = true;
    let mut channel_id = None;
    while running {
        let deadline = loop {
            match rtc.poll_output() {
                Ok(Output::Timeout(deadline)) => break deadline,
                Ok(Output::Transmit(transmit)) => {
                    let _ = socket
                        .send_to(&transmit.contents, transmit.destination)
                        .await;
                }
                Ok(Output::Event(Event::ChannelOpen(id, _))) => {
                    channel_id = Some(id);
                    if let Some(mut channel) = rtc.channel(id) {
                        let body = json!({
                            peer_id.to_string(): {
                                "p": 42, "V": true, "m": false, "s": 777_u32,
                                "j": {"p": true},
                                "sp": {"x": 100, "y": 200, "z": 300},
                                "sh": {"x": 0, "y": 0, "z": 0, "w": 100}
                            }
                        })
                        .to_string();
                        let _ = channel.write(false, body.as_bytes());
                        let _ = channel.write(
                            false,
                            json!({"m": {peer_id.to_string(): false}})
                                .to_string()
                                .as_bytes(),
                        );
                        let _ = channel.write(
                            false,
                            json!({"ug": {peer_id.to_string(): 180}})
                                .to_string()
                                .as_bytes(),
                        );
                        let _ = channel.write(false, b"{\"ping\":true}");
                    }
                }
                Ok(Output::Event(Event::ChannelData(data))) if !data.binary => {
                    if let Ok(text) = String::from_utf8(data.data) {
                        lock(&messages).push(text);
                    }
                }
                Ok(Output::Event(Event::MediaData(media)))
                    if media.params.spec().codec == Codec::Opus =>
                {
                    let pt = rtc
                        .writer(media.mid)
                        .and_then(|writer| writer.match_params(media.params));
                    if let Some(pt) = pt
                        && let Some(writer) = rtc.writer(media.mid)
                    {
                        let _ =
                            writer.write(pt, media.network_time, media.time, media.data.clone());
                    }
                }
                Ok(Output::Event(Event::Closed)) | Err(_) => {
                    running = false;
                    break Instant::now();
                }
                Ok(Output::Event(_)) => {}
            }
        };
        if !running {
            break;
        }
        tokio::select! {
            _ = &mut shutdown => { rtc.disconnect(); running = false; }
            receive = socket.recv_from(&mut buffer) => if let Ok((count, source)) = receive
                && let (Ok(destination), Ok(contents)) = (socket.local_addr(), buffer[..count].try_into())
            {
                let _ = rtc.handle_input(Input::Receive(Instant::now(), Receive { proto: Protocol::Udp, source, destination, contents }));
            },
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)) => { let _ = rtc.handle_input(Input::Timeout(Instant::now())); }
        }
    }
    if let Some(id) = channel_id.and_then(|id| rtc.channel(id).map(|_| id)) {
        let _ = id;
    }
    active.fetch_sub(1, Ordering::AcqRel);
}

#[cfg(test)]
mod tests {
    use super::*;

    struct InvalidAnswerSignaling {
        close_calls: AtomicUsize,
    }

    impl VoiceSignaling for InvalidAnswerSignaling {
        fn provision(&self, _: ProvisionRequest) -> SignalFuture<'_, ProvisionResponse> {
            Box::pin(async {
                Ok(ProvisionResponse {
                    answer_sdp: "not an SDP answer".into(),
                    viewer_session: UUID::new_with_string(
                        "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee".into(),
                    )
                    .unwrap(),
                    channel: Some("provisioned-channel".into()),
                    credentials: Some(VoiceSecret::new("provisioned-secret").unwrap()),
                })
            })
        }

        fn complete(&self, _: SignalingCompleteRequest) -> SignalFuture<'_, ()> {
            Box::pin(async { Err(WebRtcError::Protocol("completion must not run")) })
        }

        fn close(&self, request: CloseRequest) -> SignalFuture<'_, ()> {
            Box::pin(async move {
                assert!(request.logout);
                assert!(request.credentials.is_some());
                self.close_calls.fetch_add(1, Ordering::AcqRel);
                Ok(())
            })
        }
    }

    fn wav_fixture() -> Vec<u8> {
        let mut cursor = Cursor::new(Vec::new());
        {
            let spec = hound::WavSpec {
                channels: 2,
                sample_rate: 24_000,
                bits_per_sample: 16,
                sample_format: hound::SampleFormat::Int,
            };
            let mut writer = hound::WavWriter::new(&mut cursor, spec).unwrap();
            for index in 0..2_400 {
                let sample = (((index as f32 / 24_000.0) * 440.0 * std::f32::consts::TAU).sin()
                    * 8_000.0) as i16;
                writer.write_sample(sample).unwrap();
                writer.write_sample(sample).unwrap();
            }
            writer.finalize().unwrap();
        }
        cursor.into_inner()
    }

    #[test]
    fn wav_is_resampled_and_encoded_as_valid_opus() {
        let frames = encode_wav_bytes(&wav_fixture()).unwrap();
        assert_eq!(frames.len(), 5);
        let mut decoder = Decoder::new(SAMPLE_RATE, Channels::Mono).unwrap();
        let mut samples = vec![0_i16; FRAME_SAMPLES * 6];
        assert_eq!(
            decoder
                .decode(Some(frames[0].0.as_slice()), samples.as_mut_slice(), false)
                .unwrap(),
            FRAME_SAMPLES
        );
    }

    #[tokio::test]
    async fn loopback_exercises_secure_peer_data_media_and_teardown() {
        let peer = UUID::new_with_string("11111111-2222-4333-8444-555555555555".into()).unwrap();
        let signaling = LoopbackSignaling::new(peer);
        let mut session = WebRtcVoiceSession::connect(
            signaling.clone(),
            VoiceSessionConfig {
                timeout: Duration::from_secs(10),
                ..VoiceSessionConfig::default()
            },
        )
        .await
        .unwrap();
        tokio::time::timeout(Duration::from_secs(10), async {
            while !session.snapshot().data_channel_ready {
                assert!(session.next_event().await.is_some());
            }
        })
        .await
        .unwrap();
        let frames = encode_wav_bytes(&wav_fixture()).unwrap();
        session.play_frames(frames, false).await.unwrap();
        session.set_peer_mute(peer, true).await.unwrap();
        session.set_peer_gain(peer, 500).await.unwrap();
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                let state = session.snapshot();
                if state.data_channel_ready
                    && state.known_peers.contains(&peer)
                    && state.received_audio_frames > 0
                {
                    break;
                }
                let _ = session.next_event().await;
            }
        })
        .await
        .unwrap();
        session.shutdown().await.unwrap();
        signaling.wait_closed(Duration::from_secs(2)).await.unwrap();
        assert_eq!(session.snapshot().active_tasks, 0);
        assert!(
            signaling
                .received_messages()
                .iter()
                .any(|message| message.contains("\"ug\"") && message.contains("220"))
        );
    }

    #[tokio::test]
    async fn dropping_session_cancels_peer_and_signaling_tasks() {
        let peer = UUID::new_with_string("11111111-2222-4333-8444-555555555555".into()).unwrap();
        let signaling = LoopbackSignaling::new(peer);
        let session = WebRtcVoiceSession::connect(
            signaling.clone(),
            VoiceSessionConfig {
                timeout: Duration::from_secs(10),
                ..VoiceSessionConfig::default()
            },
        )
        .await
        .unwrap();
        let snapshot = Arc::clone(&session.snapshot);
        drop(session);
        signaling.wait_closed(Duration::from_secs(2)).await.unwrap();
        tokio::time::timeout(Duration::from_secs(2), async {
            while lock(&snapshot).active_tasks != 0 {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
        assert_eq!(signaling.active_tasks(), 0);
    }

    #[tokio::test]
    async fn concurrent_provisioning_admits_one_session_without_leaking_the_loser() {
        let peer = UUID::new_with_string("11111111-2222-4333-8444-555555555555".into()).unwrap();
        let signaling = LoopbackSignaling::new(peer);
        let config = VoiceSessionConfig {
            timeout: Duration::from_secs(10),
            ..VoiceSessionConfig::default()
        };
        let (first, second) = tokio::join!(
            WebRtcVoiceSession::connect(signaling.clone(), config.clone()),
            WebRtcVoiceSession::connect(signaling.clone(), config),
        );
        let (mut session, rejected) = match (first, second) {
            (Ok(session), Err(error)) | (Err(error), Ok(session)) => (session, error),
            (Ok(_), Ok(_)) => panic!("duplicate provisioning created two sessions"),
            (Err(first), Err(second)) => {
                panic!("both provisioning attempts failed: {first}; {second}")
            }
        };
        assert!(matches!(rejected, WebRtcError::Signaling(_)));

        session.shutdown().await.unwrap();
        signaling.wait_closed(Duration::from_secs(2)).await.unwrap();
        assert_eq!(session.snapshot().active_tasks, 0);
        assert_eq!(signaling.active_tasks(), 0);
    }

    #[tokio::test]
    async fn failed_answer_closes_provisioned_server_session() {
        let signaling = Arc::new(InvalidAnswerSignaling {
            close_calls: AtomicUsize::new(0),
        });
        let result = WebRtcVoiceSession::connect(
            signaling.clone(),
            VoiceSessionConfig {
                timeout: Duration::from_secs(2),
                ..VoiceSessionConfig::default()
            },
        )
        .await;
        assert!(matches!(result, Err(WebRtcError::Rtc(_))));
        assert_eq!(signaling.close_calls.load(Ordering::Acquire), 1);
    }

    #[test]
    fn secrets_and_protocol_shapes_are_safe() {
        assert_eq!(
            format!("{:?}", VoiceSecret::new("token").unwrap()),
            "VoiceSecret(<redacted>)"
        );
        let request = SignalingCompleteRequest {
            voice_server_type: "webrtc".into(),
            viewer_session: "id".into(),
            candidate: SignalingComplete { completed: true },
        };
        assert_eq!(
            serde_json::to_value(request).unwrap()["candidate"]["completed"],
            true
        );
    }
}