librtmp2 0.1.1

use std::collections::HashMap;
use std::sync::Mutex;
use std::time::{Duration, Instant};

use crate::buffer::Buffer;
use crate::chunk::reader::{chunk_read_owned, ChunkMessage};
use crate::chunk::state::{ChunkRegistry, DEFAULT_CHUNK_SIZE};
use crate::chunk::writer::chunk_write;
use crate::handshake::{self, Handshake, HandshakeState};
use crate::message::command;
use crate::message::control::{self, UCTRL_PING_REQUEST, UCTRL_PING_RESPONSE};
use crate::message::message as msg_dispatch;
use crate::session::state_machine;
use crate::session::stream::Stream;
use crate::transport::Transport;
use crate::types::*;

pub const MAX_STREAMS_PER_CONN: u32 = 16;
pub const MAX_PENDING_RELAY_FRAMES: usize = 1024;
pub const MAX_PENDING_RELAY_BYTES: usize = 8 * 1024 * 1024;
/// Cap complete messages handled per `read_messages` call so one TCP recv batch
/// cannot drain thousands of tiny control messages in a single state-machine step.
const MAX_MESSAGES_PER_READ: usize = 256;

const SERVER_WINDOW_ACK_SIZE: u32 = 2_500_000;
const SERVER_PEER_BANDWIDTH: u32 = 2_500_000;
const PEER_BANDWIDTH_DYNAMIC: u8 = 2;
const PING_INTERVAL: Duration = Duration::from_secs(5);
const PING_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_PENDING_PINGS: usize = 4;
/// Cap inbound Ping-Request reflections per connection to prevent trivial
/// outbound bandwidth/CPU amplification from unauthenticated peers.
const MAX_INBOUND_PING_RESPONSES: usize = 8;
const INBOUND_PING_WINDOW: Duration = Duration::from_secs(1);

pub struct RelayFrame {
    pub frame_type: FrameType,
    pub timestamp: u32,
    pub payload: Vec<u8>,
    pub app: String,
    pub stream_name: String,
    pub publisher_conn_id: u64,
}

pub struct Conn {
    pub state: ConnState,
    pub handshake: Handshake,
    pub recv_buffer: Buffer,
    pub send_buffer: Buffer,
    pub chunk_reg: ChunkRegistry,
    /// Target chunk size announced to the peer (from server config).
    pub chunk_size: u32,
    /// Active outbound chunk size; stays at the RTMP default until SetChunkSize
    /// is negotiated on the wire.
    active_chunk_size: u32,
    pub window_ack_size: u32,
    pub bytes_received: u32,
    pub bytes_at_last_ack: u32,
    /// Audio/video payload bytes received (excludes handshake/control overhead).
    pub media_bytes_received: u64,
    /// Audio/video payload bytes sent to this peer.
    pub media_bytes_sent: u64,
    pub client_fd: i32,
    /// Stable per-connection id (monotonic, never reused while the server runs).
    pub conn_id: u64,
    /// Peer socket address for logging (not persisted).
    pub remote_addr: String,
    pub transport: Option<Transport>,
    pub app: String,
    /// Canonical relay route key. When set, publisher/player media is matched
    /// on this value instead of the RTMP stream name (e.g. separate publish/play keys).
    pub relay_key: String,
    pub next_stream_id: u32,
    pub current_stream: Option<Box<Stream>>,
    pub connect_cb_fired: bool,
    pub send_mutex: Mutex<()>,
    pub pending_relay: Vec<RelayFrame>,
    pub needs_init_frames: bool,
    pub detected_video_codec: Option<String>,
    pub detected_audio_codec: Option<String>,
    pub relay_enabled: bool,
    /// When true, media relay stays off until the integrator sets `relay_enabled`
    /// after its own post-auth bookkeeping (used by librtmp2-server).
    pub defer_media_relay: bool,
    /// Cap on queued relay payload bytes for this connection.
    pub max_pending_relay_bytes: usize,
    pub on_frame_cb: Option<fn(&Frame)>,
    /// When set, must return true before audio/video is queued for relay.
    pub on_media_cb: Option<fn(u64, FrameType, Option<&str>) -> bool>,
    pub on_connect_cb: Option<fn()>,
    pub on_publish_cb: Option<fn(conn_id: u64, app: &str, stream_name: &str) -> bool>,
    pub on_play_cb: Option<fn(conn_id: u64, app: &str, stream_name: &str) -> bool>,
    /// Cache keys to evict after the publisher renames its stream.
    pub pending_cache_evictions: Vec<(String, String)>,
    /// Last measured client↔server RTT in milliseconds (RTMP UserControl ping).
    pub rtt_ms: f64,
    pending_pings: HashMap<u32, Instant>,
    last_ping_sent: Option<Instant>,
    next_ping_token: u32,
    inbound_ping_responses: usize,
    inbound_ping_window_start: Option<Instant>,
}

