photon_backend/storage/
in_proc.rs

1//! In-process storage port — broadcast bus, replay buffer, checkpoints (no external deps).
2//!
3//! Default Embedded adapter. Events stay inside this process — do **not** use for multi-process
4//! or fleet delivery (use a broker adapter; see
5//! [Getting started → Brokered](https://docs.rs/uf-photon/latest/photon/#brokered-publisher--worker-binaries)).
6
7use std::collections::{HashMap, VecDeque};
8use std::pin::Pin;
9use std::sync::Arc;
10
11use async_stream::stream;
12use async_trait::async_trait;
13use chrono::Utc;
14use dashmap::DashMap;
15use futures::stream::Stream;
16use serde_json::Value;
17use tokio::sync::{broadcast, RwLock};
18use uuid::Uuid;
19
20use super::partition::topic_filter_matches;
21use super::port::{StorageCapabilities, StoragePort};
22use crate::error::Result;
23use crate::event::{open_stored_event, seal_event_for_storage, TransportCrypto};
24use crate::models::Event;
25
26const DEFAULT_REPLAY_BUFFER_SIZE: usize = 1000;
27
28fn partition_key(topic_name: &str, topic_key: Option<&str>) -> String {
29    format!("{}:{}", topic_name, topic_key.unwrap_or("__null__"))
30}
31
32fn replay_key(topic_name: &str, topic_key_filter: Option<&str>) -> String {
33    partition_key(topic_name, topic_key_filter)
34}
35
36/// In-process storage for the `mem` adapter tier.
37///
38/// Single-process only: live fanout and a bounded replay buffer live in memory. Photon installs
39/// this by default when you omit
40/// [`PhotonBuilder::storage_port`](https://docs.rs/uf-photon/latest/photon/struct.PhotonBuilder.html#method.storage_port).
41///
42/// **When not to use:** multiple binaries or hosts must share a topic log — pick NATS/Kafka/Fluvio
43/// (Brokered) or `SQLite` for durable single-host Embedded.
44///
45/// Getting started: [Embedded](https://docs.rs/uf-photon/latest/photon/#embedded-one-binary).
46///
47/// # Examples
48///
49/// ## Embedded host (publish + handlers)
50///
51/// One binary owns both publish and `#[subscribe]` dispatch via `start_executor`.
52///
53/// ```rust,ignore
54/// use std::sync::Arc;
55///
56/// use photon_backend::{InProcStoragePort, StoragePort, TransportCrypto};
57/// use photon_core::JsonIdentityFactory;
58/// use photon_runtime::Photon;
59///
60/// # fn main() -> photon_backend::Result<()> {
61/// let port: Arc<dyn StoragePort> = Arc::new(InProcStoragePort::new(
62///     TransportCrypto::from_env()?,
63/// ));
64/// let photon = Photon::builder()
65///     .storage_port(port)
66///     .auto_registry()
67///     .build()?;
68/// photon.start_executor(Arc::new(JsonIdentityFactory))?;
69/// // EventType { … }.publish_on(&photon).await?;
70/// # let _ = photon;
71/// # Ok(())
72/// # }
73/// ```
74///
75/// Default path (same mem port, no explicit `storage_port`):
76/// `Photon::builder().auto_registry().build()?`.
77pub struct InProcStoragePort {
78    crypto: TransportCrypto,
79    tx: broadcast::Sender<Event>,
80    replay_buffer: Arc<RwLock<HashMap<String, VecDeque<Event>>>>,
81    replay_buffer_size: usize,
82    events: Arc<DashMap<String, Event>>,
83    seq_counters: Arc<DashMap<String, i64>>,
84    checkpoints: Arc<DashMap<String, i64>>,
85    delivery_pins: Arc<DashMap<String, i64>>,
86}
87
88impl InProcStoragePort {
89    /// Create a new in-process port with default replay buffer size.
90    #[must_use]
91    pub fn new(crypto: TransportCrypto) -> Self {
92        let (tx, _) = broadcast::channel(1024);
93        Self {
94            crypto,
95            tx,
96            replay_buffer: Arc::new(RwLock::new(HashMap::new())),
97            replay_buffer_size: DEFAULT_REPLAY_BUFFER_SIZE,
98            events: Arc::new(DashMap::new()),
99            seq_counters: Arc::new(DashMap::new()),
100            checkpoints: Arc::new(DashMap::new()),
101            delivery_pins: Arc::new(DashMap::new()),
102        }
103    }
104
105    fn checkpoint_key(sub: &str, topic: &str, topic_key: Option<&str>) -> String {
106        format!("{sub}:{}:{}", topic, topic_key.unwrap_or("__null__"))
107    }
108
109    #[allow(clippy::significant_drop_tightening)]
110    async fn push_replay(&self, event: &Event) {
111        let key = replay_key(&event.topic_name, event.topic_key.as_deref());
112        let event = event.clone();
113        let mut buf = self.replay_buffer.write().await;
114        let queue = buf.entry(key).or_default();
115        queue.push_back(event);
116        while queue.len() > self.replay_buffer_size {
117            queue.pop_front();
118        }
119    }
120
121    fn next_seq(&self, topic_name: &str, topic_key: Option<&str>) -> i64 {
122        let key = partition_key(topic_name, topic_key);
123        let mut entry = self.seq_counters.entry(key).or_insert(0);
124        *entry += 1;
125        *entry
126    }
127}
128
129#[async_trait]
130impl StoragePort for InProcStoragePort {
131    fn capabilities(&self) -> StorageCapabilities {
132        StorageCapabilities::mem()
133    }
134
135    async fn append(
136        &self,
137        topic_name: &str,
138        topic_key: Option<&str>,
139        actor_json: Value,
140        payload_json: Value,
141    ) -> Result<Event> {
142        let seq = self.next_seq(topic_name, topic_key);
143        let event = Event {
144            event_id: Uuid::new_v4().to_string(),
145            topic_name: topic_name.to_string(),
146            topic_key: topic_key.map(String::from),
147            seq,
148            actor_json,
149            payload_json,
150            created_at: Utc::now(),
151        };
152        let (plain, sealed) = seal_event_for_storage(&self.crypto, event)?;
153        self.events.insert(sealed.event_id.clone(), sealed.clone());
154        self.push_replay(&sealed).await;
155        let _ = self.tx.send(plain.clone());
156        Ok(plain)
157    }
158
159    fn subscribe(
160        &self,
161        topic_name: String,
162        topic_key_filter: Option<String>,
163        after_seq: Option<i64>,
164    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
165        let replay_key = replay_key(&topic_name, topic_key_filter.as_deref());
166        let replay_buffer = Arc::clone(&self.replay_buffer);
167        let crypto = self.crypto.clone();
168        let mut live_rx = self.tx.subscribe();
169        let topic = topic_name;
170        let filter = topic_key_filter;
171        let delivery_pins = Arc::clone(&self.delivery_pins);
172
173        Box::pin(stream! {
174            if let Some(seq) = after_seq {
175                let buf = replay_buffer.read().await;
176                if let Some(queue) = buf.get(&replay_key) {
177                    for evt in queue {
178                        if evt.seq > seq && topic_filter_matches(evt, &topic, filter.as_ref()) {
179                            yield open_stored_event(&crypto, evt.clone());
180                        }
181                    }
182                }
183            }
184
185            loop {
186                match live_rx.recv().await {
187                    Ok(ev) => {
188                        if !topic_filter_matches(&ev, &topic, filter.as_ref()) {
189                            continue;
190                        }
191                        if after_seq.is_some_and(|s| ev.seq <= s) {
192                            continue;
193                        }
194                        let pin_key = partition_key(&ev.topic_name, ev.topic_key.as_deref());
195                        delivery_pins.insert(pin_key, ev.seq);
196                        yield Ok(ev);
197                    }
198                    Err(broadcast::error::RecvError::Lagged(_)) => {
199                        // Live fanout dropped messages; catch up from in-memory replay buffer.
200                        let pin_key = partition_key(&topic, filter.as_deref());
201                        let after = delivery_pins
202                            .get(&pin_key)
203                            .map(|v| *v)
204                            .or(after_seq)
205                            .unwrap_or(0);
206                        let buf = replay_buffer.read().await;
207                        if let Some(queue) = buf.get(&replay_key) {
208                            for evt in queue {
209                                if evt.seq > after
210                                    && topic_filter_matches(evt, &topic, filter.as_ref())
211                                {
212                                    let pk = partition_key(
213                                        &evt.topic_name,
214                                        evt.topic_key.as_deref(),
215                                    );
216                                    delivery_pins.insert(pk, evt.seq);
217                                    yield open_stored_event(&crypto, evt.clone());
218                                }
219                            }
220                        }
221                    }
222                    Err(broadcast::error::RecvError::Closed) => break,
223                }
224            }
225        })
226    }
227
228    async fn get_event(&self, event_id: &str) -> Result<Option<Event>> {
229        self.events
230            .get(event_id)
231            .map(|event| open_stored_event(&self.crypto, event.clone()))
232            .transpose()
233    }
234
235    async fn list_by_topic(
236        &self,
237        topic_name: &str,
238        topic_key: Option<&str>,
239        after_seq: Option<i64>,
240        limit: usize,
241    ) -> Result<Vec<Event>> {
242        if limit == 0 {
243            return Ok(Vec::new());
244        }
245        let after = after_seq.unwrap_or(0);
246        let mut matched: Vec<Event> = Vec::new();
247        for entry in self.events.iter() {
248            let sealed = entry.value();
249            if sealed.topic_name != topic_name {
250                continue;
251            }
252            if topic_key.is_some_and(|k| sealed.topic_key.as_deref() != Some(k)) {
253                continue;
254            }
255            if sealed.seq <= after {
256                continue;
257            }
258            matched.push(open_stored_event(&self.crypto, sealed.clone())?);
259        }
260        matched.sort_by_key(|e| e.seq);
261        matched.truncate(limit);
262        Ok(matched)
263    }
264
265    async fn list_recent(&self, limit: usize) -> Result<Vec<Event>> {
266        if limit == 0 {
267            return Ok(Vec::new());
268        }
269        let mut matched: Vec<Event> = Vec::with_capacity(self.events.len().min(limit * 2));
270        for entry in self.events.iter() {
271            matched.push(open_stored_event(&self.crypto, entry.value().clone())?);
272        }
273        matched.sort_by(|a, b| b.created_at.cmp(&a.created_at).then(b.seq.cmp(&a.seq)));
274        matched.truncate(limit);
275        Ok(matched)
276    }
277
278    async fn load_checkpoint(
279        &self,
280        subscription_name: &str,
281        topic_name: &str,
282        topic_key: Option<&str>,
283    ) -> Result<Option<i64>> {
284        let key = Self::checkpoint_key(subscription_name, topic_name, topic_key);
285        Ok(self.checkpoints.get(&key).map(|v| *v))
286    }
287
288    async fn commit_checkpoint(
289        &self,
290        subscription_name: &str,
291        topic_name: &str,
292        topic_key: Option<&str>,
293        last_seq: i64,
294    ) -> Result<()> {
295        let key = Self::checkpoint_key(subscription_name, topic_name, topic_key);
296        self.checkpoints
297            .entry(key)
298            .and_modify(|existing| *existing = (*existing).max(last_seq))
299            .or_insert(last_seq);
300        Ok(())
301    }
302
303    async fn truncate_before(
304        &self,
305        topic_name: &str,
306        topic_key: Option<&str>,
307        truncate_bound: i64,
308    ) -> Result<u64> {
309        let key = replay_key(topic_name, topic_key);
310        let mut removed = 0u64;
311        let mut buf = self.replay_buffer.write().await;
312        if let Some(queue) = buf.get_mut(&key) {
313            while queue.front().is_some_and(|e| e.seq < truncate_bound) {
314                if let Some(ev) = queue.pop_front() {
315                    self.events.remove(&ev.event_id);
316                    removed += 1;
317                }
318            }
319        }
320        drop(buf);
321        Ok(removed)
322    }
323
324    async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
325        let key = partition_key(topic_name, topic_key);
326        self.delivery_pins.get(&key).map(|v| *v)
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333
334    #[tokio::test]
335    async fn append_stores_sealed_fields_and_returns_plaintext() {
336        let port = InProcStoragePort::new(TransportCrypto::from_bytes([7; 32]));
337        let marker = "SECRET_PLAINTEXT_MARKER_xyz";
338        let appended = port
339            .append(
340                "test.sealed",
341                None,
342                serde_json::json!({"actor": "test"}),
343                serde_json::json!({"message": marker}),
344            )
345            .await
346            .expect("append");
347
348        assert!(appended.payload_json.to_string().contains(marker));
349        let stored = port.events.get(&appended.event_id).expect("stored event");
350        assert!(!stored.payload_json.to_string().contains(marker));
351
352        let fetched = port
353            .get_event(&appended.event_id)
354            .await
355            .expect("get event")
356            .expect("event");
357        assert_eq!(fetched.payload_json, appended.payload_json);
358    }
359
360    #[tokio::test]
361    async fn list_by_topic_and_list_recent_honor_limit_and_order() {
362        let port = InProcStoragePort::new(TransportCrypto::from_bytes([9; 32]));
363        let topic = "test.list";
364        let mut ids = Vec::new();
365        for i in 0..5 {
366            let ev = port
367                .append(
368                    topic,
369                    None,
370                    serde_json::json!({}),
371                    serde_json::json!({"n": i}),
372                )
373                .await
374                .expect("append");
375            ids.push(ev.event_id);
376        }
377        let other = port
378            .append(
379                "test.other",
380                None,
381                serde_json::json!({}),
382                serde_json::json!({"n": 99}),
383            )
384            .await
385            .expect("append other");
386
387        let page = port
388            .list_by_topic(topic, None, None, 3)
389            .await
390            .expect("list_by_topic");
391        assert_eq!(page.len(), 3);
392        assert_eq!(page[0].event_id, ids[0]);
393        assert_eq!(page[1].event_id, ids[1]);
394        assert_eq!(page[2].event_id, ids[2]);
395        assert!(page.windows(2).all(|w| w[0].seq < w[1].seq));
396
397        let after = port
398            .list_by_topic(topic, None, Some(2), 10)
399            .await
400            .expect("after_seq");
401        assert!(after.iter().all(|e| e.seq > 2));
402        assert!(after.iter().all(|e| e.topic_name == topic));
403
404        let recent = port.list_recent(2).await.expect("list_recent");
405        assert_eq!(recent.len(), 2);
406        assert!(recent[0].created_at >= recent[1].created_at);
407        assert!(recent
408            .iter()
409            .any(|e| e.event_id == other.event_id || e.topic_name == topic));
410
411        let empty = port
412            .list_by_topic(topic, None, None, 0)
413            .await
414            .expect("zero limit");
415        assert!(empty.is_empty());
416    }
417}
photon_backend/storage/in_proc.rs

photon_backend/storage/
in_proc.rs
