【AI开发之Rust】第 18 课：SQLite 持久化与本地缓存 —— 给 store 填上真实现

第 17 课的 HistoryStore 还是内存假实现——App 一重启历史全丢。本课把它换成 SQLite 本地数据库，回答三件事：

① rusqlite：怎么连、怎么建表、怎么读写（参数化查询防注入）
② 表结构 & 迁移：schema 版本升级不丢数据
③ 把 sqlite 实现塞回 core：服务层测试一行不改（17 课接口红利的兑现）

同时处理 SQLite 在真实 App 里最常见的"最后 20%"：忙锁（database is locked）、连接共享、并发读写。这些坑恰恰是 UniFFI（19-21 课）把 core 暴露给多个 UI 线程后必然撞上的。

💡 为什么本地存储选 SQLite 而不是"写个 JSON 文件"？历史记录会持续增长，需要随机查询单会话/删除/排序——JSON 全量读改写在大数据量下又慢又容易坏。SQLite 单文件、零服务、SQL 能力完整，是移动/桌面端本地存储的事实标准。会话与消息的结构化适合表；而"聊天原样快照"之类大对象适合存成 JSON 列/独立文件——两种都会在本课出现。

18.2 rusqlite 起步：打开连接 + 最小读写

<span># core/Cargo.toml 追加</span>
<span>[dependencies]</span>
<span>rusqlite</span> = { version = <span>"0.32"</span>, features = [<span>"bundled"</span>] }   <span># bundled：免系统 SQLite，跨端更省心</span>
chrono 不再需要——时间用 i64 毫秒（第 17 课 now_ms）

<span>// store/sqlite.rs —— 新的模块</span>
<span>use</span> rusqlite::Connection;
<span>use</span> std::path::Path;

<span>pub</span> <span>struct</span> <span>SqliteStore</span> {
    conn: Connection,
}

<span>impl</span> <span>SqliteStore</span> {
    <span>/// 打开（或创建）数据库文件</span>
    <span>pub</span> <span>fn</span> <span>open</span>(path: <span>impl</span> <span>AsRef</span><Path>) <span>-></span> <span>Result</span><<span>Self</span>, rusqlite::Error> {
        <span>let</span> <span></span><span>conn</span> = Connection::<span>open</span>(path)?;
        conn.<span>pragma_update</span>(<span>None</span>, <span>"journal_mode"</span>, <span>"WAL"</span>)?;      <span>// 见 18.5</span>
        conn.<span>pragma_update</span>(<span>None</span>, <span>"foreign_keys"</span>, <span>"ON"</span>)?;
        <span>let</span> <span></span><span>store</span> = SqliteStore { conn };
        store.<span>migrate</span>()?;                                       <span>// 见 18.3</span>
        <span>Ok</span>(store)
    }
}

最小读写样例（临时库，:memory: 可用于测试）：

<span>use</span> rusqlite::{params, Connection};

<span>fn</span> <span>main</span>() <span>-></span> rusqlite::<span>Result</span><()> {
    <span>let</span> <span></span><span>conn</span> = Connection::<span>open_in_memory</span>()?;

    conn.<span>execute</span>(<span>"CREATE TABLE kv (k TEXT PRIMARY KEY, v TEXT NOT NULL)"</span>, [])?;

    <span>// 参数化插入（占位符 ?1，值作为参数传入 → 自动防注入）</span>
    conn.<span>execute</span>(<span>"INSERT INTO kv (k, v) VALUES (?1, ?2)"</span>, params![<span>"greeting"</span>, <span>"你好"</span>])?;

    <span>// 参数化查询</span>
    <span>let</span> <span></span><span>v</span>: <span>String</span> = conn.<span>query_row</span>(
        <span>"SELECT v FROM kv WHERE k = ?1"</span>,
        [<span>"greeting"</span>],
        |row| row.<span>get</span>(<span>0</span>),
    )?;
    <span>println!</span>(<span>"{v}"</span>);

    <span>// 多行读取（迭代 Statement）</span>
    <span>let</span> <span>mut </span><span>stmt</span> = conn.<span>prepare</span>(<span>"SELECT k, v FROM kv"</span>)?;
    <span>let</span> <span></span><span>rows</span> = stmt.<span>query_map</span>([], |row| {
        <span>Ok</span>((row.get::<_, <span>String</span>>(<span>0</span>)?, row.get::<_, <span>String</span>>(<span>1</span>)?))
    })?;
    <span>for</span> <span>pair</span> <span>in</span> rows {
        <span>println!</span>(<span>"{:?}"</span>, pair?);
    }
    <span>Ok</span>(())
}