impl Conn {
    pub fn new() -> Self {
        let mut chunk_reg = ChunkRegistry::new();
        chunk_reg.init();
        Self {
            state: ConnState::TcpAccepted,
            handshake: Handshake::default(),
            recv_buffer: Buffer::new(),
            send_buffer: Buffer::new(),
            chunk_reg,
            chunk_size: DEFAULT_CHUNK_SIZE,
            active_chunk_size: DEFAULT_CHUNK_SIZE,
            window_ack_size: 0,
            bytes_received: 0,
            bytes_at_last_ack: 0,
            media_bytes_received: 0,
            media_bytes_sent: 0,
            client_fd: -1,
            conn_id: 0,
            remote_addr: String::new(),
            transport: None,
            app: String::new(),
            relay_key: String::new(),
            next_stream_id: 0,
            current_stream: None,
            connect_cb_fired: false,
            send_mutex: Mutex::new(()),
            pending_relay: Vec::new(),
            needs_init_frames: false,
            detected_video_codec: None,
            detected_audio_codec: None,
            relay_enabled: false,
            defer_media_relay: false,
            max_pending_relay_bytes: MAX_PENDING_RELAY_BYTES,
            on_frame_cb: None,
            on_media_cb: None,
            on_connect_cb: None,
            on_publish_cb: None,
            on_play_cb: None,
            pending_cache_evictions: Vec::new(),
            rtt_ms: 0.0,
            pending_pings: HashMap::new(),
            last_ping_sent: None,
            next_ping_token: 1,
            inbound_ping_responses: 0,
            inbound_ping_window_start: None,
        }
    }

    fn pending_relay_bytes(&self) -> usize {
        self.pending_relay.iter().map(|f| f.payload.len()).sum()
    }

    /// Key used to route relayed media between publishers and players.
    pub fn relay_route_key(&self) -> String {
        if !self.relay_key.is_empty() {
            return self.relay_key.clone();
        }
        self.current_stream
            .as_ref()
            .map(|s| s.name.clone())
            .unwrap_or_default()
    }

    /// Queue eviction of the current publish route's cache key when
    /// `current_stream` is about to be repurposed or replaced while still
    /// marked as actively publishing (a fresh `createStream`, or switching
    /// to `play`, both abandon whatever was being published). Unlike a
    /// `publish` rename there is no new route to keep serving, so this
    /// always evicts when the connection was publishing. No-op otherwise.
    fn evict_active_publish_route(&mut self) {
        let was_publishing = self
            .current_stream
            .as_ref()
            .map(|s| s.is_publishing)
            .unwrap_or(false);
        if !was_publishing {
            return;
        }
        let route_key = self.relay_route_key();
        if !route_key.is_empty() {
            self.pending_cache_evictions
                .push((self.app.clone(), route_key));
        }
    }

    fn queue_relay_frame(&mut self, frame_type: FrameType, timestamp: u32, payload: &[u8]) -> Result<()> {
        if self.pending_relay.len() >= MAX_PENDING_RELAY_FRAMES
            || self.pending_relay_bytes() + payload.len() > self.max_pending_relay_bytes
        {
            return Err(ErrorCode::Internal);
        }
        self.pending_relay.push(RelayFrame {
            frame_type,
            timestamp,
            payload: payload.to_vec(),
            app: self.app.clone(),
            stream_name: self.relay_route_key(),
            publisher_conn_id: self.conn_id,
        });
        Ok(())
    }

    fn media_allowed(&self, frame_type: FrameType) -> bool {
        let Some(cb) = self.on_media_cb else {
            return true;
        };
        let codec = match frame_type {
            FrameType::Video => self.detected_video_codec.as_deref(),
            FrameType::Audio => self.detected_audio_codec.as_deref(),
            _ => None,
        };
        cb(self.conn_id, frame_type, codec)
    }

    fn handle_media_frame(
        &mut self,
        frame_type: FrameType,
        timestamp: u32,
        payload: &[u8],
    ) -> Result<()> {
        if !self.relay_enabled
            || !self
                .current_stream
                .as_ref()
                .map(|s| s.is_publishing)
                .unwrap_or(false)
        {
            return Ok(());
        }

        match frame_type {
            FrameType::Video if self.detected_video_codec.is_none() => {
                self.detected_video_codec = detect_video_codec(payload);
            }
            FrameType::Audio if self.detected_audio_codec.is_none() => {
                self.detected_audio_codec = detect_audio_codec(payload);
            }
            _ => {}
        }

        if !self.media_allowed(frame_type) {
            return Err(ErrorCode::Auth);
        }

        self.media_bytes_received = self
            .media_bytes_received
            .saturating_add(payload.len() as u64);

        if let Some(cb) = self.on_frame_cb {
            let owned = payload.to_vec();
            let frame = Frame {
                frame_type,
                timestamp,
                size: owned.len() as u32,
                data: owned.as_ptr(),
                ..Default::default()
            };
            cb(&frame);
        }

        if self.queue_relay_frame(frame_type, timestamp, payload).is_err() {
            return Err(ErrorCode::Internal);
        }
        Ok(())
    }

    pub fn get_fd(&self) -> i32 { self.client_fd }

