photon_backend_sqlite/
port.rs

1//! `SQLite` storage port — write-through persistence with in-memory live fanout.
2
3use std::pin::Pin;
4use std::sync::Arc;
5
6use async_stream::stream;
7use async_trait::async_trait;
8use chrono::Utc;
9use dashmap::DashMap;
10use futures::stream::Stream;
11use serde_json::Value;
12use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
13use sqlx::Row;
14use tokio::sync::broadcast;
15use uuid::Uuid;
16
17use photon_backend::models::Event;
18use photon_backend::{
19    open_stored_event, seal_event_for_storage, topic_filter_matches, PhotonError, Result,
20    StorageCapabilities, StoragePort, TransportCrypto,
21};
22
23use crate::config::sqlite_path_from_env;
24
25fn partition_key(topic_name: &str, topic_key: Option<&str>) -> String {
26    format!("{}:{}", topic_name, topic_key.unwrap_or("__null__"))
27}
28
29fn checkpoint_key(sub: &str, topic: &str, topic_key: Option<&str>) -> String {
30    format!("{sub}:{}:{}", topic, topic_key.unwrap_or("__null__"))
31}
32
33fn map_sqlx(err: sqlx::Error) -> PhotonError {
34    PhotonError::persistence("sqlite", err)
35}
36
37/// Embedded `SQLite` storage for the `sqlite` adapter tier.
38///
39/// Write-through persistence with in-memory live fanout. Enable the `sqlite` feature on the
40/// `photon` public crate and install via
41/// [`PhotonBuilder::storage_port`](https://docs.rs/uf-photon/latest/photon/struct.PhotonBuilder.html#method.storage_port).
42///
43/// Path / env: see module `config` (`PHOTON_SQLITE_PATH`).
44///
45/// Getting started: [Embedded durable](https://docs.rs/uf-photon/latest/photon/#embedded-one-binary).
46/// Runnable: `cargo run -p uf-photon --example embedded_sqlite --features runtime,sqlite`.
47///
48/// # Examples
49///
50/// ## Embedded host (publish + handlers)
51///
52/// One binary owns both publish and `#[subscribe]` dispatch via `start_executor`.
53///
54/// ```rust,ignore
55/// use std::sync::Arc;
56///
57/// use photon_backend_sqlite::SqliteStoragePort;
58/// use photon_core::JsonIdentityFactory;
59/// use photon_runtime::Photon;
60///
61/// # async fn boot() -> photon_backend::Result<()> {
62/// let port = Arc::new(SqliteStoragePort::open("/var/lib/photon/events.db").await?);
63/// let photon = Photon::builder()
64///     .storage_port(port)
65///     .auto_registry()
66///     .build()?;
67/// photon.start_executor(Arc::new(JsonIdentityFactory))?;
68/// // EventType { … }.publish_on(&photon).await?;
69/// # let _ = photon;
70/// # Ok(())
71/// # }
72/// ```
73pub struct SqliteStoragePort {
74    pool: SqlitePool,
75    crypto: TransportCrypto,
76    tx: broadcast::Sender<Event>,
77    events: Arc<DashMap<String, Event>>,
78    delivery_pins: Arc<DashMap<String, i64>>,
79}
80
81impl SqliteStoragePort {
82    /// Open or create a database at `path`.
83    ///
84    /// # Errors
85    ///
86    /// Returns an error if the database cannot be opened or migrated.
87    pub async fn open(path: &str) -> Result<Self> {
88        if path.trim().is_empty() {
89            return Err(PhotonError::Internal(
90                "SQLite database path must not be empty".into(),
91            ));
92        }
93        let options = SqliteConnectOptions::new()
94            .filename(path)
95            .create_if_missing(true);
96        let pool = SqlitePoolOptions::new()
97            .max_connections(5)
98            .connect_with(options)
99            .await
100            .map_err(map_sqlx)?;
101        Self::with_pool(pool).await
102    }
103
104    /// Open using [`sqlite_path_from_env`].
105    ///
106    /// # Errors
107    ///
108    /// Returns an error if the database cannot be opened or migrated.
109    pub async fn from_env() -> Result<Self> {
110        Self::open(&sqlite_path_from_env()).await
111    }
112
113    async fn with_pool(pool: SqlitePool) -> Result<Self> {
114        Self::migrate(&pool).await?;
115        let (tx, _) = broadcast::channel(1024);
116        Ok(Self {
117            pool,
118            crypto: TransportCrypto::from_env()?,
119            tx,
120            events: Arc::new(DashMap::new()),
121            delivery_pins: Arc::new(DashMap::new()),
122        })
123    }
124
125    async fn migrate(pool: &SqlitePool) -> Result<()> {
126        sqlx::query(
127            "CREATE TABLE IF NOT EXISTS events (
128                event_id TEXT PRIMARY KEY,
129                topic_name TEXT NOT NULL,
130                topic_key TEXT,
131                seq INTEGER NOT NULL,
132                actor_json TEXT NOT NULL,
133                payload_json TEXT NOT NULL,
134                created_at TEXT NOT NULL
135            )",
136        )
137        .execute(pool)
138        .await
139        .map_err(map_sqlx)?;
140
141        sqlx::query(
142            "CREATE INDEX IF NOT EXISTS idx_events_topic_seq
143             ON events(topic_name, topic_key, seq)",
144        )
145        .execute(pool)
146        .await
147        .map_err(map_sqlx)?;
148
149        sqlx::query(
150            "CREATE TABLE IF NOT EXISTS checkpoints (
151                checkpoint_key TEXT PRIMARY KEY,
152                last_seq INTEGER NOT NULL
153            )",
154        )
155        .execute(pool)
156        .await
157        .map_err(map_sqlx)?;
158
159        sqlx::query(
160            "CREATE TABLE IF NOT EXISTS seq_counters (
161                partition_key TEXT PRIMARY KEY,
162                next_seq INTEGER NOT NULL
163            )",
164        )
165        .execute(pool)
166        .await
167        .map_err(map_sqlx)?;
168
169        Ok(())
170    }
171
172    async fn next_seq(&self, topic_name: &str, topic_key: Option<&str>) -> Result<i64> {
173        let pk = partition_key(topic_name, topic_key);
174        let row = sqlx::query(
175            "INSERT INTO seq_counters (partition_key, next_seq) VALUES (?, 1)
176             ON CONFLICT(partition_key) DO UPDATE SET next_seq = seq_counters.next_seq + 1
177             RETURNING next_seq",
178        )
179        .bind(&pk)
180        .fetch_one(&self.pool)
181        .await
182        .map_err(map_sqlx)?;
183        Ok(row.get::<i64, _>(0))
184    }
185
186    async fn load_replay_events(
187        pool: &SqlitePool,
188        crypto: &TransportCrypto,
189        topic_name: &str,
190        topic_key_filter: Option<&str>,
191        after_seq: i64,
192    ) -> Result<Vec<Event>> {
193        let rows = if let Some(key) = topic_key_filter {
194            sqlx::query(
195                "SELECT event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at
196                 FROM events
197                 WHERE topic_name = ? AND topic_key = ? AND seq > ?
198                 ORDER BY seq ASC",
199            )
200            .bind(topic_name)
201            .bind(key)
202            .bind(after_seq)
203            .fetch_all(pool)
204            .await
205        } else {
206            sqlx::query(
207                "SELECT event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at
208                 FROM events
209                 WHERE topic_name = ? AND seq > ?
210                 ORDER BY seq ASC",
211            )
212            .bind(topic_name)
213            .bind(after_seq)
214            .fetch_all(pool)
215            .await
216        }
217        .map_err(map_sqlx)?;
218
219        rows.iter()
220            .map(row_to_event)
221            .map(|event| event.and_then(|event| open_stored_event(crypto, event)))
222            .collect()
223    }
224
225    async fn fetch_event_by_id(&self, event_id: &str) -> Result<Option<Event>> {
226        let row = sqlx::query(
227            "SELECT event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at
228             FROM events WHERE event_id = ?",
229        )
230        .bind(event_id)
231        .fetch_optional(&self.pool)
232        .await
233        .map_err(map_sqlx)?;
234        row.as_ref()
235            .map(row_to_event)
236            .transpose()?
237            .map(|event| open_stored_event(&self.crypto, event))
238            .transpose()
239    }
240}
241
242fn row_to_event(row: &sqlx::sqlite::SqliteRow) -> Result<Event> {
243    let created_raw: String = row.get("created_at");
244    let created_at = chrono::DateTime::parse_from_rfc3339(&created_raw)
245        .map_err(|e| PhotonError::persistence("sqlite decode", e))?
246        .with_timezone(&Utc);
247    let actor_json: String = row.get("actor_json");
248    let payload_json: String = row.get("payload_json");
249    Ok(Event {
250        event_id: row.get("event_id"),
251        topic_name: row.get("topic_name"),
252        topic_key: row.get("topic_key"),
253        seq: row.get("seq"),
254        actor_json: serde_json::from_str(&actor_json)?,
255        payload_json: serde_json::from_str(&payload_json)?,
256        created_at,
257    })
258}
259
260#[async_trait]
261impl StoragePort for SqliteStoragePort {
262    fn capabilities(&self) -> StorageCapabilities {
263        StorageCapabilities::sqlite()
264    }
265
266    async fn append(
267        &self,
268        topic_name: &str,
269        topic_key: Option<&str>,
270        actor_json: Value,
271        payload_json: Value,
272    ) -> Result<Event> {
273        let seq = self.next_seq(topic_name, topic_key).await?;
274        let event = Event {
275            event_id: Uuid::new_v4().to_string(),
276            topic_name: topic_name.to_string(),
277            topic_key: topic_key.map(String::from),
278            seq,
279            actor_json,
280            payload_json,
281            created_at: Utc::now(),
282        };
283        let (plain, sealed) = seal_event_for_storage(&self.crypto, event)?;
284
285        sqlx::query(
286            "INSERT INTO events
287             (event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at)
288             VALUES (?, ?, ?, ?, ?, ?, ?)",
289        )
290        .bind(&sealed.event_id)
291        .bind(&sealed.topic_name)
292        .bind(&sealed.topic_key)
293        .bind(sealed.seq)
294        .bind(sealed.actor_json.to_string())
295        .bind(sealed.payload_json.to_string())
296        .bind(sealed.created_at.to_rfc3339())
297        .execute(&self.pool)
298        .await
299        .map_err(map_sqlx)?;
300
301        self.events.insert(sealed.event_id.clone(), sealed);
302        let _ = self.tx.send(plain.clone());
303        Ok(plain)
304    }
305
306    fn subscribe(
307        &self,
308        topic_name: String,
309        topic_key_filter: Option<String>,
310        after_seq: Option<i64>,
311    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
312        let pool = self.pool.clone();
313        let crypto = self.crypto.clone();
314        let mut live_rx = self.tx.subscribe();
315        let topic = topic_name.clone();
316        let filter = topic_key_filter;
317        let delivery_pins = Arc::clone(&self.delivery_pins);
318
319        Box::pin(stream! {
320            if let Some(seq) = after_seq {
321                match Self::load_replay_events(&pool, &crypto, &topic_name, filter.as_deref(), seq)
322                    .await
323                {
324                    Ok(events) => {
325                        for evt in events {
326                            if topic_filter_matches(&evt, &topic, filter.as_ref()) {
327                                yield Ok(evt);
328                            }
329                        }
330                    }
331                    Err(e) => yield Err(e),
332                }
333            }
334
335            loop {
336                match live_rx.recv().await {
337                    Ok(ev) => {
338                        if !topic_filter_matches(&ev, &topic, filter.as_ref()) {
339                            continue;
340                        }
341                        if after_seq.is_some_and(|s| ev.seq <= s) {
342                            continue;
343                        }
344                        let pin_key = partition_key(&ev.topic_name, ev.topic_key.as_deref());
345                        delivery_pins.insert(pin_key, ev.seq);
346                        yield Ok(ev);
347                    }
348                    Err(broadcast::error::RecvError::Lagged(_)) => {
349                        // Live fanout dropped messages; catch up from durable store.
350                        let pin_key = partition_key(&topic, filter.as_deref());
351                        let after = delivery_pins
352                            .get(&pin_key)
353                            .map(|v| *v)
354                            .or(after_seq)
355                            .unwrap_or(0);
356                        match Self::load_replay_events(
357                            &pool,
358                            &crypto,
359                            &topic_name,
360                            filter.as_deref(),
361                            after,
362                        )
363                        .await
364                        {
365                            Ok(events) => {
366                                for evt in events {
367                                    if topic_filter_matches(&evt, &topic, filter.as_ref()) {
368                                        let pk = partition_key(
369                                            &evt.topic_name,
370                                            evt.topic_key.as_deref(),
371                                        );
372                                        delivery_pins.insert(pk, evt.seq);
373                                        yield Ok(evt);
374                                    }
375                                }
376                            }
377                            Err(e) => yield Err(e),
378                        }
379                    }
380                    Err(broadcast::error::RecvError::Closed) => break,
381                }
382            }
383        })
384    }
385
386    async fn get_event(&self, event_id: &str) -> Result<Option<Event>> {
387        if let Some(ev) = self.events.get(event_id) {
388            return Ok(Some(open_stored_event(&self.crypto, ev.clone())?));
389        }
390        self.fetch_event_by_id(event_id).await
391    }
392
393    async fn load_checkpoint(
394        &self,
395        subscription_name: &str,
396        topic_name: &str,
397        topic_key: Option<&str>,
398    ) -> Result<Option<i64>> {
399        let key = checkpoint_key(subscription_name, topic_name, topic_key);
400        let row = sqlx::query("SELECT last_seq FROM checkpoints WHERE checkpoint_key = ?")
401            .bind(&key)
402            .fetch_optional(&self.pool)
403            .await
404            .map_err(map_sqlx)?;
405        Ok(row.map(|r| r.get::<i64, _>(0)))
406    }
407
408    async fn commit_checkpoint(
409        &self,
410        subscription_name: &str,
411        topic_name: &str,
412        topic_key: Option<&str>,
413        last_seq: i64,
414    ) -> Result<()> {
415        let key = checkpoint_key(subscription_name, topic_name, topic_key);
416        sqlx::query(
417            "INSERT INTO checkpoints (checkpoint_key, last_seq) VALUES (?, ?)
418             ON CONFLICT(checkpoint_key) DO UPDATE
419             SET last_seq = MAX(checkpoints.last_seq, excluded.last_seq)",
420        )
421        .bind(&key)
422        .bind(last_seq)
423        .execute(&self.pool)
424        .await
425        .map_err(map_sqlx)?;
426        Ok(())
427    }
428
429    async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
430        let key = partition_key(topic_name, topic_key);
431        self.delivery_pins.get(&key).map(|v| *v)
432    }
433}
434
435#[cfg(test)]
436mod tests {
437    use super::*;
438
439    #[tokio::test]
440    async fn open_rejects_empty_path() {
441        assert!(matches!(
442            SqliteStoragePort::open(" ").await,
443            Err(PhotonError::Internal(_))
444        ));
445    }
446}
photon_backend_sqlite/port.rs

photon_backend_sqlite/
port.rs
