klarkxy/nana-story
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

restore: import verified wave3 baseline

Browse Source
This commit is contained in:
Codex
2026-07-28 03:16:01 -04:00
parent 4157f8790d
commit cf9507a9dd
41 changed files with 12251 additions and 982 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/nana-store/src/lib.rs
+736 -79
View File
@@ -2,12 +2,18 @@ use std::{
collections::BTreeMap,
path::Path,
sync::{Mutex, MutexGuard},
time::Duration,
};
use nana_domain::{RuntimeState, StoryNode, stable_json_hash};
use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior};
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
use thiserror::Error;
const SCHEMA_VERSION: i64 = 1;
#[cfg(test)]
const BUSY_TIMEOUT_MILLIS: i64 = 5_000;
const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug, Error, PartialEq, Eq)]
pub enum StoreError {
#[error("story not found: {0}")]
@@ -62,7 +68,7 @@ struct MemoryData {
/// 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
/// `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 {
@@ -183,27 +189,33 @@ impl StoryStore for InMemoryStoryStore {
}
}
/// Durable SQLite implementation of the append-only story store.
/// 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
/// `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>,
}
#[derive(Clone, Copy)]
enum DatabaseKind {
File,
InMemory,
}
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)
Self::from_connection(connection, DatabaseKind::File)
}
/// 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)
Self::from_connection(connection, DatabaseKind::InMemory)
}
/// Returns the current head node for a branch, if the branch exists.
@@ -265,8 +277,13 @@ impl SqliteStoryStore {
Ok(state)
}
fn from_connection(connection: Connection) -> Result<Self, StoreError> {
initialize_schema(&connection)?;
fn from_connection(
mut connection: Connection,
database_kind: DatabaseKind,
) -> Result<Self, StoreError> {
configure_connection(&connection)?;
initialize_schema(&mut connection)?;
configure_journal(&connection, database_kind)?;
Ok(Self {
connection: Mutex::new(connection),
})
@@ -284,8 +301,7 @@ impl StoryStore for SqliteStoryStore {
let state_json = serde_json::to_string(state)?;
let mut connection = self.lock()?;
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let node_exists = transaction.query_row(
"SELECT EXISTS(
@@ -412,13 +428,7 @@ impl StoryStore for SqliteStoryStore {
})?;
let node = deserialize_node(&stored.3)?;
validate_loaded_node(
&node,
story_id,
&stored.0,
&stored.1,
stored.2.as_deref(),
)?;
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"))?;
@@ -452,56 +462,478 @@ impl StoryStore for SqliteStoryStore {
}
}
fn initialize_schema(connection: &Connection) -> Result<(), StoreError> {
connection.execute_batch(
"PRAGMA foreign_keys = ON;
BEGIN IMMEDIATE;
fn configure_connection(connection: &Connection) -> Result<(), StoreError> {
connection.busy_timeout(BUSY_TIMEOUT)?;
connection.execute_batch("PRAGMA foreign_keys = ON;")?;
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;",
)?;
let foreign_keys_enabled =
connection.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, bool>(0))?;
if !foreign_keys_enabled {
return Err(connection_configuration(
"foreign key enforcement could not be enabled",
));
}
Ok(())
}
fn validate_materialized_state(
node: &StoryNode,
state: &RuntimeState,
fn configure_journal(
connection: &Connection,
database_kind: DatabaseKind,
) -> Result<(), StoreError> {
if matches!(database_kind, DatabaseKind::File) {
let journal_mode = connection.query_row("PRAGMA journal_mode = WAL", [], |row| {
row.get::<_, String>(0)
})?;
if !journal_mode.eq_ignore_ascii_case("wal") {
return Err(connection_configuration(&format!(
"file database refused WAL journal mode and selected {journal_mode}"
)));
}
}
// NORMAL preserves WAL's crash-safety guarantees while avoiding a full
// filesystem sync on every commit. For in-memory databases this is a
// harmless connection-local setting; WAL itself is intentionally skipped
// because SQLite keeps their journal mode as MEMORY.
connection.execute_batch("PRAGMA synchronous = NORMAL;")?;
Ok(())
}
fn initialize_schema(connection: &mut Connection) -> Result<(), StoreError> {
match schema_version(connection)? {
SCHEMA_VERSION => validate_schema(connection, SCHEMA_VERSION),
0 => initialize_unversioned_schema(connection),
found => Err(unsupported_schema_version(found)),
}
}
fn initialize_unversioned_schema(connection: &mut Connection) -> Result<(), StoreError> {
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let version = schema_version(&transaction)?;
// Another connection may have initialized the file while this connection
// waited for the write lock.
if version == SCHEMA_VERSION {
validate_schema(&transaction, SCHEMA_VERSION)?;
transaction.commit()?;
return Ok(());
}
if version != 0 {
return Err(unsupported_schema_version(version));
}
if schema_has_user_objects(&transaction)? {
// Wave 2 databases have this exact unversioned layout. Validate every
// required table, column, foreign key, and index before adopting them;
// a partial legacy database must never be repaired with IF NOT EXISTS.
validate_schema(&transaction, 0)?;
} else {
transaction.execute_batch(
"CREATE TABLE 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 one_root_per_story
ON nodes (story_id)
WHERE parent_id IS NULL;
CREATE TABLE 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 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
);",
)?;
validate_schema(&transaction, SCHEMA_VERSION)?;
}
transaction.execute_batch(&format!("PRAGMA user_version = {SCHEMA_VERSION};"))?;
transaction.commit()?;
Ok(())
}
fn schema_version(connection: &Connection) -> Result<i64, StoreError> {
connection
.query_row("PRAGMA user_version", [], |row| row.get(0))
.map_err(StoreError::from)
}
fn schema_has_user_objects(connection: &Connection) -> Result<bool, StoreError> {
connection
.query_row(
"SELECT EXISTS(
SELECT 1
FROM sqlite_schema
WHERE name NOT LIKE 'sqlite_%'
AND type IN ('table', 'index', 'view', 'trigger')
)",
[],
|row| row.get(0),
)
.map_err(StoreError::from)
}
#[derive(Debug, PartialEq, Eq)]
struct ColumnLayout {
name: String,
declared_type: String,
not_null: bool,
primary_key_position: i64,
}
#[derive(Debug, PartialEq, Eq)]
struct ForeignKeyLayout {
sequence: i64,
referenced_table: String,
from_column: String,
to_column: String,
on_delete: String,
}
struct ExpectedColumn<'a> {
name: &'a str,
declared_type: &'a str,
not_null: bool,
primary_key_position: i64,
}
struct ExpectedForeignKey<'a> {
sequence: i64,
referenced_table: &'a str,
from_column: &'a str,
to_column: &'a str,
on_delete: &'a str,
}
fn validate_schema(connection: &Connection, version: i64) -> Result<(), StoreError> {
validate_nodes_schema(connection, version)?;
validate_materialized_states_schema(connection, version)?;
validate_branch_heads_schema(connection, version)?;
validate_root_index(connection, version)?;
Ok(())
}
fn validate_nodes_schema(connection: &Connection, version: i64) -> Result<(), StoreError> {
validate_table(
connection,
version,
"nodes",
&[
ExpectedColumn {
name: "story_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 1,
},
ExpectedColumn {
name: "node_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 2,
},
ExpectedColumn {
name: "branch_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 0,
},
ExpectedColumn {
name: "parent_id",
declared_type: "TEXT",
not_null: false,
primary_key_position: 0,
},
ExpectedColumn {
name: "node_json",
declared_type: "TEXT",
not_null: true,
primary_key_position: 0,
},
],
&[
ExpectedForeignKey {
sequence: 0,
referenced_table: "nodes",
from_column: "story_id",
to_column: "story_id",
on_delete: "RESTRICT",
},
ExpectedForeignKey {
sequence: 1,
referenced_table: "nodes",
from_column: "parent_id",
to_column: "node_id",
on_delete: "RESTRICT",
},
],
)
}
fn validate_materialized_states_schema(
connection: &Connection,
version: i64,
) -> Result<(), StoreError> {
validate_table(
connection,
version,
"materialized_states",
&[
ExpectedColumn {
name: "story_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 1,
},
ExpectedColumn {
name: "node_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 2,
},
ExpectedColumn {
name: "state_json",
declared_type: "TEXT",
not_null: true,
primary_key_position: 0,
},
],
&[
ExpectedForeignKey {
sequence: 0,
referenced_table: "nodes",
from_column: "story_id",
to_column: "story_id",
on_delete: "RESTRICT",
},
ExpectedForeignKey {
sequence: 1,
referenced_table: "nodes",
from_column: "node_id",
to_column: "node_id",
on_delete: "RESTRICT",
},
],
)
}
fn validate_branch_heads_schema(connection: &Connection, version: i64) -> Result<(), StoreError> {
validate_table(
connection,
version,
"branch_heads",
&[
ExpectedColumn {
name: "story_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 1,
},
ExpectedColumn {
name: "branch_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 2,
},
ExpectedColumn {
name: "head_node_id",
declared_type: "TEXT",
not_null: true,
primary_key_position: 0,
},
],
&[
ExpectedForeignKey {
sequence: 0,
referenced_table: "nodes",
from_column: "story_id",
to_column: "story_id",
on_delete: "RESTRICT",
},
ExpectedForeignKey {
sequence: 1,
referenced_table: "nodes",
from_column: "head_node_id",
to_column: "node_id",
on_delete: "RESTRICT",
},
],
)
}
fn validate_table(
connection: &Connection,
version: i64,
table: &str,
expected_columns: &[ExpectedColumn<'_>],
expected_foreign_key: &[ExpectedForeignKey<'_>],
) -> Result<(), StoreError> {
let columns = table_columns(connection, table)?;
let columns_match = columns.len() == expected_columns.len()
&& columns
.iter()
.zip(expected_columns)
.all(|(actual, expected)| {
actual.name == expected.name
&& actual.declared_type == expected.declared_type
&& actual.not_null == expected.not_null
&& actual.primary_key_position == expected.primary_key_position
});
if !columns_match {
return Err(incomplete_schema(
version,
&format!("table {table} has an unexpected column layout"),
));
}
let foreign_keys = table_foreign_keys(connection, table)?;
let foreign_keys_match = foreign_keys.len() == expected_foreign_key.len()
&& foreign_keys
.iter()
.zip(expected_foreign_key)
.all(|(actual, expected)| {
actual.sequence == expected.sequence
&& actual.referenced_table == expected.referenced_table
&& actual.from_column == expected.from_column
&& actual.to_column == expected.to_column
&& actual.on_delete == expected.on_delete
});
if !foreign_keys_match {
return Err(incomplete_schema(
version,
&format!("table {table} has an unexpected foreign key layout"),
));
}
Ok(())
}
fn table_columns(connection: &Connection, table: &str) -> Result<Vec<ColumnLayout>, StoreError> {
let mut statement = connection.prepare(&format!("PRAGMA table_info('{table}')"))?;
let rows = statement.query_map([], |row| {
Ok(ColumnLayout {
name: row.get(1)?,
declared_type: row.get(2)?,
not_null: row.get::<_, i64>(3)? != 0,
primary_key_position: row.get(5)?,
})
})?;
rows.collect::<Result<Vec<_>, _>>()
.map_err(StoreError::from)
}
fn table_foreign_keys(
connection: &Connection,
table: &str,
) -> Result<Vec<ForeignKeyLayout>, StoreError> {
let mut statement = connection.prepare(&format!("PRAGMA foreign_key_list('{table}')"))?;
let rows = statement.query_map([], |row| {
Ok(ForeignKeyLayout {
sequence: row.get(1)?,
referenced_table: row.get(2)?,
from_column: row.get(3)?,
to_column: row.get(4)?,
on_delete: row.get(6)?,
})
})?;
rows.collect::<Result<Vec<_>, _>>()
.map_err(StoreError::from)
}
fn validate_root_index(connection: &Connection, version: i64) -> Result<(), StoreError> {
let index_layout = connection
.query_row(
"SELECT pragma_index_list.[unique], pragma_index_list.partial, sqlite_schema.sql
FROM pragma_index_list('nodes')
JOIN sqlite_schema
ON sqlite_schema.type = 'index'
AND sqlite_schema.name = pragma_index_list.name
WHERE pragma_index_list.name = 'one_root_per_story'",
[],
|row| {
Ok((
row.get::<_, bool>(0)?,
row.get::<_, bool>(1)?,
row.get::<_, String>(2)?,
))
},
)
.optional()?;
let Some((is_unique, is_partial, index_sql)) = index_layout else {
return Err(incomplete_schema(
version,
"unique partial index one_root_per_story is missing or incompatible",
));
};
let normalized_sql = index_sql
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.to_ascii_lowercase();
if !is_unique
|| !is_partial
|| normalized_sql
!= "create unique index one_root_per_story on nodes (story_id) where parent_id is null"
{
return Err(incomplete_schema(
version,
"unique partial index one_root_per_story is missing or incompatible",
));
}
let mut statement = connection
.prepare("SELECT name FROM pragma_index_info('one_root_per_story') ORDER BY seqno")?;
let columns = statement
.query_map([], |row| row.get::<_, String>(0))?
.collect::<Result<Vec<_>, _>>()?;
if columns != vec!["story_id".to_owned()] {
return Err(incomplete_schema(
version,
"index one_root_per_story targets unexpected columns",
));
}
Ok(())
}
fn incomplete_schema(version: i64, detail: &str) -> StoreError {
StoreError::Sqlite(format!(
"story database schema version {version} is incomplete or incompatible: {detail}"
))
}
fn unsupported_schema_version(found: i64) -> StoreError {
StoreError::Sqlite(format!(
"unsupported story database schema version {found}; newest supported version is \
{SCHEMA_VERSION}"
))
}
fn connection_configuration(detail: &str) -> StoreError {
StoreError::Sqlite(format!(
"SQLite connection is not safely configured: {detail}"
))
}
fn validate_materialized_state(node: &StoryNode, state: &RuntimeState) -> Result<(), StoreError> {
if node.story_id != state.story_id {
return Err(StoreError::StateMismatch("story_id"));
}
@@ -582,10 +1014,15 @@ mod tests {
time::{SystemTime, UNIX_EPOCH},
};
use nana_domain::{RuntimeState, StateDelta, StoryNode, stable_json_hash};
use rusqlite::params;
use nana_domain::{
PresentationSnapshot, RuntimeState, StateDelta, StoryNode, stable_json_hash,
};
use rusqlite::{Connection, params};
use super::{InMemoryStoryStore, SqliteStoryStore, StoreError, StoryStore};
use super::{
BUSY_TIMEOUT_MILLIS, InMemoryStoryStore, SCHEMA_VERSION, SqliteStoryStore, StoreError,
StoryStore,
};
fn state(node: &str, branch: &str) -> RuntimeState {
RuntimeState {
@@ -612,7 +1049,7 @@ mod tests {
parent_id: parent_id.map(ToOwned::to_owned),
action_id: format!("action_{id}"),
user_input: String::new(),
beats: Vec::new(),
presentation: PresentationSnapshot::default(),
delta: StateDelta { ops: Vec::new() },
state_hash: stable_json_hash(
&serde_json::to_vec(&materialized).expect("serializable test state"),
@@ -857,9 +1294,7 @@ mod tests {
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");
store.append_node(&root, &root_state).expect("root append");
assert_eq!(
store.append_node(&root, &root_state),
@@ -968,10 +1403,7 @@ mod tests {
.connection
.lock()
.expect("SQLite connection lock")
.execute(
"UPDATE nodes SET node_json = ?1",
params![corrupted_json],
)
.execute("UPDATE nodes SET node_json = ?1", params![corrupted_json])
.expect("corrupt test node");
assert_eq!(
@@ -1027,10 +1459,8 @@ mod tests {
.unwrap_or_default()
.as_nanos();
Self {
path: std::env::temp_dir().join(format!(
"nana-store-{}-{nonce}.sqlite3",
std::process::id()
)),
path: std::env::temp_dir()
.join(format!("nana-store-{}-{nonce}.sqlite3", std::process::id())),
}
}
@@ -1041,12 +1471,230 @@ mod tests {
impl Drop for TemporaryDatabase {
fn drop(&mut self) {
if self.path.exists() {
fs::remove_file(&self.path).expect("remove temporary SQLite database");
for suffix in ["", "-wal", "-shm"] {
let mut path = self.path.as_os_str().to_os_string();
path.push(suffix);
let path = PathBuf::from(path);
if path.exists() {
fs::remove_file(path).expect("remove temporary SQLite database file");
}
}
}
}
#[test]
fn sqlite_initializes_a_new_database_with_version_and_safe_pragmas() {
let database = TemporaryDatabase::new();
let store = SqliteStoryStore::open(database.path()).expect("new file SQLite store");
let connection = store.connection.lock().expect("SQLite connection lock");
assert_eq!(
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.expect("schema version"),
SCHEMA_VERSION
);
assert!(
connection
.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, bool>(0))
.expect("foreign key pragma")
);
assert_eq!(
connection
.query_row("PRAGMA busy_timeout", [], |row| row.get::<_, i64>(0))
.expect("busy timeout pragma"),
BUSY_TIMEOUT_MILLIS
);
assert_eq!(
connection
.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))
.expect("journal mode")
.to_ascii_lowercase(),
"wal"
);
assert_eq!(
connection
.query_row("PRAGMA synchronous", [], |row| row.get::<_, i64>(0))
.expect("synchronous pragma"),
1
);
}
#[test]
fn sqlite_in_memory_uses_safe_pragmas_without_requesting_wal() {
let store = SqliteStoryStore::open_in_memory().expect("in-memory SQLite store");
let connection = store.connection.lock().expect("SQLite connection lock");
assert_eq!(
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.expect("schema version"),
SCHEMA_VERSION
);
assert!(
connection
.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, bool>(0))
.expect("foreign key pragma")
);
assert_eq!(
connection
.query_row("PRAGMA busy_timeout", [], |row| row.get::<_, i64>(0))
.expect("busy timeout pragma"),
BUSY_TIMEOUT_MILLIS
);
assert_eq!(
connection
.query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0))
.expect("journal mode")
.to_ascii_lowercase(),
"memory"
);
}
#[test]
fn sqlite_adopts_a_complete_unversioned_database_without_losing_data() {
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
.connection
.lock()
.expect("SQLite connection lock")
.execute_batch("PRAGMA user_version = 0;")
.expect("simulate Wave 2 database");
}
let reopened =
SqliteStoryStore::open(database.path()).expect("adopt complete legacy database");
assert_eq!(
reopened
.load_state("story_demo", "branch_main")
.expect("legacy state")
.current_node,
"node_001"
);
assert_eq!(
reopened
.connection
.lock()
.expect("SQLite connection lock")
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.expect("adopted schema version"),
SCHEMA_VERSION
);
}
#[test]
fn sqlite_rejects_an_unknown_future_schema_version_without_changing_it() {
let database = TemporaryDatabase::new();
{
let connection = Connection::open(database.path()).expect("raw SQLite database");
connection
.execute_batch("PRAGMA user_version = 99;")
.expect("future schema version");
}
assert!(matches!(
SqliteStoryStore::open(database.path()),
Err(StoreError::Sqlite(message))
if message.contains("unsupported story database schema version 99")
));
let connection = Connection::open(database.path()).expect("reopen raw SQLite database");
assert_eq!(
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.expect("future schema version"),
99
);
}
#[test]
fn sqlite_rejects_an_incomplete_unversioned_schema_without_rebuilding_it() {
let database = TemporaryDatabase::new();
{
let connection = Connection::open(database.path()).expect("raw SQLite database");
connection
.execute_batch(
"CREATE TABLE 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)
);",
)
.expect("partial legacy schema");
}
assert!(matches!(
SqliteStoryStore::open(database.path()),
Err(StoreError::Sqlite(message))
if message.contains("schema version 0 is incomplete or incompatible")
));
let connection = Connection::open(database.path()).expect("reopen raw SQLite database");
assert_eq!(
connection
.query_row(
"SELECT COUNT(*) FROM sqlite_schema
WHERE type = 'table'
AND name IN ('nodes', 'materialized_states', 'branch_heads')",
[],
|row| row.get::<_, i64>(0),
)
.expect("application table count"),
1
);
assert_eq!(
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.expect("unchanged schema version"),
0
);
}
#[test]
fn sqlite_rejects_a_damaged_current_schema_without_rebuilding_it() {
let database = TemporaryDatabase::new();
{
let store = SqliteStoryStore::open(database.path()).expect("file SQLite store");
store
.connection
.lock()
.expect("SQLite connection lock")
.execute_batch("DROP TABLE branch_heads;")
.expect("damage schema");
}
assert!(matches!(
SqliteStoryStore::open(database.path()),
Err(StoreError::Sqlite(message))
if message.contains("schema version 1 is incomplete or incompatible")
));
let connection = Connection::open(database.path()).expect("reopen raw SQLite database");
assert!(
!connection
.query_row(
"SELECT EXISTS(
SELECT 1 FROM sqlite_schema
WHERE type = 'table' AND name = 'branch_heads'
)",
[],
|row| row.get::<_, bool>(0),
)
.expect("branch_heads presence")
);
}
#[test]
fn sqlite_reopens_and_preserves_nodes_states_and_heads() {
let database = TemporaryDatabase::new();
@@ -1067,6 +1715,15 @@ mod tests {
}
let reopened = SqliteStoryStore::open(database.path()).expect("reopened SQLite store");
assert_eq!(
reopened
.connection
.lock()
.expect("SQLite connection lock")
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.expect("reopened schema version"),
SCHEMA_VERSION
);
assert_eq!(
reopened
.branch_head("story_demo", "branch_main")