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 parking_lot::Mutex;
12use rusqlite::{params, Connection, TransactionBehavior};
13
14use crate::api::{CommitMode, EventStore, PersistedEvent, Snapshot};
15
16/// `EventStore` 的 SQLite 具体实现。
17pub struct SqliteEventStore {
18    _db_path: PathBuf,
19    conn: Mutex<Connection>,
20    commit_mode: CommitMode,
21}
22
23impl SqliteEventStore {
24    /// 打开（或创建）数据库并初始化表结构与 PRAGMA。
25    pub fn open(
26        db_path: impl Into<PathBuf>,
27        commit_mode: CommitMode,
28    ) -> anyhow::Result<Arc<Self>> {
29        let db_path = db_path.into();
30        // 确保父目录存在
31        if let Some(parent) = db_path.parent() {
32            if !parent.as_os_str().is_empty() {
33                std::fs::create_dir_all(parent)?;
34            }
35        }
36        let conn = Connection::open(&db_path)?;
37        conn.pragma_update(None, "journal_mode", &"WAL")?;
38        conn.pragma_update(None, "synchronous", &"NORMAL")?;
39        conn.pragma_update(None, "busy_timeout", &5000i64)?;
40        conn.execute_batch(
41            r#"
42            CREATE TABLE IF NOT EXISTS events (
43              lsn INTEGER PRIMARY KEY AUTOINCREMENT,
44              tr_id INTEGER NOT NULL,
45              doc_id TEXT NOT NULL,
46              ts INTEGER NOT NULL,
47              actor TEXT,
48              idempotency_key TEXT NOT NULL UNIQUE,
49              meta TEXT NOT NULL,
50              payload BLOB NOT NULL,
51              checksum INTEGER NOT NULL
52            );
53            CREATE INDEX IF NOT EXISTS ix_events_doc_lsn ON events(doc_id, lsn);
54
55            CREATE TABLE IF NOT EXISTS snapshots (
56              doc_id TEXT NOT NULL,
57              upto_lsn INTEGER NOT NULL,
58              created_at INTEGER NOT NULL,
59              state_blob BLOB NOT NULL,
60              version INTEGER NOT NULL,
61              PRIMARY KEY (doc_id, upto_lsn)
62            );
63            CREATE INDEX IF NOT EXISTS ix_snapshots_doc_created ON snapshots(doc_id, created_at DESC);
64            "#,
65        )?;
66
67        Ok(Arc::new(Self {
68            _db_path: db_path,
69            conn: Mutex::new(conn),
70            commit_mode,
71        }))
72    }
73}
74
75#[async_trait]
76impl EventStore for SqliteEventStore {
77    /// 以独立事务追加一条事件记录。
78    async fn append(
79        &self,
80        ev: PersistedEvent,
81    ) -> anyhow::Result<i64> {
82        let mut conn = self.conn.lock();
83        let tx =
84            conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
85        tx.execute(
86            "INSERT INTO events (tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
87            params![
88                ev.tr_id as i64,
89                ev.doc_id,
90                ev.ts as i64,
91                ev.actor,
92                ev.idempotency_key,
93                ev.meta.to_string(),
94                ev.payload,
95                ev.checksum as i64
96            ],
97        )?;
98        tx.commit()?;
99        let lsn = conn.last_insert_rowid();
100        match self.commit_mode {
101            CommitMode::SyncDurable => {
102                conn.pragma_update(None, "wal_checkpoint", &"TRUNCATE").ok();
103            },
104            _ => {},
105        }
106        Ok(lsn)
107    }
108
109    /// 在单事务内批量追加多条事件以提升吞吐。
110    async fn append_batch(
111        &self,
112        evs: Vec<PersistedEvent>,
113    ) -> anyhow::Result<i64> {
114        if evs.is_empty() {
115            return Ok(0);
116        }
117        let mut conn = self.conn.lock();
118        let tx =
119            conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
120        {
121            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)")?;
122            for ev in evs.into_iter() {
123                stmt.execute(params![
124                    ev.tr_id as i64,
125                    ev.doc_id,
126                    ev.ts as i64,
127                    ev.actor,
128                    ev.idempotency_key,
129                    ev.meta.to_string(),
130                    ev.payload,
131                    ev.checksum as i64
132                ])?;
133            }
134        }
135        tx.commit()?;
136        let lsn = conn.last_insert_rowid();
137        Ok(lsn)
138    }
139
140    /// 读取指定文档在 `from_lsn` 之后的有序事件流。
141    async fn load_since(
142        &self,
143        doc_id: &str,
144        from_lsn: i64,
145        limit: u32,
146    ) -> anyhow::Result<Vec<PersistedEvent>> {
147        let conn = self.conn.lock();
148        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")?;
149        let rows =
150            stmt.query_map(params![doc_id, from_lsn, limit as i64], |row| {
151                let meta_str: String = row.get(6)?;
152                let meta: serde_json::Value = serde_json::from_str(&meta_str)
153                    .unwrap_or(serde_json::json!({}));
154                Ok(PersistedEvent {
155                    lsn: row.get(0)?,
156                    tr_id: row.get::<_, i64>(1)? as u64,
157                    doc_id: row.get(2)?,
158                    ts: row.get(3)?,
159                    actor: row.get(4).ok(),
160                    idempotency_key: row.get(5)?,
161                    meta,
162                    payload: row.get(7)?,
163                    checksum: row.get::<_, i64>(8)? as u32,
164                })
165            })?;
166        Ok(rows.filter_map(|r| r.ok()).collect())
167    }
168
169    /// 返回该文档的最新快照（若存在）。
170    async fn latest_snapshot(
171        &self,
172        doc_id: &str,
173    ) -> anyhow::Result<Option<Snapshot>> {
174        let conn = self.conn.lock();
175        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")?;
176        let mut rows = stmt.query(params![doc_id])?;
177        if let Some(row) = rows.next()? {
178            Ok(Some(Snapshot {
179                doc_id: row.get(0)?,
180                upto_lsn: row.get(1)?,
181                created_at: row.get(2)?,
182                state_blob: row.get(3)?,
183                version: row.get(4)?,
184            }))
185        } else {
186            Ok(None)
187        }
188    }
189
190    /// 原子写入/替换快照。
191    async fn write_snapshot(
192        &self,
193        snap: Snapshot,
194    ) -> anyhow::Result<()> {
195        let mut conn = self.conn.lock();
196        let tx =
197            conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
198        tx.execute(
199            "INSERT OR REPLACE INTO snapshots (doc_id, upto_lsn, created_at, state_blob, version) VALUES (?1, ?2, ?3, ?4, ?5)",
200            params![snap.doc_id, snap.upto_lsn, snap.created_at, snap.state_blob, snap.version],
201        )?;
202        tx.commit()?;
203        Ok(())
204    }
205
206    /// 最佳努力的历史压缩：删除最新快照之前的事件。
207    async fn compact(
208        &self,
209        doc_id: &str,
210    ) -> anyhow::Result<()> {
211        let conn = self.conn.lock();
212        // 删除最老的事件到最近快照
213        let upto: Option<i64> = conn.query_row(
214            "SELECT upto_lsn FROM snapshots WHERE doc_id = ?1 ORDER BY upto_lsn DESC LIMIT 1",
215            params![doc_id],
216            |r| r.get(0),
217        ).ok();
218        if let Some(upto_lsn) = upto {
219            conn.execute(
220                "DELETE FROM events WHERE doc_id = ?1 AND lsn <= ?2",
221                params![doc_id, upto_lsn],
222            )?;
223        }
224        Ok(())
225    }
226}
mf_persistence/sqlite.rs

mf_persistence/
sqlite.rs
