pulses 0.2.0

//! Redis Streams broker backend.
//!
//! Uses consumer groups (`XREADGROUP`) for at-least-once delivery, `XAUTOCLAIM`
//! to reclaim messages abandoned by crashed consumers, and a per-stream
//! `<stream>-dlq` stream for dead letters. A single auto-reconnecting
//! [`ConnectionManager`] is reused across all operations.

use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use redis::AsyncCommands;
use redis::RedisError;
use redis::RedisResult;
use redis::Value;
use redis::aio::ConnectionManager;
use redis::streams::StreamAutoClaimOptions;
use redis::streams::StreamAutoClaimReply;
use redis::streams::StreamId;
use redis::streams::StreamReadOptions;
use redis::streams::StreamReadReply;
use smallvec::SmallVec;

use crate::core::Broker;
use crate::core::Envelope;
use crate::core::Subscription;

/// Suffix appended to a stream name to form its dead-letter stream.
const DEAD_LETTER_SUFFIX: &str = "-dlq";
/// Field name used to carry the payload in dead-letter entries.
const DLQ_PAYLOAD_FIELD: &[u8] = b"payload";
/// Maximum `XAUTOCLAIM` pages walked per stream per reclaim cycle, bounding the
/// work done in a single pass over a large pending list.
const MAX_RECLAIM_PAGES: usize = 64;

/// Consumer-group subscription parameters for [`RedisBroker`].
///
/// `streams` is populated automatically by the runtime from the registered
/// handlers, so callers usually only set the group and consumer (e.g. via
/// [`RedisSubscription::for_group`]).
#[derive(Debug, Clone)]
pub struct RedisSubscription {
    /// Streams to consume. Filled in by the runtime from handler `STREAMS`.
    pub streams: Vec<Arc<str>>,
    /// Consumer group name.
    pub group: Arc<str>,
    /// This consumer's name within the group.
    pub consumer: Arc<str>,
    /// How long `XREADGROUP` blocks waiting for messages (`0` = non-blocking).
    pub block: Duration,
    /// Maximum messages to read per poll.
    pub count: usize,
    /// Entry field that carries the message payload; every other field becomes
    /// an [`Envelope`] header.
    pub payload_field: Bytes,
    /// Minimum idle time before a pending message is eligible for reclaim.
    ///
    /// Must exceed the maximum in-process handler retry time (see
    /// [`Outcome::Retry`](crate::Outcome::Retry)); otherwise the reclaimer may
    /// redeliver a message that a handler is still retrying.
    pub min_idle: Duration,
    /// Maximum messages reclaimed per `XAUTOCLAIM` page.
    pub reclaim_count: usize,
}

impl Default for RedisSubscription {
    fn default() -> Self {
        Self {
            streams: Vec::new(),
            group: Arc::from("default-group"),
            consumer: Arc::from("default-consumer"),
            block: Duration::from_secs(2),
            count: 256,
            payload_field: Bytes::from_static(b"payload"),
            min_idle: Duration::from_secs(60),
            reclaim_count: 256,
        }
    }
}

impl RedisSubscription {
    /// Convenience constructor selecting a consumer group and consumer name,
    /// leaving all other fields at their defaults.
    #[must_use]
    pub fn for_group(group: impl Into<Arc<str>>, consumer: impl Into<Arc<str>>) -> Self {
        Self { group: group.into(), consumer: consumer.into(), ..Self::default() }
    }
}

impl Subscription for RedisSubscription {
    fn set_streams(&mut self, streams: Vec<Arc<str>>) { self.streams = streams; }
}

/// Token identifying a delivered Redis Streams entry for acknowledgement.
///
/// Opaque by design (it is meant to be handed back to the runtime), but its
/// components are exposed read-only for logging and assertions.
#[derive(Debug, Clone)]
pub struct RedisAckToken {
    pub(crate) stream: Arc<str>,
    pub(crate) group: Arc<str>,
    pub(crate) id: Arc<str>,
}

impl RedisAckToken {
    /// Source stream of the message.
    #[must_use]
    pub fn stream(&self) -> &str { &self.stream }

    /// Consumer group the message was delivered to.
    #[must_use]
    pub fn group(&self) -> &str { &self.group }

    /// Stream entry id.
    #[must_use]
    pub fn id(&self) -> &str { &self.id }
}

/// A [`Broker`] backed by Redis Streams.
#[derive(Clone)]
pub struct RedisBroker {
    connection: ConnectionManager,
}

impl std::fmt::Debug for RedisBroker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Intentionally opaque: the connection URL may embed credentials.
        f.debug_struct("RedisBroker").finish_non_exhaustive()
    }
}

impl RedisBroker {
    /// Connect to Redis and establish a multiplexed, auto-reconnecting
    /// connection. The URL (which may contain credentials) is not retained.
    pub async fn connect(url: impl AsRef<str>) -> RedisResult<Self> {
        let client = redis::Client::open(url.as_ref())?;
        let connection = client.get_connection_manager().await?;
        Ok(Self { connection })
    }

