ruststream-fred 0.7.0

//! The broker ladder: [`RedisBroker`] -> [`ConnectedRedisBroker`] -> [`ClosedRedisBroker`].
//!
//! Construction is synchronous and I/O-free: the named constructors only record the topology and
//! its options. All network work happens in the consuming [`Broker::connect`], and the connected
//! form owns the live `fred` pool, so subscriptions and publishers exist only once a connection
//! does.

use std::num::NonZeroUsize;
use std::sync::Arc;
#[cfg(feature = "testing")]
use std::sync::Mutex;

use fred::clients::{Client, Pool};
use fred::interfaces::{ClientLike, EventInterface, PubsubInterface, StreamsInterface};
#[cfg(feature = "credential-provider")]
use fred::types::config::CredentialProvider;
#[cfg(any(
    feature = "tls-rustls",
    feature = "tls-rustls-ring",
    feature = "tls-native-tls"
))]
use fred::types::config::TlsConfig;
use fred::types::config::{Config, PerformanceConfig, ServerConfig};
use fred::types::{ClusterHash, CustomCommand, Message, Value, Version};
use fred::util::redis_keyslot;
#[cfg(feature = "testing")]
use ruststream::testing::{Coordinator, InProcess, TestableBroker};
use ruststream::{AddressedCopies, Broker, ConnectedBroker, DescribeServer, ServerSpec, Subscribe};
#[cfg(feature = "testing")]
use ruststream::{OutgoingMessage, RawMessage};
use tokio::runtime::Handle;
use tokio::sync::broadcast::Receiver;

use crate::connection::{Closing, Connection, ReadConnection};
#[cfg(feature = "testing")]
use crate::convert::{fields_for_publish, parts_from_fields};
use crate::delay::DelayConfig;
#[cfg(feature = "testing")]
use crate::envelope::{frame, unframe};
#[cfg(feature = "testing")]
use crate::in_process::{self, Subscription, Write};
use crate::loopback::{Loopback, Tap};
use crate::recovery::RecoveryConfig;

use crate::{
    error::RedisError,
    list::{ListWire, RedisList, RedisListPublish, RedisListPublisher, RedisListSubscriber},
    pipeline::{Owner, Rounds},
    publisher::{RedisDefaultPublisher, RedisPublisher},
    pubsub::{
        PubSubMode, PubSubWire, RedisPubSub, RedisPubSubPattern, RedisPubSubPublish,
        RedisPubSubPublisher, RedisPubSubSubscriber,
    },
    route::{Recorded, Route, Routes},
    stream::{ReadMode, RedisStream},
    subscriber::RedisSubscriber,
};

/// Default `fred` connection-pool size when the caller does not set one.
const DEFAULT_POOL_SIZE: usize = 4;

/// The first Redis release whose `XREADGROUP` takes a `CLAIM` option.
const CLAIM_MIN_VERSION: Version = Version::new(8, 4, 0);

/// How the broker should connect, recorded synchronously and resolved into a `fred` config at
/// [`Broker::connect`] time so construction stays I/O- and failure-free.
#[derive(Debug, Clone)]
enum Topology {
    /// A single server, addressed by URL (`redis://host:port`).
    Standalone(String),
    /// A Redis Cluster, addressed by one or more `host:port` seed nodes.
    Cluster(Vec<String>),
    /// Sentinel-managed replication: the monitored primary's `service` name plus the `host:port`
    /// of each sentinel.
    Sentinel { service: String, hosts: Vec<String> },
    /// A pool supplied already-connected via [`RedisBroker::from_pool`]; `connect` adopts it
    /// instead of dialing.
    Preconnected(Pool),
}

/// Parses a `host:port` address (tolerating a `redis://` / `rediss://` scheme prefix) into the
/// `(host, port)` pair `fred`'s server-config constructors expect. Falls back to `default_port`
/// when no port is given.
fn parse_server(addr: &str, default_port: u16) -> Result<(String, u16), RedisError> {
    let trimmed = addr
        .trim()
        .trim_start_matches("rediss://")
        .trim_start_matches("redis://");
    let (host, port) = match trimmed.rsplit_once(':') {
        Some((host, port)) => {
            let port = port.parse::<u16>().map_err(|_| {
                RedisError::Connect(format!("invalid port in redis address `{addr}`").into())
            })?;
            (host, port)
        }
        None => (trimmed, default_port),
    };
    if host.is_empty() {
        return Err(RedisError::Connect(
            format!("missing host in redis address `{addr}`").into(),
        ));
    }
    Ok((host.to_owned(), port))
}

fn parse_servers(addrs: &[String], default_port: u16) -> Result<Vec<(String, u16)>, RedisError> {
    if addrs.is_empty() {
        return Err(RedisError::Connect("no redis addresses provided".into()));
    }
    addrs
        .iter()
        .map(|addr| parse_server(addr, default_port))
        .collect()
}

/// Authentication and TLS settings recorded on the broker and folded into the `fred` [`Config`]
/// at connect time, on every topology. Fields with no value are left untouched, so credentials
/// supplied through a standalone `redis://user:pass@host` URL survive unless overridden here.
#[derive(Clone, Default)]
struct AuthConfig {
    /// ACL username for the data nodes (`Config.username`).
    username: Option<String>,
    /// Password for the data nodes (`Config.password`).
    password: Option<String>,
    /// ACL username for authenticating to the sentinels, distinct from the data-node username.
    #[cfg(feature = "sentinel-auth")]
    sentinel_username: Option<String>,
    /// Password for authenticating to the sentinels, distinct from the data-node password.
    #[cfg(feature = "sentinel-auth")]
    sentinel_password: Option<String>,
    /// Explicit TLS configuration (`Config.tls`).
    #[cfg(any(
        feature = "tls-rustls",
        feature = "tls-rustls-ring",
        feature = "tls-native-tls"
    ))]
    tls: Option<TlsConfig>,
    /// Dynamic/rotating credential provider (`Config.credential_provider`).
    #[cfg(feature = "credential-provider")]
    credential_provider: Option<Arc<dyn CredentialProvider>>,
}

// Redacts secrets: passwords never appear, and TLS / credential-provider show only presence. The
// usernames are identifiers (not secrets) and are kept to aid debugging.
impl std::fmt::Debug for AuthConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut s = f.debug_struct("AuthConfig");
        s.field("username", &self.username);
        s.field("password", &self.password.as_ref().map(|_| "<redacted>"));
        #[cfg(feature = "sentinel-auth")]
        {
            s.field("sentinel_username", &self.sentinel_username);
            s.field(
                "sentinel_password",
                &self.sentinel_password.as_ref().map(|_| "<redacted>"),
            );
        }
        #[cfg(any(
            feature = "tls-rustls",
            feature = "tls-rustls-ring",
            feature = "tls-native-tls"
        ))]
        s.field("tls", &self.tls.as_ref().map(|_| "<configured>"));
        #[cfg(feature = "credential-provider")]
        s.field(
            "credential_provider",
            &self.credential_provider.as_ref().map(|_| "<configured>"),
        );
        s.finish()
    }
}

/// An unconnected Redis broker: the recorded topology and its options, no I/O performed yet.
///
/// Build it with [`standalone`](Self::standalone), [`cluster`](Self::cluster),
/// [`sentinel`](Self::sentinel), or [`from_pool`](Self::from_pool). All four are synchronous and
/// perform no I/O, so a Redis service composes with the synchronous `#[ruststream::app]` builder;
/// the runtime dials once at startup through the consuming [`Broker::connect`], which yields the
/// [`ConnectedRedisBroker`] every subscription and publisher is reached from.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_fred::stream::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(RedisStream::new("orders").group("workers"))]
/// async fn handle(order: &Order) -> HandlerOutcome {
///     println!("order {}", order.id);
///     HandlerOutcome::ack()
/// }
///
/// // The broker is a plain value here; the runtime connects it at startup and shuts it down
/// // when the service stops.
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         RedisBroker::standalone("redis://localhost:6379"),
///         |b| {
///             b.include(handle);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone)]
#[must_use]
pub struct RedisBroker {
    topology: Topology,
    pool_size: usize,
    default_group: Option<String>,
    auth: AuthConfig,
}

impl RedisBroker {
    /// Creates a standalone-topology broker that connects to `url` when [`Broker::connect`] runs.
    pub fn standalone(url: impl Into<String>) -> Self {
        Self::with_topology(Topology::Standalone(url.into()))
    }

