// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use iggy::prelude::{
    AutoLogin as RustAutoLogin, Credentials as RustCredentials,
    HttpClientConfig as RustHttpClientConfig, HttpClientConfigBuilder,
    QuicClientConfig as RustQuicClientConfig, QuicClientConfigBuilder,
    QuicClientReconnectionConfig as RustQuicClientReconnectionConfig,
    TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder,
    TcpClientReconnectionConfig as RustTcpClientReconnectionConfig,
    WebSocketClientConfig as RustWebSocketClientConfig, WebSocketClientConfigBuilder,
    WebSocketClientReconnectionConfig as RustWebSocketClientReconnectionConfig,
    WebSocketConfig as RustWebSocketFramingConfig,
};
use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyDelta;
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
use pyo3_stub_gen::impl_stub_type;
use secrecy::SecretString;
use std::fmt::Display;
use std::net::SocketAddr;
use std::sync::Arc;

use crate::duration::{
    duration_repr, iggy_duration_to_py_delta, millis_repr, millis_to_py_delta,
    py_delta_to_iggy_duration, py_delta_to_millis, reject_zero,
};

/// The credentials replayed by the client every time it (re)connects.
///
/// `IggyClient` only recovers a lost session when it has credentials to replay,
/// so a long-running consumer should pass one of the enabled variants.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct AutoLogin {
    pub(crate) inner: RustAutoLogin,
}

#[gen_stub_pymethods]
#[pymethods]
impl AutoLogin {
    /// No automatic login. `login_user()` must be called by hand after every connect.
    #[staticmethod]
    fn disabled() -> Self {
        Self {
            inner: RustAutoLogin::Disabled,
        }
    }

    /// Log in with the given username and password on every connect.
    #[staticmethod]
    fn username_password(username: String, password: String) -> Self {
        Self {
            inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword(
                username,
                SecretString::from(password),
            )),
        }
    }

    /// Log in with the given personal access token on every connect.
    #[staticmethod]
    fn personal_access_token(token: String) -> Self {
        Self {
            inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(
                SecretString::from(token),
            )),
        }
    }

    /// Whether automatic login is enabled.
    #[getter]
    fn enabled(&self) -> bool {
        matches!(self.inner, RustAutoLogin::Enabled(_))
    }

    /// The username to log in with, or `None` for the disabled and token variants.
    #[gen_stub(override_return_type(type_repr = "builtins.str | None"))]
    #[getter]
    fn username(&self) -> Option<String> {
        match &self.inner {
            RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => {
                Some(username.clone())
            }
            _ => None,
        }
    }

    fn __repr__(&self) -> String {
        match &self.inner {
            RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(),
            RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => {
                format!("AutoLogin.username_password({username:?}, ...)")
            }
            RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => {
                "AutoLogin.personal_access_token(...)".to_owned()
            }
        }
    }
}

/// How the TCP client reconnects after the connection to the server is lost.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct TcpReconnectionConfig {
    pub(crate) inner: RustTcpClientReconnectionConfig,
}

#[gen_stub_pymethods]
#[pymethods]
impl TcpReconnectionConfig {
    /// Constructs a reconnection policy.
    ///
    /// Args:
    ///     enabled: Whether to reconnect at all. Defaults to enabled.
    ///     max_retries: Passes over the known endpoints after the first, or
    ///         `None` for unlimited; `0` still makes that first pass. One pass
    ///         tries the endpoint the client is on, the address it was
    ///         configured with, and every node the roster named, so this counts
    ///         passes rather than dials. Defaults
    ///         to unlimited, which means a call awaited while the server is
    ///         down never returns: `connect()`, `send_messages()` and
    ///         `poll_messages()` all wait inside the retry loop. Set a finite
    ///         number for request/reply style usage, so a call fails instead.
    ///     interval: Delay between passes. Defaults to 1 second. The first pass
    ///         runs at once when more than one endpoint is known.
    ///     reestablish_after: Cooldown before redialing the endpoint of the last
    ///         successful connection, measured from when it was established, so
    ///         a session that outlived the interval is redialed at once. Owed to
    ///         that endpoint alone. Defaults to 5 seconds.
    ///
    /// Raises:
    ///     ValueError: If a duration is negative, if `max_retries` is outside the
    ///         range of an unsigned 32-bit integer, or if `interval` is zero.
    ///     OverflowError: If `max_retries` does not fit a signed 64-bit integer,
    ///         raised by the underlying conversion before this constructor runs.
    #[new]
    #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] enabled: Option<bool>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<i64>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        reestablish_after: Option<Py<PyDelta>>,
    ) -> PyResult<Self> {
        let defaults = RustTcpClientReconnectionConfig::default();
        let enabled = enabled.unwrap_or(defaults.enabled);
        let max_retries = max_retries
            .map(|max_retries| u32_param(max_retries, "max_retries"))
            .transpose()?;
        let interval = interval
            .as_ref()
            .map(py_delta_to_iggy_duration)
            .transpose()?
            .map(|interval| reject_zero(interval, "interval"))
            .transpose()?
            .unwrap_or(defaults.interval);
        Ok(Self {
            inner: RustTcpClientReconnectionConfig {
                enabled,
                max_retries,
                interval,
                reestablish_after: reestablish_after
                    .as_ref()
                    .map(py_delta_to_iggy_duration)
                    .transpose()?
                    .unwrap_or(defaults.reestablish_after),
            },
        })
    }

    #[getter]
    fn enabled(&self) -> bool {
        self.inner.enabled
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn max_retries(&self) -> Option<u32> {
        self.inner.max_retries
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.interval.get())
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.reestablish_after)
    }

    fn __repr__(&self) -> String {
        let max_retries = match self.inner.max_retries {
            Some(max_retries) => max_retries.to_string(),
            None => "None".to_owned(),
        };
        format!(
            "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})",
            python_bool(self.inner.enabled),
            duration_repr(self.inner.interval.get()),
            duration_repr(self.inner.reestablish_after),
        )
    }
}

/// Configuration for the TCP transport, accepted by `IggyClient(...)`.
///
/// Every field is keyword-only and optional.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct TcpConfig {
    inner: Arc<RustTcpClientConfig>,
}

impl TcpConfig {
    /// The configuration in the shape `TcpClient::create` expects.
    pub(crate) fn client_config(&self) -> Arc<RustTcpClientConfig> {
        self.inner.clone()
    }
}

