ruststream_fred/

pubsub.rs

//! Redis Pub/Sub transport: fire-and-forget fan-out with no acknowledgement.
//!
//! Unlike Streams, Pub/Sub has no durability, no consumer groups, and no ack: a message reaches
//! whichever subscribers are connected at publish time, and `ack` / `nack` report
//! [`AckError::Unsupported`]. Two delivery modes exist, explicit because they do not interoperate:
//!
//! * [`PubSubMode::Classic`] - `SUBSCRIBE` / `PUBLISH`, broadcast to every node; supports patterns
//!   (`PSUBSCRIBE`). The only option on standalone and sentinel.
//! * [`PubSubMode::Sharded`] - `SSUBSCRIBE` / `SPUBLISH` (Redis 7+), slot-local so it scales across
//!   a cluster, but has no pattern support.
//!
//! A channel and a pattern are two descriptors, [`RedisPubSub`] and [`RedisPubSubPattern`],
//! because they differ in more than a flag: a pattern reads many channels and names none, so
//! sharded delivery has no meaning there and neither does a destination a retry copy could go to.
//!
//! Headers travel in a frame around the payload: a binary frame by default, or a readable
//! codec-serialized envelope when a codec is set with [`RedisPubSub::codec`] /
//! [`RedisPubSubPublish::codec`]. Both framings are lossless; the envelope writes a field whose
//! bytes are valid UTF-8 as text and any other bytes as themselves.

use std::fmt::{Debug, Formatter};
use std::future::{Future, ready};
use std::num::NonZeroUsize;
use std::sync::Arc;

use bytes::Bytes;
use fred::clients::{Client, Pool};
use fred::interfaces::{ClientLike, PubsubInterface};
use fred::types::{Message, MessageKind};
use futures::Stream;
use futures::stream::unfold;
#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::Bindings;
use ruststream::codec::Codec;
use ruststream::{
    AckError, AddressedCopies, BatchSubscriber, BufferedSubscriber, HeaderMap, IncomingMessage,
    Lend, NamedCopies, OutgoingMessage, PairError, Partitioned, PublishPolicy, Publisher,
    RedeliveryAddress, RedeliveryAddressed, RetryDeclaration, SubscriptionSource,
};
use tokio::runtime::Handle;
use tokio::sync::broadcast::Receiver;
use tokio::sync::broadcast::error::{RecvError, TryRecvError};
use tracing::warn;

use crate::broker::{ConnectedRedisBroker, RedisCore};
use crate::envelope::{SharedEnvelope, frame, unframe};
use crate::loopback::{InFlight, Tap};
use crate::partition::{RedisPublishOptions, resolved_headers};
use crate::pipeline::{PubSubForm, RoundMessage, Window};
use crate::publisher::joins_round;
use crate::route::Route;
use crate::{error::RedisError, message::PARTITION_KEY_HEADER};

/// This form's publish policy, [`RedisPubSubPublish`], under the mount-site name every form gives
/// its own. Its options, the delivery mode and the framing codec, still chain off it.
pub use crate::pubsub::RedisPubSubPublish as Publish;

/// The core prelude plus everything a Redis Pub/Sub service writes.
///
/// The broker, the [`RedisPubSub`] descriptor with its [`PubSubMode`], this form's [`Publish`]
/// policy, and its per-delivery context. Pub/Sub carries no core capability traits.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_fred::pubsub::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Event {
///     kind: String,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "events.seen")]
/// struct Seen {
///     kind: String,
/// }
///
/// #[subscriber(RedisPubSub::new("events").mode(PubSubMode::Sharded), reply)]
/// async fn on_event(event: &Event) -> Seen {
///     Seen {
///         kind: event.kind.clone(),
///     }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
///         RedisBroker::cluster(["10.0.0.1:6379"]),
///         |b| {
///             // Sharded in, sharded out: an `SPUBLISH` reaches only sharded subscribers.
///             b.include(on_event)
///                 .out_reply(Publish::new().mode(PubSubMode::Sharded));
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// Two vocabularies that do not mix. A handler body imports `ruststream::prelude::*` and bounds an
/// injected slot with the broker capability trait it needs (`Out<impl Publisher>`); a routes file
/// globs this prelude and names the policy by its mount-site word, the same word on every form.
///
/// A file that also globs another form's prelude sees an ambiguous `Publish`; use
/// [`crate::prelude`] and write `pubsub::Publish` there.
pub mod prelude {
    pub use ruststream::prelude::*;

    pub use super::{PubSubMode, Publish, RedisPubSub, RedisPubSubPattern};
    // `keys` arrives as the module, not as a glob: its members are short words a service also uses
    // for its own types, and `Ctx<keys::Channel>` reads as what it is at the use site.
    pub use crate::context::{PipelineContext, PoolContext, PubSubContext, keys};
    pub use crate::pipeline::{Bindable, InRound, RedisPipelineSteps};
    pub use crate::{PARTITION_KEY_HEADER, RedisBroker, RedisPublishOptions, RedisPublishSteps};

