mf_persistence/
sqlite.rs

1//! 基于 SQLite 的 `EventStore` 实现。
2//!
3//! 默认使用 WAL 日志模式与 IMMEDIATE 事务以获得低延迟写入。
4//! 采用“单连接 + 互斥”顺序写，确保追加顺序并避免写锁竞争；
5//! 读取按需准备语句，受益于 SQLite 页缓存。
6
7use std::path::PathBuf;
8use std::sync::Arc;
9
10use async_trait::async_trait;
11use rusqlite::{params, TransactionBehavior};
12use r2d2::Pool;
13use r2d2_sqlite::SqliteConnectionManager;
14use crate::api::{CommitMode, EventStore, PersistedEvent, Snapshot};
15
16/// `EventStore` 的 SQLite 具体实现。
17pub struct SqliteEventStore {
18    _db_path: PathBuf,
19    pool: Pool<SqliteConnectionManager>,
20    commit_mode: CommitMode,
21}
22
23impl SqliteEventStore {
24    /// 打开（或创建）数据库并初始化表结构与 PRAGMA。
25    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(db_path), fields(
26        crate_name = "persistence",
27        commit_mode = ?commit_mode
28    )))]
29    pub fn open(
30        db_path: impl Into<PathBuf>,
31        commit_mode: CommitMode,
32    ) -> anyhow::Result<Arc<Self>> {
33        let db_path = db_path.into();
34        // 确保父目录存在
35        if let Some(parent) = db_path.parent() {
36            if !parent.as_os_str().is_empty() {
37                std::fs::create_dir_all(parent)?;
38            }
39        }
40        let manager = SqliteConnectionManager::file(&db_path);
41        let pool = Pool::new(manager)?;
42        pool.get()?.execute_batch(
43            r#"
44            CREATE TABLE IF NOT EXISTS events (
45              lsn INTEGER PRIMARY KEY AUTOINCREMENT,
46              tr_id TEXT NOT NULL,
47              doc_id TEXT NOT NULL,
48              ts INTEGER NOT NULL,
49              actor TEXT,
50              idempotency_key TEXT NOT NULL UNIQUE,
51              meta TEXT NOT NULL,
52              payload BLOB NOT NULL,
53              checksum INTEGER NOT NULL
54            );
55            CREATE INDEX IF NOT EXISTS ix_events_doc_lsn ON events(doc_id, lsn);
56
57            CREATE TABLE IF NOT EXISTS snapshots (
58              doc_id TEXT NOT NULL,
59              upto_lsn INTEGER NOT NULL,
60              created_at INTEGER NOT NULL,
61              state_blob BLOB NOT NULL,
62              version INTEGER NOT NULL,
63              PRIMARY KEY (doc_id, upto_lsn)
64            );
65            CREATE INDEX IF NOT EXISTS ix_snapshots_doc_created ON snapshots(doc_id, created_at DESC);
66            "#,
67        )?;
68
69        Ok(Arc::new(Self { _db_path: db_path, pool, commit_mode }))
70    }
71}
72
73#[async_trait]
74impl EventStore for SqliteEventStore {
75    /// 以独立事务追加一条事件记录。
76    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, ev), fields(
77        crate_name = "persistence",
78        tr_id = %ev.tr_id,
79        doc_id = %ev.doc_id
80    )))]
81    async fn append(
82        &self,
83        ev: PersistedEvent,
84    ) -> anyhow::Result<i64> {
85        let mut conn = self.pool.get()?;
86        let tx =
87            conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
88        tx.execute(
89            "INSERT INTO events (tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
90            params![
91                ev.tr_id.to_string(), // ✅ UUID 转字符串存储
92                ev.doc_id,
93                { ev.ts },
94                ev.actor,
95                ev.idempotency_key,
96                ev.meta.to_string(),
97                ev.payload,
98                ev.checksum as i64
99            ],
100        )?;
101        tx.commit()?;
102        let lsn = conn.last_insert_rowid();
103        if let CommitMode::SyncDurable = self.commit_mode {
104            conn.pragma_update(None, "wal_checkpoint", "TRUNCATE").ok();
105        }
106        Ok(lsn)
107    }
108
109    /// 在单事务内批量追加多条事件以提升吞吐。
110    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, evs), fields(
111        crate_name = "persistence",
112        batch_size = evs.len()
113    )))]
114    async fn append_batch(
115        &self,
116        evs: Vec<PersistedEvent>,
117    ) -> anyhow::Result<i64> {
118        if evs.is_empty() {
119            return Ok(0);
120        }
121        let mut conn = self.pool.get()?;
122        let tx =
123            conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
124        {
125            let mut stmt = tx.prepare("INSERT INTO events (tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)")?;
126            for ev in evs.into_iter() {
127                stmt.execute(params![
128                    ev.tr_id.to_string(), // ✅ UUID 转字符串存储
129                    ev.doc_id,
130                    { ev.ts },
131                    ev.actor,
132                    ev.idempotency_key,
133                    ev.meta.to_string(),
134                    ev.payload,
135                    ev.checksum as i64
136                ])?;
137            }
138        }
139        tx.commit()?;
140        let lsn = conn.last_insert_rowid();
141        Ok(lsn)
142    }
143
144    /// 读取指定文档在 `from_lsn` 之后的有序事件流。
145    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self), fields(
146        crate_name = "persistence",
147        doc_id = %doc_id,
148        from_lsn = from_lsn,
149        limit = limit
150    )))]
151    async fn load_since(
152        &self,
153        doc_id: &str,
154        from_lsn: i64,
155        limit: u32,
156    ) -> anyhow::Result<Vec<PersistedEvent>> {
157        let conn = self.pool.get()?;
158        let mut stmt = conn.prepare("SELECT lsn, tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum FROM events WHERE doc_id = ?1 AND lsn > ?2 ORDER BY lsn ASC LIMIT ?3")?;
159        let rows =
160            stmt.query_map(params![doc_id, from_lsn, limit as i64], |row| {
161                let meta_str: String = row.get(6)?;
162                let meta: serde_json::Value = serde_json::from_str(&meta_str)
163                    .unwrap_or(serde_json::json!({}));
164                let tr_id_str: String = row.get(1)?;
165                let tr_id = uuid::Uuid::parse_str(&tr_id_str).map_err(|e| {
166                    rusqlite::Error::FromSqlConversionFailure(
167                        1,
168                        rusqlite::types::Type::Text,
169                        Box::new(e),
170                    )
171                })?;
172                Ok(PersistedEvent {
173                    lsn: row.get(0)?,
174                    tr_id, // ✅ 从字符串解析 UUID
175                    doc_id: row.get(2)?,
176                    ts: row.get(3)?,
177                    actor: row.get(4).ok(),
178                    idempotency_key: row.get(5)?,
179                    meta,
180                    payload: row.get(7)?,
181                    checksum: row.get::<_, i64>(8)? as u32,
182                })
183            })?;
184        Ok(rows.filter_map(|r| r.ok()).collect())
185    }
186
187    /// 返回该文档的最新快照（若存在）。
188    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self), fields(
189        crate_name = "persistence",
190        doc_id = %doc_id
191    )))]
192    async fn latest_snapshot(
193        &self,
194        doc_id: &str,
195    ) -> anyhow::Result<Option<Snapshot>> {
196        let conn = self.pool.get()?;
197        let mut stmt = conn.prepare("SELECT doc_id, upto_lsn, created_at, state_blob, version FROM snapshots WHERE doc_id = ?1 ORDER BY created_at DESC LIMIT 1")?;
198        let mut rows = stmt.query(params![doc_id])?;
199        if let Some(row) = rows.next()? {
200            Ok(Some(Snapshot {
201                doc_id: row.get(0)?,
202                upto_lsn: row.get(1)?,
203                created_at: row.get(2)?,
204                state_blob: row.get(3)?,
205                version: row.get(4)?,
206            }))
207        } else {
208            Ok(None)
209        }
210    }
211
212    /// 原子写入/替换快照。
213    #[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, snap), fields(
214        crate_name = "persistence",
215        doc_id = %snap.doc_id,
216        upto_lsn = snap.upto_lsn
217    )))]
218    async fn write_snapshot(
219        &self,
220        snap: Snapshot,
221    ) -> anyhow::Result<()> {
222        let mut conn = self.pool.get()?;
223        let tx =
224            conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
225        tx.execute(
226            "INSERT OR REPLACE INTO snapshots (doc_id, upto_lsn, created_at, state_blob, version) VALUES (?1, ?2, ?3, ?4, ?5)",
227            params![snap.doc_id, snap.upto_lsn, snap.created_at, snap.state_blob, snap.version],
228        )?;
229        tx.commit()?;
230        Ok(())
231    }
232
233    /// 最佳努力的历史压缩：删除最新快照之前的事件。
234    async fn compact(
235        &self,
236        doc_id: &str,
237    ) -> anyhow::Result<()> {
238        let conn = self.pool.get()?;
239        // 删除最老的事件到最近快照
240        let upto: Option<i64> = conn.query_row(
241            "SELECT upto_lsn FROM snapshots WHERE doc_id = ?1 ORDER BY upto_lsn DESC LIMIT 1",
242            params![doc_id],
243            |r| r.get(0),
244        ).ok();
245        if let Some(upto_lsn) = upto {
246            conn.execute(
247                "DELETE FROM events WHERE doc_id = ?1 AND lsn <= ?2",
248                params![doc_id, upto_lsn],
249            )?;
250        }
251        Ok(())
252    }
253}
mf_persistence/sqlite.rs

mf_persistence/
sqlite.rs
