mf_persistence/
sqlite.rs

1//! 基于 SQLite 的 `EventStore` 实现，使用 RBatis 异步访问。
2//!
3//! 默认启用 WAL 模式配合 IMMEDIATE 事务，保证顺序写入与良好吞吐，
4//! 并在 `CommitMode::SyncDurable` 下触发 WAL checkpoint 以获得更强持久性。
5
6use std::path::PathBuf;
7use std::sync::Arc;
8
9use async_trait::async_trait;
10use rbatis::{executor::Executor, RBatis};
11use rbdc_sqlite::Driver;
12use rbs::Value;
13use serde::{Deserialize, Serialize};
14
15use crate::api::{CommitMode, EventStore, PersistedEvent, Snapshot};
16
17const INIT_SQL: &str = r#"
18    PRAGMA journal_mode = WAL;
19    PRAGMA synchronous = NORMAL;
20    PRAGMA cache_size = -64000;
21    PRAGMA temp_store = MEMORY;
22
23    CREATE TABLE IF NOT EXISTS events (
24      lsn INTEGER PRIMARY KEY AUTOINCREMENT,
25      tr_id INTEGER NOT NULL,
26      doc_id TEXT NOT NULL,
27      ts INTEGER NOT NULL,
28      actor TEXT,
29      idempotency_key TEXT NOT NULL UNIQUE,
30      meta TEXT NOT NULL,
31      payload BLOB NOT NULL,
32      checksum INTEGER NOT NULL
33    );
34    CREATE INDEX IF NOT EXISTS ix_events_doc_lsn ON events(doc_id, lsn);
35
36    CREATE TABLE IF NOT EXISTS snapshots (
37      doc_id TEXT NOT NULL,
38      upto_lsn INTEGER NOT NULL,
39      created_at INTEGER NOT NULL,
40      state_blob BLOB NOT NULL,
41      version INTEGER NOT NULL,
42      PRIMARY KEY (doc_id, upto_lsn)
43    );
44    CREATE INDEX IF NOT EXISTS ix_snapshots_doc_created
45        ON snapshots(doc_id, created_at DESC);
46"#;
47
48const INSERT_EVENT_SQL: &str = "\
49    INSERT INTO events \
50    (tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum) \
51    VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)";
52
53const UPSERT_SNAPSHOT_SQL: &str = "\
54    INSERT OR REPLACE INTO snapshots \
55    (doc_id, upto_lsn, created_at, state_blob, version) \
56    VALUES (?1, ?2, ?3, ?4, ?5)";
57
58/// `EventStore` 的 SQLite 具体实现。
59pub struct SqliteEventStore {
60    _db_path: PathBuf,
61    pool: Arc<RBatis>,
62    commit_mode: CommitMode,
63}
64
65impl SqliteEventStore {
66    /// 打开（或创建）数据库并初始化表结构与 PRAGMA。
67    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(db_path), fields(
68        crate_name = "persistence",
69        commit_mode = ?commit_mode
70    )))]
71    pub async fn open(
72        db_path: impl Into<PathBuf>,
73        commit_mode: CommitMode,
74    ) -> anyhow::Result<Arc<Self>> {
75        let db_path = db_path.into();
76        if let Some(parent) = db_path.parent() {
77            if !parent.as_os_str().is_empty() {
78                std::fs::create_dir_all(parent)?;
79            }
80        }
81
82        let rb = RBatis::new();
83        rb.link(Driver {}, &format!("sqlite://{}", db_path.display())).await?;
84        Self::init_schema(&rb).await?;
85
86        Ok(Arc::new(Self {
87            _db_path: db_path,
88            pool: Arc::new(rb),
89            commit_mode,
90        }))
91    }
92
93    async fn init_schema(rb: &RBatis) -> anyhow::Result<()> {
94        let conn = rb.acquire().await?;
95        conn.exec(INIT_SQL, vec![]).await?;
96        Ok(())
97    }
98
99    async fn insert_event<E>(
100        &self,
101        exec: &E,
102        ev: &PersistedEvent,
103    ) -> anyhow::Result<i64>
104    where
105        E: Executor + ?Sized,
106    {
107        let params = vec![
108            to_value(ev.tr_id.to_string()),
109            to_value(ev.doc_id.clone()),
110            to_value(ev.ts),
111            to_value(ev.actor.clone()),
112            to_value(ev.idempotency_key.clone()),
113            to_value(serde_json::to_string(&ev.meta)?),
114            to_value(ev.payload.clone()),
115            to_value(ev.checksum as i64),
116        ];
117        let exec_result = exec.exec(INSERT_EVENT_SQL, params).await?;
118        Ok(exec_result.last_insert_id.as_i64().unwrap_or_default())
119    }
120
121    async fn ensure_sync(&self) -> anyhow::Result<()> {
122        if matches!(self.commit_mode, CommitMode::SyncDurable) {
123            let conn = self.pool.acquire().await?;
124            conn.exec("PRAGMA wal_checkpoint(TRUNCATE)", vec![]).await.ok();
125        }
126        Ok(())
127    }
128}
129
130#[async_trait]
131impl EventStore for SqliteEventStore {
132    /// 以独立事务追加一条事件记录。
133    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, ev), fields(
134        crate_name = "persistence",
135        tr_id = %ev.tr_id,
136        doc_id = %ev.doc_id
137    )))]
138    async fn append(
139        &self,
140        ev: PersistedEvent,
141    ) -> anyhow::Result<i64> {
142        let tx = self.pool.acquire_begin().await?;
143        let lsn = self.insert_event(&tx, &ev).await?;
144        tx.commit().await?;
145        self.ensure_sync().await?;
146        Ok(lsn)
147    }
148
149    /// 在单事务内批量追加多条事件以提升吞吐。
150    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, evs), fields(
151        crate_name = "persistence",
152        batch_size = evs.len()
153    )))]
154    async fn append_batch(
155        &self,
156        evs: Vec<PersistedEvent>,
157    ) -> anyhow::Result<i64> {
158        if evs.is_empty() {
159            return Ok(0);
160        }
161        let tx = self.pool.acquire_begin().await?;
162        let mut last_lsn = 0;
163        for ev in &evs {
164            last_lsn = self.insert_event(&tx, ev).await?;
165        }
166        tx.commit().await?;
167        self.ensure_sync().await?;
168        Ok(last_lsn)
169    }
170
171    /// 读取指定文档在 `from_lsn` 之后的有序事件流。
172    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self), fields(
173        crate_name = "persistence",
174        doc_id = %doc_id,
175        from_lsn = from_lsn,
176        limit = limit
177    )))]
178    async fn load_since(
179        &self,
180        doc_id: &str,
181        from_lsn: i64,
182        limit: u32,
183    ) -> anyhow::Result<Vec<PersistedEvent>> {
184        let conn = self.pool.acquire().await?;
185        let rows: Vec<EventRow> = conn
186            .query_decode(
187                "SELECT lsn, tr_id, doc_id, ts, actor, idempotency_key, \
188                 meta, payload, checksum \
189                 FROM events \
190                 WHERE doc_id = ?1 AND lsn > ?2 \
191                 ORDER BY lsn ASC LIMIT ?3",
192                vec![
193                    to_value(doc_id),
194                    to_value(from_lsn),
195                    to_value(limit as i64),
196                ],
197            )
198            .await?;
199        Ok(rows
200            .into_iter()
201            .map(PersistedEvent::try_from)
202            .collect::<anyhow::Result<Vec<_>>>()?)
203    }
204
205    /// 返回该文档的最新快照（若存在）。
206    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self), fields(
207        crate_name = "persistence",
208        doc_id = %doc_id
209    )))]
210    async fn latest_snapshot(
211        &self,
212        doc_id: &str,
213    ) -> anyhow::Result<Option<Snapshot>> {
214        let conn = self.pool.acquire().await?;
215        let rows: Vec<SnapshotRow> = conn
216            .query_decode(
217                "SELECT doc_id, upto_lsn, created_at, state_blob, version \
218                 FROM snapshots \
219                 WHERE doc_id = ?1 \
220                 ORDER BY created_at DESC LIMIT 1",
221                vec![to_value(doc_id)],
222            )
223            .await?;
224        Ok(rows.into_iter().next().map(Into::into))
225    }
226
227    /// 原子写入/替换快照。
228    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, snap), fields(
229        crate_name = "persistence",
230        doc_id = %snap.doc_id,
231        upto_lsn = snap.upto_lsn
232    )))]
233    async fn write_snapshot(
234        &self,
235        snap: Snapshot,
236    ) -> anyhow::Result<()> {
237        let tx = self.pool.acquire_begin().await?;
238        tx.exec(
239            UPSERT_SNAPSHOT_SQL,
240            vec![
241                to_value(snap.doc_id),
242                to_value(snap.upto_lsn),
243                to_value(snap.created_at),
244                to_value(snap.state_blob),
245                to_value(snap.version),
246            ],
247        )
248        .await?;
249        tx.commit().await?;
250        Ok(())
251    }
252
253    /// 最佳努力的历史压缩：删除最新快照之前的事件。
254    async fn compact(
255        &self,
256        doc_id: &str,
257    ) -> anyhow::Result<()> {
258        let conn = self.pool.acquire().await?;
259        let upto_rows: Vec<UptoRow> = conn
260            .query_decode(
261                "SELECT upto_lsn FROM snapshots \
262                 WHERE doc_id = ?1 \
263                 ORDER BY upto_lsn DESC LIMIT 1",
264                vec![to_value(doc_id)],
265            )
266            .await?;
267        if let Some(row) = upto_rows.into_iter().next() {
268            conn.exec(
269                "DELETE FROM events WHERE doc_id = ?1 AND lsn <= ?2",
270                vec![to_value(doc_id), to_value(row.upto_lsn)],
271            )
272            .await?;
273        }
274        Ok(())
275    }
276}
277
278fn to_value<T: serde::Serialize>(value: T) -> Value {
279    rbs::value_def(value)
280}
281
282#[derive(Debug, Clone, Serialize, Deserialize)]
283struct EventRow {
284    lsn: i64,
285    tr_id: u64,
286    doc_id: String,
287    ts: i64,
288    actor: Option<String>,
289    idempotency_key: String,
290    meta: String,
291    payload: Vec<u8>,
292    checksum: i64,
293}
294
295impl TryFrom<EventRow> for PersistedEvent {
296    type Error = anyhow::Error;
297
298    fn try_from(row: EventRow) -> anyhow::Result<Self> {
299        Ok(Self {
300            lsn: row.lsn,
301            tr_id: row.tr_id,
302            doc_id: row.doc_id,
303            ts: row.ts,
304            actor: row.actor,
305            idempotency_key: row.idempotency_key,
306            meta: serde_json::from_str(&row.meta)?,
307            payload: row.payload,
308            checksum: row.checksum as u32,
309        })
310    }
311}
312
313#[derive(Debug, Clone, Serialize, Deserialize)]
314struct SnapshotRow {
315    doc_id: String,
316    upto_lsn: i64,
317    created_at: i64,
318    state_blob: Vec<u8>,
319    version: i32,
320}
321
322impl From<SnapshotRow> for Snapshot {
323    fn from(row: SnapshotRow) -> Self {
324        Self {
325            doc_id: row.doc_id,
326            upto_lsn: row.upto_lsn,
327            created_at: row.created_at,
328            state_blob: row.state_blob,
329            version: row.version,
330        }
331    }
332}
333
334#[derive(Debug, Clone, Serialize, Deserialize)]
335struct UptoRow {
336    upto_lsn: i64,
337}
mf_persistence/sqlite.rs

mf_persistence/
sqlite.rs
