// 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 bytes::Bytes;
use iggy::prelude::{
    AutoCommit as RustAutoCommit, Consumer as RustConsumer, IggyClient as RustIggyClient,
    IggyExpiry as RustIggyExpiry, IggyMessage as RustMessage, MaxTopicSize as RustMaxTopicSize,
    Partitioning as RustPartitioning, PollingStrategy as RustPollingStrategy, *,
};
use pyo3::PyRef;
use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyDelta, PyList, PyType};
use pyo3_async_runtimes::tokio::future_into_py;
use pyo3_stub_gen::define_stub_info_gatherer;
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
use std::collections::BTreeMap;
use std::fmt::Display;
use std::str::FromStr;
use std::sync::Arc;

use crate::config::PyClientConfig;
use crate::consumer::{
    AutoCommit, Consumer as PyConsumer, ConsumerGroup as PyConsumerGroup,
    ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer,
};
use crate::duration::{py_delta_to_iggy_duration, reject_zero};
use crate::identifier::PyIdentifier;
use crate::options::OptionSpec as PyOptionSpec;
use crate::partitioning::PyPartitioning;
use crate::permissions::Permissions as PyPermissions;
use crate::producer::{IggyProducer, ProducerMode, RetryInterval, u32_param as producer_u32_param};
use crate::receive_message::{PollingStrategy, ReceiveMessage};
use crate::send_message::{SendMessage, SendMessagesResponse as PySendMessagesResponse};
use crate::stats::Stats as PyStats;
use crate::stream::{Stream, StreamDetails};
use crate::topic::{IggyExpiry, MaxTopicSize, Topic, TopicDetails};
use crate::user::{
    UserInfo as PyUserInfo, UserInfoDetails as PyUserInfoDetails, UserStatus as PyUserStatus,
};
use tokio::sync::Mutex;

/// A Python class representing the Iggy client.
/// It provides asynchronous functionality through the contained runtime.
#[gen_stub_pyclass]
#[pyclass]
pub struct IggyClient {
    inner: Arc<RustIggyClient>,
}

/// Converts SDK errors to the RuntimeError exposed by the Python API.
fn to_runtime_error<E: Display>(error: E) -> PyErr {
    PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(error.to_string())
}

fn to_value_error<E: Display>(error: E) -> PyErr {
    PyErr::new::<pyo3::exceptions::PyValueError, _>(error.to_string())
}

/// Resolves the shared `create_topic`/`update_topic` parameters, applying
/// server defaults where the caller left them unset.
fn resolve_topic_params(
    compression_algorithm: Option<String>,
    message_expiry: Option<&IggyExpiry>,
    max_topic_size: Option<&MaxTopicSize>,
) -> PyResult<(CompressionAlgorithm, RustIggyExpiry, RustMaxTopicSize)> {
    let compression_algorithm = match compression_algorithm {
        Some(algo) => CompressionAlgorithm::from_str(&algo).map_err(to_runtime_error)?,
        None => CompressionAlgorithm::default(),
    };

    let expiry = message_expiry
        .map(RustIggyExpiry::try_from)
        .transpose()?
        .unwrap_or(RustIggyExpiry::ServerDefault);

    let max_size = max_topic_size
        .map(RustMaxTopicSize::try_from)
        .transpose()?
        .unwrap_or(RustMaxTopicSize::ServerDefault);

    Ok((compression_algorithm, expiry, max_size))
}

