qmux 0.4.0

use bytes::{Buf, BufMut, Bytes, BytesMut};
use web_transport_proto::VarInt;

use crate::{Error, StreamId, TransportParams, Version};

// QMux frame type IDs (QUIC v1 compatible)
const RESET_STREAM: VarInt = VarInt::from_u32(0x04);
// RESET_STREAM_AT (draft-ietf-quic-reliable-stream-reset), permitted by QMux
// draft-02. Decode-only: we accept and validate it, then treat it as a plain
// reset (see the decode arms for why the reliable-delivery guarantee is
// automatically satisfied on a reliable, ordered transport).
const RESET_STREAM_AT: u64 = 0x24;
const STOP_SENDING: VarInt = VarInt::from_u32(0x05);
const STREAM_BASE: u32 = 0x08;
const MAX_DATA: VarInt = VarInt::from_u32(0x10);
const MAX_STREAM_DATA: VarInt = VarInt::from_u32(0x11);
const MAX_STREAMS_BIDI: VarInt = VarInt::from_u32(0x12);
const MAX_STREAMS_UNI: VarInt = VarInt::from_u32(0x13);
const DATA_BLOCKED: VarInt = VarInt::from_u32(0x14);
const STREAM_DATA_BLOCKED: VarInt = VarInt::from_u32(0x15);
const STREAMS_BLOCKED_BIDI: VarInt = VarInt::from_u32(0x16);
const STREAMS_BLOCKED_UNI: VarInt = VarInt::from_u32(0x17);
const APPLICATION_CLOSE: VarInt = VarInt::from_u32(0x1d);
const CONNECTION_CLOSE: VarInt = VarInt::from_u32(0x1c);
// DATAGRAM frames (RFC 9221). 0x30 has no length (the payload runs to the end of
// the record); 0x31 prefixes a length varint. Datagrams are only ever sent on
// QMux01, where each rides its own record, so the record boundary would already
// delimit a lone 0x30 datagram. We still emit 0x31 so the frame stays
// self-delimiting even if a record ever carries a frame after it — the length
// varint costs 1-2 bytes. Both forms decode.
const DATAGRAM_LEN: VarInt = VarInt::from_u32(0x31);

// QX_TRANSPORT_PARAMETERS magic: "\xffQMX\r\n\r\n"
// This exceeds u32 range, so we use try_from at decode time and a pre-computed const for encode.
const QX_TRANSPORT_PARAMETERS: u64 = 0x3f5153300d0a0d0a;
// SAFETY: 0x3f5153300d0a0d0a < 2^62 (VarInt max), verified by the assertion below.
const QX_TRANSPORT_PARAMETERS_VI: VarInt =
    unsafe { VarInt::from_u64_unchecked(QX_TRANSPORT_PARAMETERS) };
const _: () = assert!(
    QX_TRANSPORT_PARAMETERS < (1 << 62),
    "QX_TRANSPORT_PARAMETERS must fit in VarInt"
);

// QX_PING frame types (draft-01)
const QX_PING_REQUEST: u64 = 0x348c67529ef8c7bd;
const QX_PING_REQUEST_VI: VarInt = unsafe { VarInt::from_u64_unchecked(QX_PING_REQUEST) };
const _: () = assert!(
    QX_PING_REQUEST < (1 << 62),
    "QX_PING_REQUEST must fit in VarInt"
);
const QX_PING_RESPONSE: u64 = 0x348c67529ef8c7be;
const QX_PING_RESPONSE_VI: VarInt = unsafe { VarInt::from_u64_unchecked(QX_PING_RESPONSE) };
const _: () = assert!(
    QX_PING_RESPONSE < (1 << 62),
    "QX_PING_RESPONSE must fit in VarInt"
);

/// Stream data frame carrying payload bytes for a specific stream.
#[derive(Debug, Clone)]
pub struct Stream {
    /// The stream this data belongs to.
    pub id: StreamId,
    /// Byte offset of this payload within the stream.
    ///
    /// Senders populate this field and always emit the OFF bit for QMux. The
    /// receive path preserves it but does not yet enforce continuity.
    pub offset: u64,
    /// The payload bytes.
    pub data: Bytes,
    /// Whether this is the final frame on the stream.
    pub fin: bool,
}