    #[cfg(any(
        feature = "tls-rustls",
        feature = "tls-rustls-ring",
        feature = "tls-native-tls"
    ))]
    pub use crate::{TlsConfig, TlsConnector};
}

/// Pub/Sub delivery mode. Defaults to [`Classic`](Self::Classic).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum PubSubMode {
    /// `SUBSCRIBE` / `PUBLISH`: cluster-wide broadcast, pattern-capable, does not scale by slot.
    #[default]
    Classic,
    /// `SSUBSCRIBE` / `SPUBLISH` (Redis 7+): slot-local sharded delivery, no patterns.
    Sharded,
}

impl PubSubMode {
    /// The word the generated `AsyncAPI` document names this mode by.
    #[cfg(feature = "asyncapi")]
    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::Classic => "classic",
            Self::Sharded => "sharded",
        }
    }
}

/// Describes one Pub/Sub subscription on a single channel, against a [`ConnectedRedisBroker`].
///
/// A glob is [`RedisPubSubPattern`] instead.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_fred::pubsub::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Event {
///     kind: String,
/// }
///
/// // Classic: `SUBSCRIBE`, broadcast to every node of a cluster.
/// #[subscriber(RedisPubSub::new("events"))]
/// async fn on_event(event: &Event) -> HandlerOutcome {
///     println!("event: {}", event.kind);
///     HandlerOutcome::ack()
/// }
///
/// // Sharded: `SSUBSCRIBE`, served by the node that owns the channel's slot.
/// #[subscriber(RedisPubSub::new("metrics").mode(PubSubMode::Sharded))]
/// async fn on_metric(event: &Event) -> HandlerOutcome {
///     println!("metric: {}", event.kind);
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
///         RedisBroker::cluster(["10.0.0.1:6379"]),
///         |b| {
///             b.include(on_event);
///             b.include(on_metric);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Clone)]
#[must_use]
pub struct RedisPubSub {
    channel: String,
    mode: PubSubMode,
    codec: Option<SharedEnvelope>,
    /// Where the mount site sends a spent delivery, taken from its retry declaration when the
    /// subscription opens, so the broker publishes to that name as the channel it is.
    dead_letter: Option<String>,
    /// `None` keeps the capacity `fred` gives a client's broadcast channels by default.
    buffer: Option<NonZeroUsize>,
}

impl Debug for RedisPubSub {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPubSub")
            .field("channel", &self.channel)
            .field("mode", &self.mode)
            .field("codec", &self.codec.is_some())
            .field("dead_letter", &self.dead_letter)
            .field("buffer", &self.buffer)
            .finish()
    }
}

impl RedisPubSub {
    /// A subscription on the exact `channel`.
    pub fn new(channel: impl Into<String>) -> Self {
        Self {
            channel: channel.into(),
            mode: PubSubMode::default(),
            codec: None,
            dead_letter: None,
            buffer: None,
        }
    }

    /// Sets the delivery mode. Defaults to [`PubSubMode::Classic`].
    pub const fn mode(mut self, mode: PubSubMode) -> Self {
        self.mode = mode;
        self
    }

    /// Decodes the header/payload envelope with `codec` (must match the publisher). Without it the
    /// default binary framing is used. Either way the payload arrives as it was published.
    pub fn codec(mut self, codec: impl Codec + 'static) -> Self {
        self.codec = Some(Arc::new(codec));
        self
    }

    /// Sets how many messages the subscription holds that its handler has not taken yet.
    ///
    /// Redis pushes a Pub/Sub message the moment it is published, and the subscription keeps it
    /// until the handler asks for the next one. The buffer bounds how many may wait. Once it is
    /// full, each new message drops the oldest, and the subscription logs a warning with the
    /// channel and the count it lost. The default is `fred`'s own
    /// (`PerformanceConfig::broadcast_channel_capacity`, 32 messages in `fred` 10.1).
    ///
    /// Raise it for a handler that falls behind bursts: set it to the largest burst the handler
    /// must absorb. The room is reserved when the subscription opens, about 600 bytes per message
    /// of buffer whatever the payloads weigh, because `fred` sizes all eight of a client's
    /// notification channels by it.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream::nonzero;
    /// use ruststream_fred::pubsub::prelude::*;
    /// # #[derive(serde::Deserialize)]
    /// # struct Event { kind: String }
    ///
    /// // A handler that falls behind a burst of up to ten thousand events loses none of them.
    /// #[subscriber(RedisPubSub::new("events").buffer(nonzero!(10_000)))]
    /// async fn on_event(event: &Event) -> HandlerOutcome {
    ///     println!("event {}", event.kind);
    ///     HandlerOutcome::ack()
    /// }
    /// #
    /// # fn app() -> RustStream {
    /// #     RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
    /// #         RedisBroker::standalone("redis://localhost:6379"),
    /// #         |b| {
    /// #             b.include(on_event);
    /// #         },
    /// #     )
    /// # }
    /// # }
    /// ```
    pub const fn buffer(mut self, messages: NonZeroUsize) -> Self {
        self.buffer = Some(messages);
        self
    }

