tauri-plugin-deeprespond-im 1.0.0-alpha

//! 对象仓库的结构（《1.本地存储》第 3、8 节）：JavaScript 在 db_open 时交来 StoreSchema 列表（与 Web SDK 相同），
//! 插件按 Flutter SDK 的物理结构建表：每个对象仓库一张 `WITHOUT ROWID` 表，列为 `k`（编码后的主键）、`v`（记录的 JSON）、
//! `i_{索引名}`（编码后的索引键，取不到时为 NULL），每个索引一个只含非 NULL 行的部分索引。结构版本记在 `_dr_schema` 中。

use std::collections::HashMap;
use std::sync::Arc;

use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior};
use serde::Deserialize;
use serde_json::Value;

use super::keys::{extract_key, KeyPath};
use crate::error::{Error, Result};

/// JavaScript 交来的对象仓库结构（同 Web SDK 的 `StoreSchema`）。
#[derive(Clone, Debug, Deserialize)]
pub struct StoreSchema {
    pub name: String,
    #[serde(rename = "keyPath", alias = "key_path")]
    pub key_path: KeyPath,
    #[serde(default)]
    pub indexes: Option<Vec<IndexSchema>>,
}

#[derive(Clone, Debug, Deserialize)]
pub struct IndexSchema {
    pub name: String,
    #[serde(rename = "keyPath", alias = "key_path")]
    pub key_path: KeyPath,
    /// IndexedDB 的唯一索引。Web SDK 的结构中没有用到，与 Flutter SDK 一样不检查。
    #[serde(default)]
    #[allow(dead_code)]
    pub unique: Option<bool>,
}

/// 一个索引：列名为 `i_{索引名}`。
#[derive(Debug)]
pub struct IndexDef {
    pub name: String,
    pub key_path: KeyPath,
    pub column: String,
}

/// 一个对象仓库，写入的 SQL 预先拼好。
#[derive(Debug)]
pub struct StoreDef {
    pub name: String,
    pub key_path: KeyPath,
    pub indexes: Vec<IndexDef>,
    pub put_sql: String,
}

impl StoreDef {
    pub fn index(&self, name: &str) -> Option<&IndexDef> {
        self.indexes.iter().find(|i| i.name == name)
    }
}

/// 校验过的结构。
#[derive(Debug)]
pub struct Schema {
    list: Vec<StoreSchema>,
    stores: HashMap<String, Arc<StoreDef>>,
}

impl Schema {
    /// 校验名称并预先拼好 SQL：名称直接写进 SQL（与 Flutter 相同，索引列不加引号），只允许字母、数字和下划线。
    pub fn compile(list: Vec<StoreSchema>) -> Result<Schema> {
        if list.is_empty() {
            return Err(invalid("没有对象仓库"));
        }
        let mut stores = HashMap::new();
        for store in &list {
            if !valid_name(&store.name) || store.name.starts_with("_dr_") || store.name.starts_with("sqlite_") {
                return Err(invalid(format!("对象仓库的名称不合法：{}", store.name)));
            }
            check_key_path(&store.key_path)?;
            let mut indexes = Vec::new();
            for index in store.indexes.iter().flatten() {
                if !valid_name(&index.name) || indexes.iter().any(|i: &IndexDef| i.name == index.name) {
                    return Err(invalid(format!("索引的名称不合法或重复：{}.{}", store.name, index.name)));
                }
                check_key_path(&index.key_path)?;
                indexes.push(IndexDef { name: index.name.clone(), key_path: index.key_path.clone(), column: format!("i_{}", index.name) });
            }
            let mut columns = vec!["k".to_string(), "v".to_string()];
            columns.extend(indexes.iter().map(|i| i.column.clone()));
            let updates: Vec<String> = columns.iter().skip(1).map(|c| format!("{c} = excluded.{c}")).collect();
            let put_sql = format!(
                "INSERT INTO \"{}\" ({}) VALUES ({}) ON CONFLICT(k) DO UPDATE SET {}",
                store.name,
                columns.join(", "),
                vec!["?"; columns.len()].join(", "),
                updates.join(", ")
            );
            let def = StoreDef { name: store.name.clone(), key_path: store.key_path.clone(), indexes, put_sql };
            if stores.insert(store.name.clone(), Arc::new(def)).is_some() {
                return Err(invalid(format!("对象仓库重复：{}", store.name)));
            }
        }
        Ok(Schema { list, stores })
    }

