ruststream-fred 0.7.0

//! Builder describing one Redis Streams subscription.
//!
//! A subscription always reads through a consumer group. Three read modes are selected by
//! constructor, never by a runtime flag, because they return disjoint message sets:
//!
//! * [`RedisStream::new`] reads fresh entries off the tail (`XREADGROUP > ...`).
//! * [`RedisStream::reclaim`] reads stale pending entries another consumer never acked
//!   (`XAUTOCLAIM`, idle at least `min_idle`) - the crash-recovery path.
//! * [`RedisStream::claiming`] reads both in one call (`XREADGROUP ... CLAIM`, Redis 8.4 and
//!   later): stale pending entries first, then fresh ones, each with its idle time and delivery
//!   count.
//!
//! Inferring the mode from a numeric parameter would be a footgun (a stray idle timeout could
//! silently stop fresh delivery), so the mode is part of the constructor name. On a server older
//! than 8.4, recovery is a separate `reclaim` subscriber beside the `new` one on the same group:
//! "two handlers per group".

use std::future::{Future, ready};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::Bindings;
use ruststream::{
    AddressedCopies, RedeliveryAddress, RedeliveryAddressed, RetryDeclaration, SubscriptionSource,
};

#[cfg(feature = "asyncapi")]
use crate::asyncapi;
use crate::broker::ConnectedRedisBroker;
use crate::delay::{DelayConfig, DelayedRetry};
use crate::{error::RedisError, subscriber::RedisSubscriber};

/// This form's publish policy, [`RedisPublish`](crate::RedisPublish), under the mount-site name
/// every form gives its own.
pub use crate::publisher::RedisPublish as Publish;

/// The same policy under the transactional mount-site name: a stream publisher buffers on the
/// handle and owns transactions as it is, so the plain policy *is* the transactional one and
/// there is no second type to transition to.
pub use crate::publisher::RedisPublish as TransactionalPublish;

/// The core prelude plus everything a Redis Streams service writes.
///
/// The broker, the [`RedisStream`] descriptor and its options, the group seek types with the
/// delivery and batch contexts that carry them, this form's [`Publish`] policy, and the
/// transaction, position and seek capabilities.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_fred::stream::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize, Serialize, Outgoing)]
/// #[outgoing(name = "orders.priced")]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(RedisStream::new("orders").group("workers"), reply)]
/// async fn price(orders: &[Order]) -> Result<Vec<Order>, HandlerOutcome> {
///     Ok(orders.iter().map(|order| Order { id: order.id }).collect())
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         RedisBroker::standalone("redis://localhost:6379"),
///         |b| {
///             // One batch's replies land together, as one MULTI / EXEC block.
///             b.include(price.batch(nonzero!(32)))
///                 .out_reply(TransactionalPublish)
///                 .transactional();
///         },
///     )
/// }
/// # }
/// # 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>`, `Out<impl
/// TransactionalPublisher>`); 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 `stream::Publish` there.
pub mod prelude {
    pub use ruststream::prelude::*;

    pub use ruststream::{
        OwnedTransactions, Positioned, Seeker, Transaction, TransactionalPublisher,
    };

    pub use super::{Publish, RedisStream, StreamStart, TransactionalPublish};
    // `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::SeekHandle>` reads as what it is at the use site.
    pub use crate::context::{
        PipelineContext, PoolContext, StreamBatchContext, StreamContext, keys,
    };
    pub use crate::pipeline::{Bindable, InRound, RedisPipelineSteps};
    pub use crate::{
        DelayedRetry, PARTITION_KEY_HEADER, RedisBroker, RedisGroupPosition, RedisGroupSeeker,
        RedisPublishOptions, RedisPublishSteps, RedisSubscribeExt,
    };

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

const DEFAULT_BLOCK: Duration = Duration::from_secs(5);

/// Generates an automatic consumer name when the caller does not set one. Distinct names keep
/// each in-process subscriber's pending list separate within a shared group.
fn auto_consumer() -> String {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("ruststream-{n}")
}

/// Where a freshly created consumer group starts reading from. Only consulted when the group does
/// not yet exist; an existing group keeps its own cursor.
#[derive(Debug, Clone, Default)]
pub enum StreamStart {
    /// Only entries added after the group is created (`$`). The default.
    #[default]
    New,
    /// Every entry currently in the stream (`0`).
    Beginning,
    /// A specific entry ID, exclusive.
    Id(String),
}

impl StreamStart {
    pub(crate) fn as_id(&self) -> &str {
        match self {
            Self::New => "$",
            Self::Beginning => "0",
            Self::Id(id) => id,
        }
    }
}

#[derive(Debug, Clone)]
pub(crate) enum ReadMode {
    /// `XREADGROUP >` - fresh tail.
    Fresh,
    /// `XAUTOCLAIM` of entries idle at least this long.
    Reclaim { min_idle: Duration },
    /// `XREADGROUP ... CLAIM` - entries idle at least this long, then the fresh tail, in one read.
    Claiming { min_idle: Duration },
}

impl ReadMode {
    /// How a retry redelivers on a subscription reading this way.
    pub(crate) const fn requeue(&self) -> RequeueMode {
        match self {
            Self::Fresh | Self::Reclaim { .. } => RequeueMode::Republish,
            Self::Claiming { min_idle } => RequeueMode::LeavePending {
                min_idle: *min_idle,
            },
        }
    }
}

/// What a retry does with an entry, which is a property of the read mode.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RequeueMode {
    /// Append a copy to the stream and ack the original. The retry is immediate, and the copy is
    /// a new entry whose delivery count starts over.
    Republish,
    /// Leave the entry in the pending entries list. The subscription's own next read claims it
    /// back once it has been idle `min_idle`, with the server's delivery count one higher, which
    /// is how a claiming subscription retries without duplicating the entry - and what makes a
    /// delayed retry native there, at that granularity.
    LeavePending { min_idle: Duration },
}