/// Abruptly terminates the sending side of a stream with an error code.
#[derive(Debug, Clone)]
pub struct ResetStream {
    /// The stream being reset.
    pub id: StreamId,
    /// Application-defined error code.
    pub code: VarInt,
    /// Total bytes sent on the stream before the reset (for flow control accounting).
    pub final_size: u64,
    /// `Some(reliable_size)` when this was decoded from a RESET_STREAM_AT frame
    /// (`0x24`, draft-02); `None` for a plain RESET_STREAM (`0x04`). The session
    /// uses this to enforce that RESET_STREAM_AT is only accepted when we
    /// advertised the `reset_stream_at` transport parameter. We never emit
    /// RESET_STREAM_AT, so the encoder ignores this field.
    pub reliable_size: Option<u64>,
}

/// Requests that the peer stop sending on a stream.
#[derive(Debug, Clone)]
pub struct StopSending {
    /// The stream to stop.
    pub id: StreamId,
    /// Application-defined error code.
    pub code: VarInt,
}

/// Transport CONNECTION_CLOSE (0x1c): a protocol violation or transport error the
/// sender detected. The receiver surfaces it as an *abnormal* close. Per RFC 9000
/// §19.19 the transport variant carries the Frame Type field (we always send 0).
#[derive(Debug, Clone)]
pub struct ConnectionClose {
    /// Error code (RFC 9000 transport error-code space).
    pub code: VarInt,
    /// Human-readable reason for closing.
    pub reason: String,
}

/// APPLICATION_CLOSE (0x1d): a graceful, app-initiated close carrying an
/// application code/reason. The receiver surfaces it as a *clean* session close.
/// Per RFC 9000 §19.19 the application variant omits the Frame Type field.
///
/// The close subtype, not the code, carries the graceful/abnormal distinction —
/// codes are application-defined, so a graceful close may use any value.
#[derive(Debug, Clone)]
pub struct ApplicationClose {
    /// Application-defined error code.
    pub code: VarInt,
    /// Human-readable reason for closing.
    pub reason: String,
}

/// A QX_PING frame for connection liveness probing (draft-01).
#[derive(Debug, Clone)]
pub struct Ping {
    /// Monotonically increasing sequence number.
    pub sequence: u64,
    /// Whether this is a response (true) or request (false).
    pub response: bool,
}

/// An unreliable datagram (RFC 9221).
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct Datagram {
    /// The payload bytes.
    pub data: Bytes,
    /// Whether the frame carried an explicit length varint on the wire (the
    /// `0x31` form) rather than the no-length `0x30` form, whose payload is
    /// delimited by the enclosing record. We always *emit* `0x31`; this is
    /// preserved on decode so [`frame_size`](Datagram::frame_size) can report
    /// the exact encoded size for `max_datagram_frame_size` validation.
    pub length_prefixed: bool,
}

impl Datagram {
    /// The encoded on-wire frame size in bytes: type byte + optional length
    /// varint + payload. This is what `max_datagram_frame_size` (RFC 9221)
    /// bounds, so the receive path compares against it.
    pub(crate) fn frame_size(&self) -> u64 {
        let len = self.data.len() as u64;
        // The type is a 1-byte varint (0x30/0x31); the length varint is only
        // present in the 0x31 form.
        let header = if self.length_prefixed {
            1 + super::varint_size(len)
        } else {
            1
        };
        header + len
    }
}

/// Build the length-prefixed (`0x31`) datagram we always emit. Decoders that
/// observe the no-length `0x30` form construct [`Datagram`] directly instead.
impl From<Bytes> for Datagram {
    fn from(data: Bytes) -> Self {
        Self {
            data,
            length_prefixed: true,
        }
    }
}

/// A QUIC-compatible frame for multiplexed transport.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum Frame {
    ResetStream(ResetStream),
    StopSending(StopSending),
    ConnectionClose(ConnectionClose),
    ApplicationClose(ApplicationClose),
    Stream(Stream),
    MaxData(u64),
    MaxStreamData {
        id: StreamId,
        max: u64,
    },
    MaxStreamsBidi(u64),
    MaxStreamsUni(u64),
    DataBlocked(u64),
    StreamDataBlocked {
        id: StreamId,
        limit: u64,
    },
    StreamsBlockedBidi(u64),
    StreamsBlockedUni(u64),
    TransportParameters(TransportParams),
    Ping(Ping),
    /// An unreliable datagram (RFC 9221).
    Datagram(Datagram),
}

