use anyhow::{Context, anyhow};
use bytes::BytesMut;
use pulse_core::{
    codec::{Encoding, SessionCodec},
    control::{AcceptPayload, ConnectPayload, StreamStatus},
    plp::{Header, PLP_VERSION, Packet, PacketFlags, PacketType},
    plp_conn::{ConnError, ConnEvent, PlpConn},
};
use quinn::{Connection, RecvStream, SendStream};
use std::collections::HashMap;
use std::sync::Arc;
use tracing::{debug, error, info, warn};

pub async fn handle_connection(
    connection: Connection,
    _state: Arc<crate::AppState>,
) -> Result<(), anyhow::Error> {
    let client_addr = connection.remote_address();
    info!("Starting session for client: {}", client_addr);

    let conn_sm = Arc::new(tokio::sync::Mutex::new(PlpConn::new_server()));
    let session_id = Arc::new(tokio::sync::Mutex::new(String::new()));
    let datagram_conn = pulse_core::plp_datagram::PlpDatagramConn::new(connection.clone());

    loop {
        let dgram_fut = async {
            if datagram_conn.max_datagram_size().is_some() {
                datagram_conn.recv_packet().await
            } else {
                std::future::pending().await
            }
        };

        tokio::select! {
            bi = connection.accept_bi() => {
                match bi {
                    Ok((send, recv)) => {
                        let state_clone = Arc::clone(&_state);
                        let client_addr_str = client_addr.to_string();
                        let conn_sm_clone = Arc::clone(&conn_sm);
                        let session_id_clone = Arc::clone(&session_id);
                        tokio::spawn(async move {
                            if let Err(e) = handle_stream(send, recv, state_clone, client_addr_str, conn_sm_clone, session_id_clone).await {
                                warn!("Stream error: {}", e);
                            }
                        });
                    }
                    Err(quinn::ConnectionError::ApplicationClosed { .. }) => {
                        info!("Connection closed gracefully by client: {}", client_addr);
                        break;
                    }
                    Err(e) => {
                        error!("Connection lost with client {}: {}", client_addr, e);
                        break;
                    }
                }
            }
            dgram = dgram_fut => {
                match dgram {
                    Ok(pkt) => {
                        let state_clone = Arc::clone(&_state);
                        let client_addr_str = client_addr.to_string();
                        let conn_sm_clone = Arc::clone(&conn_sm);
                        let session_id_clone = Arc::clone(&session_id);
                        let dconn_clone = datagram_conn.clone();
                        tokio::spawn(async move {
                            if let Err(e) = handle_plp_packet(
                                &mut ReplyWriter::Datagram(&dconn_clone),
                                &state_clone,
                                &client_addr_str,
                                None,
                                pkt,
                                &conn_sm_clone,
                                &session_id_clone,
                            ).await {
                                warn!("Datagram error: {}", e);
                            }
                        });
                    }
                    Err(e) => {
                        debug!("Datagram receive error: {}", e);
                    }
                }
            }
        }
    }

    Ok(())
}