/// Describes one Redis Streams subscription against a [`ConnectedRedisBroker`].
///
/// # Examples
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_fred::stream::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// // Fresh tail: the normal worker, reading new entries.
/// #[subscriber(RedisStream::new("orders").group("workers"))]
/// async fn handle(order: &Order) -> HandlerOutcome {
///     println!("order {}", order.id);
///     HandlerOutcome::ack()
/// }
///
/// // Recovery on the same group: entries a crashed worker left pending for over 30s.
/// #[subscriber(RedisStream::reclaim("orders", Duration::from_secs(30)).group("workers"))]
/// async fn recover(order: &Order) -> HandlerOutcome {
///     println!("recovered order {}", order.id);
///     HandlerOutcome::ack()
/// }
///
/// #[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);
///             b.include(recover);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone)]
#[must_use]
pub struct RedisStream {
    key: String,
    group: Option<String>,
    consumer: Option<String>,
    block: Option<Duration>,
    start: StreamStart,
    mode: ReadMode,
    delayed_retry: Option<DelayedRetry>,
    /// Where the mount site sends a spent delivery, taken from its retry declaration when the
    /// subscription opens, so the broker writes that name as the stream it is.
    dead_letter: Option<String>,
}

impl RedisStream {
    /// A fresh-tail subscription on `key`: reads new entries via `XREADGROUP >`.
    ///
    /// A consumer group is required; set it with [`group`](Self::group).
    pub fn new(key: impl Into<String>) -> Self {
        Self {
            key: key.into(),
            group: None,
            consumer: None,
            block: None,
            start: StreamStart::New,
            mode: ReadMode::Fresh,
            delayed_retry: None,
            dead_letter: None,
        }
    }

    /// A recovery subscription on `key`: reclaims pending entries idle at least `min_idle` via
    /// `XAUTOCLAIM`. Run it alongside a [`new`](Self::new) subscriber on the same group to pick up
    /// messages a consumer fetched but died before acking.
    ///
    /// `min_idle` has no default and must exceed the longest legitimate handler runtime: set it too
    /// low and a healthy consumer's in-flight message gets reclaimed and processed twice.
    pub fn reclaim(key: impl Into<String>, min_idle: Duration) -> Self {
        Self {
            key: key.into(),
            group: None,
            consumer: None,
            block: None,
            start: StreamStart::New,
            mode: ReadMode::Reclaim { min_idle },
            delayed_retry: None,
            dead_letter: None,
        }
    }