impl Frame {
    /// Encode the frame into bytes using the given wire format version.
    ///
    /// For QMux01, this encodes the raw frame without a record wrapper —
    /// the transport layer is responsible for delimiting records (size
    /// varint on TCP/TLS; implicit on WebSocket message boundaries).
    pub fn encode(&self, version: Version) -> Result<Bytes, Error> {
        // Reject record-layer frames (QX_PING, DATAGRAM) for versions
        // that don't use records, so a misrouted call can't accidentally emit
        // draft-01+ wire bytes on a draft-00 / webtransport session.
        if !version.uses_records() {
            match self {
                Frame::Ping(_) | Frame::Datagram(_) => return Err(Error::InvalidFrameType(0)),
                _ => {}
            }
        }

        let mut buf = BytesMut::new();

        match version {
            Version::WebTransport => self.encode_wt(&mut buf)?,
            Version::QMux00 | Version::QMux01 | Version::QMux02 => self.encode_qmux(&mut buf)?,
        }

        Ok(buf.freeze())
    }

    /// Decode all frames from a QMux record payload (draft-01).
    ///
    /// A record contains one or more frames concatenated together.
    /// Returns a Vec of decoded frames (skipping PADDING and other ignored frames).
    pub fn decode_record(mut data: Bytes) -> Result<Vec<Self>, Error> {
        let mut frames = Vec::new();

        while data.has_remaining() {
            if let Some(frame) = Self::decode_qmux_one(&mut data)? {
                frames.push(frame);
            }
        }

        Ok(frames)
    }