#[gen_stub_pymethods]
#[pymethods]
impl TcpConfig {
    /// Constructs a TCP configuration.
    ///
    /// Args:
    ///     server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`.
    ///     auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`.
    ///     reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`.
    ///     heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds.
    ///     tls_enabled: Whether to connect over TLS. Defaults to disabled.
    ///     tls_domain: Domain to validate the certificate against. Empty means it is
    ///         taken from `server_address`.
    ///     tls_ca_file: Path to the CA file for TLS. Read only when `tls_enabled`
    ///         and `tls_validate_certificate` are both on; with either one off it
    ///         is kept but never consulted, so pairing it with
    ///         `tls_validate_certificate=False` pins nothing.
    ///     tls_validate_certificate: Whether to validate the server certificate.
    ///         Defaults to validating. Disabling this accepts any certificate the
    ///         server presents, including self-signed and mismatched ones, and
    ///         takes precedence over `tls_ca_file`; intended for local development
    ///         only.
    ///     nodelay: Disable the Nagle algorithm for the TCP socket. Defaults to
    ///         leaving it on.
    ///
    /// Raises:
    ///     ValueError: If `server_address` is not a valid `host:port` pair, if a
    ///         duration is negative, or if `heartbeat_interval` is zero.
    #[new]
    #[pyo3(signature = (
        *,
        server_address=None,
        auto_login=None,
        reconnection=None,
        heartbeat_interval=None,
        tls_enabled=None,
        tls_domain=None,
        tls_ca_file=None,
        tls_validate_certificate=None,
        nodelay=None,
    ))]
    #[allow(clippy::too_many_arguments)]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option<
            String,
        >,
        #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option<AutoLogin>,
        #[gen_stub(override_type(type_repr = "TcpReconnectionConfig | None"))] reconnection: Option<
            TcpReconnectionConfig,
        >,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        heartbeat_interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option<bool>,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option<String>,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option<String>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
        tls_validate_certificate: Option<bool>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] nodelay: Option<bool>,
    ) -> PyResult<Self> {
        // The builder starts from `TcpClientConfig::default()`, and its `build()`
        // trims and validates the address whether or not one was set here.
        let mut builder = TcpClientConfigBuilder::new();
        if let Some(server_address) = server_address {
            builder = builder.with_server_address(server_address);
        }
        let mut inner = builder
            .build()
            .map_err(|e| invalid_address("server_address", e))?;
        if let Some(auto_login) = auto_login {
            inner.auto_login = auto_login.inner;
        }
        if let Some(reconnection) = reconnection {
            inner.reconnection = reconnection.inner;
        }
        if let Some(heartbeat_interval) = heartbeat_interval {
            inner.heartbeat_interval = reject_zero(
                py_delta_to_iggy_duration(&heartbeat_interval)?,
                "heartbeat_interval",
            )?;
        }
        if let Some(tls_enabled) = tls_enabled {
            inner.tls_enabled = tls_enabled;
        }
        if let Some(tls_domain) = tls_domain {
            inner.tls_domain = tls_domain;
        }
        if tls_ca_file.is_some() {
            inner.tls_ca_file = tls_ca_file;
        }
        if let Some(tls_validate_certificate) = tls_validate_certificate {
            inner.tls_validate_certificate = tls_validate_certificate;
        }
        if let Some(nodelay) = nodelay {
            inner.nodelay = nodelay;
        }

        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    #[getter]
    fn server_address(&self) -> String {
        self.inner.server_address.clone()
    }

    #[getter]
    fn auto_login(&self) -> AutoLogin {
        AutoLogin {
            inner: self.inner.auto_login.clone(),
        }
    }

    #[getter]
    fn reconnection(&self) -> TcpReconnectionConfig {
        TcpReconnectionConfig {
            inner: self.inner.reconnection.clone(),
        }
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get())
    }

    #[getter]
    fn tls_enabled(&self) -> bool {
        self.inner.tls_enabled
    }

    #[getter]
    fn tls_domain(&self) -> String {
        self.inner.tls_domain.clone()
    }

    #[gen_stub(override_return_type(type_repr = "builtins.str | None"))]
    #[getter]
    fn tls_ca_file(&self) -> Option<String> {
        self.inner.tls_ca_file.clone()
    }

    #[getter]
    fn tls_validate_certificate(&self) -> bool {
        self.inner.tls_validate_certificate
    }

    #[getter]
    fn nodelay(&self) -> bool {
        self.inner.nodelay
    }

    fn __repr__(&self) -> String {
        let tls_ca_file = match &self.inner.tls_ca_file {
            Some(tls_ca_file) => format!("{tls_ca_file:?}"),
            None => "None".to_owned(),
        };
        format!(
            "TcpConfig(server_address={:?}, auto_login={}, reconnection={}, heartbeat_interval={}, tls_enabled={}, tls_domain={:?}, tls_ca_file={tls_ca_file}, tls_validate_certificate={}, nodelay={})",
            self.inner.server_address,
            self.auto_login().__repr__(),
            self.reconnection().__repr__(),
            duration_repr(self.inner.heartbeat_interval.get()),
            python_bool(self.inner.tls_enabled),
            self.inner.tls_domain,
            python_bool(self.inner.tls_validate_certificate),
            python_bool(self.inner.nodelay),
        )
    }
}

/// How the QUIC client reconnects after the connection to the server is lost.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct QuicReconnectionConfig {
    pub(crate) inner: RustQuicClientReconnectionConfig,
}