    /// Creates a Redis Cluster broker from one or more `host:port` seed nodes.
    ///
    /// Only one reachable node is needed; `fred` discovers the rest of the cluster on connect.
    pub fn cluster(nodes: impl IntoIterator<Item = impl Into<String>>) -> Self {
        Self::with_topology(Topology::Cluster(
            nodes.into_iter().map(Into::into).collect(),
        ))
    }

    /// Creates a Sentinel-backed broker that tracks the primary named `service`, discovering it
    /// through the given sentinel `host:port` addresses.
    pub fn sentinel(
        service: impl Into<String>,
        sentinels: impl IntoIterator<Item = impl Into<String>>,
    ) -> Self {
        Self::with_topology(Topology::Sentinel {
            service: service.into(),
            hosts: sentinels.into_iter().map(Into::into).collect(),
        })
    }

    /// Wraps an already-connected `fred` pool. Useful for advanced configuration (TLS, cluster,
    /// sentinel, custom performance and reconnection policies).
    ///
    /// [`Broker::connect`] adopts the pool instead of dialing; the config it was built from is
    /// reused for the dedicated clients Pub/Sub subscriptions need.
    pub fn from_pool(pool: Pool) -> Self {
        Self::with_topology(Topology::Preconnected(pool))
    }

    fn with_topology(topology: Topology) -> Self {
        Self {
            topology,
            pool_size: DEFAULT_POOL_SIZE,
            default_group: None,
            auth: AuthConfig::default(),
        }
    }

    /// Sets the connection-pool size: the connections publishes, settlements and handler commands
    /// share. Defaults to 4. Each subscription reads on a connection of its own besides.
    pub const fn pool(mut self, size: usize) -> Self {
        self.pool_size = size;
        self
    }

    /// Sets a broker-wide default consumer group, enabling the bare-string `#[subscriber("key")]`
    /// form (Redis Streams always read through a group). Without it a bare-string subscription
    /// returns [`RedisError::InvalidOptions`]; name the group per subscription with
    /// [`RedisStream::group`] instead.
    pub fn default_group(mut self, group: impl Into<String>) -> Self {
        self.default_group = Some(group.into());
        self
    }

    /// Sets the ACL `username` and `password` used to authenticate on connect, applied on every
    /// topology (standalone, cluster, sentinel).
    ///
    /// This maps onto `fred`'s `Config.username` / `Config.password`, so authentication works
    /// beyond the standalone `redis://user:pass@host` URL, which the bare `cluster` / `sentinel`
    /// seed lists cannot express. Credentials set here override any in a standalone URL.
    ///
    /// For a password-only `AUTH` (the legacy `requirepass`, no ACL user) use
    /// [`password`](Self::password).
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream_fred::stream::prelude::*;
    /// # use serde::Deserialize;
    /// # #[derive(Deserialize)]
    /// # struct Order {
    /// #     id: u64,
    /// # }
    /// #
    /// # #[subscriber(RedisStream::new("orders").group("workers"))]
    /// # async fn handle(order: &Order) -> HandlerOutcome {
    /// #     println!("order {}", order.id);
    /// #     HandlerOutcome::ack()
    /// # }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
    ///         RedisBroker::cluster(["10.0.0.1:6379"]).credentials("worker", "s3cr3t"),
    ///         |b| {
    ///             b.include(handle);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.auth.username = Some(username.into());
        self.auth.password = Some(password.into());
        self
    }

    /// Sets a password-only `AUTH` (no ACL username; the legacy `requirepass` form), on every
    /// topology. Use [`credentials`](Self::credentials) for an ACL user plus password.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream_fred::stream::prelude::*;
    /// # use serde::Deserialize;
    /// # #[derive(Deserialize)]
    /// # struct Order {
    /// #     id: u64,
    /// # }
    /// #
    /// # #[subscriber(RedisStream::new("orders").group("workers"))]
    /// # async fn handle(order: &Order) -> HandlerOutcome {
    /// #     println!("order {}", order.id);
    /// #     HandlerOutcome::ack()
    /// # }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
    ///         RedisBroker::sentinel("mymaster", ["10.0.0.1:26379"]).password("s3cr3t"),
    ///         |b| {
    ///             b.include(handle);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn password(mut self, password: impl Into<String>) -> Self {
        self.auth.password = Some(password.into());
        self
    }

    /// Sets the TLS configuration used on connect, on every topology. Accepts a `fred`
    /// [`TlsConfig`] or anything convertible into one (for example a `TlsConnector`).
    ///
    /// Available behind the `tls-rustls`, `tls-rustls-ring`, or `tls-native-tls` feature; a
    /// standalone broker can also enable TLS through a `rediss://` / `valkeys://` URL. The
    /// `fred` re-exports [`TlsConfig`](crate::TlsConfig) / [`TlsConnector`](crate::TlsConnector)
    /// provide `default_rustls()` / `default_native_tls()` shorthands for system-trust setups.
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(feature = "tls-rustls")]
    /// # mod demo {
    /// use std::error::Error;
    ///
    /// use ruststream_fred::stream::prelude::*;
    /// # use serde::Deserialize;
    /// # #[derive(Deserialize)]
    /// # struct Order {
    /// #     id: u64,
    /// # }
    /// #
    /// # #[subscriber(RedisStream::new("orders").group("workers"))]
    /// # async fn handle(order: &Order) -> HandlerOutcome {
    /// #     println!("order {}", order.id);
    /// #     HandlerOutcome::ack()
    /// # }
    ///
    /// pub async fn serve() -> Result<(), Box<dyn Error>> {
    ///     // System trust roots, no client certificate.
    ///     let tls = TlsConnector::default_rustls()?;
    ///     RustStream::new(AppInfo::new("orders", "0.1.0"))
    ///         .with_broker(RedisBroker::cluster(["10.0.0.1:6379"]).tls(tls), |b| {
    ///             b.include(handle);
    ///         })
    ///         .run()
    ///         .await?;
    ///     Ok(())
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    #[cfg(any(
        feature = "tls-rustls",
        feature = "tls-rustls-ring",
        feature = "tls-native-tls"
    ))]
    pub fn tls(mut self, tls: impl Into<TlsConfig>) -> Self {
        self.auth.tls = Some(tls.into());
        self
    }

    /// Sets distinct credentials for authenticating to the sentinel nodes, separate from the
    /// data-node [`credentials`](Self::credentials). Only meaningful on the sentinel topology.
    ///
    /// Available behind the `sentinel-auth` feature.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream_fred::stream::prelude::*;
    /// # use serde::Deserialize;
    /// # #[derive(Deserialize)]
    /// # struct Order {
    /// #     id: u64,
    /// # }
    /// #
    /// # #[subscriber(RedisStream::new("orders").group("workers"))]
    /// # async fn handle(order: &Order) -> HandlerOutcome {
    /// #     println!("order {}", order.id);
    /// #     HandlerOutcome::ack()
    /// # }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     // The data nodes and the sentinels each check their own pair.
    ///     let broker = RedisBroker::sentinel("mymaster", ["10.0.0.1:26379"])
    ///         .credentials("worker", "data-pass")
    ///         .sentinel_credentials("sentinel-user", "sentinel-pass");
    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
    ///         broker,
    ///         |b| {
    ///             b.include(handle);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    #[cfg(feature = "sentinel-auth")]
    pub fn sentinel_credentials(
        mut self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.auth.sentinel_username = Some(username.into());
        self.auth.sentinel_password = Some(password.into());
        self
    }

    /// Sets a password-only credential for authenticating to the sentinel nodes. Use
    /// [`sentinel_credentials`](Self::sentinel_credentials) for an ACL user plus password.
    ///
    /// Available behind the `sentinel-auth` feature.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream_fred::stream::prelude::*;
    /// # use serde::Deserialize;
    /// # #[derive(Deserialize)]
    /// # struct Order {
    /// #     id: u64,
    /// # }
    /// #
    /// # #[subscriber(RedisStream::new("orders").group("workers"))]
    /// # async fn handle(order: &Order) -> HandlerOutcome {
    /// #     println!("order {}", order.id);
    /// #     HandlerOutcome::ack()
    /// # }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     let broker = RedisBroker::sentinel("mymaster", ["10.0.0.1:26379"])
    ///         .password("data-pass")
    ///         .sentinel_password("sentinel-pass");
    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
    ///         broker,
    ///         |b| {
    ///             b.include(handle);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    #[cfg(feature = "sentinel-auth")]
    pub fn sentinel_password(mut self, password: impl Into<String>) -> Self {
        self.auth.sentinel_password = Some(password.into());
        self
    }

    /// Sets a dynamic credential provider that supplies (and can rotate) the username/password on
    /// each `AUTH` / `HELLO`, for IAM-style auth. Takes precedence over static
    /// [`credentials`](Self::credentials).
    ///
    /// Available behind the `credential-provider` feature.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use std::sync::Arc;
    ///
    /// use ruststream_fred::stream::prelude::*;
    /// # use serde::Deserialize;
    /// # use fred::error::Error as FredError;
    /// # use fred::types::config::Server;
    /// # use ruststream_fred::CredentialProvider;
    /// # #[derive(Debug)]
    /// # struct IamTokens;
    /// # #[async_trait::async_trait]
    /// # impl CredentialProvider for IamTokens {
    /// #     async fn fetch(
    /// #         &self,
    /// #         _server: Option<&Server>,
    /// #     ) -> Result<(Option<String>, Option<String>), FredError> {
    /// #         Ok((Some("worker".to_owned()), Some("token".to_owned())))
    /// #     }
    /// # }
    /// # #[derive(Deserialize)]
    /// # struct Order {
    /// #     id: u64,
    /// # }
    /// #
    /// # #[subscriber(RedisStream::new("orders").group("workers"))]
    /// # async fn handle(order: &Order) -> HandlerOutcome {
    /// #     println!("order {}", order.id);
    /// #     HandlerOutcome::ack()
    /// # }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     // `IamTokens` fetches a fresh token from the identity service on every `AUTH`.
    ///     let broker = RedisBroker::standalone("redis://localhost:6379")
    ///         .credential_provider(Arc::new(IamTokens));
    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
    ///         broker,
    ///         |b| {
    ///             b.include(handle);
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    #[cfg(feature = "credential-provider")]
    pub fn credential_provider(mut self, provider: Arc<dyn CredentialProvider>) -> Self {
        self.auth.credential_provider = Some(provider);
        self
    }

    /// Builds the `fred` config for this broker's topology, then folds in the auth/TLS settings.
    fn build_config(&self) -> Result<Config, RedisError> {
        let mut config = match &self.topology {
            Topology::Standalone(url) => {
                Config::from_url(url).map_err(|err| RedisError::Connect(Box::new(err)))?
            }
            Topology::Cluster(nodes) => {
                let hosts = parse_servers(nodes, 6379)?;
                Config {
                    server: ServerConfig::new_clustered(hosts),
                    ..Config::default()
                }
            }
            Topology::Sentinel { service, hosts } => {
                let hosts = parse_servers(hosts, 26379)?;
                Config {
                    server: ServerConfig::new_sentinel(hosts, service.clone()),
                    ..Config::default()
                }
            }
            // A caller-supplied pool carries the config it was built from; reusing it keeps the
            // Pub/Sub path (which dials a dedicated client per subscription) available.
            Topology::Preconnected(pool) => pool.next().client_config(),
        };
        self.apply_auth(&mut config);
        Ok(config)
    }

    /// Folds the recorded auth/TLS settings into `config`. Each setting is applied only when set,
    /// so credentials carried by a standalone URL survive unless explicitly overridden.
    fn apply_auth(&self, config: &mut Config) {
        if self.auth.username.is_some() {
            config.username.clone_from(&self.auth.username);
        }
        if self.auth.password.is_some() {
            config.password.clone_from(&self.auth.password);
        }
        #[cfg(any(
            feature = "tls-rustls",
            feature = "tls-rustls-ring",
            feature = "tls-native-tls"
        ))]
        if self.auth.tls.is_some() {
            config.tls.clone_from(&self.auth.tls);
        }
        #[cfg(feature = "credential-provider")]
        if self.auth.credential_provider.is_some() {
            config
                .credential_provider
                .clone_from(&self.auth.credential_provider);
        }
        #[cfg(feature = "sentinel-auth")]
        if let ServerConfig::Sentinel {
            username, password, ..
        } = &mut config.server
        {
            if self.auth.sentinel_username.is_some() {
                username.clone_from(&self.auth.sentinel_username);
            }
            if self.auth.sentinel_password.is_some() {
                password.clone_from(&self.auth.sentinel_password);
            }
        }
    }
}

