platform/storage/
db.rs

1//! 数据库连接和操作管理
2//!
3//! 提供基于 sqlx 的数据库连接池和基本操作
4
5use anyhow::Result;
6use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
7use std::path::Path;
8use std::str::FromStr;
9use std::time::Duration;
10
11/// 数据库管理器
12#[derive(Clone)]
13pub struct Database {
14    pool: SqlitePool,
15}
16
17impl Database {
18    /// 创建新的数据库实例
19    ///
20    /// # Arguments
21    /// * `path` - 数据库文件存储目录路径，必须已存在
22    ///   主数据库文件将存储为 `{path}/actrix.db`
23    pub async fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
24        let db_file = path.as_ref().join("actrix.db");
25
26        // 创建连接选项并启用 WAL 模式
27        let options = SqliteConnectOptions::from_str(&format!("sqlite:{}", db_file.display()))?
28            .create_if_missing(true)
29            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
30            .synchronous(sqlx::sqlite::SqliteSynchronous::Normal)
31            .busy_timeout(Duration::from_secs(5));
32
33        // 创建连接池
34        let pool = SqlitePoolOptions::new()
35            .max_connections(10)
36            .connect_with(options)
37            .await?;
38
39        let db = Self { pool };
40
41        // 初始化数据库表结构
42        db.initialize_schema().await?;
43
44        Ok(db)
45    }
46
47    /// 初始化数据库表结构
48    async fn initialize_schema(&self) -> Result<()> {
49        // 创建 Realm 表
50        sqlx::query(
51            "CREATE TABLE IF NOT EXISTS realm (
52                id INTEGER PRIMARY KEY AUTOINCREMENT,
53                name TEXT NOT NULL,
54                status TEXT NOT NULL DEFAULT 'Active',
55                enabled INTEGER NOT NULL DEFAULT 1,
56                expires_at INTEGER,
57                created_at INTEGER NOT NULL,
58                updated_at INTEGER,
59                secret_current TEXT NOT NULL DEFAULT '',
60                secret_previous_hash TEXT,
61                secret_previous_valid_until INTEGER
62            )",
63        )
64        .execute(&self.pool)
65        .await?;
66
67        // Set autoincrement start to 2^25 = 33554432
68        // Only insert if not already present (fresh database)
69        sqlx::query("INSERT OR IGNORE INTO sqlite_sequence(name, seq) VALUES('realm', 33554431)")
70            .execute(&self.pool)
71            .await?;
72
73        // 创建访问控制列表表
74        sqlx::query(
75            "CREATE TABLE IF NOT EXISTS actoracl (
76                rowid INTEGER PRIMARY KEY AUTOINCREMENT,
77                realm_id INTEGER NOT NULL,
78                source_realm_id INTEGER,
79                from_type TEXT NOT NULL,
80                to_type TEXT NOT NULL,
81                access INTEGER NOT NULL
82            )",
83        )
84        .execute(&self.pool)
85        .await?;
86
87        // 创建索引
88        sqlx::query(
89            "CREATE INDEX IF NOT EXISTS idx_realm_name
90             ON realm(name)",
91        )
92        .execute(&self.pool)
93        .await?;
94
95        sqlx::query(
96            "CREATE INDEX IF NOT EXISTS idx_actoracl_realm_id
97             ON actoracl(realm_id)",
98        )
99        .execute(&self.pool)
100        .await?;
101
102        sqlx::query(
103            "CREATE INDEX IF NOT EXISTS idx_actoracl_lookup
104             ON actoracl(realm_id, source_realm_id, from_type, to_type)",
105        )
106        .execute(&self.pool)
107        .await?;
108
109        // Pending registration data: AIS writes, signaling reads on WS connect
110        sqlx::query(
111            "CREATE TABLE IF NOT EXISTS pending_registration (
112                serial_number INTEGER PRIMARY KEY,
113                realm_id INTEGER NOT NULL,
114                service_spec_blob BLOB,
115                ws_address TEXT,
116                created_at INTEGER NOT NULL
117            )",
118        )
119        .execute(&self.pool)
120        .await?;
121
122        // Migrate: add ws_address column if it doesn't exist (for existing databases)
123        let _ = sqlx::query("ALTER TABLE pending_registration ADD COLUMN ws_address TEXT")
124            .execute(&self.pool)
125            .await; // intentionally ignore error (column may already exist)
126
127        // MFR (Manufacturer Registry) tables
128        // name = GitHub user/org login (lowercased), serves as manufacturer identity
129        sqlx::query(
130            "CREATE TABLE IF NOT EXISTS mfr (
131                id           INTEGER PRIMARY KEY AUTOINCREMENT,
132                name         TEXT    NOT NULL UNIQUE,
133                public_key   TEXT    NOT NULL DEFAULT '',
134                contact      TEXT,
135                status       TEXT    NOT NULL DEFAULT 'pending',
136                created_at   INTEGER NOT NULL,
137                updated_at   INTEGER,
138                verified_at  INTEGER,
139                suspended_at INTEGER,
140                revoked_at   INTEGER,
141                key_expires_at  INTEGER
142            )",
143        )
144        .execute(&self.pool)
145        .await?;
146
147        // GitHub verification challenge for identity verification
148        sqlx::query(
149            "CREATE TABLE IF NOT EXISTS mfr_challenge (
150                id          INTEGER PRIMARY KEY AUTOINCREMENT,
151                mfr_id      INTEGER NOT NULL REFERENCES mfr(id),
152                token       TEXT    NOT NULL,
153                verify_url  TEXT    NOT NULL DEFAULT '',
154                expires_at  INTEGER NOT NULL,
155                verified_at INTEGER,
156                created_at  INTEGER NOT NULL
157            )",
158        )
159        .execute(&self.pool)
160        .await?;
161
162        sqlx::query(
163            "CREATE TABLE IF NOT EXISTS mfr_package (
164                id           INTEGER PRIMARY KEY AUTOINCREMENT,
165                mfr_id       INTEGER NOT NULL REFERENCES mfr(id),
166                manufacturer TEXT    NOT NULL,
167                name         TEXT    NOT NULL,
168                version      TEXT    NOT NULL,
169                type_str     TEXT    NOT NULL,
170                target       TEXT    NOT NULL,
171                manifest     TEXT    NOT NULL,
172                signature    TEXT    NOT NULL,
173                status       TEXT    NOT NULL DEFAULT 'active',
174                published_at INTEGER NOT NULL,
175                revoked_at   INTEGER,
176                UNIQUE(manufacturer, name, version, target)
177            )",
178        )
179        .execute(&self.pool)
180        .await?;
181
182        // Migrate: add target column if it doesn't exist (for existing databases)
183        let _ = sqlx::query(
184            "ALTER TABLE mfr_package ADD COLUMN target TEXT NOT NULL DEFAULT 'wasm32-wasip1'",
185        )
186        .execute(&self.pool)
187        .await; // intentionally ignore error (column may already exist)
188
189        sqlx::query("CREATE INDEX IF NOT EXISTS idx_mfr_package_type ON mfr_package(type_str)")
190            .execute(&self.pool)
191            .await?;
192
193        sqlx::query(
194            "CREATE INDEX IF NOT EXISTS idx_mfr_package_mfr ON mfr_package(mfr_id, status)",
195        )
196        .execute(&self.pool)
197        .await?;
198
199        // Migrate: add proto_files column for proto filing (JSON text, nullable)
200        let _ = sqlx::query("ALTER TABLE mfr_package ADD COLUMN proto_files TEXT")
201            .execute(&self.pool)
202            .await; // intentionally ignore error (column may already exist)
203
204        // Migrate: add key_id column (auto-assigned on activate/renew)
205        let _ = sqlx::query("ALTER TABLE mfr ADD COLUMN key_id TEXT NOT NULL DEFAULT ''")
206            .execute(&self.pool)
207            .await;
208
209        // MFR key history: stores retired public keys for JWKS-style rotation
210        sqlx::query(
211            "CREATE TABLE IF NOT EXISTS mfr_key_history (
212                id           INTEGER PRIMARY KEY AUTOINCREMENT,
213                mfr_id       INTEGER NOT NULL REFERENCES mfr(id),
214                key_id       TEXT    NOT NULL,
215                public_key   TEXT    NOT NULL,
216                status       TEXT    NOT NULL DEFAULT 'retired',
217                created_at   INTEGER NOT NULL,
218                retired_at   INTEGER NOT NULL
219            )",
220        )
221        .execute(&self.pool)
222        .await?;
223
224        sqlx::query(
225            "CREATE INDEX IF NOT EXISTS idx_mfr_key_history_lookup
226             ON mfr_key_history(mfr_id, key_id)",
227        )
228        .execute(&self.pool)
229        .await?;
230
231        // Publish nonce table for Challenge-Response authentication on /mfr/pkg/publish
232        sqlx::query(
233            "CREATE TABLE IF NOT EXISTS mfr_publish_nonce (
234                id         INTEGER PRIMARY KEY AUTOINCREMENT,
235                mfr_id     INTEGER NOT NULL REFERENCES mfr(id),
236                nonce      BLOB    NOT NULL UNIQUE,
237                status     TEXT    NOT NULL DEFAULT 'pending',
238                created_at INTEGER NOT NULL,
239                expires_at INTEGER NOT NULL
240            )",
241        )
242        .execute(&self.pool)
243        .await?;
244
245        sqlx::query(
246            "CREATE INDEX IF NOT EXISTS idx_mfr_publish_nonce_expires
247             ON mfr_publish_nonce(expires_at)",
248        )
249        .execute(&self.pool)
250        .await?;
251
252        // AIS unpublished package manufacturer-proof nonce table. AIS inserts
253        // after manufacturer_auth_signature verification; the unique index is the
254        // replay guard.
255        sqlx::query(
256            "CREATE TABLE IF NOT EXISTS ais_manufacturer_auth_nonce (
257                id           INTEGER PRIMARY KEY AUTOINCREMENT,
258                manufacturer TEXT    NOT NULL,
259                key_id       TEXT    NOT NULL,
260                nonce        BLOB    NOT NULL,
261                created_at   INTEGER NOT NULL,
262                expires_at   INTEGER NOT NULL,
263                UNIQUE(manufacturer, key_id, nonce)
264            )",
265        )
266        .execute(&self.pool)
267        .await?;
268
269        sqlx::query(
270            "CREATE INDEX IF NOT EXISTS idx_ais_manufacturer_auth_nonce_expires
271             ON ais_manufacturer_auth_nonce(expires_at)",
272        )
273        .execute(&self.pool)
274        .await?;
275
276        // Backfill key_id for existing MFRs that have a public_key but empty key_id.
277        // This runs on every startup but is a no-op when all rows already have a key_id.
278        {
279            use base64::Engine as _;
280            use sha2::{Digest, Sha256};
281
282            let rows: Vec<(i64, String)> = sqlx::query_as(
283                "SELECT id, public_key FROM mfr WHERE key_id = '' AND public_key != ''",
284            )
285            .fetch_all(&self.pool)
286            .await
287            .unwrap_or_default();
288
289            for (id, public_key_b64) in rows {
290                if let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(&public_key_b64)
291                {
292                    let hash = Sha256::digest(&bytes);
293                    let hex_str: String = hash.iter().map(|b| format!("{b:02x}")).collect();
294                    let key_id = format!("mfr-{}", &hex_str[..16]);
295                    let _ = sqlx::query("UPDATE mfr SET key_id = ? WHERE id = ?")
296                        .bind(&key_id)
297                        .bind(id)
298                        .execute(&self.pool)
299                        .await;
300                    crate::recording::info!(
301                        "backfilled key_id from public_key fingerprint: id={}, key_id={}",
302                        id,
303                        key_id
304                    );
305                }
306            }
307        }
308
309        // AIS renewal token table — only stores SHA-256(token), never the raw token.
310        sqlx::query(
311            "CREATE TABLE IF NOT EXISTS ais_renewal_token (
312                id INTEGER PRIMARY KEY AUTOINCREMENT,
313                actor_id TEXT NOT NULL,
314                token_hash BLOB NOT NULL UNIQUE,
315                expires_at INTEGER NOT NULL,
316                created_at INTEGER NOT NULL
317            )",
318        )
319        .execute(&self.pool)
320        .await?;
321
322        sqlx::query(
323            "CREATE INDEX IF NOT EXISTS idx_ais_renewal_token_actor
324             ON ais_renewal_token(actor_id)",
325        )
326        .execute(&self.pool)
327        .await?;
328
329        sqlx::query(
330            "CREATE INDEX IF NOT EXISTS idx_ais_renewal_token_expires
331             ON ais_renewal_token(expires_at)",
332        )
333        .execute(&self.pool)
334        .await?;
335
336        Ok(())
337    }
338
339    /// 获取数据库连接池
340    pub fn get_pool(&self) -> &SqlitePool {
341        &self.pool
342    }
343
344    /// 执行 SQL 语句并返回影响的行数
345    pub async fn execute(&self, sql: &str) -> Result<u64> {
346        let result = sqlx::query(sql).execute(&self.pool).await?;
347        Ok(result.rows_affected())
348    }
349}
350
351use tokio::sync::OnceCell;
352
353/// 全局数据库实例
354static GLOBAL_DATABASE: OnceCell<Database> = OnceCell::const_new();
355
356/// 设置全局数据库路径
357pub async fn set_db_path(path: &Path) -> Result<()> {
358    let database = Database::new(path).await?;
359    GLOBAL_DATABASE
360        .set(database)
361        .map_err(|_| anyhow::anyhow!("Database already initialized"))?;
362    Ok(())
363}
364
365/// 获取全局数据库实例
366pub fn get_database() -> &'static Database {
367    GLOBAL_DATABASE
368        .get()
369        .expect("Database not initialized. Call set_db_path first.")
370}
371
372/// 检查数据库是否已初始化
373pub fn is_database_initialized() -> bool {
374    GLOBAL_DATABASE.get().is_some()
375}
platform/storage/db.rs

platform/storage/
db.rs
