tauri-plugin-deeprespond-im 1.0.0-alpha

//! 本地存储的测试（《4.实现与测试》第 7 节）：键的编码与 Flutter SDK 共用测试数据；事务的提交、回滚、超时、
//! 页面实例消失；读写并发；结构降级、文件损坏、加密设置变化；删除；磁盘已满。直接测 StoreManager，不经 Tauri 命令。

use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde_json::{json, Value};

use super::keys::{encode_key, KeyRangeArg};
use super::*;
use crate::secure::{Fallback, KeyStore, Secure};

#[derive(Default)]
struct MemKeys(Mutex<Option<String>>);

impl KeyStore for MemKeys {
    fn get(&self) -> std::result::Result<Option<String>, String> {
        Ok(self.0.lock().unwrap().clone())
    }
    fn set(&self, value: &str) -> std::result::Result<(), String> {
        *self.0.lock().unwrap() = Some(value.into());
        Ok(())
    }
}

type FullLog = Arc<Mutex<Vec<(String, usize)>>>;

fn manager(dir: &Path, idle_ms: u64) -> (StoreManager, FullLog) {
    let secure = Arc::new(Secure::new(dir.to_path_buf(), Fallback::File, Box::new(MemKeys::default())));
    let full: FullLog = Arc::default();
    let log = full.clone();
    let on_full: OnFull = Arc::new(move |app_key, freed| log.lock().unwrap().push((app_key, freed)));
    (StoreManager::with_idle(dir.to_path_buf(), secure, on_full, Duration::from_millis(idle_ms)), full)
}

const APP: &str = "org#app";

fn engine(instance: u64) -> Caller {
    Caller { label: crate::hub::ENGINE_LABEL.into(), instance, engine: true }
}

fn window(instance: u64) -> Caller {
    Caller { label: "main".into(), instance, engine: false }
}

fn 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()
}

async fn open_v(m: &StoreManager, caller: &Caller, user: &str, version: u32) -> Result<Opened> {
    m.open(caller, OpenRequest { app_key: APP.into(), username: user.into(), version, schema: schema(), encrypt: false }).await
}

async fn open(m: &StoreManager, caller: &Caller, user: &str) -> Opened {
    open_v(m, caller, user, 1).await.unwrap()
}

fn op(value: Value) -> WriteOp {
    serde_json::from_value(value).unwrap()
}

fn put(store: &str, value: Value) -> WriteOp {
    op(json!({ "op": "put", "store": store, "value": value }))
}

fn message(seq: i64) -> Value {
    json!({ "conversation_id": if seq % 2 == 0 { "c1" } else { "c2" }, "seq": seq, "_local": { "client_msg_id": format!("m{seq}"), "created_ms": 1000 - seq } })
}

fn req(op: &str, store: &str) -> ReadRequest {
    ReadRequest { op: op.into(), store: store.into(), key: None, index: None, range: None, limit: None, direction: None, cursor: None }
}

fn range(value: Value) -> Option<KeyRangeArg> {
    Some(serde_json::from_value(value).unwrap())
}

fn to_value(result: ReadResult) -> Value {
    serde_json::from_str(&serde_json::to_string(&result).unwrap()).unwrap()
}

fn seqs(items: &Value) -> Vec<i64> {
    items.as_array().unwrap().iter().map(|m| m["seq"].as_i64().unwrap()).collect()
}

async fn write_all(m: &StoreManager, caller: &Caller, db: u64, stores: &[&str], ops: Vec<WriteOp>) {
    let tx = m.begin(caller, db, stores.iter().map(|s| s.to_string()).collect(), "readwrite", None).await.unwrap();
    m.commit(caller, tx, ops).await.unwrap();
}

async fn read_one(m: &StoreManager, caller: &Caller, db: u64, store: &str, key: Value) -> Value {
    let tx = m.begin(caller, db, vec![store.into()], "readonly", None).await.unwrap();
    let result = m.read(caller, tx, ReadRequest { key: Some(key), ..req("get", store) }).await.unwrap();
    m.commit(caller, tx, Vec::new()).await.unwrap();
    to_value(result)
}

fn conformance_path() -> std::path::PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("../../conformance/storage-keys.json")
}