⚠️ 永远用参数化，绝不要字符串拼接 SQL：

<span>// ❌ 注入风险 + 引号转义地狱</span>
<span>let</span> <span></span><span>sql</span> = <span>format!</span>(<span>"INSERT INTO kv VALUES ('{}')"</span>, user_input);
<span>// ✅ 参数绑定：rusqlite 负责转义，SQL 与数据分离</span>
conn.<span>execute</span>(<span>"INSERT INTO kv VALUES (?1)"</span>, params![user_input])?;

18.3 表结构与迁移：schema 版本化

18.3.1 建表 DDL

<span>-- sessions：会话表</span>
<span>CREATE</span> <span>TABLE</span> IF <span>NOT</span> <span>EXISTS</span> sessions (
    id           TEXT <span>PRIMARY</span> KEY,     <span>-- "s-<ms>-<rand>"</span>
    title        TEXT <span>NOT</span> <span>NULL</span>,
    created_at_ms <span>INTEGER</span> <span>NOT</span> <span>NULL</span>
);

<span>-- messages：消息表</span>
<span>CREATE</span> <span>TABLE</span> IF <span>NOT</span> <span>EXISTS</span> messages (
    id           TEXT <span>PRIMARY</span> KEY,
    session_id   TEXT <span>NOT</span> <span>NULL</span> <span>REFERENCES</span> sessions(id) <span>ON</span> <span>DELETE</span> CASCADE,
    role         TEXT <span>NOT</span> <span>NULL</span>,        <span>-- 'user' / 'assistant' / 'system'</span>
    content      TEXT <span>NOT</span> <span>NULL</span>,
    created_at_ms <span>INTEGER</span> <span>NOT</span> <span>NULL</span>
);

<span>CREATE</span> INDEX IF <span>NOT</span> <span>EXISTS</span> idx_messages_session
    <span>ON</span> messages (session_id, created_at_ms);

• ON DELETE CASCADE：删会话自动连带删消息（17 课 delete_session 就两个动作，其实库层一步到位）；
• 索引建在 (session_id, created_at_ms) 上：查单会话历史只走索引，随数据增长不退化；
• role 用 TEXT 存，读出来再转 Role 枚举（写入反过来）——不存 SQLite 不认识的 Rust enum。

18.3.2 用 PRAGMA user_version 做轻量迁移

<span>impl</span> <span>SqliteStore</span> {
    <span>/// 简单迁移：记录 schema 版本号，逐版升级</span>
    <span>fn</span> <span>migrate</span>(&<span>self</span>) <span>-></span> rusqlite::<span>Result</span><()> {
        <span>let</span> <span></span><span>cur</span>: <span>i64</span> = <span>self</span>.conn.<span>query_row</span>(<span>"PRAGMA user_version"</span>, [], |r| r.<span>get</span>(<span>0</span>))?;

        <span>if</span> cur < <span>1</span> {
            <span>self</span>.conn.<span>execute_batch</span>(
                <span>"BEGIN;
                 CREATE TABLE IF NOT EXISTS sessions (...同 18.3.1...);
                 CREATE TABLE IF NOT EXISTS messages (...);
                 CREATE INDEX ...;
                 PRAGMA user_version = 1;
                 COMMIT;"</span>,
            )?;
        }
        <span>// 将来加字段/加表：if cur < 2 { ... PRAGMA user_version = 2; }</span>
        <span>Ok</span>(())
    }
}

