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 load_checkpoint(
236        &self,
237        subscription_name: &str,
238        topic_name: &str,
239        topic_key: Option<&str>,
240    ) -> Result<Option<i64>> {
241        let key = Self::checkpoint_key(subscription_name, topic_name, topic_key);
242        Ok(self.checkpoints.get(&key).map(|v| *v))
243    }
244
245    async fn commit_checkpoint(
246        &self,
247        subscription_name: &str,
248        topic_name: &str,
249        topic_key: Option<&str>,
250        last_seq: i64,
251    ) -> Result<()> {
252        let key = Self::checkpoint_key(subscription_name, topic_name, topic_key);
253        self.checkpoints
254            .entry(key)
255            .and_modify(|existing| *existing = (*existing).max(last_seq))
256            .or_insert(last_seq);
257        Ok(())
258    }
259
260    async fn truncate_before(
261        &self,
262        topic_name: &str,
263        topic_key: Option<&str>,
264        truncate_bound: i64,
265    ) -> Result<u64> {
266        let key = replay_key(topic_name, topic_key);
267        let mut removed = 0u64;
268        let mut buf = self.replay_buffer.write().await;
269        if let Some(queue) = buf.get_mut(&key) {
270            while queue.front().is_some_and(|e| e.seq < truncate_bound) {
271                if let Some(ev) = queue.pop_front() {
272                    self.events.remove(&ev.event_id);
273                    removed += 1;
274                }
275            }
276        }
277        drop(buf);
278        Ok(removed)
279    }
280
281    async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
282        let key = partition_key(topic_name, topic_key);
283        self.delivery_pins.get(&key).map(|v| *v)
284    }
285}
286
287#[cfg(test)]
288mod tests {
289    use super::*;
290
291    #[tokio::test]
292    async fn append_stores_sealed_fields_and_returns_plaintext() {
293        let port = InProcStoragePort::new(TransportCrypto::from_bytes([7; 32]));
294        let marker = "SECRET_PLAINTEXT_MARKER_xyz";
295        let appended = port
296            .append(
297                "test.sealed",
298                None,
299                serde_json::json!({"actor": "test"}),
300                serde_json::json!({"message": marker}),
301            )
302            .await
303            .expect("append");
304
305        assert!(appended.payload_json.to_string().contains(marker));
306        let stored = port.events.get(&appended.event_id).expect("stored event");
307        assert!(!stored.payload_json.to_string().contains(marker));
308
309        let fetched = port
310            .get_event(&appended.event_id)
311            .await
312            .expect("get event")
313            .expect("event");
314        assert_eq!(fetched.payload_json, appended.payload_json);
315    }
316}
photon_backend/storage/in_proc.rs

photon_backend/storage/
in_proc.rs