    pub fn recv(&mut self, data: &[u8]) -> Result<()> {
        self.recv_buffer.write(data).map_err(|_| ErrorCode::Internal)?;
        self.bytes_received = self.bytes_received.wrapping_add(data.len() as u32);
        let mut max_iter = 256;
        let mut no_progress = 0;
        while max_iter > 0 {
            max_iter -= 1;
            let avail = self.recv_buffer.available();
            if avail == 0 && self.state != ConnState::Handshake { break; }
            let before = avail;
            let rc = self.process();
            if rc < 0 {
                return Err(match rc {
                    -1 => ErrorCode::Io,
                    -2 => ErrorCode::Timeout,
                    -3 => ErrorCode::Protocol,
                    -4 => ErrorCode::Handshake,
                    -5 => ErrorCode::Chunk,
                    -6 => ErrorCode::Amf,
                    -7 => ErrorCode::Unsupported,
                    -8 => ErrorCode::Auth,
                    -9 => ErrorCode::Internal,
                    _ => ErrorCode::Internal,
                });
            }
            if rc == 0 {
                let after = self.recv_buffer.available();
                if after == before {
                    no_progress += 1;
                    if no_progress > 3 { break; }
                } else {
                    no_progress = 0;
                }
                if after == 0 && self.state < ConnState::Closing { break; }
            } else {
                no_progress = 0;
            }
        }
        if self.window_ack_size > 0
            && self.bytes_received.wrapping_sub(self.bytes_at_last_ack) >= self.window_ack_size
        {
            self.send_acknowledgement(self.bytes_received)?;
            self.bytes_at_last_ack = self.bytes_received;
        }
        Ok(())
    }

    pub fn process(&mut self) -> i32 {
        match self.state {
            ConnState::TcpAccepted | ConnState::Handshake => self.do_handshake(),
            ConnState::Connected
            | ConnState::AppConnected
            | ConnState::StreamCreated
            | ConnState::Publishing
            | ConnState::Playing
            | ConnState::CapsNegotiated => self.read_messages(),
            ConnState::Closing | ConnState::Closed => 0,
        }
    }

    pub fn do_handshake(&mut self) -> i32 {
        match self.handshake.state {
            HandshakeState::ServerWaitC0 => {
                handshake::server_init(&mut self.handshake);
                match handshake::server_read_c0(&mut self.handshake, &mut self.recv_buffer) {
                    Ok(()) => { self.state = ConnState::Handshake; self.do_handshake_recurse() }
                    Err(ErrorCode::Io) => 0,
                    Err(e) => e as i32,
                }
            }
            HandshakeState::ServerWaitC1 => self.do_handshake_recurse(),
            HandshakeState::ServerWaitC2 => match handshake::server_read_c2(&mut self.handshake, &mut self.recv_buffer) {
                Ok(()) => { self.state = ConnState::Connected; 1 }
                Err(ErrorCode::Io) => 0,
                Err(e) => e as i32,
            },
            HandshakeState::Done => { self.state = ConnState::Connected; 1 }
            _ => -1,
        }
    }

    fn do_handshake_recurse(&mut self) -> i32 {
        match handshake::server_read_c1(&mut self.handshake, &mut self.recv_buffer) {
            Ok(()) => {
                if self.client_fd >= 0 {
                    let s0 = [0x03u8];
                    if self.send_buffer.write(&s0).is_err() { return ErrorCode::Internal as i32; }
                    let out_data = self.handshake.out.peek();
                    if self.send_buffer.write(out_data).is_err() { return ErrorCode::Internal as i32; }
                }
                self.handshake.out.reset();
                1
            }
            Err(ErrorCode::Io) => 0,
            Err(e) => e as i32,
        }
    }

    pub fn read_messages(&mut self) -> i32 {
        let mut processed = 0usize;
        loop {
            if processed >= MAX_MESSAGES_PER_READ {
                break;
            }
            let mut msg = ChunkMessage::default();
            match chunk_read_owned(&mut self.recv_buffer, &mut self.chunk_reg, &mut msg) {
                Ok((0, _)) => break,
                Ok((1, payload_owned)) => {
                    if msg.is_complete {
                        processed += 1;
                        if let Err(e) = self.handle_message(&msg, &payload_owned) {
                            return match e {
                                ErrorCode::Auth => -8,
                                _ => -3,
                            };
                        }
                        let _ = self.flush();
                    }
                }
                Ok(_) => break,
                Err(ErrorCode::Chunk) => return -5,
                Err(_) => return -1,
            }
        }
        1
    }

    fn handle_message(&mut self, msg: &ChunkMessage, payload: &[u8]) -> Result<()> {
        match msg.msg_type_id {
            msg_dispatch::RTMP_MSG_SET_CHUNK_SIZE
            | msg_dispatch::RTMP_MSG_ABORT_MESSAGE
            | msg_dispatch::RTMP_MSG_ACKNOWLEDGEMENT
            | msg_dispatch::RTMP_MSG_WINDOW_ACK_SIZE
            | msg_dispatch::RTMP_MSG_SET_PEER_BANDWIDTH => self.handle_control(msg.msg_type_id, payload),
            msg_dispatch::RTMP_MSG_USER_CONTROL => self.handle_user_control(payload),
            msg_dispatch::RTMP_MSG_AMF0_COMMAND => self.handle_command(payload),
            msg_dispatch::RTMP_MSG_AMF3_COMMAND => {
                if !payload.is_empty() && payload[0] == 0x00 {
                    self.handle_command(&payload[1..])
                } else {
                    self.handle_command(payload)
                }
            }
            msg_dispatch::RTMP_MSG_AUDIO => self.handle_media_frame(FrameType::Audio, msg.timestamp, payload),
            msg_dispatch::RTMP_MSG_VIDEO => self.handle_media_frame(FrameType::Video, msg.timestamp, payload),
            _ => Ok(()),
        }
    }