💡 真实项目可用 rusqlite_migration 这类 crate 管理有序迁移脚本；但"用户版本号 + 逐段 execute_batch"的原生写法已经足够支撑本课程规模，且零额外依赖。关键是养成习惯：任何 schema 变更都必须是"新增一段 if 版本号"的迁移，而不是改旧 SQL——否则老用户升级必炸。

18.4 实现 HistoryStore for SqliteStore

<span>use</span> crate::models::{Message, Role, Session};
<span>use</span> crate::store::{HistoryStore, StoreError};
<span>use</span> rusqlite::params;

<span>impl</span> <span>HistoryStore</span> <span>for</span> <span>SqliteStore</span> {
    <span>fn</span> <span>create_session</span>(&<span>self</span>, session: &Session) <span>-></span> <span>Result</span><(), StoreError> {
        <span>self</span>.conn
            .<span>execute</span>(
                <span>"INSERT INTO sessions (id, title, created_at_ms) VALUES (?1, ?2, ?3)"</span>,
                params![session.id, session.title, session.created_at_ms],
            )
            .<span>map_err</span>(map_err)?;
        <span>Ok</span>(())
    }

    <span>fn</span> <span>list_sessions</span>(&<span>self</span>, limit: <span>u32</span>) <span>-></span> <span>Result</span><<span>Vec</span><Session>, StoreError> {
        <span>let</span> <span>mut </span><span>stmt</span> = <span>self</span>
            .conn
            .<span>prepare</span>(<span>"SELECT id, title, created_at_ms FROM sessions ORDER BY created_at_ms DESC LIMIT ?1"</span>)
            .<span>map_err</span>(map_err)?;
        <span>let</span> <span></span><span>rows</span> = stmt
            .<span>query_map</span>([limit], |row| {
                <span>Ok</span>(Session {
                    id: row.<span>get</span>(<span>0</span>)?,
                    title: row.<span>get</span>(<span>1</span>)?,
                    created_at_ms: row.<span>get</span>(<span>2</span>)?,
                })
            })
            .<span>map_err</span>(map_err)?;
        rows.collect::<<span>Result</span><<span>Vec</span><_>, _>>().<span>map_err</span>(map_err)
    }

    <span>fn</span> <span>delete_session</span>(&<span>self</span>, session_id: &<span>str</span>) <span>-></span> <span>Result</span><(), StoreError> {
        <span>self</span>.conn
            .<span>execute</span>(<span>"DELETE FROM sessions WHERE id = ?1"</span>, params![session_id])
            .<span>map_err</span>(map_err)?;          <span>// messages 靠 CASCADE 一起删</span>
        <span>Ok</span>(())
    }

    <span>fn</span> <span>append_message</span>(&<span>self</span>, msg: &Message) <span>-></span> <span>Result</span><(), StoreError> {
        <span>let</span> <span></span><span>role</span> = <span>role_to_db</span>(&msg.role);   <span>// Role → &str</span>
        <span>self</span>.conn
            .<span>execute</span>(
                <span>"INSERT INTO messages (id, session_id, role, content, created_at_ms)
                 VALUES (?1, ?2, ?3, ?4, ?5)"</span>,
                params![msg.id, msg.session_id, role, msg.content, msg.created_at_ms],
            )
            .<span>map_err</span>(map_err)?;
        <span>Ok</span>(())
    }

    <span>fn</span> <span>list_messages</span>(&<span>self</span>, session_id: &<span>str</span>) <span>-></span> <span>Result</span><<span>Vec</span><Message>, StoreError> {
        <span>let</span> <span>mut </span><span>stmt</span> = <span>self</span>
            .conn
            .<span>prepare</span>(
                <span>"SELECT id, session_id, role, content, created_at_ms
                 FROM messages WHERE session_id = ?1 ORDER BY created_at_ms ASC"</span>,
            )
            .<span>map_err</span>(map_err)?;
        <span>let</span> <span></span><span>rows</span> = stmt
            .<span>query_map</span>([session_id], |row| {
                <span>let</span> <span></span><span>role</span>: <span>String</span> = row.<span>get</span>(<span>2</span>)?;
                <span>Ok</span>(Message {
                    id: row.<span>get</span>(<span>0</span>)?,
                    session_id: row.<span>get</span>(<span>1</span>)?,
                    role: <span>role_from_db</span>(&role),
                    content: row.<span>get</span>(<span>3</span>)?,
                    created_at_ms: row.<span>get</span>(<span>4</span>)?,
                })
            })
            .<span>map_err</span>(map_err)?;
        rows.collect::<<span>Result</span><<span>Vec</span><_>, _>>().<span>map_err</span>(map_err)
    }

    <span>fn</span> <span>clear_messages</span>(&<span>self</span>, session_id: &<span>str</span>) <span>-></span> <span>Result</span><(), StoreError> {
        <span>self</span>.conn
            .<span>execute</span>(<span>"DELETE FROM messages WHERE session_id = ?1"</span>, params![session_id])
            .<span>map_err</span>(map_err)?;
        <span>Ok</span>(())
    }
}