    /// A subscription on `key` that claims and reads in one call: `XREADGROUP ... CLAIM` first
    /// takes over the group's entries idle at least `min_idle`, longest idle first, then fills the
    /// rest of the read with fresh ones. One subscription, one consumer, one handler.
    ///
    /// Every delivery carries the server's own
    /// [`IDLE_MS_HEADER`](crate::IDLE_MS_HEADER) and
    /// [`DELIVERY_COUNT_HEADER`](crate::DELIVERY_COUNT_HEADER), both as of the delivery before
    /// this one: a fresh entry reports zero for both, and the first claim back reports one earlier
    /// delivery. The count is therefore the number of attempts that did not finish, which is what
    /// a handler reads to give up on a message itself.
    ///
    /// This is Redis's own redelivery, so this mode is the one that settles a retry natively.
    /// `retry()` leaves the entry in the pending entries list rather than appending a copy, so the
    /// next read brings it back under its own id with the count one higher, no sooner than
    /// `min_idle` from now. `retry_after(delay)` takes the same path and the entry comes back
    /// after `min_idle`, so a delay shorter than that rounds up to it; name a
    /// [`delayed_retry`](Self::delayed_retry) queue to have the delay honoured exactly instead.
    /// The delivery reports its count to the runtime, which is what a `max_attempts` declaration
    /// at the mount site caps.
    ///
    /// `min_idle` has no default and must exceed the longest legitimate handler runtime: below it,
    /// a message a healthy consumer is still working on is taken away and processed twice.
    ///
    /// Redis 8.4.0 added the option. A registration of this mode against an older server refuses
    /// to start, naming both versions; on such a server use [`new`](Self::new) with a
    /// [`reclaim`](Self::reclaim) subscription beside it.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use std::time::Duration;
    ///
    /// use ruststream_fred::stream::prelude::*;
    /// use serde::Deserialize;
    ///
    /// #[derive(Deserialize)]
    /// struct Order {
    ///     id: u64,
    ///     in_stock: bool,
    /// }
    ///
    /// // New orders and the ones a dead worker took and never finished, in one subscription. A
    /// // retry stays pending and comes back after 30s under its own id.
    /// #[subscriber(RedisStream::claiming("orders", Duration::from_secs(30)).group("workers"))]
    /// async fn handle(order: &Order) -> HandlerOutcome {
    ///     if !order.in_stock {
    ///         return HandlerOutcome::retry();
    ///     }
    ///     println!("order {}", order.id);
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
    ///         RedisBroker::standalone("redis://localhost:6379"),
    ///         |b| {
    ///             // The server's delivery count feeds the cap.
    ///             b.include(handle)
    ///                 .max_attempts(nonzero!(5u32))
    ///                 .dead_letter("orders.dlq");
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn claiming(key: impl Into<String>, min_idle: Duration) -> Self {
        Self {
            key: key.into(),
            group: None,
            consumer: None,
            block: None,
            start: StreamStart::New,
            mode: ReadMode::Claiming { min_idle },
            delayed_retry: None,
            dead_letter: None,
        }
    }

    /// Sets the consumer group. Required for every subscription.
    pub fn group(mut self, group: impl Into<String>) -> Self {
        self.group = Some(group.into());
        self
    }

    /// Sets this consumer's name within the group. Defaults to an auto-generated unique name.
    pub fn consumer(mut self, consumer: impl Into<String>) -> Self {
        self.consumer = Some(consumer.into());
        self
    }

    /// How long one read blocks waiting for entries. Defaults to 5 seconds. In the fresh-tail and
    /// claiming modes this is the `XREADGROUP` server-side block, which wakes on a new entry and,
    /// under `CLAIM`, on an entry ageing past the threshold; in reclaim mode `XAUTOCLAIM` does not
    /// block, so this is the poll interval slept between scans that find nothing to reclaim.
    ///
    /// Also reachable at the mount site, after the batch size, through
    /// [`RedisSubscribeExt`](crate::RedisSubscribeExt).
    pub const fn block(mut self, block: Duration) -> Self {
        self.block = Some(block);
        self
    }

    /// Where a newly created group starts reading. Ignored if the group already exists. Only
    /// meaningful for the modes that read the tail, [`new`](Self::new) and
    /// [`claiming`](Self::claiming).
    pub fn start_id(mut self, start: StreamStart) -> Self {
        self.start = start;
        self
    }

    /// Opts this subscription into durable, crash-safe delayed retry backed by a ZSET delay queue.
    ///
    /// Off by default: without it, `retry_after(delay)` / `nack_after(delay)` degrade to the
    /// runtime's broker-agnostic deferred re-publish, which the mount site binds with `out_retry`
    /// and which is at-most-once over the delay window. With it, a delayed delivery is `ZADD`ed to
    /// the named ZSET and replayed from there once due, so the retry survives a process crash. See
    /// [`DelayedRetry`] for the key and TTL requirements.
    ///
    /// The sweeper that replays due entries runs inside this subscription's read loop, so its
    /// granularity is the read [`block`](Self::block) interval.
    pub fn delayed_retry(mut self, retry: DelayedRetry) -> Self {
        self.delayed_retry = Some(retry);
        self
    }

    /// The stream key this subscription reads.
    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    pub(crate) fn group_or_err(&self) -> Result<&str, RedisError> {
        self.group.as_deref().ok_or_else(|| {
            RedisError::InvalidOptions(format!(
                "stream subscription on `{}` requires a consumer group: call .group(name)",
                self.key
            ))
        })
    }

    pub(crate) fn consumer_or_auto(&self) -> String {
        self.consumer.clone().unwrap_or_else(auto_consumer)
    }

    pub(crate) fn block_or_default(&self) -> Duration {
        self.block.unwrap_or(DEFAULT_BLOCK)
    }

