h3x 0.6.1

use std::{
    collections::{BTreeMap, btree_map},
    convert::Infallible,
    iter,
    pin::pin,
};

use bytes::Buf;
use futures::TryStreamExt;
use snafu::Snafu;
use tokio::io::{AsyncBufRead, AsyncRead, AsyncWrite};

use crate::{
    buflist::BufList,
    codec::{DecodeExt, DecodeFrom, EncodeExt, EncodeInto, StreamDecodeError},
    connection::StreamError,
    dhttp::{frame::Frame, stream::UnidirectionalStream},
    error::{Code, H3ConnectionError, H3CriticalStreamClosed, H3FrameDecodeError},
    quic,
    varint::VarInt,
};

/// ``` ignore
/// Setting {
///   Identifier (i),
///   Value (i),
/// }
/// ```
///
/// <https://datatracker.ietf.org/doc/html/rfc9114#name-settings>
pub struct Setting {
    pub id: VarInt,
    pub value: VarInt,
}

impl Setting {
    pub const fn new(id: VarInt, value: VarInt) -> Self {
        Self { id, value }
    }

    pub fn check(&self) -> Result<(), InvalidSettingValue> {
        if is_boolean_setting(self.id)
            && self.value != VarInt::from_u32(0)
            && self.value != VarInt::from_u32(1)
        {
            return Err(InvalidSettingValue::BoolSetting {
                id: self.id,
                value: self.value,
            });
        }
        Ok(())
    }
}

impl From<(VarInt, VarInt)> for Setting {
    fn from((id, value): (VarInt, VarInt)) -> Self {
        Self::new(id, value)
    }
}

#[derive(Snafu, Debug, Clone, Copy)]
pub enum InvalidSettingValue {
    #[snafu(display("boolean setting {id} must have value 0 or 1, got {value}"))]
    BoolSetting { id: VarInt, value: VarInt },
}

impl H3ConnectionError for InvalidSettingValue {
    fn code(&self) -> Code {
        Code::H3_SETTINGS_ERROR
    }
}

const fn is_boolean_setting(id: VarInt) -> bool {
    let id = id.into_inner();
    id == crate::extended_connect::settings::EnableConnectProtocol::ID.into_inner()
        || id == crate::dhttp::datagram::settings::H3Datagram::ID.into_inner()
        || is_webtransport_boolean_setting(id)
}

#[cfg(feature = "webtransport")]
const fn is_webtransport_boolean_setting(id: u64) -> bool {
    id == crate::dhttp::webtransport::settings::EnableWebTransport::ID.into_inner()
}

#[cfg(not(feature = "webtransport"))]
const fn is_webtransport_boolean_setting(_id: u64) -> bool {
    false
}

impl<S: AsyncRead + Send> DecodeFrom<S> for Setting {
    type Error = StreamError;

    async fn decode_from(stream: S) -> Result<Self, Self::Error> {
        let decode = async move {
            let mut stream = pin!(stream);
            let id = stream.decode_one().await?;
            let value = stream.decode_one().await?;
            Ok(Setting { id, value })
        };
        let setting = decode.await.map_err(|error: StreamDecodeError| {
            error
                .escalate_critical_close(|| H3CriticalStreamClosed::Control.into())
                .into_stream_error(|decode_error| {
                    H3FrameDecodeError {
                        source: decode_error,
                    }
                    .into()
                })
        })?;

        setting.check()?;
        Ok(setting)
    }
}

impl<S: AsyncWrite + Send> EncodeInto<S> for Setting {
    type Output = ();

    type Error = StreamError;

    async fn encode_into(self, stream: S) -> Result<Self::Output, Self::Error> {
        let Setting { id, value } = self;
        let encode = async move {
            let mut stream = pin!(stream);
            stream.as_mut().encode_one(id).await?;
            stream.as_mut().encode_one(value).await?;
            Ok(())
        };
        encode
            .await
            .map_err(|error: quic::StreamError| match error {
                quic::StreamError::Reset { .. } => H3CriticalStreamClosed::Control.into(),
                quic::StreamError::Connection { .. } => error.into(),
            })
    }
}