#[gen_stub_pymethods]
#[pymethods]
impl QuicReconnectionConfig {
    /// Constructs a reconnection policy.
    ///
    /// Args:
    ///     enabled: Whether to reconnect at all. Defaults to enabled.
    ///     max_retries: Redials of the configured server address after the first
    ///         attempt, or `None` for unlimited; `0` still makes that first
    ///         attempt. Unlike the TCP transport, QUIC redials the one address
    ///         it was configured with rather than walking a cluster roster, so
    ///         this counts dials. Defaults to unlimited, which means a call
    ///         awaited while the server is down never returns: `connect()`
    ///         waits inside the retry loop, as do `send_messages()` and
    ///         `poll_messages()` once auto-login is configured. Set a finite
    ///         number for request/reply style usage, so a call fails instead.
    ///     interval: Delay before each redial. Defaults to 1 second.
    ///     reestablish_after: Cooldown before redialing after a previously
    ///         successful connection, measured from when it was established, so
    ///         a session that outlived the interval is redialed at once.
    ///         Defaults to 5 seconds.
    ///
    /// Raises:
    ///     ValueError: If a duration is negative, if `max_retries` is outside the
    ///         range of an unsigned 32-bit integer, or if `interval` is zero.
    ///     OverflowError: If `max_retries` does not fit a signed 64-bit integer,
    ///         raised by the underlying conversion before this constructor runs.
    #[new]
    #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] enabled: Option<bool>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<i64>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        reestablish_after: Option<Py<PyDelta>>,
    ) -> PyResult<Self> {
        let defaults = RustQuicClientReconnectionConfig::default();
        let enabled = enabled.unwrap_or(defaults.enabled);
        let max_retries = max_retries
            .map(|max_retries| u32_param(max_retries, "max_retries"))
            .transpose()?;
        let interval = interval
            .as_ref()
            .map(py_delta_to_iggy_duration)
            .transpose()?
            .map(|interval| reject_zero(interval, "interval"))
            .transpose()?
            .unwrap_or(defaults.interval);
        Ok(Self {
            inner: RustQuicClientReconnectionConfig {
                enabled,
                max_retries,
                interval,
                reestablish_after: reestablish_after
                    .as_ref()
                    .map(py_delta_to_iggy_duration)
                    .transpose()?
                    .unwrap_or(defaults.reestablish_after),
            },
        })
    }

    #[getter]
    fn enabled(&self) -> bool {
        self.inner.enabled
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn max_retries(&self) -> Option<u32> {
        self.inner.max_retries
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.interval.get())
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.reestablish_after)
    }

    fn __repr__(&self) -> String {
        let max_retries = match self.inner.max_retries {
            Some(max_retries) => max_retries.to_string(),
            None => "None".to_owned(),
        };
        format!(
            "QuicReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})",
            python_bool(self.inner.enabled),
            duration_repr(self.inner.interval.get()),
            duration_repr(self.inner.reestablish_after),
        )
    }
}

/// quinn clamps `TransportConfig::initial_mtu` up to this floor rather than
/// rejecting a smaller value, so `QuicConfig` rejects it instead: otherwise the
/// getter would read back a value that is not the one actually in effect.
const QUINN_MIN_INITIAL_MTU: u16 = 1200;

/// Configuration for the QUIC transport, accepted by `IggyClient(...)`.
///
/// Every field is keyword-only and optional.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct QuicConfig {
    inner: Arc<RustQuicClientConfig>,
}

impl QuicConfig {
    /// The configuration in the shape `QuicClient::create` expects.
    pub(crate) fn client_config(&self) -> Arc<RustQuicClientConfig> {
        self.inner.clone()
    }
}

