use std::{
    collections::BTreeMap,
    path::Path,
    sync::{Mutex, MutexGuard},
};

use nana_domain::{RuntimeState, StoryNode, stable_json_hash};
use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior};
use thiserror::Error;

#[derive(Debug, Error, PartialEq, Eq)]
pub enum StoreError {
    #[error("story not found: {0}")]
    StoryNotFound(String),
    #[error("branch not found: {story_id}/{branch_id}")]
    BranchNotFound { story_id: String, branch_id: String },
    #[error("node already exists: {0}")]
    NodeAlreadyExists(String),
    #[error("parent node not found: {0}")]
    ParentNotFound(String),
    #[error("branch head changed: expected {expected}, got {actual}")]
    StaleBranchHead { expected: String, actual: String },
    #[error("node and materialized state disagree: {0}")]
    StateMismatch(&'static str),
    #[error("materialized state hash does not match node: {node_id}")]
    StateHashMismatch { node_id: String },
    #[error("sqlite storage error: {0}")]
    Sqlite(String),
    #[error("JSON serialization error: {0}")]
    Serialization(String),
    #[error("storage lock is poisoned")]
    Poisoned,
}

impl From<rusqlite::Error> for StoreError {
    fn from(error: rusqlite::Error) -> Self {
        Self::Sqlite(error.to_string())
    }
}

impl From<serde_json::Error> 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>;

    fn load_state(&self, story_id: &str, branch_id: &str) -> Result<RuntimeState, StoreError>;

    fn load_node(&self, story_id: &str, node_id: &str) -> Result<StoryNode, StoreError>;
}

#[derive(Debug, Default)]
struct MemoryData {
    nodes: BTreeMap<(String, String), StoryNode>,
    states: BTreeMap<(String, String), RuntimeState>,
    branch_heads: BTreeMap<(String, String), String>,
}

/// Deterministic test and development store.
///
/// Production persistence will implement the same append-only semantics with
/// SQLite. Keeping this implementation in the crate lets the engine/runtime
/// integration tests exercise branch rules without depending on a database.
#[derive(Debug, Default)]
pub struct InMemoryStoryStore {
    data: Mutex<MemoryData>,
}

impl InMemoryStoryStore {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    pub fn branch_head(
        &self,
        story_id: &str,
        branch_id: &str,
    ) -> Result<Option<String>, StoreError> {
        let data = self.lock()?;
        Ok(data
            .branch_heads
            .get(&(story_id.to_owned(), branch_id.to_owned()))
            .cloned())
    }

    pub fn load_state_at_node(
        &self,
        story_id: &str,
        node_id: &str,
    ) -> Result<RuntimeState, StoreError> {
        let data = self.lock()?;
        data.states
            .get(&(story_id.to_owned(), node_id.to_owned()))
            .cloned()
            .ok_or_else(|| StoreError::ParentNotFound(node_id.to_owned()))
    }

    fn lock(&self) -> Result<MutexGuard<'_, MemoryData>, StoreError> {
        self.data.lock().map_err(|_| StoreError::Poisoned)
    }
}

impl StoryStore for InMemoryStoryStore {
    fn append_node(&self, node: &StoryNode, state: &RuntimeState) -> Result<(), StoreError> {
        validate_materialized_state(node, state)?;

        let node_key = (node.story_id.clone(), node.id.clone());
        let branch_key = (node.story_id.clone(), node.branch_id.clone());
        let mut data = self.lock()?;

        if data.nodes.contains_key(&node_key) {
            return Err(StoreError::NodeAlreadyExists(node.id.clone()));
        }

        if let Some(parent_id) = &node.parent_id {
            if !data
                .nodes
                .contains_key(&(node.story_id.clone(), parent_id.clone()))
            {
                return Err(StoreError::ParentNotFound(parent_id.clone()));
            }
        } else if data
            .nodes
            .keys()
            .any(|(story_id, _)| story_id == &node.story_id)
        {
            return Err(StoreError::ParentNotFound(
                "a non-root node must name its parent".to_owned(),
            ));
        }

        if let Some(current_head) = data.branch_heads.get(&branch_key) {
            let expected_parent = node.parent_id.as_deref().unwrap_or("<none>");
            if current_head != expected_parent {
                return Err(StoreError::StaleBranchHead {
                    expected: current_head.clone(),
                    actual: expected_parent.to_owned(),
                });
            }
        }

        // All validation happens before any map is changed. This mirrors the
        // all-or-nothing transaction boundary required from the SQLite store.
        data.nodes.insert(node_key.clone(), node.clone());
        data.states.insert(node_key, state.clone());
        data.branch_heads.insert(branch_key, node.id.clone());
        Ok(())
    }

