openrtc 2.5.2

//! Target-neutral OpenRTC media protocol primitives.
//!
//! This module deliberately contains no DOM, capture, codec-library, socket,
//! or physical-carrier types. Native applications and browser WASM use the
//! same framing, generation, replay, queue, and integrity rules; platform
//! edges only produce and consume encoded chunks.

use std::collections::{HashMap, VecDeque};
#[cfg(not(target_arch = "wasm32"))]
use std::{future::Future, pin::Pin};

pub const MEDIA_PROTOCOL: &str = "openrtc-media/1";
pub const PEER_STREAM_CHANNEL: &str = crate::stream_metadata::PUBLIC_PEER_STREAM_CHANNEL_ID;
pub const MEDIA_WIRE_VERSION: u8 = 1;
pub const MAX_MEDIA_CHUNK_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_CONTROL_FRAME_BYTES: usize = 64 * 1024;
pub const MAX_JAVASCRIPT_SAFE_INTEGER: u64 = (1_u64 << 53) - 1;

const CHUNK_MAGIC: [u8; 4] = *b"ORTM";
const CHUNK_HEADER_BYTES: usize = 72;
const FLAG_KEYFRAME: u8 = 1;
const FLAG_DISCARDABLE: u8 = 1 << 1;
const CAPABILITY_METADATA_KEY: &str = "openrtcCapability";
#[cfg(not(target_arch = "wasm32"))]
const MAX_NATIVE_MEDIA_WRITE_DURATION: std::time::Duration = std::time::Duration::from_secs(1);

fn capability_key(kind: &str, id: &str) -> anyhow::Result<String> {
    let kind = kind.trim();
    let id = id.trim();
    if !matches!(kind, "devices" | "space" | "room" | "ticket") {
        anyhow::bail!("media capability kind is invalid");
    }
    if id.is_empty()
        || id.len() > 160
        || !id
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || b"_.:@-".contains(&byte))
    {
        anyhow::bail!("media capability id is invalid");
    }
    Ok(format!("{kind}:{id}"))
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct PublicationId(pub [u8; 16]);

impl PublicationId {
    pub fn from_bytes(bytes: [u8; 16]) -> Self {
        Self(bytes)
    }
}

impl std::fmt::Display for PublicationId {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "{}", hex::encode(self.0))
    }
}

impl std::str::FromStr for PublicationId {
    type Err = MediaProtocolError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        let bytes = hex::decode(value).map_err(|_| MediaProtocolError::InvalidPublicationId)?;
        let publication: [u8; 16] = bytes
            .try_into()
            .map_err(|_| MediaProtocolError::InvalidPublicationId)?;
        Ok(Self(publication))
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum MediaKind {
    Audio,
    Video,
    Screen,
    Data,
}

impl MediaKind {
    fn wire(self) -> u8 {
        match self {
            Self::Audio => 1,
            Self::Video => 2,
            Self::Screen => 3,
            Self::Data => 4,
        }
    }

