diff --git a/crates/nana-store/Cargo.toml b/crates/nana-store/Cargo.toml index 05bee4f..afab3a9 100644 --- a/crates/nana-store/Cargo.toml +++ b/crates/nana-store/Cargo.toml @@ -7,6 +7,8 @@ license.workspace = true [dependencies] nana-domain.workspace = true +rusqlite.workspace = true +serde_json.workspace = true thiserror.workspace = true [lints] diff --git a/crates/nana-store/src/lib.rs b/crates/nana-store/src/lib.rs index bdb49ef..420a10d 100644 --- a/crates/nana-store/src/lib.rs +++ b/crates/nana-store/src/lib.rs @@ -1,9 +1,11 @@ use std::{ collections::BTreeMap, + path::Path, sync::{Mutex, MutexGuard}, }; use nana_domain::{RuntimeState, StoryNode}; +use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; use thiserror::Error; #[derive(Debug, Error, PartialEq, Eq)] @@ -20,10 +22,26 @@ pub enum StoreError { StaleBranchHead { expected: String, actual: String }, #[error("node and materialized state disagree: {0}")] StateMismatch(&'static str), + #[error("sqlite storage error: {0}")] + Sqlite(String), + #[error("JSON serialization error: {0}")] + Serialization(String), #[error("storage lock is poisoned")] Poisoned, } +impl From for StoreError { + fn from(error: rusqlite::Error) -> Self { + Self::Sqlite(error.to_string()) + } +} + +impl From for StoreError { + fn from(error: serde_json::Error) -> Self { + Self::Serialization(error.to_string()) + } +} + pub trait StoryStore: Send + Sync { fn append_node(&self, node: &StoryNode, state: &RuntimeState) -> Result<(), StoreError>; @@ -153,6 +171,275 @@ impl StoryStore for InMemoryStoryStore { } } +/// Durable SQLite implementation of the append-only story store. +/// +/// A connection is kept behind a mutex so one store value preserves the same +/// `Send + Sync` contract as the in-memory implementation. SQLite still +/// provides the cross-connection transaction boundary when a database is +/// reopened or shared by multiple store values. +pub struct SqliteStoryStore { + connection: Mutex, +} + +impl SqliteStoryStore { + /// Opens or creates a story database at `path`. + pub fn open(path: impl AsRef) -> Result { + let connection = Connection::open(path)?; + Self::from_connection(connection) + } + + /// Opens a fresh in-memory story database. + pub fn open_in_memory() -> Result { + let connection = Connection::open_in_memory()?; + Self::from_connection(connection) + } + + /// Returns the current head node for a branch, if the branch exists. + pub fn branch_head( + &self, + story_id: &str, + branch_id: &str, + ) -> Result, StoreError> { + let connection = self.lock()?; + connection + .query_row( + "SELECT head_node_id + FROM branch_heads + WHERE story_id = ?1 AND branch_id = ?2", + params![story_id, branch_id], + |row| row.get(0), + ) + .optional() + .map_err(StoreError::from) + } + + /// Loads the materialized state associated with a specific node. + pub fn load_state_at_node( + &self, + story_id: &str, + node_id: &str, + ) -> Result { + let connection = self.lock()?; + let stored = connection + .query_row( + "SELECT nodes.branch_id, materialized_states.state_json + FROM nodes + LEFT JOIN materialized_states + ON materialized_states.story_id = nodes.story_id + AND materialized_states.node_id = nodes.node_id + WHERE nodes.story_id = ?1 AND nodes.node_id = ?2", + params![story_id, node_id], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Option>(1)?, + )) + }, + ) + .optional()? + .ok_or_else(|| StoreError::ParentNotFound(node_id.to_owned()))?; + + let state_json = stored + .1 + .ok_or(StoreError::StateMismatch("node has no materialized state"))?; + let state = deserialize_state(&state_json)?; + validate_loaded_state(&state, story_id, node_id, &stored.0)?; + Ok(state) + } + + fn from_connection(connection: Connection) -> Result { + initialize_schema(&connection)?; + Ok(Self { + connection: Mutex::new(connection), + }) + } + + fn lock(&self) -> Result, StoreError> { + self.connection.lock().map_err(|_| StoreError::Poisoned) + } +} + +impl StoryStore for SqliteStoryStore { + fn append_node(&self, node: &StoryNode, state: &RuntimeState) -> Result<(), StoreError> { + validate_materialized_state(node, state)?; + let node_json = serde_json::to_string(node)?; + let state_json = serde_json::to_string(state)?; + + let mut connection = self.lock()?; + let transaction = + connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + + let node_exists = transaction.query_row( + "SELECT EXISTS( + SELECT 1 FROM nodes WHERE story_id = ?1 AND node_id = ?2 + )", + params![node.story_id, node.id], + |row| row.get::<_, bool>(0), + )?; + if node_exists { + return Err(StoreError::NodeAlreadyExists(node.id.clone())); + } + + if let Some(parent_id) = &node.parent_id { + let parent_exists = transaction.query_row( + "SELECT EXISTS( + SELECT 1 FROM nodes WHERE story_id = ?1 AND node_id = ?2 + )", + params![node.story_id, parent_id], + |row| row.get::<_, bool>(0), + )?; + if !parent_exists { + return Err(StoreError::ParentNotFound(parent_id.clone())); + } + } else { + let story_exists = transaction.query_row( + "SELECT EXISTS(SELECT 1 FROM nodes WHERE story_id = ?1)", + params![node.story_id], + |row| row.get::<_, bool>(0), + )?; + if story_exists { + return Err(StoreError::ParentNotFound( + "a non-root node must name its parent".to_owned(), + )); + } + } + + let current_head = transaction + .query_row( + "SELECT head_node_id + FROM branch_heads + WHERE story_id = ?1 AND branch_id = ?2", + params![node.story_id, node.branch_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if let Some(current_head) = current_head { + let expected_parent = node.parent_id.as_deref().unwrap_or(""); + if current_head != expected_parent { + return Err(StoreError::StaleBranchHead { + expected: current_head, + actual: expected_parent.to_owned(), + }); + } + } + + transaction.execute( + "INSERT INTO nodes ( + story_id, node_id, branch_id, parent_id, node_json + ) VALUES (?1, ?2, ?3, ?4, ?5)", + params![ + node.story_id, + node.id, + node.branch_id, + node.parent_id, + node_json + ], + )?; + transaction.execute( + "INSERT INTO materialized_states (story_id, node_id, state_json) + VALUES (?1, ?2, ?3)", + params![node.story_id, node.id, state_json], + )?; + transaction.execute( + "INSERT INTO branch_heads (story_id, branch_id, head_node_id) + VALUES (?1, ?2, ?3) + ON CONFLICT(story_id, branch_id) + DO UPDATE SET head_node_id = excluded.head_node_id", + params![node.story_id, node.branch_id, node.id], + )?; + transaction.commit()?; + Ok(()) + } + + fn load_state(&self, story_id: &str, branch_id: &str) -> Result { + let connection = self.lock()?; + let story_exists = connection.query_row( + "SELECT EXISTS(SELECT 1 FROM nodes WHERE story_id = ?1)", + params![story_id], + |row| row.get::<_, bool>(0), + )?; + if !story_exists { + return Err(StoreError::StoryNotFound(story_id.to_owned())); + } + + let stored = connection + .query_row( + "SELECT branch_heads.head_node_id, materialized_states.state_json + FROM branch_heads + LEFT JOIN materialized_states + ON materialized_states.story_id = branch_heads.story_id + AND materialized_states.node_id = branch_heads.head_node_id + WHERE branch_heads.story_id = ?1 + AND branch_heads.branch_id = ?2", + params![story_id, branch_id], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Option>(1)?, + )) + }, + ) + .optional()? + .ok_or_else(|| StoreError::BranchNotFound { + story_id: story_id.to_owned(), + branch_id: branch_id.to_owned(), + })?; + + let state_json = stored + .1 + .ok_or(StoreError::StateMismatch("branch head has no state"))?; + let state = deserialize_state(&state_json)?; + validate_loaded_state(&state, story_id, &stored.0, branch_id)?; + Ok(state) + } +} + +fn initialize_schema(connection: &Connection) -> Result<(), StoreError> { + connection.execute_batch( + "PRAGMA foreign_keys = ON; + BEGIN IMMEDIATE; + + CREATE TABLE IF NOT EXISTS nodes ( + story_id TEXT NOT NULL, + node_id TEXT NOT NULL, + branch_id TEXT NOT NULL, + parent_id TEXT, + node_json TEXT NOT NULL, + PRIMARY KEY (story_id, node_id), + FOREIGN KEY (story_id, parent_id) + REFERENCES nodes (story_id, node_id) + ON DELETE RESTRICT + ); + + CREATE UNIQUE INDEX IF NOT EXISTS one_root_per_story + ON nodes (story_id) + WHERE parent_id IS NULL; + + CREATE TABLE IF NOT EXISTS materialized_states ( + story_id TEXT NOT NULL, + node_id TEXT NOT NULL, + state_json TEXT NOT NULL, + PRIMARY KEY (story_id, node_id), + FOREIGN KEY (story_id, node_id) + REFERENCES nodes (story_id, node_id) + ON DELETE RESTRICT + ); + + CREATE TABLE IF NOT EXISTS branch_heads ( + story_id TEXT NOT NULL, + branch_id TEXT NOT NULL, + head_node_id TEXT NOT NULL, + PRIMARY KEY (story_id, branch_id), + FOREIGN KEY (story_id, head_node_id) + REFERENCES nodes (story_id, node_id) + ON DELETE RESTRICT + ); + + COMMIT;", + )?; + Ok(()) +} + fn validate_materialized_state( node: &StoryNode, state: &RuntimeState, @@ -169,13 +456,41 @@ fn validate_materialized_state( Ok(()) } +fn deserialize_state(state_json: &str) -> Result { + serde_json::from_str(state_json).map_err(StoreError::from) +} + +fn validate_loaded_state( + state: &RuntimeState, + story_id: &str, + node_id: &str, + branch_id: &str, +) -> Result<(), StoreError> { + if state.story_id != story_id { + return Err(StoreError::StateMismatch("story_id")); + } + if state.current_node != node_id { + return Err(StoreError::StateMismatch("current_node")); + } + if state.current_branch != branch_id { + return Err(StoreError::StateMismatch("current_branch")); + } + Ok(()) +} + #[cfg(test)] mod tests { - use std::collections::BTreeMap; + use std::{ + collections::BTreeMap, + fs, + path::{Path, PathBuf}, + time::{SystemTime, UNIX_EPOCH}, + }; use nana_domain::{RuntimeState, StateDelta, StoryNode}; + use rusqlite::params; - use super::{InMemoryStoryStore, StoreError, StoryStore}; + use super::{InMemoryStoryStore, SqliteStoryStore, StoreError, StoryStore}; fn state(node: &str, branch: &str) -> RuntimeState { RuntimeState { @@ -207,11 +522,62 @@ mod tests { } } - #[test] - fn appends_and_loads_the_branch_head() { - let store = InMemoryStoryStore::new(); + trait InspectableStoryStore: StoryStore { + fn inspected_branch_head( + &self, + story_id: &str, + branch_id: &str, + ) -> Result, StoreError>; + + fn inspected_state_at_node( + &self, + story_id: &str, + node_id: &str, + ) -> Result; + } + + impl InspectableStoryStore for InMemoryStoryStore { + fn inspected_branch_head( + &self, + story_id: &str, + branch_id: &str, + ) -> Result, StoreError> { + self.branch_head(story_id, branch_id) + } + + fn inspected_state_at_node( + &self, + story_id: &str, + node_id: &str, + ) -> Result { + self.load_state_at_node(story_id, node_id) + } + } + + impl InspectableStoryStore for SqliteStoryStore { + fn inspected_branch_head( + &self, + story_id: &str, + branch_id: &str, + ) -> Result, StoreError> { + self.branch_head(story_id, branch_id) + } + + fn inspected_state_at_node( + &self, + story_id: &str, + node_id: &str, + ) -> Result { + self.load_state_at_node(story_id, node_id) + } + } + + fn assert_appends_and_loads_the_branch_head(store: &impl InspectableStoryStore) { store - .append_node(&node("node_001", None, "branch_main"), &state("node_001", "branch_main")) + .append_node( + &node("node_001", None, "branch_main"), + &state("node_001", "branch_main"), + ) .expect("root append"); store .append_node( @@ -227,13 +593,21 @@ mod tests { .current_node, "node_002" ); + assert_eq!( + store + .inspected_state_at_node("story_demo", "node_001") + .expect("root state") + .current_node, + "node_001" + ); } - #[test] - fn forks_from_an_old_node_without_moving_the_source_head() { - let store = InMemoryStoryStore::new(); + fn assert_forks_from_an_old_node(store: &impl InspectableStoryStore) { store - .append_node(&node("node_001", None, "branch_main"), &state("node_001", "branch_main")) + .append_node( + &node("node_001", None, "branch_main"), + &state("node_001", "branch_main"), + ) .expect("root append"); store .append_node( @@ -250,23 +624,24 @@ mod tests { assert_eq!( store - .branch_head("story_demo", "branch_main") + .inspected_branch_head("story_demo", "branch_main") .expect("main head"), Some("node_002".to_owned()) ); assert_eq!( store - .branch_head("story_demo", "branch_rewind") + .inspected_branch_head("story_demo", "branch_rewind") .expect("fork head"), Some("node_003".to_owned()) ); } - #[test] - fn rejects_a_stale_append_without_moving_the_head() { - let store = InMemoryStoryStore::new(); + fn assert_rejects_a_stale_append(store: &impl InspectableStoryStore) { store - .append_node(&node("node_001", None, "branch_main"), &state("node_001", "branch_main")) + .append_node( + &node("node_001", None, "branch_main"), + &state("node_001", "branch_main"), + ) .expect("root append"); store .append_node( @@ -287,27 +662,228 @@ mod tests { ); assert_eq!( store - .branch_head("story_demo", "branch_main") + .inspected_branch_head("story_demo", "branch_main") .expect("main head"), Some("node_002".to_owned()) ); + assert_eq!( + store.inspected_state_at_node("story_demo", "node_stale"), + Err(StoreError::ParentNotFound("node_stale".to_owned())) + ); } - #[test] - fn rejects_state_that_does_not_match_the_node() { - let store = InMemoryStoryStore::new(); + fn assert_rejects_state_mismatches(store: &impl InspectableStoryStore) { + let root = node("node_001", None, "branch_main"); + + let mut wrong_story = state("node_001", "branch_main"); + wrong_story.story_id = "another_story".to_owned(); assert_eq!( - store.append_node( - &node("node_001", None, "branch_main"), - &state("different_node", "branch_main"), - ), + store.append_node(&root, &wrong_story), + Err(StoreError::StateMismatch("story_id")) + ); + + assert_eq!( + store.append_node(&root, &state("different_node", "branch_main")), Err(StoreError::StateMismatch("current_node")) ); + assert_eq!( + store.append_node(&root, &state("node_001", "different_branch")), + Err(StoreError::StateMismatch("current_branch")) + ); assert_eq!( store - .branch_head("story_demo", "branch_main") + .inspected_branch_head("story_demo", "branch_main") .expect("empty store"), None ); } + + #[test] + fn memory_appends_and_loads_the_branch_head() { + let store = InMemoryStoryStore::new(); + assert_appends_and_loads_the_branch_head(&store); + } + + #[test] + fn sqlite_appends_and_loads_the_branch_head() { + let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store"); + assert_appends_and_loads_the_branch_head(&store); + } + + #[test] + fn memory_forks_from_an_old_node_without_moving_the_source_head() { + let store = InMemoryStoryStore::new(); + assert_forks_from_an_old_node(&store); + } + + #[test] + fn sqlite_forks_from_an_old_node_without_moving_the_source_head() { + let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store"); + assert_forks_from_an_old_node(&store); + } + + #[test] + fn memory_rejects_a_stale_append_without_moving_the_head() { + let store = InMemoryStoryStore::new(); + assert_rejects_a_stale_append(&store); + } + + #[test] + fn sqlite_rejects_a_stale_append_without_moving_the_head() { + let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store"); + assert_rejects_a_stale_append(&store); + } + + #[test] + fn memory_rejects_state_that_does_not_match_the_node() { + let store = InMemoryStoryStore::new(); + assert_rejects_state_mismatches(&store); + } + + #[test] + fn sqlite_rejects_state_that_does_not_match_the_node() { + let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store"); + assert_rejects_state_mismatches(&store); + } + + #[test] + fn sqlite_enforces_node_and_parent_rules_atomically() { + let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store"); + let root = node("node_001", None, "branch_main"); + let root_state = state("node_001", "branch_main"); + store + .append_node(&root, &root_state) + .expect("root append"); + + assert_eq!( + store.append_node(&root, &root_state), + Err(StoreError::NodeAlreadyExists("node_001".to_owned())) + ); + assert_eq!( + store.append_node( + &node("node_missing_parent", Some("unknown"), "branch_fork"), + &state("node_missing_parent", "branch_fork"), + ), + Err(StoreError::ParentNotFound("unknown".to_owned())) + ); + assert_eq!( + store.append_node( + &node("node_second_root", None, "branch_other"), + &state("node_second_root", "branch_other"), + ), + Err(StoreError::ParentNotFound( + "a non-root node must name its parent".to_owned() + )) + ); + assert_eq!( + store + .branch_head("story_demo", "branch_main") + .expect("main head"), + Some("node_001".to_owned()) + ); + assert_eq!( + store + .branch_head("story_demo", "branch_fork") + .expect("failed fork"), + None + ); + } + + #[test] + fn sqlite_reports_corrupt_json_without_panicking() { + let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store"); + store + .append_node( + &node("node_001", None, "branch_main"), + &state("node_001", "branch_main"), + ) + .expect("root append"); + store + .connection + .lock() + .expect("SQLite connection lock") + .execute( + "UPDATE materialized_states SET state_json = ?1", + params!["{"], + ) + .expect("corrupt test state"); + + assert!(matches!( + store.load_state("story_demo", "branch_main"), + Err(StoreError::Serialization(_)) + )); + } + + struct TemporaryDatabase { + path: PathBuf, + } + + impl TemporaryDatabase { + fn new() -> Self { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + Self { + path: std::env::temp_dir().join(format!( + "nana-store-{}-{nonce}.sqlite3", + std::process::id() + )), + } + } + + fn path(&self) -> &Path { + &self.path + } + } + + impl Drop for TemporaryDatabase { + fn drop(&mut self) { + if self.path.exists() { + fs::remove_file(&self.path).expect("remove temporary SQLite database"); + } + } + } + + #[test] + fn sqlite_reopens_and_preserves_nodes_states_and_heads() { + let database = TemporaryDatabase::new(); + { + let store = SqliteStoryStore::open(database.path()).expect("file SQLite store"); + store + .append_node( + &node("node_001", None, "branch_main"), + &state("node_001", "branch_main"), + ) + .expect("root append"); + store + .append_node( + &node("node_002", Some("node_001"), "branch_main"), + &state("node_002", "branch_main"), + ) + .expect("child append"); + } + + let reopened = SqliteStoryStore::open(database.path()).expect("reopened SQLite store"); + assert_eq!( + reopened + .branch_head("story_demo", "branch_main") + .expect("persisted head"), + Some("node_002".to_owned()) + ); + assert_eq!( + reopened + .load_state("story_demo", "branch_main") + .expect("persisted head state") + .current_node, + "node_002" + ); + assert_eq!( + reopened + .load_state_at_node("story_demo", "node_001") + .expect("persisted root state") + .current_node, + "node_001" + ); + } }