    fn handle_control(&mut self, msg_type_id: u8, payload: &[u8]) -> Result<()> {
        match msg_type_id {
            msg_dispatch::RTMP_MSG_SET_CHUNK_SIZE => {
                if payload.len() >= 4 {
                    if let Ok(cs) = control::read_set_chunk_size(payload) {
                        self.chunk_reg.set_all_chunk_size(cs);
                    }
                }
            }
            msg_dispatch::RTMP_MSG_ABORT_MESSAGE => {
                if payload.len() >= 4 {
                    if let Ok(csid) = control::read_abort_message(payload) {
                        self.chunk_reg.reset_stream(csid);
                    }
                }
            }
            msg_dispatch::RTMP_MSG_WINDOW_ACK_SIZE => {
                if payload.len() >= 4 {
                    if let Ok(win) = control::read_window_ack_size(payload) {
                        self.window_ack_size = win;
                    }
                }
            }
            msg_dispatch::RTMP_MSG_ACKNOWLEDGEMENT => {
                if payload.len() >= 4 {
                    let _ = control::read_acknowledgement_size(payload);
                }
            }
            msg_dispatch::RTMP_MSG_SET_PEER_BANDWIDTH => {
                if payload.len() >= 5 {
                    let _ = control::read_set_peer_bandwidth(payload);
                }
            }
            _ => {}
        }
        Ok(())
    }

    /// Apply a chunk size to outbound writes and inbound reassembly.
    pub fn apply_chunk_size(&mut self, chunk_size: u32) {
        self.chunk_size = chunk_size;
        self.active_chunk_size = chunk_size;
        self.chunk_reg.set_all_chunk_size(chunk_size);
    }

    fn activate_announced_chunk_size(&mut self) {
        self.active_chunk_size = self.chunk_size;
        self.chunk_reg.set_all_chunk_size(self.chunk_size);
    }

    fn handle_user_control(&mut self, payload: &[u8]) -> Result<()> {
        if payload.len() < 6 {
            return Ok(());
        }
        let (event_type, param1, _) = control::read_user_control(payload, false)?;
        match event_type {
            UCTRL_PING_RESPONSE => {
                if let Some(sent_at) = self.pending_pings.remove(&param1) {
                    self.rtt_ms = sent_at.elapsed().as_secs_f64() * 1000.0;
                }
            }
            UCTRL_PING_REQUEST => {
                let now = Instant::now();
                if let Some(start) = self.inbound_ping_window_start {
                    if now.duration_since(start) >= INBOUND_PING_WINDOW {
                        self.inbound_ping_window_start = Some(now);
                        self.inbound_ping_responses = 0;
                    }
                } else {
                    self.inbound_ping_window_start = Some(now);
                }
                if self.inbound_ping_responses >= MAX_INBOUND_PING_RESPONSES {
                    return Err(ErrorCode::Protocol);
                }
                self.inbound_ping_responses += 1;
                self.send_user_control_ping_response(param1)?;
            }
            _ => {}
        }
        Ok(())
    }

    /// Send an RTMP ping when due and measure RTT from the client's response.
    pub fn maybe_send_ping(&mut self) -> Result<()> {
        if self.state < ConnState::AppConnected {
            return Ok(());
        }
        let now = Instant::now();
        if self
            .last_ping_sent
            .is_some_and(|t| now.duration_since(t) < PING_INTERVAL)
        {
            return Ok(());
        }

        self.pending_pings
            .retain(|_, sent| now.duration_since(*sent) < PING_TIMEOUT);
        if self.pending_pings.len() >= MAX_PENDING_PINGS {
            return Ok(());
        }

        let token = self.next_ping_token;
        self.next_ping_token = self.next_ping_token.wrapping_add(1);
        self.send_user_control_ping_request(token)?;
        self.pending_pings.insert(token, now);
        self.last_ping_sent = Some(now);
        Ok(())
    }