    /// The channel this subscription listens on.
    #[must_use]
    pub fn channel(&self) -> &str {
        &self.channel
    }

    pub(crate) const fn buffer_size(&self) -> Option<NonZeroUsize> {
        self.buffer
    }

    pub(crate) const fn delivery_mode(&self) -> PubSubMode {
        self.mode
    }

    pub(crate) fn codec_handle(&self) -> Option<SharedEnvelope> {
        self.codec.clone()
    }

    /// How a name this subscription reads or dead-letters to is written: a publish in its mode
    /// and its framing.
    pub(crate) fn route(&self) -> Route {
        Route::Channel {
            mode: self.mode,
            envelope: self.codec.clone(),
        }
    }

    /// The dead-letter destination the mount site declared, once the registration has handed its
    /// declaration over.
    pub(crate) fn dead_letter(&self) -> Option<&str> {
        self.dead_letter.as_deref()
    }

    /// Takes the dead-letter destination out of the mount site's declaration.
    fn with_declaration(mut self, declaration: &RetryDeclaration) -> Self {
        self.dead_letter = declaration.dead_letter().map(str::to_owned);
        self
    }

    /// Where a publisher reaches this subscription again, shared by both broker forms.
    fn redelivery_channel(&self) -> RedeliveryAddress {
        RedeliveryAddress::new(self.channel.clone())
    }

    /// What this subscription adds to its channel in the generated `AsyncAPI` document, shared by
    /// both broker forms: the delivery mode it subscribes in and how headers are framed.
    #[cfg(feature = "asyncapi")]
    fn describe(&self) -> Bindings {
        crate::asyncapi::channel(&crate::asyncapi::PubSubSubscription::new(
            self.mode.as_str(),
            false,
            crate::asyncapi::Envelope::of(self.codec.as_ref()),
        ))
    }
}

/// Describes one Pub/Sub subscription on a glob pattern (`PSUBSCRIBE`), against a
/// [`ConnectedRedisBroker`].
///
/// Patterns are classic-only: sharded Pub/Sub routes by slot and has no `PSUBSCRIBE`, so this
/// descriptor carries no mode to get wrong.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_fred::pubsub::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Event {
///     kind: String,
/// }
///
/// #[subscriber(RedisPubSubPattern::new("events.*"))]
/// async fn on_any_event(event: &Event, Ctx(channel): Ctx<keys::Channel>) -> HandlerOutcome {
///     println!("{} on {channel}", event.kind);
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
///         RedisBroker::standalone("redis://localhost:6379"),
///         |b| {
///             // A glob is no channel a `PUBLISH` can name, so the mount site says where a
///             // retry copy goes.
///             b.include(on_any_event)
///                 .out_retry(Publish::default())
///                 .to("events.retry");
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// Sharded Pub/Sub has no pattern subscription, so a pattern takes no mode:
///
/// ```compile_fail
/// use ruststream_fred::{PubSubMode, RedisPubSubPattern};
///
/// let events = RedisPubSubPattern::new("events.*").mode(PubSubMode::Sharded);
/// ```
#[derive(Clone)]
#[must_use]
pub struct RedisPubSubPattern {
    pattern: String,
    codec: Option<SharedEnvelope>,
    /// `None` keeps the capacity `fred` gives a client's broadcast channels by default.
    buffer: Option<NonZeroUsize>,
}

impl Debug for RedisPubSubPattern {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPubSubPattern")
            .field("pattern", &self.pattern)
            .field("codec", &self.codec.is_some())
            .field("buffer", &self.buffer)
            .finish()
    }
}

impl RedisPubSubPattern {
    /// A subscription on every channel matching the glob `pattern`.
    pub fn new(pattern: impl Into<String>) -> Self {
        Self {
            pattern: pattern.into(),
            codec: None,
            buffer: None,
        }
    }

    /// Decodes the header/payload envelope with `codec` (must match the publisher). Without it the
    /// default binary framing is used. Either way the payload arrives as it was published.
    pub fn codec(mut self, codec: impl Codec + 'static) -> Self {
        self.codec = Some(Arc::new(codec));
        self
    }