impl Broker for RedisBroker {
    type Error = RedisError;
    type Connected = ConnectedRedisBroker;

    /// Opens (or adopts) the connection pool, consuming the unconnected form.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::Connect`] when the recorded topology cannot be turned into a `fred`
    /// config or the pool cannot reach the server.
    async fn connect(self) -> Result<Self::Connected, Self::Error> {
        let config = self.build_config()?;
        // Read off the resolved config rather than the recorded topology, so a caller-supplied
        // pool that is itself clustered is treated as the cluster it is.
        let clustered = config.server.is_clustered();
        let pool = if let Topology::Preconnected(pool) = &self.topology {
            pool.clone()
        } else {
            let pool = Pool::new(config.clone(), None, None, None, self.pool_size)
                .map_err(|err| RedisError::Connect(Box::new(err)))?;
            pool.init()
                .await
                .map_err(|err| RedisError::Connect(Box::new(err)))?;
            pool
        };
        Ok(ConnectedRedisBroker {
            core: Arc::new(RedisCore {
                connection: Connection::new(pool),
                closing: Arc::default(),
                runtime: Handle::current(),
                config,
                default_group: self.default_group,
                clustered,
                routes: Routes::default(),
                rounds: Arc::default(),
                loopback: Loopback::default(),
                #[cfg(feature = "testing")]
                opened: Mutex::default(),
            }),
        })
    }
}

/// The in-process mode: the production connected broker over an in-process Redis, the transport
/// a test runs the service's own app against.
///
/// Every setting of this broker is read the way [`Broker::connect`] reads it: the topology's
/// address is parsed (a URL `connect` refuses is refused here), the pool keeps its size, the
/// default group and the cluster flag carry over, and a cluster topology refuses what spans hash
/// slots. What the model answers is described in the crate's testing overview.
#[cfg(feature = "testing")]
impl InProcess for RedisBroker {
    async fn connect_in_process(self) -> Result<Self::Connected, Self::Error> {
        let config = self.build_config()?;
        let clustered = config.server.is_clustered();
        let pool_size = match &self.topology {
            Topology::Preconnected(pool) => pool.size(),
            _ => self.pool_size,
        };
        let (pool, config, server) = in_process::connect(&config, pool_size, clustered).await?;
        Ok(ConnectedRedisBroker {
            core: Arc::new(RedisCore {
                connection: Connection::new(pool),
                closing: Arc::default(),
                runtime: Handle::current(),
                config,
                default_group: self.default_group,
                clustered,
                routes: Routes::default(),
                rounds: Arc::default(),
                loopback: Loopback::to(server),
                opened: Mutex::default(),
            }),
        })
    }
}

#[cfg(feature = "testing")]
ruststream::register_testable_broker!(RedisBroker);

/// `DescribeServer` reports the host and port a client dials (the first seed for cluster and
/// sentinel). Nothing else a configured address may carry reaches the description: a broker built
/// from an already-connected pool describes no host at all.
impl DescribeServer for RedisBroker {
    fn describe_server(&self) -> ServerSpec {
        let host = match &self.topology {
            Topology::Standalone(url) => describe_address(url, 6379),
            Topology::Cluster(nodes) => nodes
                .first()
                .map_or_else(String::new, |node| describe_address(node, 6379)),
            Topology::Sentinel { hosts, .. } => hosts
                .first()
                .map_or_else(String::new, |host| describe_address(host, 26379)),
            // A caller-supplied pool carries its address inside `fred`'s own config, which this
            // description does not reach into.
            Topology::Preconnected(_) => String::new(),
        };
        let mut spec = ServerSpec::new(host, "redis");
        // An address this broker cannot name is left out of the document rather than published as
        // an empty string, which a reader takes for a coordinate and tries to dial.
        if spec.host.as_deref() == Some("") {
            spec.host = None;
        }
        spec
    }
}