<span>/// 把 rusqlite 错误统一转成 StoreError（第 17 课的契约类型）</span>
<span>fn</span> <span>map_err</span>(e: rusqlite::Error) <span>-></span> StoreError {
    <span>match</span> e {
        rusqlite::Error::QueryReturnedNoRows => StoreError::<span>NotFound</span>(<span>"记录不存在"</span>.<span>into</span>()),
        other => StoreError::<span>Io</span>(other.<span>to_string</span>()),
    }
}

<span>fn</span> <span>role_to_db</span>(role: &Role) <span>-></span> &<span>'static</span> <span>str</span> {
    <span>match</span> role { Role::User => <span>"user"</span>, Role::Assistant => <span>"assistant"</span>, Role::System => <span>"system"</span> }
}

<span>fn</span> <span>role_from_db</span>(s: &<span>str</span>) <span>-></span> Role {
    <span>match</span> s { <span>"assistant"</span> => Role::Assistant, <span>"system"</span> => Role::System, _ => Role::User }
}

💡 读 DB 行 → Rust struct 的样板循环（query_map + row.get）在代码里很常见。量大后可考虑 serde 的 rusqlite 适配（derive 直接映射行），但先学会手写版，错误类型和列顺序都由你掌控。

18.5 并发与"database is locked"：真实 App 必修课

18.5.1 问题从哪来

第 19 课起，UI 的多个线程/任务可能同时调 append_message（用户连发多条）、list_sessions（刷新列表）、delete_session。SQLite 默认一次只允许一个写者，写锁没释放时另一写会报 database is locked。

rusqlite 的 Connection 默认不是 Sync（含内部缓存），直接放进 Box<dyn HistoryStore> 并跨线程调用会编译不过。三种处理姿势：

本课采用 B：把 Connection 包在 Mutex 里，天然满足 Sync 且把写冲突变成"排队"（加上 18.2 已开的 WAL 模式，读不阻塞写）：

<span>use</span> std::sync::Mutex;

<span>pub</span> <span>struct</span> <span>SqliteStore</span> {
    conn: Mutex<Connection>,          <span>// 内部串行化；对外仍是 Sync</span>
}

