tauri-plugin-deeprespond-im 1.0.0-alpha

//! 事务中的读写，在连接所在的线程中执行。SQL 与 Flutter 的 Tx（object_db.dart）相同：按主键或 `(索引列, k)` 排序，
//! 范围是编码后的字节比较，遍历按上一批最后一条（索引键 + 主键）继续。记录以 JSON 文本原样返回，不重复解析。

use std::sync::Arc;

use rusqlite::types::{ToSql, Value as SqlValue};
use rusqlite::{params_from_iter, Connection};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use serde_json::Value;

use super::keys::{encoded_len, extract_key, require_key, KeyRangeArg, Range};
use super::schema::{Schema, StoreDef};
use crate::error::{Error, Result};

/// iterate 每批最多的条数（《1.本地存储》第 5.2 节）。
pub const PAGE_SIZE: u64 = 200;

/// tx_write、tx_commit 中的一项写入。
#[derive(Debug, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub enum WriteOp {
    Put {
        store: String,
        value: Value,
    },
    Delete {
        store: String,
        #[serde(default)]
        key: Option<Value>,
        #[serde(default)]
        range: Option<KeyRangeArg>,
    },
    Clear {
        store: String,
    },
}

/// 校验过、键已编码的写入：主键不合法等错误在交给数据库线程之前就报出。
pub enum Prepared {
    Put { def: Arc<StoreDef>, key: Vec<u8>, json: String, indexes: Vec<Option<Vec<u8>>> },
    Delete { store: String, key: Vec<u8> },
    DeleteRange { store: String, range: Range },
    Clear { store: String },
}

/// 事务范围内的对象仓库。
pub fn store_in_scope<'a>(schema: &'a Schema, scope: &[String], name: &str) -> Result<&'a Arc<StoreDef>> {
    if !scope.iter().any(|s| s == name) {
        return Err(Error::validation("store_not_in_scope", format!("对象仓库 {name} 不在事务的范围内")));
    }
    schema.store(name).ok_or_else(|| Error::validation("unknown_store", format!("没有对象仓库 {name}")))
}

pub fn prepare(schema: &Schema, scope: &[String], ops: Vec<WriteOp>) -> Result<Vec<Prepared>> {
    let mut out = Vec::with_capacity(ops.len());
    for op in ops {
        out.push(match op {
            WriteOp::Put { store, value } => {
                let def = store_in_scope(schema, scope, &store)?.clone();
                let key = extract_key(&value, &def.key_path)
                    .ok_or_else(|| Error::validation("invalid_key", format!("记录没有合法的主键（{store}，keyPath {:?}）", def.key_path.paths())))?;
                let indexes = def.indexes.iter().map(|i| extract_key(&value, &i.key_path)).collect();
                let json = serde_json::to_string(&value)?;
                Prepared::Put { def, key, json, indexes }
            }
            WriteOp::Delete { store, key, range } => {
                store_in_scope(schema, scope, &store)?;
                match (range, key) {
                    (Some(range), _) => Prepared::DeleteRange { store, range: range.encode()? },
                    // key 是 { lower, upper } 对象的也按范围处理（与 Web SDK 的 isRange 相同）；其他对象不是合法的键
                    (None, Some(Value::Object(map))) if map.contains_key("lower") || map.contains_key("upper") => {
                        let range: KeyRangeArg = serde_json::from_value(Value::Object(map))?;
                        Prepared::DeleteRange { store, range: range.encode()? }
                    }
                    (None, Some(key)) => Prepared::Delete { store, key: require_key(&key)? },
                    (None, None) => return Err(Error::validation("invalid_key", "删除时没有给出键或范围")),
                }
            }
            WriteOp::Clear { store } => {
                store_in_scope(schema, scope, &store)?;
                Prepared::Clear { store }
            }
        });
    }
    Ok(out)
}