    /// Sets how many messages the subscription holds that its handler has not taken yet.
    ///
    /// Redis pushes a Pub/Sub message the moment it is published, and the subscription keeps it
    /// until the handler asks for the next one. The buffer bounds how many may wait. Once it is
    /// full, each new message drops the oldest, and the subscription logs a warning with the
    /// channel and the count it lost. The default is `fred`'s own
    /// (`PerformanceConfig::broadcast_channel_capacity`, 32 messages in `fred` 10.1).
    ///
    /// Raise it for a handler that falls behind bursts: set it to the largest burst the handler
    /// must absorb. The room is reserved when the subscription opens, about 600 bytes per message
    /// of buffer whatever the payloads weigh, because `fred` sizes all eight of a client's
    /// notification channels by it.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use ruststream::nonzero;
    /// use ruststream_fred::pubsub::prelude::*;
    /// # #[derive(serde::Deserialize)]
    /// # struct Event { kind: String }
    ///
    /// // A handler that falls behind a burst of up to ten thousand events loses none of them.
    /// #[subscriber(RedisPubSubPattern::new("events.*").buffer(nonzero!(10_000)))]
    /// async fn on_event(event: &Event) -> HandlerOutcome {
    ///     println!("event {}", event.kind);
    ///     HandlerOutcome::ack()
    /// }
    /// #
    /// # fn app() -> RustStream {
    /// #     RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
    /// #         RedisBroker::standalone("redis://localhost:6379"),
    /// #         |b| {
    /// #             b.include(on_event);
    /// #         },
    /// #     )
    /// # }
    /// # }
    /// ```
    pub const fn buffer(mut self, messages: NonZeroUsize) -> Self {
        self.buffer = Some(messages);
        self
    }

    /// The glob this subscription matches channels against.
    #[must_use]
    pub fn pattern(&self) -> &str {
        &self.pattern
    }

    pub(crate) const fn buffer_size(&self) -> Option<NonZeroUsize> {
        self.buffer
    }

    pub(crate) fn codec_handle(&self) -> Option<SharedEnvelope> {
        self.codec.clone()
    }

    /// Patterns are classic-only, so the mode in the document is the only one they have.
    #[cfg(feature = "asyncapi")]
    fn describe(&self) -> Bindings {
        crate::asyncapi::channel(&crate::asyncapi::PubSubSubscription::new(
            PubSubMode::Classic.as_str(),
            true,
            crate::asyncapi::Envelope::of(self.codec.as_ref()),
        ))
    }
}

impl SubscriptionSource<ConnectedRedisBroker> for RedisPubSub {
    type Subscriber = RedisPubSubSubscriber;
    /// Pub/Sub cannot acknowledge, so every retry is a copy this process publishes, and the
    /// channel is where it goes.
    type Copies = AddressedCopies;

    fn name(&self) -> &str {
        self.channel()
    }

    fn declare_retry(self, declaration: &RetryDeclaration) -> Self {
        self.with_declaration(declaration)
    }

    async fn subscribe(
        self,
        connected: &ConnectedRedisBroker,
    ) -> Result<Self::Subscriber, RedisError> {
        connected.subscribe_pubsub(self).await
    }

    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self) -> Bindings {
        self.describe()
    }
}

/// The channel itself.
///
/// The answer is about the address, not about the command: reaching a
/// [`PubSubMode::Sharded`] subscription still takes a publisher in the same mode, which is the
/// policy the mount site names with `out_retry`.
impl RedeliveryAddressed<ConnectedRedisBroker> for RedisPubSub {
    fn redelivery_address(
        &self,
        _connected: &ConnectedRedisBroker,
    ) -> impl Future<Output = Result<RedeliveryAddress, RedisError>> {
        ready(Ok(self.redelivery_channel()))
    }
}

/// A pattern subscription reads every channel the glob matches, and a glob is not a channel a
/// `PUBLISH` can name, so the mount site names where a retry copy goes: `.out_retry(policy).to(..)`
/// for a fixed channel, or a transform that reads the delivery's own channel.
impl SubscriptionSource<ConnectedRedisBroker> for RedisPubSubPattern {
    type Subscriber = RedisPubSubSubscriber;
    type Copies = NamedCopies;

    fn name(&self) -> &str {
        self.pattern()
    }

    async fn subscribe(
        self,
        connected: &ConnectedRedisBroker,
    ) -> Result<Self::Subscriber, RedisError> {
        connected.subscribe_pubsub_pattern(self).await
    }

    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self) -> Bindings {
        self.describe()
    }
}

/// A Pub/Sub subscription backed by a dedicated `fred` client, so its message stream and channel
/// state are isolated from other subscribers and from the publishing pool.
///
/// Pub/Sub delivers one message at a time, so the batches a [`BatchSubscriber`] hands out are
/// assembled on the client by the core's [`BufferedSubscriber`]: it fills a batch up to the size
/// the mount site asked for and closes a partial one on its own deadline.
pub struct RedisPubSubSubscriber(BufferedSubscriber<PubSubWire>);