#[test]
fn keys_match_shared_test_data() {
    let text = std::fs::read_to_string(conformance_path()).expect("sdk/conformance/storage-keys.json");
    let data: Value = serde_json::from_str(&text).unwrap();
    let cases = data["cases"].as_array().unwrap();
    assert!(cases.len() >= 50);
    for case in cases {
        assert_eq!(hex::encode(encode_key(&case["key"]).unwrap()), case["hex"].as_str().unwrap(), "{}", case["key"]);
    }
    let order: Vec<Vec<u8>> = data["order"].as_array().unwrap().iter().map(|k| encode_key(k).unwrap()).collect();
    assert_eq!(order.len(), cases.len());
    for pair in order.windows(2) {
        assert!(pair[0] < pair[1], "顺序不对：{} 应小于 {}", hex::encode(&pair[0]), hex::encode(&pair[1]));
    }
}

/// 生成共用测试数据：`cargo test store::tests::generate_shared_test_data -- --ignored`，再用 Flutter 的测试核对。
#[test]
#[ignore]
fn generate_shared_test_data() {
    let keys = json!([
        -1e300, -9007199254740991i64, -1000, -1.5, -1, -0.5, -5e-324, 0, 5e-324, 1e-300, 0.1, 0.5, 1, 1.5, 2, 10, 255, 256, 1000,
        1700000000000i64, 9007199254740991i64, 1e21, 1.7976931348623157e308,
        "", "\u{0}", "\u{1}", "\u{2}", "A", "Z", "a", "a\u{0}", "a\u{0}b", "a\u{1}", "a\u{1}b", "a\u{2}", "ab", "abc", "b", "~", "\u{7f}",
        "é", "中", "中文", "日本", "😀", "a😀", "conv_1", "conv_10", "conv_2",
        [], [0], [1], [1, 2], [1, "a"], [2], ["a"], ["a", 1], ["a", []], ["c1", 5], ["c1", 10], ["c1", "m1"], ["c1", []], ["c2", 1],
        [[]], [[], 1], [[1]], [["a"]], [1, [2, ["x"]]]
    ]);
    let keys = keys.as_array().unwrap();
    let cases: Vec<Value> = keys.iter().map(|k| json!({ "key": k, "hex": hex::encode(encode_key(k).unwrap()) })).collect();
    let mut order: Vec<&Value> = keys.iter().collect();
    order.sort_by_key(|k| encode_key(k).unwrap());
    let doc = json!({
        "description": "对象仓库的键编码（Flutter SDK 的 keys.dart 与 Tauri 插件的 store/keys.rs 共用）：cases 为键与编码后的十六进制，order 为全部键按升序排列",
        "cases": cases,
        "order": order,
    });
    std::fs::write(conformance_path(), serde_json::to_string_pretty(&doc).unwrap() + "\n").unwrap();
}