/// The `host:port` a client dials, with everything else an address may carry stripped: the scheme,
/// any `user:password@`, the database path and the query.
///
/// [`ServerSpec::host_from_url`] does the stripping, so the rule that keeps a credential out of a
/// published `AsyncAPI` document lives in one place for every broker. What is added here is the
/// Redis default port, which the core cannot know.
fn describe_address(addr: &str, default_port: u16) -> String {
    let host = ServerSpec::host_from_url(addr.trim());
    if host.is_empty() {
        return String::new();
    }
    if has_port(&host) {
        host
    } else {
        format!("{host}:{default_port}")
    }
}

/// Whether an address already names a port, allowing for a bracketed IPv6 literal (`[::1]:6379`).
fn has_port(host: &str) -> bool {
    host.rsplit_once(']').map_or_else(
        || {
            host.rsplit_once(':')
                .is_some_and(|(_, port)| !port.is_empty())
        },
        |(_, after_bracket)| after_bracket.starts_with(':'),
    )
}

/// The live connection shared by the connected broker and every handle derived from it.
pub(crate) struct RedisCore {
    /// The pool, and whether [`ConnectedBroker::shutdown`] closed it. The ladder makes owner-side
    /// misuse a compile error, but publishers and deliveries handed out before the shutdown alias
    /// the connection and outlive it, so they ask it before issuing a command.
    connection: Arc<Connection>,
    /// The closes of the subscriptions' read connections still running, which the shutdown
    /// waits for before it closes the pool.
    closing: Arc<Closing>,
    /// The runtime `connect` ran on. Every task the broker starts for itself is spawned here, so
    /// a call from a dedicated thread's runtime (a handler under `threads(n)`) leaves no task
    /// behind that dies when that runtime stops.
    runtime: Handle,
    /// The config the pool was built from; Pub/Sub subscriptions dial their dedicated client
    /// from it.
    config: Config,
    default_group: Option<String>,
    /// Whether the connection is to a cluster, which is where a command spanning two keys needs
    /// them on one hash slot: a `MULTI` block cannot span slots at all, and a reliable list's
    /// claim is a `BLMOVE` between its queue and its processing list.
    clustered: bool,
    /// How each name this connection's subscriptions read or dead-letter to is written, for the
    /// default publisher.
    routes: Routes,
    /// The rounds of the deliveries in flight on this connection's pipelined subscriptions, which
    /// a reply or a bound slot publish joins.
    rounds: Arc<Rounds>,
    /// The in-process server, on a connection the test harness made in process.
    loopback: Loopback,
    /// The subscriptions opened on this connection, which the test harness asks where a publish
    /// is delivered.
    #[cfg(feature = "testing")]
    opened: Mutex<Vec<Opened>>,
}

/// What one subscription reads, for the harness's routing question.
#[cfg(feature = "testing")]
#[derive(Debug, Clone)]
enum Opened {
    /// A consumer of `group` on the stream `key`.
    Stream { key: String, group: String },
    /// A consumer of the list `key`.
    List { key: String },
    /// A subscription to one channel.
    Channel { name: String, sharded: bool },
    /// A subscription to every channel a glob matches.
    Pattern { glob: String },
}

impl RedisCore {
    /// The rounds a publish of this connection may join.
    pub(crate) fn rounds(&self) -> &Arc<Rounds> {
        &self.rounds
    }

    /// The route table the default publisher reads.
    pub(crate) const fn routes(&self) -> &Routes {
        &self.routes
    }

    /// Records a subscription's own name and its dead-letter destination with `route`.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::InvalidOptions`] when either name is already written another way.
    pub(crate) fn record_routes(
        &self,
        name: &str,
        dead_letter: Option<&str>,
        route: &Route,
    ) -> Result<Recorded<'_>, RedisError> {
        self.routes.record_subscription(name, dead_letter, route)
    }

    /// The live pool, or [`RedisError::ShutDown`] once the connection was torn down.
    pub(crate) fn pool(&self) -> Result<Pool, RedisError> {
        self.connection.live().cloned()
    }

    /// The shared connection a subscription hands its deliveries, or [`RedisError::ShutDown`]
    /// once it was torn down.
    pub(crate) fn connection(&self) -> Result<Arc<Connection>, RedisError> {
        self.connection.live()?;
        Ok(Arc::clone(&self.connection))
    }

    /// Cluster cannot offer multi-key transactions, because buffered keys may hash to different
    /// nodes, so its publishers reject `begin_transaction`.
    pub(crate) const fn transactions_supported(&self) -> bool {
        !self.clustered
    }

    pub(crate) const fn clustered(&self) -> bool {
        self.clustered
    }

    /// The in-process server, when the connection is one.
    pub(crate) const fn loopback(&self) -> &Loopback {
        &self.loopback
    }

    #[cfg(feature = "testing")]
    fn opened(&self, subscription: Opened) {
        self.opened
            .lock()
            .expect("redis subscription record poisoned")
            .push(subscription);
    }

    /// The version the server reported, or the in-process server's own.
    #[cfg_attr(
        not(feature = "testing"),
        allow(
            clippy::unused_self,
            reason = "the connection is read under `testing` only"
        )
    )]
    fn server_version(&self, pool: &Pool) -> Option<Version> {
        #[cfg(feature = "testing")]
        if self.loopback.server().is_some() {
            return Some(in_process::VERSION);
        }
        pool.server_version()
    }
}

impl std::fmt::Debug for RedisCore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisCore")
            .field("pool", self.connection.pool())
            .field("closed", &self.connection.is_closed())
            .finish_non_exhaustive()
    }
}

/// The typed witness that [`Broker::connect`] succeeded: it owns the live `fred` pool.
///
/// Every subscription and every publisher is reached from here, so "not connected" is not
/// representable. [`ConnectedBroker::shutdown`] consumes it, which makes a publish or subscribe
/// after shutdown a compile error for the owner of the handle.
#[derive(Debug)]
pub struct ConnectedRedisBroker {
    core: Arc<RedisCore>,
}

impl ConnectedRedisBroker {
    /// Opens a stream subscription described by `def`.
    ///
    /// Ensures the consumer group exists (`XGROUP CREATE ... MKSTREAM`, ignoring an
    /// already-existing group) before returning the subscriber.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::ShutDown`] when the connection was already torn down,
    /// [`RedisError::InvalidOptions`] when `def` names no consumer group, or
    /// [`RedisError::Subscribe`] when the group cannot be created.
    pub async fn subscribe(&self, def: RedisStream) -> Result<RedisSubscriber, RedisError> {
        let connection = self.core.connection()?;
        let pool = connection.pool();
        let group = def.group_or_err()?.to_owned();
        let consumer = def.consumer_or_auto();
        if let ReadMode::Claiming { .. } = def.mode() {
            require_claim_support(self.core.server_version(pool).as_ref(), def.key())?;
        }
        let recorded = self
            .core
            .record_routes(def.key(), def.dead_letter(), &Route::Stream)?;
        ensure_group(pool, def.key(), &group, def.start().as_id()).await?;
        let reader = self.read_connection().await?;
        let delay = def.delay_config().map(|cfg| cfg.on(self.core.loopback()));
        let tap = self.stream_tap(&def, &group, &consumer, delay.as_ref());
        recorded.keep();
        Ok(RedisSubscriber::new(
            connection,
            reader,
            def.key().to_owned(),
            group,
            consumer,
            def.block_or_default(),
            def.mode(),
            delay,
            tap,
        ))
    }