    fn from_wire(value: u8) -> Result<Self, MediaProtocolError> {
        match value {
            1 => Ok(Self::Audio),
            2 => Ok(Self::Video),
            3 => Ok(Self::Screen),
            4 => Ok(Self::Data),
            _ => Err(MediaProtocolError::UnsupportedKind(value)),
        }
    }
}

impl std::str::FromStr for MediaKind {
    type Err = MediaProtocolError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value.to_ascii_lowercase().as_str() {
            "audio" => Ok(Self::Audio),
            "video" => Ok(Self::Video),
            "screen" => Ok(Self::Screen),
            "data" => Ok(Self::Data),
            _ => Err(MediaProtocolError::UnsupportedKind(0)),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum MediaCodec {
    Opus,
    H264,
    Vp8,
    Vp9,
    Av1,
    Pcm,
    Opaque,
}

impl MediaCodec {
    fn wire(self) -> u8 {
        match self {
            Self::Opus => 1,
            Self::H264 => 2,
            Self::Vp8 => 3,
            Self::Vp9 => 4,
            Self::Av1 => 5,
            Self::Pcm => 6,
            Self::Opaque => 255,
        }
    }

    fn from_wire(value: u8) -> Result<Self, MediaProtocolError> {
        match value {
            1 => Ok(Self::Opus),
            2 => Ok(Self::H264),
            3 => Ok(Self::Vp8),
            4 => Ok(Self::Vp9),
            5 => Ok(Self::Av1),
            6 => Ok(Self::Pcm),
            255 => Ok(Self::Opaque),
            _ => Err(MediaProtocolError::UnsupportedCodec(value)),
        }
    }
}

impl std::str::FromStr for MediaCodec {
    type Err = MediaProtocolError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value.to_ascii_lowercase().as_str() {
            "opus" => Ok(Self::Opus),
            "h264" | "avc1" => Ok(Self::H264),
            "vp8" => Ok(Self::Vp8),
            "vp9" => Ok(Self::Vp9),
            "av1" | "av01" => Ok(Self::Av1),
            "pcm" => Ok(Self::Pcm),
            "opaque" => Ok(Self::Opaque),
            _ => Err(MediaProtocolError::UnsupportedCodec(0)),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EncodedMediaChunk {
    pub publication_id: PublicationId,
    pub media_generation: u32,
    pub sequence: u64,
    pub timestamp_us: u64,
    pub duration_us: u32,
    pub kind: MediaKind,
    pub codec: MediaCodec,
    pub keyframe: bool,
    pub discardable: bool,
    pub payload: Vec<u8>,
}

impl EncodedMediaChunk {
    pub fn encode(&self) -> Result<Vec<u8>, MediaProtocolError> {
        if self.payload.len() > MAX_MEDIA_CHUNK_BYTES {
            return Err(MediaProtocolError::ChunkTooLarge(self.payload.len()));
        }
        let mut encoded = Vec::with_capacity(CHUNK_HEADER_BYTES + self.payload.len());
        encoded.extend_from_slice(&CHUNK_MAGIC);
        encoded.push(MEDIA_WIRE_VERSION);
        encoded.push(1);
        let mut flags = 0;
        if self.keyframe {
            flags |= FLAG_KEYFRAME;
        }
        if self.discardable {
            flags |= FLAG_DISCARDABLE;
        }
        encoded.push(flags);
        encoded.push(self.kind.wire());
        encoded.push(self.codec.wire());
        encoded.extend_from_slice(&[0; 3]);
        encoded.extend_from_slice(&self.publication_id.0);
        encoded.extend_from_slice(&self.media_generation.to_be_bytes());
        encoded.extend_from_slice(&self.sequence.to_be_bytes());
        encoded.extend_from_slice(&self.timestamp_us.to_be_bytes());
        encoded.extend_from_slice(&self.duration_us.to_be_bytes());
        encoded.extend_from_slice(&(self.payload.len() as u32).to_be_bytes());
        let digest = blake3::hash(&self.payload);
        encoded.extend_from_slice(&digest.as_bytes()[..16]);
        encoded.extend_from_slice(&self.payload);
        Ok(encoded)
    }

    pub fn decode(encoded: &[u8]) -> Result<Self, MediaProtocolError> {
        if encoded.len() < CHUNK_HEADER_BYTES {
            return Err(MediaProtocolError::Truncated);
        }
        if encoded[..4] != CHUNK_MAGIC {
            return Err(MediaProtocolError::InvalidMagic);
        }
        if encoded[4] != MEDIA_WIRE_VERSION {
            return Err(MediaProtocolError::UnsupportedVersion(encoded[4]));
        }
        if encoded[5] != 1 {
            return Err(MediaProtocolError::UnsupportedFrameType(encoded[5]));
        }
        let flags = encoded[6];
        let kind = MediaKind::from_wire(encoded[7])?;
        let codec = MediaCodec::from_wire(encoded[8])?;
        let mut publication = [0; 16];
        publication.copy_from_slice(&encoded[12..28]);
        let media_generation = u32::from_be_bytes(encoded[28..32].try_into().unwrap());
        let sequence = u64::from_be_bytes(encoded[32..40].try_into().unwrap());
        let timestamp_us = u64::from_be_bytes(encoded[40..48].try_into().unwrap());
        let duration_us = u32::from_be_bytes(encoded[48..52].try_into().unwrap());
        let payload_len = u32::from_be_bytes(encoded[52..56].try_into().unwrap()) as usize;
        if payload_len > MAX_MEDIA_CHUNK_BYTES {
            return Err(MediaProtocolError::ChunkTooLarge(payload_len));
        }
        let total = CHUNK_HEADER_BYTES
            .checked_add(payload_len)
            .ok_or(MediaProtocolError::ChunkTooLarge(payload_len))?;
        if encoded.len() != total {
            return Err(MediaProtocolError::LengthMismatch {
                declared: payload_len,
                actual: encoded.len().saturating_sub(CHUNK_HEADER_BYTES),
            });
        }
        let payload = &encoded[CHUNK_HEADER_BYTES..];
        let digest = blake3::hash(payload);
        if encoded[56..72] != digest.as_bytes()[..16] {
            return Err(MediaProtocolError::IntegrityFailure);
        }
        Ok(Self {
            publication_id: PublicationId(publication),
            media_generation,
            sequence,
            timestamp_us,
            duration_us,
            kind,
            codec,
            keyframe: flags & FLAG_KEYFRAME != 0,
            discardable: flags & FLAG_DISCARDABLE != 0,
            payload: payload.to_vec(),
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", rename_all = "kebab-case")]
pub enum MediaControlFrame {
    Publish {
        publication_id: PublicationId,
        media_generation: u32,
        kind: MediaKind,
        codec: MediaCodec,
        clock_rate: u32,
        coded_width: Option<u32>,
        coded_height: Option<u32>,
        channels: Option<u16>,
    },
    SetEnabled {
        publication_id: PublicationId,
        media_generation: u32,
        enabled: bool,
    },
    RequestKeyframe {
        publication_id: PublicationId,
        media_generation: u32,
    },
    Stop {
        publication_id: PublicationId,
        media_generation: u32,
        reason: String,
    },
}

impl MediaControlFrame {
    pub fn encode(&self) -> Result<Vec<u8>, MediaProtocolError> {
        let encoded = serde_json::to_vec(self)
            .map_err(|error| MediaProtocolError::Control(error.to_string()))?;
        if encoded.len() > MAX_CONTROL_FRAME_BYTES {
            return Err(MediaProtocolError::ControlTooLarge(encoded.len()));
        }
        Ok(encoded)
    }

    pub fn decode(encoded: &[u8]) -> Result<Self, MediaProtocolError> {
        if encoded.len() > MAX_CONTROL_FRAME_BYTES {
            return Err(MediaProtocolError::ControlTooLarge(encoded.len()));
        }
        serde_json::from_slice(encoded)
            .map_err(|error| MediaProtocolError::Control(error.to_string()))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MediaProtocolError {
    InvalidPublicationId,
    InvalidMagic,
    Truncated,
    UnsupportedVersion(u8),
    UnsupportedFrameType(u8),
    UnsupportedKind(u8),
    UnsupportedCodec(u8),
    ChunkTooLarge(usize),
    ControlTooLarge(usize),
    LengthMismatch { declared: usize, actual: usize },
    IntegrityFailure,
    Replay,
    RetiredGeneration,
    Control(String),
}

impl std::fmt::Display for MediaProtocolError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "{self:?}")
    }
}

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

pub fn ensure_javascript_safe_integer(value: u64, field: &str) -> Result<u64, MediaProtocolError> {
    if value > MAX_JAVASCRIPT_SAFE_INTEGER {
        return Err(MediaProtocolError::Control(format!(
            "media {field} exceeds the JavaScript safe integer boundary"
        )));
    }
    Ok(value)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MediaGenerationFence {
    pub transport_stable_id: u64,
    pub transport_generation: u64,
    pub route_generation: u64,
    pub media_generation: u32,
}

impl MediaGenerationFence {
    pub fn accepts(self, current: Self) -> bool {
        self == current
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct MediaStats {
    pub accepted: u64,
    pub queued: u64,
    pub transmitted: u64,
    pub received: u64,
    pub decoded: u64,
    pub rendered: u64,
    pub dropped_stale: u64,
    pub dropped_overflow: u64,
    pub integrity_failures: u64,
    pub malformed_frames: u64,
    pub replay_drops: u64,
    pub video_recoveries: u64,
    pub queued_bytes: usize,
    pub max_queued: u64,
    pub max_queued_bytes: usize,
    pub max_queue_age_ms: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MediaDependencyDecision {
    Accept,
    DropUntilKeyframe,
    Recovered,
}

#[derive(Debug)]
struct MediaDependencyGate {
    visual: bool,
    needs_keyframe: bool,
    recovery_pending: bool,
    highest_sequence: Option<u64>,
}

impl MediaDependencyGate {
    fn new(kind: MediaKind) -> Self {
        let visual = matches!(kind, MediaKind::Video | MediaKind::Screen);
        Self {
            visual,
            needs_keyframe: visual,
            recovery_pending: false,
            highest_sequence: None,
        }
    }

    fn observe_discontinuity(&mut self) {
        if self.visual {
            self.needs_keyframe = true;
            self.recovery_pending = true;
        }
    }

    fn observe_chunk(&mut self, chunk: &EncodedMediaChunk) -> MediaDependencyDecision {
        if let Some(highest) = self.highest_sequence {
            if chunk.sequence > highest.saturating_add(1) {
                self.observe_discontinuity();
            }
            self.highest_sequence = Some(highest.max(chunk.sequence));
        } else {
            self.highest_sequence = Some(chunk.sequence);
        }

        if !self.visual || !self.needs_keyframe {
            return MediaDependencyDecision::Accept;
        }
        if !chunk.keyframe {
            return MediaDependencyDecision::DropUntilKeyframe;
        }
        self.needs_keyframe = false;
        if std::mem::take(&mut self.recovery_pending) {
            MediaDependencyDecision::Recovered
        } else {
            MediaDependencyDecision::Accept
        }
    }
}

impl MediaStats {
    fn record_decode_failure(&mut self, error: &MediaProtocolError) {
        if *error == MediaProtocolError::IntegrityFailure {
            self.integrity_failures = self.integrity_failures.saturating_add(1);
        } else {
            self.malformed_frames = self.malformed_frames.saturating_add(1);
        }
    }
}

#[derive(Debug, Clone)]
struct QueuedChunk {
    chunk: EncodedMediaChunk,
    accepted_at_ms: u64,
    bytes: usize,
}

/// Bounded live-media queue. Stale discardable video is removed before audio.
#[derive(Debug)]
pub struct MediaQueue {
    items: VecDeque<QueuedChunk>,
    max_items: usize,
    max_bytes: usize,
    max_age_ms: u64,
    stats: MediaStats,
}

impl MediaQueue {
    pub fn new(max_items: usize, max_bytes: usize, max_age_ms: u64) -> Self {
        Self {
            items: VecDeque::new(),
            max_items: max_items.max(1),
            max_bytes: max_bytes.max(1),
            max_age_ms: max_age_ms.max(1),
            stats: MediaStats::default(),
        }
    }

    pub fn push(&mut self, chunk: EncodedMediaChunk, now_ms: u64) {
        self.drop_stale(now_ms);
        let bytes = chunk.payload.len().saturating_add(CHUNK_HEADER_BYTES);
        self.stats.accepted = self.stats.accepted.saturating_add(1);
        if bytes > self.max_bytes {
            self.stats.dropped_overflow = self.stats.dropped_overflow.saturating_add(1);
            return;
        }
        while self.items.len() >= self.max_items
            || self.stats.queued_bytes.saturating_add(bytes) > self.max_bytes
        {
            let removable = self
                .items
                .iter()
                .position(|entry| entry.chunk.kind != MediaKind::Audio && entry.chunk.discardable);
            let removed = removable
                .and_then(|index| self.items.remove(index))
                .or_else(|| self.items.pop_front());
            let Some(removed) = removed else {
                break;
            };
            self.stats.queued_bytes = self.stats.queued_bytes.saturating_sub(removed.bytes);
            self.stats.dropped_overflow = self.stats.dropped_overflow.saturating_add(1);
        }
        self.stats.queued_bytes = self.stats.queued_bytes.saturating_add(bytes);
        self.items.push_back(QueuedChunk {
            chunk,
            accepted_at_ms: now_ms,
            bytes,
        });
        self.update_queue_gauges();
    }

    pub fn pop(&mut self, now_ms: u64) -> Option<EncodedMediaChunk> {
        self.drop_stale(now_ms);
        let entry = self.items.pop_front()?;
        self.stats.max_queue_age_ms = self
            .stats
            .max_queue_age_ms
            .max(now_ms.saturating_sub(entry.accepted_at_ms));
        self.stats.queued_bytes = self.stats.queued_bytes.saturating_sub(entry.bytes);
        self.update_queue_gauges();
        Some(entry.chunk)
    }

    pub fn stats(&self) -> MediaStats {
        self.stats
    }

    fn drop_stale(&mut self, now_ms: u64) {
        let mut removed = 0_u64;
        let mut removed_bytes = 0_usize;
        self.items.retain(|entry| {
            let keep = now_ms.saturating_sub(entry.accepted_at_ms) <= self.max_age_ms;
            if !keep {
                removed = removed.saturating_add(1);
                removed_bytes = removed_bytes.saturating_add(entry.bytes);
            }
            keep
        });
        self.stats.queued_bytes = self.stats.queued_bytes.saturating_sub(removed_bytes);
        self.stats.dropped_stale = self.stats.dropped_stale.saturating_add(removed);
        self.update_queue_gauges();
    }

    fn update_queue_gauges(&mut self) {
        self.stats.queued = self.items.len() as u64;
        self.stats.max_queued = self.stats.max_queued.max(self.stats.queued);
        self.stats.max_queued_bytes = self.stats.max_queued_bytes.max(self.stats.queued_bytes);
    }
}

#[derive(Debug, Clone, Copy, Default)]
struct ReplayState {
    generation: u32,
    highest: u64,
    seen: u128,
    initialized: bool,
}

/// Generation-aware bounded replay window for incoming encoded chunks.
#[derive(Debug, Default)]
pub struct MediaReplayGuard {
    publications: HashMap<PublicationId, ReplayState>,
}

impl MediaReplayGuard {
    pub fn accept(&mut self, chunk: &EncodedMediaChunk) -> Result<(), MediaProtocolError> {
        let state = self.publications.entry(chunk.publication_id).or_default();
        if state.initialized && chunk.media_generation < state.generation {
            return Err(MediaProtocolError::RetiredGeneration);
        }
        if !state.initialized || chunk.media_generation > state.generation {
            *state = ReplayState {
                generation: chunk.media_generation,
                highest: chunk.sequence,
                seen: 1,
                initialized: true,
            };
            return Ok(());
        }
        if chunk.sequence > state.highest {
            let shift = chunk.sequence.saturating_sub(state.highest).min(128) as u32;
            state.seen = (if shift == 128 { 0 } else { state.seen << shift }) | 1;
            state.highest = chunk.sequence;
            return Ok(());
        }
        let distance = state.highest - chunk.sequence;
        if distance >= 128 || state.seen & (1_u128 << distance) != 0 {
            return Err(MediaProtocolError::Replay);
        }
        state.seen |= 1_u128 << distance;
        Ok(())
    }
}

#[derive(Debug, Clone)]
struct PortableMediaSenderState {
    publication: MediaPublicationConfig,
    next_sequence: u64,
    active: bool,
}

#[derive(Debug, Clone)]
struct PortableMediaReceiverState {
    publication: MediaPublicationConfig,
    active: bool,
}

/// One target-neutral receiver admission and replay owner. Native
/// `MediaConnection` instances and the WASM portable session both use this
/// state so a replacement cannot be interpreted differently at either edge.
#[derive(Debug, Default)]
struct MediaReceiverAdmission {
    publications: HashMap<PublicationId, PortableMediaReceiverState>,
    replay: MediaReplayGuard,
}

impl MediaReceiverAdmission {
    fn decode_control(&mut self, encoded: &[u8]) -> Result<MediaControlFrame, MediaProtocolError> {
        let control = MediaControlFrame::decode(encoded)?;
        self.apply_control(&control)?;
        Ok(control)
    }

    fn apply_control(&mut self, control: &MediaControlFrame) -> Result<(), MediaProtocolError> {
        match control {
            MediaControlFrame::Publish {
                publication_id,
                media_generation,
                kind,
                codec,
                clock_rate,
                coded_width,
                coded_height,
                channels,
            } => {
                if let Some(current) = self.publications.get(publication_id) {
                    if *media_generation < current.publication.media_generation {
                        return Err(MediaProtocolError::RetiredGeneration);
                    }
                    if *media_generation == current.publication.media_generation {
                        return Err(MediaProtocolError::Control(
                            "duplicate media publication generation".to_string(),
                        ));
                    }
                }
                self.publications.insert(
                    *publication_id,
                    PortableMediaReceiverState {
                        publication: MediaPublicationConfig {
                            publication_id: *publication_id,
                            media_generation: *media_generation,
                            kind: *kind,
                            codec: *codec,
                            clock_rate: *clock_rate,
                            coded_width: *coded_width,
                            coded_height: *coded_height,
                            channels: *channels,
                        },
                        active: true,
                    },
                );
            }
            MediaControlFrame::Stop {
                publication_id,
                media_generation,
                ..
            } => self.retire(*publication_id, *media_generation),
            MediaControlFrame::SetEnabled { .. } | MediaControlFrame::RequestKeyframe { .. } => {}
        }
        Ok(())
    }

    fn decode_chunk(&mut self, encoded: &[u8]) -> Result<EncodedMediaChunk, MediaProtocolError> {
        let chunk = EncodedMediaChunk::decode(encoded)?;
        self.accept_chunk(&chunk)?;
        Ok(chunk)
    }

    fn accept_chunk(&mut self, chunk: &EncodedMediaChunk) -> Result<(), MediaProtocolError> {
        let admitted = self
            .publications
            .get(&chunk.publication_id)
            .filter(|state| state.active)
            .ok_or_else(|| {
                MediaProtocolError::Control(
                    "media chunk does not match an active publication".to_string(),
                )
            })?;
        if admitted.publication.media_generation != chunk.media_generation {
            return Err(MediaProtocolError::RetiredGeneration);
        }
        if admitted.publication.kind != chunk.kind || admitted.publication.codec != chunk.codec {
            return Err(MediaProtocolError::Control(
                "media chunk changed immutable publication metadata".to_string(),
            ));
        }
        self.replay.accept(chunk)
    }

    fn retire(&mut self, publication_id: PublicationId, media_generation: u32) {
        if let Some(current) = self.publications.get_mut(&publication_id) {
            if current.publication.media_generation == media_generation {
                // Keep the publication and replay high-water marks as a
                // tombstone for the lifetime of this logical media session.
                current.active = false;
            }
        }
    }
}

/// Shared target-neutral state for browser-produced media carried by either
/// WASM directly or a thin native IPC projection.
///
/// Owner: OpenRTC Rust core.
/// Consumers: the WASM bridge and native host adapters such as
/// `openrtc-tauri-plugin`.
/// Introduced: 2026-08-26.
///
/// Platform edges own capture, codecs, and rendering. This state alone owns
/// publication generations, sequence numbers, receiver admission, replay
/// protection, and integrity validation so native frontends never recreate
/// protocol lifecycle in TypeScript.
#[derive(Debug, Default)]
pub struct PortableMediaSession {
    senders: HashMap<PublicationId, PortableMediaSenderState>,
    receivers: MediaReceiverAdmission,
}

impl PortableMediaSession {
    pub fn begin_publication(
        &mut self,
        mut publication: MediaPublicationConfig,
    ) -> Result<(MediaPublicationConfig, Vec<u8>), MediaProtocolError> {
        publication.media_generation =
            self.senders
                .get(&publication.publication_id)
                .map_or(Ok(1), |state| {
                    state
                        .publication
                        .media_generation
                        .checked_add(1)
                        .ok_or_else(|| {
                            MediaProtocolError::Control(
                                "media publication generation exhausted".to_string(),
                            )
                        })
                })?;
        let control = publication.control().encode()?;
        self.senders.insert(
            publication.publication_id,
            PortableMediaSenderState {
                publication: publication.clone(),
                next_sequence: 0,
                active: true,
            },
        );
        Ok((publication, control))
    }

    pub fn encode_sample(
        &mut self,
        publication_id: PublicationId,
        sample: EncodedMediaSample,
    ) -> Result<Vec<u8>, MediaProtocolError> {
        let state = self
            .senders
            .get_mut(&publication_id)
            .ok_or_else(|| MediaProtocolError::Control("unknown media publication".to_string()))?;
        if !state.active {
            return Err(MediaProtocolError::Control(
                "media publication is paused".to_string(),
            ));
        }
        let next_sequence = state.next_sequence.checked_add(1).ok_or_else(|| {
            MediaProtocolError::Control("media publication sequence exhausted".to_string())
        })?;
        let encoded = EncodedMediaChunk {
            publication_id,
            media_generation: state.publication.media_generation,
            sequence: state.next_sequence,
            timestamp_us: sample.timestamp_us,
            duration_us: sample.duration_us,
            kind: state.publication.kind,
            codec: state.publication.codec,
            keyframe: sample.keyframe,
            discardable: sample.discardable,
            payload: sample.payload,
        }
        .encode()?;
        state.next_sequence = next_sequence;
        Ok(encoded)
    }

    pub fn pause_publication(
        &mut self,
        publication_id: PublicationId,
    ) -> Result<(), MediaProtocolError> {
        let state = self
            .senders
            .get_mut(&publication_id)
            .ok_or_else(|| MediaProtocolError::Control("unknown media publication".to_string()))?;
        state.active = false;
        Ok(())
    }

    pub fn retire_publication(&mut self, publication_id: PublicationId) {
        if let Some(state) = self.senders.get_mut(&publication_id) {
            // Keep the generation high-water mark so reusing a stable
            // publication ID cannot reopen generation 1 to delayed traffic.
            state.active = false;
        }
    }

    pub fn retire_receiver(&mut self, publication_id: PublicationId, media_generation: u32) {
        self.receivers.retire(publication_id, media_generation);
    }

    pub fn decode_control(
        &mut self,
        encoded: &[u8],
    ) -> Result<MediaControlFrame, MediaProtocolError> {
        self.receivers.decode_control(encoded)
    }

    pub fn decode_chunk(
        &mut self,
        encoded: &[u8],
    ) -> Result<EncodedMediaChunk, MediaProtocolError> {
        self.receivers.decode_chunk(encoded)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MediaPublicationConfig {
    pub publication_id: PublicationId,
    pub media_generation: u32,
    pub kind: MediaKind,
    pub codec: MediaCodec,
    pub clock_rate: u32,
    pub coded_width: Option<u32>,
    pub coded_height: Option<u32>,
    pub channels: Option<u16>,
}

impl MediaPublicationConfig {
    pub fn control(&self) -> MediaControlFrame {
        MediaControlFrame::Publish {
            publication_id: self.publication_id,
            media_generation: self.media_generation,
            kind: self.kind,
            codec: self.codec,
            clock_rate: self.clock_rate,
            coded_width: self.coded_width,
            coded_height: self.coded_height,
            channels: self.channels,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EncodedMediaSample {
    pub timestamp_us: u64,
    pub duration_us: u32,
    pub keyframe: bool,
    pub discardable: bool,
    pub payload: Vec<u8>,
}

/// Native capture/encoder edge for the transport-agnostic media session.
///
/// Implementations may wrap a camera, microphone, file, software codec, or
/// hardware codec. OpenRTC owns publication and transport lifecycle; a source
/// only reports its encoded publication contract and yields bounded samples.
#[cfg(not(target_arch = "wasm32"))]
pub trait MediaSource: Send {
    fn publication(&self) -> MediaPublicationConfig;

    fn next_sample(
        &mut self,
    ) -> Pin<Box<dyn Future<Output = anyhow::Result<Option<EncodedMediaSample>>> + Send + '_>>;
}

/// Native decoder/render edge. A successful call means the chunk was decoded
/// and committed to playout; transport-specific handles never cross this port.
#[cfg(not(target_arch = "wasm32"))]
pub trait MediaSink: Send {
    fn render(
        &mut self,
        publication: &MediaPublicationConfig,
        chunk: EncodedMediaChunk,
    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + '_>>;
}

/// Native Rust facade over the same admitted protected peer stream used by the
/// browser edge. Capture and codec adapters provide encoded samples; this type
/// owns only publication generation, framing, and bounded transport writes.
pub struct MediaSender {
    client: std::sync::Arc<crate::Client>,
    peer_id: String,
    capability_key: Option<String>,
    connection_id: String,
    endpoint_id: iroh::EndpointId,
    transport_stable_id: u64,
    config: MediaPublicationConfig,
    sequence: u64,
    send: Option<crate::application_crypto_streams::PeerSendStream>,
    stats: MediaStats,
    #[cfg(not(target_arch = "wasm32"))]
    source: Option<Box<dyn MediaSource>>,
}

impl MediaSender {
    async fn open(
        client: std::sync::Arc<crate::Client>,
        peer_id: String,
        capability_key: Option<String>,
        config: MediaPublicationConfig,
    ) -> anyhow::Result<Self> {
        let (snapshot, endpoint_id, mut send, recv) =
            client.open_peer_bi_current(&peer_id, Some(10_000)).await?;
        let connection_id = snapshot.active_connection_id.ok_or_else(|| {
            anyhow::anyhow!("media peer {peer_id} has no active logical connection")
        })?;
        let transport_stable_id = snapshot.active_transport_stable_id.ok_or_else(|| {
            anyhow::anyhow!("media peer {peer_id} has no active transport generation")
        })?;
        // The reverse half is reserved for future receiver constraints and
        // keyframe requests. Version 1 sends no reverse control yet.
        drop(recv);
        let envelope = encode_media_stream_envelope(capability_key.as_deref())?;
        #[cfg(not(target_arch = "wasm32"))]
        within_media_write_deadline(async {
            send.write_all(&envelope).await?;
            write_media_packet(&mut send, &config.control().encode()?).await
        })
        .await?;
        #[cfg(target_arch = "wasm32")]
        {
            send.write_all(&envelope).await?;
            write_media_packet(&mut send, &config.control().encode()?).await?;
        }
        Ok(Self {
            client,
            peer_id,
            capability_key,
            connection_id,
            endpoint_id,
            transport_stable_id,
            config,
            sequence: 0,
            send: Some(send),
            stats: MediaStats::default(),
            #[cfg(not(target_arch = "wasm32"))]
            source: None,
        })
    }

    pub fn publication(&self) -> &MediaPublicationConfig {
        &self.config
    }

    pub async fn send(&mut self, sample: EncodedMediaSample) -> anyhow::Result<()> {
        self.stats.accepted = self.stats.accepted.saturating_add(1);
        let mut recovered = false;
        loop {
            let next_sequence = self
                .sequence
                .checked_add(1)
                .ok_or_else(|| anyhow::anyhow!("media publication sequence exhausted"))?;
            let chunk = EncodedMediaChunk {
                publication_id: self.config.publication_id,
                media_generation: self.config.media_generation,
                sequence: self.sequence,
                timestamp_us: sample.timestamp_us,
                duration_us: sample.duration_us,
                kind: self.config.kind,
                codec: self.config.codec,
                keyframe: sample.keyframe,
                discardable: sample.discardable,
                payload: sample.payload.clone(),
            };
            let encoded = chunk.encode()?;
            let send = self
                .send
                .as_mut()
                .ok_or_else(|| anyhow::anyhow!("media sender is stopped"))?;
            self.stats.queued = 1;
            self.stats.queued_bytes = encoded.len();
            #[cfg(not(target_arch = "wasm32"))]
            let write = within_media_write_deadline(write_media_packet(send, &encoded)).await;
            #[cfg(target_arch = "wasm32")]
            let write = write_media_packet(send, &encoded).await;
            self.stats.queued = 0;
            self.stats.queued_bytes = 0;
            match write {
                Ok(()) => {
                    self.sequence = next_sequence;
                    self.stats.transmitted = self.stats.transmitted.saturating_add(1);
                    return Ok(());
                }
                Err(error) if !recovered => {
                    recovered = true;
                    self.recover_after_transport_loss().await.map_err(|recovery| {
                        anyhow::anyhow!(
                            "media write failed ({error}); physical-generation recovery failed: {recovery}"
                        )
                    })?;
                }
                Err(error) => return Err(error),
            }
        }
    }

    /// Pull and transmit one sample from the installed native source. Returns
    /// `false` after the source reaches its natural end.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn send_next(&mut self) -> anyhow::Result<bool> {
        let source = self
            .source
            .as_mut()
            .ok_or_else(|| anyhow::anyhow!("media sender has no native source"))?;
        let Some(sample) = source.next_sample().await? else {
            return Ok(false);
        };
        self.send(sample).await?;
        Ok(true)
    }

    pub async fn replace(&mut self, mut config: MediaPublicationConfig) -> anyhow::Result<()> {
        self.finish_current()?;
        config.publication_id = self.config.publication_id;
        config.media_generation = self
            .config
            .media_generation
            .checked_add(1)
            .ok_or_else(|| anyhow::anyhow!("media publication generation exhausted"))?;
        let mut replacement = Self::open(
            self.client.clone(),
            self.peer_id.clone(),
            self.capability_key.clone(),
            config,
        )
        .await?;
        replacement.stats = self.stats;
        *self = replacement;
        Ok(())
    }

    async fn recover_after_transport_loss(&mut self) -> anyhow::Result<()> {
        let prior_connection_id = self.connection_id.clone();
        let prior_endpoint_id = self.endpoint_id;
        let prior_transport_stable_id = self.transport_stable_id;
        self.send.take();
        self.client
            .wait_for_peer_transport_replacement(
                &self.peer_id,
                &prior_connection_id,
                prior_endpoint_id,
                prior_transport_stable_id,
                10_000,
            )
            .await
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "peer {} did not settle a replacement transport within 10000ms",
                    self.peer_id
                )
            })?;
        let mut config = self.config.clone();
        config.media_generation = config
            .media_generation
            .checked_add(1)
            .ok_or_else(|| anyhow::anyhow!("media publication generation exhausted"))?;
        let mut replacement = Self::open(
            self.client.clone(),
            self.peer_id.clone(),
            self.capability_key.clone(),
            config,
        )
        .await?;
        replacement.stats = self.stats;
        *self = replacement;
        Ok(())
    }

    /// Replace the native capture/codec edge while retaining publication ID
    /// and advancing only the media generation.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn replace_track(&mut self, source: Box<dyn MediaSource>) -> anyhow::Result<()> {
        let config = source.publication();
        self.replace(config).await?;
        self.source = Some(source);
        Ok(())
    }

    pub fn stats(&self) -> MediaStats {
        self.stats
    }

    pub fn stop(&mut self) -> anyhow::Result<()> {
        self.finish_current()
    }

    fn finish_current(&mut self) -> anyhow::Result<()> {
        if let Some(send) = self.send.take() {
            send.finish()?;
        }
        Ok(())
    }
}

impl Drop for MediaSender {
    fn drop(&mut self) {
        let _ = self.finish_current();
    }
}

#[derive(Clone)]
pub struct MediaConnection {
    client: std::sync::Arc<crate::Client>,
    peer_id: String,
    capability_key: Option<String>,
    #[cfg(not(target_arch = "wasm32"))]
    receiver_admission: std::sync::Arc<std::sync::Mutex<MediaReceiverAdmission>>,
}

impl MediaConnection {
    pub fn new(client: std::sync::Arc<crate::Client>, peer_id: impl Into<String>) -> Self {
        Self {
            client,
            peer_id: peer_id.into(),
            capability_key: None,
            #[cfg(not(target_arch = "wasm32"))]
            receiver_admission: Default::default(),
        }
    }

    /// Bind native media to the same public avenue capability as the logical
    /// peer connection. Browser and multi-WebView consumers reject unscoped
    /// streams, so cross-runtime applications should use this constructor (or
    /// a native capability handle's `media_connection` convenience method).
    pub fn for_capability(
        client: std::sync::Arc<crate::Client>,
        peer_id: impl Into<String>,
        kind: &str,
        id: &str,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            client,
            peer_id: peer_id.into(),
            capability_key: Some(capability_key(kind, id)?),
            #[cfg(not(target_arch = "wasm32"))]
            receiver_admission: Default::default(),
        })
    }

    pub async fn add_encoded_track(
        &self,
        config: MediaPublicationConfig,
    ) -> anyhow::Result<MediaSender> {
        MediaSender::open(
            self.client.clone(),
            self.peer_id.clone(),
            self.capability_key.clone(),
            config,
        )
        .await
    }

    /// Add a native encoded media source to this logical peer connection.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn add_track(&self, source: Box<dyn MediaSource>) -> anyhow::Result<MediaSender> {
        let config = source.publication();
        let mut sender = MediaSender::open(
            self.client.clone(),
            self.peer_id.clone(),
            self.capability_key.clone(),
            config,
        )
        .await?;
        sender.source = Some(source);
        Ok(sender)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub fn remove_track(&self, sender: &mut MediaSender) -> anyhow::Result<()> {
        sender.stop()
    }

    /// Secure one media stream selected by the host's single incoming
    /// dispatcher from `Client::incoming_streams`. The method validates the
    /// channel envelope again before decoding, but never installs a second
    /// raw-stream consumer.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn accept_incoming(
        &self,
        incoming: crate::native_node::IncomingStream,
    ) -> anyhow::Result<MediaReceiver> {
        let (snapshot, expected_endpoint_id) = self
            .client
            .resolve_settled_peer_endpoint(&self.peer_id, Some(0))
            .await?;
        if incoming.endpoint_id != expected_endpoint_id
            || snapshot.active_transport_stable_id != Some(incoming.transport_stable_id)
        {
            anyhow::bail!("incoming media stream belongs to a retired peer generation");
        }
        let crate::native_node::IncomingStreamType::Bi(send, recv) = incoming.stream else {
            anyhow::bail!("portable media requires a bidirectional control/data stream");
        };
        let (send, mut recv) = self
            .client
            .wrap_bi_with_prefix(
                &incoming.endpoint_id,
                incoming.transport_stable_id,
                send,
                recv,
                &incoming.recv_prefix,
            )
            .await?;
        send.finish()?;
        if !read_and_validate_media_envelope(&mut recv, self.capability_key.as_deref()).await? {
            anyhow::bail!("incoming stream is not {MEDIA_PROTOCOL}");
        }
        MediaReceiver::from_stream_with_admission(recv, self.receiver_admission.clone()).await
    }
}

/// Generation/replay/integrity-aware decoder for an already-admitted incoming
/// media stream. The caller obtains the stream from OpenRTC's single incoming
/// classifier; this type never reads the physical connection directly.
pub struct MediaReceiver {
    recv: crate::application_crypto_streams::PeerRecvStream,
    publication: MediaPublicationConfig,
    admission: std::sync::Arc<std::sync::Mutex<MediaReceiverAdmission>>,
    dependency_gate: MediaDependencyGate,
    retired: bool,
    stats: MediaStats,
}

impl MediaReceiver {
    pub async fn from_stream(
        recv: crate::application_crypto_streams::PeerRecvStream,
    ) -> anyhow::Result<Self> {
        Self::from_stream_with_admission(recv, Default::default()).await
    }

    async fn from_stream_with_admission(
        mut recv: crate::application_crypto_streams::PeerRecvStream,
        admission: std::sync::Arc<std::sync::Mutex<MediaReceiverAdmission>>,
    ) -> anyhow::Result<Self> {
        let encoded = read_media_packet(&mut recv).await?;
        let control = admission
            .lock()
            .map_err(|_| anyhow::anyhow!("media receiver admission lock poisoned"))?
            .decode_control(&encoded)?;
        let MediaControlFrame::Publish {
            publication_id,
            media_generation,
            kind,
            codec,
            clock_rate,
            coded_width,
            coded_height,
            channels,
        } = control
        else {
            anyhow::bail!("media stream must begin with publish control");
        };
        Ok(Self {
            recv,
            dependency_gate: MediaDependencyGate::new(kind),
            publication: MediaPublicationConfig {
                publication_id,
                media_generation,
                kind,
                codec,
                clock_rate,
                coded_width,
                coded_height,
                channels,
            },
            admission,
            retired: false,
            stats: MediaStats::default(),
        })
    }

    pub fn publication(&self) -> &MediaPublicationConfig {
        &self.publication
    }

    pub async fn next_chunk(&mut self) -> anyhow::Result<Option<EncodedMediaChunk>> {
        loop {
            let encoded = match read_media_packet_or_eof(&mut self.recv).await? {
                Some(encoded) => encoded,
                None => {
                    self.retire();
                    return Ok(None);
                }
            };
            let chunk = match EncodedMediaChunk::decode(&encoded) {
                Ok(chunk) => chunk,
                Err(error) => {
                    self.stats.record_decode_failure(&error);
                    self.dependency_gate.observe_discontinuity();
                    continue;
                }
            };
            if let Err(error) = self
                .admission
                .lock()
                .map_err(|_| anyhow::anyhow!("media receiver admission lock poisoned"))?
                .accept_chunk(&chunk)
            {
                match error {
                    MediaProtocolError::Replay => {
                        self.stats.replay_drops = self.stats.replay_drops.saturating_add(1);
                        continue;
                    }
                    MediaProtocolError::RetiredGeneration => {
                        self.stats.dropped_stale = self.stats.dropped_stale.saturating_add(1);
                        continue;
                    }
                    _ => return Err(error.into()),
                }
            }
            self.stats.received = self.stats.received.saturating_add(1);
            match self.dependency_gate.observe_chunk(&chunk) {
                MediaDependencyDecision::Accept => return Ok(Some(chunk)),
                MediaDependencyDecision::DropUntilKeyframe => {
                    self.stats.dropped_stale = self.stats.dropped_stale.saturating_add(1);
                }
                MediaDependencyDecision::Recovered => {
                    self.stats.video_recoveries = self.stats.video_recoveries.saturating_add(1);
                    return Ok(Some(chunk));
                }
            }
        }
    }

    /// Decode/render the next protected chunk through a native platform sink.
    /// Returns `false` when the remote publication ends cleanly.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn render_next(&mut self, sink: &mut dyn MediaSink) -> anyhow::Result<bool> {
        let Some(chunk) = self.next_chunk().await? else {
            return Ok(false);
        };
        sink.render(&self.publication, chunk).await?;
        self.stats.decoded = self.stats.decoded.saturating_add(1);
        self.stats.rendered = self.stats.rendered.saturating_add(1);
        Ok(true)
    }

    pub fn stats(&self) -> MediaStats {
        self.stats
    }

    fn retire(&mut self) {
        if self.retired {
            return;
        }
        self.retired = true;
        if let Ok(mut admission) = self.admission.lock() {
            admission.retire(
                self.publication.publication_id,
                self.publication.media_generation,
            );
        }
    }
}

impl Drop for MediaReceiver {
    fn drop(&mut self) {
        self.retire();
    }
}

async fn write_media_packet(
    send: &mut crate::application_crypto_streams::PeerSendStream,
    payload: &[u8],
) -> anyhow::Result<()> {
    if payload.len() > MAX_MEDIA_CHUNK_BYTES + CHUNK_HEADER_BYTES {
        anyhow::bail!("media packet exceeds the bounded stream frame limit");
    }
    send.write_all(&(payload.len() as u32).to_be_bytes())
        .await?;
    send.write_all(payload).await?;
    Ok(())
}

#[cfg(not(target_arch = "wasm32"))]
async fn within_media_write_deadline<F>(write: F) -> anyhow::Result<()>
where
    F: Future<Output = anyhow::Result<()>>,
{
    tokio::time::timeout(MAX_NATIVE_MEDIA_WRITE_DURATION, write)
        .await
        .map_err(|_| {
            anyhow::anyhow!(
                "media protected write exceeded the {}ms queue-age bound",
                MAX_NATIVE_MEDIA_WRITE_DURATION.as_millis(),
            )
        })?
}

async fn read_media_packet(
    recv: &mut crate::application_crypto_streams::PeerRecvStream,
) -> anyhow::Result<Vec<u8>> {
    read_media_packet_or_eof(recv)
        .await?
        .ok_or_else(|| anyhow::anyhow!("media stream ended before its first frame"))
}

async fn read_media_packet_or_eof(
    recv: &mut crate::application_crypto_streams::PeerRecvStream,
) -> anyhow::Result<Option<Vec<u8>>> {
    let mut length = [0u8; 4];
    let mut offset = 0;
    while offset < length.len() {
        let read = recv.read(&mut length[offset..]).await?;
        if read == 0 {
            return if offset == 0 {
                Ok(None)
            } else {
                Err(anyhow::anyhow!("truncated media frame length"))
            };
        }
        offset += read;
    }
    let length = u32::from_be_bytes(length) as usize;
    if length > MAX_MEDIA_CHUNK_BYTES + CHUNK_HEADER_BYTES {
        anyhow::bail!("incoming media packet exceeds the bounded stream frame limit");
    }
    let mut payload = vec![0u8; length];
    let mut offset = 0;
    while offset < payload.len() {
        let read = recv.read(&mut payload[offset..]).await?;
        if read == 0 {
            anyhow::bail!("truncated media frame payload");
        }
        offset += read;
    }
    Ok(Some(payload))
}

#[cfg(not(target_arch = "wasm32"))]
async fn read_and_validate_media_envelope(
    recv: &mut crate::application_crypto_streams::PeerRecvStream,
    expected_capability_key: Option<&str>,
) -> anyhow::Result<bool> {
    let mut header = [0_u8; crate::stream_metadata::CHANNEL_ENVELOPE_HEADER_BYTES];
    read_exact_peer(recv, &mut header).await?;
    if header[0] != crate::stream_metadata::CHANNEL_ENVELOPE_MAGIC
        || header[1] != crate::stream_metadata::CHANNEL_ENVELOPE_VERSION
    {
        return Ok(false);
    }
    let channel_len = u16::from_be_bytes([header[2], header[3]]) as usize;
    let metadata_len = u16::from_be_bytes([header[4], header[5]]) as usize;
    if channel_len == 0 || channel_len > 128 || metadata_len > 4 * 1024 {
        anyhow::bail!("incoming media stream envelope exceeds its bounds");
    }
    let mut body = vec![0_u8; channel_len + metadata_len];
    read_exact_peer(recv, &mut body).await?;
    let mut envelope = Vec::with_capacity(header.len() + body.len());
    envelope.extend_from_slice(&header);
    envelope.extend_from_slice(&body);
    Ok(matches_media_stream_envelope(
        &envelope,
        expected_capability_key,
    ))
}

fn encode_media_stream_envelope(
    capability_key: Option<&str>,
) -> Result<Vec<u8>, crate::stream_metadata::EncodeError> {
    let mut metadata =
        serde_json::Map::from_iter([("protocol".to_string(), serde_json::json!(MEDIA_PROTOCOL))]);
    if let Some(capability_key) = capability_key {
        metadata.insert(
            CAPABILITY_METADATA_KEY.to_string(),
            serde_json::json!(capability_key),
        );
    }
    crate::stream_metadata::encode_envelope(PEER_STREAM_CHANNEL, Some(&metadata))
}

#[cfg(not(target_arch = "wasm32"))]
fn matches_media_stream_envelope(encoded: &[u8], expected_capability_key: Option<&str>) -> bool {
    let crate::stream_metadata::DecodeDecision::Decoded(prefix) =
        crate::stream_metadata::decode_prefix(encoded)
    else {
        return false;
    };
    if prefix.channel.channel_id != PEER_STREAM_CHANNEL {
        return false;
    }
    let Some(metadata) = prefix.channel.metadata.as_ref() else {
        return false;
    };
    if metadata.get("protocol").and_then(serde_json::Value::as_str) != Some(MEDIA_PROTOCOL) {
        return false;
    }
    expected_capability_key.is_none_or(|expected| {
        metadata
            .get(CAPABILITY_METADATA_KEY)
            .and_then(serde_json::Value::as_str)
            == Some(expected)
    })
}

#[cfg(not(target_arch = "wasm32"))]
async fn read_exact_peer(
    recv: &mut crate::application_crypto_streams::PeerRecvStream,
    output: &mut [u8],
) -> anyhow::Result<()> {
    let mut offset = 0;
    while offset < output.len() {
        let read = recv.read(&mut output[offset..]).await?;
        if read == 0 {
            anyhow::bail!("incoming media stream ended with a truncated envelope");
        }
        offset += read;
    }
    Ok(())
}

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

    fn chunk(kind: MediaKind, sequence: u64, generation: u32) -> EncodedMediaChunk {
        EncodedMediaChunk {
            publication_id: PublicationId([7; 16]),
            media_generation: generation,
            sequence,
            timestamp_us: sequence * 20_000,
            duration_us: 20_000,
            kind,
            codec: if kind == MediaKind::Audio {
                MediaCodec::Opus
            } else {
                MediaCodec::H264
            },
            keyframe: sequence % 100 == 0,
            discardable: kind != MediaKind::Audio && sequence % 100 != 0,
            payload: sequence.to_be_bytes().repeat(8),
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn native_media_write_deadline_rejects_a_stalled_protected_write() {
        let error = within_media_write_deadline(std::future::pending::<anyhow::Result<()>>())
            .await
            .unwrap_err();
        assert!(error.to_string().contains("queue-age bound"));
    }

    #[test]
    fn media_chunk_round_trips_and_detects_corruption() {
        let original = chunk(MediaKind::Video, 42, 3);
        let encoded = original.encode().unwrap();
        assert_eq!(EncodedMediaChunk::decode(&encoded).unwrap(), original);
        let mut corrupted = encoded;
        *corrupted.last_mut().unwrap() ^= 1;
        assert_eq!(
            EncodedMediaChunk::decode(&corrupted),
            Err(MediaProtocolError::IntegrityFailure)
        );
    }

    #[test]
    fn media_stats_distinguish_integrity_failures_from_malformed_frames() {
        let mut stats = MediaStats::default();
        stats.record_decode_failure(&MediaProtocolError::IntegrityFailure);
        stats.record_decode_failure(&MediaProtocolError::Truncated);
        stats.record_decode_failure(&MediaProtocolError::LengthMismatch {
            declared: 10,
            actual: 2,
        });
        assert_eq!(stats.integrity_failures, 1);
        assert_eq!(stats.malformed_frames, 2);
    }

    #[test]
    fn video_dependency_gate_recovers_on_keyframes_without_blocking_audio() {
        let mut video = MediaDependencyGate::new(MediaKind::Video);
        let mut first_delta = chunk(MediaKind::Video, 0, 1);
        first_delta.keyframe = false;
        assert_eq!(
            video.observe_chunk(&first_delta),
            MediaDependencyDecision::DropUntilKeyframe
        );

        let mut first_keyframe = chunk(MediaKind::Video, 1, 1);
        first_keyframe.keyframe = true;
        assert_eq!(
            video.observe_chunk(&first_keyframe),
            MediaDependencyDecision::Accept
        );

        let mut after_gap = chunk(MediaKind::Video, 3, 1);
        after_gap.keyframe = false;
        assert_eq!(
            video.observe_chunk(&after_gap),
            MediaDependencyDecision::DropUntilKeyframe
        );
        let mut recovery = chunk(MediaKind::Video, 4, 1);
        recovery.keyframe = true;
        assert_eq!(
            video.observe_chunk(&recovery),
            MediaDependencyDecision::Recovered
        );

        video.observe_discontinuity();
        let mut corrupt_recovery = chunk(MediaKind::Video, 5, 1);
        corrupt_recovery.keyframe = true;
        assert_eq!(
            video.observe_chunk(&corrupt_recovery),
            MediaDependencyDecision::Recovered
        );

        let mut audio = MediaDependencyGate::new(MediaKind::Audio);
        let audio_start = chunk(MediaKind::Audio, 0, 1);
        assert_eq!(
            audio.observe_chunk(&audio_start),
            MediaDependencyDecision::Accept
        );
        audio.observe_discontinuity();
        let audio_after_gap = chunk(MediaKind::Audio, 2, 1);
        assert_eq!(
            audio.observe_chunk(&audio_after_gap),
            MediaDependencyDecision::Accept
        );
    }

    #[test]
    fn media_uses_the_cross_runtime_logical_peer_stream_envelope() {
        let capability = "devices:user-1";
        let encoded =
            encode_media_stream_envelope(Some(capability)).expect("media stream envelope");
        assert!(matches_media_stream_envelope(&encoded, Some(capability)));
        assert!(!matches_media_stream_envelope(
            &encoded,
            Some("devices:user-2")
        ));

        let direct_legacy = crate::stream_metadata::encode_envelope(MEDIA_PROTOCOL, None)
            .expect("legacy media envelope");
        assert!(!matches_media_stream_envelope(
            &direct_legacy,
            Some(capability)
        ));
    }

    #[test]
    fn replay_guard_is_generation_fenced_and_allows_bounded_reordering() {
        let mut guard = MediaReplayGuard::default();
        guard.accept(&chunk(MediaKind::Video, 10, 1)).unwrap();
        guard.accept(&chunk(MediaKind::Video, 12, 1)).unwrap();
        guard.accept(&chunk(MediaKind::Video, 11, 1)).unwrap();
        assert_eq!(
            guard.accept(&chunk(MediaKind::Video, 11, 1)),
            Err(MediaProtocolError::Replay)
        );
        guard.accept(&chunk(MediaKind::Video, 0, 2)).unwrap();
        assert_eq!(
            guard.accept(&chunk(MediaKind::Video, 13, 1)),
            Err(MediaProtocolError::RetiredGeneration)
        );
    }

    #[test]
    fn portable_session_owns_publication_generation_and_receiver_replay() {
        let publication_id = PublicationId([11; 16]);
        let publication = MediaPublicationConfig {
            publication_id,
            media_generation: 0,
            kind: MediaKind::Video,
            codec: MediaCodec::Vp8,
            clock_rate: 1_000_000,
            coded_width: Some(640),
            coded_height: Some(480),
            channels: None,
        };
        let mut sender = PortableMediaSession::default();
        let mut receiver = PortableMediaSession::default();
        let (first, first_control) = sender.begin_publication(publication.clone()).unwrap();
        assert_eq!(first.media_generation, 1);
        receiver.decode_control(&first_control).unwrap();

        let encoded = sender
            .encode_sample(
                publication_id,
                EncodedMediaSample {
                    timestamp_us: 1_000,
                    duration_us: 33_333,
                    keyframe: true,
                    discardable: false,
                    payload: vec![1, 2, 3],
                },
            )
            .unwrap();
        assert_eq!(receiver.decode_chunk(&encoded).unwrap().sequence, 0);
        assert_eq!(
            receiver.decode_chunk(&encoded),
            Err(MediaProtocolError::Replay)
        );

        let (replacement, replacement_control) = sender.begin_publication(publication).unwrap();
        assert_eq!(replacement.media_generation, 2);
        receiver.decode_control(&replacement_control).unwrap();
        assert_eq!(
            receiver.decode_control(&first_control),
            Err(MediaProtocolError::RetiredGeneration)
        );
    }

    #[test]
    fn receiver_admission_retains_tombstones_and_immutable_metadata() {
        let publication_id = PublicationId([13; 16]);
        let publication = MediaPublicationConfig {
            publication_id,
            media_generation: 1,
            kind: MediaKind::Video,
            codec: MediaCodec::Vp8,
            clock_rate: 1_000_000,
            coded_width: Some(640),
            coded_height: Some(480),
            channels: None,
        };
        let mut admission = MediaReceiverAdmission::default();
        let first_control = publication.control().encode().unwrap();
        admission.decode_control(&first_control).unwrap();

        let valid = EncodedMediaChunk {
            publication_id,
            media_generation: 1,
            sequence: 0,
            timestamp_us: 0,
            duration_us: 33_333,
            kind: MediaKind::Video,
            codec: MediaCodec::Vp8,
            keyframe: true,
            discardable: false,
            payload: vec![1, 2, 3],
        };
        let mut changed_codec = valid.clone();
        changed_codec.codec = MediaCodec::H264;
        assert!(matches!(
            admission.decode_chunk(&changed_codec.encode().unwrap()),
            Err(MediaProtocolError::Control(message))
                if message.contains("immutable publication metadata")
        ));
        admission.decode_chunk(&valid.encode().unwrap()).unwrap();

        admission.retire(publication_id, 1);
        let mut delayed = valid.clone();
        delayed.sequence = 1;
        assert!(admission.decode_chunk(&delayed.encode().unwrap()).is_err());
        assert!(matches!(
            admission.decode_control(&first_control),
            Err(MediaProtocolError::Control(message))
                if message.contains("duplicate media publication generation")
        ));

        let replacement = MediaPublicationConfig {
            media_generation: 2,
            ..publication
        };
        admission
            .decode_control(&replacement.control().encode().unwrap())
            .unwrap();
        assert_eq!(
            admission.decode_chunk(&delayed.encode().unwrap()),
            Err(MediaProtocolError::RetiredGeneration)
        );
        let next = EncodedMediaChunk {
            media_generation: 2,
            sequence: 0,
            ..valid
        };
        admission.decode_chunk(&next.encode().unwrap()).unwrap();
    }

    #[test]
    fn portable_sender_retains_generation_and_fails_closed_on_exhaustion() {
        let publication_id = PublicationId([17; 16]);
        let publication = MediaPublicationConfig {
            publication_id,
            media_generation: 0,
            kind: MediaKind::Audio,
            codec: MediaCodec::Opus,
            clock_rate: 48_000,
            coded_width: None,
            coded_height: None,
            channels: Some(1),
        };
        let mut session = PortableMediaSession::default();
        assert_eq!(
            session
                .begin_publication(publication.clone())
                .unwrap()
                .0
                .media_generation,
            1
        );
        session.retire_publication(publication_id);
        assert_eq!(
            session
                .begin_publication(publication.clone())
                .unwrap()
                .0
                .media_generation,
            2
        );

        session
            .senders
            .get_mut(&publication_id)
            .unwrap()
            .next_sequence = u64::MAX;
        assert!(matches!(
            session.encode_sample(
                publication_id,
                EncodedMediaSample {
                    timestamp_us: 0,
                    duration_us: 20_000,
                    keyframe: true,
                    discardable: false,
                    payload: vec![1],
                },
            ),
            Err(MediaProtocolError::Control(message))
                if message.contains("sequence exhausted")
        ));

        session
            .senders
            .get_mut(&publication_id)
            .unwrap()
            .publication
            .media_generation = u32::MAX;
        assert!(matches!(
            session.begin_publication(publication),
            Err(MediaProtocolError::Control(message))
                if message.contains("generation exhausted")
        ));
    }

    #[test]
    fn javascript_media_edges_reject_lossy_u64_values() {
        assert_eq!(
            ensure_javascript_safe_integer(MAX_JAVASCRIPT_SAFE_INTEGER, "timestamp").unwrap(),
            MAX_JAVASCRIPT_SAFE_INTEGER
        );
        assert!(matches!(
            ensure_javascript_safe_integer(MAX_JAVASCRIPT_SAFE_INTEGER + 1, "timestamp"),
            Err(MediaProtocolError::Control(message))
                if message.contains("JavaScript safe integer boundary")
        ));
    }

    #[test]
    fn bounded_queue_drops_discardable_video_before_audio() {
        let mut queue = MediaQueue::new(2, 10_000, 1_000);
        queue.push(chunk(MediaKind::Video, 1, 1), 0);
        queue.push(chunk(MediaKind::Audio, 2, 1), 0);
        queue.push(chunk(MediaKind::Audio, 3, 1), 0);
        assert_eq!(queue.pop(1).unwrap().sequence, 2);
        assert_eq!(queue.pop(1).unwrap().sequence, 3);
        assert_eq!(queue.stats().dropped_overflow, 1);
    }

    #[test]
    fn bounded_queue_evicts_stale_entries_even_after_a_fresh_item() {
        let mut queue = MediaQueue::new(4, 10_000, 50);
        queue.push(chunk(MediaKind::Audio, 1, 1), 100);
        queue.push(chunk(MediaKind::Video, 2, 1), 0);

        assert_eq!(queue.pop(101).unwrap().sequence, 1);
        assert!(queue.pop(101).is_none());
        assert_eq!(queue.stats().dropped_stale, 1);
        assert_eq!(queue.stats().queued_bytes, 0);
    }

    #[test]
    fn bounded_queue_enforces_byte_age_and_plateau_gauges() {
        let item_bytes = CHUNK_HEADER_BYTES + chunk(MediaKind::Audio, 1, 1).payload.len();
        let mut queue = MediaQueue::new(4, item_bytes * 2, 50);

        queue.push(chunk(MediaKind::Audio, 1, 1), 0);
        queue.push(chunk(MediaKind::Video, 2, 1), 0);
        queue.push(chunk(MediaKind::Audio, 3, 1), 0);
        assert_eq!(queue.stats().queued, 2);
        assert_eq!(queue.stats().queued_bytes, item_bytes * 2);
        assert_eq!(queue.stats().max_queued, 2);
        assert_eq!(queue.stats().max_queued_bytes, item_bytes * 2);
        assert_eq!(queue.pop(50).unwrap().sequence, 1);
        assert!(queue.pop(51).is_none());
        assert_eq!(queue.stats().queued, 0);
        assert_eq!(queue.stats().queued_bytes, 0);
        assert_eq!(queue.stats().max_queue_age_ms, 50);
        assert_eq!(queue.stats().dropped_stale, 1);

        for cycle in 0..100 {
            let now = 100 + cycle * 10;
            queue.push(chunk(MediaKind::Audio, cycle + 10, 1), now);
            assert!(queue.stats().queued_bytes <= item_bytes * 2);
            assert!(queue.pop(now + 1).is_some());
        }
        assert_eq!(queue.stats().queued, 0);
        assert_eq!(queue.stats().queued_bytes, 0);

        let mut oversized = chunk(MediaKind::Video, 1, 1);
        oversized.payload.resize(item_bytes * 2, 9);
        queue.push(oversized, 2_000);
        assert_eq!(queue.stats().queued, 0);
        assert_eq!(queue.stats().queued_bytes, 0);
        assert_eq!(queue.stats().dropped_overflow, 2);
    }

    #[derive(Debug, Default)]
    struct OraclePlaybackState {
        decoded: u64,
        rendered: u64,
        last_rendered_sequence: Option<u64>,
        awaiting_keyframe_since_us: Option<u64>,
        max_recovery_us: u64,
    }

    fn oracle_payload(chunk: &EncodedMediaChunk) -> Vec<u8> {
        let mut payload = Vec::with_capacity(38);
        payload.extend_from_slice(&chunk.publication_id.0);
        payload.extend_from_slice(&chunk.media_generation.to_be_bytes());
        payload.extend_from_slice(&chunk.sequence.to_be_bytes());
        payload.extend_from_slice(&chunk.timestamp_us.to_be_bytes());
        payload.push(chunk.kind.wire());
        payload.push(chunk.codec.wire());
        payload
    }

    fn mark_oracle_video_discontinuity(state: &mut OraclePlaybackState, timestamp_us: u64) {
        state.awaiting_keyframe_since_us.get_or_insert(timestamp_us);
    }

    fn render_oracle_chunk(chunk: EncodedMediaChunk, now_us: u64, state: &mut OraclePlaybackState) {
        assert_eq!(chunk.payload, oracle_payload(&chunk));
        if chunk.kind == MediaKind::Video && state.awaiting_keyframe_since_us.is_some() {
            if !chunk.keyframe {
                return;
            }
            let started = state
                .awaiting_keyframe_since_us
                .take()
                .expect("video recovery start");
            state.max_recovery_us = state.max_recovery_us.max(now_us.saturating_sub(started));
        }
        state.decoded = state.decoded.saturating_add(1);
        state.rendered = state.rendered.saturating_add(1);
        state.last_rendered_sequence = Some(chunk.sequence);
    }

    #[test]
    fn deterministic_two_way_oracle_advances_past_ten_minutes_without_corruption() {
        let mut left = MediaReplayGuard::default();
        let mut right = MediaReplayGuard::default();
        let frames = 10 * 60 * 50;
        for sequence in 0..frames {
            let outgoing_left = chunk(MediaKind::Audio, sequence, 1);
            let outgoing_right = EncodedMediaChunk {
                publication_id: PublicationId([9; 16]),
                ..chunk(MediaKind::Video, sequence, 1)
            };
            let received_left =
                EncodedMediaChunk::decode(&outgoing_right.encode().unwrap()).unwrap();
            let received_right =
                EncodedMediaChunk::decode(&outgoing_left.encode().unwrap()).unwrap();
            left.accept(&received_left).unwrap();
            right.accept(&received_right).unwrap();
        }
    }

    #[test]
    fn two_way_audio_video_stays_within_latency_bounds_through_churn_and_switches() {
        const TEN_MINUTES_US: u64 = 10 * 60 * 1_000_000;
        const VIDEO_INTERVAL_US: u64 = 33_333;
        const MAX_EXPECTED_LATENCY_US: u64 = 180_000;
        const MAX_ESTIMATED_LATENCY_US: u64 = 300_000;
        const MAX_VIDEO_RECOVERY_US: u64 = 1_000_000;
        const MAX_QUEUE_ITEMS: usize = 8;
        const MAX_QUEUE_BYTES: usize = 8 * 1024;
        const MAX_QUEUE_AGE_MS: u64 = 1_000;

        let publications = [
            (PublicationId([1; 16]), MediaKind::Audio, MediaCodec::Opus),
            (PublicationId([2; 16]), MediaKind::Video, MediaCodec::Vp8),
            (PublicationId([3; 16]), MediaKind::Audio, MediaCodec::Opus),
            (PublicationId([4; 16]), MediaKind::Video, MediaCodec::Vp8),
        ];
        let mut guards: HashMap<PublicationId, MediaReplayGuard> = publications
            .iter()
            .map(|(publication, _, _)| (*publication, MediaReplayGuard::default()))
            .collect();
        let mut generations: HashMap<PublicationId, u32> = publications
            .iter()
            .map(|(publication, _, _)| (*publication, 1))
            .collect();
        let mut queues: HashMap<PublicationId, MediaQueue> = publications
            .iter()
            .map(|(publication, _, _)| {
                (
                    *publication,
                    MediaQueue::new(MAX_QUEUE_ITEMS, MAX_QUEUE_BYTES, MAX_QUEUE_AGE_MS),
                )
            })
            .collect();
        let mut playback: HashMap<PublicationId, OraclePlaybackState> = publications
            .iter()
            .map(|(publication, _, _)| (*publication, OraclePlaybackState::default()))
            .collect();
        let mut latencies = Vec::new();
        let mut corruption_failures = 0_u64;
        let mut injected_loss = 0_u64;
        let mut replay_drops = 0_u64;
        let mut route_switches = 0_u64;
        let mut reconnects = 0_u64;

        for sequence in 0..(TEN_MINUTES_US / VIDEO_INTERVAL_US) {
            let timestamp_us = sequence * VIDEO_INTERVAL_US;
            if sequence > 0 && sequence % 900 == 0 {
                // A physical carrier switch keeps publication/media generation
                // stable because the logical Iroh connection is unchanged.
                route_switches += 1;
            }
            if sequence > 0 && sequence % 3_600 == 0 {
                // A genuine peer transport replacement resumes each stable
                // publication intent on one new media generation.
                reconnects += 1;
                for generation in generations.values_mut() {
                    *generation += 1;
                }
            }

            for (index, (publication_id, kind, codec)) in publications.iter().enumerate() {
                let generation = generations[publication_id];
                let mut chunk = EncodedMediaChunk {
                    publication_id: *publication_id,
                    media_generation: generation,
                    sequence,
                    timestamp_us,
                    duration_us: VIDEO_INTERVAL_US as u32,
                    kind: *kind,
                    codec: *codec,
                    keyframe: *kind == MediaKind::Video && sequence % 30 == 0,
                    discardable: *kind == MediaKind::Video && sequence % 30 != 0,
                    payload: Vec::new(),
                };
                chunk.payload = oracle_payload(&chunk);
                if sequence > 0 && sequence % 997 == 0 && *kind == MediaKind::Video {
                    mark_oracle_video_discontinuity(
                        playback.get_mut(publication_id).unwrap(),
                        timestamp_us,
                    );
                    injected_loss += 1;
                    continue;
                }
                let mut encoded = chunk.encode().unwrap();
                if sequence > 0
                    && (sequence + index as u64) % 4_999 == 0
                    && *kind == MediaKind::Video
                {
                    *encoded.last_mut().unwrap() ^= 1;
                    assert_eq!(
                        EncodedMediaChunk::decode(&encoded),
                        Err(MediaProtocolError::IntegrityFailure)
                    );
                    mark_oracle_video_discontinuity(
                        playback.get_mut(publication_id).unwrap(),
                        timestamp_us,
                    );
                    corruption_failures += 1;
                    continue;
                }
                let received = EncodedMediaChunk::decode(&encoded).unwrap();
                guards
                    .get_mut(publication_id)
                    .unwrap()
                    .accept(&received)
                    .unwrap();
                if sequence > 0 && sequence % 887 == 0 && index == 0 {
                    assert_eq!(
                        guards.get_mut(publication_id).unwrap().accept(&received),
                        Err(MediaProtocolError::Replay)
                    );
                    replay_drops += 1;
                }
                queues
                    .get_mut(publication_id)
                    .unwrap()
                    .push(received, timestamp_us / 1_000);
            }

            // Hold a short burst at every logical carrier switch, then drain
            // it. This exercises real queue count/byte/age gauges without
            // inventing a second carrier lifecycle in the oracle.
            if sequence % 900 >= 3 {
                for (index, (publication_id, _, _)) in publications.iter().enumerate() {
                    while let Some(received) = queues
                        .get_mut(publication_id)
                        .unwrap()
                        .pop(timestamp_us / 1_000)
                    {
                        let queue_delay_us = timestamp_us.saturating_sub(received.timestamp_us);
                        render_oracle_chunk(
                            received,
                            timestamp_us,
                            playback.get_mut(publication_id).unwrap(),
                        );

                        // Deterministic carrier-independent estimate. Exact
                        // carrier and glass-to-glass lanes measure real time.
                        let latency_us = 45_000
                            + ((sequence * 17 + index as u64 * 13) % 100_000)
                            + queue_delay_us;
                        assert!(latency_us <= MAX_ESTIMATED_LATENCY_US);
                        latencies.push(latency_us);
                    }
                }
            }
        }

        let final_timestamp_us = TEN_MINUTES_US + MAX_QUEUE_AGE_MS * 1_000;
        for (publication_id, _, _) in &publications {
            while let Some(received) = queues
                .get_mut(publication_id)
                .unwrap()
                .pop(final_timestamp_us / 1_000)
            {
                render_oracle_chunk(
                    received,
                    final_timestamp_us,
                    playback.get_mut(publication_id).unwrap(),
                );
            }
        }

        latencies.sort_unstable();
        let p95 = latencies[latencies.len() * 95 / 100];
        assert!(p95 <= MAX_EXPECTED_LATENCY_US, "p95 latency was {p95}us");
        assert!(corruption_failures > 0);
        assert!(injected_loss > 0);
        assert!(replay_drops > 0);
        assert!(route_switches >= 10);
        assert!(reconnects >= 4);
        assert_eq!(guards.len(), 4, "both directions retained audio and video");
        for (publication_id, kind, _) in publications {
            let state = &playback[&publication_id];
            assert!(state.decoded > 0 && state.rendered > 0);
            assert!(state.awaiting_keyframe_since_us.is_none());
            assert!(
                state.max_recovery_us <= MAX_VIDEO_RECOVERY_US,
                "publication {publication_id} recovery was {}us",
                state.max_recovery_us
            );
            if kind == MediaKind::Audio {
                assert_eq!(state.decoded, TEN_MINUTES_US / VIDEO_INTERVAL_US);
            }
            assert!(
                state.last_rendered_sequence.unwrap() + 2 >= TEN_MINUTES_US / VIDEO_INTERVAL_US,
                "publication {publication_id} stopped advancing"
            );

            let stats = queues[&publication_id].stats();
            assert_eq!(stats.queued, 0);
            assert_eq!(stats.queued_bytes, 0);
            assert!(stats.max_queued <= MAX_QUEUE_ITEMS as u64);
            assert!(stats.max_queued_bytes <= MAX_QUEUE_BYTES);
            assert!(stats.max_queue_age_ms <= MAX_QUEUE_AGE_MS);
        }
        assert!(generations
            .values()
            .all(|generation| *generation == 1 + reconnects as u32));
    }
}