    /// Issue a `PING`, returning the server's reply.
    pub async fn ping(&self) -> RedisResult<String> {
        let mut connection = self.connection.clone();
        redis::cmd("PING").query_async(&mut connection).await
    }

    /// Idempotently create the consumer group on every subscribed stream
    /// (`MKSTREAM`), ignoring `BUSYGROUP` (already exists) but propagating
    /// genuine failures.
    async fn ensure_groups(&self, sub: &RedisSubscription) -> RedisResult<()> {
        let mut connection = self.connection.clone();
        for stream in &sub.streams {
            let result: RedisResult<()> =
                connection.xgroup_create_mkstream(stream.as_ref(), sub.group.as_ref(), "0").await;

            if let Err(error) = result
                && error.code() != Some("BUSYGROUP")
            {
                return Err(error);
            }
        }
        Ok(())
    }
}

impl Broker for RedisBroker {
    type AckToken = RedisAckToken;
    type Subscription = RedisSubscription;
    type Error = RedisError;

    async fn poll(&self, sub: &Self::Subscription) -> Result<Vec<(Envelope, RedisAckToken)>, RedisError> {
        if sub.streams.is_empty() {
            return Ok(Vec::new());
        }

        let mut connection = self.connection.clone();
        let keys: Vec<&str> = sub.streams.iter().map(|stream| stream.as_ref()).collect();
        let ids: Vec<&str> = vec![">"; keys.len()];

        let mut options = StreamReadOptions::default().group(sub.group.as_ref(), sub.consumer.as_ref());
        if sub.count > 0 {
            options = options.count(sub.count);
        }
        let block_ms = duration_to_millis(sub.block);
        if block_ms > 0 {
            options = options.block(block_ms);
        }

        let reply: Option<StreamReadReply> = match connection.xread_options(&keys, &ids, &options).await {
            Ok(reply) => reply,
            Err(error) if is_nogroup(&error) => {
                self.ensure_groups(sub).await?;
                connection.xread_options(&keys, &ids, &options).await?
            }
            Err(error) => return Err(error),
        };

        let Some(reply) = reply else {
            return Ok(Vec::new());
        };

        let mut messages = Vec::new();
        for stream_key in reply.keys {
            let stream: Arc<str> = Arc::from(stream_key.key.as_str());
            for entry in stream_key.ids {
                messages.push(entry_to_message(&stream, entry, sub));
            }
        }
        Ok(messages)
    }

    async fn ack(&self, token: RedisAckToken) -> Result<(), RedisError> {
        let mut connection = self.connection.clone();
        let acknowledged: i64 =
            connection.xack(token.stream.as_ref(), token.group.as_ref(), &[token.id.as_ref()]).await?;
        if acknowledged == 0 {
            tracing::debug!(stream = %token.stream, group = %token.group, id = %token.id, "XACK matched no entry");
        }
        Ok(())
    }

    async fn reclaim(&self, sub: &Self::Subscription) -> Result<Vec<(Envelope, RedisAckToken)>, RedisError> {
        if sub.streams.is_empty() {
            return Ok(Vec::new());
        }

        let mut connection = self.connection.clone();
        let min_idle_ms = duration_to_millis(sub.min_idle);
        let mut messages = Vec::new();

        for stream in &sub.streams {
            let mut cursor = String::from("0-0");
            for _ in 0..MAX_RECLAIM_PAGES {
                let options = StreamAutoClaimOptions::default().count(sub.reclaim_count.max(1));
                let reply: StreamAutoClaimReply = match connection
                    .xautoclaim_options(
                        stream.as_ref(),
                        sub.group.as_ref(),
                        sub.consumer.as_ref(),
                        min_idle_ms,
                        cursor.as_str(),
                        options,
                    )
                    .await
                {
                    Ok(reply) => reply,
                    Err(error) if is_nogroup(&error) => {
                        self.ensure_groups(sub).await?;
                        break;
                    }
                    Err(error) => return Err(error),
                };

                let next_cursor = reply.next_stream_id.clone();

                // Entries that were XDEL'd while still pending stay in the PEL
                // forever unless acked; ack them best-effort so the PEL drains.
                if !reply.deleted_ids.is_empty() {
                    let deleted: Vec<&str> = reply.deleted_ids.iter().map(String::as_str).collect();
                    let acked: RedisResult<i64> = connection.xack(stream.as_ref(), sub.group.as_ref(), &deleted).await;
                    if let Err(error) = acked {
                        tracing::warn!(%error, stream = %stream, "failed to ack deleted PEL entries");
                    }
                }

                for entry in reply.claimed {
                    messages.push(entry_to_message(stream, entry, sub));
                }

                if next_cursor.is_empty() || next_cursor == "0-0" {
                    break;
                }
                cursor = next_cursor;
            }
        }
        Ok(messages)
    }