/// 依次执行写入；出错时停在出错的一项，由调用方决定撤销到哪里。
pub fn apply(conn: &Connection, ops: &[Prepared]) -> Result<()> {
    for op in ops {
        match op {
            Prepared::Put { def, key, json, indexes } => {
                let mut stmt = conn.prepare_cached(&def.put_sql)?;
                let mut args: Vec<&dyn ToSql> = Vec::with_capacity(2 + indexes.len());
                args.push(key);
                args.push(json);
                for index in indexes {
                    args.push(index);
                }
                stmt.execute(args.as_slice())?;
            }
            Prepared::Delete { store, key } => {
                conn.prepare_cached(&format!("DELETE FROM \"{store}\" WHERE k = ?"))?.execute([key])?;
            }
            Prepared::DeleteRange { store, range } => {
                let mut conditions = Vec::new();
                let mut args = Vec::new();
                range_where("k", range, &mut conditions, &mut args);
                let mut sql = format!("DELETE FROM \"{store}\"");
                if !conditions.is_empty() {
                    sql.push_str(" WHERE ");
                    sql.push_str(&conditions.join(" AND "));
                }
                conn.prepare_cached(&sql)?.execute(params_from_iter(args))?;
            }
            Prepared::Clear { store } => {
                conn.prepare_cached(&format!("DELETE FROM \"{store}\""))?.execute([])?;
            }
        }
    }
    Ok(())
}

/// tx_write 的一批：要么全部生效，要么全部撤销（保存点），事务本身继续。
pub fn apply_batch(conn: &Connection, ops: &[Prepared]) -> Result<()> {
    if ops.is_empty() {
        return Ok(());
    }
    conn.execute_batch("SAVEPOINT drim_batch")?;
    match apply(conn, ops) {
        Ok(()) => {
            conn.execute_batch("RELEASE drim_batch")?;
            Ok(())
        }
        Err(err) => {
            // 磁盘已满等错误时 SQLite 可能已经回滚了整个事务，这时保存点也不在了
            if !conn.is_autocommit() {
                let _ = conn.execute_batch("ROLLBACK TO drim_batch; RELEASE drim_batch");
            }
            Err(err)
        }
    }
}

/// 读取的条件：对象仓库、按哪一列（None 为主键）、范围、方向、条数。
pub struct Query {
    pub store: String,
    pub column: Option<String>,
    pub range: Range,
    pub reverse: bool,
    pub limit: Option<u64>,
}

/// 遍历的游标：上一批最后一条的索引键（按索引遍历时）和主键。
pub struct Cursor {
    column: Option<Vec<u8>>,
    key: Vec<u8>,
}

impl Cursor {
    /// 十六进制的“索引键 + 主键”：编码后的键自带结尾，按长度拆开，不需要分隔符。
    pub fn parse(text: &str, indexed: bool) -> Result<Cursor> {
        let bad = || Error::validation("invalid_cursor", "游标不合法");
        let bytes = hex::decode(text).map_err(|_| bad())?;
        let (column, key) = if indexed {
            let n = encoded_len(&bytes).ok_or_else(bad)?;
            (Some(bytes[..n].to_vec()), bytes[n..].to_vec())
        } else {
            (None, bytes)
        };
        if encoded_len(&key) != Some(key.len()) {
            return Err(bad());
        }
        Ok(Cursor { column, key })
    }

    fn encode(column: Option<&[u8]>, key: &[u8]) -> String {
        let mut bytes = column.map(<[u8]>::to_vec).unwrap_or_default();
        bytes.extend_from_slice(key);
        hex::encode(bytes)
    }
}

/// tx_read 的结果。
#[derive(Debug, Serialize)]
#[serde(untagged)]
pub enum ReadResult {
    One(Option<Box<RawValue>>),
    Many(Vec<Box<RawValue>>),
    Count(u64),
    Page { items: Vec<Box<RawValue>>, cursor: Option<String> },
}

fn range_where(column: &str, range: &Range, conditions: &mut Vec<String>, args: &mut Vec<SqlValue>) {
    if let Some((key, open)) = &range.lower {
        conditions.push(format!("{column} {} ?", if *open { ">" } else { ">=" }));
        args.push(SqlValue::Blob(key.clone()));
    }
    if let Some((key, open)) = &range.upper {
        conditions.push(format!("{column} {} ?", if *open { "<" } else { "<=" }));
        args.push(SqlValue::Blob(key.clone()));
    }
}