    fn load_state(&self, story_id: &str, branch_id: &str) -> Result<RuntimeState, StoreError> {
        let data = self.lock()?;
        let story_exists = data
            .nodes
            .keys()
            .any(|(stored_story_id, _)| stored_story_id == story_id);
        if !story_exists {
            return Err(StoreError::StoryNotFound(story_id.to_owned()));
        }

        let head = data
            .branch_heads
            .get(&(story_id.to_owned(), branch_id.to_owned()))
            .ok_or_else(|| StoreError::BranchNotFound {
                story_id: story_id.to_owned(),
                branch_id: branch_id.to_owned(),
            })?;
        data.states
            .get(&(story_id.to_owned(), head.clone()))
            .cloned()
            .ok_or(StoreError::StateMismatch("branch head has no state"))
    }

    fn load_node(&self, story_id: &str, node_id: &str) -> Result<StoryNode, StoreError> {
        let data = self.lock()?;
        data.nodes
            .get(&(story_id.to_owned(), node_id.to_owned()))
            .cloned()
            .ok_or_else(|| StoreError::ParentNotFound(node_id.to_owned()))
    }
}

/// 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<Connection>,
}

impl SqliteStoryStore {
    /// Opens or creates a story database at `path`.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let connection = Connection::open(path)?;
        Self::from_connection(connection)
    }

    /// Opens a fresh in-memory story database.
    pub fn open_in_memory() -> Result<Self, StoreError> {
        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<Option<String>, 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<RuntimeState, StoreError> {
        let connection = self.lock()?;
        let stored = connection
            .query_row(
                "SELECT nodes.branch_id, nodes.parent_id, nodes.node_json,
                        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<String>>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, Option<String>>(3)?,
                    ))
                },
            )
            .optional()?
            .ok_or_else(|| StoreError::ParentNotFound(node_id.to_owned()))?;

        let node = deserialize_node(&stored.2)?;
        validate_loaded_node(&node, story_id, node_id, &stored.0, stored.1.as_deref())?;
        let state_json = stored
            .3
            .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)?;
        validate_state_hash(&node, &state)?;
        Ok(state)
    }

    fn from_connection(connection: Connection) -> Result<Self, StoreError> {
        initialize_schema(&connection)?;
        Ok(Self {
            connection: Mutex::new(connection),
        })
    }

    fn lock(&self) -> Result<MutexGuard<'_, Connection>, 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("<none>");
            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<RuntimeState, StoreError> {
        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, nodes.branch_id, nodes.parent_id,
                        nodes.node_json, materialized_states.state_json
                 FROM branch_heads
                 JOIN nodes
                   ON nodes.story_id = branch_heads.story_id
                  AND nodes.node_id = branch_heads.head_node_id
                  AND nodes.branch_id = branch_heads.branch_id
                 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::<_, String>(1)?,
                        row.get::<_, Option<String>>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, Option<String>>(4)?,
                    ))
                },
            )
            .optional()?
            .ok_or_else(|| StoreError::BranchNotFound {
                story_id: story_id.to_owned(),
                branch_id: branch_id.to_owned(),
            })?;

        let node = deserialize_node(&stored.3)?;
        validate_loaded_node(
            &node,
            story_id,
            &stored.0,
            &stored.1,
            stored.2.as_deref(),
        )?;
        let state_json = stored
            .4
            .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)?;
        validate_state_hash(&node, &state)?;
        Ok(state)
    }

    fn load_node(&self, story_id: &str, node_id: &str) -> Result<StoryNode, StoreError> {
        let connection = self.lock()?;
        let stored = connection
            .query_row(
                "SELECT branch_id, parent_id, node_json
                 FROM nodes
                 WHERE story_id = ?1 AND node_id = ?2",
                params![story_id, node_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, Option<String>>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .optional()?
            .ok_or_else(|| StoreError::ParentNotFound(node_id.to_owned()))?;
        let node = deserialize_node(&stored.2)?;
        validate_loaded_node(&node, story_id, node_id, &stored.0, stored.1.as_deref())?;
        Ok(node)
    }
}

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,
) -> Result<(), StoreError> {
    if node.story_id != state.story_id {
        return Err(StoreError::StateMismatch("story_id"));
    }
    if node.id != state.current_node {
        return Err(StoreError::StateMismatch("current_node"));
    }
    if node.branch_id != state.current_branch {
        return Err(StoreError::StateMismatch("current_branch"));
    }
    validate_state_hash(node, state)?;
    Ok(())
}