#[derive(Default, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Settings {
    map: BTreeMap<VarInt, VarInt>,
}

impl<S> DecodeFrom<S> for Settings
where
    for<'s> &'s mut S: AsyncBufRead,
    S: Send,
{
    type Error = StreamError;

    async fn decode_from(stream: S) -> Result<Self, Self::Error> {
        let mut stream = pin!(stream.into_decode_stream::<Setting, StreamError>());
        let mut settings = Settings::default();
        while let Some(setting) = stream.try_next().await? {
            settings.set(setting);
        }
        Ok(settings)
    }
}

impl EncodeInto<BufList> for &Settings {
    type Output = Frame<BufList>;

    type Error = Infallible;

    async fn encode_into(self, stream: BufList) -> Result<Self::Output, Self::Error> {
        assert!(!stream.has_remaining());
        let mut frame = Frame::new(Frame::SETTINGS_FRAME_TYPE, stream)
            .expect("SETTINGS frame type is a valid VarInt");
        for setting in self {
            frame
                .encode_one(setting)
                .await
                .expect("encoding a Setting into a BufList is infallible");
        }
        Ok(frame)
    }
}

impl EncodeInto<BufList> for Settings {
    type Output = Frame<BufList>;

    type Error = Infallible;

    async fn encode_into(self, stream: BufList) -> Result<Self::Output, Self::Error> {
        (&self).encode_into(stream).await
    }
}

impl Settings {
    /// Typed access to a setting value. The return type depends on the setting:
    ///
    /// - Concrete setting types (`crate::qpack::settings::QpackMaxTableCapacity`,
    ///   `MaxFieldSectionSize`, …) apply defaults and return their associated `Value` type.
    /// - A raw [`VarInt`] identifier returns `Option<VarInt>` with no default fallback.
    ///
    /// ```ignore
    /// settings.get(crate::qpack::settings::QpackMaxTableCapacity) // → VarInt (with default)
    /// settings.get(MaxFieldSectionSize)                           // → Option<VarInt>
    /// settings.get(VarInt::from_u32(0x06))                        // → Option<VarInt> (raw)
    /// ```
    pub fn get<S: SettingId>(&self, id: S) -> S::Value {
        id.value_from(self)
    }

    pub(crate) fn get_raw(&self, id: VarInt) -> Option<VarInt> {
        self.map.get(&id).copied()
    }

    pub fn max_field_section_size(&self) -> Option<VarInt> {
        self.get(MaxFieldSectionSize)
    }

    pub fn set(&mut self, Setting { id, value }: Setting) {
        self.map.insert(id, value);
    }

    pub fn with(mut self, setting: Setting) -> Self {
        self.set(setting);
        self
    }

    pub fn with_all(mut self, settings: impl IntoIterator<Item = Setting>) -> Self {
        self.extend(settings);
        self
    }
}

impl IntoIterator for Settings {
    type Item = Setting;

    type IntoIter = iter::Map<btree_map::IntoIter<VarInt, VarInt>, fn((VarInt, VarInt)) -> Setting>;

    fn into_iter(self) -> Self::IntoIter {
        self.map
            .into_iter()
            .map(|(id, value)| Setting { id, value })
    }
}

impl<'s> IntoIterator for &'s Settings {
    type Item = Setting;

    type IntoIter = iter::Map<
        btree_map::Iter<'s, VarInt, VarInt>,
        for<'v> fn((&'v VarInt, &'v VarInt)) -> Setting,
    >;

    fn into_iter(self) -> Self::IntoIter {
        self.map.iter().map(|(&id, &value)| Setting { id, value })
    }
}

impl FromIterator<Setting> for Settings {
    fn from_iter<T: IntoIterator<Item = Setting>>(iter: T) -> Self {
        Self {
            map: iter
                .into_iter()
                .map(|Setting { id, value }| (id, value))
                .collect::<BTreeMap<_, _>>(),
        }
    }
}