async fn handle_stream(
    mut send: SendStream,
    mut recv: RecvStream,
    state: Arc<crate::AppState>,
    client_addr: String,
    conn_sm: Arc<tokio::sync::Mutex<PlpConn>>,
    session_id: Arc<tokio::sync::Mutex<String>>,
) -> Result<(), anyhow::Error> {
    let stream_id = send.id();
    info!("New bidirectional stream opened: {:?}", stream_id);

    let mut plp_buf = BytesMut::with_capacity(8 * 1024);

    // First, determine whether this stream is speaking PLP framing.
    // If not, fall back to legacy raw-buffer echo for backwards compatibility.
    let mode = match read_some(&mut recv, &mut plp_buf).await? {
        Some(()) => match try_decode_one_plp_packet(&mut plp_buf) {
            Ok(Some(pkt)) => {
                // We already consumed the first packet from plp_buf.
                StreamMode::Plp { first_packet: pkt }
            }
            Ok(None) => {
                // Not enough bytes to decide; keep reading in PLP mode and decide later.
                StreamMode::PlpUndecided
            }
            Err(pulse_core::plp::Error::InvalidVersion(_))
            | Err(pulse_core::plp::Error::InvalidPacketType(_)) => StreamMode::Legacy,
            Err(e) => return Err(anyhow!(e)).context("failed to decode initial PLP packet"),
        },
        None => {
            info!("Stream {:?} closed before any bytes", stream_id);
            return Ok(());
        }
    };

    match mode {
        StreamMode::Legacy => {
            legacy_raw_echo_loop(
                &mut send,
                &mut recv,
                &state,
                &client_addr,
                stream_id,
                plp_buf,
            )
            .await?;
        }
        StreamMode::Plp { first_packet } => {
            plp_loop(
                &mut send,
                &mut recv,
                &state,
                &client_addr,
                stream_id,
                plp_buf,
                Some(first_packet),
                conn_sm,
                session_id,
            )
            .await?;
        }
        StreamMode::PlpUndecided => {
            plp_loop(
                &mut send,
                &mut recv,
                &state,
                &client_addr,
                stream_id,
                plp_buf,
                None,
                conn_sm,
                session_id,
            )
            .await?;
        }
    }

    info!("Stream {:?} finished", stream_id);

    Ok(())
}

enum StreamMode {
    Legacy,
    Plp { first_packet: Packet },
    PlpUndecided,
}

async fn legacy_raw_echo_loop(
    send: &mut SendStream,
    recv: &mut RecvStream,
    state: &crate::AppState,
    client_addr: &str,
    quic_stream_id: quinn::StreamId,
    mut already_read: BytesMut,
) -> Result<(), anyhow::Error> {
    let mut buf = vec![0u8; 4096];

    if !already_read.is_empty() {
        debug!(
            "Legacy mode: processing {} buffered bytes on stream {:?}",
            already_read.len(),
            quic_stream_id
        );
        let packet = crate::sfu::ForwardPacket {
            source: client_addr.to_string(),
            stream_id: format!("{:?}", quic_stream_id),
            sequence: 0,
            payload: already_read.to_vec(),
            lossless: false,
        };
        let processed = state.media_pipeline.process_frame(packet).await?;
        send.write_all(&processed.payload).await?;
        already_read.clear();
    }

    while let Some(bytes_read) = recv.read(&mut buf).await? {
        debug!(
            "Received {} bytes on stream {:?}",
            bytes_read, quic_stream_id
        );

        let packet = crate::sfu::ForwardPacket {
            source: client_addr.to_string(),
            stream_id: format!("{:?}", quic_stream_id),
            sequence: 0,
            payload: buf[..bytes_read].to_vec(),
            lossless: false,
        };

        let processed = state.media_pipeline.process_frame(packet).await?;
        send.write_all(&processed.payload).await?;
    }

    Ok(())
}

async fn plp_loop(
    send: &mut SendStream,
    recv: &mut RecvStream,
    state: &crate::AppState,
    client_addr: &str,
    quic_stream_id: quinn::StreamId,
    mut plp_buf: BytesMut,
    first_packet: Option<Packet>,
    conn_sm: Arc<tokio::sync::Mutex<PlpConn>>,
    session_id: Arc<tokio::sync::Mutex<String>>,
) -> Result<(), anyhow::Error> {
    // If we already have a first packet decoded, handle it immediately.
    if let Some(pkt) = first_packet {
        handle_plp_packet(
            &mut ReplyWriter::Stream(send),
            state,
            client_addr,
            Some(quic_stream_id),
            pkt,
            &conn_sm,
            &session_id,
        )
        .await?;
    }

    loop {
        // Try decode from buffered bytes first.
        if let Some(pkt) = match try_decode_one_plp_packet(&mut plp_buf) {
            Ok(v) => v,
            Err(pulse_core::plp::Error::BufferTooShort) => None,
            Err(pulse_core::plp::Error::InvalidVersion(_))
            | Err(pulse_core::plp::Error::InvalidPacketType(_)) => {
                warn!(
                    "Invalid PLP framing on stream {:?}; switching to legacy mode",
                    quic_stream_id
                );
                return legacy_raw_echo_loop(
                    send,
                    recv,
                    state,
                    client_addr,
                    quic_stream_id,
                    plp_buf,
                )
                .await;
            }
            Err(e) => return Err(anyhow!(e)).context("failed to decode PLP packet"),
        } {
            handle_plp_packet(
                &mut ReplyWriter::Stream(send),
                state,
                client_addr,
                Some(quic_stream_id),
                pkt,
                &conn_sm,
                &session_id,
            )
            .await?;
            continue;
        }

        // Need more bytes from the network.
        if read_some(recv, &mut plp_buf).await?.is_none() {
            return Ok(());
        }
    }
}

