use std::future::Future;
use std::pin::Pin;
use crate::error::{DbError, Result};
use crate::schema::ddl::*;
pub const SCHEMA_VERSION: u32 = 7;
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>,
}
const STEPS: &[Step] = &[
Step {
from: 0,
to: SCHEMA_VERSION,
name: "baseline-0.5.4",
apply: |conn| Box::pin(baseline(conn)),
},
Step {
from: 2,
to: 3,
name: "analytics-annotations",
apply: |conn| Box::pin(add_analytics_annotations(conn)),
},
Step {
from: 3,
to: 4,
name: "traversal-covering-index",
apply: |conn| Box::pin(add_traversal_cover(conn)),
},
Step {
from: 4,
to: 5,
name: "concepts-fts",
apply: |conn| Box::pin(add_concepts_fts(conn)),
},
Step {
from: 5,
to: 6,
name: "single-open-interval-index",
apply: |conn| Box::pin(add_open_interval_index(conn)),
},
Step {
from: 6,
to: 7,
name: "links-weight-check",
apply: |conn| Box::pin(add_weight_check(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<()> {
let tx = conn
.transaction_with_behavior(libsql::TransactionBehavior::Immediate)
.await?;
let res: Result<()> = async {
(step.apply)(&tx).await?;
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(())
}
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"
);
}
}