#[gen_stub_pymethods]
#[pymethods]
impl QuicConfig {
    /// Constructs a QUIC configuration.
    ///
    /// Args:
    ///     server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8080`.
    ///     client_address: `host:port` to bind the local UDP socket to. Defaults to
    ///         `127.0.0.1:0`, which binds to any available port. That exact value,
    ///         passed or defaulted, binds `[::1]:0` instead when `server_address`
    ///         resolves to IPv6, so the socket in use may not be the address read
    ///         back here. Any other value binds as given.
    ///     server_name: Server name used for the QUIC/TLS handshake. Defaults to
    ///         `localhost`.
    ///     auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`.
    ///     reconnection: Reconnection policy. Defaults to `QuicReconnectionConfig()`.
    ///     heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds.
    ///     response_buffer_size: Size of the response buffer in bytes. Defaults to 10 MB.
    ///     max_concurrent_bidi_streams: Maximum number of concurrent bidirectional
    ///         streams. Defaults to 10,000.
    ///     datagram_send_buffer_size: Size of the datagram send buffer in bytes.
    ///         Defaults to 100,000.
    ///     initial_mtu: Initial MTU in bytes. Defaults to 1200.
    ///     send_window: Send window size in bytes. Defaults to 100,000.
    ///     receive_window: Receive window size in bytes. Defaults to 100,000.
    ///     keep_alive_interval: Interval between QUIC keep-alive pings, or a zero
    ///         duration to disable them. Defaults to 5 seconds.
    ///     max_idle_timeout: How long the connection tolerates silence before it is
    ///         considered dead, or a zero duration to use quinn's own default (30
    ///         seconds) instead, since `configure()` skips the setter entirely when
    ///         zero. Defaults to 10 seconds.
    ///     validate_certificate: Whether to validate the server certificate. Defaults
    ///         to disabled; only the TCP transport validates by default.
    ///
    /// Raises:
    ///     ValueError: If `server_address` or `client_address` is not a valid
    ///         `host:port` pair, if a duration is negative, if
    ///         `heartbeat_interval` is zero, if `keep_alive_interval` or
    ///         `max_idle_timeout` is not a whole number of milliseconds, if
    ///         `initial_mtu` is below quinn's minimum of 1200, or if a numeric
    ///         field is outside the range of its underlying wire type.
    ///     OverflowError: If a numeric field does not fit a signed 64-bit integer,
    ///         raised by the underlying conversion before this constructor runs.
    #[new]
    #[pyo3(signature = (
        *,
        server_address=None,
        client_address=None,
        server_name=None,
        auto_login=None,
        reconnection=None,
        heartbeat_interval=None,
        response_buffer_size=None,
        max_concurrent_bidi_streams=None,
        datagram_send_buffer_size=None,
        initial_mtu=None,
        send_window=None,
        receive_window=None,
        keep_alive_interval=None,
        max_idle_timeout=None,
        validate_certificate=None,
    ))]
    #[allow(clippy::too_many_arguments)]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option<
            String,
        >,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] client_address: Option<
            String,
        >,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_name: Option<String>,
        #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option<AutoLogin>,
        #[gen_stub(override_type(type_repr = "QuicReconnectionConfig | None"))]
        reconnection: Option<QuicReconnectionConfig>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        heartbeat_interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] response_buffer_size: Option<
            i64,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))]
        max_concurrent_bidi_streams: Option<i64>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))]
        datagram_send_buffer_size: Option<i64>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] initial_mtu: Option<i64>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] send_window: Option<i64>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] receive_window: Option<i64>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        keep_alive_interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        max_idle_timeout: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] validate_certificate: Option<
            bool,
        >,
    ) -> PyResult<Self> {
        // The builder starts from `QuicClientConfig::default()`, and its `build()`
        // trims and validates the server address whether or not one was set here.
        let mut builder = QuicClientConfigBuilder::new();
        if let Some(server_address) = server_address {
            builder = builder.with_server_address(server_address);
        }
        let mut inner = builder
            .build()
            .map_err(|e| invalid_address("server_address", e))?;
        if let Some(client_address) = client_address {
            // Trimmed like the server address, but otherwise kept verbatim rather
            // than re-serialized from the parsed `SocketAddr`: `QuicClient::create`
            // compares this against the literal default to decide whether to bind
            // an IPv6 socket for an IPv6 server, and a rewritten string would not
            // match.
            let client_address = client_address.trim();
            client_address
                .parse::<SocketAddr>()
                .map_err(|e| invalid_address("client_address", e))?;
            inner.client_address = client_address.to_owned();
        }
        if let Some(server_name) = server_name {
            inner.server_name = server_name;
        }
        if let Some(auto_login) = auto_login {
            inner.auto_login = auto_login.inner;
        }
        if let Some(reconnection) = reconnection {
            inner.reconnection = reconnection.inner;
        }
        if let Some(heartbeat_interval) = heartbeat_interval {
            inner.heartbeat_interval = reject_zero(
                py_delta_to_iggy_duration(&heartbeat_interval)?,
                "heartbeat_interval",
            )?;
        }
        if let Some(response_buffer_size) = response_buffer_size {
            inner.response_buffer_size = u64_param(response_buffer_size, "response_buffer_size")?;
        }
        if let Some(max_concurrent_bidi_streams) = max_concurrent_bidi_streams {
            inner.max_concurrent_bidi_streams =
                varint_param(max_concurrent_bidi_streams, "max_concurrent_bidi_streams")?;
        }
        if let Some(datagram_send_buffer_size) = datagram_send_buffer_size {
            inner.datagram_send_buffer_size =
                u64_param(datagram_send_buffer_size, "datagram_send_buffer_size")?;
        }
        if let Some(initial_mtu) = initial_mtu {
            let initial_mtu = u16_param(initial_mtu, "initial_mtu")?;
            if initial_mtu < QUINN_MIN_INITIAL_MTU {
                return Err(PyValueError::new_err(format!(
                    "'initial_mtu' must be at least {QUINN_MIN_INITIAL_MTU}; quinn silently \
                     raises anything smaller to that floor, so the getter would no longer \
                     match the value actually in effect"
                )));
            }
            inner.initial_mtu = initial_mtu;
        }
        if let Some(send_window) = send_window {
            inner.send_window = u64_param(send_window, "send_window")?;
        }
        if let Some(receive_window) = receive_window {
            inner.receive_window = varint_param(receive_window, "receive_window")?;
        }
        if let Some(keep_alive_interval) = keep_alive_interval {
            inner.keep_alive_interval =
                py_delta_to_millis(&keep_alive_interval, "keep_alive_interval")?;
        }
        if let Some(max_idle_timeout) = max_idle_timeout {
            inner.max_idle_timeout = py_delta_to_millis(&max_idle_timeout, "max_idle_timeout")?;
        }
        if let Some(validate_certificate) = validate_certificate {
            inner.validate_certificate = validate_certificate;
        }

        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    #[getter]
    fn server_address(&self) -> String {
        self.inner.server_address.clone()
    }

    #[getter]
    fn client_address(&self) -> String {
        self.inner.client_address.clone()
    }

    #[getter]
    fn server_name(&self) -> String {
        self.inner.server_name.clone()
    }

    #[getter]
    fn auto_login(&self) -> AutoLogin {
        AutoLogin {
            inner: self.inner.auto_login.clone(),
        }
    }

    #[getter]
    fn reconnection(&self) -> QuicReconnectionConfig {
        QuicReconnectionConfig {
            inner: self.inner.reconnection.clone(),
        }
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get())
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int"))]
    #[getter]
    fn response_buffer_size(&self) -> u64 {
        self.inner.response_buffer_size
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int"))]
    #[getter]
    fn max_concurrent_bidi_streams(&self) -> u64 {
        self.inner.max_concurrent_bidi_streams
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int"))]
    #[getter]
    fn datagram_send_buffer_size(&self) -> u64 {
        self.inner.datagram_send_buffer_size
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int"))]
    #[getter]
    fn initial_mtu(&self) -> u16 {
        self.inner.initial_mtu
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int"))]
    #[getter]
    fn send_window(&self) -> u64 {
        self.inner.send_window
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int"))]
    #[getter]
    fn receive_window(&self) -> u64 {
        self.inner.receive_window
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn keep_alive_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        millis_to_py_delta(py, self.inner.keep_alive_interval)
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn max_idle_timeout<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        millis_to_py_delta(py, self.inner.max_idle_timeout)
    }

    #[getter]
    fn validate_certificate(&self) -> bool {
        self.inner.validate_certificate
    }

    fn __repr__(&self) -> String {
        format!(
            "QuicConfig(server_address={:?}, client_address={:?}, server_name={:?}, auto_login={}, reconnection={}, heartbeat_interval={}, response_buffer_size={}, max_concurrent_bidi_streams={}, datagram_send_buffer_size={}, initial_mtu={}, send_window={}, receive_window={}, keep_alive_interval={}, max_idle_timeout={}, validate_certificate={})",
            self.inner.server_address,
            self.inner.client_address,
            self.inner.server_name,
            self.auto_login().__repr__(),
            self.reconnection().__repr__(),
            duration_repr(self.inner.heartbeat_interval.get()),
            self.inner.response_buffer_size,
            self.inner.max_concurrent_bidi_streams,
            self.inner.datagram_send_buffer_size,
            self.inner.initial_mtu,
            self.inner.send_window,
            self.inner.receive_window,
            millis_repr(self.inner.keep_alive_interval),
            millis_repr(self.inner.max_idle_timeout),
            python_bool(self.inner.validate_certificate),
        )
    }
}

/// Configuration for the HTTP transport, accepted by `IggyClient(...)`.
///
/// Every field is keyword-only and optional.
///
/// There is no `AutoLogin` and no reconnection policy, and `connect()` does not
/// dial: it only starts the heartbeat, so `login_user(...)` has to follow it.
///
/// HTTP is single-consumer only. `consumer_group(...)` fails with
/// `Feature is unavailable`, and so does a `Consumer.Group(...)` poll unless it
/// names an explicit `partition_id`. With one, the consumer kind is not carried
/// on the HTTP wire, so the poll is served as an ordinary consumer named after
/// the group, with no membership, no partition assignment, and no rebalancing
/// behind it. Pass `Consumer.Single(...)` explicitly.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct HttpConfig {
    inner: Arc<RustHttpClientConfig>,
}

impl HttpConfig {
    /// The configuration in the shape `HttpClient::create` expects.
    pub(crate) fn client_config(&self) -> Arc<RustHttpClientConfig> {
        self.inner.clone()
    }
}

#[gen_stub_pymethods]
#[pymethods]
impl HttpConfig {
    /// Constructs an HTTP configuration.
    ///
    /// Args:
    ///     api_url: Base URL of the Iggy HTTP API, as `scheme://host[:port]`
    ///         only - no path, query, fragment, or credentials. Defaults to
    ///         `http://127.0.0.1:3000`.
    ///     retries: Number of retries to perform on transient errors, each one
    ///         replaying the full request (including its body) via automatic
    ///         middleware. Defaults to 3. Delivery is therefore at-least-once:
    ///         if the original request actually committed but its response
    ///         was lost (e.g. to a timeout), a retried call applies the same
    ///         operation again. Set to 0 to disable automatic replay and match
    ///         the other transports, which surface the failure instead of
    ///         silently resending.
    ///     jwt: JWT token for A2A (Agent-to-Agent) authentication. Defaults to
    ///         `None`. Stored trimmed, since a token read from a file carries a
    ///         trailing newline that the `Authorization` header value rejects.
    ///         Rejected if empty or whitespace-only: accepting it would make
    ///         `has_jwt` report `True` while every call still fails
    ///         `Unauthenticated`.
    ///     heartbeat_interval: Interval between the client's liveness probes
    ///         (a bare `GET /ping`). Defaults to 5 seconds. Unlike TCP/QUIC,
    ///         HTTP has no persistent connection or session for this to keep
    ///         alive; it only proves the server is reachable.
    ///
    /// Raises:
    ///     ValueError: If `api_url` is not a valid URL, if `retries` is outside
    ///         the range of an unsigned 32-bit integer, if `jwt` is empty or
    ///         whitespace-only, if a duration is negative, or if
    ///         `heartbeat_interval` is zero.
    ///     OverflowError: If `retries` does not fit a signed 64-bit integer,
    ///         raised by the underlying conversion before this constructor runs.
    #[new]
    #[pyo3(signature = (*, api_url=None, retries=None, jwt=None, heartbeat_interval=None))]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] api_url: Option<String>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] retries: Option<i64>,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] jwt: Option<String>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        heartbeat_interval: Option<Py<PyDelta>>,
    ) -> PyResult<Self> {
        // The builder starts from `HttpClientConfig::default()`, and its `build()`
        // trims and validates the API URL whether or not one was set here.
        let mut builder = HttpClientConfigBuilder::new();
        if let Some(api_url) = api_url {
            builder = builder.with_api_url(api_url);
        }
        let mut inner = builder
            .build()
            .map_err(|e| PyValueError::new_err(format!("'api_url' is not a valid URL: {e}")))?;
        if let Some(retries) = retries {
            inner.retries = u32_param(retries, "retries")?;
        }
        if let Some(jwt) = jwt {
            let jwt = jwt.trim();
            if jwt.is_empty() {
                return Err(PyValueError::new_err(
                    "'jwt' must not be empty or whitespace-only",
                ));
            }
            inner.jwt = Some(jwt.to_owned());
        }
        if let Some(heartbeat_interval) = heartbeat_interval {
            inner.heartbeat_interval = reject_zero(
                py_delta_to_iggy_duration(&heartbeat_interval)?,
                "heartbeat_interval",
            )?;
        }

        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    #[getter]
    fn api_url(&self) -> String {
        self.inner.api_url.clone()
    }

    #[getter]
    fn retries(&self) -> u32 {
        self.inner.retries
    }

    /// Whether a JWT is configured, without exposing the token itself.
    #[getter]
    fn has_jwt(&self) -> bool {
        self.inner.jwt.is_some()
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get())
    }

    fn __repr__(&self) -> String {
        let jwt = if self.inner.jwt.is_some() {
            "..."
        } else {
            "None"
        };
        format!(
            "HttpConfig(api_url={:?}, retries={}, jwt={jwt}, heartbeat_interval={})",
            self.inner.api_url,
            self.inner.retries,
            duration_repr(self.inner.heartbeat_interval.get()),
        )
    }
}

/// How the WebSocket client reconnects after the connection to the server is lost.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct WebSocketReconnectionConfig {
    pub(crate) inner: RustWebSocketClientReconnectionConfig,
}

#[gen_stub_pymethods]
#[pymethods]
impl WebSocketReconnectionConfig {
    /// Constructs a reconnection policy.
    ///
    /// Args:
    ///     enabled: Whether to reconnect at all. Defaults to enabled.
    ///     max_retries: Redials of the configured server address after the first
    ///         attempt, or `None` for unlimited; `0` still makes that first
    ///         attempt. Unlike the TCP transport, WebSocket redials the one
    ///         address it was configured with rather than walking a cluster
    ///         roster, so this counts dials. Defaults to unlimited, which means
    ///         a call awaited while the server is down never returns:
    ///         `connect()` waits inside the retry loop, as do `send_messages()`
    ///         and `poll_messages()` once auto-login is configured. Set a finite
    ///         number for request/reply style usage, so a call fails instead.
    ///     interval: Delay before each redial. Defaults to 1 second.
    ///     reestablish_after: Cooldown before redialing after a previously
    ///         successful connection, measured from when it was established, so
    ///         a session that outlived the interval is redialed at once. Applied
    ///         from the first redial onward, not to the initial connect.
    ///         Defaults to 5 seconds.
    ///
    /// Raises:
    ///     ValueError: If a duration is negative, if `max_retries` is outside the
    ///         range of an unsigned 32-bit integer, or if `interval` is zero.
    ///     OverflowError: If `max_retries` does not fit a signed 64-bit integer,
    ///         raised by the underlying conversion before this constructor runs.
    #[new]
    #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] enabled: Option<bool>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<i64>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        reestablish_after: Option<Py<PyDelta>>,
    ) -> PyResult<Self> {
        let defaults = RustWebSocketClientReconnectionConfig::default();
        let enabled = enabled.unwrap_or(defaults.enabled);
        let max_retries = max_retries
            .map(|max_retries| u32_param(max_retries, "max_retries"))
            .transpose()?;
        let interval = interval
            .as_ref()
            .map(py_delta_to_iggy_duration)
            .transpose()?
            .map(|interval| reject_zero(interval, "interval"))
            .transpose()?
            .unwrap_or(defaults.interval);
        Ok(Self {
            inner: RustWebSocketClientReconnectionConfig {
                enabled,
                max_retries,
                interval,
                reestablish_after: reestablish_after
                    .as_ref()
                    .map(py_delta_to_iggy_duration)
                    .transpose()?
                    .unwrap_or(defaults.reestablish_after),
            },
        })
    }

    #[getter]
    fn enabled(&self) -> bool {
        self.inner.enabled
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn max_retries(&self) -> Option<u32> {
        self.inner.max_retries
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.interval.get())
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.reestablish_after)
    }

    fn __repr__(&self) -> String {
        let max_retries = match self.inner.max_retries {
            Some(max_retries) => max_retries.to_string(),
            None => "None".to_owned(),
        };
        format!(
            "WebSocketReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})",
            python_bool(self.inner.enabled),
            duration_repr(self.inner.interval.get()),
            duration_repr(self.inner.reestablish_after),
        )
    }
}