impl Extend<Setting> for Settings {
    fn extend<T: IntoIterator<Item = Setting>>(&mut self, iter: T) {
        self.map
            .extend(iter.into_iter().map(|Setting { id, value }| (id, value)));
    }
}

/// Trait for typed HTTP/3 setting identifiers.
///
/// Each setting has an associated `Value` type that encodes whether a default
/// exists: settings with defaults use `VarInt` (always returns a value),
/// while optional settings use `Option<VarInt>`.
pub trait SettingId {
    /// The value type returned when querying this setting.
    type Value;

    /// The wire-format setting identifier (RFC 9114 / RFC 9204).
    fn id(&self) -> VarInt;

    /// Extract the typed value from `Settings`, applying defaults if applicable.
    fn value_from(&self, settings: &Settings) -> Self::Value;
}

/// Raw access by [`VarInt`] identifier — returns the explicitly set value, or
/// `None` if not present. No default-value fallback is applied.
impl SettingId for VarInt {
    type Value = Option<VarInt>;

    fn id(&self) -> VarInt {
        *self
    }

    fn value_from(&self, settings: &Settings) -> Option<VarInt> {
        settings.get_raw(*self)
    }
}

/// `SETTINGS_MAX_FIELD_SECTION_SIZE` (0x06). No default (unlimited).
///
/// An HTTP/3 implementation MAY impose a limit on the maximum size of the
/// message header it will accept on an individual HTTP message. The size
/// of a field list is calculated based on the uncompressed size of fields,
/// including the length of the name and value in bytes plus an overhead of
/// 32 bytes for each field.
///
/// <https://datatracker.ietf.org/doc/html/rfc9114#name-header-size-constraints>
pub struct MaxFieldSectionSize;

impl MaxFieldSectionSize {
    pub const ID: VarInt = VarInt::from_u32(0x06);

    pub const fn setting(value: VarInt) -> Setting {
        Setting::new(Self::ID, value)
    }
}

impl SettingId for MaxFieldSectionSize {
    type Value = Option<VarInt>;

    fn id(&self) -> VarInt {
        Self::ID
    }

    fn value_from(&self, settings: &Settings) -> Option<VarInt> {
        settings.get_raw(Self::ID)
    }
}

impl UnidirectionalStream<()> {
    /// A control stream is indicated by a stream type of 0x00. Data on this
    ///  stream consists of HTTP/3 frames, as defined in Section 7.2.
    ///
    /// <https://datatracker.ietf.org/doc/html/rfc9114#name-control-streams>
    pub const CONTROL_STREAM_TYPE: VarInt = VarInt::from_u32(0x00);
}

impl<S: ?Sized> UnidirectionalStream<S> {
    pub const fn is_control_stream(&self) -> bool {
        self.r#type().into_inner() == UnidirectionalStream::CONTROL_STREAM_TYPE.into_inner()
    }

    pub async fn initial_control_stream(stream: S) -> Result<Self, StreamError>
    where
        S: AsyncWrite + Unpin + Sized + Send,
    {
        Self::initial(UnidirectionalStream::CONTROL_STREAM_TYPE, stream)
            .await
            .map_err(|error| error.map_stream_reset(|_| H3CriticalStreamClosed::Control.into()))
    }
}

#[cfg(test)]
mod tests {
    use std::{
        io,
        pin::Pin,
        task::{Context, Poll},
    };

    use bytes::Buf;
    use tokio::io::AsyncWriteExt;

    use super::*;
    #[cfg(feature = "webtransport")]
    use crate::dhttp::webtransport::settings::EnableWebTransport;
    use crate::{
        codec::{DecodeError, DecodeExt, EncodeExt},
        connection,
        dhttp::datagram::settings::H3Datagram,
        extended_connect::settings::EnableConnectProtocol,
        quic,
        varint::VarInt,
    };

    #[derive(Clone)]
    struct FailWrite {
        error: quic::StreamError,
    }

    impl FailWrite {
        fn reset(code: VarInt) -> Self {
            Self {
                error: quic::StreamError::Reset { code },
            }
        }