fn deserialize_node(node_json: &str) -> Result<StoryNode, StoreError> {
    serde_json::from_str(node_json).map_err(StoreError::from)
}

fn deserialize_state(state_json: &str) -> Result<RuntimeState, StoreError> {
    serde_json::from_str(state_json).map_err(StoreError::from)
}

fn validate_loaded_node(
    node: &StoryNode,
    story_id: &str,
    node_id: &str,
    branch_id: &str,
    parent_id: Option<&str>,
) -> Result<(), StoreError> {
    if node.story_id != story_id {
        return Err(StoreError::StateMismatch("story_id"));
    }
    if node.id != node_id {
        return Err(StoreError::StateMismatch("current_node"));
    }
    if node.branch_id != branch_id {
        return Err(StoreError::StateMismatch("current_branch"));
    }
    if node.parent_id.as_deref() != parent_id {
        return Err(StoreError::StateMismatch("parent_id"));
    }
    Ok(())
}

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

fn validate_state_hash(node: &StoryNode, state: &RuntimeState) -> Result<(), StoreError> {
    let bytes = serde_json::to_vec(state)?;
    if stable_json_hash(&bytes) != node.state_hash {
        return Err(StoreError::StateHashMismatch {
            node_id: node.id.clone(),
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        fs,
        path::{Path, PathBuf},
        time::{SystemTime, UNIX_EPOCH},
    };

    use nana_domain::{RuntimeState, StateDelta, StoryNode, stable_json_hash};
    use rusqlite::params;

    use super::{InMemoryStoryStore, SqliteStoryStore, StoreError, StoryStore};

    fn state(node: &str, branch: &str) -> RuntimeState {
        RuntimeState {
            story_id: "story_demo".to_owned(),
            current_node: node.to_owned(),
            current_branch: branch.to_owned(),
            world_flags: BTreeMap::new(),
            relationships: BTreeMap::new(),
            relationship_states: Vec::new(),
            promises: Vec::new(),
            knowledge: Vec::new(),
            items: Vec::new(),
            clocks: Vec::new(),
            checks: Vec::new(),
        }
    }

    fn node(id: &str, parent_id: Option<&str>, branch: &str) -> StoryNode {
        let materialized = state(id, branch);
        StoryNode {
            id: id.to_owned(),
            story_id: "story_demo".to_owned(),
            branch_id: branch.to_owned(),
            parent_id: parent_id.map(ToOwned::to_owned),
            action_id: format!("action_{id}"),
            user_input: String::new(),
            beats: Vec::new(),
            delta: StateDelta { ops: Vec::new() },
            state_hash: stable_json_hash(
                &serde_json::to_vec(&materialized).expect("serializable test state"),
            ),
        }
    }

    trait InspectableStoryStore: StoryStore {
        fn inspected_branch_head(
            &self,
            story_id: &str,
            branch_id: &str,
        ) -> Result<Option<String>, StoreError>;

        fn inspected_state_at_node(
            &self,
            story_id: &str,
            node_id: &str,
        ) -> Result<RuntimeState, StoreError>;
    }

    impl InspectableStoryStore for InMemoryStoryStore {
        fn inspected_branch_head(
            &self,
            story_id: &str,
            branch_id: &str,
        ) -> Result<Option<String>, StoreError> {
            self.branch_head(story_id, branch_id)
        }

        fn inspected_state_at_node(
            &self,
            story_id: &str,
            node_id: &str,
        ) -> Result<RuntimeState, StoreError> {
            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<Option<String>, StoreError> {
            self.branch_head(story_id, branch_id)
        }

        fn inspected_state_at_node(
            &self,
            story_id: &str,
            node_id: &str,
        ) -> Result<RuntimeState, StoreError> {
            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"),
            )
            .expect("root append");
        store
            .append_node(
                &node("node_002", Some("node_001"), "branch_main"),
                &state("node_002", "branch_main"),
            )
            .expect("child append");

        assert_eq!(
            store
                .load_state("story_demo", "branch_main")
                .expect("stored state")
                .current_node,
            "node_002"
        );
        assert_eq!(
            store
                .inspected_state_at_node("story_demo", "node_001")
                .expect("root state")
                .current_node,
            "node_001"
        );
        assert_eq!(
            store
                .load_node("story_demo", "node_002")
                .expect("stored node")
                .parent_id
                .as_deref(),
            Some("node_001")
        );
    }

    fn assert_forks_from_an_old_node(store: &impl InspectableStoryStore) {
        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("main append");
        store
            .append_node(
                &node("node_003", Some("node_001"), "branch_rewind"),
                &state("node_003", "branch_rewind"),
            )
            .expect("fork append");

        assert_eq!(
            store
                .inspected_branch_head("story_demo", "branch_main")
                .expect("main head"),
            Some("node_002".to_owned())
        );
        assert_eq!(
            store
                .inspected_branch_head("story_demo", "branch_rewind")
                .expect("fork head"),
            Some("node_003".to_owned())
        );
    }

    fn assert_rejects_a_stale_append(store: &impl InspectableStoryStore) {
        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("main append");

        assert_eq!(
            store.append_node(
                &node("node_stale", Some("node_001"), "branch_main"),
                &state("node_stale", "branch_main"),
            ),
            Err(StoreError::StaleBranchHead {
                expected: "node_002".to_owned(),
                actual: "node_001".to_owned(),
            })
        );
        assert_eq!(
            store
                .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()))
        );
    }

    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(&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
                .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(_))
        ));
    }

    #[test]
    fn sqlite_rejects_valid_state_json_with_a_mismatched_hash() {
        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");
        let mut corrupted = state("node_001", "branch_main");
        corrupted
            .world_flags
            .insert("silently_changed".to_owned(), true);
        let corrupted_json = serde_json::to_string(&corrupted).expect("corrupt test JSON");
        store
            .connection
            .lock()
            .expect("SQLite connection lock")
            .execute(
                "UPDATE materialized_states SET state_json = ?1",
                params![corrupted_json],
            )
            .expect("corrupt test state");

        assert_eq!(
            store.load_state("story_demo", "branch_main"),
            Err(StoreError::StateHashMismatch {
                node_id: "node_001".to_owned()
            })
        );
    }

    #[test]
    fn sqlite_rejects_node_json_that_disagrees_with_structure_columns() {
        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");
        let mut corrupted = node("node_001", None, "branch_main");
        corrupted.branch_id = "hidden_branch".to_owned();
        let corrupted_json = serde_json::to_string(&corrupted).expect("corrupt node JSON");
        store
            .connection
            .lock()
            .expect("SQLite connection lock")
            .execute(
                "UPDATE nodes SET node_json = ?1",
                params![corrupted_json],
            )
            .expect("corrupt test node");

        assert_eq!(
            store.load_node("story_demo", "node_001"),
            Err(StoreError::StateMismatch("current_branch"))
        );
    }

    #[test]
    fn sqlite_rejects_a_branch_head_pointing_at_another_branch() {
        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
            .append_node(
                &node("node_other", Some("node_001"), "branch_other"),
                &state("node_other", "branch_other"),
            )
            .expect("other branch append");
        store
            .connection
            .lock()
            .expect("SQLite connection lock")
            .execute(
                "UPDATE branch_heads
                 SET head_node_id = 'node_other'
                 WHERE story_id = 'story_demo' AND branch_id = 'branch_main'",
                [],
            )
            .expect("corrupt branch head");

        assert_eq!(
            store.load_state("story_demo", "branch_main"),
            Err(StoreError::BranchNotFound {
                story_id: "story_demo".to_owned(),
                branch_id: "branch_main".to_owned()
            })
        );
    }

    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"
        );
    }
}