    pub fn store(&self, name: &str) -> Option<&Arc<StoreDef>> {
        self.stores.get(name)
    }
}

fn invalid(message: impl Into<String>) -> Error {
    Error::validation("invalid_schema", message)
}

fn valid_name(name: &str) -> bool {
    !name.is_empty() && name.len() <= 64 && name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')
}

fn check_key_path(path: &KeyPath) -> Result<()> {
    let paths = path.paths();
    if paths.is_empty() || paths.iter().any(|p| p.is_empty()) {
        return Err(invalid("keyPath 为空"));
    }
    Ok(())
}

/// 建表和建索引的 SQL，与 Flutter 的 ObjectDatabase._createSql 逐字相同。
pub fn create_sql(store: &StoreDef) -> Vec<String> {
    let mut columns = vec!["k BLOB PRIMARY KEY NOT NULL".to_string(), "v TEXT NOT NULL".to_string()];
    columns.extend(store.indexes.iter().map(|i| format!("{} BLOB", i.column)));
    let mut out = vec![format!("CREATE TABLE IF NOT EXISTS \"{}\" ({}) WITHOUT ROWID", store.name, columns.join(", "))];
    for index in &store.indexes {
        out.push(format!(
            "CREATE INDEX IF NOT EXISTS \"{name}__{index}\" ON \"{name}\" ({col}, k) WHERE {col} IS NOT NULL",
            name = store.name,
            index = index.name,
            col = index.column
        ));
    }
    out
}

/// 读出结构版本（新文件为 0）。第一次读取表时，密钥不对、文件损坏在这里以 NOTADB/CORRUPT 失败。
pub fn read_version(conn: &Connection) -> rusqlite::Result<u32> {
    conn.execute_batch("CREATE TABLE IF NOT EXISTS _dr_schema (version INTEGER NOT NULL)")?;
    let version: Option<i64> = conn.query_row("SELECT version FROM _dr_schema", [], |r| r.get(0)).optional()?;
    Ok(version.unwrap_or(0).clamp(0, u32::MAX as i64) as u32)
}

/// 升级到 version：在一个事务中建立缺少的表和索引；已有的表缺少索引列的，加列并按记录补上索引键；
/// 结构中没有的对象仓库删除（与 IndexedDB 的升级相同，Web SDK 的内存实现也是这样）。
pub fn upgrade(conn: &mut Connection, schema: &Schema, version: u32) -> rusqlite::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    for store in &schema.list {
        let def = &schema.stores[&store.name];
        let columns = table_columns(&tx, &def.name)?;
        if !columns.is_empty() {
            for index in &def.indexes {
                if !columns.contains(&index.column) {
                    tx.execute_batch(&format!("ALTER TABLE \"{}\" ADD COLUMN {} BLOB", def.name, index.column))?;
                    backfill(&tx, def, index)?;
                }
            }
        }
        for sql in create_sql(def) {
            tx.execute_batch(&sql)?;
        }
    }
    let existing: Vec<String> = {
        let mut stmt = tx.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name <> '_dr_schema'")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        rows.collect::<rusqlite::Result<_>>()?
    };
    for name in existing {
        if schema.stores.contains_key(&name) {
            continue;
        }
        // 只删除对象仓库的表（有 k、v 两列），其他表（App 自己建的）不动
        let columns = table_columns(&tx, &name)?;
        if columns.iter().any(|c| c == "k") && columns.iter().any(|c| c == "v") {
            log::info!("结构升级：删除结构中已没有的对象仓库 {name}");
            tx.execute_batch(&format!("DROP TABLE \"{}\"", name.replace('"', "\"\"")))?;
        }
    }
    tx.execute("DELETE FROM _dr_schema", [])?;
    tx.execute("INSERT INTO _dr_schema (version) VALUES (?1)", params![version])?;
    tx.commit()
}

fn table_columns(conn: &Connection, table: &str) -> rusqlite::Result<Vec<String>> {
    let mut stmt = conn.prepare("SELECT name FROM pragma_table_info(?1)")?;
    let rows = stmt.query_map([table], |r| r.get::<_, String>(0))?;
    rows.collect()
}