    /// Registers a stream subscription with the in-process server, and records it for the
    /// harness's routing question.
    #[cfg_attr(
        not(feature = "testing"),
        allow(
            clippy::unused_self,
            reason = "the connection is read under `testing` only"
        )
    )]
    fn stream_tap(
        &self,
        def: &RedisStream,
        group: &str,
        consumer: &str,
        delay: Option<&DelayConfig>,
    ) -> Tap {
        #[cfg(feature = "testing")]
        {
            self.core.opened(Opened::Stream {
                key: def.key().to_owned(),
                group: group.to_owned(),
            });
            if let Some(server) = self.core.loopback().server() {
                let (claim, fresh) = match def.mode() {
                    ReadMode::Fresh => (None, true),
                    ReadMode::Reclaim { min_idle } => (Some(min_idle), false),
                    ReadMode::Claiming { min_idle } => (Some(min_idle), true),
                };
                let reader = server.attach_stream(
                    def.key(),
                    group,
                    consumer,
                    claim,
                    fresh,
                    delay.map(DelayConfig::zset_key),
                );
                return Tap::registered(Arc::clone(server), reader);
            }
            Tap::default()
        }
        #[cfg(not(feature = "testing"))]
        {
            let _ = (def, group, consumer, delay);
            Tap::default()
        }
    }

    /// Opens a Pub/Sub subscription on one channel, described by `def`, on a dedicated client.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::ShutDown`] when the connection was already torn down,
    /// [`RedisError::Connect`] when the dedicated client cannot connect, or
    /// [`RedisError::Subscribe`] when the subscribe command fails.
    pub async fn subscribe_pubsub(
        &self,
        def: RedisPubSub,
    ) -> Result<RedisPubSubSubscriber, RedisError> {
        Ok(RedisPubSubSubscriber::new(self.open_pubsub(def).await?))
    }

    /// Subscribes a dedicated client to the channel `def` names and hands back its wire.
    pub(crate) async fn open_pubsub(&self, def: RedisPubSub) -> Result<PubSubWire, RedisError> {
        let codec = def.codec_handle();
        let recorded = self
            .core
            .record_routes(def.channel(), def.dead_letter(), &def.route())?;
        let client = self.new_client(def.buffer_size()).await?;
        // Opened before the subscribe, because the messages arrive over a broadcast channel whose
        // receiver sees only what is sent after it exists. The connection is dedicated to this one
        // subscription, so opening the stream early can pick up nothing else.
        let rx = client.message_rx();
        let channel = def.channel().to_owned();
        let sharded = def.delivery_mode() == PubSubMode::Sharded;
        match def.delivery_mode() {
            PubSubMode::Classic => {
                client
                    .subscribe(channel)
                    .await
                    .map_err(RedisError::subscribe)?;
                confirm_subscribed(&client).await?;
            }
            // `SSUBSCRIBE` is answered by the server, so `fred` waits for it and there is nothing
            // left to confirm here.
            PubSubMode::Sharded => {
                client
                    .ssubscribe(channel)
                    .await
                    .map_err(RedisError::subscribe)?;
            }
        }
        let (rx, tap) = self.pubsub_tap(
            rx,
            PubSubTarget::Channel {
                name: def.channel(),
                sharded,
            },
            def.buffer_size(),
        );
        let pool = self.core.pool()?;
        recorded.keep();
        Ok(PubSubWire::new(
            def.channel().to_owned(),
            client,
            rx,
            codec,
            pool,
            tap,
            self.runtime().clone(),
        ))
    }

    /// Registers a Pub/Sub subscription with the in-process server, whose messages then arrive
    /// on a channel of its own instead of the client's, and records it for the harness's routing
    /// question.
    #[cfg_attr(
        not(feature = "testing"),
        allow(
            clippy::unused_self,
            reason = "the connection is read under `testing` only"
        )
    )]
    fn pubsub_tap(
        &self,
        rx: Receiver<Message>,
        target: PubSubTarget<'_>,
        buffer: Option<NonZeroUsize>,
    ) -> (Receiver<Message>, Tap) {
        #[cfg(feature = "testing")]
        {
            let (opened, subscription) = match target {
                PubSubTarget::Channel { name, sharded } => {
                    let bytes = bytes::Bytes::copy_from_slice(name.as_bytes());
                    (
                        Opened::Channel {
                            name: name.to_owned(),
                            sharded,
                        },
                        if sharded {
                            Subscription::Sharded(bytes)
                        } else {
                            Subscription::Channel(bytes)
                        },
                    )
                }
                PubSubTarget::Pattern(glob) => (
                    Opened::Pattern {
                        glob: glob.to_owned(),
                    },
                    Subscription::Pattern(bytes::Bytes::copy_from_slice(glob.as_bytes())),
                ),
            };
            self.core.opened(opened);
            if let Some(server) = self.core.loopback().server() {
                let (rx, reader) = server.attach_pubsub(subscription, buffer);
                return (rx, Tap::registered(Arc::clone(server), reader));
            }
            (rx, Tap::default())
        }
        #[cfg(not(feature = "testing"))]
        {
            let _ = (target, buffer);
            (rx, Tap::default())
        }
    }

    /// Opens a Pub/Sub subscription on a glob (`PSUBSCRIBE`), described by `def`, on a dedicated
    /// client.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::ShutDown`] when the connection was already torn down,
    /// [`RedisError::Connect`] when the dedicated client cannot connect, or
    /// [`RedisError::Subscribe`] when the subscribe command fails.
    pub async fn subscribe_pubsub_pattern(
        &self,
        def: RedisPubSubPattern,
    ) -> Result<RedisPubSubSubscriber, RedisError> {
        Ok(RedisPubSubSubscriber::new(
            self.open_pubsub_pattern(def).await?,
        ))
    }

    /// Subscribes a dedicated client to the glob `def` names and hands back its wire.
    pub(crate) async fn open_pubsub_pattern(
        &self,
        def: RedisPubSubPattern,
    ) -> Result<PubSubWire, RedisError> {
        let codec = def.codec_handle();
        let client = self.new_client(def.buffer_size()).await?;
        // Opened before the subscribe, for the reason `subscribe_pubsub` gives.
        let rx = client.message_rx();
        client
            .psubscribe(def.pattern().to_owned())
            .await
            .map_err(RedisError::subscribe)?;
        confirm_subscribed(&client).await?;
        let (rx, tap) =
            self.pubsub_tap(rx, PubSubTarget::Pattern(def.pattern()), def.buffer_size());
        Ok(PubSubWire::new(
            def.pattern().to_owned(),
            client,
            rx,
            codec,
            self.core.pool()?,
            tap,
            self.runtime().clone(),
        ))
    }

    /// Opens a list (work-queue) subscription described by `def`.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::ShutDown`] when the connection was already torn down,
    /// [`RedisError::InvalidOptions`] when `def` names a recovery ZSET without a `min_idle`, or
    /// [`RedisError::Connect`] when the connection the subscription reads on cannot connect.
    pub async fn subscribe_list(&self, def: RedisList) -> Result<RedisListSubscriber, RedisError> {
        self.open_list(def).await.map(RedisListSubscriber::new)
    }

    /// Validates `def` against this connection and hands back its wire.
    pub(crate) async fn open_list(&self, def: RedisList) -> Result<ListWire, RedisError> {
        let connection = self.core.connection()?;
        let recovery = def
            .recovery_config()?
            .map(|cfg| cfg.on(self.core.loopback()));
        let reliable = def.is_reliable();
        let processing = def.processing_or_default();
        if reliable && self.core.clustered() {
            require_one_slot(def.key(), &processing)?;
        }
        self.core
            .record_routes(def.key(), def.dead_letter(), &def.route())?
            .keep();
        let block = def.block_or_default();
        let codec = def.codec_handle();
        let reader = self.read_connection().await?;
        let tap = self.list_tap(def.key(), recovery.as_ref());
        Ok(ListWire::new(
            connection,
            reader,
            def.into_key(),
            reliable,
            processing,
            block,
            codec,
            recovery,
            tap,
        ))
    }

    /// Registers a list subscription with the in-process server, and records it for the
    /// harness's routing question.
    #[cfg_attr(
        not(feature = "testing"),
        allow(
            clippy::unused_self,
            reason = "the connection is read under `testing` only"
        )
    )]
    fn list_tap(&self, key: &str, recovery: Option<&RecoveryConfig>) -> Tap {
        #[cfg(feature = "testing")]
        {
            self.core.opened(Opened::List {
                key: key.to_owned(),
            });
            if let Some(server) = self.core.loopback().server() {
                let reader = server.attach_list(
                    key,
                    recovery.map(|cfg| (cfg.zset_key.as_str(), cfg.min_idle)),
                );
                return Tap::registered(Arc::clone(server), reader);
            }
            Tap::default()
        }
        #[cfg(not(feature = "testing"))]
        {
            let _ = (key, recovery);
            Tap::default()
        }
    }

    /// Returns a stream publisher (`XADD`) bound to this connection.
    #[must_use]
    pub fn publisher(&self) -> RedisPublisher {
        RedisPublisher::new(Arc::clone(&self.core))
    }

    /// Returns the publisher of the broker's default policy, [`RedisDefaultPublish`], which
    /// writes each name the way this connection's subscriptions read it.
    ///
    /// [`RedisDefaultPublish`]: crate::RedisDefaultPublish
    #[must_use]
    pub fn default_publisher(&self) -> RedisDefaultPublisher {
        RedisDefaultPublisher::new(Arc::clone(&self.core))
    }

    /// Returns a Pub/Sub publisher configured by `publish` (mode and envelope codec).
    #[must_use]
    pub fn pubsub_publisher(&self, publish: RedisPubSubPublish) -> RedisPubSubPublisher {
        RedisPubSubPublisher::new(Arc::clone(&self.core), publish)
    }

    /// Returns a list publisher (`LPUSH`) configured by `publish` (envelope codec and key TTL).
    #[must_use]
    pub fn list_publisher(&self, publish: RedisListPublish) -> RedisListPublisher {
        RedisListPublisher::new(Arc::clone(&self.core), publish)
    }

    /// The rounds a publish of this connection may join.
    pub(crate) fn rounds(&self) -> &Arc<Rounds> {
        self.core.rounds()
    }

    /// The runtime this connection was opened on, where the broker spawns its own tasks.
    pub(crate) fn runtime(&self) -> &Handle {
        &self.core.runtime
    }

    /// What a pipelined subscription's window takes from this connection, or
    /// [`RedisError::ShutDown`] once it was torn down.
    pub(crate) fn window_owner(&self) -> Result<Owner, RedisError> {
        Ok(Owner {
            rounds: Arc::clone(self.rounds()),
            runtime: self.runtime().clone(),
            connection: self.core.connection()?,
        })
    }

    /// Returns a clone of the underlying pool, for advanced operations not covered by the
    /// wrapper.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::ShutDown`] once the connection was torn down.
    pub fn pool_handle(&self) -> Result<Pool, RedisError> {
        self.core.pool()
    }

    /// The connection a stream or list subscription reads on, apart from the pool its publishes
    /// and settlements go out on.
    async fn read_connection(&self) -> Result<ReadConnection, RedisError> {
        #[cfg(feature = "testing")]
        let blocks = self.core.loopback().server().is_none();
        #[cfg(not(feature = "testing"))]
        let blocks = true;
        Ok(ReadConnection::new(
            self.new_client(None).await?,
            self.runtime().clone(),
            Arc::clone(&self.core.closing),
            blocks,
        ))
    }

    /// Builds and connects a dedicated `fred` client (used for Pub/Sub, which needs an isolated
    /// message stream and channel state per subscriber, and for the reads of a stream or list
    /// subscription).
    ///
    /// `buffer` is how many messages its Pub/Sub channel holds for a reader that has not taken them
    /// yet; `None` leaves `fred`'s own setting.
    async fn new_client(&self, buffer: Option<NonZeroUsize>) -> Result<Client, RedisError> {
        // The dedicated client is a second connection to the same server: refuse to dial one for
        // a connection whose owner already shut down.
        let _ = self.core.pool()?;
        let performance = buffer.map(|buffer| PerformanceConfig {
            broadcast_channel_capacity: buffer.get(),
            ..PerformanceConfig::default()
        });
        let client = Client::new(self.core.config.clone(), performance, None, None);
        // `init` spawns the task that drives the connection on the runtime it is polled on; it is
        // polled on the broker's own runtime, so the connection outlives a caller's runtime. The
        // price is one task spawned per subscription opened.
        let connecting = client.clone();
        self.core
            .runtime
            .spawn(async move { connecting.init().await })
            .await
            .map_err(|err| RedisError::Connect(Box::new(err)))?
            .map_err(|err| RedisError::Connect(Box::new(err)))?;
        Ok(client)
    }
}

