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    async fn fetch_list_by_topic(
242        &self,
243        topic_name: &str,
244        topic_key: Option<&str>,
245        after_seq: Option<i64>,
246        limit: usize,
247    ) -> Result<Vec<Event>> {
248        if limit == 0 {
249            return Ok(Vec::new());
250        }
251        let after = after_seq.unwrap_or(0);
252        let limit_i = i64::try_from(limit).unwrap_or(i64::MAX);
253        let rows = if let Some(key) = topic_key {
254            sqlx::query(
255                "SELECT event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at
256                 FROM events
257                 WHERE topic_name = ? AND topic_key = ? AND seq > ?
258                 ORDER BY seq ASC
259                 LIMIT ?",
260            )
261            .bind(topic_name)
262            .bind(key)
263            .bind(after)
264            .bind(limit_i)
265            .fetch_all(&self.pool)
266            .await
267        } else {
268            sqlx::query(
269                "SELECT event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at
270                 FROM events
271                 WHERE topic_name = ? AND seq > ?
272                 ORDER BY seq ASC
273                 LIMIT ?",
274            )
275            .bind(topic_name)
276            .bind(after)
277            .bind(limit_i)
278            .fetch_all(&self.pool)
279            .await
280        }
281        .map_err(map_sqlx)?;
282
283        rows.iter()
284            .map(row_to_event)
285            .map(|event| event.and_then(|event| open_stored_event(&self.crypto, event)))
286            .collect()
287    }
288
289    async fn fetch_list_recent(&self, limit: usize) -> Result<Vec<Event>> {
290        if limit == 0 {
291            return Ok(Vec::new());
292        }
293        let limit_i = i64::try_from(limit).unwrap_or(i64::MAX);
294        let rows = sqlx::query(
295            "SELECT event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at
296             FROM events
297             ORDER BY created_at DESC, seq DESC
298             LIMIT ?",
299        )
300        .bind(limit_i)
301        .fetch_all(&self.pool)
302        .await
303        .map_err(map_sqlx)?;
304
305        rows.iter()
306            .map(row_to_event)
307            .map(|event| event.and_then(|event| open_stored_event(&self.crypto, event)))
308            .collect()
309    }
310}
311
312fn row_to_event(row: &sqlx::sqlite::SqliteRow) -> Result<Event> {
313    let created_raw: String = row.get("created_at");
314    let created_at = chrono::DateTime::parse_from_rfc3339(&created_raw)
315        .map_err(|e| PhotonError::persistence("sqlite decode", e))?
316        .with_timezone(&Utc);
317    let actor_json: String = row.get("actor_json");
318    let payload_json: String = row.get("payload_json");
319    Ok(Event {
320        event_id: row.get("event_id"),
321        topic_name: row.get("topic_name"),
322        topic_key: row.get("topic_key"),
323        seq: row.get("seq"),
324        actor_json: serde_json::from_str(&actor_json)?,
325        payload_json: serde_json::from_str(&payload_json)?,
326        created_at,
327    })
328}
329
330#[async_trait]
331impl StoragePort for SqliteStoragePort {
332    fn capabilities(&self) -> StorageCapabilities {
333        StorageCapabilities::sqlite()
334    }
335
336    async fn append(
337        &self,
338        topic_name: &str,
339        topic_key: Option<&str>,
340        actor_json: Value,
341        payload_json: Value,
342    ) -> Result<Event> {
343        let seq = self.next_seq(topic_name, topic_key).await?;
344        let event = Event {
345            event_id: Uuid::new_v4().to_string(),
346            topic_name: topic_name.to_string(),
347            topic_key: topic_key.map(String::from),
348            seq,
349            actor_json,
350            payload_json,
351            created_at: Utc::now(),
352        };
353        let (plain, sealed) = seal_event_for_storage(&self.crypto, event)?;
354
355        sqlx::query(
356            "INSERT INTO events
357             (event_id, topic_name, topic_key, seq, actor_json, payload_json, created_at)
358             VALUES (?, ?, ?, ?, ?, ?, ?)",
359        )
360        .bind(&sealed.event_id)
361        .bind(&sealed.topic_name)
362        .bind(&sealed.topic_key)
363        .bind(sealed.seq)
364        .bind(sealed.actor_json.to_string())
365        .bind(sealed.payload_json.to_string())
366        .bind(sealed.created_at.to_rfc3339())
367        .execute(&self.pool)
368        .await
369        .map_err(map_sqlx)?;
370
371        self.events.insert(sealed.event_id.clone(), sealed);
372        let _ = self.tx.send(plain.clone());
373        Ok(plain)
374    }
375
376    fn subscribe(
377        &self,
378        topic_name: String,
379        topic_key_filter: Option<String>,
380        after_seq: Option<i64>,
381    ) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
382        let pool = self.pool.clone();
383        let crypto = self.crypto.clone();
384        let mut live_rx = self.tx.subscribe();
385        let topic = topic_name.clone();
386        let filter = topic_key_filter;
387        let delivery_pins = Arc::clone(&self.delivery_pins);
388
389        Box::pin(stream! {
390            if let Some(seq) = after_seq {
391                match Self::load_replay_events(&pool, &crypto, &topic_name, filter.as_deref(), seq)
392                    .await
393                {
394                    Ok(events) => {
395                        for evt in events {
396                            if topic_filter_matches(&evt, &topic, filter.as_ref()) {
397                                yield Ok(evt);
398                            }
399                        }
400                    }
401                    Err(e) => yield Err(e),
402                }
403            }
404
405            loop {
406                match live_rx.recv().await {
407                    Ok(ev) => {
408                        if !topic_filter_matches(&ev, &topic, filter.as_ref()) {
409                            continue;
410                        }
411                        if after_seq.is_some_and(|s| ev.seq <= s) {
412                            continue;
413                        }
414                        let pin_key = partition_key(&ev.topic_name, ev.topic_key.as_deref());
415                        delivery_pins.insert(pin_key, ev.seq);
416                        yield Ok(ev);
417                    }
418                    Err(broadcast::error::RecvError::Lagged(_)) => {
419                        // Live fanout dropped messages; catch up from durable store.
420                        let pin_key = partition_key(&topic, filter.as_deref());
421                        let after = delivery_pins
422                            .get(&pin_key)
423                            .map(|v| *v)
424                            .or(after_seq)
425                            .unwrap_or(0);
426                        match Self::load_replay_events(
427                            &pool,
428                            &crypto,
429                            &topic_name,
430                            filter.as_deref(),
431                            after,
432                        )
433                        .await
434                        {
435                            Ok(events) => {
436                                for evt in events {
437                                    if topic_filter_matches(&evt, &topic, filter.as_ref()) {
438                                        let pk = partition_key(
439                                            &evt.topic_name,
440                                            evt.topic_key.as_deref(),
441                                        );
442                                        delivery_pins.insert(pk, evt.seq);
443                                        yield Ok(evt);
444                                    }
445                                }
446                            }
447                            Err(e) => yield Err(e),
448                        }
449                    }
450                    Err(broadcast::error::RecvError::Closed) => break,
451                }
452            }
453        })
454    }
455
456    async fn get_event(&self, event_id: &str) -> Result<Option<Event>> {
457        if let Some(ev) = self.events.get(event_id) {
458            return Ok(Some(open_stored_event(&self.crypto, ev.clone())?));
459        }
460        self.fetch_event_by_id(event_id).await
461    }
462
463    async fn list_by_topic(
464        &self,
465        topic_name: &str,
466        topic_key: Option<&str>,
467        after_seq: Option<i64>,
468        limit: usize,
469    ) -> Result<Vec<Event>> {
470        self.fetch_list_by_topic(topic_name, topic_key, after_seq, limit)
471            .await
472    }
473
474    async fn list_recent(&self, limit: usize) -> Result<Vec<Event>> {
475        self.fetch_list_recent(limit).await
476    }
477
478    async fn load_checkpoint(
479        &self,
480        subscription_name: &str,
481        topic_name: &str,
482        topic_key: Option<&str>,
483    ) -> Result<Option<i64>> {
484        let key = checkpoint_key(subscription_name, topic_name, topic_key);
485        let row = sqlx::query("SELECT last_seq FROM checkpoints WHERE checkpoint_key = ?")
486            .bind(&key)
487            .fetch_optional(&self.pool)
488            .await
489            .map_err(map_sqlx)?;
490        Ok(row.map(|r| r.get::<i64, _>(0)))
491    }
492
493    async fn commit_checkpoint(
494        &self,
495        subscription_name: &str,
496        topic_name: &str,
497        topic_key: Option<&str>,
498        last_seq: i64,
499    ) -> Result<()> {
500        let key = checkpoint_key(subscription_name, topic_name, topic_key);
501        sqlx::query(
502            "INSERT INTO checkpoints (checkpoint_key, last_seq) VALUES (?, ?)
503             ON CONFLICT(checkpoint_key) DO UPDATE
504             SET last_seq = MAX(checkpoints.last_seq, excluded.last_seq)",
505        )
506        .bind(&key)
507        .bind(last_seq)
508        .execute(&self.pool)
509        .await
510        .map_err(map_sqlx)?;
511        Ok(())
512    }
513
514    async fn delivery_seq_pin(&self, topic_name: &str, topic_key: Option<&str>) -> Option<i64> {
515        let key = partition_key(topic_name, topic_key);
516        self.delivery_pins.get(&key).map(|v| *v)
517    }
518}
519
520#[cfg(test)]
521mod tests {
522    use super::*;
523
524    #[tokio::test]
525    async fn open_rejects_empty_path() {
526        assert!(matches!(
527            SqliteStoragePort::open(" ").await,
528            Err(PhotonError::Internal(_))
529        ));
530    }
531}
photon_backend_sqlite/port.rs

photon_backend_sqlite/
port.rs