    pub(crate) const fn start(&self) -> &StreamStart {
        &self.start
    }

    pub(crate) fn mode(&self) -> ReadMode {
        self.mode.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
    }

    pub(crate) fn delay_config(&self) -> Option<DelayConfig> {
        self.delayed_retry.as_ref().map(DelayConfig::from_retry)
    }

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

    /// What this subscription adds to its channel in the generated `AsyncAPI` document, shared by
    /// both broker forms: the group and consumer it reads through, its read mode and, on the two
    /// modes that claim, the idle threshold they claim at.
    #[cfg(feature = "asyncapi")]
    fn describe(&self) -> Bindings {
        let (mode, min_idle) = match self.mode {
            ReadMode::Fresh => (asyncapi::ReadMode::Fresh, None),
            ReadMode::Reclaim { min_idle } => (asyncapi::ReadMode::Reclaim, Some(min_idle)),
            ReadMode::Claiming { min_idle } => (asyncapi::ReadMode::Claiming, Some(min_idle)),
        };
        asyncapi::channel(&asyncapi::StreamSubscription::new(
            self.group.as_deref(),
            self.consumer.as_deref(),
            mode,
            min_idle.map(|idle| u64::try_from(idle.as_millis()).unwrap_or(u64::MAX)),
        ))
    }
}

impl SubscriptionSource<ConnectedRedisBroker> for RedisStream {
    type Subscriber = RedisSubscriber;
    /// A retry copy is an `XADD` this process makes, and the stream key is where it goes.
    type Copies = AddressedCopies;

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

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

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

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

/// The stream key, in every read mode: an `XADD` there is read by the consumer group this
/// subscription belongs to.
///
/// Under [`reclaim`](RedisStream::reclaim) the copy is read by the group's tail reader rather
/// than by this subscription, which only ever takes entries already pending on another consumer.
/// That is the topology the mode is written for - a `reclaim` subscriber runs beside a
/// [`new`](RedisStream::new) one on the same group - so the copy reaches the handler either way.
impl RedeliveryAddressed<ConnectedRedisBroker> for RedisStream {
    fn redelivery_address(
        &self,
        _connected: &ConnectedRedisBroker,
    ) -> impl Future<Output = Result<RedeliveryAddress, RedisError>> {
        ready(Ok(self.redelivery_key()))
    }
}

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

    #[test]
    fn group_is_required() {
        let err = RedisStream::new("orders").group_or_err().unwrap_err();
        assert!(matches!(err, RedisError::InvalidOptions(msg) if msg.contains("consumer group")));
    }

    #[test]
    fn group_set_resolves() {
        let s = RedisStream::new("orders").group("workers");
        assert_eq!(s.group_or_err().expect("group set"), "workers");
    }

    #[test]
    fn start_maps_to_redis_ids() {
        assert_eq!(StreamStart::New.as_id(), "$");
        assert_eq!(StreamStart::Beginning.as_id(), "0");
        assert_eq!(StreamStart::Id("5-0".into()).as_id(), "5-0");
    }

    #[test]
    fn reclaim_carries_min_idle() {
        let s = RedisStream::reclaim("orders", Duration::from_secs(30)).group("g");
        assert!(matches!(s.mode(), ReadMode::Reclaim { min_idle } if min_idle.as_secs() == 30));
    }

    #[test]
    fn claiming_carries_min_idle() {
        let s = RedisStream::claiming("orders", Duration::from_secs(30)).group("g");
        assert!(matches!(s.mode(), ReadMode::Claiming { min_idle } if min_idle.as_secs() == 30));
    }

    /// Every read mode answers with the stream key: an `XADD` there is read by the group, which
    /// is what a retry copy has to reach.
    #[test]
    fn every_read_mode_addresses_its_copies_to_the_stream_key() {
        let idle = Duration::from_secs(30);
        for source in [
            RedisStream::new("orders"),
            RedisStream::claiming("orders", idle),
            RedisStream::reclaim("orders", idle),
        ] {
            assert_eq!(source.redelivery_key().as_str(), "orders");
        }
    }

    /// The retry a read mode performs is the read mode's own: the claiming mode leaves the entry
    /// pending so its next read claims it back, and the other two append a copy.
    #[test]
    fn the_claiming_mode_retries_by_leaving_the_entry_pending() {
        let idle = Duration::from_secs(30);
        assert_eq!(
            RedisStream::claiming("orders", idle).mode().requeue(),
            RequeueMode::LeavePending { min_idle: idle }
        );
        assert_eq!(
            RedisStream::new("orders").mode().requeue(),
            RequeueMode::Republish
        );
        assert_eq!(
            RedisStream::reclaim("orders", idle).mode().requeue(),
            RequeueMode::Republish
        );
    }
}