impl ConnectedBroker for ConnectedRedisBroker {
    type Error = RedisError;
    type Closed = ClosedRedisBroker;

    /// Closes every pooled connection and marks the shared connection dead, so publishers handed
    /// out earlier report [`RedisError::ShutDown`] instead of running against a closed pool.
    ///
    /// # Errors
    ///
    /// Returns [`RedisError::Connect`] when the `QUIT` roundtrip fails.
    async fn shutdown(self) -> Result<Self::Closed, Self::Error> {
        self.core.connection.close();
        self.core.closing.finish().await;
        let connections_closed = self.core.connection.pool().clients().len();
        self.core
            .connection
            .pool()
            .quit()
            .await
            .map_err(|err| RedisError::Connect(Box::new(err)))?;
        // The in-process server goes down with the connection, after the `QUIT` it answers.
        #[cfg(feature = "testing")]
        if let Some(server) = self.core.loopback().server() {
            server.close();
        }
        Ok(ClosedRedisBroker { connections_closed })
    }
}

/// What a Pub/Sub subscription listens to, as its registration names it.
#[cfg_attr(
    not(feature = "testing"),
    allow(dead_code, reason = "read by the in-process registration only")
)]
#[derive(Clone, Copy)]
enum PubSubTarget<'a> {
    Channel { name: &'a str, sharded: bool },
    Pattern(&'a str),
}

/// The harness's view of the connection: the in-process server it injects into and reads the
/// writes of, and, in both modes, where a publish is delivered.
///
/// # Panics
///
/// `inject` and `published` panic on a broker connected with `connect`: the harness drives only
/// the connection `connect_in_process` produced, and a live connection has no log to read and no
/// synchronous way to take a message.
#[cfg(feature = "testing")]
impl TestableBroker for ConnectedRedisBroker {
    fn install_coordinator(&self, coordinator: Coordinator) {
        if let Some(server) = self.core.loopback().server() {
            server.install(coordinator);
        }
    }

    /// Writes the message the way the default publisher writes its name: an `XADD`, or the
    /// `LPUSH` or `PUBLISH` of the subscription that reads it, in that subscription's framing. A
    /// name only pattern subscriptions read goes out with a raw `PUBLISH`, as the producers they
    /// listen to publish it: an `XADD` would reach none of them.
    fn inject(&self, message: OutgoingMessage<'_>) {
        let server = self.in_process("inject");
        let name = message.name();
        let (payload, headers) = (message.payload(), message.headers());
        let written = match self.core.routes().route(name) {
            Route::Stream if self.read_by_patterns_alone(name) => {
                server.inject_publish(name, frame(None, payload, headers), false)
            }
            Route::Stream => {
                server.inject_stream(name, fields_for_publish(payload.to_vec(), headers))
            }
            Route::List { envelope } => {
                server.inject_list(name, frame(envelope.as_ref(), payload, headers))
            }
            Route::Channel { mode, envelope } => server.inject_publish(
                name,
                frame(envelope.as_ref(), payload, headers),
                mode == PubSubMode::Sharded,
            ),
        };
        if let Err(err) = written {
            panic!("the injected message to {name:?} is not one Redis takes: {err}");
        }
    }

    /// Every write to `name`, decoded the way the subscription reading it decodes: an entry's
    /// fields, or a list element or a channel message in its subscription's framing. A requeue's
    /// copy and a delayed entry the sweep adds back are writes too.
    fn published(&self, name: &str) -> Vec<RawMessage> {
        let server = self.in_process("published");
        let envelope = match self.core.routes().route(name) {
            Route::Stream => None,
            Route::List { envelope } | Route::Channel { envelope, .. } => envelope,
        };
        server
            .published(name)
            .into_iter()
            .map(|write| {
                let (payload, headers) = match write {
                    Write::Fields(fields) => parts_from_fields(
                        fields
                            .into_iter()
                            .map(|(field, value)| {
                                (String::from_utf8_lossy(&field).into_owned(), value.to_vec())
                            })
                            .collect(),
                    ),
                    Write::Body(body) => unframe(envelope.as_ref(), &body),
                };
                RawMessage::new(name.to_owned(), payload).with_headers(headers)
            })
            .collect()
    }

    /// Redis's routing: a stream entry is delivered once to every consumer group reading the
    /// stream (to one consumer of each), a list element to one consumer, and a channel message to
    /// every subscription of the channel and every pattern subscription whose glob matches it. A
    /// name no subscription reads exactly is reached by the pattern subscriptions alone.
    ///
    /// Subscriptions sharing a name are told apart by nothing but the name, so a publish is owed
    /// by the first position of each name as many times as that name receives it. In process
    /// that is where it goes: a group's new entries go to its earliest open subscription.
    fn routes(&self, destination: &str, subscriptions: &[&str]) -> Vec<usize> {
        let opened = self
            .core
            .opened
            .lock()
            .expect("redis subscription record poisoned")
            .clone();
        let mut owed = Vec::new();
        let mut owe = |name: &str, times: usize| {
            if let Some(at) = subscriptions.iter().position(|held| *held == name) {
                owed.extend(std::iter::repeat_n(at, times));
            }
        };
        let groups: std::collections::BTreeSet<&str> = opened
            .iter()
            .filter_map(|sub| match sub {
                Opened::Stream { key, group } if key == destination => Some(group.as_str()),
                _ => None,
            })
            .collect();
        let patterns = match self.core.routes().route(destination) {
            Route::Stream => {
                owe(destination, groups.len());
                groups.is_empty()
            }
            Route::List { .. } => {
                let read = opened
                    .iter()
                    .any(|sub| matches!(sub, Opened::List { key } if key == destination));
                owe(destination, usize::from(read));
                false
            }
            Route::Channel { mode, .. } => {
                let sharded = mode == PubSubMode::Sharded;
                let reached = opened
                    .iter()
                    .filter(|sub| {
                        matches!(sub, Opened::Channel { name, sharded: held }
                            if name == destination && *held == sharded)
                    })
                    .count();
                owe(destination, reached);
                !sharded
            }
        };
        if patterns {
            for sub in &opened {
                if let Opened::Pattern { glob } = sub
                    && in_process::glob_matches(glob.as_bytes(), destination.as_bytes())
                {
                    owe(glob, 1);
                }
            }
        }
        owed
    }
}

#[cfg(feature = "testing")]
impl ConnectedRedisBroker {
    /// Whether `name` is read by a pattern subscription and by no stream group: what the harness
    /// then publishes to the channel instead of adding to a stream.
    fn read_by_patterns_alone(&self, name: &str) -> bool {
        let opened = self
            .core
            .opened
            .lock()
            .expect("redis subscription record poisoned");
        let grouped = opened
            .iter()
            .any(|sub| matches!(sub, Opened::Stream { key, .. } if key == name));
        !grouped
            && opened.iter().any(|sub| {
                matches!(sub, Opened::Pattern { glob }
                    if in_process::glob_matches(glob.as_bytes(), name.as_bytes()))
            })
    }

    /// The in-process server, which is all the harness injects into and reads.
    fn in_process(&self, what: &str) -> &Arc<in_process::Server> {
        self.core.loopback().server().unwrap_or_else(|| {
            panic!(
                "TestableBroker::{what} reached a broker connected with `connect`; the harness \
                 drives the connection `connect_in_process` produces"
            )
        })
    }
}

/// The terminal witness returned by shutting down a [`ConnectedRedisBroker`].
///
/// It has no publish or subscribe surface; it carries the teardown diagnostics as plain data.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ClosedRedisBroker {
    connections_closed: usize,
}