/// 拼出 SELECT：与 Flutter 的 Tx._query 相同；条数用参数，SQL 文本稳定，可以缓存预编译的语句。
fn select_sql(q: &Query, select: &str, after: Option<&Cursor>, limit: Option<u64>) -> Result<(String, Vec<SqlValue>)> {
    let column = q.column.as_deref().unwrap_or("k");
    let mut conditions = Vec::new();
    let mut args = Vec::new();
    if q.column.is_some() {
        conditions.push(format!("{column} IS NOT NULL"));
    }
    range_where(column, &q.range, &mut conditions, &mut args);
    let op = if q.reverse { "<" } else { ">" };
    if let Some(after) = after {
        match (&q.column, &after.column) {
            (None, _) => {
                conditions.push(format!("k {op} ?"));
                args.push(SqlValue::Blob(after.key.clone()));
            }
            (Some(_), Some(c)) => {
                conditions.push(format!("({column} {op} ? OR ({column} = ? AND k {op} ?))"));
                args.extend([SqlValue::Blob(c.clone()), SqlValue::Blob(c.clone()), SqlValue::Blob(after.key.clone())]);
            }
            (Some(_), None) => return Err(Error::validation("invalid_cursor", "游标不合法")),
        }
    }
    let dir = if q.reverse { "DESC" } else { "ASC" };
    let order = match q.column {
        None => format!("k {dir}"),
        Some(_) => format!("{column} {dir}, k {dir}"),
    };
    let mut sql = format!("SELECT {select} FROM \"{}\"", q.store);
    if !conditions.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(&conditions.join(" AND "));
    }
    sql.push_str(" ORDER BY ");
    sql.push_str(&order);
    if let Some(limit) = limit {
        sql.push_str(" LIMIT ?");
        args.push(SqlValue::Integer(limit.min(i64::MAX as u64) as i64));
    }
    Ok((sql, args))
}

fn raw(text: String) -> Result<Box<RawValue>> {
    RawValue::from_string(text).map_err(|e| Error::storage("invalid_record", format!("记录不是合法的 JSON：{e}")))
}

pub fn get(conn: &Connection, store: &str, key: &[u8]) -> Result<Option<Box<RawValue>>> {
    let mut stmt = conn.prepare_cached(&format!("SELECT v FROM \"{store}\" WHERE k = ?"))?;
    let mut rows = stmt.query([key])?;
    match rows.next()? {
        Some(row) => Ok(Some(raw(row.get(0)?)?)),
        None => Ok(None),
    }
}

pub fn get_all(conn: &Connection, q: &Query) -> Result<Vec<Box<RawValue>>> {
    if q.limit == Some(0) {
        return Ok(Vec::new());
    }
    let (sql, args) = select_sql(q, "v", None, q.limit)?;
    let mut stmt = conn.prepare_cached(&sql)?;
    let texts = stmt.query_map(params_from_iter(args), |r| r.get::<_, String>(0))?.collect::<rusqlite::Result<Vec<_>>>()?;
    texts.into_iter().map(raw).collect()
}

pub fn count(conn: &Connection, q: &Query) -> Result<u64> {
    let column = q.column.as_deref().unwrap_or("k");
    let mut conditions = Vec::new();
    let mut args = Vec::new();
    if q.column.is_some() {
        conditions.push(format!("{column} IS NOT NULL"));
    }
    range_where(column, &q.range, &mut conditions, &mut args);
    let mut sql = format!("SELECT COUNT(*) FROM \"{}\"", q.store);
    if !conditions.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(&conditions.join(" AND "));
    }
    let n: i64 = conn.prepare_cached(&sql)?.query_row(params_from_iter(args), |r| r.get(0))?;
    Ok(n.max(0) as u64)
}

/// 一批遍历：最多 min(200, limit) 条；多取一条判断后面还有没有，没有时游标为 null，JavaScript 不必再取一次空批。
pub fn iterate(conn: &Connection, q: &Query, after: Option<&Cursor>) -> Result<ReadResult> {
    let page = q.limit.unwrap_or(PAGE_SIZE).min(PAGE_SIZE);
    if page == 0 {
        return Ok(ReadResult::Page { items: Vec::new(), cursor: None });
    }
    let select = match &q.column {
        None => "v, k".to_string(),
        Some(column) => format!("v, k, {column} AS c"),
    };
    let (sql, args) = select_sql(q, &select, after, Some(page + 1))?;
    let mut stmt = conn.prepare_cached(&sql)?;
    let indexed = q.column.is_some();
    let mut rows = stmt
        .query_map(params_from_iter(args), |r| {
            let column: Option<Vec<u8>> = if indexed { r.get(2)? } else { None };
            Ok((r.get::<_, String>(0)?, r.get::<_, Vec<u8>>(1)?, column))
        })?
        .collect::<rusqlite::Result<Vec<_>>>()?;
    let more = rows.len() as u64 > page;
    rows.truncate(page as usize);
    let cursor = match (more, rows.last()) {
        (true, Some((_, key, column))) => Some(Cursor::encode(column.as_deref(), key)),
        _ => None,
    };
    let items = rows.into_iter().map(|(v, _, _)| raw(v)).collect::<Result<Vec<_>>>()?;
    Ok(ReadResult::Page { items, cursor })
}