#[tokio::test]
async fn crud_index_range_direction_and_cursor() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let db = open(&m, &e, "Alice").await;
    assert_eq!((db.old_version, db.reset), (0, None));
    write_all(&m, &e, db.db, &["messages"], (1..=450).map(|seq| put("messages", message(seq))).collect()).await;

    let tx = m.begin(&e, db.db, vec!["messages".into()], "readonly", None).await.unwrap();
    let (mm, ec) = (&m, &e);
    let r = move |q: ReadRequest| mm.read(ec, tx, q);
    assert_eq!(to_value(r(ReadRequest { key: Some(json!(["c1", 10])), ..req("get", "messages") }).await.unwrap())["seq"], 10);
    assert_eq!(to_value(r(ReadRequest { key: Some(json!(["c1", 11])), ..req("get", "messages") }).await.unwrap()), Value::Null);
    let c1 = to_value(r(ReadRequest { range: range(json!({ "lower": ["c1"], "upper": ["c1", []] })), ..req("get_all", "messages") }).await.unwrap());
    assert_eq!(seqs(&c1).len(), 225);
    assert_eq!(seqs(&c1)[0], 2);
    let last = to_value(r(ReadRequest { range: range(json!({ "lower": ["c2"], "upper": ["c2", []] })), direction: Some("prev".into()), limit: Some(1), ..req("get_all", "messages") }).await.unwrap());
    assert_eq!(seqs(&last), vec![449]);
    let by_created = to_value(r(ReadRequest { index: Some("created".into()), limit: Some(3), ..req("get_all", "messages") }).await.unwrap());
    assert_eq!(seqs(&by_created), vec![450, 449, 448]);
    let n = to_value(r(ReadRequest { index: Some("created".into()), range: range(json!({ "lower": 900, "upper": 950, "upperOpen": true })), ..req("count", "messages") }).await.unwrap());
    assert_eq!(n, 50);
    assert_eq!(to_value(r(req("count", "messages")).await.unwrap()), 450);
    // 按范围倒序遍历：一批 200 条，与 Flutter 的测试相同
    let page = to_value(r(ReadRequest { range: range(json!({ "lower": ["c1", 0], "upper": ["c1", 400] })), direction: Some("prev".into()), ..req("iterate", "messages") }).await.unwrap());
    assert_eq!(seqs(&page["items"]).len(), 200);
    assert_eq!(seqs(&page["items"])[0], 400);
    // c1 在范围内正好 200 条：多取一条判断出后面没有了，游标为 null，不必再取一次空批
    assert_eq!(page["cursor"], Value::Null);
    // 按索引倒序遍历全部：200、200、50，最后一批的游标为 null
    let mut all = Vec::new();
    let mut cursor: Option<String> = None;
    let mut batches = Vec::new();
    loop {
        let page = to_value(r(ReadRequest { index: Some("created".into()), direction: Some("prev".into()), cursor: cursor.clone(), ..req("iterate", "messages") }).await.unwrap());
        batches.push(page["items"].as_array().unwrap().len());
        all.extend(seqs(&page["items"]));
        match page["cursor"].as_str() {
            Some(c) => cursor = Some(c.to_string()),
            None => break,
        }
    }
    assert_eq!(batches, vec![200, 200, 50]);
    assert_eq!(all, (1..=450).collect::<Vec<i64>>());
    // limit 小于一批：只取 limit 条，后面还有时给出游标
    let small = to_value(r(ReadRequest { index: Some("client".into()), range: range(json!({ "lower": ["c2"], "upper": ["c2", []] })), limit: Some(5), ..req("iterate", "messages") }).await.unwrap());
    assert_eq!(small["items"].as_array().unwrap().len(), 5);
    assert!(small["cursor"].is_string());
    let bad = r(ReadRequest { cursor: Some("zz".into()), ..req("iterate", "messages") }).await.unwrap_err();
    assert_eq!(bad.reason.as_deref(), Some("invalid_cursor"));
    let unknown = r(ReadRequest { index: Some("nope".into()), ..req("get_all", "messages") }).await.unwrap_err();
    assert_eq!(unknown.reason.as_deref(), Some("unknown_index"));
    let scope = r(req("get_all", "meta")).await.unwrap_err();
    assert_eq!(scope.reason.as_deref(), Some("store_not_in_scope"));
    m.abort(&e, tx);
}