impl ClosedRedisBroker {
    /// How many pooled connections the teardown closed.
    #[must_use]
    pub const fn connections_closed(&self) -> usize {
        self.connections_closed
    }
}

// By-name subscription capability for the bare string `#[subscriber("key")]` form. Redis Streams
// always read through a consumer group, so this requires a broker-wide default group.
#[allow(
    clippy::use_self,
    reason = "the type name disambiguates the inherent subscribe from this trait method"
)]
impl Subscribe for ConnectedRedisBroker {
    type Subscriber = RedisSubscriber;
    /// A bare name opens a consumer group over that stream key, and an `XADD` there is read by the
    /// group again, so the name is both ends and a retry copy needs nothing from the mount site.
    type Copies = AddressedCopies;

    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error> {
        let group = self
            .core
            .default_group
            .clone()
            .ok_or_else(|| missing_default_group(name, "RedisBroker"))?;
        ConnectedRedisBroker::subscribe(self, RedisStream::new(name).group(group)).await
    }
}

/// The refusal of a bare-string subscription on a broker with no default group: Redis Streams read
/// through a consumer group, and a bare name names none. `broker` is the type whose
/// `default_group` setting supplies one.
pub(crate) fn missing_default_group(name: &str, broker: &str) -> RedisError {
    RedisError::InvalidOptions(format!(
        "bare-string subscription on `{name}` needs a broker-wide default group: call \
         {broker}::default_group(name), or subscribe with RedisStream::new(name).group(group)"
    ))
}

/// Refuses a [`RedisStream::claiming`] subscription the connected server cannot serve.
///
/// The version is the one the `fred` client read from `INFO server` during its handshake, so the
/// check costs no round trip and runs once, where the subscription resolves against the connected
/// broker. Without it the mistake surfaces on the first read, as a bare syntax error from a server
/// that does not know the option.
///
/// A server that reports no version is refused the same way: the option either exists or the
/// service does not start, and a guess is what the check is here to avoid.
fn require_claim_support(version: Option<&Version>, key: &str) -> Result<(), RedisError> {
    match version {
        Some(version) if *version >= CLAIM_MIN_VERSION => Ok(()),
        Some(version) => Err(RedisError::ServerTooOld(format!(
            "RedisStream::claiming on {key:?} needs Redis {CLAIM_MIN_VERSION} or later \
             (XREADGROUP CLAIM); the connected server is {version}"
        ))),
        None => Err(RedisError::ServerTooOld(format!(
            "RedisStream::claiming on {key:?} needs Redis {CLAIM_MIN_VERSION} or later \
             (XREADGROUP CLAIM); the connected server reports no version"
        ))),
    }
}

/// Waits for the server to apply the `SUBSCRIBE` or `PSUBSCRIBE` the client just issued.
///
/// `fred` answers both the moment the frame is queued: Redis confirms them out of band, on the
/// message stream, and the client drops that confirmation without surfacing it. A publish issued
/// the instant `subscribe(..)` returned therefore reaches a server that has not routed the channel
/// here yet, and Pub/Sub keeps nothing for a subscriber who is not there: the message is gone,
/// with no error on either side. That is a lost delivery on a subscription the caller was told was
/// open, so it is worth the round trip this costs once per subscription.
///
/// Redis serves one connection's commands in order, so a reply to a command sent after the
/// subscribe cannot come back before the subscribe was applied. The `PING` carries the hash `fred`
/// pins the legacy Pub/Sub commands to, so on a cluster it is answered by the node the subscribe
/// reached rather than by another one.
async fn confirm_subscribed(client: &Client) -> Result<(), RedisError> {
    let node = ClusterHash::Custom(redis_keyslot(client.id().as_bytes()));
    let _: Value = client
        .custom(CustomCommand::new("PING", node, false), Vec::<Value>::new())
        .await
        .map_err(RedisError::subscribe)?;
    Ok(())
}

/// Refuses a reliable list on a cluster whose two keys live on different nodes.
///
/// A claim is a `BLMOVE` from the queue to the processing list, and Redis refuses a command that
/// spans hash slots. Nothing says so until the first delivery, so the subscription is refused
/// where it is opened and the message names the hash tag that puts both keys on one node.
fn require_one_slot(key: &str, processing: &str) -> Result<(), RedisError> {
    if redis_keyslot(key.as_bytes()) == redis_keyslot(processing.as_bytes()) {
        return Ok(());
    }
    Err(RedisError::InvalidOptions(format!(
        "a reliable list on `{key}` cannot run on a cluster while its processing list \
         `{processing}` hashes to another slot: a claim moves the entry between the two in one \
         command. Put both on one slot with a hash tag, as in RedisList::new(\"{{{key}}}\"), which \
         the default processing key follows"
    )))
}

