photon_backend/storage/
in_proc.rs

1//! In-process storage port — broadcast bus, replay buffer, checkpoints (no external deps).
2//!
3//! Default Mode 1 adapter. Events stay inside this process — do **not** use for multi-process
4//! or fleet delivery (use a broker adapter; see
5//! [Getting started → Mode 2](https://docs.rs/uf-photon/latest/photon/#mode-2--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::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/// (Mode 2) or `SQLite` for durable single-host Mode 1.
44///
45/// Getting started: [Mode 1](https://docs.rs/uf-photon/latest/photon/#mode-1--embedded-one-binary).
46///
47/// # Examples
48///
49/// ## Mode 1 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 _ = self.crypto.encrypt(&actor_json, &payload_json)?;
143        let seq = self.next_seq(topic_name, topic_key);
144        let event = Event {
145            event_id: Uuid::new_v4().to_string(),
146            topic_name: topic_name.to_string(),
147            topic_key: topic_key.map(String::from),
148            seq,
149            actor_json,
150            payload_json,
151            created_at: Utc::now(),
152        };
153        self.events
154            .insert(event.event_id.clone(), event.clone());
155        self.push_replay(&event).await;
156        let _ = self.tx.send(event.clone());
157        Ok(event)
158    }
159
160    fn subscribe(
161        &self,
162        topic_name: String,
163        topic_key_filter: Option<String>,
164        after_seq: Option<i64>,
165    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
166        let replay_key = replay_key(&topic_name, topic_key_filter.as_deref());
167        let replay_buffer = Arc::clone(&self.replay_buffer);
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 Ok(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 Ok(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        Ok(self.events.get(event_id).map(|e| e.clone()))
230    }
231
232    async fn load_checkpoint(
233        &self,
234        subscription_name: &str,
235        topic_name: &str,
236        topic_key: Option<&str>,
237    ) -> Result<Option<i64>> {
238        let key = Self::checkpoint_key(subscription_name, topic_name, topic_key);
239        Ok(self.checkpoints.get(&key).map(|v| *v))
240    }
241
242    async fn commit_checkpoint(
243        &self,
244        subscription_name: &str,
245        topic_name: &str,
246        topic_key: Option<&str>,
247        last_seq: i64,
248    ) -> Result<()> {
249        let key = Self::checkpoint_key(subscription_name, topic_name, topic_key);
250        self.checkpoints.insert(key, last_seq);
251        Ok(())
252    }
253
254    async fn truncate_before(
255        &self,
256        topic_name: &str,
257        topic_key: Option<&str>,
258        truncate_bound: i64,
259    ) -> Result<u64> {
260        let key = replay_key(topic_name, topic_key);
261        let mut removed = 0u64;
262        let mut buf = self.replay_buffer.write().await;
263        if let Some(queue) = buf.get_mut(&key) {
264            while queue
265                .front()
266                .is_some_and(|e| e.seq < truncate_bound)
267            {
268                if let Some(ev) = queue.pop_front() {
269                    self.events.remove(&ev.event_id);
270                    removed += 1;
271                }
272            }
273        }
274        drop(buf);
275        Ok(removed)
276    }
277
278    async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
279        let key = partition_key(topic_name, topic_key);
280        self.delivery_pins.get(&key).map(|v| *v)
281    }
282}
photon_backend/storage/in_proc.rs

photon_backend/storage/
in_proc.rs