#[tokio::test]
async fn delete_range_clear_commit_and_abort() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let db = open(&m, &e, "alice").await.db;
    let stores = ["messages", "meta"];
    write_all(&m, &e, db, &stores, (1..=20).map(|seq| put("messages", message(seq))).chain([put("meta", json!({ "key": "owner", "value": 1 }))]).collect()).await;

    // 删除一个范围（c1 的 seq 4～10）和一个键；abort 的写入不生效
    write_all(&m, &e, db, &stores, vec![
        op(json!({ "op": "delete", "store": "messages", "range": { "lower": ["c1", 4], "upper": ["c1", 10] } })),
        op(json!({ "op": "delete", "store": "messages", "key": ["c2", 1] })),
    ])
    .await;
    let tx = m.begin(&e, db, vec!["messages".into(), "meta".into()], "readwrite", Some("strict")).await.unwrap();
    m.write(&e, tx, vec![op(json!({ "op": "clear", "store": "meta" })), put("messages", message(99))]).await.unwrap();
    // 主键不合法：这一批失败，事务继续
    let err = m.write(&e, tx, vec![put("messages", message(100)), put("messages", json!({ "conversation_id": "c1" }))]).await.unwrap_err();
    assert_eq!((err.code.as_str(), err.reason.as_deref()), ("local_validation", Some("invalid_key")));
    let inside = to_value(m.read(&e, tx, req("count", "meta")).await.unwrap());
    assert_eq!(inside, 0);
    m.abort(&e, tx);
    m.abort(&e, tx);

    let rtx = m.begin(&e, db, vec!["messages".into(), "meta".into()], "readonly", None).await.unwrap();
    let left = to_value(m.read(&e, rtx, req("get_all", "messages")).await.unwrap());
    let mut got = seqs(&left);
    got.sort();
    assert_eq!(got, vec![2, 3, 5, 7, 9, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]);
    assert_eq!(to_value(m.read(&e, rtx, req("count", "meta")).await.unwrap()), 1);
    m.commit(&e, rtx, Vec::new()).await.unwrap();
    // 结束后的事务：tx_finished
    let ended = m.read(&e, rtx, req("count", "meta")).await.unwrap_err();
    assert_eq!(ended.reason.as_deref(), Some("tx_finished"));

    // tx_commit 带上 ops：先执行再提交；clear 后再写
    let tx = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    m.commit(&e, tx, vec![op(json!({ "op": "clear", "store": "meta" })), put("meta", json!({ "key": "k2", "value": "中文" }))]).await.unwrap();
    assert_eq!(read_one(&m, &e, db, "meta", json!("k2")).await, json!({ "key": "k2", "value": "中文" }));
    assert_eq!(read_one(&m, &e, db, "meta", json!("owner")).await, Value::Null);
    // commit 中的写入失败：整个事务回滚
    let tx = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    m.write(&e, tx, vec![put("meta", json!({ "key": "k3" }))]).await.unwrap();
    assert!(m.commit(&e, tx, vec![put("meta", json!({ "nokey": 1 }))]).await.is_err());
    assert_eq!(read_one(&m, &e, db, "meta", json!("k3")).await, Value::Null);
    // 只读事务不能写
    let rtx = m.begin(&e, db, vec!["meta".into()], "readonly", None).await.unwrap();
    assert_eq!(m.write(&e, rtx, vec![put("meta", json!({ "key": "x" }))]).await.unwrap_err().reason.as_deref(), Some("readonly_transaction"));
    m.abort(&e, rtx);
}

#[tokio::test]
async fn only_engine_writes() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let w = window(2);
    let db = open(&m, &engine(1), "alice").await.db;
    let wdb = open(&m, &w, "alice").await;
    assert_eq!(wdb.old_version, 1);
    let err = m.begin(&w, wdb.db, vec!["meta".into()], "readwrite", None).await.unwrap_err();
    assert_eq!((err.code.as_str(), err.reason.as_deref()), ("permission_denied", Some("not_engine_window")));
    // 句柄属于打开它的页面实例
    assert_eq!(m.begin(&w, db, vec!["meta".into()], "readonly", None).await.unwrap_err().reason.as_deref(), Some("db_not_found"));
    let tx = m.begin(&w, wdb.db, vec!["meta".into()], "readonly", None).await.unwrap();
    assert_eq!(m.read(&engine(1), tx, req("count", "meta")).await.unwrap_err().reason.as_deref(), Some("tx_not_found"));
    m.abort(&w, tx);
}

#[tokio::test]
async fn idle_write_transaction_is_rolled_back() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 300);
    let e = engine(1);
    let db = open(&m, &e, "alice").await.db;
    let tx = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    m.write(&e, tx, vec![put("meta", json!({ "key": "a" }))]).await.unwrap();
    tokio::time::sleep(Duration::from_millis(800)).await;
    let err = m.write(&e, tx, vec![put("meta", json!({ "key": "b" }))]).await.unwrap_err();
    assert_eq!(err.reason.as_deref(), Some("tx_timeout"));
    assert!(m.commit(&e, tx, Vec::new()).await.is_err());
    // 写连接已放开：新的写事务立即开始，之前的写入不在
    let next = tokio::time::timeout(Duration::from_secs(1), m.begin(&e, db, vec!["meta".into()], "readwrite", None)).await.unwrap().unwrap();
    m.commit(&e, next, Vec::new()).await.unwrap();
    assert_eq!(read_one(&m, &e, db, "meta", json!("a")).await, Value::Null);
    // 只读事务同样超时，连接还回池中
    let rtx = m.begin(&e, db, vec!["meta".into()], "readonly", None).await.unwrap();
    tokio::time::sleep(Duration::from_millis(800)).await;
    assert_eq!(m.read(&e, rtx, req("count", "meta")).await.unwrap_err().reason.as_deref(), Some("tx_timeout"));
    assert_eq!(m.core.txs.len(), 0);
}

