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

photon_backend_sqlite/
port.rs