/// Frame- and buffer-level options passed through to the underlying WebSocket
/// implementation, accepted by `WebSocketConfig`'s `framing` argument.
///
/// Every field is keyword-only and optional; unset fields fall back to the
/// underlying WebSocket library's own defaults.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct WebSocketFramingConfig {
    pub(crate) inner: RustWebSocketFramingConfig,
}

#[gen_stub_pymethods]
#[pymethods]
impl WebSocketFramingConfig {
    /// Constructs a WebSocket framing configuration.
    ///
    /// Args:
    ///     read_buffer_size: Read buffer size in bytes. Defaults to 128 KiB.
    ///     write_buffer_size: Write buffer size in bytes. Defaults to 128 KiB.
    ///     max_write_buffer_size: Maximum write buffer size in bytes. Defaults to
    ///         unbounded, which reads back as the largest value a pointer-sized
    ///         unsigned integer holds rather than as `None`.
    ///     max_message_size: Maximum message size in bytes, or an explicit `None`
    ///         to lift the limit entirely. Omitting the argument is not the same
    ///         as passing `None`: it keeps the underlying default of 64 MiB.
    ///         Lifting the limit lets a peer queue an arbitrarily large message
    ///         in memory, so prefer a finite value.
    ///     max_frame_size: Maximum frame size in bytes, or an explicit `None` to
    ///         lift the limit entirely. Omitting the argument keeps the
    ///         underlying default of 16 MiB, with the same caveat as
    ///         `max_message_size`.
    ///     accept_unmasked_frames: Whether to accept unmasked frames. Defaults to
    ///         `False`; clients should typically keep this off for RFC compliance.
    ///
    /// Raises:
    ///     ValueError: If a numeric field is outside the range of a pointer-sized
    ///         unsigned integer, or if `max_write_buffer_size` does not come out
    ///         greater than `write_buffer_size`. tungstenite enforces the same
    ///         invariant with an `assert!` at connect time, which would otherwise
    ///         surface as an unrecoverable Rust panic instead of a `ValueError`.
    ///     OverflowError: If a numeric field does not fit a signed 128-bit integer,
    ///         raised by the underlying conversion before this constructor runs.
    #[new]
    #[pyo3(signature = (
        *,
        read_buffer_size=None,
        write_buffer_size=None,
        max_write_buffer_size=None,
        max_message_size=64 << 20,
        max_frame_size=16 << 20,
        accept_unmasked_frames=None,
    ))]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] read_buffer_size: Option<
            i128,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] write_buffer_size: Option<
            i128,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_write_buffer_size: Option<
            i128,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_message_size: Option<
            i128,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_frame_size: Option<i128>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
        accept_unmasked_frames: Option<bool>,
    ) -> PyResult<Self> {
        let mut inner = RustWebSocketFramingConfig::default();
        if let Some(read_buffer_size) = read_buffer_size {
            inner.read_buffer_size = Some(usize_param(read_buffer_size, "read_buffer_size")?);
        }
        if let Some(write_buffer_size) = write_buffer_size {
            inner.write_buffer_size = Some(usize_param(write_buffer_size, "write_buffer_size")?);
        }
        if let Some(max_write_buffer_size) = max_write_buffer_size {
            inner.max_write_buffer_size =
                Some(usize_param(max_write_buffer_size, "max_write_buffer_size")?);
        }
        // Assigned unconditionally, unlike the buffer sizes above: `None` here
        // means "no limit", and pyo3 cannot tell an omitted argument from an
        // explicit `None` on its own. The signature defaults carry the
        // underlying limits instead, so omission lands on `Some(default)` and
        // only an explicit `None` reaches this as `None`.
        inner.max_message_size = max_message_size
            .map(|max_message_size| usize_param(max_message_size, "max_message_size"))
            .transpose()?;
        inner.max_frame_size = max_frame_size
            .map(|max_frame_size| usize_param(max_frame_size, "max_frame_size"))
            .transpose()?;
        if let Some(accept_unmasked_frames) = accept_unmasked_frames {
            inner.accept_unmasked_frames = accept_unmasked_frames;
        }
        if let (Some(write_buffer_size), Some(max_write_buffer_size)) =
            (inner.write_buffer_size, inner.max_write_buffer_size)
            && max_write_buffer_size <= write_buffer_size
        {
            return Err(PyValueError::new_err(format!(
                "'max_write_buffer_size' ({max_write_buffer_size}) must be greater than \
                 'write_buffer_size' ({write_buffer_size})"
            )));
        }

        Ok(Self { inner })
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn read_buffer_size(&self) -> Option<usize> {
        self.inner.read_buffer_size
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn write_buffer_size(&self) -> Option<usize> {
        self.inner.write_buffer_size
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn max_write_buffer_size(&self) -> Option<usize> {
        self.inner.max_write_buffer_size
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn max_message_size(&self) -> Option<usize> {
        self.inner.max_message_size
    }

    #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
    #[getter]
    fn max_frame_size(&self) -> Option<usize> {
        self.inner.max_frame_size
    }

    #[getter]
    fn accept_unmasked_frames(&self) -> bool {
        self.inner.accept_unmasked_frames
    }

    fn __repr__(&self) -> String {
        let optional_usize = |value: Option<usize>| match value {
            Some(value) => value.to_string(),
            None => "None".to_owned(),
        };
        format!(
            "WebSocketFramingConfig(read_buffer_size={}, write_buffer_size={}, max_write_buffer_size={}, max_message_size={}, max_frame_size={}, accept_unmasked_frames={})",
            optional_usize(self.inner.read_buffer_size),
            optional_usize(self.inner.write_buffer_size),
            optional_usize(self.inner.max_write_buffer_size),
            optional_usize(self.inner.max_message_size),
            optional_usize(self.inner.max_frame_size),
            python_bool(self.inner.accept_unmasked_frames),
        )
    }
}

/// Configuration for the WebSocket transport, accepted by `IggyClient(...)`.
///
/// Every field is keyword-only and optional.
#[gen_stub_pyclass]
#[pyclass(from_py_object)]
#[derive(Clone)]
pub struct WebSocketConfig {
    inner: Arc<RustWebSocketClientConfig>,
}

impl WebSocketConfig {
    /// The configuration in the shape `WebSocketClient::create` expects.
    pub(crate) fn client_config(&self) -> Arc<RustWebSocketClientConfig> {
        self.inner.clone()
    }
}

#[gen_stub_pymethods]
#[pymethods]
impl WebSocketConfig {
    /// Constructs a WebSocket configuration.
    ///
    /// Args:
    ///     server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8092`.
    ///     auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`.
    ///     reconnection: Reconnection policy. Defaults to `WebSocketReconnectionConfig()`.
    ///     heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds.
    ///     framing: Frame- and buffer-level options. Defaults to `WebSocketFramingConfig()`.
    ///     tls_enabled: Whether to connect over TLS. Defaults to disabled.
    ///     tls_domain: Domain to validate the certificate against. Defaults to
    ///         `localhost`. Empty means it is taken from the IP `server_address`
    ///         resolves to.
    ///     tls_ca_file: Path to the CA file for TLS. Read only when `tls_enabled`
    ///         and `tls_validate_certificate` are both on; with either one off it
    ///         is kept but never consulted, so pairing it with
    ///         `tls_validate_certificate=False` pins nothing.
    ///     tls_validate_certificate: Whether to validate the server certificate.
    ///         Defaults to `False`; only the TCP transport validates by default.
    ///         Disabling this accepts any certificate the server presents,
    ///         including self-signed and mismatched ones, and takes precedence
    ///         over `tls_ca_file`.
    ///
    /// Raises:
    ///     ValueError: If `server_address` is not a valid `host:port` pair, if a
    ///         duration is negative, or if `heartbeat_interval` is zero.
    #[new]
    #[pyo3(signature = (
        *,
        server_address=None,
        auto_login=None,
        reconnection=None,
        heartbeat_interval=None,
        framing=None,
        tls_enabled=None,
        tls_domain=None,
        tls_ca_file=None,
        tls_validate_certificate=None,
    ))]
    #[allow(clippy::too_many_arguments)]
    fn new(
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option<
            String,
        >,
        #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option<AutoLogin>,
        #[gen_stub(override_type(type_repr = "WebSocketReconnectionConfig | None"))]
        reconnection: Option<WebSocketReconnectionConfig>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        heartbeat_interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "WebSocketFramingConfig | None"))] framing: Option<
            WebSocketFramingConfig,
        >,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option<bool>,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option<String>,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option<String>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
        tls_validate_certificate: Option<bool>,
    ) -> PyResult<Self> {
        // The builder starts from `WebSocketClientConfig::default()`, and its
        // `build()` trims and validates the address whether or not one was set here.
        let mut builder = WebSocketClientConfigBuilder::new();
        if let Some(server_address) = server_address {
            builder = builder.with_server_address(server_address);
        }
        let mut inner = builder
            .build()
            .map_err(|e| invalid_address("server_address", e))?;
        if let Some(auto_login) = auto_login {
            inner.auto_login = auto_login.inner;
        }
        if let Some(reconnection) = reconnection {
            inner.reconnection = reconnection.inner;
        }
        if let Some(heartbeat_interval) = heartbeat_interval {
            inner.heartbeat_interval = reject_zero(
                py_delta_to_iggy_duration(&heartbeat_interval)?,
                "heartbeat_interval",
            )?;
        }
        if let Some(framing) = framing {
            inner.ws_config = framing.inner;
        }
        if let Some(tls_enabled) = tls_enabled {
            inner.tls_enabled = tls_enabled;
        }
        if let Some(tls_domain) = tls_domain {
            inner.tls_domain = tls_domain;
        }
        if tls_ca_file.is_some() {
            inner.tls_ca_file = tls_ca_file;
        }
        if let Some(tls_validate_certificate) = tls_validate_certificate {
            inner.tls_validate_certificate = tls_validate_certificate;
        }

        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    #[getter]
    fn server_address(&self) -> String {
        self.inner.server_address.clone()
    }

    #[getter]
    fn auto_login(&self) -> AutoLogin {
        AutoLogin {
            inner: self.inner.auto_login.clone(),
        }
    }

    #[getter]
    fn reconnection(&self) -> WebSocketReconnectionConfig {
        WebSocketReconnectionConfig {
            inner: self.inner.reconnection.clone(),
        }
    }

    #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
    #[getter]
    fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
        iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get())
    }

    #[getter]
    fn framing(&self) -> WebSocketFramingConfig {
        WebSocketFramingConfig {
            inner: self.inner.ws_config.clone(),
        }
    }

    #[getter]
    fn tls_enabled(&self) -> bool {
        self.inner.tls_enabled
    }

    #[getter]
    fn tls_domain(&self) -> String {
        self.inner.tls_domain.clone()
    }

    #[gen_stub(override_return_type(type_repr = "builtins.str | None"))]
    #[getter]
    fn tls_ca_file(&self) -> Option<String> {
        self.inner.tls_ca_file.clone()
    }

    #[getter]
    fn tls_validate_certificate(&self) -> bool {
        self.inner.tls_validate_certificate
    }

    fn __repr__(&self) -> String {
        let tls_ca_file = match &self.inner.tls_ca_file {
            Some(tls_ca_file) => format!("{tls_ca_file:?}"),
            None => "None".to_owned(),
        };
        format!(
            "WebSocketConfig(server_address={:?}, auto_login={}, reconnection={}, heartbeat_interval={}, framing={}, tls_enabled={}, tls_domain={:?}, tls_ca_file={tls_ca_file}, tls_validate_certificate={})",
            self.inner.server_address,
            self.auto_login().__repr__(),
            self.reconnection().__repr__(),
            duration_repr(self.inner.heartbeat_interval.get()),
            self.framing().__repr__(),
            python_bool(self.inner.tls_enabled),
            self.inner.tls_domain,
            python_bool(self.inner.tls_validate_certificate),
        )
    }
}