#[tokio::test]
async fn long_operation_is_not_idle() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 300);
    let e = engine(1);
    let db = open(&m, &e, "alice").await.db;
    let tx = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    // 清空大表这类操作本身执行很久：执行期间不算空闲，结束后重新计时
    m.core.run_in_tx(&e, tx, |_| {
        std::thread::sleep(Duration::from_millis(900));
        Ok(())
    })
    .await
    .unwrap();
    m.write(&e, tx, vec![put("meta", json!({ "key": "a" }))]).await.unwrap();
    m.commit(&e, tx, Vec::new()).await.unwrap();
    assert_eq!(read_one(&m, &e, db, "meta", json!("a")).await, json!({ "key": "a" }));
}

#[tokio::test]
async fn reads_during_write_see_committed_snapshot() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let w = window(2);
    let db = open(&m, &e, "alice").await.db;
    let wdb = open(&m, &w, "alice").await.db;
    write_all(&m, &e, db, &["meta"], vec![put("meta", json!({ "key": "a", "value": 1 }))]).await;
    let tx = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    m.write(&e, tx, vec![put("meta", json!({ "key": "a", "value": 2 })), put("meta", json!({ "key": "b" }))]).await.unwrap();
    // 写事务进行中：界面窗口的读取不被阻塞，读到写事务开始前的数据
    let before = m.begin(&w, wdb, vec!["meta".into()], "readonly", None).await.unwrap();
    let seen = tokio::time::timeout(Duration::from_secs(2), m.read(&w, before, ReadRequest { key: Some(json!("a")), ..req("get", "meta") })).await.unwrap().unwrap();
    assert_eq!(to_value(seen)["value"], 1);
    assert_eq!(to_value(m.read(&w, before, req("count", "meta")).await.unwrap()), 1);
    m.commit(&e, tx, Vec::new()).await.unwrap();
    // 提交后：已经开始的只读事务仍是开始时的快照，新的读到提交后的数据
    assert_eq!(to_value(m.read(&w, before, ReadRequest { key: Some(json!("a")), ..req("get", "meta") }).await.unwrap())["value"], 1);
    m.abort(&w, before);
    assert_eq!(read_one(&m, &w, wdb, "meta", json!("a")).await["value"], 2);
}

#[tokio::test]
async fn write_transactions_queue_and_readers_are_pooled() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let m = Arc::new(m);
    let e = engine(1);
    let db = open(&m, &e, "alice").await.db;
    let first = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    let second = tokio::spawn({
        let (m, e) = (m.clone(), e.clone());
        async move { m.begin(&e, db, vec!["meta".into()], "readwrite", None).await }
    });
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(!second.is_finished(), "第二个写事务应排队");
    m.write(&e, first, vec![put("meta", json!({ "key": "a" }))]).await.unwrap();
    m.commit(&e, first, Vec::new()).await.unwrap();
    let second = tokio::time::timeout(Duration::from_secs(2), second).await.unwrap().unwrap().unwrap();
    assert_eq!(to_value(m.read(&e, second, ReadRequest { key: Some(json!("a")), ..req("get", "meta") }).await.unwrap())["key"], "a");
    m.abort(&e, second);

    // 4 个只读连接：第 5 个只读事务等其中一个结束
    let mut readers = Vec::new();
    for _ in 0..4 {
        readers.push(m.begin(&e, db, vec!["meta".into()], "readonly", None).await.unwrap());
    }
    let fifth = tokio::spawn({
        let (m, e) = (m.clone(), e.clone());
        async move { m.begin(&e, db, vec!["meta".into()], "readonly", None).await }
    });
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(!fifth.is_finished());
    m.abort(&e, readers[0]);
    let fifth = tokio::time::timeout(Duration::from_secs(2), fifth).await.unwrap().unwrap().unwrap();
    assert_eq!(to_value(m.read(&e, fifth, req("count", "meta")).await.unwrap()), 1);
}