/// Creates the consumer group, treating an already-existing group as success.
async fn ensure_group(
    pool: &Pool,
    key: &str,
    group: &str,
    start_id: &str,
) -> Result<(), RedisError> {
    let result: Result<String, fred::error::Error> =
        pool.xgroup_create(key, group, start_id, true).await;
    match result {
        Ok(_) => Ok(()),
        // BUSYGROUP: the group already exists, which is the steady-state case.
        Err(err) if err.details().contains("BUSYGROUP") => Ok(()),
        Err(err) => Err(RedisError::subscribe(err)),
    }
}

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

    /// The refusal names the subscription, the mode and both versions, so the operator reads
    /// what to change without reaching for the source.
    #[test]
    fn a_claiming_subscription_names_both_versions_when_it_refuses() {
        let err = require_claim_support(Some(&Version::new(7, 4, 2)), "orders")
            .expect_err("a server below 8.4.0 must be refused");
        assert_eq!(
            err.to_string(),
            "RedisStream::claiming on \"orders\" needs Redis 8.4.0 or later (XREADGROUP CLAIM); \
             the connected server is 7.4.2"
        );
    }

    /// A server that reports no version is refused too: the option either exists or the service
    /// does not start.
    #[test]
    fn a_server_with_no_version_is_refused() {
        let err = require_claim_support(None, "orders")
            .expect_err("a server reporting no version must be refused");
        assert!(
            err.to_string().contains("reports no version"),
            "the refusal has to say the version is missing: {err}"
        );
    }

    /// 8.4.0 is the floor, not a version to be above, and a later release keeps the option.
    #[test]
    fn the_floor_release_and_later_are_accepted() {
        for version in [
            Version::new(8, 4, 0),
            Version::new(8, 4, 6),
            Version::new(9, 0, 0),
        ] {
            assert!(
                require_claim_support(Some(&version), "orders").is_ok(),
                "{version} has XREADGROUP CLAIM"
            );
        }
        for version in [Version::new(8, 3, 9), Version::new(7, 4, 2)] {
            assert!(
                require_claim_support(Some(&version), "orders").is_err(),
                "{version} does not have XREADGROUP CLAIM"
            );
        }
    }

    #[test]
    fn describe_server_reports_redis() {
        let broker = RedisBroker::standalone("redis://localhost:6379");
        let spec = broker.describe_server();
        assert_eq!(spec.protocol, "redis");
        assert_eq!(spec.host.as_deref(), Some("localhost:6379"));
    }

    /// Every address shape this crate accepts reduces to the host and the port a client dials.
    #[test]
    fn a_description_is_the_host_and_port_of_every_address_shape() {
        let cases = [
            ("redis://localhost:6379", "localhost:6379"),
            ("rediss://redis.example.com:6380", "redis.example.com:6380"),
            ("redis://localhost", "localhost:6379"),
            ("redis://localhost:6379/3", "localhost:6379"),
            ("redis://localhost:6379/?timeout=5s", "localhost:6379"),
            (
                "valkeys://valkey.example.com:6380",
                "valkey.example.com:6380",
            ),
            ("10.0.0.1:6379", "10.0.0.1:6379"),
            ("  redis://localhost:6379  ", "localhost:6379"),
            ("redis://[::1]:6379", "[::1]:6379"),
            ("redis://[::1]", "[::1]:6379"),
        ];

        for (url, expected) in cases {
            let spec = RedisBroker::standalone(url).describe_server();
            assert_eq!(
                spec.host.as_deref(),
                Some(expected),
                "the description of {url:?} is wrong"
            );
        }

        let cluster = RedisBroker::cluster(["10.0.0.1", "10.0.0.2:7001"]).describe_server();
        assert_eq!(cluster.host.as_deref(), Some("10.0.0.1:6379"));

        let sentinel = RedisBroker::sentinel("mymaster", ["10.0.0.9"]).describe_server();
        assert_eq!(sentinel.host.as_deref(), Some("10.0.0.9:26379"));
    }

    /// The description ends up in the generated `AsyncAPI` document, which teams publish, so a URL
    /// carrying credentials must leave neither the user name nor the password behind. The password
    /// here holds an `@` of its own, the case a split on the first separator gets wrong.
    #[test]
    fn a_description_never_carries_the_credentials_of_a_url() {
        let cases = [
            (
                RedisBroker::standalone("redis://alice:p@ss@redis.example.com:6379/0"),
                "redis.example.com:6379",
            ),
            (
                RedisBroker::cluster(["redis://alice:p@ss@10.0.0.1:7000"]),
                "10.0.0.1:7000",
            ),
            (
                RedisBroker::sentinel("mymaster", ["redis://alice:p@ss@10.0.0.9:26379"]),
                "10.0.0.9:26379",
            ),
        ];

        for (broker, expected) in cases {
            let host = broker.describe_server().host.expect("a described host");
            assert_eq!(host, expected, "the description kept more than the address");
            assert!(
                !host.contains('@'),
                "{host:?} still carries a userinfo part"
            );
            assert!(
                !host.contains("alice"),
                "{host:?} still carries the user name"
            );
            assert!(
                !host.contains("p@ss"),
                "{host:?} still carries the password"
            );
        }
    }

    // Credentials must reach the fred config on every topology, not just the standalone URL.
    #[test]
    fn credentials_apply_to_all_topologies() {
        let brokers = [
            RedisBroker::standalone("redis://localhost:6379").credentials("alice", "s3cr3t"),
            RedisBroker::cluster(["127.0.0.1:7000"]).credentials("alice", "s3cr3t"),
            RedisBroker::sentinel("mymaster", ["127.0.0.1:26379"]).credentials("alice", "s3cr3t"),
        ];
        for broker in brokers {
            let config = broker.build_config().expect("config builds");
            assert_eq!(config.username.as_deref(), Some("alice"));
            assert_eq!(config.password.as_deref(), Some("s3cr3t"));
        }
    }

    #[test]
    fn password_only_sets_password_without_username() {
        let config = RedisBroker::cluster(["127.0.0.1:7000"])
            .password("requirepass")
            .build_config()
            .expect("config builds");
        assert_eq!(config.username, None);
        assert_eq!(config.password.as_deref(), Some("requirepass"));
    }

    // Programmatic credentials win over a standalone URL's userinfo.
    #[test]
    fn programmatic_credentials_override_standalone_url() {
        let config = RedisBroker::standalone("redis://urluser:urlpass@localhost:6379")
            .credentials("acluser", "aclpass")
            .build_config()
            .expect("config builds");
        assert_eq!(config.username.as_deref(), Some("acluser"));
        assert_eq!(config.password.as_deref(), Some("aclpass"));
    }

    // Without an override the URL's credentials are left untouched.
    #[test]
    fn url_credentials_preserved_without_override() {
        let config = RedisBroker::standalone("redis://urluser:urlpass@localhost:6379")
            .build_config()
            .expect("config builds");
        assert_eq!(config.username.as_deref(), Some("urluser"));
        assert_eq!(config.password.as_deref(), Some("urlpass"));
    }

    #[test]
    fn debug_redacts_password() {
        let broker =
            RedisBroker::standalone("redis://localhost:6379").credentials("alice", "s3cr3t");
        let rendered = format!("{broker:?}");
        assert!(
            !rendered.contains("s3cr3t"),
            "password must not appear in Debug output: {rendered}"
        );
        // The username is an identifier, not a secret, and is kept for debugging.
        assert!(
            rendered.contains("alice"),
            "expected username in: {rendered}"
        );
    }

    #[cfg(feature = "sentinel-auth")]
    #[test]
    fn sentinel_credentials_apply_to_sentinel_server() {
        let config = RedisBroker::sentinel("mymaster", ["127.0.0.1:26379"])
            .credentials("datauser", "datapass")
            .sentinel_credentials("sentineluser", "sentinelpass")
            .build_config()
            .expect("config builds");
        // Data-node credentials sit on the top-level config.
        assert_eq!(config.username.as_deref(), Some("datauser"));
        let ServerConfig::Sentinel {
            username, password, ..
        } = &config.server
        else {
            panic!("expected a sentinel server config");
        };
        assert_eq!(username.as_deref(), Some("sentineluser"));
        assert_eq!(password.as_deref(), Some("sentinelpass"));
    }

    #[cfg(feature = "credential-provider")]
    #[derive(Debug)]
    struct StaticCredentials;

    #[cfg(feature = "credential-provider")]
    #[async_trait::async_trait]
    impl CredentialProvider for StaticCredentials {
        async fn fetch(
            &self,
            _server: Option<&fred::types::config::Server>,
        ) -> Result<(Option<String>, Option<String>), fred::error::Error> {
            Ok((Some("rotating".into()), Some("token".into())))
        }
    }

    #[cfg(feature = "credential-provider")]
    #[test]
    fn credential_provider_is_applied() {
        let provider: Arc<dyn CredentialProvider> = Arc::new(StaticCredentials);
        let config = RedisBroker::cluster(["127.0.0.1:7000"])
            .credential_provider(provider)
            .build_config()
            .expect("config builds");
        assert!(config.credential_provider.is_some());
    }
}