    /// Decode a single QMux frame from a buffer, advancing past the consumed bytes.
    ///
    /// Unlike `decode_qmux`, this correctly handles multiple frames in a record
    /// by not consuming trailing bytes for STREAM frames without the LEN bit.
    fn decode_qmux_one(data: &mut Bytes) -> Result<Option<Self>, Error> {
        let frame_type = VarInt::decode(data)?.into_inner();

        // PADDING: single zero byte, already consumed by VarInt decode
        if frame_type == 0x00 {
            return Ok(None);
        }

        // STREAM frames: 0x08-0x0f
        if (0x08..=0x0f).contains(&frame_type) {
            let has_off = frame_type & 0x04 != 0;
            let has_len = frame_type & 0x02 != 0;
            let has_fin = frame_type & 0x01 != 0;

            let id = StreamId(VarInt::decode(data)?);

            let offset = if has_off {
                VarInt::decode(data)?.into_inner()
            } else {
                0
            };

            let stream_data = if has_len {
                let len = VarInt::decode(data)?.into_inner();
                if (data.remaining() as u64) < len {
                    return Err(Error::Short);
                }
                data.split_to(len as usize)
            } else {
                // No LEN bit: rest of record is payload
                data.split_to(data.remaining())
            };

            return Ok(Some(Frame::Stream(Stream {
                id,
                offset,
                data: stream_data,
                fin: has_fin,
            })));
        }

        match frame_type {
            // RESET_STREAM
            0x04 => {
                let id = StreamId(VarInt::decode(data)?);
                let code = VarInt::decode(data)?;
                let final_size = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::ResetStream(ResetStream {
                    id,
                    code,
                    final_size,
                    reliable_size: None,
                })))
            }
            // RESET_STREAM_AT (draft-02). QMux runs on a reliable, ordered
            // transport, so every STREAM frame up to `final_size` has already
            // been delivered by the time this frame arrives — the Reliable Size
            // guarantee is automatically satisfied and we treat it as a plain
            // reset. `reliable_size > final_size` is a FRAME_ENCODING_ERROR; the
            // session additionally rejects it unless we advertised reset_stream_at.
            RESET_STREAM_AT => {
                let id = StreamId(VarInt::decode(data)?);
                let code = VarInt::decode(data)?;
                let final_size = VarInt::decode(data)?.into_inner();
                let reliable_size = VarInt::decode(data)?.into_inner();
                if reliable_size > final_size {
                    return Err(Error::FrameEncoding);
                }
                Ok(Some(Frame::ResetStream(ResetStream {
                    id,
                    code,
                    final_size,
                    reliable_size: Some(reliable_size),
                })))
            }
            // STOP_SENDING
            0x05 => {
                let id = StreamId(VarInt::decode(data)?);
                let code = VarInt::decode(data)?;
                Ok(Some(Frame::StopSending(StopSending { id, code })))
            }
            // CONNECTION_CLOSE (0x1c): carries the Frame Type field (RFC 9000 §19.19).
            0x1c => {
                let code = VarInt::decode(data)?;
                let _frame_type = VarInt::decode(data)?;
                let reason_len = VarInt::decode(data)?.into_inner();
                if (data.remaining() as u64) < reason_len {
                    return Err(Error::Short);
                }
                let reason =
                    String::from_utf8_lossy(&data.split_to(reason_len as usize)).into_owned();
                Ok(Some(Frame::ConnectionClose(ConnectionClose {
                    code,
                    reason,
                })))
            }
            // APPLICATION_CLOSE (0x1d): no Frame Type field (RFC 9000 §19.19).
            0x1d => {
                let code = VarInt::decode(data)?;
                let reason_len = VarInt::decode(data)?.into_inner();
                if (data.remaining() as u64) < reason_len {
                    return Err(Error::Short);
                }
                let reason =
                    String::from_utf8_lossy(&data.split_to(reason_len as usize)).into_owned();
                Ok(Some(Frame::ApplicationClose(ApplicationClose {
                    code,
                    reason,
                })))
            }
            // MAX_DATA
            0x10 => {
                let max = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::MaxData(max)))
            }
            // MAX_STREAM_DATA
            0x11 => {
                let id = StreamId(VarInt::decode(data)?);
                let max = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::MaxStreamData { id, max }))
            }
            // MAX_STREAMS (bidi)
            0x12 => {
                let max = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::MaxStreamsBidi(max)))
            }
            // MAX_STREAMS (uni)
            0x13 => {
                let max = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::MaxStreamsUni(max)))
            }
            // DATA_BLOCKED
            0x14 => {
                let limit = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::DataBlocked(limit)))
            }
            // STREAM_DATA_BLOCKED
            0x15 => {
                let id = StreamId(VarInt::decode(data)?);
                let limit = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::StreamDataBlocked { id, limit }))
            }
            // STREAMS_BLOCKED (bidi)
            0x16 => {
                let limit = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::StreamsBlockedBidi(limit)))
            }
            // STREAMS_BLOCKED (uni)
            0x17 => {
                let limit = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::StreamsBlockedUni(limit)))
            }
            // DATAGRAM without length — rest of record is payload
            0x30 => {
                let payload = data.split_to(data.remaining());
                Ok(Some(Frame::Datagram(Datagram {
                    data: payload,
                    length_prefixed: false,
                })))
            }
            // DATAGRAM with length
            0x31 => {
                let len = VarInt::decode(data)?.into_inner();
                if (data.remaining() as u64) < len {
                    return Err(Error::Short);
                }
                let payload = data.split_to(len as usize);
                Ok(Some(Frame::Datagram(Datagram {
                    data: payload,
                    length_prefixed: true,
                })))
            }
            // QX_TRANSPORT_PARAMETERS
            0x3f5153300d0a0d0a => {
                let len = VarInt::decode(data)?.into_inner();
                if (data.remaining() as u64) < len {
                    return Err(Error::Short);
                }
                let payload = data.split_to(len as usize);
                let params = TransportParams::decode(payload)?;
                Ok(Some(Frame::TransportParameters(params)))
            }
            // QX_PING request
            QX_PING_REQUEST => {
                let sequence = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::Ping(Ping {
                    sequence,
                    response: false,
                })))
            }
            // QX_PING response
            QX_PING_RESPONSE => {
                let sequence = VarInt::decode(data)?.into_inner();
                Ok(Some(Frame::Ping(Ping {
                    sequence,
                    response: true,
                })))
            }
            _ => Err(Error::InvalidFrameType(frame_type)),
        }
    }

    fn encode_wt(&self, buf: &mut BytesMut) -> Result<(), Error> {
        match self {
            Frame::Stream(s) => {
                buf.put_u8(if s.fin { 0x09 } else { 0x08 });
                s.id.0.encode(buf);
                buf.put_slice(&s.data);
            }
            Frame::ResetStream(r) => {
                buf.put_u8(0x04);
                r.id.0.encode(buf);
                r.code.encode(buf);
            }
            Frame::StopSending(s) => {
                buf.put_u8(0x05);
                s.id.0.encode(buf);
                s.code.encode(buf);
            }
            // The legacy WebTransport format keeps the reason as the rest of the
            // buffer (no Frame Type / length fields); only the type byte differs.
            Frame::ConnectionClose(c) => {
                buf.put_u8(0x1c);
                c.code.encode(buf);
                buf.put_slice(c.reason.as_bytes());
            }
            Frame::ApplicationClose(c) => {
                buf.put_u8(0x1d);
                c.code.encode(buf);
                buf.put_slice(c.reason.as_bytes());
            }
            // Flow control frames are QMux-only, not valid for WebTransport version
            _ => return Err(Error::InvalidFrameType(0)),
        }
        Ok(())
    }

    fn encode_qmux(&self, buf: &mut BytesMut) -> Result<(), Error> {
        match self {
            Frame::Stream(s) => {
                // Always set OFF (0x04) and LEN (0x02). Receivers preserve the
                // offset but intentionally defer continuity enforcement.
                let frame_type =
                    VarInt::from_u32(STREAM_BASE | 0x04 | 0x02 | if s.fin { 0x01 } else { 0 });
                frame_type.encode(buf);
                s.id.0.encode(buf);
                VarInt::try_from(s.offset)?.encode(buf);
                VarInt::try_from(s.data.len())?.encode(buf);
                buf.put_slice(&s.data);
            }
            Frame::ResetStream(r) => {
                RESET_STREAM.encode(buf);
                r.id.0.encode(buf);
                r.code.encode(buf);
                VarInt::try_from(r.final_size)?.encode(buf);
            }
            Frame::StopSending(s) => {
                STOP_SENDING.encode(buf);
                s.id.0.encode(buf);
                s.code.encode(buf);
            }
            Frame::ConnectionClose(c) => {
                CONNECTION_CLOSE.encode(buf);
                c.code.encode(buf);
                // Frame Type field — present for 0x1c (RFC 9000 §19.19). We don't
                // track which frame tripped the error, so it's always 0 (PADDING).
                VarInt::from(0u32).encode(buf);
                let reason_bytes = c.reason.as_bytes();
                VarInt::try_from(reason_bytes.len())?.encode(buf);
                buf.put_slice(reason_bytes);
            }
            Frame::ApplicationClose(c) => {
                APPLICATION_CLOSE.encode(buf);
                c.code.encode(buf);
                // No Frame Type field for 0x1d (RFC 9000 §19.19).
                let reason_bytes = c.reason.as_bytes();
                VarInt::try_from(reason_bytes.len())?.encode(buf);
                buf.put_slice(reason_bytes);
            }
            Frame::MaxData(max) => {
                MAX_DATA.encode(buf);
                VarInt::try_from(*max)?.encode(buf);
            }
            Frame::MaxStreamData { id, max } => {
                MAX_STREAM_DATA.encode(buf);
                id.0.encode(buf);
                VarInt::try_from(*max)?.encode(buf);
            }
            Frame::MaxStreamsBidi(max) => {
                MAX_STREAMS_BIDI.encode(buf);
                VarInt::try_from(*max)?.encode(buf);
            }
            Frame::MaxStreamsUni(max) => {
                MAX_STREAMS_UNI.encode(buf);
                VarInt::try_from(*max)?.encode(buf);
            }
            Frame::DataBlocked(limit) => {
                DATA_BLOCKED.encode(buf);
                VarInt::try_from(*limit)?.encode(buf);
            }
            Frame::StreamDataBlocked { id, limit } => {
                STREAM_DATA_BLOCKED.encode(buf);
                id.0.encode(buf);
                VarInt::try_from(*limit)?.encode(buf);
            }
            Frame::StreamsBlockedBidi(limit) => {
                STREAMS_BLOCKED_BIDI.encode(buf);
                VarInt::try_from(*limit)?.encode(buf);
            }
            Frame::StreamsBlockedUni(limit) => {
                STREAMS_BLOCKED_UNI.encode(buf);
                VarInt::try_from(*limit)?.encode(buf);
            }
            Frame::TransportParameters(params) => {
                QX_TRANSPORT_PARAMETERS_VI.encode(buf);
                let payload = params.encode()?;
                VarInt::try_from(payload.len())?.encode(buf);
                buf.put_slice(&payload);
            }
            Frame::Ping(ping) => {
                if ping.response {
                    QX_PING_RESPONSE_VI.encode(buf);
                } else {
                    QX_PING_REQUEST_VI.encode(buf);
                }
                VarInt::try_from(ping.sequence)?.encode(buf);
            }
            Frame::Datagram(datagram) => {
                // Length-prefixed form (0x31): self-delimiting regardless of
                // framing. See DATAGRAM_LEN for why we emit it uniformly.
                DATAGRAM_LEN.encode(buf);
                VarInt::try_from(datagram.data.len())?.encode(buf);
                buf.put_slice(&datagram.data);
            }
        }

        Ok(())
    }

    /// Decode a frame from bytes using the given wire format version.
    ///
    /// Returns `Ok(None)` for recognized but ignored frame types (e.g. flow control).
    pub fn decode(data: Bytes, version: Version) -> Result<Option<Self>, Error> {
        if data.is_empty() {
            return Err(Error::Short);
        }

        match version {
            Version::WebTransport => Self::decode_wt(data).map(Some),
            Version::QMux00 | Version::QMux01 | Version::QMux02 => Self::decode_qmux(data),
        }
    }

    fn decode_wt(mut data: Bytes) -> Result<Self, Error> {
        let frame_type = data.get_u8();

        match frame_type {
            0x04 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let code = VarInt::decode(&mut data)?;
                // WebTransport wire format has no final_size; flow control is QMux-only.
                Ok(Frame::ResetStream(ResetStream {
                    id,
                    code,
                    final_size: 0,
                    reliable_size: None,
                }))
            }
            0x05 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let code = VarInt::decode(&mut data)?;
                Ok(Frame::StopSending(StopSending { id, code }))
            }
            0x08 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                Ok(Frame::Stream(Stream {
                    id,
                    offset: 0,
                    data,
                    fin: false,
                }))
            }
            0x09 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                Ok(Frame::Stream(Stream {
                    id,
                    offset: 0,
                    data,
                    fin: true,
                }))
            }
            0x1c => {
                let code = VarInt::decode(&mut data)?;
                let reason = String::from_utf8_lossy(&data).into_owned();
                Ok(Frame::ConnectionClose(ConnectionClose { code, reason }))
            }
            0x1d => {
                let code = VarInt::decode(&mut data)?;
                let reason = String::from_utf8_lossy(&data).into_owned();
                Ok(Frame::ApplicationClose(ApplicationClose { code, reason }))
            }
            _ => Err(Error::InvalidFrameType(frame_type as u64)),
        }
    }

    fn decode_qmux(mut data: Bytes) -> Result<Option<Self>, Error> {
        let frame_type = VarInt::decode(&mut data)?.into_inner();

        // STREAM frames: 0x08-0x0f
        if (0x08..=0x0f).contains(&frame_type) {
            let has_off = frame_type & 0x04 != 0;
            let has_len = frame_type & 0x02 != 0;
            let has_fin = frame_type & 0x01 != 0;

            let id = StreamId(VarInt::decode(&mut data)?);

            let offset = if has_off {
                VarInt::decode(&mut data)?.into_inner()
            } else {
                0
            };

            let stream_data = if has_len {
                let len = VarInt::decode(&mut data)?.into_inner();
                if (data.remaining() as u64) < len {
                    return Err(Error::Short);
                }
                data.split_to(len as usize)
            } else {
                data.split_to(data.remaining())
            };

            return Ok(Some(Frame::Stream(Stream {
                id,
                offset,
                data: stream_data,
                fin: has_fin,
            })));
        }

        match frame_type {
            // PADDING
            0x00 => Ok(None),
            // RESET_STREAM
            0x04 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let code = VarInt::decode(&mut data)?;
                let final_size = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::ResetStream(ResetStream {
                    id,
                    code,
                    final_size,
                    reliable_size: None,
                })))
            }
            // RESET_STREAM_AT (draft-02); see the record-decoder arm for why we
            // treat it as a plain reset.
            RESET_STREAM_AT => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let code = VarInt::decode(&mut data)?;
                let final_size = VarInt::decode(&mut data)?.into_inner();
                let reliable_size = VarInt::decode(&mut data)?.into_inner();
                if reliable_size > final_size {
                    return Err(Error::FrameEncoding);
                }
                Ok(Some(Frame::ResetStream(ResetStream {
                    id,
                    code,
                    final_size,
                    reliable_size: Some(reliable_size),
                })))
            }
            // STOP_SENDING
            0x05 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let code = VarInt::decode(&mut data)?;
                Ok(Some(Frame::StopSending(StopSending { id, code })))
            }
            // CONNECTION_CLOSE (0x1c): carries the Frame Type field (RFC 9000 §19.19).
            0x1c => {
                let code = VarInt::decode(&mut data)?;
                let _frame_type = VarInt::decode(&mut data)?;
                let reason_len = VarInt::decode(&mut data)?.into_inner();
                if (data.remaining() as u64) < reason_len {
                    return Err(Error::Short);
                }
                let reason =
                    String::from_utf8_lossy(&data.split_to(reason_len as usize)).into_owned();
                Ok(Some(Frame::ConnectionClose(ConnectionClose {
                    code,
                    reason,
                })))
            }
            // APPLICATION_CLOSE (0x1d): no Frame Type field (RFC 9000 §19.19).
            0x1d => {
                let code = VarInt::decode(&mut data)?;
                let reason_len = VarInt::decode(&mut data)?.into_inner();
                if (data.remaining() as u64) < reason_len {
                    return Err(Error::Short);
                }
                let reason =
                    String::from_utf8_lossy(&data.split_to(reason_len as usize)).into_owned();
                Ok(Some(Frame::ApplicationClose(ApplicationClose {
                    code,
                    reason,
                })))
            }
            // MAX_DATA
            0x10 => {
                let max = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::MaxData(max)))
            }
            // MAX_STREAM_DATA
            0x11 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let max = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::MaxStreamData { id, max }))
            }
            // MAX_STREAMS (bidi)
            0x12 => {
                let max = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::MaxStreamsBidi(max)))
            }
            // MAX_STREAMS (uni)
            0x13 => {
                let max = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::MaxStreamsUni(max)))
            }
            // DATA_BLOCKED
            0x14 => {
                let limit = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::DataBlocked(limit)))
            }
            // STREAM_DATA_BLOCKED
            0x15 => {
                let id = StreamId(VarInt::decode(&mut data)?);
                let limit = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::StreamDataBlocked { id, limit }))
            }
            // STREAMS_BLOCKED (bidi)
            0x16 => {
                let limit = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::StreamsBlockedBidi(limit)))
            }
            // STREAMS_BLOCKED (uni)
            0x17 => {
                let limit = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::StreamsBlockedUni(limit)))
            }
            // DATAGRAM without length — rest of message is payload
            0x30 => {
                let payload = data.split_to(data.remaining());
                Ok(Some(Frame::Datagram(Datagram {
                    data: payload,
                    length_prefixed: false,
                })))
            }
            // DATAGRAM with length
            0x31 => {
                let len = VarInt::decode(&mut data)?.into_inner();
                if (data.remaining() as u64) < len {
                    return Err(Error::Short);
                }
                let payload = data.split_to(len as usize);
                Ok(Some(Frame::Datagram(Datagram {
                    data: payload,
                    length_prefixed: true,
                })))
            }
            // QX_TRANSPORT_PARAMETERS
            0x3f5153300d0a0d0a => {
                let len = VarInt::decode(&mut data)?.into_inner();
                if (data.remaining() as u64) < len {
                    return Err(Error::Short);
                }
                let payload = data.split_to(len as usize);
                let params = TransportParams::decode(payload)?;
                Ok(Some(Frame::TransportParameters(params)))
            }
            // QX_PING request
            QX_PING_REQUEST => {
                let sequence = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::Ping(Ping {
                    sequence,
                    response: false,
                })))
            }
            // QX_PING response
            QX_PING_RESPONSE => {
                let sequence = VarInt::decode(&mut data)?.into_inner();
                Ok(Some(Frame::Ping(Ping {
                    sequence,
                    response: true,
                })))
            }
            _ => Err(Error::InvalidFrameType(frame_type)),
        }
    }
}

impl From<Stream> for Frame {
    fn from(stream: Stream) -> Self {
        Frame::Stream(stream)
    }
}

impl From<ResetStream> for Frame {
    fn from(reset: ResetStream) -> Self {
        Frame::ResetStream(reset)
    }
}

impl From<StopSending> for Frame {
    fn from(stop: StopSending) -> Self {
        Frame::StopSending(stop)
    }
}

impl From<ConnectionClose> for Frame {
    fn from(close: ConnectionClose) -> Self {
        Frame::ConnectionClose(close)
    }
}

impl From<ApplicationClose> for Frame {
    fn from(close: ApplicationClose) -> Self {
        Frame::ApplicationClose(close)
    }
}