#[tokio::test]
async fn release_instance_rolls_back_and_closes() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let old = engine(1);
    let db = open(&m, &old, "alice").await.db;
    let tx = m.begin(&old, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    m.write(&old, tx, vec![put("meta", json!({ "key": "half" }))]).await.unwrap();
    // 引擎崩溃：回滚它的事务、关闭它的句柄（没有其他句柄时关闭连接）
    m.release_instance(crate::hub::ENGINE_LABEL, 1);
    assert!(m.core.state.lock().unwrap().files.is_empty());
    assert!(m.write(&old, tx, vec![put("meta", json!({ "key": "x" }))]).await.is_err());
    assert_eq!(m.begin(&old, db, vec!["meta".into()], "readonly", None).await.unwrap_err().reason.as_deref(), Some("db_not_found"));
    // 已经不在的实例不能再打开
    assert_eq!(open_v(&m, &old, "alice", 1).await.unwrap_err().code, "aborted");
    let new = engine(3);
    let db = open(&m, &new, "alice").await.db;
    let tx = tokio::time::timeout(Duration::from_secs(1), m.begin(&new, db, vec!["meta".into()], "readwrite", None)).await.unwrap().unwrap();
    m.abort(&new, tx);
    assert_eq!(read_one(&m, &new, db, "meta", json!("half")).await, Value::Null);
}

#[tokio::test]
async fn schema_downgrade_and_corruption_reset() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let first = open_v(&m, &e, "alice", 2).await.unwrap();
    assert_eq!((first.old_version, first.reset), (0, None));
    write_all(&m, &e, first.db, &["meta"], vec![put("meta", json!({ "key": "a" }))]).await;
    m.close(&e, first.db);
    let again = open_v(&m, &e, "alice", 2).await.unwrap();
    assert_eq!((again.old_version, again.reset), (2, None));
    m.close(&e, again.db);
    // App 降级：版本比代码新，删除重建
    let down = open_v(&m, &e, "alice", 1).await.unwrap();
    assert_eq!((down.old_version, down.reset), (0, Some("schema_downgrade")));
    assert_eq!(read_one(&m, &e, down.db, "meta", json!("a")).await, Value::Null);
    // 文件损坏：删除重建（先等连接全部关闭，否则旧连接关闭时的检查点会写回文件头）
    m.close_all();
    let path = crate::paths::db_path(dir.path(), APP, "alice");
    let _ = std::fs::remove_file(path.with_extension("db-wal"));
    let _ = std::fs::remove_file(path.with_extension("db-shm"));
    std::fs::write(&path, vec![0x5au8; 8192]).unwrap();
    let broken = open_v(&m, &e, "alice", 1).await.unwrap();
    assert_eq!(broken.reset, Some("corrupted"));
    write_all(&m, &e, broken.db, &["meta"], vec![put("meta", json!({ "key": "b" }))]).await;
    assert_eq!(read_one(&m, &e, broken.db, "meta", json!("b")).await, json!({ "key": "b" }));
}

#[tokio::test]
async fn users_json_and_encryption_changed() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let users = crate::paths::app_dir(dir.path(), APP).join("users.json");
    std::fs::create_dir_all(users.parent().unwrap()).unwrap();
    std::fs::write(&users, r#"{"v":1,"items":[{"username":"alice","created_at":1,"last_used_at":1,"pending_delete":true,"extra":"keep"}]}"#).unwrap();
    let db = open(&m, &e, "Alice").await.db;
    let doc: Value = serde_json::from_str(&std::fs::read_to_string(&users).unwrap()).unwrap();
    let item = &doc["items"][0];
    assert_eq!((item["encrypted"].as_bool(), item["pending_delete"].as_bool(), item["extra"].as_str()), (Some(false), Some(false), Some("keep")));
    assert!(item["last_used_at"].as_i64().unwrap() > 1);
    m.close(&e, db);
    // 记下的是加密的，这次不加密：删除重建
    let text = std::fs::read_to_string(&users).unwrap().replace("\"encrypted\":false", "\"encrypted\":true");
    std::fs::write(&users, text).unwrap();
    let reopened = open(&m, &e, "alice").await;
    assert_eq!(reopened.reset, Some("encryption_changed"));
}

