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

photon_backend_sqlite/
port.rs