#[gen_stub_pymethods]
#[pymethods]
impl IggyClient {
    /// Constructs a new IggyClient from a TCP server address, a `TcpConfig`, a
    /// `QuicConfig`, an `HttpConfig`, or a `WebSocketConfig`. Construction is
    /// synchronous; async methods return asyncio awaitables backed by the shared
    /// Tokio runtime.
    ///
    /// Args:
    ///     conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, an
    ///         `HttpConfig`, or a `WebSocketConfig`. Defaults to `127.0.0.1:8090`
    ///         over TCP with auto-login disabled. A malformed address is reported
    ///         differently depending on the form: the string form raises
    ///         `RuntimeError` here, while every config type raises `ValueError`
    ///         when it is constructed, before any of them reaches this call. Neither
    ///         exception is a subclass of the other.
    ///
    /// Raises:
    ///     RuntimeError: If the address passed as a string is not a valid
    ///         `host:port` pair, or if a `QuicConfig` client cannot bind its
    ///         local UDP socket (for example the port is already in use).
    #[new]
    #[pyo3(signature = (conn=None))]
    fn new(
        #[gen_stub(override_type(
            type_repr = "TcpConfig | QuicConfig | HttpConfig | WebSocketConfig | builtins.str | None"
        ))]
        conn: Option<PyClientConfig>,
    ) -> PyResult<Self> {
        let wrapper = match conn {
            Some(PyClientConfig::Tcp(config)) => ClientWrapper::Tcp(
                TcpClient::create(config.client_config()).map_err(to_runtime_error)?,
            ),
            Some(PyClientConfig::ServerAddress(server_address)) => {
                let config = Arc::new(
                    TcpClientConfigBuilder::new()
                        .with_server_address(server_address)
                        .build()
                        .map_err(to_runtime_error)?,
                );
                ClientWrapper::Tcp(TcpClient::create(config).map_err(to_runtime_error)?)
            }
            Some(PyClientConfig::Quic(config)) => {
                // `quinn::Endpoint::client` (invoked eagerly by `QuicClient::create`) looks
                // up the current Tokio runtime via `Handle::try_current()` and fails with
                // `CannotCreateEndpoint` if none is active. This method runs synchronously
                // from Python without one, so enter the runtime pyo3-async-runtimes uses
                // for our own async methods before building the endpoint.
                let _guard = pyo3_async_runtimes::tokio::get_runtime().enter();
                ClientWrapper::Quic(
                    QuicClient::create(config.client_config()).map_err(to_runtime_error)?,
                )
            }
            Some(PyClientConfig::Http(config)) => ClientWrapper::Http(
                HttpClient::create(config.client_config()).map_err(to_runtime_error)?,
            ),
            Some(PyClientConfig::WebSocket(config)) => ClientWrapper::WebSocket(
                WebSocketClient::create(config.client_config()).map_err(to_runtime_error)?,
            ),
            None => ClientWrapper::Tcp(
                TcpClient::create(Arc::new(TcpClientConfig::default()))
                    .map_err(to_runtime_error)?,
            ),
        };
        Ok(Self {
            inner: Arc::new(RustIggyClient::new(wrapper)),
        })
    }

    /// Constructs a new IggyClient from a connection string.
    /// Returns an error if the connection string provided is invalid.
    // TODO: add examples for connection strings or at least a link to the doc page where
    // connection strings are explained.
    #[classmethod]
    #[pyo3(signature = (connection_string))]
    fn from_connection_string(
        _cls: &Bound<'_, PyType>,
        connection_string: String,
    ) -> PyResult<Self> {
        // The QUIC transport builds its endpoint eagerly and needs a Tokio runtime context
        // to do so (see the `QuicConfig` arm of `new()` above for details); entering it here
        // is a no-op for the other transports since the protocol isn't known until the
        // connection string is parsed.
        let _guard = pyo3_async_runtimes::tokio::get_runtime().enter();
        let client =
            RustIggyClient::from_connection_string(&connection_string).map_err(to_runtime_error)?;
        Ok(Self {
            inner: Arc::new(client),
        })
    }

    /// Sends a ping request to the server to check connectivity.
    /// Raises `RuntimeError` if the connection fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn ping<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(
            py,
            async move { inner.ping().await.map_err(to_runtime_error) },
        )
    }

    /// Get the statistics and details of the server and its running process.
    ///
    /// Requires an authenticated session whose user holds the `read_servers`
    /// or `manage_servers` global permission.
    ///
    /// Returns:
    ///     An awaitable that resolves to `Stats`.
    ///
    /// Raises:
    ///     RuntimeError: If the client is not connected, the session is not
    ///         authenticated, the user lacks the permission, or the request
    ///         fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[Stats]", imports=("collections.abc")))]
    fn get_stats<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            let stats = inner.get_stats().await.map_err(to_runtime_error)?;
            Ok(PyStats::from(stats))
        })
    }

    /// Describe the option catalog for a resource scope.
    ///
    /// This is the discovery surface for the `options` argument on
    /// `create_topic`/`update_topic`: a key outside the catalog is refused at
    /// create, and the binary transports carry only the error code back.
    ///
    /// Args:
    ///     scope: One of `"topic"`, `"stream"`, `"user"`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `list[OptionSpec]`, empty for a scope
    ///     with no keys yet.
    ///
    /// Raises:
    ///     ValueError: If the scope name is not one of the three above.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[OptionSpec]]", imports=("collections.abc")))]
    fn describe_options<'a>(&self, py: Python<'a>, scope: &str) -> PyResult<Bound<'a, PyAny>> {
        let scope = crate::options::options_scope_from_str(scope)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let specs = inner
                .describe_options(scope)
                .await
                .map_err(to_runtime_error)?;
            Ok(specs
                .into_iter()
                .map(PyOptionSpec::from)
                .collect::<Vec<_>>())
        })
    }

    /// Logs in the user with the given credentials.
    /// Raises `RuntimeError` on failure.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn login_user<'a>(
        &self,
        py: Python<'a>,
        username: String,
        password: String,
    ) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner
                .login_user(&username, &password)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Get the info about a specific user by unique ID or username.
    ///
    /// Args:
    ///     user_id: User identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `UserInfoDetails` if the user exists,
    ///     or `None` otherwise.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[UserInfoDetails | None]", imports=("collections.abc")))]
    fn get_user<'a>(&self, py: Python<'a>, user_id: PyIdentifier) -> PyResult<Bound<'a, PyAny>> {
        let user_id = Identifier::try_from(user_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let user = inner.get_user(&user_id).await.map_err(to_runtime_error)?;
            Ok(user.map(PyUserInfoDetails::from))
        })
    }

    /// Get the info about all the users.
    ///
    /// Returns:
    ///     An awaitable that resolves to `list[UserInfo]`.
    ///
    /// Raises:
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[UserInfo]]", imports=("collections.abc")))]
    fn get_users<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let users = inner.get_users().await.map_err(to_runtime_error)?;
            Ok(users.into_iter().map(PyUserInfo::from).collect::<Vec<_>>())
        })
    }

    /// Create a new user.
    ///
    /// Args:
    ///     username: Username as `str`.
    ///     password: Password as `str`.
    ///     status: User status as `UserStatus | None`; defaults to `UserStatus.Active`.
    ///     permissions: Permissions as `Permissions | None`; the user has none when `None`.
    ///
    /// Returns:
    ///     An awaitable that resolves to the created `UserInfoDetails`.
    ///
    /// Raises:
    ///     RuntimeError: If an argument is invalid or the request fails.
    #[pyo3(signature = (username, password, status=None, permissions=None))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[UserInfoDetails]", imports=("collections.abc")))]
    fn create_user<'a>(
        &self,
        py: Python<'a>,
        username: String,
        password: String,
        #[gen_stub(override_type(type_repr = "UserStatus | None"))] status: Option<PyUserStatus>,
        #[gen_stub(override_type(type_repr = "Permissions | None"))] permissions: Option<
            PyPermissions,
        >,
    ) -> PyResult<Bound<'a, PyAny>> {
        let status = status.map_or(UserStatus::Active, UserStatus::from);
        let permissions = permissions.map(|permissions| permissions.inner);
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let user = inner
                .create_user(&username, &password, status, permissions)
                .await
                .map_err(to_runtime_error)?;
            Ok(PyUserInfoDetails::from(user))
        })
    }

    /// Update a user by unique ID or username.
    ///
    /// Args:
    ///     user_id: User identifier as `str | int`.
    ///     username: New username as `str | None`; unchanged when `None`.
    ///     status: New status as `UserStatus | None`; unchanged when `None`.
    ///     options: Reserved for future use. Additional option keys as
    ///         `dict[str, str] | None`, forwarded to the server. No user update
    ///         option key exists yet, so a current server rejects every key.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the user is updated.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[pyo3(signature = (user_id, username=None, status=None, options=None))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn update_user<'a>(
        &self,
        py: Python<'a>,
        user_id: PyIdentifier,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] username: Option<String>,
        #[gen_stub(override_type(type_repr = "UserStatus | None"))] status: Option<PyUserStatus>,
        #[gen_stub(override_type(type_repr = "builtins.dict[builtins.str, builtins.str] | None"))]
        options: Option<BTreeMap<String, String>>,
    ) -> PyResult<Bound<'a, PyAny>> {
        let user_id = Identifier::try_from(user_id)?;
        let status = status.map(UserStatus::from);
        let update_options = UserUpdateOptions {
            raw: options.unwrap_or_default(),
        };
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .update_user(&user_id, username.as_deref(), status, &update_options)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Delete a user by unique ID or username.
    ///
    /// Args:
    ///     user_id: User identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the user is deleted.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn delete_user<'a>(&self, py: Python<'a>, user_id: PyIdentifier) -> PyResult<Bound<'a, PyAny>> {
        let user_id = Identifier::try_from(user_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .delete_user(&user_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Update the permissions of a user by unique ID or username.
    ///
    /// This is a full replacement: the given permissions overwrite the previous
    /// ones, and `None` removes them entirely.
    ///
    /// Args:
    ///     user_id: User identifier as `str | int`.
    ///     permissions: New permissions as `Permissions | None`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the permissions are updated.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[pyo3(signature = (user_id, permissions))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn update_permissions<'a>(
        &self,
        py: Python<'a>,
        user_id: PyIdentifier,
        #[gen_stub(override_type(type_repr = "Permissions | None"))] permissions: Option<
            PyPermissions,
        >,
    ) -> PyResult<Bound<'a, PyAny>> {
        let user_id = Identifier::try_from(user_id)?;
        let permissions = permissions.map(|permissions| permissions.inner);
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .update_permissions(&user_id, permissions)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Change the password of a user by unique ID or username.
    ///
    /// Args:
    ///     user_id: User identifier as `str | int`.
    ///     current_password: Current password as `str`.
    ///     new_password: New password as `str`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the password is changed.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the current password is wrong or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn change_password<'a>(
        &self,
        py: Python<'a>,
        user_id: PyIdentifier,
        current_password: String,
        new_password: String,
    ) -> PyResult<Bound<'a, PyAny>> {
        let user_id = Identifier::try_from(user_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .change_password(&user_id, &current_password, &new_password)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Log out the currently authenticated user.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the user is logged out.
    ///
    /// Raises:
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn logout_user<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner.logout_user().await.map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Connects the IggyClient to its service and starts the heartbeat task.
    /// Raises `RuntimeError` if the connection fails. Over HTTP there is no
    /// connection to establish, so only the heartbeat starts and this call
    /// succeeds even against an unreachable server.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn connect<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner.connect().await.map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Closes the current connection. Repeated calls are safe. Call `connect`
    /// to use the client again. Over TCP and QUIC, that `connect` first waits
    /// for the rest of `reestablish_after` (5 s by default). The sign-in made
    /// with `login_user` is dropped, so it must be repeated after reconnecting.
    /// A client configured with auto-login credentials signs in again on
    /// `connect`. Over HTTP there is no connection to close and this call does
    /// nothing.
    ///
    /// Known issue: unless reconnection is disabled, the heartbeat of a client
    /// with auto-login credentials connects it again and signs in within one
    /// heartbeat interval. See https://github.com/apache/iggy/issues/4287.
    ///
    /// Raises:
    ///     RuntimeError: If the connection cannot be closed.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn disconnect<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner.disconnect().await.map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Closes the connection. Shut down background producers with
    /// `IggyProducer.shutdown()` and stop iterating consumers before this call,
    /// because they share the connection. Otherwise background producers drop
    /// queued messages and consumer iterators hang. Later requests fail with
    /// `RuntimeError`. Repeated calls are safe. Over HTTP there is no
    /// connection to close, but the heartbeat that `connect` started keeps
    /// sending pings until the client is dropped.
    ///
    /// Raises:
    ///     RuntimeError: If the client cannot be shut down.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn shutdown<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner.shutdown().await.map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Creates a new stream with the provided ID and name.
    /// Raises `RuntimeError` if the stream cannot be created.
    #[pyo3(signature = (name))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn create_stream<'a>(&self, py: Python<'a>, name: String) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner.create_stream(&name).await.map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Gets stream by id.
    /// Returns the stream details, or `None` if the stream does not exist.
    /// Raises `RuntimeError` on failure.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[StreamDetails | None]", imports=("collections.abc")))]
    fn get_stream<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let stream = inner
                .get_stream(&stream_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(stream.map(StreamDetails::from))
        })
    }

    /// Return all streams.
    ///
    /// Returns:
    ///     A list of `Stream` summaries.
    ///
    /// Raises:
    ///     RuntimeError: If the client is not authenticated, the user lacks global
    ///         `read_streams` or `manage_streams` permission, or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[Stream]]", imports=("collections.abc")))]
    fn get_streams<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            let streams = inner.get_streams().await.map_err(to_runtime_error)?;
            Ok(streams.into_iter().map(Stream::from).collect::<Vec<_>>())
        })
    }

    /// Rename a stream selected by name or numeric ID.
    ///
    /// `stream_id` accepts a stream name as `str` or numeric ID as `int`. A
    /// decimal-only string is interpreted as a numeric ID. `name` must be unique
    /// and contain between 1 and 255 UTF-8 bytes. Renaming a stream to its current
    /// name succeeds without changing it.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     name: New stream name as `str`.
    ///     options: Additional option keys as `dict[str, str] | None`, forwarded
    ///         to the server. Current server versions reject all stream update
    ///         option keys.
    ///
    /// Returns:
    ///     None.
    ///
    /// Raises:
    ///     TypeError: If `stream_id` is not `str` or an integer in
    ///         `0..=2**32 - 1`, or `name` is not `str`.
    ///     ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes.
    ///     RuntimeError: If the client is not authenticated, the user lacks global
    ///         `manage_streams` or per-stream `manage_stream` permission, the
    ///         stream does not exist, the new name is invalid or already used, or
    ///         the request fails.
    #[pyo3(signature = (stream_id, name, options=None))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn update_stream<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        name: String,
        #[gen_stub(override_type(type_repr = "builtins.dict[builtins.str, builtins.str] | None"))]
        options: Option<BTreeMap<String, String>>,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let update_options = StreamUpdateOptions {
            raw: options.unwrap_or_default(),
        };
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner
                .update_stream(&stream_id, &name, &update_options)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Delete a stream selected by name or numeric ID.
    ///
    /// Deletion removes the stream and all of its topics, partitions, and messages.
    /// `stream_id` accepts a stream name as `str` or numeric ID as `int`. A
    /// decimal-only string is interpreted as a numeric ID.
    ///
    /// Returns:
    ///     None.
    ///
    /// Raises:
    ///     TypeError: If `stream_id` is not `str` or an integer in
    ///         `0..=2**32 - 1`.
    ///     ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes.
    ///     RuntimeError: If the client is not authenticated, the user lacks global
    ///         `manage_streams` or per-stream `manage_stream` permission, the
    ///         stream does not exist, or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn delete_stream<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner
                .delete_stream(&stream_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Delete all messages from every topic in a stream.
    ///
    /// The stream, topics, and partitions remain available. Repeated purges of an
    /// existing empty stream succeed. `stream_id` accepts a stream name as `str`
    /// or numeric ID as `int`. A decimal-only string is interpreted as a numeric
    /// ID.
    ///
    /// Returns:
    ///     None.
    ///
    /// Raises:
    ///     TypeError: If `stream_id` is not `str` or an integer in
    ///         `0..=2**32 - 1`.
    ///     ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes.
    ///     RuntimeError: If the client is not authenticated, the user lacks global
    ///         `manage_streams` or per-stream `manage_stream` permission, the
    ///         stream does not exist, or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn purge_stream<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let inner = self.inner.clone();
        future_into_py(py, async move {
            inner
                .purge_stream(&stream_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Creates a new topic with the given parameters.
    ///
    /// Args:
    ///     stream: Stream identifier as `str | int`.
    ///     name: Topic name as `str`.
    ///     partitions_count: Number of partitions as `int`, at most 1000.
    ///     compression_algorithm: Compression algorithm as `str | None`.
    ///     message_expiry: Message expiry as `IggyExpiry | None`.
    ///     max_topic_size: Maximum topic size as `MaxTopicSize | None`.
    ///     segment_size: Per-topic segment size in bytes as `int | None`.
    ///     durability: Message completion policy, defaulting to replicated.
    ///     consumer_offset_durability: Independent offset policy, defaulting to replicated.
    ///     messages_required_to_save: Message-count flush threshold as `int | None`.
    ///     size_of_messages_required_to_save: Byte flush threshold as `int | None`.
    ///     preallocate_segments: Reserve segment bytes on open as `bool | None`.
    ///     options: Additional option keys as `dict[str, str] | None`, sent
    ///         verbatim so a newer server key can be set from this build.
    ///
    /// Every option left as `None` resolves against the server default at
    /// admission.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the topic is created.
    ///
    /// Raises:
    ///     ValueError: If `message_expiry` or `max_topic_size` is out of range.
    ///     PyRuntimeError: If another argument is invalid or the request fails.
    #[pyo3(
        signature = (stream, name, partitions_count, compression_algorithm = None, message_expiry = None, max_topic_size = None, segment_size = None, durability = None, consumer_offset_durability = None, messages_required_to_save = None, size_of_messages_required_to_save = None, preallocate_segments = None, options = None)
    )]
    #[allow(clippy::too_many_arguments)]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn create_topic<'a>(
        &self,
        py: Python<'a>,
        stream: PyIdentifier,
        name: String,
        partitions_count: u32,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] compression_algorithm: Option<
            String,
        >,
        #[gen_stub(override_type(type_repr = "IggyExpiry | None"))] message_expiry: Option<
            &IggyExpiry,
        >,
        #[gen_stub(override_type(type_repr = "MaxTopicSize | None"))] max_topic_size: Option<
            &MaxTopicSize,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] segment_size: Option<u64>,
        #[gen_stub(override_type(type_repr = "Durability | None"))] durability: Option<
            &Bound<'_, PyAny>,
        >,
        #[gen_stub(override_type(type_repr = "Durability | None"))]
        consumer_offset_durability: Option<&Bound<'_, PyAny>>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))]
        messages_required_to_save: Option<u32>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))]
        size_of_messages_required_to_save: Option<u64>,
        #[gen_stub(override_type(type_repr = "builtins.bool | None"))] preallocate_segments: Option<
            bool,
        >,
        #[gen_stub(override_type(type_repr = "builtins.dict[builtins.str, builtins.str] | None"))]
        options: Option<BTreeMap<String, String>>,
    ) -> PyResult<Bound<'a, PyAny>> {
        let (compression_algorithm, expiry, max_size) =
            resolve_topic_params(compression_algorithm, message_expiry, max_topic_size)?;

        let topic_options = TopicCreateOptions {
            partitions_count: Some(partitions_count),
            // `None` is what tells admission to resolve the server default, so
            // the sentinels the resolver returns must collapse back to it.
            compression_algorithm: (compression_algorithm != CompressionAlgorithm::default())
                .then_some(compression_algorithm),
            message_expiry: (expiry != RustIggyExpiry::ServerDefault).then_some(expiry),
            max_topic_size: (max_size != RustMaxTopicSize::ServerDefault).then_some(max_size),
            segment_size: segment_size.map(IggyByteSize::from),
            durability: crate::durability::Durability::try_from(durability)?.0,
            consumer_offset_durability: crate::durability::Durability::try_from(
                consumer_offset_durability,
            )?
            .0,
            messages_required_to_save,
            size_of_messages_required_to_save: size_of_messages_required_to_save
                .map(IggyByteSize::from),
            preallocate_segments,
            raw: options.unwrap_or_default(),
        };

        let stream = Identifier::try_from(stream)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .create_topic(&stream, &name, &topic_options)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Gets topic by stream and id.
    /// Returns the topic details, or `None` if the topic does not exist.
    /// Raises `RuntimeError` on failure.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[TopicDetails | None]", imports=("collections.abc")))]
    fn get_topic<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let topic = inner
                .get_topic(&stream_id, &topic_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(topic.map(TopicDetails::from))
        })
    }

    /// Get all topics in a stream.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `list[Topic]`.
    ///
    /// Raises:
    ///     RuntimeError: If the identifier is invalid or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[Topic]]", imports=("collections.abc")))]
    fn get_topics<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let topics = inner
                .get_topics(&stream_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(topics.into_iter().map(Topic::from).collect::<Vec<_>>())
        })
    }

    /// Update an existing topic.
    ///
    /// A patch, not a replacement: every setting rides the options block, so a
    /// field left unset keeps the topic's current value rather than resetting
    /// it to a server default.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     name: New topic name as `str`.
    ///     compression_algorithm: Compression algorithm as `str | None`.
    ///     message_expiry: Message expiry as `IggyExpiry | None`.
    ///     max_topic_size: Maximum topic size as `MaxTopicSize | None`.
    ///     options: Additional option keys as `dict[str, str] | None`, sent
    ///         verbatim so an updatable server key can be set from this build.
    ///         A create-only key is refused by name.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the topic is updated.
    ///
    /// Raises:
    ///     ValueError: If `message_expiry` or `max_topic_size` is out of range.
    ///     PyRuntimeError: If another argument is invalid or the request fails.
    #[pyo3(
        signature = (stream_id, topic_id, name, compression_algorithm = None, message_expiry = None, max_topic_size = None, options = None)
    )]
    #[allow(clippy::too_many_arguments)]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn update_topic<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        name: String,
        #[gen_stub(override_type(type_repr = "builtins.str | None"))] compression_algorithm: Option<
            String,
        >,
        #[gen_stub(override_type(type_repr = "IggyExpiry | None"))] message_expiry: Option<
            &IggyExpiry,
        >,
        #[gen_stub(override_type(type_repr = "MaxTopicSize | None"))] max_topic_size: Option<
            &MaxTopicSize,
        >,
        #[gen_stub(override_type(type_repr = "builtins.dict[builtins.str, builtins.str] | None"))]
        options: Option<BTreeMap<String, String>>,
    ) -> PyResult<Bound<'a, PyAny>> {
        // Absent stays absent: a key the caller did not pass is left alone
        // server-side rather than reset to a default.
        let compression_algorithm = compression_algorithm
            .map(|algo| CompressionAlgorithm::from_str(&algo).map_err(to_runtime_error))
            .transpose()?;
        let update_options = TopicUpdateOptions {
            compression_algorithm,
            message_expiry: message_expiry.map(RustIggyExpiry::try_from).transpose()?,
            max_topic_size: max_topic_size.map(RustMaxTopicSize::try_from).transpose()?,
            raw: options.unwrap_or_default(),
        };

        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .update_topic(&stream_id, &topic_id, &name, &update_options)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Delete a topic from a stream.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the topic is deleted.
    ///
    /// Raises:
    ///     RuntimeError: If an identifier is invalid or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn delete_topic<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .delete_topic(&stream_id, &topic_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Purge all messages from a topic.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the topic is purged.
    ///
    /// Raises:
    ///     RuntimeError: If an identifier is invalid or the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn purge_topic<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .purge_topic(&stream_id, &topic_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Create partitions for a topic. New partition IDs continue from one past the
    /// current highest ID; IDs removed by deletion can be reused. Existing consumer
    /// groups are immediately rebalanced across all partitions, advancing their
    /// generation and dropping pending revocations.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     partitions_count: Number of partitions to create as `int`, between 1 and
    ///         1000 inclusive.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the partitions are committed;
    ///     storage materialization completes asynchronously.
    ///
    /// Raises:
    ///     ValueError: If an identifier is invalid.
    ///     OverflowError: If `partitions_count` is outside the unsigned 32-bit range.
    ///     RuntimeError: If the client is not authenticated, lacks global
    ///         `manage_streams` or `manage_topics`, per-stream `manage_stream` or
    ///         `manage_topics`, or per-topic `manage_topic` permission, or the
    ///         request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn create_partitions<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        partitions_count: u32,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .create_partitions(&stream_id, &topic_id, partitions_count)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Delete the last partitions from a topic, including all messages stored in them.
    /// Existing consumer groups are immediately rebalanced across the remaining
    /// partitions, advancing their generation and dropping pending revocations.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     partitions_count: Number of partitions to delete as `int` from the end of
    ///         the topic; must be between 1 and 1000 inclusive and no greater than
    ///         its current count.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when deletion is accepted; storage
    ///     teardown completes asynchronously.
    ///
    /// Raises:
    ///     ValueError: If an identifier is invalid.
    ///     OverflowError: If `partitions_count` is outside the unsigned 32-bit range.
    ///     RuntimeError: If the client is not authenticated, lacks global
    ///         `manage_streams` or `manage_topics`, per-stream `manage_stream` or
    ///         `manage_topics`, or per-topic `manage_topic` permission, or the
    ///         request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn delete_partitions<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        partitions_count: u32,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .delete_partitions(&stream_id, &topic_id, partitions_count)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Create a consumer group for a stream and topic.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     name: Consumer group name as `str`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the consumer group is created.
    ///
    /// Raises:
    ///     ValueError: If an identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn create_consumer_group<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        name: String,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .create_consumer_group(&stream_id, &topic_id, &name)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Retrieve details for a consumer group from the specified stream and topic.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     group_id: Consumer group identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `ConsumerGroupDetails` if the consumer group exists,
    ///     or `None` otherwise.
    ///
    /// Raises:
    ///     ValueError: If an identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[ConsumerGroupDetails | None]", imports=("collections.abc")))]
    fn get_consumer_group<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        group_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let group_id = Identifier::try_from(group_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let group = inner
                .get_consumer_group(&stream_id, &topic_id, &group_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(group.map(PyConsumerGroupDetails::from))
        })
    }

    /// Get all consumer groups for the specified stream and topic.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `list[ConsumerGroup]`.
    ///
    /// Raises:
    ///     ValueError: If an identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[ConsumerGroup]]", imports=("collections.abc")))]
    fn get_consumer_groups<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let groups = inner
                .get_consumer_groups(&stream_id, &topic_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(groups
                .into_iter()
                .map(PyConsumerGroup::from)
                .collect::<Vec<_>>())
        })
    }

    /// Delete a consumer group for a stream and topic.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     group_id: Consumer group identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the consumer group is deleted.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn delete_consumer_group<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        group_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let group_id = Identifier::try_from(group_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .delete_consumer_group(&stream_id, &topic_id, &group_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Join a consumer group for a stream and topic.
    ///
    /// This method only registers the current client as a group member. To consume messages
    /// as a group, use `consumer_group()`, which enables auto-join by default.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     group_id: Consumer group identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the client joins the consumer group.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails, including `Feature is unavailable` on HTTP transport.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn join_consumer_group<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        group_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let group_id = Identifier::try_from(group_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .join_consumer_group(&stream_id, &topic_id, &group_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Leave a consumer group for a stream and topic.
    ///
    /// Args:
    ///     stream_id: Stream identifier as `str | int`.
    ///     topic_id: Topic identifier as `str | int`.
    ///     group_id: Consumer group identifier as `str | int`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `None` when the client leaves the consumer group.
    ///
    /// Note:
    ///     Consumers created from this client for the same group share one server-side
    ///     membership. Leaving revokes that membership. Consumers with auto-join enabled
    ///     rejoin on their next poll.
    ///
    /// Raises:
    ///     ValueError: If a string identifier is invalid.
    ///     RuntimeError: If the request fails, including `Feature is unavailable` on HTTP transport.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
    fn leave_consumer_group<'a>(
        &self,
        py: Python<'a>,
        stream_id: PyIdentifier,
        topic_id: PyIdentifier,
        group_id: PyIdentifier,
    ) -> PyResult<Bound<'a, PyAny>> {
        let stream_id = Identifier::try_from(stream_id)?;
        let topic_id = Identifier::try_from(topic_id)?;
        let group_id = Identifier::try_from(group_id)?;
        let inner = self.inner.clone();

        future_into_py(py, async move {
            inner
                .leave_consumer_group(&stream_id, &topic_id, &group_id)
                .await
                .map_err(to_runtime_error)?;
            Ok(())
        })
    }

    /// Sends a batch of messages to a topic using the selected partitioning strategy.
    ///
    /// Args:
    ///     stream: Stream identifier as `str | int`.
    ///     topic: Topic identifier as `str | int`.
    ///     partitioning: A `Partitioning` strategy or an integer partition ID.
    ///         Use `Partitioning.balanced()`, `Partitioning.partition_id(id)`, or
    ///         `Partitioning.messages_key(key)`. An integer is shorthand for
    ///         `Partitioning.partition_id(id)`.
    ///     messages: Messages to send as `list[SendMessage]`.
    ///
    /// Returns:
    ///     An awaitable that resolves to `SendMessagesResponse`. Its confirmations
    ///     report the committed partition and batch base offset. The list is empty
    ///     when the server reports no offsets.
    ///
    /// Raises:
    ///     ValueError: If a string stream or topic identifier is invalid.
    ///     TypeError: If `partitioning` or `messages` has an unsupported type.
    ///     OverflowError: If a numeric stream, topic, or partition ID is outside
    ///         the supported unsigned 32-bit range.
    ///     RuntimeError: If the request fails.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[SendMessagesResponse]", imports=("collections.abc")))]
    fn send_messages<'a>(
        &self,
        py: Python<'a>,
        stream: PyIdentifier,
        topic: PyIdentifier,
        #[gen_stub(override_type(type_repr = "Partitioning | builtins.int"))]
        partitioning: PyPartitioning,
        #[gen_stub(override_type(type_repr = "list[SendMessage]"))] messages: &Bound<'_, PyList>,
    ) -> PyResult<Bound<'a, PyAny>> {
        let messages: Vec<SendMessage> = messages
            .iter()
            .map(|item| {
                let msg: PyRef<'_, SendMessage> = item.extract()?;
                Ok::<_, PyErr>(msg.clone())
            })
            .collect::<Result<Vec<_>, _>>()?;
        let mut messages: Vec<RustMessage> = messages
            .into_iter()
            .map(|message| message.inner)
            .collect::<Vec<_>>();

        let stream = Identifier::try_from(stream)?;
        let topic = Identifier::try_from(topic)?;
        let partitioning = RustPartitioning::from(partitioning);
        let inner = self.inner.clone();

        future_into_py(py, async move {
            let response = inner
                .send_messages(&stream, &topic, &partitioning, messages.as_mut())
                .await
                .map_err(to_runtime_error)?;
            Ok(PySendMessagesResponse::from(response))
        })
    }

    /// Creates and initializes a high-level producer bound to a stream and topic.
    ///
    /// This is a Python port of the Rust high-level producer API. For detailed
    /// producer semantics, see https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
    /// `None` selects direct mode. `BackgroundProducerConfig` starts background
    /// workers and makes successful sends mean queue acceptance rather than a
    /// server commit. Replicated topics accept producer writes. Set
    /// `topic_durability=Durability.PERSISTED` for an automatically created topic
    /// when sends require crash-safe retry receipts.
    ///
    /// Raises `ValueError` for invalid names or numeric ranges and `RuntimeError`
    /// when stream/topic initialization fails.
    #[allow(clippy::too_many_arguments)]
    #[pyo3(signature = (
        stream,
        topic,
        partitioning=None,
        mode=None,
        create_stream_if_not_exists=true,
        create_topic_if_not_exists=true,
        topic_partitions_count=1,
        topic_message_expiry=None,
        topic_max_size=None,
        send_retries=Some(3),
        send_retry_interval=RetryInterval::default(),
        topic_durability=None,
    ))]
    #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[IggyProducer]", imports=("collections.abc")))]
    fn producer<'a>(
        &self,
        py: Python<'a>,
        stream: &str,
        topic: &str,
        #[gen_stub(override_type(type_repr = "Partitioning | None"))] partitioning: Option<
            &crate::partitioning::Partitioning,
        >,
        #[gen_stub(override_type(
            type_repr = "DirectProducerConfig | BackgroundProducerConfig | None"
        ))]
        mode: Option<ProducerMode>,
        create_stream_if_not_exists: bool,
        create_topic_if_not_exists: bool,
        topic_partitions_count: i64,
        #[gen_stub(override_type(type_repr = "IggyExpiry | None"))] topic_message_expiry: Option<
            &IggyExpiry,
        >,
        #[gen_stub(override_type(type_repr = "MaxTopicSize | None"))] topic_max_size: Option<
            &MaxTopicSize,
        >,
        send_retries: Option<i64>,
        send_retry_interval: RetryInterval,
        #[gen_stub(override_type(type_repr = "Durability | None"))] topic_durability: Option<
            &Bound<'_, PyAny>,
        >,
    ) -> PyResult<Bound<'a, PyAny>> {
        let mode = mode.unwrap_or_default();

        let topic_partitions_count =
            producer_u32_param(topic_partitions_count, "topic_partitions_count")?;
        let topic_message_expiry = topic_message_expiry
            .map(RustIggyExpiry::try_from)
            .transpose()?
            .unwrap_or(RustIggyExpiry::ServerDefault);
        let topic_max_size = topic_max_size
            .map(RustMaxTopicSize::try_from)
            .transpose()?
            .unwrap_or(RustMaxTopicSize::ServerDefault);
        let send_retries = send_retries
            .map(|retries| producer_u32_param(retries, "send_retries"))
            .transpose()?
            .filter(|retries| *retries != 0);
        let send_retry_interval = send_retry_interval.resolve()?;

        let mut builder = self
            .inner
            .producer(stream, topic)
            .map_err(to_value_error)?
            .send_retries(send_retries, send_retry_interval)
            .topic_durability(crate::durability::Durability::try_from(topic_durability)?.0);

        builder = match mode {
            ProducerMode::Direct(config) => builder.direct((&config).into()),
            ProducerMode::Background(config) => builder.background((&config).try_into()?),
        };

        if let Some(partitioning) = partitioning {
            builder = builder.partitioning(partitioning.inner.as_ref().clone());
        }
        if create_stream_if_not_exists {
            builder = builder.create_stream_if_not_exists();
        } else {
            builder = builder.do_not_create_stream_if_not_exists();
        }
        if create_topic_if_not_exists {
            builder = builder.create_topic_if_not_exists(
                topic_partitions_count,
                topic_message_expiry,
                topic_max_size,
            );
        } else {
            builder = builder.do_not_create_topic_if_not_exists();
        }

        future_into_py(py, async move {
            // A background build starts Tokio worker tasks, so it must happen
            // while this future is executing on the Rust runtime.
            let producer = builder.build();
            producer.init().await.map_err(to_runtime_error)?;
            Ok(IggyProducer::new(producer))
        })
    }

    /// Polls for messages from the specified topic on behalf of the given consumer.
    /// Omitting `partition_id` reads partition 0 for a regular consumer, and
    /// polls the member's assigned partitions for a consumer group.
    /// Returns a list of received messages or a RuntimeError on failure.
    #[allow(clippy::too_many_arguments)]
    #[pyo3(signature = (stream, topic, *, consumer, polling_strategy, count, auto_commit, partition_id = None))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[ReceiveMessage]]", imports=("collections.abc")))]
    fn poll_messages<'a>(
        &self,
        py: Python<'a>,
        stream: PyIdentifier,
        topic: PyIdentifier,
        consumer: &PyConsumer,
        polling_strategy: &PollingStrategy,
        count: u32,
        auto_commit: bool,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] partition_id: Option<u32>,
    ) -> PyResult<Bound<'a, PyAny>> {
        let consumer = RustConsumer::try_from(consumer)?;
        let stream = Identifier::try_from(stream)?;
        let topic = Identifier::try_from(topic)?;
        let strategy: RustPollingStrategy = polling_strategy.into();

        let inner = self.inner.clone();

        future_into_py(py, async move {
            let polled_messages = inner
                .poll_messages(
                    &stream,
                    &topic,
                    partition_id,
                    &consumer,
                    &strategy,
                    count,
                    auto_commit,
                )
                .await
                .map_err(to_runtime_error)?;
            let partition_id = polled_messages.partition_id;
            let messages = polled_messages
                .messages
                .into_iter()
                .map(|m| ReceiveMessage {
                    inner: m,
                    partition_id,
                })
                .collect::<Vec<_>>();
            Ok(messages)
        })
    }

    /// Creates a new consumer group consumer.
    /// `partition_id` is ignored for a consumer group: the member reads the partitions
    /// the server assigns to it.
    /// Returns the consumer or a RuntimeError on failure. Raises `ValueError` if
    /// `poll_interval`, `polling_retry_interval`, `init_retry_interval` or an
    /// `AutoCommit` interval is negative, or if any of those except `poll_interval`
    /// is zero.
    ///
    /// Consumer groups are not available over HTTP. With `auto_join_consumer_group`
    /// left on, this call fails at the join with `Feature is unavailable`.
    /// Turning it off is not a workaround: the join is skipped, but a group
    /// member always polls without a partition, so the first poll fails with
    /// the same error. Use `Consumer.Single(...)` with `poll_messages(...)`
    /// instead - a `Consumer.Group(...)` poll with an explicit `partition_id`
    /// does reach the server, but is served as an ordinary consumer named
    /// after the group.
    #[allow(clippy::too_many_arguments)]
    #[pyo3(signature = (
        name,
        stream,
        topic,
        partition_id=None,
        polling_strategy=None,
        batch_length=None,
        auto_commit=None,
        create_consumer_group_if_not_exists=true,
        auto_join_consumer_group=true,
        poll_interval=None,
        polling_retry_interval=None,
        init_retries=None,
        init_retry_interval=None,
        allow_replay=false,
    ))]
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[IggyConsumer]", imports=("collections.abc")))]
    fn consumer_group<'a>(
        &self,
        py: Python<'a>,
        name: &str,
        stream: &str,
        topic: &str,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] partition_id: Option<u32>,
        #[gen_stub(override_type(type_repr = "PollingStrategy | None"))] polling_strategy: Option<
            &PollingStrategy,
        >,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] batch_length: Option<u32>,
        #[gen_stub(override_type(type_repr = "AutoCommit | None"))] auto_commit: Option<
            &AutoCommit,
        >,
        create_consumer_group_if_not_exists: bool,
        auto_join_consumer_group: bool,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        poll_interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        polling_retry_interval: Option<Py<PyDelta>>,
        #[gen_stub(override_type(type_repr = "builtins.int | None"))] init_retries: Option<u32>,
        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
        init_retry_interval: Option<Py<PyDelta>>,
        allow_replay: bool,
    ) -> PyResult<Bound<'a, PyAny>> {
        let mut builder = self
            .inner
            .consumer_group(name, stream, topic)
            .map_err(to_runtime_error)?
            .without_encryptor()
            .partition(partition_id);

        if create_consumer_group_if_not_exists {
            builder = builder.create_consumer_group_if_not_exists()
        } else {
            builder = builder.do_not_create_consumer_group_if_not_exists()
        };
        if auto_join_consumer_group {
            builder = builder.auto_join_consumer_group()
        } else {
            builder = builder.do_not_auto_join_consumer_group()
        };
        if let Some(polling_strategy) = polling_strategy {
            builder = builder.polling_strategy(polling_strategy.into())
        };
        if let Some(batch_length) = batch_length {
            builder = builder.batch_length(batch_length)
        };
        if let Some(auto_commit) = auto_commit {
            builder = builder.auto_commit(RustAutoCommit::try_from(auto_commit)?)
        };
        if let Some(poll_interval) = poll_interval {
            builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval)?)
        } else {
            builder = builder.without_poll_interval()
        };
        if let Some(polling_retry_interval) = polling_retry_interval {
            builder = builder.polling_retry_interval(reject_zero(
                py_delta_to_iggy_duration(&polling_retry_interval)?,
                "polling_retry_interval",
            )?)
        }
        if init_retries.is_some() && init_retry_interval.is_none() {
            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
                "'init_retry_interval' is required if 'init_retries' is set",
            ));
        }
        if init_retries.is_none() && init_retry_interval.is_some() {
            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
                "'init_retries' is required if 'init_retry_interval' is set",
            ));
        }
        if let (Some(init_retries), Some(init_retry_interval)) = (init_retries, init_retry_interval)
        {
            builder = builder.init_retries(
                init_retries,
                reject_zero(
                    py_delta_to_iggy_duration(&init_retry_interval)?,
                    "init_retry_interval",
                )?,
            );
        }
        if allow_replay {
            builder = builder.allow_replay()
        }
        let mut consumer = builder.build();

        future_into_py(py, async move {
            consumer.init().await.map_err(to_runtime_error)?;
            let state = consumer.state();
            let name = consumer.name().to_string();
            let stream = PyIdentifier::try_from(consumer.stream())?;
            let topic = PyIdentifier::try_from(consumer.topic())?;
            Ok(IggyConsumer {
                inner: Arc::new(Mutex::new(consumer)),
                state,
                name,
                stream,
                topic,
            })
        })
    }

    /// Send a command code with a payload and return the raw response bytes.
    ///
    /// Session-control codes are rejected client-side. HTTP transport does not
    /// support raw binary commands.
    ///
    /// Args:
    ///     code: Command code as `int`.
    ///     payload: Request payload as `bytes`.
    ///
    /// Returns:
    ///     An awaitable that resolves to the raw response `bytes`.
    ///
    /// Raises:
    ///     RuntimeError: If the command cannot be sent or the server returns an error.
    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[bytes]", imports=("collections.abc")))]
    fn send_binary_request<'a>(
        &self,
        py: Python<'a>,
        code: u32,
        #[gen_stub(override_type(type_repr = "builtins.bytes"))] payload: Vec<u8>,
    ) -> PyResult<Bound<'a, PyAny>> {
        let inner = self.inner.clone();
        future_into_py(py, async move {
            let response = inner
                .send_binary_request(code, Bytes::from(payload))
                .await
                .map_err(to_runtime_error)?;
            Ok(Python::attach(|py| PyBytes::new(py, &response).unbind()))
        })
    }
}

define_stub_info_gatherer!(stub_info);