/// 新加的索引列：按已有记录取出索引键写入。
fn backfill(conn: &Connection, store: &StoreDef, index: &IndexDef) -> rusqlite::Result<()> {
    let rows: Vec<(Vec<u8>, String)> = {
        let mut stmt = conn.prepare(&format!("SELECT k, v FROM \"{}\"", store.name))?;
        let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
        rows.collect::<rusqlite::Result<_>>()?
    };
    let mut update = conn.prepare(&format!("UPDATE \"{}\" SET {} = ?1 WHERE k = ?2", store.name, index.column))?;
    for (k, v) in rows {
        let value: Value = serde_json::from_str(&v).unwrap_or(Value::Null);
        if let Some(key) = extract_key(&value, &index.key_path) {
            update.execute(params![key, k])?;
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn messages_schema() -> Vec<StoreSchema> {
        serde_json::from_value(json!([
            { "name": "messages", "keyPath": ["conversation_id", "seq"], "indexes": [
                { "name": "client", "keyPath": ["conversation_id", "_local.client_msg_id"] },
                { "name": "created", "keyPath": "_local.created_ms" }
            ] },
            { "name": "meta", "keyPath": "key" }
        ]))
        .unwrap()
    }

    #[test]
    fn sql_matches_flutter() {
        let schema = Schema::compile(messages_schema()).unwrap();
        let sql = create_sql(schema.store("messages").unwrap());
        assert_eq!(sql[0], "CREATE TABLE IF NOT EXISTS \"messages\" (k BLOB PRIMARY KEY NOT NULL, v TEXT NOT NULL, i_client BLOB, i_created BLOB) WITHOUT ROWID");
        assert_eq!(sql[2], "CREATE INDEX IF NOT EXISTS \"messages__created\" ON \"messages\" (i_created, k) WHERE i_created IS NOT NULL");
        assert_eq!(
            schema.store("messages").unwrap().put_sql,
            "INSERT INTO \"messages\" (k, v, i_client, i_created) VALUES (?, ?, ?, ?) ON CONFLICT(k) DO UPDATE SET v = excluded.v, i_client = excluded.i_client, i_created = excluded.i_created"
        );
    }

    #[test]
    fn rejects_bad_names() {
        for bad in [json!([{ "name": "a\"b", "keyPath": "k" }]), json!([{ "name": "_dr_schema", "keyPath": "k" }]), json!([{ "name": "a", "keyPath": "" }]), json!([])] {
            let list: Vec<StoreSchema> = serde_json::from_value(bad).unwrap();
            assert_eq!(Schema::compile(list).unwrap_err().reason.as_deref(), Some("invalid_schema"));
        }
    }

    #[test]
    fn upgrade_adds_index_and_drops_removed_store() {
        let mut conn = Connection::open_in_memory().unwrap();
        assert_eq!(read_version(&conn).unwrap(), 0);
        let v1: Vec<StoreSchema> = serde_json::from_value(json!([{ "name": "users", "keyPath": "username" }, { "name": "old", "keyPath": "id" }])).unwrap();
        upgrade(&mut conn, &Schema::compile(v1).unwrap(), 1).unwrap();
        conn.execute("INSERT INTO users (k, v) VALUES (?1, ?2)", params![vec![1u8], r#"{"username":"a","nick":"x"}"#]).unwrap();
        let v2: Vec<StoreSchema> = serde_json::from_value(json!([{ "name": "users", "keyPath": "username", "indexes": [{ "name": "nick", "keyPath": "nick" }] }])).unwrap();
        upgrade(&mut conn, &Schema::compile(v2).unwrap(), 2).unwrap();
        assert_eq!(read_version(&conn).unwrap(), 2);
        let nick: Vec<u8> = conn.query_row("SELECT i_nick FROM users", [], |r| r.get(0)).unwrap();
        assert_eq!(Some(nick), crate::store::keys::encode_key(&json!("x")));
        let old: i64 = conn.query_row("SELECT count(*) FROM sqlite_master WHERE name = 'old'", [], |r| r.get(0)).unwrap();
        assert_eq!(old, 0);
    }
}