impl Debug for RedisPubSubSubscriber {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPubSubSubscriber")
            .finish_non_exhaustive()
    }
}

impl RedisPubSubSubscriber {
    pub(crate) fn new(wire: PubSubWire) -> Self {
        Self(BufferedSubscriber::new(wire))
    }
}

impl ruststream::Subscriber for RedisPubSubSubscriber {
    type Message = RedisPubSubMessage;
    type Error = RedisError;

    /// Yields one message per Pub/Sub delivery.
    ///
    /// # Cancel safety
    ///
    /// Dropping the returned stream between items is safe. What arrives while no stream is polling
    /// waits in the subscription's [buffer](crate::RedisPubSub::buffer); past it the oldest is
    /// lost, and while the subscription is disconnected everything is, as Redis Pub/Sub keeps
    /// nothing for a subscriber that is not there.
    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
        self.0.stream()
    }
}

impl BatchSubscriber for RedisPubSubSubscriber {
    type Batch = Vec<RedisPubSubMessage>;

    /// Yields batches of at most `size` deliveries, assembled as they arrive.
    ///
    /// # Cancel safety
    ///
    /// Same as [`Subscriber::stream`](ruststream::Subscriber::stream). A batch abandoned mid-fill
    /// loses the deliveries it holds, as any Pub/Sub delivery nobody is polling for is lost.
    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, Self::Error>> + Send + '_ {
        self.0.batches(size)
    }
}

/// The wire side of a Pub/Sub subscription: the dedicated client and the channel it feeds.
pub(crate) struct PubSubWire {
    /// The channel or the glob it reads, for the warning a lost message raises.
    name: String,
    client: Client,
    rx: Receiver<Message>,
    codec: Option<SharedEnvelope>,
    /// The broker's pool, handed to every delivery for `Ctx<keys::FredPool>`.
    pool: Pool,
    /// The subscription's registration with the in-process server; empty on a real connection.
    tap: Tap,
    /// The broker's runtime, where the dedicated client is closed once the subscription stops.
    runtime: Handle,
}

impl Debug for PubSubWire {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PubSubWire").finish_non_exhaustive()
    }
}

impl PubSubWire {
    pub(crate) fn new(
        name: String,
        client: Client,
        rx: Receiver<Message>,
        codec: Option<SharedEnvelope>,
        pool: Pool,
        tap: Tap,
        runtime: Handle,
    ) -> Self {
        Self {
            name,
            client,
            rx,
            codec,
            pool,
            tap,
            runtime,
        }
    }

    /// The window's client: a connection of the pool the flushes go out on.
    pub(crate) fn round_client(&self) -> Client {
        self.pool.next().clone()
    }

    /// Yields one delivery per message, each with a slot in `window`, telling the window how
    /// many more have arrived behind it, and reports a failed flush of the window once.
    pub(crate) fn round_stream<'a>(
        &'a mut self,
        window: &'a Arc<Window<PubSubForm>>,
    ) -> impl Stream<Item = Result<RoundMessage<PubSubForm>, RedisError>> + Send + 'a {
        let codec = self.codec.clone();
        let (pool, tap, name) = (&self.pool, &self.tap, self.name.as_str());
        unfold((&mut self.rx, codec), move |(rx, codec)| async move {
            loop {
                if let Some(failure) = window.take_failure() {
                    return Some((Err(failure), (rx, codec)));
                }
                match rx.recv().await {
                    Ok(msg) => {
                        let delivery = to_message(&msg, codec.as_ref(), pool, tap);
                        window.yielded(rx.len());
                        let round = window.open();
                        return Some((
                            Ok(RoundMessage::new(delivery, Arc::clone(window), round)),
                            (rx, codec),
                        ));
                    }
                    // The receiver fell behind the broadcast buffer; skip the gap and keep reading.
                    Err(RecvError::Lagged(skipped)) => lagged(name, tap, skipped),
                    Err(RecvError::Closed) => return None,
                }
            }
        })
    }
}

/// Reports the messages a subscription that fell behind its buffer lost, and forgets them.
fn lagged(name: &str, tap: &Tap, skipped: u64) {
    warn!(
        subscription = name,
        lost = skipped,
        "a Redis Pub/Sub subscription fell behind its buffer and lost the oldest messages; raise \
         its buffer or speed up its handler"
    );
    #[cfg(feature = "testing")]
    tap.discard(usize::try_from(skipped).unwrap_or(usize::MAX));
    #[cfg(not(feature = "testing"))]
    let _ = tap;
}