fn python_bool(value: bool) -> &'static str {
    if value { "True" } else { "False" }
}

/// Rejects an address that is not a valid `host:port`, naming the argument it
/// came from: neither the builder's error nor `SocketAddr`'s mentions which one.
fn invalid_address(parameter: &str, error: impl Display) -> PyErr {
    PyValueError::new_err(format!("'{parameter}' is not a valid 'host:port': {error}"))
}

/// Converts a Python int to the unsigned 32-bit integer `max_retries`/`retries`
/// expect, naming the parameter in the error so a caller can tell which
/// argument was out of range. A value too large even for `i64` still raises
/// pyo3's own unnamed `OverflowError` before this ever runs.
fn u32_param(value: i64, parameter: &str) -> PyResult<u32> {
    u32::try_from(value).map_err(|_| {
        PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u32::MAX))
    })
}

/// Converts a Python int to the unsigned 64-bit integer a QUIC transport
/// field expects, naming the parameter in the error so a caller can tell
/// which argument was out of range. The bound in the message is `i64::MAX`
/// rather than `u64::MAX` because pyo3 extracts the argument as an `i64`
/// first: anything above that never reaches here, raising `OverflowError`
/// on the way in. Every one of these fields is a buffer or window size, so
/// the unreachable half of the range has no practical use.
fn u64_param(value: i64, parameter: &str) -> PyResult<u64> {
    u64::try_from(value).map_err(|_| {
        PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", i64::MAX))
    })
}