    pub fn handle_command(&mut self, payload: &[u8]) -> Result<()> {
        let mut buf = Buffer::from_slice(payload);
        let mut name_buf = [0u8; 64];
        if command::peek_name(&mut buf, &mut name_buf).is_err() {
            return Ok(());
        }
        let name = std::str::from_utf8(&name_buf).unwrap_or("").trim_end_matches('\0');
        match name {
            "connect" => {
                // A connection may only negotiate its app namespace once. Without
                // this guard a peer that's already publishing/playing under an
                // authorized app could send a second `connect` with a different
                // app, silently repointing `self.app` (and therefore relay/cache
                // routing) to a namespace `on_publish_cb`/`on_play_cb` never
                // authorized -- cache poisoning / cross-app playback.
                if self.state >= ConnState::AppConnected {
                    return Ok(());
                }
                let mut info = ConnectInfo::default();
                command::read_connect(&mut buf, &mut info)?;
                let app_len = info.app.iter().position(|&b| b == 0).unwrap_or(0);
                self.app = std::str::from_utf8(&info.app[..app_len]).unwrap_or("").to_string();
                let _ = state_machine::conn_transition(&mut self.state, ConnState::AppConnected);
                self.send_connect_response(info.transaction_id)?;
                if !self.connect_cb_fired {
                    self.connect_cb_fired = true;
                    if let Some(cb) = self.on_connect_cb { cb(); }
                }
            }
            "createStream" => {
                if self.state < ConnState::AppConnected {
                    return self.send_onstatus(0, "error", "NetStream.Failed", "connect required before createStream");
                }
                let txn = command::read_create_stream(&mut buf)?;
                if self.next_stream_id >= MAX_STREAMS_PER_CONN {
                    self.send_onstatus(0, "error", "NetStream.Failed", "Too many streams")?;
                } else {
                    // A fresh createStream replaces current_stream outright;
                    // if the stream it's replacing was actively publishing,
                    // that route is being abandoned and must be evicted now
                    // rather than left as a dangling cache-key tracking
                    // entry until this connection eventually disconnects.
                    self.evict_active_publish_route();
                    self.next_stream_id += 1;
                    let stream_id = self.next_stream_id;
                    self.current_stream = Some(Box::new(Stream::new(stream_id)));
                    let _ = state_machine::conn_transition(&mut self.state, ConnState::StreamCreated);
                    self.send_create_stream_response(txn, stream_id)?;
                }
            }
            "publish" => {
                let mut stream_name = [0u8; 256];
                let mut publish_type = [0u8; 64];
                command::read_publish(&mut buf, &mut stream_name, &mut publish_type)?;
                let name_str = std::str::from_utf8(&stream_name).unwrap_or("").trim_end_matches('\0').to_string();
                if self.current_stream.is_none() {
                    return self.send_onstatus(
                        0,
                        "error",
                        "NetStream.Publish.BadConnection",
                        "No stream created",
                    );
                }
                if let Some(cb) = self.on_publish_cb {
                    if !cb(self.conn_id, &self.app, &name_str) {
                        return self.send_onstatus(0, "error", "NetStream.Publish.BadName", "Publish not authorized");
                    }
                }
                // Use current_stream.is_publishing rather than self.state
                // here: conn_transition() only allows forward moves through
                // ConnState, so a play-then-publish sequence leaves
                // self.state stuck at Playing (Publishing < Playing) even
                // though this stream is genuinely publishing again. `play`
                // explicitly clears is_publishing when it takes over a
                // stream, so the flag accurately reflects whether this
                // specific current_stream is the one actively publishing.
                let was_publishing = self
                    .current_stream
                    .as_ref()
                    .map(|s| s.is_publishing)
                    .unwrap_or(false);
                let prev_route_key = self.relay_route_key();
                let next_route_key = if !self.relay_key.is_empty() {
                    self.relay_key.clone()
                } else {
                    name_str.clone()
                };
                if was_publishing && !prev_route_key.is_empty() && prev_route_key != next_route_key {
                    self.pending_cache_evictions
                        .push((self.app.clone(), prev_route_key));
                }
                if !self.defer_media_relay || self.on_publish_cb.is_none() {
                    self.relay_enabled = true;
                }
                {
                    if let Some(ref mut stream) = self.current_stream {
                        stream.is_publishing = true;
                        stream.name = name_str;
                    }
                    let _ = state_machine::conn_transition(&mut self.state, ConnState::Publishing);
                    let sid = self.current_stream.as_ref().map(|s| s.stream_id).unwrap_or(0);
                    self.send_onstatus(sid, "status", "NetStream.Publish.Start", "Publishing")?;
                }
            }
            "play" => {
                let mut stream_name = [0u8; 256];
                command::read_play(&mut buf, &mut stream_name)?;
                let name_str = std::str::from_utf8(&stream_name).unwrap_or("").trim_end_matches('\0').to_string();
                if self.current_stream.is_none() {
                    return self.send_onstatus(
                        0,
                        "error",
                        "NetStream.Play.BadConnection",
                        "No stream created",
                    );
                }
                if let Some(cb) = self.on_play_cb {
                    if !cb(self.conn_id, &self.app, &name_str) {
                        return self.send_onstatus(0, "error", "NetStream.Play.Failed", "Play not authorized");
                    }
                }
                if !self.defer_media_relay || self.on_play_cb.is_none() {
                    self.relay_enabled = true;
                }
                {
                    // `play` supersedes any publish role this current_stream
                    // held: evict the abandoned publish route's cache key
                    // now (there's no "next" publish route to preserve it
                    // for), and clear is_publishing so it can't be
                    // mistaken for an active publish by a later command.
                    self.evict_active_publish_route();
                    if let Some(ref mut stream) = self.current_stream {
                        stream.is_playing = true;
                        stream.is_publishing = false;
                        stream.name = name_str;
                    }
                    self.needs_init_frames = true;
                    let _ = state_machine::conn_transition(&mut self.state, ConnState::Playing);
                    let sid = self.current_stream.as_ref().map(|s| s.stream_id).unwrap_or(0);
                    self.send_onstatus(sid, "status", "NetStream.Play.Start", "Playing")?;
                }
            }
            "FCPublish" | "FCUnpublish" | "releaseStream" | "deleteStream" => {}
            _ => {}
        }
        Ok(())
    }