impl PubSubWire {
    /// Yields one batch of up to `size` messages: the first one waited for, and those already
    /// arrived behind it. The batch shares one slot of `window`.
    pub(crate) fn round_batches<'a>(
        &'a mut self,
        window: &'a Arc<Window<PubSubForm>>,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Vec<RoundMessage<PubSubForm>>, RedisError>> + Send + 'a {
        window.fill_at(size.get());
        let codec = self.codec.clone();
        let (pool, tap, name) = (&self.pool, &self.tap, self.name.as_str());
        unfold((&mut self.rx, codec), move |(rx, codec)| async move {
            loop {
                if let Some(failure) = window.take_failure() {
                    return Some((Err(failure), (rx, codec)));
                }
                let first = match rx.recv().await {
                    Ok(msg) => msg,
                    Err(RecvError::Lagged(skipped)) => {
                        lagged(name, tap, skipped);
                        continue;
                    }
                    Err(RecvError::Closed) => return None,
                };
                let mut messages = vec![first];
                while messages.len() < size.get() {
                    match rx.try_recv() {
                        Ok(msg) => messages.push(msg),
                        Err(TryRecvError::Lagged(skipped)) => lagged(name, tap, skipped),
                        Err(_) => break,
                    }
                }
                let members = u32::try_from(messages.len()).unwrap_or(u32::MAX);
                for _ in 0..members {
                    window.yielded(rx.len());
                }
                let round = window.open_batch(members);
                let batch = messages
                    .iter()
                    .map(|msg| {
                        RoundMessage::new(
                            to_message(msg, codec.as_ref(), pool, tap),
                            Arc::clone(window),
                            round.clone(),
                        )
                    })
                    .collect();
                return Some((Ok(batch), (rx, codec)));
            }
        })
    }
}

impl Drop for PubSubWire {
    fn drop(&mut self) {
        // What reached the subscription and was never read leaves with it: the in-process server
        // counted each message in flight when it sent it.
        #[cfg(feature = "testing")]
        {
            let mut unread = 0_usize;
            loop {
                match self.rx.try_recv() {
                    Ok(_) => unread += 1,
                    Err(TryRecvError::Lagged(skipped)) => {
                        unread =
                            unread.saturating_add(usize::try_from(skipped).unwrap_or(usize::MAX));
                    }
                    Err(_) => break,
                }
            }
            self.tap.discard(unread);
        }
        // The dedicated client owns a background connection task; close it on a detached task since
        // `drop` cannot await, on the broker's runtime, which is there wherever the drop happens.
        let client = self.client.clone();
        self.runtime.spawn(async move {
            let _ = client.quit().await;
        });
    }
}

fn to_message(
    msg: &Message,
    codec: Option<&SharedEnvelope>,
    pool: &Pool,
    tap: &Tap,
) -> RedisPubSubMessage {
    let raw = msg.value.as_bytes().unwrap_or(&[]);
    let (payload, headers) = unframe(codec, raw);
    RedisPubSubMessage {
        channel: msg.channel.to_string(),
        // `PMessage` is the delivery kind for a `PSUBSCRIBE` match; the message's own channel is the
        // concrete one matched, which differs from the subscription's glob pattern.
        pattern: matches!(msg.kind, MessageKind::PMessage),
        payload,
        headers,
        pool: pool.clone(),
        flight: tap.delivered(),
    }
}

impl ruststream::Subscriber for PubSubWire {
    type Message = RedisPubSubMessage;
    type Error = RedisError;

    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
        let codec = self.codec.clone();
        let (pool, tap, name) = (&self.pool, &self.tap, self.name.as_str());
        unfold((&mut self.rx, codec), move |(rx, codec)| async move {
            loop {
                match rx.recv().await {
                    Ok(msg) => {
                        let message = to_message(&msg, codec.as_ref(), pool, tap);
                        return Some((Ok(message), (rx, codec)));
                    }
                    // The receiver fell behind the broadcast buffer; skip the gap and keep reading.
                    Err(RecvError::Lagged(skipped)) => lagged(name, tap, skipped),
                    Err(RecvError::Closed) => return None,
                }
            }
        })
    }
}

/// A Pub/Sub delivery. `ack` / `nack` are unsupported (Pub/Sub has no acknowledgement).
pub struct RedisPubSubMessage {
    channel: String,
    /// Whether this delivery arrived through a `PSUBSCRIBE` pattern match (vs an exact subscribe).
    pattern: bool,
    payload: Bytes,
    headers: HeaderMap,
    /// The broker's pool, which `Ctx<keys::FredPool>` hands the handler.
    pool: Pool,
    /// The test harness's count of this delivery; empty outside the in-process mode.
    #[cfg_attr(
        not(feature = "testing"),
        allow(
            dead_code,
            reason = "held for its release on drop, which only `testing` gives"
        )
    )]
    flight: InFlight,
}