<span>impl</span> <span>SqliteStore</span> {
    <span>pub</span> <span>fn</span> <span>open</span>(path: <span>impl</span> <span>AsRef</span><std::path::Path>) <span>-></span> <span>Result</span><<span>Self</span>, rusqlite::Error> {
        <span>let</span> <span></span><span>conn</span> = Connection::<span>open</span>(path)?;
        conn.<span>pragma_update</span>(<span>None</span>, <span>"journal_mode"</span>, <span>"WAL"</span>)?;
        conn.<span>pragma_update</span>(<span>None</span>, <span>"foreign_keys"</span>, <span>"ON"</span>)?;
        <span>let</span> <span></span><span>store</span> = SqliteStore { conn: Mutex::<span>new</span>(conn) };
        store.<span>migrate</span>()?;
        <span>Ok</span>(store)
    }

    <span>fn</span> <span>lock</span>(&<span>self</span>) <span>-></span> <span>Result</span><std::sync::MutexGuard<<span>'_</span>, Connection>, StoreError> {
        <span>self</span>.conn.<span>lock</span>().<span>map_err</span>(|_| StoreError::<span>Io</span>(<span>"存储锁中毒"</span>.<span>into</span>()))
    }
}

然后每个方法开头 let conn = self.lock()?;，后续全部走 conn：

<span>impl</span> <span>HistoryStore</span> <span>for</span> <span>SqliteStore</span> {
    <span>fn</span> <span>create_session</span>(&<span>self</span>, session: &Session) <span>-></span> <span>Result</span><(), StoreError> {
        <span>let</span> <span></span><span>conn</span> = <span>self</span>.<span>lock</span>()?;
        conn.<span>execute</span>(
            <span>"INSERT INTO sessions (id, title, created_at_ms) VALUES (?1, ?2, ?3)"</span>,
            params![session.id, session.title, session.created_at_ms],
        )
        .<span>map_err</span>(map_err)?;
        <span>Ok</span>(())
    }
    <span>// ……其余方法同样先 lock()</span>
}

⚠️ 同步方法里用 std::sync::Mutex 没问题（不跨 .await 持有）。永远不要在持锁期间调用 UI/网络——锁粒度 = 一条 SQL。SQLite 单次操作毫秒级，串行化完全可接受。

18.5.2 WAL 模式是什么

<span>journal_mode</span> = WAL（Write-Ahead Logging）：
  - 写操作先追加到 .wal 文件，再择机合入主库
  - 效果：读写不互相阻塞（一个写者 + 多个读者可同时进行）
  - App 常见标配；配合 busy_timeout 可进一步缓解极端冲突

<span>// 极端冲突兜底：等锁最长 5 秒而不是立刻报错</span>
conn.<span>busy_timeout</span>(std::time::Duration::<span>from_secs</span>(<span>5</span>))?;

18.6 把真 store 接回 service：17 课测试一行不改

<span>// core 集成测试：sqlite_smoke.rs</span>
<span>use</span> my_ai_core::llm::chat::ChatLlm;
<span>use</span> my_ai_core::models::Message;
<span>use</span> my_ai_core::service::AssistantService;
<span>use</span> my_ai_core::store::{HistoryStore, SqliteStore};

<span>struct</span> <span>FakeLlm</span>;   <span>// 同 17.6</span>

