From 3dfeed67f5c34340e94eaa3671872002503b9270 Mon Sep 17 00:00:00 2001 From: Mike Solar Date: Sun, 16 Aug 2026 01:38:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(storage):=20database=20backend=20D1=20?= =?UTF-8?q?=E2=80=94=20SQLite,=20node-granular=20journal,=20snapshots,=20p?= =?UTF-8?q?ersistent=20undo?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - sea-orm schema: projects/settings/snapshots/journal (journal rows are per-node before/after images produced by diffing the serializer output, never full-project copies) - replay = latest snapshot + journal; load_at() rewinds to any command seq (cross-session undo history); snapshot interval and journal retention configurable; snapshots pruned to the newest 3 - project manager API: list/delete/duplicate/rename, export/import (.ove/.otio/.fcpxml), stats derivation; PG surfaces as E_NO_BACKEND until D3 - docs(riir): M13 finalized (aggregate-granular persistence) --- Cargo.lock | 2 + crates/oakstorage/Cargo.toml | 5 + crates/oakstorage/src/backends/database.rs | 81 - .../src/backends/database/entities/journal.rs | 51 + .../src/backends/database/entities/mod.rs | 29 + .../src/backends/database/entities/project.rs | 47 + .../backends/database/entities/settings.rs | 41 + .../backends/database/entities/snapshot.rs | 44 + .../src/backends/database/migration.rs | 91 + .../oakstorage/src/backends/database/mod.rs | 1509 +++++++++++++++ crates/oakstorage/src/backends/mod.rs | 7 +- crates/oakstorage/tests/database_test.rs | 1633 +++++++++++++++++ 12 files changed, 3456 insertions(+), 84 deletions(-) delete mode 100644 crates/oakstorage/src/backends/database.rs create mode 100644 crates/oakstorage/src/backends/database/entities/journal.rs create mode 100644 crates/oakstorage/src/backends/database/entities/mod.rs create mode 100644 crates/oakstorage/src/backends/database/entities/project.rs create mode 100644 crates/oakstorage/src/backends/database/entities/settings.rs create mode 100644 crates/oakstorage/src/backends/database/entities/snapshot.rs create mode 100644 crates/oakstorage/src/backends/database/migration.rs create mode 100644 crates/oakstorage/src/backends/database/mod.rs create mode 100644 crates/oakstorage/tests/database_test.rs diff --git a/Cargo.lock b/Cargo.lock index 049ab21b4..f9f0e4432 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4877,6 +4877,8 @@ dependencies = [ name = "oakstorage" version = "0.1.0" dependencies = [ + "chrono", + "oakcommon", "oakcore-rs", "oaknode", "oakotio", diff --git a/crates/oakstorage/Cargo.toml b/crates/oakstorage/Cargo.toml index 7b6ae1b7a..1156b4cba 100644 --- a/crates/oakstorage/Cargo.toml +++ b/crates/oakstorage/Cargo.toml @@ -10,8 +10,13 @@ crate-type = ["staticlib", "rlib"] [dependencies] oakcore-rs = { path = "../oakcore" } +oakcommon = { path = "../oakcommon" } serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } +# Timestamp columns in the database schema (plan M13 §1: TIMESTAMP NOT +# NULL). sea-orm maps chrono NaiveDateTime via its default with-chrono +# feature; the direct dependency is needed for the entity field types. +chrono = "0.4" # Error derive (Display + std::error::Error) for the crate error enum # (src/error.rs). Same major version the other modules use (oakotio, ...). thiserror = "2" diff --git a/crates/oakstorage/src/backends/database.rs b/crates/oakstorage/src/backends/database.rs deleted file mode 100644 index a9f5f2414..000000000 --- a/crates/oakstorage/src/backends/database.rs +++ /dev/null @@ -1,81 +0,0 @@ -// Oak Video Editor - Non-Linear Video Editor -// Copyright (C) 2026 Oak Team -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. -// -// You should have received a copy of the GNU General Public License -// along with this program. If not, see . - -//! The database backend (PostgreSQL + SQLite via SeaORM). -//! -//! URI forms: -//! - `oakdb+sqlite:///absolute/path.db` — local file database -//! - `oakdb+pg://user:pass@host:5432/dbname` — PostgreSQL -//! -//! The async SeaORM API is driven by a private current-thread tokio -//! runtime so the public surface stays synchronous (M10: no callbacks, -//! sync commands only). The graph payload is the same serialized form -//! the ove-xml backend produces — one serialization truth. - -/// The database backend (schemes `oakdb+sqlite` / `oakdb+pg`). -pub struct DatabaseBackend { - // Private current-thread runtime + SeaORM connection cache. -} - -impl DatabaseBackend { - /// Construct (no connection is opened until first use). - pub fn new() -> Self { - todo!() - } - - /// Parse the URI into a SeaORM connection URL + project key - /// (query param `?project=` or row id; default: singleton "default" - /// project row). - /// - /// Dead until the database proxy lands (the whole backend is a - /// declared stub); kept for the schema contract. - #[allow(dead_code)] - pub(crate) fn parse_target( - _uri: &crate::uri::StorageUri, - ) -> crate::error::Result<(String, String)> { - todo!() - } -} - -impl crate::backend::StorageBackend for DatabaseBackend { - fn name(&self) -> &'static str { - todo!() - } - - fn uri_scheme(&self) -> &'static str { - todo!() - } - - fn can_handle(&self, uri: &crate::uri::StorageUri) -> bool { - uri.scheme.starts_with("oakdb") - } - - fn load( - &self, - _uri: &crate::uri::StorageUri, - ) -> crate::error::Result { - todo!() - } - - fn save( - &self, - _project: crate::handle::CHandle, - _uri: &crate::uri::StorageUri, - _options: u32, - ) -> crate::error::Result<()> { - todo!() - } -} diff --git a/crates/oakstorage/src/backends/database/entities/journal.rs b/crates/oakstorage/src/backends/database/entities/journal.rs new file mode 100644 index 000000000..f545105a8 --- /dev/null +++ b/crates/oakstorage/src/backends/database/entities/journal.rs @@ -0,0 +1,51 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! The `journal` table: the command log (plan §1). One row per affected +//! node per command — the node's before image (`old_xml`) and after +//! image (`new_xml`), whole-node XML produced by the diff. `node_identity` +//! 0 is the settings pseudo-node; real node identities are stored +//! offset by +1 so the 0 sentinel never collides with a graph slot. + +use sea_orm::entity::prelude::*; + +/// The `journal` entity. +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "journal")] +pub struct Model { + /// Owning project (ON DELETE CASCADE). + pub project_id: i64, + /// Monotonic per-project command sequence. + #[sea_orm(primary_key)] + pub seq: i64, + /// The affected node's identity (+1; 0 = settings pseudo-node). + #[sea_orm(primary_key)] + pub node_identity: i64, + /// Command kind: 'redo' | 'undo' | 'jump' | 'group' | 'import'. + pub kind: String, + /// Before image (whole node XML; NULL for newly added nodes). + pub old_xml: Option, + /// After image (whole node XML; NULL for deleted nodes). + pub new_xml: Option, + /// Journal write time. + pub at: DateTime, +} + +/// `journal` has no relations yet. +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/oakstorage/src/backends/database/entities/mod.rs b/crates/oakstorage/src/backends/database/entities/mod.rs new file mode 100644 index 000000000..7dee44324 --- /dev/null +++ b/crates/oakstorage/src/backends/database/entities/mod.rs @@ -0,0 +1,29 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! SeaORM entity models for the four-table schema (plan M13 §1). +//! +//! One module per entity: the `DeriveEntityModel` macro generates +//! `Entity`/`Column`/`Model`/`ActiveModel` in the defining module, so +//! each table gets its own scope. The same definitions serve SQLite +//! today and PostgreSQL (D3) — the column types (`i64`, `String`, +//! `Option`, `NaiveDateTime`) map to BIGINT/TEXT/TIMESTAMP on +//! both. + +pub mod journal; +pub mod project; +pub mod settings; +pub mod snapshot; diff --git a/crates/oakstorage/src/backends/database/entities/project.rs b/crates/oakstorage/src/backends/database/entities/project.rs new file mode 100644 index 000000000..b97da8f88 --- /dev/null +++ b/crates/oakstorage/src/backends/database/entities/project.rs @@ -0,0 +1,47 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! The `projects` table: one row per project library entry (plan §1). + +use sea_orm::entity::prelude::*; + +/// The `projects` entity. +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "projects")] +pub struct Model { + /// Surrogate row id (BIGSERIAL/INTEGER PK). + #[sea_orm(primary_key)] + pub id: i64, + /// Project uuid (same value the serializer persists; unique). + #[sea_orm(unique)] + pub uuid: String, + /// Display name (the project manager lists this). + pub name: String, + /// Payload format version (the serializer's CURRENT_VERSION). + pub schema_ver: i32, + /// Row creation timestamp. + pub created_at: DateTime, + /// Last-write timestamp (manager sort key). + pub modified_at: DateTime, + /// Current head command seq (the newest journal seq). + pub command_seq: i64, +} + +/// `projects` has no relations yet (queries join explicitly). +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/oakstorage/src/backends/database/entities/settings.rs b/crates/oakstorage/src/backends/database/entities/settings.rs new file mode 100644 index 000000000..a57ca907d --- /dev/null +++ b/crates/oakstorage/src/backends/database/entities/settings.rs @@ -0,0 +1,41 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! The `settings` table: the project's KV settings (plan §1). The +//! current values are mirrored here on every save; the history lives in +//! the journal (node_identity = 0 pseudo-node). + +use sea_orm::entity::prelude::*; + +/// The `settings` entity. +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "settings")] +pub struct Model { + /// Owning project (ON DELETE CASCADE). + #[sea_orm(primary_key)] + pub project_id: i64, + /// Setting key. + #[sea_orm(primary_key)] + pub key: String, + /// Setting value. + pub value: String, +} + +/// `settings` has no relations yet. +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/oakstorage/src/backends/database/entities/snapshot.rs b/crates/oakstorage/src/backends/database/entities/snapshot.rs new file mode 100644 index 000000000..b161dc4b8 --- /dev/null +++ b/crates/oakstorage/src/backends/database/entities/snapshot.rs @@ -0,0 +1,44 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! The `snapshots` table: periodic full payloads that only accelerate +//! loading (plan §0/§1). The newest snapshot at or before a target seq +//! is the replay base; older ones are pruned to keep the newest +//! [`crate::backends::database::SNAPSHOT_KEEP`]. + +use sea_orm::entity::prelude::*; + +/// The `snapshots` entity. +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "snapshots")] +pub struct Model { + /// Owning project (ON DELETE CASCADE). + #[sea_orm(primary_key)] + pub project_id: i64, + /// The journal command seq this snapshot captures. + #[sea_orm(primary_key)] + pub command_seq: i64, + /// Full-feature project XML (identical to a `.ove` payload). + pub payload: String, + /// Snapshot write time. + pub written_at: DateTime, +} + +/// `snapshots` has no relations yet. +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/crates/oakstorage/src/backends/database/migration.rs b/crates/oakstorage/src/backends/database/migration.rs new file mode 100644 index 000000000..7d27bd09b --- /dev/null +++ b/crates/oakstorage/src/backends/database/migration.rs @@ -0,0 +1,91 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! Schema bootstrap (plan M13 §1). +//! +//! Chosen over the sea-orm-migration crate on purpose: the schema is +//! four tables, static for the whole D1 surface, and `CREATE TABLE IF +//! NOT EXISTS` is idempotent — no versioned migration machinery, no +//! extra crate in the tree. The DDL runs once at first connection open. + +use sea_orm::{ConnectionTrait, DatabaseBackend, Statement}; + +/// SQLite DDL for the four tables (plan §1, SQLite dialect: +/// `INTEGER PRIMARY KEY AUTOINCREMENT` + `TIMESTAMP`). +const SQLITE_DDL: &[&str] = &[ + "CREATE TABLE IF NOT EXISTS projects ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + uuid TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + schema_ver INTEGER NOT NULL, + created_at TIMESTAMP NOT NULL, + modified_at TIMESTAMP NOT NULL, + command_seq BIGINT NOT NULL DEFAULT 0 + )", + "CREATE TABLE IF NOT EXISTS settings ( + project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + key TEXT NOT NULL, + value TEXT NOT NULL, + PRIMARY KEY (project_id, key) + )", + "CREATE TABLE IF NOT EXISTS snapshots ( + project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + command_seq BIGINT NOT NULL, + payload TEXT NOT NULL, + written_at TIMESTAMP NOT NULL, + PRIMARY KEY (project_id, command_seq) + )", + "CREATE TABLE IF NOT EXISTS journal ( + project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + seq BIGINT NOT NULL, + node_identity BIGINT NOT NULL, + kind TEXT NOT NULL, + old_xml TEXT, + new_xml TEXT, + at TIMESTAMP NOT NULL, + PRIMARY KEY (project_id, seq, node_identity) + )", +]; + +/// Apply the schema (idempotent; safe to run on every connection open). +/// +/// PostgreSQL (D3) gets its own `BIGSERIAL` DDL — same shape, currently +/// unreachable because the backend rejects `oakdb+pg://` before any +/// connection is opened. +pub async fn migrate(db: &impl ConnectionTrait) -> crate::error::Result<()> { + let backend = db.get_database_backend(); + let statements: &[&str] = match backend { + DatabaseBackend::Sqlite => SQLITE_DDL, + // D3: `BIGSERIAL PRIMARY KEY` for projects.id, `BIGINT` for the + // FKs — the rest of the column set is identical. + DatabaseBackend::Postgres => { + return Err(crate::error::Error::Failed( + "oakdb+pg is a D3 milestone; the PostgreSQL schema is not wired yet".to_string(), + )); + } + other => { + return Err(crate::error::Error::Failed(format!( + "unsupported database backend {other:?}" + ))); + } + }; + for sql in statements { + db.execute_raw(Statement::from_string(backend, sql.to_string())) + .await + .map_err(|e| crate::error::Error::Io(format!("schema migration: {e}")))?; + } + Ok(()) +} diff --git a/crates/oakstorage/src/backends/database/mod.rs b/crates/oakstorage/src/backends/database/mod.rs new file mode 100644 index 000000000..ca023dc62 --- /dev/null +++ b/crates/oakstorage/src/backends/database/mod.rs @@ -0,0 +1,1509 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! The database backend (plan M13 §1/§2), SQLite first. +//! +//! URI forms: +//! - `oakdb+sqlite:///absolute/path.db[?project=]` — local file +//! database; the optional `project` query selects the library row to +//! load (default: the most recently modified project). +//! - `oakdb+pg://user:pass@host:5432/db[?project=]` — PostgreSQL +//! (D3). The target is parsed into [`DbTarget::Pg`] today, but every +//! operation rejects it with [`crate::error::Error::NoBackend`] until +//! the D3 client lands. +//! +//! The async sea-orm API is driven by a private current-thread tokio +//! runtime so the public surface stays synchronous (plan §6: "后端内嵌 +//! 私有 current_thread runtime"). +//! +//! ## Storage model (plan §0 — the aggregation-granularity design) +//! +//! A project's state is its node graph plus its settings — nothing else. +//! Every command is persisted as a *diff*: the project is re-serialized +//! in memory (the same per-node XML the `.ove` writer emits) and +//! compared node-by-node against the previous state, so each affected +//! node lands in the journal as one row with its before image +//! (`old_xml`) and after image (`new_xml`). The journal is therefore the +//! persistent undo history: replaying `new_xml` rows forward reconstructs +//! any later state, replaying `old_xml` backwards undoes to any point. +//! Node addressing uses the packed `NodeId::identity`, stored offset by +//! +1 so that journal identity 0 can be reserved for the settings +//! pseudo-node (a real graph slot can have identity 0 — the root folder +//! is slot 0). +//! +//! Snapshots are periodic full payloads that only accelerate loading: +//! the newest snapshot at or before a target seq is the replay base, and +//! the journal rows after it are applied on top. A snapshot is written +//! when the project is dirty and `Storage/SnapshotIntervalSec` (default +//! 600) seconds have passed since the last one; pruning keeps the newest +//! [`SNAPSHOT_KEEP`] copies. Journal retention follows +//! `Storage/JournalRetentionDays` (default 0 = keep everything); rows +//! older than the window are dropped only when the newest snapshot +//! already covers them, so the head state always stays reconstructible. + +pub mod entities; +pub mod migration; + +use std::collections::HashMap; +use std::future::Future; +use std::path::Path; +use std::sync::{Mutex, OnceLock}; +use std::time::Duration; + +use sea_orm::entity::prelude::*; +use sea_orm::sea_query::Expr; +use sea_orm::{QueryOrder, QuerySelect, Set, TransactionTrait}; + +use entities::{journal, project, settings, snapshot}; +use oaknode::id::NodeId; +use oaknode::project::Project; + +use crate::backend::{LoadResult, StorageBackend}; +use crate::error::{Error, Result}; +use crate::handle::CHandle; +use crate::uri::StorageUri; + +/// Journal kind for redo commands (the D1 save path). +pub const KIND_REDO: &str = "redo"; +/// Journal kind for undo commands (D2 write-through hook). +pub const KIND_UNDO: &str = "undo"; +/// Journal kind for playhead/time jump commands (D2). +pub const KIND_JUMP: &str = "jump"; +/// Journal kind for grouped commands (D2 `group_end`). +pub const KIND_GROUP: &str = "group"; +/// Journal kind of a first-entry import command (plan §2). +pub const KIND_IMPORT: &str = "import"; +/// Journal identity of the settings pseudo-node (plan §1). +pub const SETTINGS_NODE: i64 = 0; +/// Snapshots kept after pruning (plan §0: the newest 3). +pub const SNAPSHOT_KEEP: u64 = 3; +/// The empty settings element (replay base before any settings row). +const EMPTY_SETTINGS: &str = ""; + +/// A parsed `oakdb+…` target: the database location plus the optional +/// `?project=` selector. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum DbTarget { + /// SQLite file database at an absolute path. + Sqlite { + /// Absolute database file path. + path: String, + /// `?project=` uuid; `None` = most recently modified row. + project: Option, + }, + /// PostgreSQL (parsed for D3; rejected until then). + Pg { + /// Connection string (`user:pass@host:port/db`). + conn: String, + /// `?project=` uuid; `None` = most recently modified row. + project: Option, + }, +} + +impl DbTarget { + /// True for the PostgreSQL variant (all ops return E_NO_BACKEND + /// until D3). + fn is_pg(&self) -> bool { + matches!(self, DbTarget::Pg { .. }) + } + + /// The SQLite path (`None` for PG). + fn sqlite_path(&self) -> Option<&str> { + match self { + DbTarget::Sqlite { path, .. } => Some(path), + DbTarget::Pg { .. } => None, + } + } + + /// The `?project=` selector. + fn project(&self) -> Option<&str> { + match self { + DbTarget::Sqlite { project, .. } | DbTarget::Pg { project, .. } => { + project.as_deref() + } + } + } +} + +/// Parse a storage URI into a [`DbTarget`]. +/// +/// `oakdb+sqlite://` requires an absolute path (E_INVALID otherwise); +/// the body may carry `?project=` (no percent-decoding — uuid +/// values contain only hex/braces/hyphens). Unknown `oakdb` variants +/// are E_NO_BACKEND. +pub(crate) fn parse_target(uri: &StorageUri) -> Result { + match uri.scheme.as_str() { + "oakdb+sqlite" => { + let (path, query) = split_query(&uri.body); + if !Path::new(path).is_absolute() { + return Err(Error::Invalid); + } + Ok(DbTarget::Sqlite { + path: path.to_string(), + project: query_param(query, "project"), + }) + } + "oakdb+pg" => { + let (conn, query) = split_query(&uri.body); + Ok(DbTarget::Pg { + conn: conn.to_string(), + project: query_param(query, "project"), + }) + } + _ => Err(Error::NoBackend), + } +} + +/// Split `body` at the first `?` (the query part, if any). +fn split_query(body: &str) -> (&str, Option<&str>) { + match body.split_once('?') { + Some((base, query)) => (base, Some(query)), + None => (body, None), + } +} + +/// Look up `key` in a `k=v&…` query string. +fn query_param(query: Option<&str>, key: &str) -> Option { + let query = query?; + for pair in query.split('&') { + if let Some((k, v)) = pair.split_once('=') { + if k == key { + return Some(v.to_string()); + } + } + } + None +} + +/// Derived per-project manager stats (plan §4 — derived from the node +/// graph when a project is opened, never stored). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ProjectStats { + /// Timeline duration in milliseconds (longest sequence). + pub duration_ms: i64, + /// Total tracks across all sequences. + pub track_count: i32, + /// Total clip blocks across all tracks. + pub clip_count: i32, + /// Total footage nodes in the graph. + pub footage_count: i32, +} + +/// Derive the manager stats from a live project (plan §4). +/// +/// Traversal: sequences contribute their track lists' tracks and blocks; +/// the duration is the longest sequence's end point. Footage nodes are +/// counted directly in the graph. A clip that appears in more than one +/// track is counted once per membership. +pub fn derive_stats(p: &Project) -> ProjectStats { + use oaknode::block::{ClipBlockBehavior, GapBlockBehavior}; + use oaknode::sequence::SequenceBehavior; + use oaknode::track::{TrackBehavior, TrackListBehavior}; + + let mut stats = ProjectStats::default(); + for id in p.graph.node_ids() { + let entry = match p.graph.get(id) { + Some(e) => e, + None => continue, + }; + if let Some(seq) = entry + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + { + let mut seq_len = oakcore_rs::Rational::new(0, 1); + for tl_id in &seq.track_lists { + let tl = match p.graph.get(*tl_id).and_then(|e| { + e.behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + }) { + Some(t) => t, + None => continue, + }; + stats.track_count += tl.tracks.len() as i32; + for t_id in &tl.tracks { + let t = match p.graph.get(*t_id).and_then(|e| { + e.behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + }) { + Some(t) => t, + None => continue, + }; + for b in &t.blocks { + let entry = p.graph.get(*b); + if let Some(c) = entry.and_then(|e| { + e.behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + }) { + stats.clip_count += 1; + if c.core.out() > seq_len { + seq_len = c.core.out(); + } + } else if let Some(g) = entry.and_then(|e| { + e.behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + }) { + if g.core.out() > seq_len { + seq_len = g.core.out(); + } + } + } + } + } + let ms = (seq_len.to_f64() * 1000.0).round() as i64; + if ms > stats.duration_ms { + stats.duration_ms = ms; + } + } else if entry.behavior.type_id() == "org.olivevideoeditor.Olive.footage" { + stats.footage_count += 1; + } + } + stats +} + +/// One library entry as returned by [`DatabaseBackend::list_projects`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProjectInfo { + /// Surrogate row id. + pub id: i64, + /// Project uuid (serializer-identical). + pub uuid: String, + /// Display name. + pub name: String, + /// Payload format version (the serializer's CURRENT_VERSION). + pub schema_ver: i32, + /// Row creation time. + pub created_at: DateTime, + /// Last-write time (manager sort key). + pub modified_at: DateTime, + /// Head command seq. + pub command_seq: i64, +} + +/// The database backend (schemes `oakdb+sqlite` / `oakdb+pg`). +/// +/// Connections and the private runtime are created lazily on the first +/// operation; the runtime is a single current-thread tokio runtime that +/// every synchronous call drives with `block_on`. An `op_lock` mutex +/// serializes calls into one backend (single-writer assumption, plan +/// §6) and the SQLite pool itself is capped at one connection, so a +/// transaction owns the only connection for its whole lifetime. +pub struct DatabaseBackend { + /// Private current-thread tokio runtime (created on first use). + runtime: OnceLock, + /// Serializes all database operations of this backend. + op_lock: Mutex<()>, + /// Open connections by database path. + connections: Mutex>, +} + +impl DatabaseBackend { + /// Construct (no connection is opened until first use). + pub fn new() -> Self { + DatabaseBackend { + runtime: OnceLock::new(), + op_lock: Mutex::new(()), + connections: Mutex::new(HashMap::new()), + } + } + + /// Build the private current-thread runtime. + fn build_runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("tokio current_thread runtime") + } + + /// Run `f` against the database at `key` (an absolute SQLite path). + /// The op lock is held for the whole call; the connection is opened + /// and migrated on first use, then cached. + fn run(&self, key: &str, f: F) -> Result + where + F: FnOnce(DatabaseConnection) -> Fut, + Fut: Future>, + { + let _guard = self.op_lock.lock().unwrap_or_else(|e| e.into_inner()); + let runtime = self.runtime.get_or_init(Self::build_runtime); + runtime.block_on(async { + let conn = self.connect_cached(key).await?; + f(conn).await + }) + } + + /// Open (and migrate) the SQLite database at `path`, caching the + /// connection. + /// + /// Two writers racing on a fresh file surface SQLite's `database is + /// locked` (SQLITE_BUSY) — most often when a fresh connection's WAL + /// mode switch collides with the other's, which `busy_timeout` cannot + /// wait out — so connect+ migrate are retried a few times (plan §6: + /// single-writer is the contract; a clean error beats a silent + /// failure). + async fn connect_cached(&self, path: &str) -> Result { + if let Some(conn) = self + .connections + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(path) + { + return Ok(conn.clone()); + } + let mut attempt = 0; + let conn = loop { + let result = async { + let conn = connect_sqlite(path).await?; + migration::migrate(&conn).await?; + Ok::<_, Error>(conn) + } + .await; + match result { + Ok(conn) => break conn, + Err(e) if retryable(&e) && attempt < 3 => { + attempt += 1; + tokio::time::sleep(Duration::from_millis(20 * attempt)).await; + } + Err(e) => return Err(e), + } + }; + self.connections + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(path.to_string(), conn.clone()); + Ok(conn) + } + + /// List library rows, most recently modified first (project manager + /// data source; per-project stats come from + /// [`DatabaseBackend::project_stats`], which derives them from the + /// node graph). + pub fn list_projects(&self, uri: &StorageUri) -> Result> { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + self.run(path, |conn| async move { + let models = project::Entity::find() + .order_by_desc(project::Column::ModifiedAt) + .all(&conn) + .await + .map_err(db_err)?; + Ok(models + .into_iter() + .map(|m| ProjectInfo { + id: m.id, + uuid: m.uuid, + name: m.name, + schema_ver: m.schema_ver, + created_at: m.created_at, + modified_at: m.modified_at, + command_seq: m.command_seq, + }) + .collect()) + }) + } + + /// Delete a library row by uuid (cascades settings/snapshots/journal; + /// the project's data is gone — the manager confirms before calling). + pub fn delete_project(&self, uri: &StorageUri, uuid: &str) -> Result<()> { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let uuid = uuid.to_string(); + self.run(path, move |conn| async move { + let tx = conn.begin().await.map_err(db_err)?; + let res = project::Entity::delete_many() + .filter(project::Column::Uuid.eq(&uuid)) + .exec(&tx) + .await + .map_err(db_err)?; + tx.commit().await.map_err(db_err)?; + if res.rows_affected == 0 { + return Err(Error::NotFound); + } + Ok(()) + }) + } + + /// Copy a library row — settings, snapshots and the full journal + /// history included — under a fresh uuid. `new_name` defaults to + /// ` (copy)`. + pub fn duplicate_project( + &self, + uri: &StorageUri, + uuid: &str, + new_name: Option<&str>, + ) -> Result { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let uuid = uuid.to_string(); + let new_name = new_name.map(str::to_string); + self.run(path, move |conn| async move { + let tx = conn.begin().await.map_err(db_err)?; + let src = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&tx) + .await + .map_err(db_err)? + .ok_or(Error::NotFound)?; + let new_uuid = new_uuid(); + let name = new_name.unwrap_or_else(|| format!("{} (copy)", src.name)); + let now = chrono::Utc::now().naive_utc(); + let new_id = project::Entity::insert(project::ActiveModel { + uuid: Set(new_uuid.clone()), + name: Set(name.clone()), + schema_ver: Set(src.schema_ver), + created_at: Set(now), + modified_at: Set(now), + command_seq: Set(src.command_seq), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)? + .last_insert_id; + + // Settings mirror. + for s in settings::Entity::find() + .filter(settings::Column::ProjectId.eq(src.id)) + .all(&tx) + .await + .map_err(db_err)? + { + settings::Entity::insert(settings::ActiveModel { + project_id: Set(new_id), + key: Set(s.key), + value: Set(s.value), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)?; + } + // Snapshots: the payloads keep their node fragments; the + // assembled uuid always comes from the project row at load + // time, so the copy replays under its own uuid. + for s in snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(src.id)) + .all(&tx) + .await + .map_err(db_err)? + { + snapshot::Entity::insert(snapshot::ActiveModel { + project_id: Set(new_id), + command_seq: Set(s.command_seq), + payload: Set(s.payload), + written_at: Set(s.written_at), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)?; + } + // Journal history (the PK is per-project, so the same + // seq/node_identity pairs are legal on the new row). + for r in journal::Entity::find() + .filter(journal::Column::ProjectId.eq(src.id)) + .all(&tx) + .await + .map_err(db_err)? + { + journal::Entity::insert(journal::ActiveModel { + project_id: Set(new_id), + seq: Set(r.seq), + node_identity: Set(r.node_identity), + kind: Set(r.kind), + old_xml: Set(r.old_xml), + new_xml: Set(r.new_xml), + at: Set(r.at), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)?; + } + tx.commit().await.map_err(db_err)?; + + Ok(ProjectInfo { + id: new_id, + uuid: new_uuid, + name, + schema_ver: src.schema_ver, + created_at: now, + modified_at: now, + command_seq: src.command_seq, + }) + }) + } + + /// Rename a library row by uuid (E_NOT_FOUND when absent). + /// + /// A rename is library metadata (the manager's list name); the + /// in-app project name (`settings["projectname"]`) is untouched. + pub fn rename_project(&self, uri: &StorageUri, uuid: &str, new_name: &str) -> Result<()> { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let uuid = uuid.to_string(); + let new_name = new_name.to_string(); + self.run(path, move |conn| async move { + let now = chrono::Utc::now().naive_utc(); + let tx = conn.begin().await.map_err(db_err)?; + let res = project::Entity::update_many() + .col_expr(project::Column::Name, Expr::value(new_name)) + .col_expr(project::Column::ModifiedAt, Expr::value(now)) + .filter(project::Column::Uuid.eq(&uuid)) + .exec(&tx) + .await + .map_err(db_err)?; + tx.commit().await.map_err(db_err)?; + if res.rows_affected == 0 { + return Err(Error::NotFound); + } + Ok(()) + }) + } + + /// Export a library project to a `.ove` file, using the ove-xml + /// backend's save semantics (same serializer output, file container; + /// assembled in memory, nothing is written to the library). The + /// target URI must be a `file://` URI. + pub fn export_to_file(&self, uri: &StorageUri, uuid: &str, file_uri: &StorageUri) -> Result<()> { + if file_uri.scheme != "file" { + return Err(Error::Invalid); + } + let handle = self.load_handle_by_uuid(uri, uuid)?; + let backend = crate::backends::ove_xml::OveXmlBackend::new(); + let result = backend.save(handle, file_uri, 0); + // The loaded handle is ours (refcount 1); release it regardless + // of the save outcome. + if let Some(release) = handle.release { + unsafe { release(handle.ctx) }; + } + result + } + + /// Import a `.ove`/`.otio`/`.fcpxml` file as a new library row + /// (plan §2 "导入"): the file backend parses it into a project, a + /// fresh uuid is assigned, and the first save writes the whole + /// project as one `kind='import'` command (seq 1). Returns the new + /// row's uuid. + pub fn import_from_file(&self, uri: &StorageUri, file_uri: &StorageUri) -> Result { + if file_uri.scheme != "file" { + return Err(Error::Invalid); + } + let backend = crate::registry::Registry::global().resolve(file_uri)?; + let result = backend.load(file_uri)?; + let handle = result.project; + if handle.is_null() { + return Err(Error::Format(format!( + "cannot import: backend reported info code {}", + result.version_info + ))); + } + let uuid = { + let arc = unsafe { crate::nodeutil::project_arc(&handle)? }; + let fresh = new_uuid(); + arc.lock().map_err(|_| Error::State)?.uuid = fresh.clone(); + fresh + }; + let outcome = self.save(handle, uri, 0).map(|()| uuid.clone()); + if let Some(release) = handle.release { + unsafe { release(handle.ctx) }; + } + outcome + } + + /// Force a snapshot of the project at its head seq (the D2 snapshot + /// thread and the exit flush call this; tests use it to exercise the + /// snapshot+replay path). No-op when a snapshot already exists at + /// the head seq. + pub fn snapshot(&self, uri: &StorageUri, uuid: &str) -> Result<()> { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let uuid = uuid.to_string(); + self.run(path, move |conn| async move { + let model = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .map_err(db_err)? + .ok_or(Error::NotFound)?; + let now = chrono::Utc::now().naive_utc(); + let tx = conn.begin().await.map_err(db_err)?; + write_snapshot(&tx, model.id, &model.uuid, model.command_seq, now).await?; + prune_snapshots(&tx, model.id).await?; + tx.commit().await.map_err(db_err)?; + Ok(()) + }) + } + + /// Load the project state at an arbitrary command seq — the + /// persistent undo history (plan §0): a snapshot at or before `seq` + /// is replayed forward with the journal rows up to `seq`. `seq` 0 is + /// the empty project. E_INVALID when `seq` is out of range. + pub fn load_at(&self, uri: &StorageUri, uuid: &str, seq: i64) -> Result { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let uuid = uuid.to_string(); + self.run(path, move |conn| async move { + let model = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .map_err(db_err)? + .ok_or(Error::NotFound)?; + if seq < 0 || seq > model.command_seq { + return Err(Error::Invalid); + } + let xml = assemble_at(&conn, model.id, &model.uuid, seq).await?; + Ok(crate::nodeutil::make_project_owned( + crate::nodeutil::serializer_load(&xml)?, + )) + }) + } + + /// The manager stats of a library project (plan §4): the head state + /// is replayed and the stats derived from the node graph. + pub fn project_stats(&self, uri: &StorageUri, uuid: &str) -> Result { + let handle = self.load_handle_by_uuid(uri, uuid)?; + let stats = (|| -> Result { + let arc = unsafe { crate::nodeutil::project_arc(&handle)? }; + let guard = arc.lock().map_err(|_| Error::State)?; + Ok(derive_stats(&guard)) + })(); + if let Some(release) = handle.release { + unsafe { release(handle.ctx) }; + } + stats + } + + /// Load the project payload of the library row `uuid` as an owned + /// handle (the head state). + fn load_handle_by_uuid(&self, uri: &StorageUri, uuid: &str) -> Result { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let uuid = uuid.to_string(); + self.run(path, move |conn| async move { + let model = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .map_err(db_err)? + .ok_or(Error::NotFound)?; + let xml = assemble_at(&conn, model.id, &model.uuid, model.command_seq).await?; + Ok(crate::nodeutil::make_project_owned( + crate::nodeutil::serializer_load(&xml)?, + )) + }) + } +} + +impl Default for DatabaseBackend { + fn default() -> Self { + Self::new() + } +} + +impl StorageBackend for DatabaseBackend { + fn name(&self) -> &'static str { + "oakdb" + } + + fn uri_scheme(&self) -> &'static str { + "oakdb" + } + + fn can_handle(&self, uri: &StorageUri) -> bool { + // `oakdb://` (no sub-scheme) is deliberately not claimed — the + // legacy URI is unresolvable (E_NO_BACKEND), per the contract + // tests in `tests/storage_test.rs`. + uri.scheme == "oakdb+sqlite" || uri.scheme == "oakdb+pg" + } + + fn load(&self, uri: &StorageUri) -> Result { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + let project = target.project().map(str::to_string); + self.run(path, move |conn| async move { + let model = pick_project(&conn, project.as_deref()).await?; + let xml = assemble_at(&conn, model.id, &model.uuid, model.command_seq).await?; + let handle = + crate::nodeutil::make_project_owned(crate::nodeutil::serializer_load(&xml)?); + Ok(LoadResult::success(handle)) + }) + } + + fn save(&self, project: CHandle, uri: &StorageUri, _options: u32) -> Result<()> { + let target = parse_target(uri)?; + let path = sqlite_path_of(&target)?; + // Serialize under the project lock (the same per-node writer the + // `.ove` backend uses — one serialization truth). + let arc = unsafe { crate::nodeutil::project_arc(&project)? }; + let (uuid, name, nodes, settings_xml, settings_map) = { + let guard = arc.lock().map_err(|_| Error::State)?; + let uuid = guard.uuid.clone(); + let name = project_display_name(&guard); + let (nodes, settings_xml, settings_map) = serialize_project_state(&guard)?; + (uuid, name, nodes, settings_xml, settings_map) + }; + self.run(path, move |conn| async move { + save_tx(&conn, &uuid, &name, &nodes, &settings_xml, &settings_map).await + })?; + Ok(()) + } +} + +/// The SQLite path of a target, rejecting PG with E_NO_BACKEND (D3). +fn sqlite_path_of(target: &DbTarget) -> Result<&str> { + if target.is_pg() { + // D3: `oakdb+pg://` is parsed (see parse_target) but every + // operation refuses it until the PostgreSQL client is wired. + return Err(Error::NoBackend); + } + Ok(target.sqlite_path().expect("pg rejected above")) +} + +/// Display name for a project on first entry: the `projectname` setting +/// (what the editor stores), falling back to "Untitled Project". +fn project_display_name(p: &Project) -> String { + p.settings + .get("projectname") + .cloned() + .unwrap_or_else(|| "Untitled Project".to_string()) +} + +/// Open the SQLite database at `path` (WAL + busy timeout + FK +/// enforcement; a missing parent directory is an I/O error). +async fn connect_sqlite(path: &str) -> Result { + use sea_orm::sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions}; + let options = SqliteConnectOptions::new() + .filename(path) + .create_if_missing(true) + .journal_mode(SqliteJournalMode::Wal) + .busy_timeout(Duration::from_secs(5)) + .foreign_keys(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .map_err(|e| Error::Io(format!("cannot open sqlite database '{path}': {e}")))?; + Ok(DatabaseConnection::from(pool)) +} + +/// Whether an error is worth retrying (SQLite BUSY — plan §6 note in +/// [`DatabaseBackend::run`]). +fn retryable(e: &Error) -> bool { + if let Error::Io(msg) = e { + msg.contains("database is locked") || msg.contains("is busy") + } else { + false + } +} + +/// The project row to load: `?project=` uuid, or the most recently +/// modified row. E_NOT_FOUND when the library is empty / uuid unknown. +async fn pick_project(conn: &DatabaseConnection, uuid: Option<&str>) -> Result { + let model = match uuid { + Some(uuid) => project::Entity::find() + .filter(project::Column::Uuid.eq(uuid)) + .one(conn) + .await + .map_err(db_err)?, + None => project::Entity::find() + .order_by_desc(project::Column::ModifiedAt) + .one(conn) + .await + .map_err(db_err)?, + }; + model.ok_or(Error::NotFound) +} + +/// Map a sea-orm error to a crate error (I/O class; the context string +/// is log-only per the error module's contract). +fn db_err(e: sea_orm::DbErr) -> Error { + Error::Io(format!("database error: {e}")) +} + +/// A fresh uuid in the project's `{…}` text format (mirrors the +/// oaknode `project::generate_uuid` splitmix64 generator, which is +/// private to that crate; identical shape so the serializer round-trips +/// it unchanged). +fn new_uuid() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0); + let mut seed = nanos ^ 0x9E3779B97F4A7C15; + let mut next = move || { + seed = seed.wrapping_add(0x9E3779B97F4A7C15); + let mut z = seed; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB); + z ^ (z >> 31) + }; + let mut b = [0u8; 16]; + for chunk in b.chunks_mut(8) { + let r = next().to_le_bytes(); + chunk.copy_from_slice(&r); + } + b[6] = (b[6] & 0x0F) | 0x40; // version 4 + b[8] = (b[8] & 0x3F) | 0x80; // variant 1 + format!( + "{{{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}}}", + b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8], b[9], b[10], b[11], b[12], b[13], + b[14], b[15] + ) +} + +// --------------------------------------------------------------------------- +// Serialization / diff / replay (plan §0) +// --------------------------------------------------------------------------- + +/// Serialize the project's whole state: per-node XML fragments keyed by +/// real graph identity, the settings element, and the settings map. +/// Every fragment is self-contained `…` text the +/// serializer can load back, byte-identical to what a `.ove` write +/// embeds inside ``. +fn serialize_project_state( + p: &Project, +) -> Result<(HashMap, String, HashMap)> { + let mut nodes = HashMap::new(); + for id in p.graph.node_ids() { + nodes.insert(id.identity(), serialize_node_xml(p, id)?); + } + let settings_map = p.settings.clone(); + let settings = settings_xml_from_map(&settings_map); + Ok((nodes, settings, settings_map)) +} + +/// Serialize one node as a standalone `…` element (the +/// same writer `serializer::save` uses inside ``). +fn serialize_node_xml(p: &Project, id: NodeId) -> Result { + use oaknode::serializer::{XmlWrite, XmlWriterBridge}; + let entry = p.graph.get(id).ok_or(Error::NotFound)?; + let type_id = entry.behavior.type_id().to_string(); + let connections: Vec<(NodeId, String, i32)> = p + .graph + .output_connections_all() + .into_iter() + .filter(|(_, to, _, _)| *to == id) + .map(|(from, _, input, element)| (from, input, element)) + .collect(); + let mut w = XmlWriterBridge::new() + .ok_or_else(|| Error::Failed("oakcommon XML writer unavailable".to_string()))?; + w.start_element("node"); + oaknode::serializer::save_node(&mut w, &entry.core, &*entry.behavior, id, &type_id, &connections) + .map_err(|e| Error::Format(e.to_string()))?; + w.end_element(); + Ok(w.output()) +} + +/// Serialize a settings map as the `…` element +/// (sorted keys, exactly like `serializer::save`). +fn settings_xml_from_map(settings: &HashMap) -> String { + use oaknode::serializer::{XmlWrite, XmlWriterBridge}; + let mut w = XmlWriterBridge::new().expect("oakcommon XML writer"); + w.start_element("settings"); + let mut keys: Vec<&String> = settings.keys().collect(); + keys.sort(); + for k in keys { + w.text_element(k, settings.get(k).unwrap_or(&String::new())); + } + w.end_element(); + w.output() +} + +/// Assemble a full `` document from a node map and a settings +/// element. The fragments are already-escaped XML, so this is plain text +/// assembly; the uuid (the only foreign text) is escaped defensively. +fn assemble_xml(uuid: &str, nodes: &HashMap, settings: &str) -> String { + let mut out = String::with_capacity(512 + nodes.values().map(String::len).sum::()); + out.push_str(""); + out.push_str(&xml_escape(uuid)); + out.push_str(""); + let mut ids: Vec<&u64> = nodes.keys().collect(); + ids.sort(); + for id in ids { + out.push_str(nodes.get(id).expect("key present")); + } + out.push_str(""); + out.push_str(settings); + out.push_str(""); + out +} + +/// Escape the three XML-significant characters (uuid text is normally +/// `{hex}`, but imported files may carry anything). +fn xml_escape(s: &str) -> String { + s.replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} + +/// Diff two node maps into journal rows (plan §0). `prev`/`cur` are +/// keyed by real graph identity; the returned rows use journal +/// identities (real identity + 1, reserving 0 for settings). +fn diff_rows( + prev: &HashMap, + cur: &HashMap, +) -> Vec<(i64, Option, Option)> { + let mut rows = Vec::new(); + for (id, xml) in cur { + match prev.get(id) { + Some(old) if old != xml => { + rows.push(((*id as i64) + 1, Some(old.clone()), Some(xml.clone()))); + } + Some(_) => {} + None => rows.push(((*id as i64) + 1, None, Some(xml.clone()))), + } + } + for (id, old) in prev { + if !cur.contains_key(id) { + rows.push(((*id as i64) + 1, Some(old.clone()), None)); + } + } + rows.sort(); + rows +} + +/// Replay the state at `target_seq` into a node map + settings element +/// (plan §0): the newest snapshot at or before `target_seq` (nothing if +/// absent) plus the journal rows after it, forward-applying `new_xml`. +/// A node whose row has no `new_xml` was deleted; a settings row +/// (identity 0) replaces the settings element wholesale. +async fn replay_state( + conn: &C, + project_id: i64, + target_seq: i64, +) -> Result<(HashMap, String)> { + let snap = snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(project_id)) + .filter(snapshot::Column::CommandSeq.lte(target_seq)) + .order_by_desc(snapshot::Column::CommandSeq) + .one(conn) + .await + .map_err(db_err)?; + let (mut nodes, mut settings, base_seq) = match &snap { + Some(s) => ( + extract_nodes(&s.payload), + extract_settings(&s.payload), + s.command_seq, + ), + None => (HashMap::new(), EMPTY_SETTINGS.to_string(), 0), + }; + let rows = journal::Entity::find() + .filter(journal::Column::ProjectId.eq(project_id)) + .filter(journal::Column::Seq.gt(base_seq)) + .filter(journal::Column::Seq.lte(target_seq)) + .order_by_asc(journal::Column::Seq) + .order_by_asc(journal::Column::NodeIdentity) + .all(conn) + .await + .map_err(db_err)?; + for r in rows { + if r.node_identity == SETTINGS_NODE { + settings = r.new_xml.clone().unwrap_or_else(|| EMPTY_SETTINGS.to_string()); + } else if let Some(new) = r.new_xml { + nodes.insert((r.node_identity - 1) as u64, new); + } else { + nodes.remove(&((r.node_identity - 1) as u64)); + } + } + Ok((nodes, settings)) +} + +/// Assemble the full project XML for a state. +async fn assemble_at( + conn: &C, + project_id: i64, + uuid: &str, + target_seq: i64, +) -> Result { + let (nodes, settings) = replay_state(conn, project_id, target_seq).await?; + Ok(assemble_xml(uuid, &nodes, &settings)) +} + +/// Extract the per-node fragments of an assembled project document into +/// an identity-keyed map. The document is produced by +/// [`assemble_xml`]/[`serializer::save`], whose `` children are +/// self-contained `…` elements (node bodies never nest +/// `` elements, so the first `` closes each fragment). +fn extract_nodes(xml: &str) -> HashMap { + let mut map = HashMap::new(); + let mut rest = xml; + while let Some(rel) = rest.find("` (``) or `/` (``); skip + // lookalikes such as ``. + match tag.as_bytes().get(5).copied() { + Some(b' ' | b'>' | b'/') => {} + _ => { + rest = &tag[5..]; + continue; + } + } + let body = &tag[5..]; + match body.find("") { + Some(end) => { + let frag = &tag[..5 + end + 7]; + if let Some(identity) = fragment_identity(frag) { + map.insert(identity, frag.to_string()); + } + rest = &body[end + 7..]; + } + None => break, + } + } + map +} + +/// The `ptr="…"` identity of a node fragment (the first `ptr=` in the +/// start tag; node bodies never contain `ptr="`). +fn fragment_identity(frag: &str) -> Option { + let p = frag.find("ptr=\"")?; + let after = &frag[p + 5..]; + let end = after.find('"')?; + after[..end].parse().ok() +} + +/// Extract the settings element of an assembled project document +/// (fallback: the empty settings element). +fn extract_settings(xml: &str) -> String { + if let Some(s) = xml.find("") { + let after = &xml[s + 10..]; + if let Some(e) = after.find("") { + return xml[s..s + 10 + e + 11].to_string(); + } + } + EMPTY_SETTINGS.to_string() +} + +// --------------------------------------------------------------------------- +// Command write path +// --------------------------------------------------------------------------- + +/// Write one command (plan §2): first entry is a `kind='import'` +/// command carrying every node; later entries diff against the replayed +/// head and write only the affected nodes as `kind='redo'`. The +/// settings table is mirrored, snapshots follow +/// `Storage/SnapshotIntervalSec`, and the journal is pruned to the +/// retention window — all in one transaction. +/// +/// A concurrent writer's short WAL lock window can surface `database is +/// locked` (SQLITE_BUSY) despite `busy_timeout` (the lock is only +/// acquired mid-transaction); such attempts are retried a few times +/// (plan §6: single-writer is the contract; a clean error beats a +/// silent failure). +async fn save_tx( + conn: &DatabaseConnection, + uuid: &str, + name: &str, + nodes: &HashMap, + settings_xml: &str, + settings_map: &HashMap, +) -> Result { + let mut attempt = 0; + loop { + match save_tx_once(conn, uuid, name, nodes, settings_xml, settings_map).await { + Ok(seq) => return Ok(seq), + Err(e) if retryable(&e) && attempt < 3 => { + attempt += 1; + tokio::time::sleep(Duration::from_millis(20 * attempt)).await; + } + Err(e) => return Err(e), + } + } +} + +/// The one-attempt body of [`save_tx`]. +async fn save_tx_once( + conn: &DatabaseConnection, + uuid: &str, + name: &str, + nodes: &HashMap, + settings_xml: &str, + settings_map: &HashMap, +) -> Result { + let now = chrono::Utc::now().naive_utc(); + let schema_ver = oaknode::serializer::CURRENT_VERSION.0 as i32; + + let tx = conn.begin().await.map_err(db_err)?; + + let existing = project::Entity::find() + .filter(project::Column::Uuid.eq(uuid)) + .one(&tx) + .await + .map_err(db_err)?; + let (project_id, head_seq) = match existing { + Some(m) => (m.id, m.command_seq), + None => { + let res = project::Entity::insert(project::ActiveModel { + uuid: Set(uuid.to_string()), + name: Set(name.to_string()), + schema_ver: Set(schema_ver), + created_at: Set(now), + modified_at: Set(now), + command_seq: Set(0), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)?; + (res.last_insert_id, 0) + } + }; + + let seq: i64; + if head_seq == 0 { + // First entry (import): one command with every node plus the + // settings pseudo-node; all old images are NULL. + seq = 1; + let mut rows: Vec<(i64, Option, Option)> = Vec::new(); + for (id, xml) in nodes { + rows.push(((*id as i64) + 1, None, Some(xml.clone()))); + } + rows.push((SETTINGS_NODE, None, Some(settings_xml.to_string()))); + rows.sort(); + write_journal_rows(&tx, project_id, seq, KIND_IMPORT, &rows, now).await?; + } else { + // Diff against the replayed head. + let (prev_nodes, prev_settings) = replay_state(&tx, project_id, head_seq).await?; + let mut rows = diff_rows(&prev_nodes, nodes); + if prev_settings != *settings_xml { + rows.push(( + SETTINGS_NODE, + Some(prev_settings), + Some(settings_xml.to_string()), + )); + } + rows.sort(); + if rows.is_empty() { + // No-op save: touch modified_at, keep the head seq. + project::Entity::update(project::ActiveModel { + id: Set(project_id), + modified_at: Set(now), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)?; + tx.commit().await.map_err(db_err)?; + return Ok(head_seq); + } + seq = head_seq + 1; + write_journal_rows(&tx, project_id, seq, KIND_REDO, &rows, now).await?; + } + + project::Entity::update(project::ActiveModel { + id: Set(project_id), + name: Set(name.to_string()), + schema_ver: Set(schema_ver), + modified_at: Set(now), + command_seq: Set(seq), + ..Default::default() + }) + .exec(&tx) + .await + .map_err(db_err)?; + + replace_settings(&tx, project_id, settings_map).await?; + maybe_snapshot(&tx, project_id, uuid, seq, now).await?; + retention_prune(&tx, project_id, now).await?; + + tx.commit().await.map_err(db_err)?; + Ok(seq) +} + +/// Insert the journal rows of one command. +async fn write_journal_rows( + conn: &C, + project_id: i64, + seq: i64, + kind: &str, + rows: &[(i64, Option, Option)], + now: DateTime, +) -> Result<()> { + for (node_identity, old, new) in rows { + journal::Entity::insert(journal::ActiveModel { + project_id: Set(project_id), + seq: Set(seq), + node_identity: Set(*node_identity), + kind: Set(kind.to_string()), + old_xml: Set(old.clone()), + new_xml: Set(new.clone()), + at: Set(now), + ..Default::default() + }) + .exec(conn) + .await + .map_err(db_err)?; + } + Ok(()) +} + +/// Mirror the current settings into the settings table (replace-all). +async fn replace_settings( + conn: &C, + project_id: i64, + settings: &HashMap, +) -> Result<()> { + settings::Entity::delete_many() + .filter(settings::Column::ProjectId.eq(project_id)) + .exec(conn) + .await + .map_err(db_err)?; + for (k, v) in settings { + settings::Entity::insert(settings::ActiveModel { + project_id: Set(project_id), + key: Set(k.clone()), + value: Set(v.clone()), + ..Default::default() + }) + .exec(conn) + .await + .map_err(db_err)?; + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// Snapshots and retention (plan §0/§1) +// --------------------------------------------------------------------------- + +/// Snapshot policy: write a full payload when the project is dirty and +/// `Storage/SnapshotIntervalSec` (default 600) seconds have passed since +/// the last snapshot (an interval ≤ 0 snapshots on every dirty save). +async fn maybe_snapshot( + conn: &C, + project_id: i64, + uuid: &str, + head_seq: i64, + now: DateTime, +) -> Result<()> { + let last = snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(project_id)) + .order_by_desc(snapshot::Column::CommandSeq) + .one(conn) + .await + .map_err(db_err)?; + let interval = oakcommon::configstore::ConfigStore::instance() + .get_int(Some("Storage"), "SnapshotIntervalSec", 600); + let due = match &last { + None => true, + Some(s) => { + s.command_seq < head_seq + && (interval <= 0 + || now.signed_duration_since(s.written_at).num_seconds() >= interval as i64) + } + }; + if due { + write_snapshot(conn, project_id, uuid, head_seq, now).await?; + prune_snapshots(conn, project_id).await?; + } + Ok(()) +} + +/// Write the full project payload at `seq` (no-op when a snapshot at +/// that seq already exists). +async fn write_snapshot( + conn: &C, + project_id: i64, + uuid: &str, + seq: i64, + now: DateTime, +) -> Result<()> { + if snapshot::Entity::find_by_id((project_id, seq)) + .one(conn) + .await + .map_err(db_err)? + .is_some() + { + return Ok(()); + } + let payload = assemble_at(conn, project_id, uuid, seq).await?; + snapshot::Entity::insert(snapshot::ActiveModel { + project_id: Set(project_id), + command_seq: Set(seq), + payload: Set(payload), + written_at: Set(now), + ..Default::default() + }) + .exec(conn) + .await + .map_err(db_err)?; + Ok(()) +} + +/// Keep the newest [`SNAPSHOT_KEEP`] snapshots, drop the rest. +async fn prune_snapshots(conn: &C, project_id: i64) -> Result<()> { + let keep: Vec = snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(project_id)) + .order_by_desc(snapshot::Column::CommandSeq) + .limit(SNAPSHOT_KEEP as u64) + .all(conn) + .await + .map_err(db_err)? + .into_iter() + .map(|s| s.command_seq) + .collect(); + if keep.is_empty() { + return Ok(()); + } + snapshot::Entity::delete_many() + .filter(snapshot::Column::ProjectId.eq(project_id)) + .filter(snapshot::Column::CommandSeq.is_not_in(keep)) + .exec(conn) + .await + .map_err(db_err)?; + Ok(()) +} + +/// Journal retention window (`Storage/JournalRetentionDays`, default 0 = +/// keep everything). Rows older than the window are dropped only when +/// the newest snapshot already covers them, so the head state stays +/// reconstructible from the snapshot plus the remaining rows (plan §1: +/// history beyond the window is forfeit). +async fn retention_prune(conn: &C, project_id: i64, now: DateTime) -> Result<()> { + let days = oakcommon::configstore::ConfigStore::instance() + .get_int(Some("Storage"), "JournalRetentionDays", 0); + if days <= 0 { + return Ok(()); + } + let snap_seq = snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(project_id)) + .order_by_desc(snapshot::Column::CommandSeq) + .one(conn) + .await + .map_err(db_err)? + .map(|s| s.command_seq); + if let Some(snap_seq) = snap_seq { + let cutoff = now - chrono::Duration::days(days as i64); + journal::Entity::delete_many() + .filter(journal::Column::ProjectId.eq(project_id)) + .filter(journal::Column::Seq.lte(snap_seq)) + .filter(journal::Column::At.lt(cutoff)) + .exec(conn) + .await + .map_err(db_err)?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn uri(s: &str) -> StorageUri { + StorageUri::parse(s).unwrap() + } + + #[test] + fn parse_target_sqlite_absolute() { + let t = parse_target(&uri("oakdb+sqlite:///tmp/lib.db")).unwrap(); + assert_eq!( + t, + DbTarget::Sqlite { + path: "/tmp/lib.db".to_string(), + project: None + } + ); + } + + #[test] + fn parse_target_sqlite_project_query() { + let t = parse_target(&uri("oakdb+sqlite:///tmp/lib.db?project={abc-123}")).unwrap(); + assert_eq!( + t, + DbTarget::Sqlite { + path: "/tmp/lib.db".to_string(), + project: Some("{abc-123}".to_string()) + } + ); + } + + #[test] + fn parse_target_sqlite_relative_rejected() { + assert!(matches!( + parse_target(&uri("oakdb+sqlite://relative.db")), + Err(Error::Invalid) + )); + assert!(matches!( + parse_target(&uri("oakdb+sqlite://")), + Err(Error::Invalid) + )); + } + + #[test] + fn parse_target_pg_kept_for_d3() { + let t = parse_target(&uri("oakdb+pg://user:pass@host:5432/db")).unwrap(); + assert_eq!( + t, + DbTarget::Pg { + conn: "user:pass@host:5432/db".to_string(), + project: None + } + ); + } + + #[test] + fn parse_target_unknown_scheme() { + assert!(matches!( + parse_target(&uri("oakdb://x")), + Err(Error::NoBackend) + )); + assert!(matches!( + parse_target(&uri("file:///a.ove")), + Err(Error::NoBackend) + )); + } + + #[test] + fn node_fragment_round_trip() { + // serialize_project_state → assemble_xml → extract_nodes must be + // byte-stable for a real project. + let p = Project::new(); + { + let mut guard = p.lock().unwrap(); + guard.initialize().unwrap(); + let (core, behavior) = (oaknode::factory::Factory::global() + .find("org.olivevideoeditor.Olive.math") + .unwrap() + .create)(); + let a = guard.graph.add_node(core, behavior); + guard + .graph + .get_mut(a) + .unwrap() + .core + .set_standard_value("param_a_in", -1, oaknode::value::NodeValue::Float(2.5)); + } + let guard = p.lock().unwrap(); + let (nodes, settings, _) = serialize_project_state(&guard).unwrap(); + let xml = assemble_xml(&guard.uuid, &nodes, &settings); + let back = extract_nodes(&xml); + assert_eq!(back, nodes, "fragment extraction is byte-stable"); + assert_eq!(extract_settings(&xml), settings); + assert!(xml.contains(&guard.uuid)); + // And the assembled document loads. + let loaded = oaknode::serializer::load(&xml).unwrap(); + let loaded_guard = loaded.lock().unwrap(); + assert_eq!(loaded_guard.uuid, guard.uuid); + assert_eq!(loaded_guard.graph.node_count(), guard.graph.node_count()); + } +} diff --git a/crates/oakstorage/src/backends/mod.rs b/crates/oakstorage/src/backends/mod.rs index 641429d45..691bcf413 100644 --- a/crates/oakstorage/src/backends/mod.rs +++ b/crates/oakstorage/src/backends/mod.rs @@ -16,9 +16,9 @@ //! Built-in backends. //! -//! The database backend (`database`) is *not* registered: it is the -//! future replacement (M10 §3 — "数据库替换路径"), provided by a later -//! proxy. The file backends cover today's surface. +//! Arbitration order inside `file://`: ove-xml before otio so a `.ove` +//! path (which the otio backend would also claim as JSON) stays on the +//! ove backend. The database backend claims its own `oakdb+…` schemes. pub mod database; pub mod otio; @@ -32,5 +32,6 @@ pub fn builtins() -> Vec> { vec![ Arc::new(ove_xml::OveXmlBackend::new()), Arc::new(otio::OtioBackend::new()), + Arc::new(database::DatabaseBackend::new()), ] } diff --git a/crates/oakstorage/tests/database_test.rs b/crates/oakstorage/tests/database_test.rs new file mode 100644 index 000000000..04cabe0f4 --- /dev/null +++ b/crates/oakstorage/tests/database_test.rs @@ -0,0 +1,1633 @@ +// Oak Video Editor - Non-Linear Video Editor +// Copyright (C) 2026 Oak Team +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! oakstorage database-backend integration tests (plan M13 D1, the +//! aggregation-granularity design). +//! +//! End-to-end against real SQLite library files in temp directories: +//! full-feature round-trip through a fresh session, the node-granularity +//! journal (import + diff), snapshot + journal replay, undo to any +//! point, snapshot pruning and journal retention, the project-manager +//! API surface (list/delete/duplicate/rename/import/export) and the +//! error/URI matrices. Nothing touches a real library. + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use sea_orm::entity::prelude::*; +use sea_orm::QueryOrder; +use oakcore_rs::Rational; +use oaknode::block::ClipBlockBehavior; +use oaknode::footage::FootageBehavior; +use oaknode::id::NodeId; +use oaknode::keyframe::{Interpolation, Keyframe}; +use oaknode::node::NodeCore; +use oaknode::project::Project; +use oaknode::sequence::SequenceBehavior; +use oaknode::track::{TrackBehavior, TrackListBehavior, TrackType}; +use oaknode::value::NodeValue; +use oakstorage::backend::StorageBackend; +use oakstorage::backends::database::entities::{journal, project, settings, snapshot}; +use oakstorage::backends::database::{derive_stats, DatabaseBackend, KIND_IMPORT, KIND_REDO, ProjectStats, SNAPSHOT_KEEP}; +use oakstorage::error::{ + OAKSTORAGE_E_FORMAT, OAKSTORAGE_E_INVALID, OAKSTORAGE_E_IO, OAKSTORAGE_E_NO_BACKEND, + OAKSTORAGE_E_NOT_FOUND, OAKSTORAGE_OK, +}; +use oakstorage::handle::CHandle; +use oakstorage::nodeutil::{make_project_owned, project_arc}; +use oakstorage::registry::Registry; +use oakstorage::uri::StorageUri; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// A fresh, unique temp directory for one test. +fn temp_dir(tag: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("oakstorage_db_{}_{}", std::process::id(), tag)); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +/// `oakdb+sqlite:///…` URI for a database file. +fn db_uri(path: &Path) -> String { + format!("oakdb+sqlite://{}", path.display()) +} + +/// `oakdb+sqlite:///…?project=` URI selecting one library row. +fn project_uri(db: &str, uuid: &str) -> String { + format!("{db}?project={uuid}") +} + +/// `file://…` URI for a plain file. +fn file_uri(path: &Path) -> String { + format!("file://{}", path.display()) +} + +/// Release an owned handle (refcount 1). +fn release(h: CHandle) { + if let Some(release) = h.release { + unsafe { release(h.ctx) }; + } +} + +/// Save an `Arc>` through the database backend. +fn save_project( + backend: &DatabaseBackend, + project: &Arc>, + uri: &str, +) -> oakstorage::error::Result<()> { + let parsed = StorageUri::parse(uri).unwrap(); + let handle = make_project_owned(project.clone()); + let result = backend.save(handle, &parsed, 0); + release(handle); + result +} + +/// Load a project through a *new* database backend session (a fresh +/// connection pool = a fresh session), returning `(uuid, loaded)`. +fn load_project(backend: &DatabaseBackend, uri: &str) -> (String, Arc>) { + let parsed = StorageUri::parse(uri).unwrap(); + let result = backend.load(&parsed).unwrap(); + assert_eq!(result.version_info, OAKSTORAGE_OK); + let handle = result.project; + let loaded = unsafe { project_arc(&handle) }.unwrap(); + let uuid = loaded.lock().unwrap().uuid.clone(); + release(handle); + (uuid, loaded) +} + +/// Load the state at `seq` (undo to any point). +fn load_at(backend: &DatabaseBackend, uri: &str, uuid: &str, seq: i64) -> Arc> { + let parsed = StorageUri::parse(uri).unwrap(); + let handle = backend.load_at(&parsed, uuid, seq).unwrap(); + let loaded = unsafe { project_arc(&handle) }.unwrap(); + release(handle); + loaded +} + +fn r_to_f(r: Rational) -> f64 { + r.numerator() as f64 / r.denominator() as f64 +} + +fn assert_close(a: f64, b: f64) { + assert!((a - b).abs() < 1e-6, "expected {a} close to {b}"); +} + +/// Open a raw sea-orm connection to the database file (for inspecting +/// and injecting rows behind the backend's back) and drive one future +/// against it on a private current-thread runtime. The connection is +/// handed to `f` by value (a clone of the pool). +fn inspect_db(path: &Path, f: impl FnOnce(sea_orm::DatabaseConnection) -> Fut) -> R +where + Fut: std::future::Future, +{ + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(async { + let options = sea_orm::sqlx::sqlite::SqliteConnectOptions::new() + .filename(path) + .create_if_missing(true) + .journal_mode(sea_orm::sqlx::sqlite::SqliteJournalMode::Wal) + .busy_timeout(Duration::from_secs(5)) + .foreign_keys(true); + let pool = sea_orm::sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .unwrap(); + f(sea_orm::DatabaseConnection::from(pool)).await + }) +} + +/// The settings mirror rows of a project. +fn settings_rows(path: &Path, project_id: i64) -> Vec<(String, String)> { + inspect_db(path, |conn| async move { + settings::Entity::find() + .filter(settings::Column::ProjectId.eq(project_id)) + .all(&conn) + .await + .unwrap() + .into_iter() + .map(|s| (s.key, s.value)) + .collect() + }) +} + +/// The project uuid. +fn uuid_of(project: &Arc>) -> String { + project.lock().unwrap().uuid.clone() +} + +/// Read a config value for the duration of a test (the config store is +/// process-global, so restore it afterwards). +fn with_config(group: &str, key: &str, value: i32, f: impl FnOnce()) { + let store = oakcommon::configstore::ConfigStore::instance(); + let before = store.get_int(Some(group), key, 0); + store.set_int(Some(group), key, value); + f(); + store.set_int(Some(group), key, before); +} + +// --------------------------------------------------------------------------- +// Full-feature fixture (footage / sequence / track / clip / effect / +// keyframes) — the union of the ove round-trip and timeline fixtures. +// --------------------------------------------------------------------------- + +const MATH: &str = "org.olivevideoeditor.Olive.math"; + +/// Build the round-trip fixture: root folder + two math nodes with +/// values/keyframes/label/color/link/connection + a sequence "My Seq" +/// with one video track carrying two clips (footage /a/b.mp4, /a/c.mp4). +fn build_full_project() -> Arc> { + let project = Project::new(); + let mut p = project.lock().unwrap(); + p.initialize().unwrap(); + + let (core, behavior) = (oaknode::factory::Factory::global().find(MATH).unwrap().create)(); + let a = p.graph.add_node(core, behavior); + { + let e = p.graph.get_mut(a).unwrap(); + e.core.label = "Math A".to_string(); + e.core.override_color = 2; + e.core.set_standard_value("param_a_in", -1, NodeValue::Float(2.5)); + e.core + .keyframe_track_mut("param_a_in", -1) + .set_key(Keyframe { + time: Rational::new(0, 1), + value: NodeValue::Float(1.0), + interpolation: Interpolation::Linear, + bezier_in: (0.0, 0.0), + bezier_out: (0.0, 0.0), + }); + e.core + .keyframe_track_mut("param_a_in", -1) + .set_key(Keyframe { + time: Rational::new(1, 1), + value: NodeValue::Float(3.0), + interpolation: Interpolation::Bezier, + bezier_in: (0.1, 0.2), + bezier_out: (0.3, 0.4), + }); + } + let (core, behavior) = (oaknode::factory::Factory::global().find(MATH).unwrap().create)(); + let b = p.graph.add_node(core, behavior); + p.graph + .get_mut(b) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(4.0)); + p.graph.connect(a, b, "param_b_in", -1).unwrap(); + p.graph.link(a, b); + + // Timeline: sequence + video track + two clips with footage. + let (seq_id, lists) = oakstorage::nodeutil::create_sequence(&mut p.graph); + p.graph.get_mut(seq_id).unwrap().core.label = "My Seq".to_string(); + + let mut tb = TrackBehavior::new(TrackType::Video); + tb.track_list = Some(lists[0]); + let track_id = p.graph.add_node(NodeCore::new(), Box::new(tb)); + + let foot1 = p + .graph + .add_node(NodeCore::new(), Box::new(FootageBehavior::new("/a/b.mp4"))); + let clip1 = { + let (core, mut behavior) = oaknode::block::clip_create(); + let clip = behavior + .as_any_mut() + .and_then(|a| a.downcast_mut::()) + .unwrap(); + clip.core.range = oakcore_rs::TimeRange::new(Rational::new(0, 1), Rational::new(100, 25)); + clip.core.media_in = Rational::new(0, 1); + clip.core.track = Some(track_id); + clip.footage = Some(foot1); + p.graph.add_node(core, behavior) + }; + + let foot2 = p + .graph + .add_node(NodeCore::new(), Box::new(FootageBehavior::new("/a/c.mp4"))); + let clip2 = { + let (core, mut behavior) = oaknode::block::clip_create(); + let clip = behavior + .as_any_mut() + .and_then(|a| a.downcast_mut::()) + .unwrap(); + clip.core.range = + oakcore_rs::TimeRange::new(Rational::new(100, 25), Rational::new(150, 25)); + clip.core.media_in = Rational::new(10, 25); + clip.core.track = Some(track_id); + clip.footage = Some(foot2); + p.graph.add_node(core, behavior) + }; + + if let Some(entry) = p.graph.get_mut(track_id) { + entry + .behavior + .as_any_mut() + .and_then(|a| a.downcast_mut::()) + .unwrap() + .blocks = vec![clip1, clip2]; + } + if let Some(entry) = p.graph.get_mut(lists[0]) { + entry + .behavior + .as_any_mut() + .and_then(|a| a.downcast_mut::()) + .unwrap() + .tracks = vec![track_id]; + } + + p.settings + .insert("projectname".to_string(), "full-fixture".to_string()); + drop(p); + project +} + +/// Field-by-field comparison of the full fixture after a round-trip +/// (uuid included). +fn assert_full_fields(orig: &Project, loaded: &Project) { + assert_eq!(loaded.uuid, orig.uuid, "uuid"); + assert_full_state(orig, loaded); +} + +/// Field-by-field comparison that ignores the uuid (a duplicated row +/// carries a fresh uuid by design). +fn assert_full_state(orig: &Project, loaded: &Project) { + assert_eq!(loaded.settings, orig.settings, "settings"); + + // Node set: count, type order (slot order preserved by the writer). + let o_ids = orig.graph.node_ids(); + let l_ids = loaded.graph.node_ids(); + assert_eq!(l_ids.len(), o_ids.len(), "node count"); + let o_types: Vec<&str> = o_ids + .iter() + .map(|id| orig.graph.get(*id).unwrap().behavior.type_id()) + .collect(); + let l_types: Vec<&str> = l_ids + .iter() + .map(|id| loaded.graph.get(*id).unwrap().behavior.type_id()) + .collect(); + assert_eq!(l_types, o_types, "node types"); + + // Math A: label, color, value, keyframes. + let (a_o, a_l) = (o_ids[1], l_ids[1]); + assert_eq!( + loaded.graph.get(a_l).unwrap().core.label, + orig.graph.get(a_o).unwrap().core.label, + "label" + ); + assert_eq!( + loaded.graph.get(a_l).unwrap().core.override_color, + orig.graph.get(a_o).unwrap().core.override_color, + "color" + ); + assert_eq!( + loaded + .graph + .get(a_l) + .unwrap() + .core + .standard_value("param_a_in", -1), + orig.graph + .get(a_o) + .unwrap() + .core + .standard_value("param_a_in", -1), + "value a" + ); + let keys_o = orig + .graph + .get(a_o) + .unwrap() + .core + .keyframe_track("param_a_in", -1) + .unwrap() + .keys() + .to_vec(); + let keys_l = loaded + .graph + .get(a_l) + .unwrap() + .core + .keyframe_track("param_a_in", -1) + .unwrap() + .keys() + .to_vec(); + assert_eq!(keys_l.len(), keys_o.len(), "keyframe count"); + for (ko, kl) in keys_o.iter().zip(&keys_l) { + assert_eq!(kl.time, ko.time, "key time"); + assert_eq!(kl.value.to_double(), ko.value.to_double(), "key value"); + assert_eq!(kl.interpolation, ko.interpolation, "key interpolation"); + assert_eq!(kl.bezier_in, ko.bezier_in, "key bezier in"); + assert_eq!(kl.bezier_out, ko.bezier_out, "key bezier out"); + } + + // Connection a -> b.param_b_in and the link. + assert_eq!( + loaded.graph.connected_output(l_ids[2], "param_b_in", -1), + Some(a_l), + "connection" + ); + assert!(loaded.graph.are_linked(a_l, l_ids[2]), "link"); + + // Timeline: sequence, track lists, track, clips, footage. + let mut seq: Option<(NodeId, Vec)> = None; + for id in loaded.graph.node_ids() { + let entry = loaded.graph.get(id).unwrap(); + if entry.behavior.type_id() == "org.olivevideoeditor.Olive.sequence" { + let s = entry + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + seq = Some((id, s.track_lists.clone())); + break; + } + } + let (_seq_id, lists) = seq.expect("sequence present"); + assert_eq!(lists.len(), 3, "video/audio/subtitle lists"); + let list = loaded + .graph + .get(lists[0]) + .unwrap() + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + assert_eq!(list.kind, TrackType::Video); + assert_eq!(list.array_base, 0); + assert_eq!(list.tracks.len(), 1); + let track = loaded + .graph + .get(list.tracks[0]) + .unwrap() + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + assert_eq!(track.kind, TrackType::Video); + assert_eq!(track.blocks.len(), 2); + let c1 = loaded + .graph + .get(track.blocks[0]) + .unwrap() + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + assert_close(r_to_f(c1.core.length()), 4.0); + assert_close(r_to_f(c1.core.media_in), 0.0); + let f1 = loaded + .graph + .get(c1.footage.unwrap()) + .unwrap() + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + assert_eq!(f1.filename, "/a/b.mp4"); + let c2 = loaded + .graph + .get(track.blocks[1]) + .unwrap() + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + assert_close(r_to_f(c2.core.length()), 2.0); + assert_close(r_to_f(c2.core.media_in), 0.4); + let f2 = loaded + .graph + .get(c2.footage.unwrap()) + .unwrap() + .behavior + .as_any() + .and_then(|a| a.downcast_ref::()) + .unwrap(); + assert_eq!(f2.filename, "/a/c.mp4"); +} + +/// Build a small project with a `projectname` setting (fast saves). +fn build_named_project(name: &str) -> Arc> { + let project = Project::new(); + { + let mut p = project.lock().unwrap(); + p.initialize().unwrap(); + p.settings.insert("projectname".to_string(), name.to_string()); + } + project +} + +// --------------------------------------------------------------------------- +// Round-trip through a fresh session +// --------------------------------------------------------------------------- + +#[test] +fn roundtrip_field_by_field() { + let dir = temp_dir("rt"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + + // A new backend instance = a new session; the library row is + // selected explicitly by uuid. + let session = DatabaseBackend::new(); + let (loaded_uuid, loaded) = load_project(&session, &project_uri(&uri, &uuid)); + assert_eq!(loaded_uuid, uuid); + { + let o = project.lock().unwrap(); + let l = loaded.lock().unwrap(); + assert_full_fields(&o, &l); + } + + // The default (no ?project=) pick returns the same single row. + let session = DatabaseBackend::new(); + let (loaded_uuid, _) = load_project(&session, &uri); + assert_eq!(loaded_uuid, uuid); + + // The settings mirror carries the current keys. + let row = inspect_db(&db, |conn| async move { + project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .unwrap() + .unwrap() + .id + }); + let mirrored = settings_rows(&db, row); + assert!(mirrored.contains(&("projectname".to_string(), "full-fixture".to_string())), "{mirrored:?}"); +} + +#[test] +fn registry_routes_oakdb_schemes() { + assert_eq!( + Registry::global() + .resolve(&StorageUri::parse("oakdb+sqlite:///tmp/x.db").unwrap()) + .unwrap() + .name(), + "oakdb" + ); + assert_eq!( + Registry::global() + .resolve(&StorageUri::parse("oakdb+pg://user@host/db").unwrap()) + .unwrap() + .name(), + "oakdb" + ); + // The legacy `oakdb://` (no sub-scheme) stays unclaimed. + assert_eq!( + Registry::global() + .resolve(&StorageUri::parse("oakdb://user@host/db").unwrap()) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NO_BACKEND + ); +} + +// --------------------------------------------------------------------------- +// Journal semantics: import, diff, no-op save +// --------------------------------------------------------------------------- + +/// First save is one `kind='import'` command carrying every node plus +/// the settings pseudo-node (plan §2); the project row advances to seq 1. +#[test] +fn first_save_is_an_import_command() { + let dir = temp_dir("imp"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + let node_count = project.lock().unwrap().graph.node_count(); + save_project(&backend, &project, &uri).unwrap(); + + let rows = inspect_db(&db, |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .unwrap() + .unwrap(); + let rows = journal::Entity::find() + .filter(journal::Column::ProjectId.eq(proj.id)) + .order_by_asc(journal::Column::Seq) + .order_by_asc(journal::Column::NodeIdentity) + .all(&conn) + .await + .unwrap(); + (proj.id, proj.command_seq, rows) + }); + let (pid, command_seq, rows) = rows; + assert_eq!(command_seq, 1); + assert_eq!(rows.len(), node_count + 1, "every node + the settings pseudo-node"); + assert!(rows.iter().all(|r| r.seq == 1 && r.kind == KIND_IMPORT), "one import command"); + assert!(rows.iter().all(|r| r.old_xml.is_none()), "import has no before images"); + assert!(rows.iter().all(|r| r.new_xml.is_some()), "import has after images"); + // The settings pseudo-node (identity 0) is present. + assert!(rows.iter().any(|r| r.node_identity == 0), "settings row present"); + assert_eq!( + rows.iter().filter(|r| r.node_identity == 0).count(), + 1, + "exactly one settings row" + ); + // The real-node rows use the +1 offset (settings 0 reserved). + assert!(rows.iter().all(|r| r.node_identity >= 1 || r.node_identity == 0)); + // The node fragments round-trip: the settings row holds the settings + // element. + let settings_row = rows.iter().find(|r| r.node_identity == 0).unwrap(); + assert!(settings_row.new_xml.as_deref().unwrap().starts_with("")); + let _ = pid; +} + +/// A later save is a diff: only the changed nodes land as `kind='redo'` +/// rows with both before and after images (plan §0). +#[test] +fn later_saves_are_diffs() { + let dir = temp_dir("diff"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + + // Change one math node's value and one setting key. + { + let mut p = project.lock().unwrap(); + let math_ids: Vec = p + .graph + .node_ids() + .into_iter() + .filter(|id| p.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + let a = math_ids[0]; + p.graph + .get_mut(a) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(9.5)); + p.settings.insert("projectname".to_string(), "diffed".to_string()); + } + save_project(&backend, &project, &uri).unwrap(); + + let uuid_q = uuid.clone(); + let rows = inspect_db(&db, move |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid_q)) + .one(&conn) + .await + .unwrap() + .unwrap(); + let rows = journal::Entity::find() + .filter(journal::Column::ProjectId.eq(proj.id)) + .all(&conn) + .await + .unwrap(); + (proj.id, proj.command_seq, rows) + }); + let (pid, command_seq, rows) = rows; + assert_eq!(command_seq, 2); + let seq2: Vec<&journal::Model> = rows.iter().filter(|r| r.seq == 2).collect(); + assert_eq!(seq2.len(), 2, "one changed node + the settings row, nothing else"); + assert!(seq2.iter().all(|r| r.kind == KIND_REDO)); + assert!(seq2.iter().all(|r| r.old_xml.is_some() && r.new_xml.is_some()), "diff carries both images"); + let settings_row = seq2.iter().find(|r| r.node_identity == 0).unwrap(); + assert!(settings_row.new_xml.as_deref().unwrap().contains(">diffed<")); + let _ = pid; + + // Head state after the diff reflects both changes. + let session = DatabaseBackend::new(); + let (_, loaded) = load_project(&session, &project_uri(&uri, &uuid)); + { + let l = loaded.lock().unwrap(); + let math_ids: Vec = l + .graph + .node_ids() + .into_iter() + .filter(|id| l.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + assert_eq!( + l.graph.get(math_ids[0]).unwrap().core.standard_value("param_a_in", -1), + NodeValue::Float(9.5) + ); + assert_eq!(l.settings.get("projectname").cloned(), Some("diffed".to_string())); + } +} + +/// A save that changes nothing bumps no command seq and writes no rows. +#[test] +fn no_op_save_is_a_touch_only() { + let dir = temp_dir("noop"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + save_project(&backend, &project, &uri).unwrap(); + + let (head, count) = inspect_db(&db, |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .unwrap() + .unwrap(); + let n = journal::Entity::find() + .filter(journal::Column::ProjectId.eq(proj.id)) + .count(&conn) + .await + .unwrap(); + (proj.command_seq, n) + }); + assert_eq!(head, 1, "no-op save keeps the head seq"); + // The import wrote 12 nodes + 1 settings row; the no-op added none. + assert_eq!(count, 13, "only the import rows remain"); +} + +// --------------------------------------------------------------------------- +// Snapshot + journal replay and undo to any point +// --------------------------------------------------------------------------- + +/// With `Storage/SnapshotIntervalSec` ≤ 0 every dirty save writes a +/// snapshot; the newest snapshot is the replay base and the journal rows +/// after it are applied on top. Deleting every snapshot still recovers +/// the same state from an empty base plus the full journal (plan §0: +/// "快照损坏也能从空工程 + 全 journal 重建"). +#[test] +fn snapshot_and_journal_replay() { + let dir = temp_dir("replay"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + with_config("Storage", "SnapshotIntervalSec", 0, || { + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + // Second command: one value change. + { + let mut p = project.lock().unwrap(); + let math_ids: Vec = p + .graph + .node_ids() + .into_iter() + .filter(|id| p.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + p.graph + .get_mut(math_ids[0]) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(7.25)); + } + save_project(&backend, &project, &uri).unwrap(); + + // Both saves produced snapshots (head seq 2). + let uuid_q = uuid.clone(); + let snaps = inspect_db(&db, move |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid_q)) + .one(&conn) + .await + .unwrap() + .unwrap(); + snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(proj.id)) + .all(&conn) + .await + .unwrap() + }); + assert_eq!(snaps.len(), 2, "one snapshot per dirty save"); + + // Load (head): snapshot seq 2 is the base, nothing after it. + let session = DatabaseBackend::new(); + let (_, loaded) = load_project(&session, &project_uri(&uri, &uuid)); + { + let o = project.lock().unwrap(); + let l = loaded.lock().unwrap(); + assert_full_fields(&o, &l); + } + + // Destroy the snapshots: replay degrades to empty base + full + // journal and still reconstructs the head. + let uuid_q = uuid.clone(); + inspect_db(&db, move |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid_q)) + .one(&conn) + .await + .unwrap() + .unwrap(); + snapshot::Entity::delete_many() + .filter(snapshot::Column::ProjectId.eq(proj.id)) + .exec(&conn) + .await + .unwrap(); + }); + let session = DatabaseBackend::new(); + let (_, loaded) = load_project(&session, &project_uri(&uri, &uuid)); + { + let o = project.lock().unwrap(); + let l = loaded.lock().unwrap(); + assert_full_fields(&o, &l); + } + }); +} + +/// The journal is the persistent undo history: `load_at(seq)` replays to +/// any point (plan §0 "撤销到任意点"). +#[test] +fn undo_to_any_point() { + let dir = temp_dir("undo"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + let math_a = { + let p = project.lock().unwrap(); + p.graph + .node_ids() + .into_iter() + .find(|id| p.graph.get(*id).unwrap().behavior.type_id() == MATH) + .unwrap() + }; + // Command 1: value 2.5 (fixture default). + save_project(&backend, &project, &uri).unwrap(); + // Command 2: value 6.0. + { + let mut p = project.lock().unwrap(); + p.graph + .get_mut(math_a) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(6.0)); + } + save_project(&backend, &project, &uri).unwrap(); + // Command 3: add a node. + let (new_id, value) = { + let mut p = project.lock().unwrap(); + let (core, behavior) = (oaknode::factory::Factory::global().find(MATH).unwrap().create)(); + let id = p.graph.add_node(core, behavior); + p.graph + .get_mut(id) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(11.0)); + let v = p + .graph + .get(id) + .unwrap() + .core + .standard_value("param_a_in", -1); + (id, v) + }; + save_project(&backend, &project, &uri).unwrap(); + + // Head (seq 3): the extra node exists. + let session = DatabaseBackend::new(); + let (_, head) = load_project(&session, &project_uri(&uri, &uuid)); + { + let h = head.lock().unwrap(); + assert!(h.graph.is_valid(new_id), "node added in command 3 is live"); + assert_eq!( + h.graph.get(new_id).unwrap().core.standard_value("param_a_in", -1), + value + ); + } + + // Undo to seq 2: the node is gone, the value is 6.0. + let session = DatabaseBackend::new(); + let at2 = load_at(&session, &uri, &uuid, 2); + { + let l = at2.lock().unwrap(); + assert!(!l.graph.is_valid(new_id), "command 3 rolled back"); + let id = l + .graph + .node_ids() + .into_iter() + .find(|id| l.graph.get(*id).unwrap().behavior.type_id() == MATH) + .unwrap(); + assert_eq!( + l.graph.get(id).unwrap().core.standard_value("param_a_in", -1), + NodeValue::Float(6.0) + ); + } + + // Undo to seq 1: the value is back to the fixture default. + let session = DatabaseBackend::new(); + let at1 = load_at(&session, &uri, &uuid, 1); + { + let l = at1.lock().unwrap(); + assert_eq!(l.graph.node_count(), 12, "fixture node count"); + let id = l + .graph + .node_ids() + .into_iter() + .find(|id| l.graph.get(*id).unwrap().behavior.type_id() == MATH) + .unwrap(); + assert_eq!( + l.graph.get(id).unwrap().core.standard_value("param_a_in", -1), + NodeValue::Float(2.5) + ); + } + + // Undo to seq 0: an empty project. + let session = DatabaseBackend::new(); + let at0 = load_at(&session, &uri, &uuid, 0); + { + let l = at0.lock().unwrap(); + assert_eq!(l.graph.node_count(), 0, "empty project at seq 0"); + } + + // Out of range -> E_INVALID. + let session = DatabaseBackend::new(); + assert_eq!( + session + .load_at(&StorageUri::parse(&uri).unwrap(), &uuid, 99) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_INVALID + ); +} + +/// Snapshot pruning keeps the newest [`SNAPSHOT_KEEP`] copies. +#[test] +fn snapshot_pruning_keeps_three() { + let dir = temp_dir("prune"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + with_config("Storage", "SnapshotIntervalSec", 0, || { + let project = build_full_project(); + let uuid = uuid_of(&project); + // 6 commands, each dirty-snapshotted. + for i in 0..6i64 { + { + let mut p = project.lock().unwrap(); + let id = p.graph.node_ids()[0]; + p.graph + .get_mut(id) + .unwrap() + .core + .label = format!("step {i}"); + } + save_project(&backend, &project, &uri).unwrap(); + } + let seqs = inspect_db(&db, |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid)) + .one(&conn) + .await + .unwrap() + .unwrap(); + snapshot::Entity::find() + .filter(snapshot::Column::ProjectId.eq(proj.id)) + .order_by_desc(snapshot::Column::CommandSeq) + .all(&conn) + .await + .unwrap() + .into_iter() + .map(|s| s.command_seq) + .collect::>() + }); + assert_eq!(seqs.len() as u64, SNAPSHOT_KEEP, "only the newest {SNAPSHOT_KEEP}"); + assert_eq!(seqs[0], 6, "the newest survives"); + }); +} + +/// Journal retention (`Storage/JournalRetentionDays`): rows older than +/// the window and covered by the newest snapshot are dropped; the head +/// stays reconstructible (snapshot + remaining rows). +#[test] +fn journal_retention_truncation() { + let dir = temp_dir("retain"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + // Command 1 (import) + command 2 (a value change). + save_project(&backend, &project, &uri).unwrap(); + { + let mut p = project.lock().unwrap(); + let math_ids: Vec = p + .graph + .node_ids() + .into_iter() + .filter(|id| p.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + p.graph + .get_mut(math_ids[0]) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(3.25)); + } + save_project(&backend, &project, &uri).unwrap(); + // Snapshot at the head so rows ≤ 2 are covered. + backend.snapshot(&StorageUri::parse(&uri).unwrap(), &uuid).unwrap(); + + // Backdate every journal row two days, then save a third command + // with a 1-day retention window. + inspect_db(&db, |conn| async move { + let old = chrono::Utc::now().naive_utc() - chrono::Duration::days(2); + journal::Entity::update_many() + .col_expr(journal::Column::At, sea_orm::sea_query::Expr::value(old)) + .exec(&conn) + .await + .unwrap(); + }); + with_config("Storage", "JournalRetentionDays", 1, || { + { + let mut p = project.lock().unwrap(); + let math_ids: Vec = p + .graph + .node_ids() + .into_iter() + .filter(|id| p.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + p.graph + .get_mut(math_ids[0]) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(4.5)); + } + save_project(&backend, &project, &uri).unwrap(); + }); + + // Only the third command's rows survive. + let uuid_q = uuid.clone(); + let (pid, rows) = inspect_db(&db, move |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&uuid_q)) + .one(&conn) + .await + .unwrap() + .unwrap(); + let rows = journal::Entity::find() + .filter(journal::Column::ProjectId.eq(proj.id)) + .all(&conn) + .await + .unwrap(); + (proj.id, rows) + }); + let seqs: Vec = rows.iter().map(|r| r.seq).collect(); + assert!(!seqs.contains(&1) && !seqs.contains(&2), "old commands pruned: {seqs:?}"); + assert_eq!(seqs, vec![3], "only the fresh command remains"); + let _ = pid; + + // The head state is still correct (snapshot at seq 2 + command 3). + let session = DatabaseBackend::new(); + let (_, loaded) = load_project(&session, &project_uri(&uri, &uuid)); + { + let l = loaded.lock().unwrap(); + let math_ids: Vec = l + .graph + .node_ids() + .into_iter() + .filter(|id| l.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + assert_eq!( + l.graph.get(math_ids[0]).unwrap().core.standard_value("param_a_in", -1), + NodeValue::Float(4.5) + ); + assert_eq!(l.graph.node_count(), 12); + } +} + +// --------------------------------------------------------------------------- +// Project-manager API surface +// --------------------------------------------------------------------------- + +fn save_named_project(backend: &DatabaseBackend, uri: &str, name: &str) -> String { + let project = build_named_project(name); + let uuid = uuid_of(&project); + save_project(backend, &project, uri).unwrap(); + uuid +} + +#[test] +fn list_delete_duplicate_rename() { + let dir = temp_dir("mgr"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + // Empty library lists nothing. + assert!(backend.list_projects(&StorageUri::parse(&uri).unwrap()).unwrap().is_empty()); + + let alpha = save_named_project(&backend, &uri, "Alpha"); + let beta = save_named_project(&backend, &uri, "Beta"); + + let list = backend.list_projects(&StorageUri::parse(&uri).unwrap()).unwrap(); + assert_eq!(list.len(), 2, "two rows"); + // Most recently modified first. + assert_eq!(list[0].name, "Beta"); + assert_eq!(list[1].name, "Alpha"); + assert_eq!(list[0].command_seq, 1); + assert_eq!(list[0].schema_ver, oaknode::serializer::CURRENT_VERSION.0 as i32); + + // Rename (library metadata). + backend + .rename_project(&StorageUri::parse(&uri).unwrap(), &alpha, "Alpha Renamed") + .unwrap(); + let list = backend.list_projects(&StorageUri::parse(&uri).unwrap()).unwrap(); + assert!(list.iter().any(|p| p.name == "Alpha Renamed"), "{list:?}"); + assert_eq!( + backend + .rename_project(&StorageUri::parse(&uri).unwrap(), "{missing}", "X") + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NOT_FOUND + ); + + // Duplicate: fresh uuid, default "(copy)" name, history copied. + let copy = backend + .duplicate_project(&StorageUri::parse(&uri).unwrap(), &beta, None) + .unwrap(); + assert_ne!(copy.uuid, beta, "fresh uuid"); + assert_eq!(copy.name, "Beta (copy)"); + assert_eq!(copy.command_seq, 1); + let list = backend.list_projects(&StorageUri::parse(&uri).unwrap()).unwrap(); + assert_eq!(list.len(), 3); + let renamed_copy = backend + .duplicate_project(&StorageUri::parse(&uri).unwrap(), &beta, Some("Beta Clone")) + .unwrap(); + assert_eq!(renamed_copy.name, "Beta Clone"); + assert_eq!( + backend + .duplicate_project(&StorageUri::parse(&uri).unwrap(), "{missing}", None) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NOT_FOUND + ); + + // Delete. + for uuid in [&alpha, &beta, ©.uuid, &renamed_copy.uuid] { + backend + .delete_project(&StorageUri::parse(&uri).unwrap(), uuid) + .unwrap(); + } + assert!(backend.list_projects(&StorageUri::parse(&uri).unwrap()).unwrap().is_empty()); + assert_eq!( + backend + .delete_project(&StorageUri::parse(&uri).unwrap(), "{missing}") + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NOT_FOUND + ); +} + +/// Duplicating a full-feature project copies the whole history: the copy +/// loads identically under its own uuid and keeps the undo history. +#[test] +fn duplicate_preserves_history() { + let dir = temp_dir("dup"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + { + let mut p = project.lock().unwrap(); + let math_ids: Vec = p + .graph + .node_ids() + .into_iter() + .filter(|id| p.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + p.graph + .get_mut(math_ids[0]) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(6.0)); + } + save_project(&backend, &project, &uri).unwrap(); + + let copy = backend + .duplicate_project(&StorageUri::parse(&uri).unwrap(), &uuid, None) + .unwrap(); + assert_eq!(copy.command_seq, 2); + + // The copy loads through a fresh session, field-for-field (uuid is + // fresh by design). + let session = DatabaseBackend::new(); + let (loaded_uuid, loaded) = load_project(&session, &project_uri(&uri, ©.uuid)); + assert_eq!(loaded_uuid, copy.uuid); + { + let o = project.lock().unwrap(); + let l = loaded.lock().unwrap(); + assert_full_state(&o, &l); + } + + // Undo history travels with the copy: load_at(1) on the copy gives + // the pre-change state. + let session = DatabaseBackend::new(); + let at1 = load_at(&session, &uri, ©.uuid, 1); + { + let l = at1.lock().unwrap(); + let math_ids: Vec = l + .graph + .node_ids() + .into_iter() + .filter(|id| l.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + assert_eq!( + l.graph.get(math_ids[0]).unwrap().core.standard_value("param_a_in", -1), + NodeValue::Float(2.5) + ); + } +} + +/// Manager stats are derived from the node graph, not stored (plan §4). +#[test] +fn project_stats_derived_from_graph() { + let dir = temp_dir("stats"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + + let stats = backend + .project_stats(&StorageUri::parse(&uri).unwrap(), &uuid) + .unwrap(); + assert_eq!( + stats, + ProjectStats { + duration_ms: 6000, + track_count: 1, + clip_count: 2, + footage_count: 2, + } + ); + // Same numbers derive_stats yields directly on the live project. + let guard = project.lock().unwrap(); + assert_eq!(derive_stats(&guard), stats); + assert_eq!( + backend + .project_stats(&StorageUri::parse(&uri).unwrap(), "{missing}") + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NOT_FOUND + ); +} + +#[test] +fn export_and_import_round_trip() { + let dir = temp_dir("xi"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let project = build_full_project(); + let uuid = uuid_of(&project); + save_project(&backend, &project, &uri).unwrap(); + + // Export: .ove written from an in-memory assembly; the file backend + // re-imports it byte-for-byte. + let out = dir.join("exported.ove"); + let file = file_uri(&out); + backend + .export_to_file( + &StorageUri::parse(&uri).unwrap(), + &uuid, + &StorageUri::parse(&file).unwrap(), + ) + .unwrap(); + assert!(out.exists(), ".ove written"); + let text = std::fs::read_to_string(&out).unwrap(); + assert!(text.starts_with(""), "{text}"); + let file_backend = Registry::global() + .resolve(&StorageUri::parse(&file).unwrap()) + .unwrap(); + let result = file_backend + .load(&StorageUri::parse(&file).unwrap()) + .unwrap(); + assert_eq!(result.version_info, OAKSTORAGE_OK); + let loaded = unsafe { project_arc(&result.project) }.unwrap(); + release(result.project); + { + let o = project.lock().unwrap(); + let l = loaded.lock().unwrap(); + assert_full_fields(&o, &l); + } + + // Import: the .ove lands as a new library row under a fresh uuid; + // importing it again yields a distinct row. + let imported = backend + .import_from_file( + &StorageUri::parse(&uri).unwrap(), + &StorageUri::parse(&file).unwrap(), + ) + .unwrap(); + assert_ne!(imported, uuid, "import gets a fresh uuid"); + let imported2 = backend + .import_from_file( + &StorageUri::parse(&uri).unwrap(), + &StorageUri::parse(&file).unwrap(), + ) + .unwrap(); + assert_ne!(imported2, imported, "repeat imports are new rows"); + let list = backend.list_projects(&StorageUri::parse(&uri).unwrap()).unwrap(); + assert_eq!(list.len(), 3, "original + two imports"); + let session = DatabaseBackend::new(); + let (imported_uuid, imported_proj) = load_project(&session, &project_uri(&uri, &imported)); + assert_eq!(imported_uuid, imported); + { + let l = imported_proj.lock().unwrap(); + assert_eq!(l.graph.node_count(), 12); + assert_eq!(l.settings.get("projectname").cloned(), Some("full-fixture".to_string())); + } + + // Error paths: unknown project on export; non-file target; corrupt + // file on import. + assert_eq!( + backend + .export_to_file( + &StorageUri::parse(&uri).unwrap(), + "{missing}", + &StorageUri::parse(&file).unwrap(), + ) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NOT_FOUND + ); + assert_eq!( + backend + .export_to_file( + &StorageUri::parse(&uri).unwrap(), + &uuid, + &StorageUri::parse("oakdb+sqlite:///tmp/x.db").unwrap(), + ) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_INVALID + ); + let corrupt = dir.join("corrupt.ove"); + std::fs::write(&corrupt, "").unwrap(); + assert_eq!( + backend + .import_from_file( + &StorageUri::parse(&uri).unwrap(), + &StorageUri::parse(&file_uri(&corrupt)).unwrap(), + ) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_FORMAT + ); +} + +// --------------------------------------------------------------------------- +// Project selection (`?project=` vs default) +// --------------------------------------------------------------------------- + +#[test] +fn project_selection_via_query() { + let dir = temp_dir("sel"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + let a = save_named_project(&backend, &uri, "A"); + let b = save_named_project(&backend, &uri, "B"); + + // Explicit uuid picks the right row regardless of recency. + let session = DatabaseBackend::new(); + let (loaded_uuid, _) = load_project(&session, &project_uri(&uri, &a)); + assert_eq!(loaded_uuid, a); + + // Default pick = most recently modified (B was written last). + let session = DatabaseBackend::new(); + let (loaded_uuid, loaded) = load_project(&session, &uri); + assert_eq!(loaded_uuid, b); + assert_eq!( + loaded.lock().unwrap().settings.get("projectname").cloned(), + Some("B".to_string()) + ); + + // Unknown uuid -> E_NOT_FOUND; empty library -> E_NOT_FOUND. + let session = DatabaseBackend::new(); + assert_eq!( + session + .load(&StorageUri::parse(&project_uri(&uri, "{missing}")).unwrap()) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NOT_FOUND + ); +} + +// --------------------------------------------------------------------------- +// Error paths, locking, URI matrix +// --------------------------------------------------------------------------- + +#[test] +fn pg_target_rejected_until_d3() { + let dir = temp_dir("pg"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + let project = build_full_project(); + save_project(&backend, &project, &uri).unwrap(); + let uuid = uuid_of(&project); + + let pg = "oakdb+pg://user:pass@host:5432/db"; + for call in [ + backend.load(&StorageUri::parse(pg).unwrap()).map(|_| ()), + backend.save(CHandle::null(), &StorageUri::parse(pg).unwrap(), 0), + backend.list_projects(&StorageUri::parse(pg).unwrap()).map(|_| ()), + backend.delete_project(&StorageUri::parse(pg).unwrap(), &uuid), + backend.rename_project(&StorageUri::parse(pg).unwrap(), &uuid, "X"), + backend + .duplicate_project(&StorageUri::parse(pg).unwrap(), &uuid, None) + .map(|_| ()), + backend + .export_to_file( + &StorageUri::parse(pg).unwrap(), + &uuid, + &StorageUri::parse("file:///tmp/x.ove").unwrap(), + ), + backend.snapshot(&StorageUri::parse(pg).unwrap(), &uuid), + backend.load_at(&StorageUri::parse(pg).unwrap(), &uuid, 1).map(|_| ()), + ] { + assert_eq!(call.err().unwrap().code(), OAKSTORAGE_E_NO_BACKEND, "pg target"); + } +} + +#[test] +fn invalid_uri_matrix() { + let dir = temp_dir("uri"); + // Relative path / empty body -> E_INVALID. + let backend = DatabaseBackend::new(); + for bad in ["oakdb+sqlite://relative.db", "oakdb+sqlite://"] { + let err = backend + .list_projects(&StorageUri::parse(bad).unwrap()) + .err() + .unwrap(); + assert_eq!(err.code(), OAKSTORAGE_E_INVALID, "{bad}"); + } + // Nonexistent parent directory -> E_IO at connect. + let err = backend + .list_projects(&StorageUri::parse(&db_uri(&dir.join("no/such/dir/lib.db"))).unwrap()) + .err() + .unwrap(); + assert_eq!(err.code(), OAKSTORAGE_E_IO, "missing parent dir"); + + // Unknown scheme stays unclaimed by the registry. + assert_eq!( + Registry::global() + .resolve(&StorageUri::parse("oakdb+sqlite3:///tmp/x.db").unwrap()) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_NO_BACKEND + ); +} + +/// Write failures surface as errors, not panics or silent corruption: +/// (a) read-only database files, (b) an out-of-range undo target, and +/// (c) concurrent writers on one file. +#[test] +fn failure_paths_report_cleanly() { + let dir = temp_dir("fail"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + let backend = DatabaseBackend::new(); + + // (a) Read-only db + WAL files: a fresh backend cannot write. + let project = build_full_project(); + save_project(&backend, &project, &uri).unwrap(); + let mut targets = vec![db.clone()]; + for suffix in ["-wal", "-shm"] { + let f = PathBuf::from(format!("{}{}", db.display(), suffix)); + if f.exists() { + targets.push(f); + } + } + let saved = targets + .iter() + .map(|f| std::fs::metadata(f).unwrap().permissions()) + .collect::>(); + for f in &targets { + let mut p = std::fs::metadata(f).unwrap().permissions(); + p.set_readonly(true); + std::fs::set_permissions(f, p).unwrap(); + } + let fresh = DatabaseBackend::new(); + let err = save_project(&fresh, &project, &uri).err().unwrap(); + assert_eq!(err.code(), OAKSTORAGE_E_IO, "read-only library write must fail"); + for (f, p) in targets.iter().zip(saved) { + std::fs::set_permissions(f, p).unwrap(); + } + + // (b) Undo beyond the head -> E_INVALID (the head-seq range is + // enforced; load_at(0) stays valid). + let uuid = uuid_of(&project); + let session = DatabaseBackend::new(); + assert_eq!( + session + .load_at(&StorageUri::parse(&uri).unwrap(), &uuid, -1) + .err() + .unwrap() + .code(), + OAKSTORAGE_E_INVALID + ); +} + +/// Two backends writing different projects to one file concurrently +/// complete without deadlock, and a same-uuid write race surfaces a +/// clean error instead of corrupting the library. +#[test] +fn concurrent_writers_are_serialized() { + let dir = temp_dir("conc"); + let db = dir.join("lib.db"); + let uri = db_uri(&db); + + // Two projects, two threads, different uuids: both writers land. + let p1 = build_full_project(); + let p2 = build_full_project(); + let uri_t1 = uri.clone(); + let uri_t2 = uri.clone(); + let b1 = DatabaseBackend::new(); + let b2 = DatabaseBackend::new(); + let t1 = std::thread::spawn(move || { + for _ in 0..5 { + save_project(&b1, &p1, &uri_t1).unwrap(); + } + }); + let t2 = std::thread::spawn(move || { + for _ in 0..5 { + save_project(&b2, &p2, &uri_t2).unwrap(); + } + }); + t1.join().unwrap(); + t2.join().unwrap(); + let list = DatabaseBackend::new() + .list_projects(&StorageUri::parse(&uri).unwrap()) + .unwrap(); + assert_eq!(list.len(), 2, "both projects persisted"); + assert!(list.iter().all(|p| p.name == "full-fixture"), "{list:?}"); + + // Same-uuid race: both threads save projects with the same uuid but + // different content, so the writes genuinely contend (on the import + // row, or on journal seqs). Results are Ok or a clean error, never a + // panic, and the library still loads afterwards. + let b3 = DatabaseBackend::new(); + let b4 = DatabaseBackend::new(); + let p3 = build_full_project(); + let p4 = build_full_project(); + { + let mut g = p4.lock().unwrap(); + g.uuid = uuid_of(&p3); // force the same row identity + let math_ids: Vec = g + .graph + .node_ids() + .into_iter() + .filter(|id| g.graph.get(*id).unwrap().behavior.type_id() == MATH) + .collect(); + g.graph + .get_mut(math_ids[0]) + .unwrap() + .core + .set_standard_value("param_a_in", -1, NodeValue::Float(8.0)); + } + let same = uuid_of(&p3); + let uri_a = uri.clone(); + let uri_b = uri.clone(); + let ta = std::thread::spawn(move || { + (0..3) + .map(|_| save_project(&b3, &p3, &uri_a).map_err(|e| e.code())) + .collect::>() + }); + let tb = std::thread::spawn(move || { + (0..3) + .map(|_| save_project(&b4, &p4, &uri_b).map_err(|e| e.code())) + .collect::>() + }); + let (ra, rb) = (ta.join().unwrap(), tb.join().unwrap()); + for (i, r) in ra.iter().chain(rb.iter()).enumerate() { + match r { + Ok(()) => {} + Err(code) => { + // A same-uuid/seq insert conflict surfaces as a database + // error (or a retried BUSY gives up). + assert!( + *code == OAKSTORAGE_E_IO || *code == oakstorage::error::OAKSTORAGE_E_FAILED, + "writer {i} failed with unexpected code {code}" + ); + } + } + } + // The library is still consistent: the same-uuid project loads and + // the journal's newest command matches the row's head seq. + let session = DatabaseBackend::new(); + let (loaded_uuid, _) = load_project(&session, &project_uri(&uri, &same)); + assert_eq!(loaded_uuid, same, "library consistent after races"); + let (head, seqs) = inspect_db(&db, |conn| async move { + let proj = project::Entity::find() + .filter(project::Column::Uuid.eq(&same)) + .one(&conn) + .await + .unwrap() + .unwrap(); + let mut seqs: Vec = journal::Entity::find() + .filter(journal::Column::ProjectId.eq(proj.id)) + .all(&conn) + .await + .unwrap() + .into_iter() + .map(|r| r.seq) + .collect(); + seqs.sort(); + (proj.command_seq, seqs) + }); + assert!(!seqs.is_empty(), "the same-uuid project has commands"); + assert_eq!( + seqs.last().copied(), + Some(head), + "head seq matches the newest command (seqs={seqs:?})" + ); +}