impl Debug for RedisPubSubMessage {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPubSubMessage")
            .field("channel", &self.channel)
            .field("pattern", &self.pattern)
            .field("payload_len", &self.payload.len())
            .finish_non_exhaustive()
    }
}

impl RedisPubSubMessage {
    /// The channel this message arrived on.
    ///
    /// On a [`RedisPubSubPattern`] subscription this is the concrete channel the message was
    /// published to, which differs from the glob the subscription registered.
    #[must_use]
    pub fn channel(&self) -> &str {
        &self.channel
    }

    /// Whether this delivery arrived through a `PSUBSCRIBE` pattern match rather than an exact
    /// channel subscribe.
    #[must_use]
    pub fn from_pattern(&self) -> bool {
        self.pattern
    }

    /// The broker's connection pool, for the per-delivery context.
    pub(crate) const fn pool(&self) -> &Pool {
        &self.pool
    }

    /// Takes the harness's count of this delivery, for a window to hold until its flush.
    #[cfg(feature = "testing")]
    pub(crate) fn take_flight(&mut self) -> InFlight {
        std::mem::take(&mut self.flight)
    }
}

impl IncomingMessage for RedisPubSubMessage {
    fn payload(&self) -> &[u8] {
        &self.payload
    }

    fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    // See `RedisMessage`: the keyed worker lanes read the key from here, not from `Partitioned`.
    fn partition_key(&self) -> Option<&[u8]> {
        Partitioned::partition_key(self)
    }

    fn ack(self) -> impl Future<Output = Result<(), AckError>> {
        ready(Err(AckError::Unsupported))
    }

    fn nack(self, _requeue: bool) -> impl Future<Output = Result<(), AckError>> {
        ready(Err(AckError::Unsupported))
    }
}

impl Partitioned for RedisPubSubMessage {
    fn partition_key(&self) -> Option<&[u8]> {
        self.headers().get(PARTITION_KEY_HEADER)
    }
}

/// The declaration half of the Pub/Sub publisher: delivery mode and envelope codec, no connection.
///
/// Constructible anywhere (in a router definition, in configuration), it pairs into a
/// [`RedisPubSubPublisher`] against a [`ConnectedRedisBroker`]. The publish mode must match how
/// subscribers subscribed: a sharded publish only reaches sharded subscribers.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_fred::stream::prelude::*;
/// use ruststream_fred::{PubSubMode, RedisPubSubPublish};
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "orders.live")]
/// struct Placed {
///     id: u64,
/// }
///
/// #[subscriber(RedisStream::new("orders").group("workers"), reply)]
/// async fn announce(order: &Order) -> Placed {
///     Placed { id: order.id }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         RedisBroker::cluster(["10.0.0.1:6379"]),
///         |b| {
///             // Read from a durable stream, broadcast to whoever is listening with `SPUBLISH`.
///             b.include(announce)
///                 .out_reply(RedisPubSubPublish::new().mode(PubSubMode::Sharded));
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Clone, Default)]
#[must_use]
pub struct RedisPubSubPublish {
    mode: PubSubMode,
    codec: Option<SharedEnvelope>,
}

impl Debug for RedisPubSubPublish {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPubSubPublish")
            .field("mode", &self.mode)
            .field("codec", &self.codec.is_some())
            .finish()
    }
}

impl RedisPubSubPublish {
    /// A classic-mode policy with the default binary framing. Equivalent to [`Self::default`].
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the publish mode. Defaults to [`PubSubMode::Classic`].
    pub const fn mode(mut self, mode: PubSubMode) -> Self {
        self.mode = mode;
        self
    }

    /// Serializes the header/payload envelope with `codec` (must match the subscriber), which makes
    /// the wire value readable while the data is text. Without it the default binary framing is
    /// used. Either way a payload that is not text is published byte for byte.
    pub fn codec(mut self, codec: impl Codec + 'static) -> Self {
        self.codec = Some(Arc::new(codec));
        self
    }
}

impl PublishPolicy<ConnectedRedisBroker> for RedisPubSubPublish {
    type Live = RedisPubSubPublisher;

    fn pair(
        self,
        connected: &ConnectedRedisBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(Ok(connected.pubsub_publisher(self)))
    }

    /// The channel a `PUBLISH` goes out on, the delivery mode it goes out in, and how headers are
    /// framed beside the payload.
    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self, channel: &str) -> Bindings {
        crate::asyncapi::channel(&crate::asyncapi::Publish::pubsub(
            channel,
            self.mode.as_str(),
            crate::asyncapi::Envelope::of(self.codec.as_ref()),
        ))
    }
}