/// Converts a Python int to the unsigned 16-bit integer `initial_mtu` expects.
fn u16_param(value: i64, parameter: &str) -> PyResult<u16> {
    u16::try_from(value).map_err(|_| {
        PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u16::MAX))
    })
}

/// Converts a Python int to a `u64` that also fits `quinn::VarInt` (max
/// `2^62 - 1`), which `max_concurrent_bidi_streams` and `receive_window` are
/// narrowed into when the connection is configured. A `u64` in range for
/// `u64::MAX` but not `VarInt::MAX` would otherwise only fail there, as an
/// opaque `RuntimeError` instead of a `ValueError` naming the argument.
fn varint_param(value: i64, parameter: &str) -> PyResult<u64> {
    const VARINT_MAX: u64 = (1u64 << 62) - 1;
    let value = u64_param(value, parameter)?;
    if value > VARINT_MAX {
        return Err(PyValueError::new_err(format!(
            "'{parameter}' must be between 0 and {VARINT_MAX}"
        )));
    }
    Ok(value)
}

/// Converts a Python int to the unsigned pointer-sized integer a WebSocket
/// framing field expects, naming the parameter in the error so a caller can
/// tell which argument was out of range. Extracted as an `i128` rather than an
/// `i64` so the whole `usize` range survives the way in: `max_write_buffer_size`
/// defaults to `usize::MAX`, which an `i64` parameter would refuse to take back
/// with an unnamed `OverflowError`, breaking `eval(repr(config))`.
fn usize_param(value: i128, parameter: &str) -> PyResult<usize> {
    usize::try_from(value).map_err(|_| {
        PyValueError::new_err(format!(
            "'{parameter}' must be between 0 and {}",
            usize::MAX
        ))
    })
}