    async fn dead_letter(&self, envelope: &Envelope, reason: &str) -> Result<(), RedisError> {
        let mut connection = self.connection.clone();
        let dlq_stream = format!("{}{}", envelope.stream, DEAD_LETTER_SUFFIX);

        const RESERVED: [&[u8]; 4] = [DLQ_PAYLOAD_FIELD, b"x-dead-reason", b"x-original-stream", b"x-original-id"];

        let mut fields: Vec<(&[u8], &[u8])> = Vec::with_capacity(envelope.headers.len() + 4);
        fields.push((DLQ_PAYLOAD_FIELD, envelope.payload.as_ref()));
        for (key, value) in &envelope.headers {
            // Skip headers that would collide with reserved DLQ metadata fields.
            if RESERVED.contains(&key.as_ref()) {
                continue;
            }
            fields.push((key.as_ref(), value.as_ref()));
        }
        fields.push((b"x-dead-reason", reason.as_bytes()));
        fields.push((b"x-original-stream", envelope.stream.as_bytes()));
        fields.push((b"x-original-id", envelope.id.as_bytes()));

        let _: String = connection.xadd(dlq_stream.as_str(), "*", &fields).await?;
        Ok(())
    }
}

fn entry_to_message(stream: &Arc<str>, entry: StreamId, sub: &RedisSubscription) -> (Envelope, RedisAckToken) {
    let id: Arc<str> = Arc::from(entry.id.as_str());
    let ts_unix_ms = parse_timestamp(&id);

    let mut payload = Bytes::new();
    let mut payload_set = false;
    let mut headers: SmallVec<[(Bytes, Bytes); 8]> = SmallVec::new();

    for (field, value) in entry.map {
        let value_bytes = value_to_bytes(value);
        if !payload_set && field.as_bytes() == sub.payload_field.as_ref() {
            payload = value_bytes;
            payload_set = true;
        } else {
            headers.push((Bytes::from(field.into_bytes()), value_bytes));
        }
    }

    let envelope = Envelope { stream: stream.clone(), id: id.clone(), payload, headers, ts_unix_ms };
    let token = RedisAckToken { stream: stream.clone(), group: sub.group.clone(), id };
    (envelope, token)
}

fn value_to_bytes(value: Value) -> Bytes {
    match value {
        Value::BulkString(bytes) => Bytes::from(bytes),
        Value::SimpleString(text) => Bytes::from(text.into_bytes()),
        Value::Int(number) => Bytes::from(number.to_string().into_bytes()),
        Value::Nil => Bytes::new(),
        other => {
            tracing::warn!("unexpected redis value variant in stream entry; coercing via debug representation");
            Bytes::from(format!("{other:?}").into_bytes())
        }
    }
}

fn parse_timestamp(id: &str) -> u64 { id.split_once('-').and_then(|(millis, _)| millis.parse().ok()).unwrap_or(0) }

fn is_nogroup(error: &RedisError) -> bool { error.code() == Some("NOGROUP") }

fn duration_to_millis(duration: Duration) -> usize { duration.as_millis().min(usize::MAX as u128) as usize }

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use super::*;

    fn subscription() -> RedisSubscription {
        RedisSubscription { payload_field: Bytes::from_static(b"body"), ..RedisSubscription::default() }
    }

    #[test]
    fn parses_timestamp_from_entry_id() {
        assert_eq!(parse_timestamp("1731016450000-1"), 1_731_016_450_000);
        assert_eq!(parse_timestamp("malformed"), 0);
    }

    #[test]
    fn value_conversions_never_panic() {
        assert_eq!(value_to_bytes(Value::BulkString(b"hi".to_vec())).as_ref(), b"hi");
        assert_eq!(value_to_bytes(Value::Int(42)).as_ref(), b"42");
        assert_eq!(value_to_bytes(Value::Nil).as_ref(), b"");
    }

    #[test]
    fn entry_splits_payload_from_headers() {
        let mut map = HashMap::new();
        map.insert("body".to_string(), Value::BulkString(b"{\"x\":1}".to_vec()));
        map.insert("tenant".to_string(), Value::BulkString(b"t1".to_vec()));
        let entry = StreamId { id: "1731016450000-1".to_string(), map, ..StreamId::default() };

        let stream: Arc<str> = Arc::from("orders");
        let (envelope, token) = entry_to_message(&stream, entry, &subscription());

        assert_eq!(envelope.stream.as_ref(), "orders");
        assert_eq!(envelope.id.as_ref(), "1731016450000-1");
        assert_eq!(envelope.payload.as_ref(), b"{\"x\":1}");
        assert_eq!(envelope.ts_unix_ms, 1_731_016_450_000);
        assert_eq!(envelope.headers.len(), 1);
        assert_eq!(envelope.headers[0].0.as_ref(), b"tenant");
        assert_eq!(token.id.as_ref(), "1731016450000-1");
        assert_eq!(token.group.as_ref(), "default-group");
    }

    #[test]
    fn for_group_sets_names() {
        let sub = RedisSubscription::for_group("g", "c");
        assert_eq!(sub.group.as_ref(), "g");
        assert_eq!(sub.consumer.as_ref(), "c");
        assert!(sub.streams.is_empty());
    }
}