/// Publishes Pub/Sub messages with `PUBLISH` (classic) or `SPUBLISH` (sharded): a
/// [`RedisPubSubPublish`] policy paired with a connection.
///
/// Obtain it from
/// [`ConnectedRedisBroker::pubsub_publisher`](crate::ConnectedRedisBroker::pubsub_publisher), or
/// by pairing the policy. Like every publisher here it may outlive the connection, so publishing
/// after shutdown reports [`RedisError::ShutDown`].
#[derive(Clone)]
pub struct RedisPubSubPublisher {
    core: Arc<RedisCore>,
    mode: PubSubMode,
    codec: Option<SharedEnvelope>,
    /// What `pipeline.bind(&out)` names this publisher and its clones by.
    round_name: u64,
}

impl Debug for RedisPubSubPublisher {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedisPubSubPublisher")
            .field("mode", &self.mode)
            .field("codec", &self.codec.is_some())
            .finish_non_exhaustive()
    }
}

impl RedisPubSubPublisher {
    /// What `pipeline.bind(&out)` names this publisher and its clones by.
    pub(crate) const fn round_name(&self) -> u64 {
        self.round_name
    }

    pub(crate) fn new(core: Arc<RedisCore>, publish: RedisPubSubPublish) -> Self {
        Self {
            round_name: core.rounds().publisher(),
            core,
            mode: publish.mode,
            codec: publish.codec,
        }
    }
}

impl Publisher for RedisPubSubPublisher {
    /// The payload is read into an envelope of this crate's own before it reaches the client, so
    /// there is nothing to keep: the publish is lent the bytes where they already are.
    type Payload = Lend;

    type Error = RedisError;
    /// `PUBLISH` and `SPUBLISH` take a channel and a payload and nothing else, and which of the
    /// two is issued is the publisher's mode, fixed by the policy, because a sharded publish only
    /// reaches sharded subscribers. What a call site still says is the partition key, framed into
    /// the payload with the message's other headers.
    type Options = RedisPublishOptions;

    async fn publish(
        &self,
        msg: OutgoingMessage<'_>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        let (channel, payload, headers) = msg.into_parts();
        if let Some(round) = self
            .core
            .rounds()
            .joined(self.round_name, joins_round(options))
        {
            let body = frame(
                self.codec.as_ref(),
                payload,
                &resolved_headers(headers, options),
            );
            return round
                .publish(channel, body, self.mode == PubSubMode::Sharded)
                .await
                .map_err(RedisError::publish);
        }
        send(
            &self.core,
            self.mode,
            self.codec.as_ref(),
            channel,
            payload,
            resolved_headers(headers, options),
        )
        .await
    }
}

/// Publishes one framed message on `channel` in `mode`: the Pub/Sub publish, shared by
/// [`RedisPubSubPublisher`] and the broker's default publisher.
pub(crate) async fn send(
    core: &RedisCore,
    mode: PubSubMode,
    codec: Option<&SharedEnvelope>,
    channel: &str,
    payload: &[u8],
    headers: HeaderMap,
) -> Result<(), RedisError> {
    let pool = core.pool()?;
    let client = pool.next();
    let body = frame(codec, payload, &headers);
    let channel = channel.to_owned();
    let _: i64 = match mode {
        PubSubMode::Classic => client.publish(channel, body).await,
        PubSubMode::Sharded => client.spublish(channel, body).await,
    }
    .map_err(RedisError::publish)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::context::PubSubContext;
    use fred::types::config::Config;
    use ruststream::BuildContext;

    /// An unconnected pool (just client structs); `Pool::new` opens no sockets.
    fn offline_pool() -> Pool {
        Pool::new(Config::default(), None, None, None, 1).expect("offline pool")
    }

    #[test]
    fn build_context_reads_channel_and_pattern_flag() {
        let exact = RedisPubSubMessage {
            channel: "events".to_owned(),
            pattern: false,
            payload: Bytes::from_static(b"{}"),
            headers: HeaderMap::new(),
            pool: offline_pool(),
            flight: InFlight::default(),
        };
        let cx = PubSubContext::build(&exact);
        assert_eq!(cx.channel(), "events");
        assert!(!cx.from_pattern());

        let matched = RedisPubSubMessage {
            channel: "events.user".to_owned(),
            pattern: true,
            payload: Bytes::from_static(b"{}"),
            headers: HeaderMap::new(),
            pool: offline_pool(),
            flight: InFlight::default(),
        };
        assert!(PubSubContext::build(&matched).from_pattern());
    }

    /// A channel is one destination a `PUBLISH` names, whatever the delivery mode.
    #[test]
    fn a_channel_addresses_its_copies_to_itself() {
        assert_eq!(
            RedisPubSub::new("events").redelivery_channel().as_str(),
            "events"
        );
        assert_eq!(
            RedisPubSub::new("events")
                .mode(PubSubMode::Sharded)
                .redelivery_channel()
                .as_str(),
            "events"
        );
    }
}