    pub fn send_connect_response(&mut self, transaction_id: f64) -> Result<()> {
        let win = SERVER_WINDOW_ACK_SIZE.to_be_bytes();
        self.send_control(0x05, &win)?;
        let mut bw = [0u8; 5];
        let bw_val = SERVER_PEER_BANDWIDTH.to_be_bytes();
        bw[..4].copy_from_slice(&bw_val);
        bw[4] = PEER_BANDWIDTH_DYNAMIC;
        self.send_control(0x06, &bw)?;
        let cs = self.chunk_size.to_be_bytes();
        self.send_control(0x01, &cs)?;
        // Negotiation complete: subsequent server chunks and client sends use
        // the announced size (connect AMF above was still at 128).
        self.activate_announced_chunk_size();
        let mut amf_buf = Buffer::with_capacity(512);
        crate::amf::amf0::write_string(&mut amf_buf, "_result")?;
        crate::amf::amf0::write_number(&mut amf_buf, transaction_id)?;
        crate::amf::amf0::write_null(&mut amf_buf)?;
        crate::amf::amf0::write_object_begin(&mut amf_buf)?;
        crate::amf::amf0::write_object_key(&mut amf_buf, "level")?;
        crate::amf::amf0::write_string(&mut amf_buf, "status")?;
        crate::amf::amf0::write_object_key(&mut amf_buf, "code")?;
        crate::amf::amf0::write_string(&mut amf_buf, "NetConnection.Connect.Success")?;
        crate::amf::amf0::write_object_key(&mut amf_buf, "description")?;
        crate::amf::amf0::write_string(&mut amf_buf, "Connection succeeded.")?;
        crate::amf::amf0::write_object_end(&mut amf_buf)?;
        self.send_command(0, amf_buf.as_slice())
    }

    pub fn send_create_stream_response(&mut self, transaction_id: f64, stream_id: u32) -> Result<()> {
        let mut amf_buf = Buffer::with_capacity(256);
        command::build_create_stream_result(&mut amf_buf, transaction_id, stream_id as f64)?;
        self.send_command(0, amf_buf.as_slice())
    }

    pub fn send_onstatus(&mut self, stream_id: u32, level: &str, code: &str, description: &str) -> Result<()> {
        let mut amf_buf = Buffer::with_capacity(512);
        command::build_onstatus(&mut amf_buf, level, code, description)?;
        self.send_command(stream_id, amf_buf.as_slice())
    }

    pub fn flush(&mut self) -> Result<()> {
        if self.client_fd < 0 || self.send_buffer.available() == 0 { return Ok(()); }
        let Some(ref mut transport) = self.transport else { return Ok(()); };
        while self.send_buffer.available() > 0 {
            let pending = self.send_buffer.peek();
            let n = transport.try_send(pending, &mut 0i32)?;
            if n == 0 { break; }
            self.send_buffer.drain(n);
        }
        Ok(())
    }

    pub fn send_frame(&mut self, frame_type: FrameType, timestamp: u32, payload: &[u8]) -> Result<()> {
        let stream_id = self.current_stream.as_ref().map(|s| s.stream_id).unwrap_or(1);
        let mut cmsg = ChunkMessage::default();
        cmsg.timestamp = timestamp;
        cmsg.msg_length = payload.len() as u32;
        cmsg.msg_stream_id = stream_id;
        cmsg.fmt = 0;
        if frame_type == FrameType::Audio {
            cmsg.csid = 4;
            cmsg.msg_type_id = 0x08;
        } else {
            cmsg.csid = 6;
            cmsg.msg_type_id = 0x09;
        }
        chunk_write(
            &mut self.send_buffer,
            &cmsg,
            payload,
            payload.len(),
            self.active_chunk_size as usize,
        )?;
        self.media_bytes_sent = self
            .media_bytes_sent
            .saturating_add(payload.len() as u64);
        Ok(())
    }

    fn send_control(&mut self, ty: u8, data: &[u8]) -> Result<()> {
        let mut msg = ChunkMessage::default();
        msg.csid = 2;
        msg.fmt = 0;
        msg.msg_length = data.len() as u32;
        msg.msg_type_id = ty;
        msg.msg_stream_id = 0;
        chunk_write(&mut self.send_buffer, &msg, data, data.len(), self.active_chunk_size as usize)
    }

    fn send_command(&mut self, msg_stream_id: u32, amf_data: &[u8]) -> Result<()> {
        let mut cmd_msg = ChunkMessage::default();
        cmd_msg.csid = 3;
        cmd_msg.fmt = 0;
        cmd_msg.timestamp = 0;
        cmd_msg.msg_length = amf_data.len() as u32;
        cmd_msg.msg_type_id = 0x14;
        cmd_msg.msg_stream_id = msg_stream_id;
        chunk_write(
            &mut self.send_buffer,
            &cmd_msg,
            amf_data,
            amf_data.len(),
            self.active_chunk_size as usize,
        )
    }

    fn send_acknowledgement(&mut self, seq: u32) -> Result<()> {
        self.send_control(0x03, &seq.to_be_bytes())
    }

    fn send_user_control_ping_request(&mut self, timestamp: u32) -> Result<()> {
        let mut buf = Buffer::with_capacity(6);
        control::write_user_control_ping_request(&mut buf, timestamp)?;
        self.send_control(msg_dispatch::RTMP_MSG_USER_CONTROL, buf.as_slice())
    }

    fn send_user_control_ping_response(&mut self, timestamp: u32) -> Result<()> {
        let mut buf = Buffer::with_capacity(6);
        control::write_user_control_ping_response(&mut buf, timestamp)?;
        self.send_control(msg_dispatch::RTMP_MSG_USER_CONTROL, buf.as_slice())
    }
}

impl Default for Conn {
    fn default() -> Self { Self::new() }
}