        fn connection() -> Self {
            Self {
                error: quic::StreamError::Connection {
                    source: quic_connection_error(),
                },
            }
        }
    }

    struct FailAfterWrites {
        successful_writes_before_failure: usize,
        error: quic::StreamError,
    }

    impl FailAfterWrites {
        fn new(successful_writes_before_failure: usize, error: quic::StreamError) -> Self {
            Self {
                successful_writes_before_failure,
                error,
            }
        }
    }

    impl tokio::io::AsyncWrite for FailAfterWrites {
        fn poll_write(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<io::Result<usize>> {
            if self.successful_writes_before_failure == 0 {
                return Poll::Ready(Err(io::Error::from(self.error.clone())));
            }

            self.successful_writes_before_failure -= 1;
            Poll::Ready(Ok(buf.len()))
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    struct FailRead;

    impl tokio::io::AsyncRead for FailRead {
        fn poll_read(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            _buf: &mut tokio::io::ReadBuf<'_>,
        ) -> Poll<io::Result<()>> {
            Poll::Ready(Err(DecodeError::ArithmeticOverflow.into()))
        }
    }

    impl tokio::io::AsyncWrite for FailWrite {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            _buf: &[u8],
        ) -> Poll<io::Result<usize>> {
            Poll::Ready(Err(io::Error::from(self.error.clone())))
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    struct FailBufRead {
        error: Option<connection::StreamError>,
    }

    impl FailBufRead {
        fn new(error: connection::StreamError) -> Self {
            Self { error: Some(error) }
        }
    }

    impl tokio::io::AsyncRead for FailBufRead {
        fn poll_read(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            _buf: &mut tokio::io::ReadBuf<'_>,
        ) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    impl tokio::io::AsyncBufRead for FailBufRead {
        fn poll_fill_buf(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
            let error = self
                .get_mut()
                .error
                .take()
                .expect("test stream should be polled once");
            Poll::Ready(Err(io::Error::from(error)))
        }

        fn consume(self: Pin<&mut Self>, _amt: usize) {}
    }

    fn quic_connection_error() -> quic::ConnectionError {
        quic::ConnectionError::Application {
            source: quic::ApplicationError {
                code: Code::H3_INTERNAL_ERROR,
                reason: "test failure".into(),
            },
        }
    }

    fn assert_h3_connection_code(error: StreamError, expected: Code) {
        assert!(matches!(
            error,
            StreamError::Connection {
                source: connection::ConnectionError::H3 { source },
            } if source.code() == expected
        ));
    }

    fn assert_quic_connection_error(error: StreamError) {
        assert!(matches!(
            error,
            StreamError::Connection {
                source: connection::ConnectionError::Quic { .. },
            }
        ));
    }

    #[test]
    fn boolean_setting_validation_uses_new_owner_modules() {
        for id in [EnableConnectProtocol::ID, H3Datagram::ID] {
            let err = Setting::new(id, VarInt::from_u32(2))
                .check()
                .expect_err("boolean setting value 2 must be rejected");
            assert!(matches!(err, InvalidSettingValue::BoolSetting { .. }));
        }

        #[cfg(feature = "webtransport")]
        {
            let err = Setting::new(EnableWebTransport::ID, VarInt::from_u32(2))
                .check()
                .expect_err("webtransport boolean setting value 2 must be rejected");
            assert!(matches!(err, InvalidSettingValue::BoolSetting { .. }));
        }
    }

    #[test]
    fn setting_construction_validation_and_error_metadata() {
        let setting = Setting::from((MaxFieldSectionSize::ID, VarInt::from_u32(4096)));
        assert_eq!(setting.id, MaxFieldSectionSize.id());
        assert_eq!(setting.value, VarInt::from_u32(4096));
        assert!(setting.check().is_ok());

        assert!(
            Setting::new(EnableConnectProtocol::ID, VarInt::from_u32(0))
                .check()
                .is_ok()
        );
        assert!(
            Setting::new(EnableConnectProtocol::ID, VarInt::from_u32(1))
                .check()
                .is_ok()
        );

        let error = Setting::new(EnableConnectProtocol::ID, VarInt::from_u32(2))
            .check()
            .expect_err("invalid boolean setting must fail");
        assert_eq!(error.code(), Code::H3_SETTINGS_ERROR);
        assert_eq!(
            error.to_string(),
            "boolean setting 8 must have value 0 or 1, got 2",
        );
    }

    #[test]
    fn non_boolean_settings_accept_arbitrary_values_and_validation_error_has_no_source() {
        let custom_setting = Setting::new(VarInt::from_u32(0x21), VarInt::MAX);
        assert!(custom_setting.check().is_ok());

        let error = Setting::new(H3Datagram::ID, VarInt::from_u32(42))
            .check()
            .expect_err("invalid boolean setting must fail");
        assert!(std::error::Error::source(&error).is_none());
        assert_eq!(error.code(), Code::H3_SETTINGS_ERROR);
        assert_eq!(
            error.to_string(),
            "boolean setting 51 must have value 0 or 1, got 42",
        );
    }

    #[test]
    fn settings_accessors_iterators_and_extension_paths() {
        let mut settings = Settings::default();
        assert_eq!(settings.get(VarInt::from_u32(0x1234)), None);
        assert_eq!(settings.max_field_section_size(), None);

        settings.set(MaxFieldSectionSize::setting(VarInt::from_u32(4096)));
        settings.extend([
            EnableConnectProtocol::setting(true),
            H3Datagram::setting(false),
        ]);
        settings.extend(std::iter::once(H3Datagram::setting(true)));

        assert_eq!(
            settings.get(MaxFieldSectionSize),
            Some(VarInt::from_u32(4096)),
        );
        assert_eq!(
            settings.max_field_section_size(),
            Some(VarInt::from_u32(4096)),
        );
        assert_eq!(
            settings.get(VarInt::from_u32(0x06)),
            Some(VarInt::from_u32(4096)),
        );
        assert!(settings.enable_connect_protocol());
        assert!(settings.h3_datagram());
        #[cfg(feature = "webtransport")]
        assert!(!settings.enable_webtransport());

        let borrowed: Vec<_> = (&settings).into_iter().collect();
        assert_eq!(borrowed.len(), 3);
        assert_eq!(borrowed[0].id, MaxFieldSectionSize::ID);

        let owned: Vec<_> = settings.clone().into_iter().collect();
        assert_eq!(owned.len(), borrowed.len());
        for (left, right) in owned.iter().zip(&borrowed) {
            assert_eq!(left.id, right.id);
            assert_eq!(left.value, right.value);
        }

        let rebuilt = Settings::from_iter(owned);
        assert_eq!(settings, rebuilt);
    }

    #[test]
    fn settings_with_and_with_all_compose_setting_fragments() {
        let settings = Settings::default()
            .with(MaxFieldSectionSize::setting(VarInt::from_u32(4096)))
            .with_all([
                EnableConnectProtocol::setting(true),
                H3Datagram::setting(false),
            ])
            .with(H3Datagram::setting(true));

        assert_eq!(
            settings.max_field_section_size(),
            Some(VarInt::from_u32(4096)),
        );
        assert!(settings.enable_connect_protocol());
        assert!(settings.h3_datagram());
    }

    #[test]
    fn setting_id_methods_return_wire_ids_and_typed_values() {
        let mut settings = Settings::default();
        let raw_id = VarInt::from_u32(0x21);

        assert_eq!(raw_id.id(), raw_id);
        assert_eq!(raw_id.value_from(&settings), None);
        assert_eq!(MaxFieldSectionSize.id(), MaxFieldSectionSize::ID);
        assert_eq!(MaxFieldSectionSize.value_from(&settings), None);

        settings.set(Setting::new(raw_id, VarInt::from_u32(7)));
        settings.set(MaxFieldSectionSize::setting(VarInt::from_u32(4096)));

        assert_eq!(raw_id.value_from(&settings), Some(VarInt::from_u32(7)));
        assert_eq!(settings.get(raw_id), Some(VarInt::from_u32(7)));
        assert_eq!(
            MaxFieldSectionSize.value_from(&settings),
            Some(VarInt::from_u32(4096)),
        );
    }

    #[tokio::test]
    async fn setting_decode_maps_incomplete_id_and_value_to_closed_control_stream() {
        for payload in [
            BufList::new(),
            BufList::from_buf(&[MaxFieldSectionSize::ID.into_inner() as u8][..]),
        ] {
            let error = match payload.decode::<Setting>().await {
                Ok(_) => panic!("incomplete setting must fail"),
                Err(error) => error,
            };
            assert_h3_connection_code(error, Code::H3_CLOSED_CRITICAL_STREAM);
        }
    }

    #[tokio::test]
    async fn setting_decode_maps_payload_decode_error_to_frame_decode_error() {
        let error = FailRead
            .decode::<Setting>()
            .await
            .err()
            .expect("typed decode failure should be a frame decode error");

        assert_h3_connection_code(error, Code::H3_FRAME_ERROR);
    }

    #[tokio::test]
    async fn setting_encode_maps_reset_to_closed_control_stream_and_preserves_connection_errors() {
        let mut idle_writer = FailWrite::reset(VarInt::from_u32(0));
        idle_writer.flush().await.expect("flush succeeds");
        idle_writer.shutdown().await.expect("shutdown succeeds");

        let reset_code = VarInt::from_u32(77);
        let error = Setting::new(MaxFieldSectionSize::ID, VarInt::from_u32(1))
            .encode_into(FailWrite::reset(reset_code))
            .await
            .expect_err("write reset must fail");
        assert_h3_connection_code(error, Code::H3_CLOSED_CRITICAL_STREAM);

        let error = Setting::new(MaxFieldSectionSize::ID, VarInt::from_u32(1))
            .encode_into(FailWrite::connection())
            .await
            .expect_err("connection write failure must fail");
        assert_quic_connection_error(error);
    }

    #[tokio::test]
    async fn settings_decode_propagates_stream_fill_buf_errors() {
        let reset_code = VarInt::from_u32(88);
        let reset = FailBufRead::new(StreamError::Reset { code: reset_code })
            .decode::<Settings>()
            .await
            .expect_err("fill_buf reset must fail");
        assert!(matches!(reset, StreamError::Reset { code } if code == reset_code));

        let connection =
            FailBufRead::new(connection::ConnectionError::from(quic_connection_error()).into())
                .decode::<Settings>()
                .await
                .expect_err("fill_buf connection error must fail");
        assert_quic_connection_error(connection);
    }

    #[tokio::test]
    async fn setting_encode_maps_value_write_reset_to_closed_control_stream() {
        let mut idle_writer = FailAfterWrites::new(
            1,
            quic::StreamError::Reset {
                code: VarInt::from_u32(0),
            },
        );
        idle_writer.flush().await.expect("flush succeeds");
        idle_writer.shutdown().await.expect("shutdown succeeds");

        let error = Setting::new(MaxFieldSectionSize::ID, VarInt::from_u32(4096))
            .encode_into(FailAfterWrites::new(
                1,
                quic::StreamError::Reset {
                    code: VarInt::from_u32(123),
                },
            ))
            .await
            .expect_err("value write reset must fail");

        assert_h3_connection_code(error, Code::H3_CLOSED_CRITICAL_STREAM);
    }

    #[tokio::test]
    async fn setting_encode_decode_round_trips_and_rejects_invalid_bool() {
        let mut encoded = BufList::new();
        encoded
            .encode_one(Setting::new(
                MaxFieldSectionSize::ID,
                VarInt::from_u32(4096),
            ))
            .await
            .expect("setting encoding into buflist is infallible");
        let decoded = encoded.decode::<Setting>().await.expect("setting decodes");
        assert_eq!(decoded.id, MaxFieldSectionSize::ID);
        assert_eq!(decoded.value, VarInt::from_u32(4096));

        let mut invalid = BufList::new();
        invalid
            .encode_one(Setting::new(H3Datagram::ID, VarInt::from_u32(2)))
            .await
            .expect("setting encoding into buflist is infallible");
        let error = invalid
            .decode::<Setting>()
            .await
            .err()
            .expect("invalid boolean setting must fail to decode");
        assert_h3_connection_code(error, Code::H3_SETTINGS_ERROR);
    }

    #[cfg(feature = "webtransport")]
    #[tokio::test]
    async fn setting_decode_rejects_invalid_webtransport_bool_when_feature_enabled() {
        let mut invalid = BufList::new();
        invalid
            .encode_one(Setting::new(EnableWebTransport::ID, VarInt::from_u32(2)))
            .await
            .expect("setting encoding into buflist is infallible");

        let error = invalid
            .decode::<Setting>()
            .await
            .err()
            .expect("invalid webtransport boolean setting must fail to decode");

        assert_h3_connection_code(error, Code::H3_SETTINGS_ERROR);
    }

    #[tokio::test]
    async fn settings_encode_to_frame_and_decode_payload() {
        let settings = Settings::from_iter([
            MaxFieldSectionSize::setting(VarInt::from_u32(8192)),
            EnableConnectProtocol::setting(true),
        ]);

        let frame = BufList::new()
            .encode(&settings)
            .await
            .expect("settings encoding into buflist is infallible");
        assert_eq!(frame.r#type(), Frame::SETTINGS_FRAME_TYPE);
        assert!(frame.length().into_inner() > 0);

        let decoded = frame
            .into_payload()
            .decode::<Settings>()
            .await
            .expect("settings decode from payload");
        assert_eq!(decoded, settings);

        let frame = BufList::new()
            .encode(settings.clone())
            .await
            .expect("owned settings encoding into buflist is infallible");
        assert_eq!(frame.r#type(), Frame::SETTINGS_FRAME_TYPE);
    }

    #[tokio::test]
    async fn empty_settings_encode_to_zero_length_frame_and_decode_to_default() {
        let settings = Settings::default();

        let frame = BufList::new()
            .encode(&settings)
            .await
            .expect("settings encoding into buflist is infallible");
        assert_eq!(frame.r#type(), Frame::SETTINGS_FRAME_TYPE);
        assert_eq!(frame.length(), VarInt::from_u32(0));

        let decoded = frame
            .into_payload()
            .decode::<Settings>()
            .await
            .expect("empty settings payload decodes");
        assert_eq!(decoded, settings);
    }

    #[tokio::test]
    async fn settings_decode_uses_last_value_for_duplicate_identifiers() {
        let mut encoded = BufList::new();
        encoded
            .encode_one(MaxFieldSectionSize::setting(VarInt::from_u32(1024)))
            .await
            .expect("setting encoding into buflist is infallible");
        encoded
            .encode_one(MaxFieldSectionSize::setting(VarInt::from_u32(2048)))
            .await
            .expect("setting encoding into buflist is infallible");

        let decoded = encoded
            .decode::<Settings>()
            .await
            .expect("settings payload decodes");

        assert_eq!(
            decoded.max_field_section_size(),
            Some(VarInt::from_u32(2048)),
        );
        assert_eq!(decoded.into_iter().count(), 1);
    }

    #[tokio::test]
    async fn initial_control_stream_maps_write_errors() {
        let error =
            UnidirectionalStream::initial_control_stream(FailWrite::reset(VarInt::from_u32(9)))
                .await
                .err()
                .expect("control stream reset must fail");
        assert_h3_connection_code(error, Code::H3_CLOSED_CRITICAL_STREAM);

        let error = UnidirectionalStream::initial_control_stream(FailWrite::connection())
            .await
            .err()
            .expect("control stream connection failure must fail");
        assert_quic_connection_error(error);
    }

    #[tokio::test]
    async fn control_stream_helpers_identify_and_write_stream_type() {
        let control = UnidirectionalStream::initial_control_stream(BufList::new())
            .await
            .expect("control stream initialization");

        assert!(control.is_control_stream());
        assert_eq!(control.r#type(), UnidirectionalStream::CONTROL_STREAM_TYPE);
        assert!(control.into_inner().has_remaining());
    }
}