pub enum ReplyWriter<'a> {
    Stream(&'a mut SendStream),
    Datagram(&'a pulse_core::plp_datagram::PlpDatagramConn),
}

impl<'a> ReplyWriter<'a> {
    pub async fn write_packet(&mut self, pkt: &Packet) -> Result<(), anyhow::Error> {
        match self {
            Self::Stream(send) => write_plp_packet(send, pkt).await,
            Self::Datagram(conn) => conn
                .send_packet(pkt)
                .map_err(|e| anyhow::anyhow!("Datagram send error: {}", e)),
        }
    }
}

async fn handle_plp_packet(
    sender: &mut ReplyWriter<'_>,
    state: &crate::AppState,
    client_addr: &str,
    quic_stream_id: Option<quinn::StreamId>,
    pkt: Packet,
    conn_sm_mtx: &tokio::sync::Mutex<PlpConn>,
    session_id_mtx: &tokio::sync::Mutex<String>,
) -> Result<(), anyhow::Error> {
    debug!(
        "PLP packet on {:?}: type={:?} stream_id={} seq={} len={}",
        quic_stream_id,
        pkt.header.packet_type,
        pkt.header.stream_id,
        pkt.header.sequence_number,
        pkt.header.payload_length
    );

    match pkt.header.packet_type {
        PacketType::Connect => {
            // Enforce ordering via core connection state machine.
            let mut conn_sm = conn_sm_mtx.lock().await;
            match conn_sm.on_inbound(&pkt) {
                Ok(Some(ConnEvent::Handshake(
                    pulse_core::plp_handshake::HandshakeEvent::ConnectReceived,
                ))) => {}
                Ok(_) => return Ok(()),
                Err(_) => return Ok(()),
            }

            // CONNECT payload is always JSON (handshake phase — negotiation not complete yet).
            let connect: ConnectPayload = serde_json::from_slice(&pkt.payload)
                .context("failed to parse CONNECT payload JSON")?;

            // Determine the encoding the client wants for subsequent packets.
            let negotiated = connect
                .encoding
                .as_deref()
                .map(Encoding::from_wire_name)
                .unwrap_or(Encoding::Json); // old client = JSON

            let mut session_id = session_id_mtx.lock().await;
            *session_id = format!("{}-{:?}", client_addr, quic_stream_id);

            info!(
                "CONNECT from {} — encoding={}, session={}",
                client_addr, negotiated, *session_id
            );

            // ── Auth verification ───────────────────────────────────────
            let auth_result = state.auth_verifier.verify(connect.auth.as_ref()).await;
            let auth_user = match auth_result {
                crate::auth::AuthResult::Authenticated(user) => {
                    info!("CONNECT auth OK: user={}", user.id);
                    Some(user)
                }
                crate::auth::AuthResult::Skipped => None,
                crate::auth::AuthResult::Rejected(reason) => {
                    warn!("CONNECT auth REJECTED from {}: {}", client_addr, reason);
                    let reject = pulse_core::control::RejectPayload {
                        reason: format!("authentication failed: {reason}"),
                    };
                    let payload = serde_json::to_vec(&reject)
                        .context("failed to encode REJECT payload JSON")?;
                    let resp = Packet {
                        header: Header {
                            version: PLP_VERSION,
                            packet_type: PacketType::Reject,
                            flags: PacketFlags::empty(),
                            stream_id: pkt.header.stream_id,
                            sequence_number: pkt.header.sequence_number,
                            payload_length: payload.len() as u64,
                        },
                        payload: bytes::Bytes::from(payload),
                    };
                    sender.write_packet(&resp).await?;
                    return Ok(());
                }
            };

            let streams: Vec<StreamStatus> = connect
                .streams
                .iter()
                .enumerate()
                .map(|(i, s)| StreamStatus {
                    id: (i + 1) as u16,
                    stream_type: s.stream_type.clone(),
                    accepted: true,
                })
                .collect();

            // ACCEPT is always JSON too (handshake phase).
            let accept = AcceptPayload {
                session_id: session_id.clone(),
                streams,
                relay_path: vec!["pulse-relay".to_string()],
                metrics_endpoint: "/api/v1/metrics".to_string(),
                encoding: connect.encoding.clone(),
                user: auth_user,
            };

            let payload =
                serde_json::to_vec(&accept).context("failed to encode ACCEPT payload JSON")?;
            let resp = Packet {
                header: Header {
                    version: PLP_VERSION,
                    packet_type: PacketType::Accept,
                    flags: PacketFlags::empty(),
                    stream_id: pkt.header.stream_id,
                    sequence_number: pkt.header.sequence_number,
                    payload_length: payload.len() as u64,
                },
                payload: bytes::Bytes::from(payload),
            };

            sender.write_packet(&resp).await?;

            // Mark server-side ACCEPT sent.
            let _ = conn_sm.on_outbound(PacketType::Accept, 0);
            drop(conn_sm);
        }
        PacketType::Ping => {
            let resp = Packet {
                header: Header {
                    version: PLP_VERSION,
                    packet_type: PacketType::Pong,
                    flags: PacketFlags::empty(),
                    stream_id: pkt.header.stream_id,
                    sequence_number: pkt.header.sequence_number,
                    payload_length: 0,
                },
                payload: bytes::Bytes::new(),
            };
            sender.write_packet(&resp).await?;
        }
        PacketType::StreamOpen => {
            let mut conn_sm = conn_sm_mtx.lock().await;
            match conn_sm.on_inbound(&pkt) {
                Ok(_) => {}
                Err(ConnError::HandshakeNotAccepted) => return Ok(()),
                Err(_) => return Ok(()),
            }
        }
        PacketType::StreamData => {
            let mut conn_sm = conn_sm_mtx.lock().await;
            match conn_sm.on_inbound(&pkt) {
                Ok(_) => {}
                Err(ConnError::HandshakeNotAccepted) => {
                    warn!(
                        "STREAM_DATA before CONNECT on stream {:?}; dropping",
                        quic_stream_id
                    );
                    return Ok(());
                }
                Err(ConnError::Stream(pulse_core::plp_stream::StreamError::NotOpen)) => {
                    warn!(
                        "STREAM_DATA before STREAM_OPEN on stream_id={} (quic {:?}); dropping",
                        pkt.header.stream_id, quic_stream_id
                    );
                    return Ok(());
                }
                Err(_) => return Ok(()),
            }

            // ── Payload size enforcement ──
            if pkt.payload.len() > state.max_payload_bytes {
                warn!(
                    "STREAM_DATA payload too large ({} bytes, max {}) on stream_id={} from {}; resetting stream",
                    pkt.payload.len(),
                    state.max_payload_bytes,
                    pkt.header.stream_id,
                    client_addr
                );
                let reset = Packet {
                    header: Header {
                        version: PLP_VERSION,
                        packet_type: PacketType::StreamReset,
                        flags: PacketFlags::empty(),
                        stream_id: pkt.header.stream_id,
                        sequence_number: pkt.header.sequence_number,
                        payload_length: 0,
                    },
                    payload: bytes::Bytes::new(),
                };
                sender.write_packet(&reset).await?;
                return Ok(());
            }

            let mode = conn_sm
                .stream_mode(pkt.header.stream_id)
                .unwrap_or(pulse_core::plp::StreamMode::Realtime);
            drop(conn_sm);

            let packet = crate::sfu::ForwardPacket {
                source: client_addr.to_string(),
                stream_id: pkt.header.stream_id.to_string(),
                sequence: pkt.header.sequence_number,
                payload: pkt.payload.to_vec(),
                lossless: mode == pulse_core::plp::StreamMode::Broadcast,
            };

            match mode {
                pulse_core::plp::StreamMode::Queue => {
                    debug!("QUEUE mode deferred to Phase 9");
                    return Ok(());
                }
                pulse_core::plp::StreamMode::P2p => {
                    debug!("P2P mode deferred to Phase 8");
                    return Ok(());
                }
                pulse_core::plp::StreamMode::Rpc => {
                    // Direct forwarding (echo for now via media pipeline)
                    match state.media_pipeline.process_frame(packet).await {
                        Ok(processed) => {
                            let resp = Packet {
                                header: Header {
                                    version: PLP_VERSION,
                                    packet_type: PacketType::StreamData,
                                    flags: pkt.header.flags,
                                    stream_id: pkt.header.stream_id,
                                    sequence_number: pkt.header.sequence_number,
                                    payload_length: processed.payload.len() as u64,
                                },
                                payload: bytes::Bytes::from(processed.payload),
                            };
                            sender.write_packet(&resp).await?;
                        }
                        Err(e) => {
                            error!(
                                "WASM Transform failed on stream {:?}: {}",
                                quic_stream_id, e
                            );
                        }
                    }
                }
                pulse_core::plp::StreamMode::Realtime | pulse_core::plp::StreamMode::Broadcast => {
                    // Route via SFU
                    state.sfu.forward(packet).await;
                }
            }
        }
        PacketType::StreamClose | PacketType::StreamReset => {
            let mut conn_sm = conn_sm_mtx.lock().await;
            match conn_sm.on_inbound(&pkt) {
                Ok(_) => {}
                Err(ConnError::HandshakeNotAccepted) => return Ok(()),
                Err(_) => return Ok(()),
            }
            drop(conn_sm);

            // For now, we don't maintain per-stream state in the relay session handler.
            // Echo the close/reset back so clients can observe stream teardown.
            let resp = Packet {
                header: Header {
                    version: PLP_VERSION,
                    packet_type: pkt.header.packet_type,
                    flags: PacketFlags::empty(),
                    stream_id: pkt.header.stream_id,
                    sequence_number: pkt.header.sequence_number,
                    payload_length: pkt.payload.len() as u64,
                },
                payload: pkt.payload.clone(),
            };

            sender.write_packet(&resp).await?;
        }
        PacketType::Metrics => {
            let conn_sm = conn_sm_mtx.lock().await;
            // Client-reported metrics payload (JSON), forwarded to control plane.
            // Expected fields: rtt_ms (number), packet_loss_pct (number), bitrates (object)
            if conn_sm.handshake_state() != pulse_core::plp_handshake::HandshakeState::Accepted {
                return Ok(());
            }
            drop(conn_sm);

            let v: serde_json::Value = match serde_json::from_slice(&pkt.payload) {
                Ok(v) => v,
                Err(_) => return Ok(()),
            };

            let rtt_ms = v.get("rtt_ms").and_then(|x| x.as_u64()).unwrap_or(0) as u32;
            let packet_loss_pct = v
                .get("packet_loss_pct")
                .and_then(|x| x.as_f64())
                .unwrap_or(0.0) as f32;

            let mut bitrates: HashMap<String, u32> = HashMap::new();
            if let Some(obj) = v.get("bitrates").and_then(|x| x.as_object()) {
                for (k, vv) in obj.iter() {
                    if let Some(n) = vv.as_u64() {
                        bitrates.insert(k.clone(), n as u32);
                    }
                }
            }

            let session_id = session_id_mtx.lock().await;
            let control_url = state.control_url.clone();
            let session_id_clone = session_id.clone();
            drop(session_id);

            tokio::spawn(async move {
                let client = reqwest::Client::new();
                let url = format!("{}/api/v1/metrics", control_url);
                let body = serde_json::json!({
                    "session_id": session_id_clone,
                    "rtt_ms": rtt_ms,
                    "packet_loss_pct": packet_loss_pct,
                    "bitrates": bitrates,
                });

                let _ = client.post(url).json(&body).send().await;
            });
        }
        PacketType::ServiceRequest => {
            // ── PLP Service RPC: replaces HTTP control-plane calls ──
            let conn_sm = conn_sm_mtx.lock().await;
            if conn_sm.handshake_state() != pulse_core::plp_handshake::HandshakeState::Accepted {
                return Ok(());
            }
            drop(conn_sm);

            let req: crate::service::ServiceRequest = match serde_json::from_slice(&pkt.payload) {
                Ok(r) => r,
                Err(e) => {
                    warn!("Failed to parse SERVICE_REQUEST: {}", e);
                    return Ok(());
                }
            };

            let service_state = Arc::clone(&state.service_state);
            let seq = pkt.header.sequence_number;

            // Handle asynchronously so we don't block the packet loop.
            let resp = crate::service::handle_service_request(req, &service_state).await;
            let resp_pkt = crate::service::encode_service_response(&resp, seq);
            sender.write_packet(&resp_pkt).await?;
        }
        PacketType::HttpOutbound => {
            // ── HTTP Bridge: client asks relay to make an HTTP call ──
            let conn_sm = conn_sm_mtx.lock().await;
            if conn_sm.handshake_state() != pulse_core::plp_handshake::HandshakeState::Accepted {
                return Ok(());
            }
            drop(conn_sm);

            let req: crate::http_bridge::HttpOutboundRequest =
                match serde_json::from_slice(&pkt.payload) {
                    Ok(r) => r,
                    Err(e) => {
                        warn!("Failed to parse HTTP_OUTBOUND: {}", e);
                        return Ok(());
                    }
                };

            let config = Arc::clone(&state.http_proxy_config);
            let seq = pkt.header.sequence_number;
            let session_id_str = session_id_mtx.lock().await.clone();

            let resp =
                crate::http_bridge::handle_http_outbound(req, &config, &session_id_str).await;
            let resp_pkt = crate::http_bridge::encode_http_response(&resp, seq);
            sender.write_packet(&resp_pkt).await?;
        }
        _ => {
            debug!(
                "Unhandled PLP packet type {:?} on stream {:?}",
                pkt.header.packet_type, quic_stream_id
            );
        }
    }

    Ok(())
}

async fn write_plp_packet(send: &mut SendStream, pkt: &Packet) -> Result<(), anyhow::Error> {
    let mut out = BytesMut::with_capacity((pkt.header.payload_length as usize) + 32);
    pkt.encode(&mut out);
    send.write_all(&out).await?;
    Ok(())
}

fn try_decode_one_plp_packet(buf: &mut BytesMut) -> Result<Option<Packet>, pulse_core::plp::Error> {
    let mut slice: &[u8] = &buf[..];
    match Packet::decode(&mut slice) {
        Ok(pkt) => {
            let consumed = buf.len() - slice.len();
            let _ = buf.split_to(consumed);
            Ok(Some(pkt))
        }
        Err(pulse_core::plp::Error::BufferTooShort) => Ok(None),
        Err(e) => Err(e),
    }
}

async fn read_some(recv: &mut RecvStream, buf: &mut BytesMut) -> Result<Option<()>, anyhow::Error> {
    let mut tmp = [0u8; 4096];
    match recv.read(&mut tmp).await? {
        Some(n) => {
            buf.extend_from_slice(&tmp[..n]);
            Ok(Some(()))
        }
        None => Ok(None),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{ai_processor, edge_wasm, gpu, media};
    use pulse_core::transport::PulseEndpoint;
    use rustls::pki_types::{CertificateDer, PrivateKeyDer};
    use std::net::SocketAddr;
    use std::sync::Arc;
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::sync::{RwLock, oneshot};

    #[test]
    fn plp_decode_handles_partial_then_complete_packet() {
        let payload = b"hello".to_vec();
        let pkt = Packet {
            header: Header {
                version: PLP_VERSION,
                packet_type: PacketType::Ping,
                flags: PacketFlags::empty(),
                stream_id: 7,
                sequence_number: 1,
                payload_length: payload.len() as u64,
            },
            payload: bytes::Bytes::from(payload),
        };

        let mut encoded = BytesMut::new();
        pkt.encode(&mut encoded);

        let split_at = (encoded.len() / 2).max(1);
        let mut buf = BytesMut::new();
        buf.extend_from_slice(&encoded[..split_at]);
        assert!(try_decode_one_plp_packet(&mut buf).unwrap().is_none());

        buf.extend_from_slice(&encoded[split_at..]);
        let decoded = try_decode_one_plp_packet(&mut buf).unwrap().unwrap();
        assert_eq!(decoded.header.packet_type, PacketType::Ping);
        assert_eq!(decoded.header.stream_id, 7);
        assert_eq!(decoded.payload.as_ref(), b"hello");
        assert!(buf.is_empty());
    }

    #[test]
    fn plp_decode_can_extract_two_packets_back_to_back() {
        let pkt1 = Packet {
            header: Header {
                version: PLP_VERSION,
                packet_type: PacketType::Ping,
                flags: PacketFlags::empty(),
                stream_id: 1,
                sequence_number: 1,
                payload_length: 0,
            },
            payload: bytes::Bytes::new(),
        };
        let pkt2 = Packet {
            header: Header {
                version: PLP_VERSION,
                packet_type: PacketType::Pong,
                flags: PacketFlags::empty(),
                stream_id: 1,
                sequence_number: 2,
                payload_length: 0,
            },
            payload: bytes::Bytes::new(),
        };

        let mut buf = BytesMut::new();
        pkt1.encode(&mut buf);
        pkt2.encode(&mut buf);

        let a = try_decode_one_plp_packet(&mut buf).unwrap().unwrap();
        let b = try_decode_one_plp_packet(&mut buf).unwrap().unwrap();
        assert_eq!(a.header.packet_type, PacketType::Ping);
        assert_eq!(b.header.packet_type, PacketType::Pong);
        assert!(buf.is_empty());
    }

    async fn start_mock_control_server() -> (String, oneshot::Receiver<serde_json::Value>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind mock server");
        let addr = listener.local_addr().expect("local addr");
        let (tx, rx) = oneshot::channel::<serde_json::Value>();

        tokio::spawn(async move {
            let (mut sock, _) = listener.accept().await.expect("accept");
            let mut buf = Vec::with_capacity(4096);
            let mut tmp = [0u8; 1024];

            // Read headers.
            loop {
                let n = match sock.read(&mut tmp).await {
                    Ok(0) => return,
                    Ok(n) => n,
                    Err(_) => return,
                };
                buf.extend_from_slice(&tmp[..n]);
                if buf.windows(4).any(|w| w == b"\r\n\r\n") {
                    break;
                }
                if buf.len() > 64 * 1024 {
                    return;
                }
            }

            let header_end = buf
                .windows(4)
                .position(|w| w == b"\r\n\r\n")
                .map(|p| p + 4)
                .unwrap();

            let headers = &buf[..header_end];
            let headers_str = String::from_utf8_lossy(headers);
            let mut content_len: usize = 0;
            for line in headers_str.lines() {
                let lower = line.to_ascii_lowercase();
                if let Some(rest) = lower.strip_prefix("content-length:") {
                    content_len = rest.trim().parse::<usize>().unwrap_or(0);
                }
            }

            // Read body (may already be partially in buf).
            while buf.len() < header_end + content_len {
                let n = match sock.read(&mut tmp).await {
                    Ok(0) => break,
                    Ok(n) => n,
                    Err(_) => break,
                };
                buf.extend_from_slice(&tmp[..n]);
                if buf.len() > 256 * 1024 {
                    break;
                }
            }

            let body = if content_len > 0 && buf.len() >= header_end + content_len {
                &buf[header_end..header_end + content_len]
            } else {
                &[]
            };

            if let Ok(v) = serde_json::from_slice::<serde_json::Value>(body) {
                let _ = tx.send(v);
            }

            let _ = sock
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK")
                .await;
        });

        (format!("http://{}", addr), rx)
    }

    fn generate_self_signed_cert()
    -> Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>), Box<dyn std::error::Error>>
    {
        let subject_alt_names = vec!["localhost".to_string()];
        let cert = rcgen::generate_simple_self_signed(subject_alt_names)?;

        let key = PrivateKeyDer::try_from(cert.signing_key.serialize_der())?;
        let cert_der = CertificateDer::from(cert.cert.der().to_vec());
        Ok((vec![cert_der], key))
    }

    #[tokio::test]
    async fn metrics_frames_are_forwarded_to_control_plane() {
        let _ = rustls::crypto::ring::default_provider().install_default();

        let (control_url, rx) = start_mock_control_server().await;

        // Minimal relay state (avoid expensive hardware detection in tests).
        let state = Arc::new(crate::AppState {
            gpu_manager: Arc::new(RwLock::new(gpu::GpuManager::new_for_tests())),
            ai_processor: Arc::new(ai_processor::AiProcessor::new(ai_processor::AiConfig {
                stt_endpoint: None,
                tts_endpoint: None,
                vision_endpoint: None,
            })),
            media_pipeline: Arc::new(media::MediaPipeline::new(Arc::new(
                edge_wasm::EdgeWasmEngine::new().expect("wasm engine"),
            ))),
            control_url: control_url.clone(),
            sfu: Arc::new(crate::sfu::Sfu::new()),
            max_payload_bytes: 1024 * 1024,
            service_state: Arc::new(crate::service::ServiceState::new(
                Arc::new(pulse_core::queue_storage::InMemoryBackend::new()),
            )),
            http_proxy_config: Arc::new(crate::http_bridge::HttpProxyConfig::default()),
            queue_backend: Arc::new(pulse_core::queue_storage::InMemoryBackend::new()),
            auth_verifier: crate::auth::AuthVerifier::new(crate::auth::AuthConfig::default()),
        });

        // Start relay QUIC endpoint.
        let (certs, key) = generate_self_signed_cert().expect("cert");
        let server = PulseEndpoint::new_server("127.0.0.1:0".parse().unwrap(), certs, key)
            .expect("server endpoint");
        let endpoint = server.inner().clone();
        let server_addr: SocketAddr = endpoint.local_addr().expect("local addr");

        // Accept exactly one connection and run the real session handler.
        let state_clone = Arc::clone(&state);
        let server_task = tokio::spawn(async move {
            let Some(incoming) = endpoint.accept().await else {
                return;
            };
            let Ok(conn) = incoming.await else {
                return;
            };
            let _ = handle_connection(conn, state_clone).await;
        });

        // Client: handshake via Rust SDK.
        let client = pulse_sdk::PulseClient::connect_insecure_dev(server_addr).expect("client");
        let mut session = client
            .connect_and_handshake("dev")
            .await
            .expect("handshake");

        let mut bitrates = std::collections::HashMap::new();
        bitrates.insert("audio".to_string(), 64_000);
        bitrates.insert("video".to_string(), 1_500_000);
        session
            .send_metrics(17, 0.2, bitrates)
            .await
            .expect("send metrics");

        // Assert control plane received a metrics ingest POST.
        let v = tokio::time::timeout(Duration::from_secs(2), rx)
            .await
            .expect("timeout waiting for control POST")
            .expect("receiver dropped");

        assert!(
            v.get("session_id")
                .and_then(|x| x.as_str())
                .unwrap_or("")
                .len()
                > 0
        );
        assert_eq!(v.get("rtt_ms").and_then(|x| x.as_u64()).unwrap_or(0), 17);
        let loss = v
            .get("packet_loss_pct")
            .and_then(|x| x.as_f64())
            .unwrap_or(0.0);
        assert!(
            (loss - 0.2).abs() < 1e-6,
            "expected packet_loss_pct ~= 0.2, got {loss}"
        );

        // Cleanup.
        drop(session);
        let _ = tokio::time::timeout(Duration::from_millis(200), server_task).await;
    }
}
