use std::future::Future;
use std::pin::Pin;
use crate::error::{DbError, Result};
use crate::schema::ddl::*;
pub const SCHEMA_VERSION: u32 = 8;
type StepFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;
struct Step {
from: u32,
to: u32,
name: &'static str,
apply: for<'a> fn(&'a libsql::Connection) -> StepFuture<'a>,
suspends_foreign_keys: bool,
}
const STEPS: &[Step] = &[
Step {
from: 0,
to: SCHEMA_VERSION,
name: "baseline-0.5.4",
suspends_foreign_keys: false,
apply: |conn| Box::pin(baseline(conn)),
},
Step {
from: 2,
to: 3,
name: "analytics-annotations",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_analytics_annotations(conn)),
},
Step {
from: 3,
to: 4,
name: "traversal-covering-index",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_traversal_cover(conn)),
},
Step {
from: 4,
to: 5,
name: "concepts-fts",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_concepts_fts(conn)),
},
Step {
from: 5,
to: 6,
name: "single-open-interval-index",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_open_interval_index(conn)),
},
Step {
from: 6,
to: 7,
name: "links-weight-check",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_weight_check(conn)),
},
Step {
from: 7,
to: 8,
name: "concepts-rowid-pk-and-unread-indices",
suspends_foreign_keys: true,
apply: |conn| Box::pin(add_concepts_rowid_pk(conn)),
},
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MigrationOutcome {
pub from: u32,
pub to: u32,
}
impl MigrationOutcome {
pub fn upgraded(&self) -> bool {
self.from != 0 && self.from != self.to
}
}
pub async fn run(conn: &libsql::Connection) -> Result<MigrationOutcome> {
let found = read_user_version(conn).await?;
if found > SCHEMA_VERSION {
return Err(DbError::Migration {
to: SCHEMA_VERSION,
reason: format!(
"database is at schema v{found}; this build understands v{SCHEMA_VERSION} \
and will not operate on a schema it does not know. Upgrade macrame \
rather than opening the file with an older build."
),
});
}
if found == 0 {
refuse_if_occupied(conn).await?;
}
let mut current = found;
while current != SCHEMA_VERSION {
let step = STEPS
.iter()
.find(|s| s.from == current)
.ok_or_else(|| no_path_from(current))?;
apply_step(conn, step).await?;
current = step.to;
}
verify(conn).await?;
Ok(MigrationOutcome {
from: found,
to: SCHEMA_VERSION,
})
}
pub fn current_version() -> u32 {
SCHEMA_VERSION
}
async fn refuse_if_occupied(conn: &libsql::Connection) -> Result<()> {
let mut rows = conn
.query(
"SELECT COUNT(*) FROM sqlite_master WHERE name NOT LIKE 'sqlite_%'",
(),
)
.await?;
let objects: i64 = match rows.next().await? {
Some(row) => row.get(0)?,
None => 0,
};
if objects > 0 {
return Err(DbError::Migration {
to: SCHEMA_VERSION,
reason: format!(
"database carries no macrame schema version but already holds {objects} \
object(s); refusing to lay the baseline over an unrelated database. \
Point at a new file, or delete this one deliberately."
),
});
}
Ok(())
}
fn no_path_from(current: u32) -> DbError {
let reason = if current < SCHEMA_VERSION {
format!(
"database is at schema v{current}, written by a pre-0.5.4 build: its \
timestamps are second-precision and its tables carry none of the \
canonical-form CHECK constraints (D-029). This build provides no \
migration path — create a new database."
)
} else {
format!("no migration step leads out of schema v{current}")
};
DbError::Migration {
to: SCHEMA_VERSION,
reason,
}
}
async fn apply_step(conn: &libsql::Connection, step: &Step) -> Result<()> {
if step.suspends_foreign_keys {
conn.execute("PRAGMA foreign_keys = OFF", ()).await?;
}
let res = apply_step_inner(conn, step).await;
if step.suspends_foreign_keys {
conn.execute("PRAGMA foreign_keys = ON", ()).await?;
}
res
}
async fn apply_step_inner(conn: &libsql::Connection, step: &Step) -> Result<()> {
let tx = conn
.transaction_with_behavior(libsql::TransactionBehavior::Immediate)
.await?;
let res: Result<()> = async {
(step.apply)(&tx).await?;
if step.suspends_foreign_keys {
let mut rows = tx.query("PRAGMA foreign_key_check", ()).await?;
if let Some(row) = rows.next().await? {
let table: String = row.get(0).unwrap_or_else(|_| "?".to_string());
return Err(DbError::Migration {
to: step.to,
reason: format!(
"step {:?} suspended foreign keys and left a violation \
in {table:?}; the rung is wrong, not the check",
step.name
),
});
}
}
tx.execute(&format!("PRAGMA user_version = {}", step.to), ())
.await?;
Ok(())
}
.await;
match res {
Ok(()) => {
tx.commit().await?;
Ok(())
}
Err(e) => {
let _ = tx.rollback().await;
Err(DbError::Migration {
to: step.to,
reason: format!("step {:?}: {e}", step.name),
})
}
}
}
async fn baseline(conn: &libsql::Connection) -> Result<()> {
conn.execute(CREATE_CONCEPTS_TABLE, ()).await?;
conn.execute(CREATE_LINKS_TABLE, ()).await?;
conn.execute(CREATE_LINKS_CURRENT_TABLE, ()).await?;
conn.execute(CREATE_TRANSACTION_LOG_TABLE, ()).await?;
conn.execute(CREATE_ANALYTICS_ANNOTATIONS_TABLE, ()).await?;
conn.execute(CREATE_CONCEPTS_FTS, ()).await?;
for index_ddl in CREATE_INDICES {
conn.execute(index_ddl, ()).await?;
}
for trigger_ddl in CREATE_TRIGGERS {
conn.execute(trigger_ddl, ()).await?;
}
Ok(())
}
async fn add_concepts_fts(conn: &libsql::Connection) -> Result<()> {
conn.execute(CREATE_CONCEPTS_FTS, ()).await?;
for trigger_ddl in CREATE_TRIGGERS {
conn.execute(trigger_ddl, ()).await?;
}
conn.execute(REBUILD_CONCEPTS_FTS, ()).await?;
Ok(())
}
async fn add_analytics_annotations(conn: &libsql::Connection) -> Result<()> {
conn.execute(CREATE_ANALYTICS_ANNOTATIONS_TABLE, ()).await?;
for index_ddl in CREATE_INDICES {
conn.execute(index_ddl, ()).await?;
}
Ok(())
}
async fn add_open_interval_index(conn: &libsql::Connection) -> Result<()> {
for index_ddl in CREATE_INDICES {
conn.execute(index_ddl, ()).await?;
}
Ok(())
}
const LINKS_V7: &str = r#"
CREATE TABLE links_v7 (
source_id TEXT NOT NULL REFERENCES concepts(id),
target_id TEXT NOT NULL REFERENCES concepts(id),
edge_type TEXT NOT NULL,
valid_from TEXT NOT NULL,
recorded_at TEXT NOT NULL,
valid_to TEXT NOT NULL DEFAULT '9999-12-31T23:59:59.999999Z',
weight REAL NOT NULL DEFAULT 1.0,
properties TEXT NOT NULL DEFAULT '{}',
PRIMARY KEY (source_id, target_id, edge_type, valid_from, recorded_at),
CHECK (weight >= 0.0 AND weight < 9e999 AND typeof(weight) = 'real'),
-- (the timestamp CHECK, spelled out for the same pinning reason)
CHECK (valid_from GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9][0-9][0-9][0-9][0-9]Z' AND valid_to GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9][0-9][0-9][0-9][0-9]Z' AND recorded_at GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9][0-9][0-9][0-9][0-9]Z' AND 1)
)
"#;
async fn add_weight_check(conn: &libsql::Connection) -> Result<()> {
let offending: i64 = conn
.query(
"SELECT COUNT(*) FROM links WHERE NOT (weight >= 0.0 AND weight < 9e999 AND typeof(weight) = 'real')",
(),
)
.await?
.next()
.await?
.and_then(|r| r.get(0).ok())
.unwrap_or(0);
if offending > 0 {
let example: Option<String> = conn
.query(
"SELECT source_id || ' -> ' || target_id || ' (' || edge_type || \
') weight=' || CAST(weight AS TEXT) FROM links \
WHERE NOT (weight >= 0.0 AND weight < 9e999 AND typeof(weight) = 'real') LIMIT 1",
(),
)
.await?
.next()
.await?
.and_then(|r| r.get(0).ok());
return Err(DbError::Migration {
to: 7,
reason: format!(
"{offending} row(s) in `links` hold a weight the v7 constraint \
rejects, e.g. {}. Copying them verbatim is impossible and \
altering them would violate Doctrine III, so this migration \
refuses rather than choosing on your behalf. These rows were \
writable through `assert_edge` before v7 (§4.7) — decide what \
they were meant to assert, archive them, and retry.",
example.as_deref().unwrap_or("<unreadable>")
),
});
}
conn.execute(LINKS_V7, ()).await?;
conn.execute(
"INSERT INTO links_v7 (source_id, target_id, edge_type, valid_from, \
recorded_at, valid_to, weight, properties) \
SELECT source_id, target_id, edge_type, valid_from, recorded_at, \
valid_to, weight, properties FROM links",
(),
)
.await?;
conn.execute("DROP TABLE links", ()).await?;
conn.execute("ALTER TABLE links_v7 RENAME TO links", ())
.await?;
for trigger_ddl in CREATE_TRIGGERS {
conn.execute(trigger_ddl, ()).await?;
}
Ok(())
}
const CONCEPTS_V8: &str = r#"
CREATE TABLE concepts_v8 (
rowid_pk INTEGER PRIMARY KEY,
id TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
embedding_model TEXT,
valid_from TEXT NOT NULL,
valid_to TEXT NOT NULL DEFAULT '9999-12-31T23:59:59.999999Z',
recorded_at TEXT NOT NULL,
retired INTEGER NOT NULL DEFAULT 0,
-- (the timestamp CHECK, spelled out for the same pinning reason)
CHECK (valid_from GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9][0-9][0-9][0-9][0-9]Z' AND valid_to GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9][0-9][0-9][0-9][0-9]Z' AND recorded_at GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9][0-9]:[0-9][0-9]:[0-9][0-9].[0-9][0-9][0-9][0-9][0-9][0-9]Z' AND 1)
)
"#;
const CONCEPTS_TRIGGERS_V7: &[&str] = &[
"trg_concepts_monotonic_ra",
"trg_concepts_log_insert",
"trg_concepts_log_update",
"trg_concepts_guard_delete",
"trg_concepts_fts_insert",
"trg_concepts_fts_update",
];
async fn add_concepts_rowid_pk(conn: &libsql::Connection) -> Result<()> {
conn.execute("DROP INDEX IF EXISTS idx_annotations_label", ())
.await?;
conn.execute("DROP INDEX IF EXISTS idx_lc_tgt_active", ())
.await?;
for name in CONCEPTS_TRIGGERS_V7 {
conn.execute(&format!("DROP TRIGGER IF EXISTS {name}"), ())
.await?;
}
conn.execute("DROP TABLE IF EXISTS concepts_fts", ()).await?;
conn.execute(CONCEPTS_V8, ()).await?;
conn.execute(
"INSERT INTO concepts_v8 (rowid_pk, id, title, content, embedding_model, \
valid_from, valid_to, recorded_at, retired) \
SELECT rowid, id, title, content, embedding_model, \
valid_from, valid_to, recorded_at, retired \
FROM concepts ORDER BY rowid",
(),
)
.await?;
conn.execute("DROP TABLE concepts", ()).await?;
conn.execute("ALTER TABLE concepts_v8 RENAME TO concepts", ())
.await?;
conn.execute(CREATE_CONCEPTS_FTS, ()).await?;
for trigger_ddl in CREATE_TRIGGERS {
conn.execute(trigger_ddl, ()).await?;
}
conn.execute(REBUILD_CONCEPTS_FTS, ()).await?;
Ok(())
}
async fn add_traversal_cover(conn: &libsql::Connection) -> Result<()> {
for index_ddl in CREATE_INDICES {
conn.execute(index_ddl, ()).await?;
}
conn.execute("DROP INDEX IF EXISTS idx_lc_src_active", ())
.await?;
Ok(())
}
pub(crate) const BASELINE_TABLES: &[&str] = &[
"concepts",
"links",
"links_current",
"transaction_log",
"analytics_annotations",
"concepts_fts",
];
async fn verify(conn: &libsql::Connection) -> Result<()> {
let mut rows = conn
.query(
"SELECT type, name FROM sqlite_master WHERE type IN ('table','trigger','index')",
(),
)
.await?;
let mut present: Vec<(String, String)> = Vec::new();
while let Some(row) = rows.next().await? {
present.push((row.get(0)?, row.get(1)?));
}
let has = |kind: &str, name: &str| {
present
.iter()
.any(|(k, n)| k == kind && n.eq_ignore_ascii_case(name))
};
let mut missing: Vec<String> = Vec::new();
for table in BASELINE_TABLES {
if !has("table", table) {
missing.push(format!("table {table}"));
}
}
for name in trigger_names() {
if !has("trigger", &name) {
missing.push(format!("trigger {name}"));
}
}
for name in index_names() {
if !has("index", &name) {
missing.push(format!("index {name}"));
}
}
if !missing.is_empty() {
return Err(DbError::Migration {
to: SCHEMA_VERSION,
reason: format!(
"schema verification failed: the database is stamped v{SCHEMA_VERSION} \
but is missing {}: {}",
missing.len(),
missing.join(", ")
),
});
}
Ok(())
}
fn names_after(ddl: &[&str], keyword: &str) -> Vec<String> {
ddl.iter()
.filter_map(|stmt| {
let lower = stmt.to_ascii_lowercase();
let at = lower.find(keyword)? + keyword.len();
Some(
stmt[at..]
.split_whitespace()
.next()?
.trim_matches(|c: char| !c.is_alphanumeric() && c != '_')
.to_string(),
)
})
.filter(|n| !n.is_empty())
.collect()
}
fn trigger_names() -> Vec<String> {
names_after(CREATE_TRIGGERS, "create trigger if not exists ")
}
fn index_names() -> Vec<String> {
names_after(CREATE_INDICES, "create index if not exists ")
}
async fn read_user_version(conn: &libsql::Connection) -> Result<u32> {
let mut rows = conn.query("PRAGMA user_version", ()).await?;
match rows.next().await? {
Some(row) => Ok(row.get::<u32>(0)?),
None => Ok(0),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn every_step_advances() {
for step in STEPS {
assert!(
step.to > step.from,
"step {:?} does not advance ({} -> {})",
step.name,
step.from,
step.to
);
}
}
#[test]
fn no_two_steps_share_an_origin() {
for (i, a) in STEPS.iter().enumerate() {
for b in &STEPS[i + 1..] {
assert_ne!(a.from, b.from, "two steps start at v{}", a.from);
}
}
}
#[test]
fn the_ladder_reaches_the_current_version() {
let mut current = 0;
for _ in 0..STEPS.len() {
match STEPS.iter().find(|s| s.from == current) {
Some(step) => current = step.to,
None => break,
}
}
assert_eq!(current, SCHEMA_VERSION);
}
#[test]
fn every_trigger_and_index_yields_a_name() {
let triggers = super::trigger_names();
assert_eq!(triggers.len(), CREATE_TRIGGERS.len());
assert!(
triggers.iter().all(|n| n.starts_with("trg_")),
"{triggers:?}"
);
let indices = super::index_names();
assert_eq!(indices.len(), CREATE_INDICES.len());
assert!(indices.iter().all(|n| n.starts_with("idx_")), "{indices:?}");
}
#[test]
fn legacy_v1_has_no_rung() {
assert!(
!STEPS.iter().any(|s| s.from == 1),
"a step out of v1 reintroduces pre-0.5.4 databases as a supported input"
);
}
async fn scratch() -> libsql::Connection {
let db = libsql::Builder::new_local(":memory:").build().await.unwrap();
let conn = db.connect().unwrap();
conn.execute("PRAGMA foreign_keys = ON", ()).await.unwrap();
conn.execute("CREATE TABLE parent (id TEXT PRIMARY KEY)", ())
.await
.unwrap();
conn.execute(
"CREATE TABLE child (id TEXT PRIMARY KEY, p TEXT NOT NULL \
REFERENCES parent(id))",
(),
)
.await
.unwrap();
conn.execute("INSERT INTO parent VALUES ('a')", ())
.await
.unwrap();
conn.execute("INSERT INTO child VALUES ('c', 'a')", ())
.await
.unwrap();
conn
}
#[tokio::test]
async fn a_suspending_rung_can_rebuild_a_table_with_inbound_keys() {
let conn = scratch().await;
let step = Step {
from: 0,
to: 99,
name: "test-rebuild",
suspends_foreign_keys: true,
apply: |tx| {
Box::pin(async move {
tx.execute("CREATE TABLE parent_new (rowid_pk INTEGER PRIMARY KEY, id TEXT NOT NULL UNIQUE)", ()).await?;
tx.execute("INSERT INTO parent_new (id) SELECT id FROM parent ORDER BY rowid", ()).await?;
tx.execute("DROP TABLE parent", ()).await?;
tx.execute("ALTER TABLE parent_new RENAME TO parent", ()).await?;
Ok(())
})
},
};
apply_step(&conn, &step).await.expect("the rung must apply");
let mut rows = conn
.query("SELECT rowid_pk FROM parent WHERE id = 'a'", ())
.await
.unwrap();
assert!(rows.next().await.unwrap().is_some(), "parent lost its row");
let orphan = conn
.execute("INSERT INTO child VALUES ('d', 'nonexistent')", ())
.await;
assert!(
orphan.is_err(),
"foreign keys were not restored after the rung"
);
}
#[tokio::test]
async fn a_suspending_rung_that_leaves_a_violation_is_refused() {
let conn = scratch().await;
let step = Step {
from: 0,
to: 99,
name: "test-orphan",
suspends_foreign_keys: true,
apply: |tx| {
Box::pin(async move {
tx.execute("DROP TABLE parent", ()).await?;
tx.execute("CREATE TABLE parent (id TEXT PRIMARY KEY)", ())
.await?;
Ok(())
})
},
};
let err = apply_step(&conn, &step)
.await
.expect_err("a rung that orphans a row must not commit");
let text = err.to_string();
assert!(
text.contains("suspended foreign keys and left a violation"),
"the rung must fail at `foreign_key_check`, not merely fail: {text}"
);
assert!(text.contains("test-orphan"), "the error should name the rung: {text}");
let mut rows = conn.query("PRAGMA user_version", ()).await.unwrap();
let v: u32 = rows.next().await.unwrap().unwrap().get(0).unwrap();
assert_eq!(v, 0, "a failed rung must not stamp its version");
let orphan = conn
.execute("INSERT INTO child VALUES ('d', 'nonexistent')", ())
.await;
assert!(
orphan.is_err(),
"foreign keys must be restored even when the rung failed"
);
}
}