fn detect_video_codec(payload: &[u8]) -> Option<String> {
    if payload.is_empty() { return None; }
    if payload[0] & 0x80 != 0 {
        if payload.len() >= 5 {
            if let Ok(s) = std::str::from_utf8(&payload[1..5]) { return Some(s.to_string()); }
        }
        return None;
    }
    Some(match payload[0] & 0x0F {
        7 => "avc1".to_string(),
        12 => "hvc1".to_string(),
        13 => "av01".to_string(),
        _ => return None,
    })
}

fn detect_audio_codec(payload: &[u8]) -> Option<String> {
    if payload.is_empty() { return None; }
    if (payload[0] & 0xF0) == 0x90 && payload.len() >= 5 {
        if let Ok(s) = std::str::from_utf8(&payload[1..5]) { return Some(s.to_string()); }
    }
    Some(match (payload[0] >> 4) & 0x0F {
        10 => "mp4a".to_string(),
        2 => "mp3".to_string(),
        14 => "Opus".to_string(),
        _ => return None,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::session::stream::Stream;

    #[test]
    fn relay_route_key_prefers_relay_key_over_rtmp_name() {
        let mut conn = Conn::new();
        conn.relay_key = "stream-db-id".to_string();
        conn.current_stream = Some(Box::new(Stream::new(1)));
        if let Some(ref mut stream) = conn.current_stream {
            stream.name = "pub_or_play_key".to_string();
        }
        assert_eq!(conn.relay_route_key(), "stream-db-id");
    }

    #[test]
    fn relay_route_key_falls_back_to_rtmp_stream_name() {
        let mut conn = Conn::new();
        conn.current_stream = Some(Box::new(Stream::new(1)));
        if let Some(ref mut stream) = conn.current_stream {
            stream.name = "legacy_name".to_string();
        }
        assert_eq!(conn.relay_route_key(), "legacy_name");
    }

    #[test]
    fn connect_rejects_app_names_longer_than_routing_buffer() {
        let mut conn = Conn::new();
        let mut buf = Buffer::with_capacity(512);
        let long_app = "a".repeat(256);
        command::build_connect(
            &mut buf,
            &long_app,
            "rtmp://host/app",
            "",
            "",
            "FMLE/3.0",
            0,
            0,
        )
        .unwrap();
        assert_eq!(conn.handle_command(buf.as_slice()), Err(ErrorCode::Amf));
        assert!(conn.app.is_empty());
    }

    #[test]
    fn connect_after_app_connected_does_not_repoint_app_namespace() {
        let mut conn = Conn::new();

        let mut buf = Buffer::with_capacity(256);
        command::build_connect(&mut buf, "public", "rtmp://host/public", "", "", "FMLE/3.0", 0, 0)
            .unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert_eq!(conn.app, "public");
        assert_eq!(conn.state, ConnState::AppConnected);

        // A second `connect` after the app namespace is already authorized
        // must be ignored -- it must not silently repoint `self.app` (and
        // therefore relay/cache routing) to a namespace that was never
        // passed through on_publish_cb/on_play_cb.
        let mut buf2 = Buffer::with_capacity(256);
        command::build_connect(
            &mut buf2,
            "private",
            "rtmp://host/private",
            "",
            "",
            "FMLE/3.0",
            0,
            0,
        )
        .unwrap();
        conn.handle_command(buf2.as_slice()).unwrap();
        assert_eq!(conn.app, "public");
        assert_eq!(conn.state, ConnState::AppConnected);
    }

    #[test]
    fn apply_chunk_size_updates_outbound_and_inbound() {
        let mut conn = Conn::new();
        conn.apply_chunk_size(4096);
        assert_eq!(conn.chunk_size, 4096);
        assert_eq!(conn.active_chunk_size, 4096);
        assert_eq!(conn.chunk_reg.default_chunk_size, 4096);
    }

    #[test]
    fn new_connection_starts_at_rtmp_default_chunk_size() {
        let conn = Conn::new();
        assert_eq!(conn.chunk_size, DEFAULT_CHUNK_SIZE);
        assert_eq!(conn.active_chunk_size, DEFAULT_CHUNK_SIZE);
        assert_eq!(conn.chunk_reg.default_chunk_size, DEFAULT_CHUNK_SIZE);
    }

    #[test]
    fn enhanced_av1_media_frames_accepted_while_publishing() {
        let mut conn = Conn::new();
        conn.relay_enabled = true;
        conn.current_stream = Some(Box::new(Stream::new(1)));
        if let Some(s) = conn.current_stream.as_mut() {
            s.is_publishing = true;
        }

        let av1_seq = vec![0x90, b'a', b'v', b'0', b'1', 0x01, 0x02, 0x03];
        assert!(conn
            .handle_media_frame(FrameType::Video, 0, &av1_seq)
            .is_ok());

        let aac_seq = vec![0xAF, 0x00, 0x12, 0x10];
        assert!(conn
            .handle_media_frame(FrameType::Audio, 0, &aac_seq)
            .is_ok());

        let av1_frame = vec![0x91, b'a', b'v', b'0', b'1', 0xDE, 0xAD, 0xBE, 0xEF];
        assert!(conn
            .handle_media_frame(FrameType::Video, 40, &av1_frame)
            .is_ok());
    }

    #[test]
    fn publish_rename_with_relay_key_does_not_evict_stale_rtmp_name() {
        let mut conn = Conn::new();
        conn.app = "live".to_string();
        conn.relay_key = "route-1".to_string();
        conn.current_stream = Some(Box::new(Stream::new(1)));

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "A", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert!(conn.pending_cache_evictions.is_empty());
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "A");

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "B", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();

        // relay_route_key() is pinned to relay_key, so republishing under a new
        // RTMP stream name must NOT queue an eviction for the stale RTMP name
        // ("A") -- the real cache key ("route-1") never changed.
        assert!(conn.pending_cache_evictions.is_empty());
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "B");
    }

    #[test]
    fn publish_rename_without_relay_key_evicts_old_route_key() {
        let mut conn = Conn::new();
        conn.app = "live".to_string();
        conn.current_stream = Some(Box::new(Stream::new(1)));

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "A", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert!(conn.pending_cache_evictions.is_empty());

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "B", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();

        assert_eq!(
            conn.pending_cache_evictions,
            vec![("live".to_string(), "A".to_string())]
        );
    }

    #[test]
    fn play_then_publish_does_not_evict_foreign_cache_key() {
        let mut conn = Conn::new();
        conn.app = "live".to_string();
        conn.current_stream = Some(Box::new(Stream::new(1)));

        let mut buf = Buffer::with_capacity(128);
        command::build_play(&mut buf, "victim").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "victim");
        assert!(!conn.current_stream.as_ref().unwrap().is_publishing);

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "other", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();

        // This connection only ever played "victim" -- it never published it
        // -- so publishing "other" afterwards must not queue an eviction for
        // a cache key this connection never created.
        assert!(conn.pending_cache_evictions.is_empty());
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "other");
    }

    #[test]
    fn publish_then_play_then_publish_does_not_evict_played_stream_key() {
        let mut conn = Conn::new();
        conn.app = "live".to_string();
        conn.current_stream = Some(Box::new(Stream::new(1)));

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "A", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert!(conn.pending_cache_evictions.is_empty());

        // `play` supersedes the active publish of "A" on this stream, so it
        // must evict "A" itself right away (there's no future publish
        // rename to hang that eviction off of, and is_publishing is cleared
        // so this stream can no longer be mistaken for an active publisher
        // of "A").
        let mut buf = Buffer::with_capacity(128);
        command::build_play(&mut buf, "victim").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "victim");
        assert!(!conn.current_stream.as_ref().unwrap().is_publishing);
        assert_eq!(
            conn.pending_cache_evictions,
            vec![("live".to_string(), "A".to_string())]
        );
        conn.pending_cache_evictions.clear();

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "other", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();

        // This connection was never actively publishing "victim" -- it was
        // only playing it -- so publishing "other" must not queue a second
        // eviction for a cache key it never owned.
        assert!(conn.pending_cache_evictions.is_empty());
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "other");
    }

    #[test]
    fn create_stream_evicts_active_publish_route() {
        let mut conn = Conn::new();
        conn.app = "live".to_string();
        conn.state = ConnState::AppConnected;
        conn.current_stream = Some(Box::new(Stream::new(1)));

        let mut buf = Buffer::with_capacity(128);
        command::build_publish(&mut buf, "A", "live").unwrap();
        conn.handle_command(buf.as_slice()).unwrap();
        assert!(conn.pending_cache_evictions.is_empty());
        assert!(conn.current_stream.as_ref().unwrap().is_publishing);

        // A fresh createStream replaces current_stream outright, abandoning
        // the active publish of "A" with no future rename to hang an
        // eviction off of -- it must be evicted immediately here, not left
        // as a dangling publisher_cache_keys tracking entry until this
        // connection eventually disconnects.
        let mut buf = Buffer::with_capacity(128);
        command::build_create_stream(&mut buf, 4.0).unwrap();
        conn.handle_command(buf.as_slice()).unwrap();

        assert_eq!(
            conn.pending_cache_evictions,
            vec![("live".to_string(), "A".to_string())]
        );
        assert!(!conn.current_stream.as_ref().unwrap().is_publishing);
        assert_eq!(conn.current_stream.as_ref().unwrap().name, "");
    }

    #[test]
    fn handle_control_peer_set_chunk_size_updates_inbound_only() {
        let mut conn = Conn::new();
        conn.chunk_size = 4096;
        conn.handle_control(
            msg_dispatch::RTMP_MSG_SET_CHUNK_SIZE,
            &8192u32.to_be_bytes(),
        )
        .unwrap();
        assert_eq!(conn.chunk_size, 4096);
        assert_eq!(conn.active_chunk_size, DEFAULT_CHUNK_SIZE);
        assert_eq!(conn.chunk_reg.default_chunk_size, 8192);
    }

    #[test]
    fn inbound_ping_requests_are_rate_limited() {
        let mut conn = Conn::new();
        conn.client_fd = 0;
        conn.transport = None;
        let mut ping = |token: u32| {
            let mut buf = Buffer::with_capacity(6);
            control::write_user_control_ping_request(&mut buf, token).unwrap();
            conn.handle_user_control(buf.as_slice())
        };
        for token in 0..8 {
            assert!(ping(token).is_ok(), "token {token} should be accepted");
        }
        assert!(
            matches!(ping(99), Err(ErrorCode::Protocol)),
            "9th ping in one second must be rejected"
        );
    }
}