#[cfg(not(feature = "sqlcipher"))]
#[tokio::test]
async fn encrypt_requires_sqlcipher_feature() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let err = m.open(&engine(1), OpenRequest { app_key: APP.into(), username: "alice".into(), version: 1, schema: schema(), encrypt: true }).await.unwrap_err();
    assert_eq!((err.code.as_str(), err.reason.as_deref()), ("unsupported", Some("sqlcipher_required")));
}

#[cfg(feature = "sqlcipher")]
#[tokio::test]
async fn encrypted_database_roundtrip() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let req = || OpenRequest { app_key: APP.into(), username: "alice".into(), version: 1, schema: schema(), encrypt: true };
    let db = m.open(&e, req()).await.unwrap();
    write_all(&m, &e, db.db, &["meta"], vec![put("meta", json!({ "key": "secret-value" }))]).await;
    assert_eq!(read_one(&m, &e, db.db, "meta", json!("secret-value")).await["key"], "secret-value");
    m.close_all();
    let raw = std::fs::read(crate::paths::db_path(dir.path(), APP, "alice")).unwrap();
    assert!(!raw.starts_with(b"SQLite format 3"));
    let again = m.open(&engine(2), req()).await.unwrap();
    assert_eq!(again.reset, None);
    assert_eq!(read_one(&m, &engine(2), again.db, "meta", json!("secret-value")).await["key"], "secret-value");
}

#[cfg(feature = "sqlcipher")]
#[tokio::test]
async fn encryption_switched_or_key_lost() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let req = |encrypt| OpenRequest { app_key: APP.into(), username: "alice".into(), version: 1, schema: schema(), encrypt };
    let plain = m.open(&engine(1), req(false)).await.unwrap();
    write_all(&m, &engine(1), plain.db, &["meta"], vec![put("meta", json!({ "key": "a" }))]).await;
    m.close_all();
    // users.json 中没有记下加密设置（被引擎的旧数据覆盖）：按文件头判断出是明文库
    std::fs::remove_file(crate::paths::app_dir(dir.path(), APP).join("users.json")).unwrap();
    let enc = m.open(&engine(2), req(true)).await.unwrap();
    assert_eq!(enc.reset, Some("encryption_changed"));
    m.close_all();
    // 密钥丢失：打不开，删除重建
    m.core.secure.set(APP, "db_keys", None).unwrap();
    let lost = m.open(&engine(3), req(true)).await.unwrap();
    assert_eq!(lost.reset, Some("corrupted"));
}

#[tokio::test]
async fn delete_closes_connections_and_removes_files() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let w = window(2);
    let db = open(&m, &e, "alice").await.db;
    let wdb = open(&m, &w, "alice").await.db;
    write_all(&m, &e, db, &["meta"], vec![put("meta", json!({ "key": "a" }))]).await;
    let reading = m.begin(&w, wdb, vec!["meta".into()], "readonly", None).await.unwrap();
    let outbox = crate::paths::outbox_dir(dir.path(), APP, "alice");
    std::fs::create_dir_all(&outbox).unwrap();
    std::fs::write(outbox.join("f.bin"), b"x").unwrap();
    m.delete(APP, "ALICE").await.unwrap();
    let path = crate::paths::db_path(dir.path(), APP, "alice");
    assert!(!path.exists() && !path.with_extension("db-wal").exists());
    assert!(!crate::paths::user_dir(dir.path(), APP, "alice").exists());
    let users: Value = serde_json::from_str(&std::fs::read_to_string(crate::paths::app_dir(dir.path(), APP).join("users.json")).unwrap_or("{}".into())).unwrap();
    assert!(users["items"].as_array().is_none_or(|items| items.is_empty()));
    // 界面窗口的只读事务已回滚，句柄失效
    assert!(m.read(&w, reading, req("count", "meta")).await.is_err());
    assert_eq!(m.begin(&w, wdb, vec!["meta".into()], "readonly", None).await.unwrap_err().reason.as_deref(), Some("db_closed"));
    m.close(&w, wdb);
    m.close(&e, db);
    // 重新打开是一个新库
    let fresh = open(&m, &e, "alice").await;
    assert_eq!(fresh.old_version, 0);
}