/// What `IggyClient(...)` accepts: a bare `host:port`, a full `TcpConfig`, a
/// `QuicConfig` for the QUIC transport, an `HttpConfig` for the HTTP transport,
/// or a `WebSocketConfig` for the WebSocket transport.
#[derive(FromPyObject)]
pub enum PyClientConfig {
    #[pyo3(transparent)]
    Tcp(TcpConfig),
    #[pyo3(transparent)]
    Quic(QuicConfig),
    #[pyo3(transparent)]
    Http(HttpConfig),
    #[pyo3(transparent)]
    WebSocket(WebSocketConfig),
    #[pyo3(transparent, annotation = "str")]
    ServerAddress(String),
}
impl_stub_type!(PyClientConfig = TcpConfig | QuicConfig | HttpConfig | WebSocketConfig | String);

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

    /// Mirrors the literal in `WebSocketFramingConfig::new`'s signature.
    const DEFAULT_MAX_MESSAGE_SIZE: usize = 64 << 20;

    /// Mirrors the literal in `WebSocketFramingConfig::new`'s signature.
    const DEFAULT_MAX_FRAME_SIZE: usize = 16 << 20;

    /// The signature defaults have to be literals for the generated stub to stay
    /// valid Python, so nothing but this test stops them drifting from the SDK
    /// (and so from tungstenite) on a dependency bump.
    #[test]
    fn defaults_should_match_the_sdk() {
        let defaults = RustWebSocketFramingConfig::default();

        assert_eq!(
            defaults.max_message_size,
            Some(DEFAULT_MAX_MESSAGE_SIZE),
            "'max_message_size' drifted from the SDK, update the literal in \
             WebSocketFramingConfig::new's signature too"
        );
        assert_eq!(
            defaults.max_frame_size,
            Some(DEFAULT_MAX_FRAME_SIZE),
            "'max_frame_size' drifted from the SDK, update the literal in \
             WebSocketFramingConfig::new's signature too"
        );
        assert!(
            defaults.write_buffer_size.is_some(),
            "'write_buffer_size' lost its default, so the write buffer invariant \
             check in WebSocketFramingConfig::new would stop running"
        );
        assert!(
            defaults.max_write_buffer_size.is_some(),
            "'max_write_buffer_size' lost its default, so the write buffer \
             invariant check in WebSocketFramingConfig::new would stop running"
        );
    }
}