aster_forge_events/
redis_transport.rs

1use async_trait::async_trait;
2use futures::StreamExt;
3use redis::AsyncCommands;
4use std::future::Future;
5use std::sync::Arc;
6use tokio::sync::mpsc;
7use tokio_util::sync::CancellationToken;
8
9use crate::{
10    EventConnectionObservation, EventReconnectPolicy, EventSubscriptionSource,
11    EventSubscriptionUpdate, supervise_event_subscription,
12};
13
14/// Redis event transport errors returned while creating a publisher.
15#[derive(Debug, thiserror::Error)]
16pub enum RedisEventBusError {
17    /// The Redis client could not be created from the configured URL.
18    #[error("open Redis event URL: {0}")]
19    Open(String),
20    /// A configured event topic is empty.
21    #[error("event topic must not be empty")]
22    EmptyTopic,
23    /// Publishing failed.
24    #[error("publish event payload: {0}")]
25    Publish(String),
26    /// Opening a subscription failed.
27    #[error("subscribe to event topic: {0}")]
28    Subscribe(String),
29    /// An active subscription ended unexpectedly.
30    #[error("event subscription stream ended")]
31    StreamEnded,
32}
33
34/// Receives subscriber lifecycle observations.
35pub trait EventConnectionObserver: Send + Sync {
36    /// Records one connection transition.
37    fn observe_event_connection(&self, observation: EventConnectionObservation);
38}
39
40impl<F> EventConnectionObserver for F
41where
42    F: Fn(EventConnectionObservation) + Send + Sync,
43{
44    fn observe_event_connection(&self, observation: EventConnectionObservation) {
45        self(observation);
46    }
47}
48
49/// Backwards-compatible name for the shared event reconnect policy.
50pub type RedisEventReconnectPolicy = EventReconnectPolicy;
51
52/// Redis-backed transient event publisher and reconnecting subscriber.
53#[derive(Clone)]
54pub struct RedisEventBus {
55    client: redis::Client,
56    topic: String,
57    reconnect_policy: EventReconnectPolicy,
58}
59
60/// One active Redis Pub/Sub subscription.
61pub struct RedisEventSubscription {
62    pubsub: redis::aio::PubSub,
63}
64
65impl RedisEventBus {
66    /// Creates a bus from a Redis URL and logical topic.
67    ///
68    /// # Errors
69    ///
70    /// Returns an error when the topic is blank or the Redis client rejects `url`.
71    pub fn from_url(url: &str, topic: impl Into<String>) -> Result<Self, RedisEventBusError> {
72        let topic = topic.into();
73        if topic.trim().is_empty() {
74            return Err(RedisEventBusError::EmptyTopic);
75        }
76        let client = redis::Client::open(url)
77            .map_err(|error| RedisEventBusError::Open(error.to_string()))?;
78        Ok(Self {
79            client,
80            topic,
81            reconnect_policy: EventReconnectPolicy::default(),
82        })
83    }
84
85    /// Creates a bus from a base Redis URL and raw credentials.
86    ///
87    /// The base URL must not contain userinfo. `username = None` with a password creates the
88    /// password-only URL form used by Redis deployments without ACL usernames.
89    ///
90    /// # Errors
91    ///
92    /// Returns an error when the topic is blank, the base URL or credential combination is
93    /// invalid, or the Redis client rejects the resulting URL.
94    pub fn from_credentials(
95        base_url: &str,
96        username: Option<&str>,
97        password: Option<&str>,
98        topic: impl Into<String>,
99    ) -> Result<Self, RedisEventBusError> {
100        let topic = topic.into();
101        if topic.trim().is_empty() {
102            return Err(RedisEventBusError::EmptyTopic);
103        }
104        let url = aster_forge_utils::url::url_with_credentials(
105            base_url,
106            username,
107            password,
108            "Redis event base URL",
109        )
110        .map_err(|error| RedisEventBusError::Open(error.to_string()))?;
111        let client = redis::Client::open(url).map_err(|_| {
112            RedisEventBusError::Open("invalid Redis event connection configuration".to_string())
113        })?;
114        Ok(Self {
115            client,
116            topic,
117            reconnect_policy: EventReconnectPolicy::default(),
118        })
119    }
120
121    /// Creates a bus from an existing Redis client.
122    pub fn from_client(client: redis::Client, topic: impl Into<String>) -> Self {
123        Self {
124            client,
125            topic: topic.into(),
126            reconnect_policy: EventReconnectPolicy::default(),
127        }
128    }
129
130    /// Overrides the reconnect policy.
131    #[must_use]
132    pub fn with_reconnect_policy(mut self, policy: RedisEventReconnectPolicy) -> Self {
133        self.reconnect_policy = policy;
134        self
135    }
136
137    /// Returns the configured logical topic.
138    #[must_use]
139    pub fn topic(&self) -> &str {
140        &self.topic
141    }
142
143    /// Publishes one opaque payload. Payload interpretation belongs to the product layer.
144    ///
145    /// # Errors
146    ///
147    /// Returns an error when a Redis connection cannot be opened or the publish command fails.
148    pub async fn publish(&self, payload: impl Into<String>) -> Result<(), RedisEventBusError> {
149        let mut connection = self
150            .client
151            .get_multiplexed_async_connection()
152            .await
153            .map_err(|error| RedisEventBusError::Publish(error.to_string()))?;
154        let _: usize = connection
155            .publish(&self.topic, payload.into())
156            .await
157            .map_err(|error| RedisEventBusError::Publish(error.to_string()))?;
158        Ok(())
159    }
160
161    /// Opens one Redis subscription attempt.
162    ///
163    /// # Errors
164    ///
165    /// Returns an error when the Pub/Sub connection cannot be opened or the topic subscription
166    /// command fails.
167    pub async fn subscribe(&self) -> Result<RedisEventSubscription, RedisEventBusError> {
168        let mut pubsub = self
169            .client
170            .get_async_pubsub()
171            .await
172            .map_err(|error| RedisEventBusError::Subscribe(error.to_string()))?;
173        pubsub
174            .subscribe(&self.topic)
175            .await
176            .map_err(|error| RedisEventBusError::Subscribe(error.to_string()))?;
177        Ok(RedisEventSubscription { pubsub })
178    }
179
180    /// Runs a reconnecting subscription until shutdown is cancelled.
181    ///
182    /// Malformed Redis payloads are logged and skipped. The callback is responsible for decoding
183    /// the product payload and deciding whether an event belongs to the current runtime.
184    pub async fn run_subscription<F, Fut>(
185        &self,
186        shutdown: CancellationToken,
187        observer: Option<&dyn EventConnectionObserver>,
188        mut on_payload: F,
189    ) where
190        F: FnMut(String) -> Fut,
191        Fut: Future<Output = ()>,
192    {
193        let (updates_tx, mut updates_rx) = mpsc::channel(1);
194        let supervisor = supervise_event_subscription(
195            Arc::new(self.clone()),
196            self.reconnect_policy,
197            shutdown.clone(),
198            updates_tx,
199        );
200        tokio::pin!(supervisor);
201
202        loop {
203            let update = tokio::select! {
204                () = shutdown.cancelled() => return,
205                () = &mut supervisor => return,
206                update = updates_rx.recv() => update,
207            };
208            match update {
209                Some(EventSubscriptionUpdate::Connection(observation)) => {
210                    if let Some(observer) = observer {
211                        observer.observe_event_connection(observation);
212                    }
213                }
214                Some(EventSubscriptionUpdate::Item(payload)) => on_payload(payload).await,
215                None => return,
216            }
217        }
218    }
219}
220
221impl RedisEventSubscription {
222    /// Receives one raw payload. Redis payload conversion failures are logged and skipped.
223    ///
224    /// # Errors
225    ///
226    /// Returns [`RedisEventBusError::StreamEnded`] when the active Pub/Sub stream terminates.
227    pub async fn receive(&mut self) -> Result<String, RedisEventBusError> {
228        loop {
229            let mut stream = self.pubsub.on_message();
230            let Some(message) = stream.next().await else {
231                return Err(RedisEventBusError::StreamEnded);
232            };
233            match message.get_payload::<String>() {
234                Ok(payload) => return Ok(payload),
235                Err(error) => tracing::warn!(%error, "failed to decode Redis event payload"),
236            }
237        }
238    }
239}
240
241#[async_trait]
242impl EventSubscriptionSource for RedisEventBus {
243    type Item = String;
244    type Subscription = RedisEventSubscription;
245    type Error = RedisEventBusError;
246
247    async fn subscribe(&self) -> Result<Self::Subscription, Self::Error> {
248        RedisEventBus::subscribe(self).await
249    }
250
251    async fn receive(
252        &self,
253        subscription: &mut Self::Subscription,
254    ) -> Result<Self::Item, Self::Error> {
255        subscription.receive().await
256    }
257}
258
259#[cfg(test)]
260mod tests {
261    use super::{RedisEventBus, RedisEventBusError};
262
263    #[test]
264    fn rejects_empty_topics() {
265        assert!(matches!(
266            RedisEventBus::from_url("redis://127.0.0.1", "  "),
267            Err(RedisEventBusError::EmptyTopic)
268        ));
269    }
270
271    #[test]
272    fn credentialed_bus_accepts_reserved_password_characters() {
273        let bus = RedisEventBus::from_credentials(
274            "redis://cache.example:6379/2?protocol=resp3",
275            None,
276            Some("#[]{}^+=*@:/?%\u{5bc6}\u{7801}"),
277            "aster.events",
278        )
279        .unwrap();
280
281        assert_eq!(bus.topic(), "aster.events");
282    }
283
284    #[test]
285    fn credentialed_bus_rejects_conflicting_userinfo_without_secret_leak() {
286        let raw_password = "raw#event-secret";
287        let result = RedisEventBus::from_credentials(
288            "redis://existing@cache.example:6379/0",
289            Some("replacement"),
290            Some(raw_password),
291            "aster.events",
292        );
293        let Err(error) = result else {
294            panic!("conflicting Redis credentials should be rejected");
295        };
296
297        assert!(matches!(error, RedisEventBusError::Open(_)));
298        assert!(
299            error
300                .to_string()
301                .contains("must not already include userinfo")
302        );
303        assert!(!error.to_string().contains(raw_password));
304    }
305}
aster_forge_events/redis_transport.rs

aster_forge_events/
redis_transport.rs