#[tokio::test]
async fn disk_full_frees_other_users() {
    let dir = tempfile::tempdir().unwrap();
    let (m, full) = manager(dir.path(), 5000);
    let e = engine(1);
    let bob = open(&m, &e, "bob").await.db;
    m.close(&e, bob);
    let db = open(&m, &e, "alice").await.db;
    let users = crate::paths::app_dir(dir.path(), APP).join("users.json");
    assert_eq!(serde_json::from_str::<Value>(&std::fs::read_to_string(&users).unwrap()).unwrap()["items"].as_array().unwrap().len(), 2);
    let tx = m.begin(&e, db, vec!["meta".into()], "readwrite", None).await.unwrap();
    // 用 max_page_count 让数据库“满”：之后的写入以 SQLITE_FULL 失败
    m.core
        .run_in_tx(&e, tx, |conn| {
            let pages: i64 = conn.query_row("PRAGMA page_count", [], |r| r.get(0))?;
            conn.query_row(&format!("PRAGMA max_page_count = {}", pages + 1), [], |_| Ok(()))?;
            Ok(())
        })
        .await
        .unwrap();
    let err = m.write(&e, tx, vec![put("meta", json!({ "key": "big", "value": "x".repeat(200_000) }))]).await.unwrap_err();
    assert_eq!((err.code.as_str(), err.reason.as_deref()), ("storage_error", Some("quota_exceeded")));
    m.abort(&e, tx);
    for _ in 0..100 {
        if !full.lock().unwrap().is_empty() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert_eq!(full.lock().unwrap().clone(), vec![(APP.to_string(), 1)]);
    assert!(!crate::paths::db_path(dir.path(), APP, "bob").exists());
    assert!(crate::paths::db_path(dir.path(), APP, "alice").exists());
    let doc: Value = serde_json::from_str(&std::fs::read_to_string(&users).unwrap()).unwrap();
    assert_eq!(doc["items"].as_array().unwrap().iter().map(|i| i["username"].as_str().unwrap()).collect::<Vec<_>>(), vec!["alice"]);
    // 10 分钟内只处理一次
    m.core.storage_full(APP, "alice");
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert_eq!(full.lock().unwrap().len(), 1);
}

#[tokio::test]
async fn sizes_use_hashed_names() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    open(&m, &engine(1), "alice").await;
    let sizes = m.sizes();
    let list = sizes.as_array().unwrap();
    assert_eq!(list.len(), 1);
    assert!(!list[0]["file"].as_str().unwrap().contains("alice"));
    assert_eq!(list[0]["open"], true);
    assert!(list[0]["bytes"].as_u64().unwrap() > 0);
    m.close_all();
    assert_eq!(m.sizes()[0]["open"], false);
}

#[tokio::test]
async fn records_keep_field_order() {
    let dir = tempfile::tempdir().unwrap();
    let (m, _) = manager(dir.path(), 5000);
    let e = engine(1);
    let db = open(&m, &e, "alice").await.db;
    let value: Value = serde_json::from_str(r#"{"key":"k","z":1,"a":{"y":true,"b":null},"m":[3,"x"]}"#).unwrap();
    write_all(&m, &e, db, &["meta"], vec![put("meta", value)]).await;
    let tx = m.begin(&e, db, vec!["meta".into()], "readonly", None).await.unwrap();
    let result = m.read(&e, tx, ReadRequest { key: Some(json!("k")), ..req("get", "meta") }).await.unwrap();
    // 记录以存入的 JSON 文本原样返回，字段顺序不变（Web SDK 按 JSON.stringify 比较记录）
    assert_eq!(serde_json::to_string(&result).unwrap(), r#"{"key":"k","z":1,"a":{"y":true,"b":null},"m":[3,"x"]}"#);
    m.abort(&e, tx);
}