<span>#[tokio::test]</span>
<span>async</span> <span>fn</span> <span>sqlite_roundtrip_via_service</span>() {
    <span>// 临时文件库，测完自动删</span>
    <span>let</span> <span></span><span>dir</span> = std::env::<span>temp_dir</span>();
    <span>let</span> <span></span><span>path</span> = dir.<span>join</span>(<span>format!</span>(<span>"ai_test_{}.db"</span>, std::process::<span>id</span>()));

    <span>let</span> <span></span><span>store</span> = SqliteStore::<span>open</span>(&path).<span>unwrap</span>();
    <span>let</span> <span></span><span>service</span> = AssistantService::<span>new</span>(<span>Box</span>::<span>new</span>(FakeLlm), <span>Box</span>::<span>new</span>(store), <span>"系统提示"</span>.<span>into</span>());

    <span>let</span> <span></span><span>session</span> = service.<span>new_session</span>(<span>"持久化测试"</span>).<span>await</span>.<span>unwrap</span>();
    <span>let</span> <span>mut </span><span>got</span> = <span>String</span>::<span>new</span>();
    service
        .<span>ask_stream</span>(&session.id, <span>"存下来了吗"</span>, <span>Box</span>::<span>new</span>(|d| got.<span>push_str</span>(d)))
        .<span>await</span>
        .<span>unwrap</span>();
    <span>assert!</span>(got.<span>contains</span>(<span>"你好，世界"</span>));

    <span>// 重新打开同一个文件（模拟 App 重启）：数据还在</span>
    <span>let</span> <span></span><span>store2</span> = SqliteStore::<span>open</span>(&path).<span>unwrap</span>();
    <span>let</span> <span></span><span>sessions</span> = store2.<span>list_sessions</span>(<span>10</span>).<span>unwrap</span>();
    <span>assert_eq!</span>(sessions.<span>len</span>(), <span>1</span>);
    <span>let</span> <span></span><span>msgs</span> = store2.<span>list_messages</span>(&session.id).<span>unwrap</span>();
    <span>assert_eq!</span>(msgs.<span>len</span>(), <span>2</span>);

    std::fs::<span>remove_file</span>(&path).<span>ok</span>();
}

验收瞬间：同一段 service 测试，把 MemStore::default() 换成 SqliteStore::open(...) 就跑通——17 课设计的接口抽象兑现了。

18.7 📝 动手练习

参考实现放 code/18-sqlite/（写作时同步给出）。

1. rusqlite 最小读写：:memory: 建表/插入/查询/迭代，练参数化与 query_map 收行。
2. 防注入验证：向 18.2 的插入函数传 x'); DROP TABLE kv; --，确认它只是被当作普通字符串存进去（参数化生效）。
3. 完整实现 HistoryStore for SqliteStore：按 18.4 补全全部 6 个方法，并跑通 18.6 的冒烟测试。
4. 迁移实验：先建 version=1 的库，再用"version=2：给 messages 加一列 complete INTEGER DEFAULT 1"的迁移脚本升级，验证老数据不丢、新字段可写。
5. 忙锁观察：开两个 SqliteStore 指向同一文件，线程 A 在事务里 sleep 200ms 后提交，线程 B 同时写——先不设 busy_timeout 记录报错，再设 5s 观察排队成功。
6. 并发冒烟：8 个 tokio 任务各 append 50 条消息到同一会话，结束后断言总数 = 400（体会 Mutex 串行化 + WAL 的效果）。
7. 综合（重点）：给 messages 补 complete 列并在模型加字段，让 ask_stream 断流时（FakeLlm 抛 StreamInterrupted）能落一条 complete=false 的尾巴消息——为 19 课"流中断可恢复"预演。

验收门禁：能写出带 ?1 占位 + params! 的最小读写；能说出 WAL 解决了什么、Mutex 解决了什么；能解释为什么 schema 变更必须走版本迁移而不是改旧 SQL。

✅ 本节小结

• rusqlite：bundled 特性免系统依赖；:memory: 便于测试；永远参数化 SQL；
• schema：sessions/messages 两张表 + 复合索引 + ON DELETE CASCADE；PRAGMA user_version 做增量迁移；
• 枚举映射：DB 存 TEXT，行读取时再转回 Rust enum；
• 并发：Mutex<Connection> 串行化（天然 Sync）、WAL 读写分离、busy_timeout 兜底；
• 接口红利：换 store 实现，service 测试与调用方零改动；
• 纪律：持锁不做 IO、schema 只加不改、迁移逐版本前进。

下一课预告：第 19 课《UniFFI 导出核心能力》——把 17-18 课的 core 交给各端：UniFFI 生成 Swift/Kotlin/Python 绑定、async 方法导出、复杂类型（Session/Message/enum 错误）映射、以及"如何在壳与 core 之间传回调"。这是实战篇的技术主峰。