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

photon_backend_sqlite/
port.rs
