use std::future::Future;
use std::pin::Pin;
use crate::error::{DbError, Result};
use crate::schema::ddl::*;
pub const SCHEMA_VERSION: u32 = 19;
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)),
},
Step {
from: 9,
to: 10,
name: "concepts-log-insert-marker-gated",
suspends_foreign_keys: false,
apply: |conn| Box::pin(gate_concepts_log_insert_on_marker(conn)),
},
Step {
from: 8,
to: 9,
name: "concepts-guard-marker-gated",
suspends_foreign_keys: false,
apply: |conn| Box::pin(gate_concepts_guard_on_marker(conn)),
},
Step {
from: 12,
to: 13,
name: "branches-archive-gate",
suspends_foreign_keys: false,
apply: |conn| Box::pin(gate_branches_delete_guard(conn)),
},
Step {
from: 13,
to: 14,
name: "lineage-cut-index",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_lineage_cut_index(conn)),
},
Step {
from: 17,
to: 18,
name: "branch-partition-indices",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_branch_partition_indices(conn)),
},
Step {
from: 18,
to: 19,
name: "single-open-plan-pin",
suspends_foreign_keys: false,
apply: |conn| Box::pin(pin_single_open_plan(conn)),
},
Step {
from: 16,
to: 17,
name: "txlog-fold-partition-index",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_fold_partition_index(conn)),
},
Step {
from: 15,
to: 16,
name: "log-integrity-bit",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_log_integrity(conn)),
},
Step {
from: 14,
to: 15,
name: "links-lineage-key",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_links_lineage_key(conn)),
},
Step {
from: 11,
to: 12,
name: "branch-storage",
suspends_foreign_keys: true,
apply: |conn| Box::pin(add_branch_storage(conn)),
},
Step {
from: 10,
to: 11,
name: "links-archive-indices",
suspends_foreign_keys: false,
apply: |conn| Box::pin(add_links_archive_indices(conn)),
},
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
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_BRANCHES_TABLE, ()).await?;
seed_root_branch(conn).await?;
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_LOG_INTEGRITY_TABLE, ()).await?;
conn.execute(SEED_LOG_INTEGRITY, ()).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_log_integrity(conn: &libsql::Connection) -> Result<()> {
conn.execute(CREATE_LOG_INTEGRITY_TABLE, ()).await?;
conn.execute(SEED_LOG_INTEGRITY, ()).await?;
conn.execute(CREATE_TXLOG_MARK_GAP, ()).await?;
Ok(())
}
async fn add_branch_storage(conn: &libsql::Connection) -> Result<()> {
conn.execute(CREATE_BRANCHES_TABLE, ()).await?;
seed_root_branch(conn).await?;
for table in ["concepts", "links", "transaction_log"] {
conn.execute(
&format!("ALTER TABLE {table} ADD COLUMN {BRANCH_COLUMN}"),
(),
)
.await?;
}
conn.execute("DROP TABLE links_current", ()).await?;
conn.execute(CREATE_LINKS_CURRENT_TABLE, ()).await?;
conn.execute(LC_TRAVERSAL_COVER, ()).await?;
conn.execute(LC_OPEN_INTERVAL, ()).await?;
for (name, ddl) in [
("trg_concepts_log_insert", CREATE_CONCEPTS_LOG_INSERT),
("trg_concepts_log_update", CREATE_CONCEPTS_LOG_UPDATE),
("trg_links_log_insert", CREATE_LINKS_LOG_INSERT),
("trg_links_current_sync", CREATE_LINKS_CURRENT_SYNC),
("trg_links_single_open", CREATE_LINKS_SINGLE_OPEN),
] {
conn.execute(&format!("DROP TRIGGER IF EXISTS {name}"), ())
.await?;
conn.execute(ddl, ()).await?;
}
for ddl in [
CREATE_CONCEPTS_GUARD_LINEAGE,
CREATE_CONCEPTS_GUARD_BRANCH,
CREATE_BRANCHES_GUARD_UPDATE,
CREATE_BRANCHES_GUARD_DELETE,
] {
conn.execute(ddl, ()).await?;
}
crate::integrity::rebuild::rebuild_within(conn, crate::integrity::rebuild::Verify::Yes).await?;
Ok(())
}
async fn seed_root_branch(conn: &libsql::Connection) -> Result<()> {
let now = crate::util::timestamp::format(std::time::SystemTime::now());
conn.execute(SEED_MAIN_BRANCH, libsql::params![now]).await?;
Ok(())
}
const V12_TRIGGERS: &[&str] = &[
"trg_concepts_cross_lineage",
"trg_concepts_branch_immutable",
"trg_branches_frozen_update",
"trg_branches_frozen_delete",
"trg_links_current_sync",
"trg_links_single_open",
"trg_concepts_log_insert",
"trg_concepts_log_update",
"trg_links_log_insert",
];
const LATER_TRIGGERS: &[&str] = &["trg_txlog_mark_gap"];
const TRIGGERS_V11: &[&str] = &[
r#"
CREATE TRIGGER IF NOT EXISTS trg_links_current_sync
AFTER INSERT ON links
BEGIN
INSERT INTO links_current
(source_id, target_id, edge_type, valid_from, valid_to,
weight, properties, recorded_at)
VALUES
(NEW.source_id, NEW.target_id, NEW.edge_type, NEW.valid_from,
NEW.valid_to, NEW.weight, NEW.properties, NEW.recorded_at)
ON CONFLICT(source_id, target_id, edge_type, valid_from) DO UPDATE SET
valid_to = excluded.valid_to,
weight = excluded.weight,
properties = excluded.properties,
recorded_at = excluded.recorded_at
WHERE excluded.recorded_at > links_current.recorded_at;
END;
"#,
r#"
CREATE TRIGGER IF NOT EXISTS trg_links_single_open
BEFORE INSERT ON links
WHEN NEW.valid_to = '9999-12-31T23:59:59.999999Z'
AND EXISTS (
SELECT 1 FROM links_current
WHERE source_id = NEW.source_id
AND target_id = NEW.target_id
AND edge_type = NEW.edge_type
AND valid_from <> NEW.valid_from
AND valid_to = '9999-12-31T23:59:59.999999Z'
)
BEGIN
SELECT RAISE(ABORT, 'macrame: edge already has an open interval; retire it first');
END;
"#,
concat!(
r#"
CREATE TRIGGER IF NOT EXISTS trg_concepts_log_insert
AFTER INSERT ON concepts
WHEN NOT EXISTS (
SELECT 1 FROM sqlite_master
WHERE type = 'table' AND name = '"#,
"macrame_archive_session",
r#"'
)
BEGIN
INSERT INTO transaction_log (table_name, entity_id, operation, payload, recorded_at)
VALUES ('concepts', NEW.id, 'I',
json_object('v', 2, 'title', NEW.title, 'content', NEW.content,
'valid_from', NEW.valid_from, 'valid_to', NEW.valid_to,
'retired', NEW.retired,
'embedding_model', NEW.embedding_model),
NEW.recorded_at);
END;
"#
),
r#"
CREATE TRIGGER IF NOT EXISTS trg_concepts_log_update
AFTER UPDATE ON concepts
BEGIN
INSERT INTO transaction_log (table_name, entity_id, operation, payload, recorded_at)
VALUES ('concepts', NEW.id, 'U',
json_object('v', 2, 'title', NEW.title, 'content', NEW.content,
'valid_from', NEW.valid_from, 'valid_to', NEW.valid_to,
'retired', NEW.retired,
'embedding_model', NEW.embedding_model),
NEW.recorded_at);
END;
"#,
r#"
CREATE TRIGGER IF NOT EXISTS trg_links_log_insert
AFTER INSERT ON links
BEGIN
INSERT INTO transaction_log (table_name, entity_id, operation, payload, recorded_at)
VALUES ('links',
NEW.source_id || '|' || NEW.target_id || '|' || NEW.edge_type || '|' || NEW.valid_from,
'I',
json_object('v', 1, 'source_id', NEW.source_id, 'target_id', NEW.target_id,
'edge_type', NEW.edge_type, 'valid_from', NEW.valid_from,
'valid_to', NEW.valid_to, 'weight', NEW.weight,
'properties', json(NEW.properties)),
NEW.recorded_at);
END;
"#,
];
fn triggers_before_v12() -> impl Iterator<Item = &'static str> {
CREATE_TRIGGERS
.iter()
.copied()
.filter(|t| {
!V12_TRIGGERS
.iter()
.chain(LATER_TRIGGERS.iter())
.any(|name| t.contains(name))
})
.chain(TRIGGERS_V11.iter().copied())
}
async fn add_concepts_fts(conn: &libsql::Connection) -> Result<()> {
conn.execute(CREATE_CONCEPTS_FTS, ()).await?;
for trigger_ddl in triggers_before_v12() {
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?;
Ok(())
}
async fn gate_branches_delete_guard(conn: &libsql::Connection) -> Result<()> {
conn.execute("DROP TRIGGER IF EXISTS trg_branches_frozen_delete", ())
.await?;
conn.execute(CREATE_BRANCHES_GUARD_DELETE, ()).await?;
Ok(())
}
async fn add_links_archive_indices(conn: &libsql::Connection) -> Result<()> {
create_indices(conn, &["idx_links_recorded_at", "idx_links_target"]).await
}
async fn add_lineage_cut_index(conn: &libsql::Connection) -> Result<()> {
create_indices(conn, &["idx_lc_lineage_cut"]).await
}
async fn add_fold_partition_index(conn: &libsql::Connection) -> Result<()> {
create_indices(conn, &["idx_txlog_fold_partition"]).await
}
async fn add_branch_partition_indices(conn: &libsql::Connection) -> Result<()> {
create_indices(conn, &["idx_links_branch", "idx_txlog_branch"]).await
}
async fn pin_single_open_plan(conn: &libsql::Connection) -> Result<()> {
conn.execute("DROP TRIGGER IF EXISTS trg_links_single_open", ())
.await?;
conn.execute(CREATE_LINKS_SINGLE_OPEN, ()).await?;
Ok(())
}
const LINKS_V15: &str = r#"
CREATE TABLE links_v15 (
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 '{}',
branch_id TEXT NOT NULL DEFAULT 'main' REFERENCES branches(branch_id),
PRIMARY KEY (source_id, target_id, edge_type, valid_from, recorded_at, branch_id),
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)
)
"#;
const LINKS_TRIGGERS_V15: &[&str] = &[
"trg_links_current_sync",
"trg_links_single_open",
"trg_links_log_insert",
"trg_links_guard_delete",
];
async fn add_links_lineage_key(conn: &libsql::Connection) -> Result<()> {
conn.execute(LINKS_V15, ()).await?;
conn.execute(
"INSERT INTO links_v15 (source_id, target_id, edge_type, valid_from, \
recorded_at, valid_to, weight, properties, branch_id) \
SELECT source_id, target_id, edge_type, valid_from, recorded_at, \
valid_to, weight, properties, branch_id FROM links",
(),
)
.await?;
conn.execute("DROP TABLE links", ()).await?;
conn.execute("ALTER TABLE links_v15 RENAME TO links", ())
.await?;
for ddl in [
CREATE_LINKS_CURRENT_SYNC,
CREATE_LINKS_SINGLE_OPEN,
CREATE_LINKS_LOG_INSERT,
CREATE_LINKS_GUARD_DELETE,
] {
conn.execute(ddl, ()).await?;
}
debug_assert_eq!(LINKS_TRIGGERS_V15.len(), 4);
create_indices(conn, &["idx_links_recorded_at", "idx_links_target"]).await
}
async fn add_open_interval_index(conn: &libsql::Connection) -> Result<()> {
create_indices(conn, &["idx_lc_open_interval"]).await
}
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 triggers_before_v12() {
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",
];
const CONCEPTS_GUARD_DELETE_V8: &str = r#"
CREATE TRIGGER IF NOT EXISTS trg_concepts_guard_delete
BEFORE DELETE ON concepts
BEGIN
SELECT RAISE(ABORT, 'macrame: concepts are never physically archived (D-022)');
END;
"#;
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 triggers_before_v12() {
conn.execute(trigger_ddl, ()).await?;
}
conn.execute("DROP TRIGGER IF EXISTS trg_concepts_guard_delete", ())
.await?;
conn.execute(CONCEPTS_GUARD_DELETE_V8, ()).await?;
conn.execute("DROP TRIGGER IF EXISTS trg_concepts_log_insert", ())
.await?;
conn.execute(CONCEPTS_LOG_INSERT_V9, ()).await?;
conn.execute(REBUILD_CONCEPTS_FTS, ()).await?;
Ok(())
}
const CONCEPTS_LOG_INSERT_V9: &str = r#"
CREATE TRIGGER IF NOT EXISTS trg_concepts_log_insert
AFTER INSERT ON concepts
BEGIN
INSERT INTO transaction_log (table_name, entity_id, operation, payload, recorded_at)
VALUES ('concepts', NEW.id, 'I',
json_object('v', 2, 'title', NEW.title, 'content', NEW.content,
'valid_from', NEW.valid_from, 'valid_to', NEW.valid_to,
'retired', NEW.retired,
'embedding_model', NEW.embedding_model),
NEW.recorded_at);
END;
"#;
async fn gate_concepts_log_insert_on_marker(conn: &libsql::Connection) -> Result<()> {
conn.execute("DROP TRIGGER IF EXISTS trg_concepts_log_insert", ())
.await?;
conn.execute(CREATE_CONCEPTS_LOG_INSERT, ()).await?;
Ok(())
}
async fn gate_concepts_guard_on_marker(conn: &libsql::Connection) -> Result<()> {
conn.execute("DROP TRIGGER IF EXISTS trg_concepts_guard_delete", ())
.await?;
conn.execute(CREATE_CONCEPTS_GUARD_DELETE, ()).await?;
Ok(())
}
async fn create_indices(conn: &libsql::Connection, names: &[&str]) -> Result<()> {
for name in names {
let ddl = CREATE_INDICES
.iter()
.find(|sql| sql.contains(name))
.unwrap_or_else(|| panic!("{name} is not declared in ddl::CREATE_INDICES"));
conn.execute(ddl, ()).await?;
}
Ok(())
}
async fn add_traversal_cover(conn: &libsql::Connection) -> Result<()> {
create_indices(conn, &["idx_lc_traversal_cover"]).await?;
conn.execute("DROP INDEX IF EXISTS idx_lc_src_active", ())
.await?;
Ok(())
}
pub(crate) const BASELINE_TABLES: &[&str] = &[
"branches",
"concepts",
"links",
"links_current",
"transaction_log",
"analytics_annotations",
"concepts_fts",
"log_integrity",
];
async fn verify(conn: &libsql::Connection) -> Result<()> {
let mut rows = conn
.query(
"SELECT type, name, COALESCE(sql, '') FROM sqlite_master \
WHERE type IN ('table','trigger','index')",
(),
)
.await?;
let mut present: Vec<(String, String)> = Vec::new();
let mut bodies: Vec<(String, String)> = Vec::new();
let mut links_sql = String::new();
while let Some(row) = rows.next().await? {
let (kind, name, sql): (String, String, String) = (row.get(0)?, row.get(1)?, row.get(2)?);
if kind == "table" && name.eq_ignore_ascii_case("links") {
links_sql = sql.clone();
}
if kind == "trigger" {
bodies.push((name.clone(), sql));
}
present.push((kind, name));
}
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(", ")
),
});
}
let keyed_by_lineage = links_sql
.split_once("PRIMARY KEY")
.and_then(|(_, rest)| rest.split_once(')'))
.is_some_and(|(key, _)| key.contains("branch_id"));
if !links_sql.is_empty() && !keyed_by_lineage {
return Err(DbError::Migration {
to: SCHEMA_VERSION,
reason: format!(
"schema verification failed: the database is stamped \
v{SCHEMA_VERSION} but `links` is not keyed by lineage. Its \
primary key must end in `branch_id` (v15); without it a batch \
asserting one edge key on two lineages collides, because the \
batch paths share one `recorded_at` by contract. Upgrading \
through the ladder rebuilds the table."
),
});
}
let ungated: Vec<&str> = DELETE_GUARDS
.iter()
.filter(|name| {
bodies
.iter()
.find(|(n, _)| n.eq_ignore_ascii_case(name))
.is_none_or(|(_, sql)| !sql.contains(ARCHIVE_SESSION_MARKER))
})
.copied()
.collect();
if !ungated.is_empty() {
return Err(DbError::Migration {
to: SCHEMA_VERSION,
reason: format!(
"schema verification failed: the database is stamped v{SCHEMA_VERSION} \
but {} delete guard(s) do not probe the archive-session marker: {}. \
A guard with the right name and a pre-v9 body refuses archival it \
should permit; upgrading through the ladder replaces it.",
ungated.len(),
ungated.join(", ")
),
});
}
if has("table", ARCHIVE_SESSION_MARKER) {
return Err(DbError::ArchiveSessionLeaked {
marker: ARCHIVE_SESSION_MARKER.to_string(),
});
}
Ok(())
}
const DELETE_GUARDS: &[&str] = &[
"trg_concepts_guard_delete",
"trg_links_guard_delete",
"trg_txlog_guard_delete",
"trg_branches_frozen_delete",
];
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 = 2;
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,
"the incremental ladder stops at v{current} and this build stamps \
v{SCHEMA_VERSION}: a database stored at v{current} has no rung out \
of it and cannot be opened"
);
}
#[test]
fn every_stored_version_has_a_rung_out_of_it() {
for v in 2..SCHEMA_VERSION {
assert!(
STEPS.iter().any(|s| s.from == v),
"no rung leads out of v{v}, so a database stored at v{v} cannot \
be opened by this build"
);
}
}
#[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"
);
}
}