use std::fmt::{Display, Formatter};
use std::time::Instant;
use rusqlite::Connection;
pub const SCHEMA_VERSION: u32 = 26;
pub const PRAGMA_USER_VERSION: &str = "user_version";
pub const SQLITE_SUFFIX: &str = ".sqlite";
pub const WAL_SUFFIX: &str = "-wal";
pub const LOCK_SUFFIX: &str = ".lock";
pub const JOURNAL_SUFFIX: &str = "-journal";
#[must_use]
pub fn bootstrap_steps() -> &'static [&'static str] {
&["create canonical tables", "register projection metadata", "seed rewrite-era configuration"]
}
pub const CANONICAL_TABLES: &[&str] = &[
"canonical_nodes",
"canonical_edges",
"operational_collections",
"operational_mutations",
"operational_state",
];
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct Migration {
pub step_id: u32,
pub sql: &'static str,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MigrationStepReport {
pub step_id: u32,
pub duration_ms: Option<u64>,
pub failed: bool,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MigrationReport {
pub schema_version_before: u32,
pub schema_version_after: u32,
pub migration_steps: Vec<MigrationStepReport>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MigrationFailureReport {
pub schema_version_before: u32,
pub schema_version_current: u32,
pub migration_steps: Vec<MigrationStepReport>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum MigrationError {
IncompatibleSchemaVersion { seen: u32, supported: u32 },
MigrationError(MigrationFailureReport),
Storage { message: &'static str },
}
impl Display for MigrationError {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Self::IncompatibleSchemaVersion { seen, supported } => {
write!(f, "database schema version {seen} is incompatible with supported version {supported}")
}
Self::MigrationError(report) => write!(
f,
"schema migration failed at step {}",
report.migration_steps.last().map_or(0, |step| step.step_id)
),
Self::Storage { message } => write!(f, "schema storage error: {message}"),
}
}
}
impl std::error::Error for MigrationError {}
pub const MIGRATIONS: &[Migration] = &[
Migration {
step_id: 1,
sql: "CREATE TABLE IF NOT EXISTS _fathomdb_schema_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL)",
},
Migration {
step_id: 2,
sql: "CREATE TABLE IF NOT EXISTS _fathomdb_migrations(step_id INTEGER PRIMARY KEY, applied_at_ms INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS canonical_nodes(write_cursor INTEGER NOT NULL, kind TEXT NOT NULL, body TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS canonical_edges(write_cursor INTEGER NOT NULL, kind TEXT NOT NULL, from_id TEXT NOT NULL, to_id TEXT NOT NULL);",
},
Migration {
step_id: 3,
sql: "CREATE TABLE IF NOT EXISTS _fathomdb_embedder_profiles(profile TEXT PRIMARY KEY, name TEXT NOT NULL, revision TEXT NOT NULL, dimension INTEGER NOT NULL)",
},
Migration {
step_id: 4,
sql: "CREATE TABLE IF NOT EXISTS operational_collections(
name TEXT PRIMARY KEY,
kind TEXT NOT NULL CHECK(kind IN ('append_only_log', 'latest_state')),
schema_json TEXT NOT NULL,
retention_json TEXT NOT NULL,
format_version INTEGER NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS operational_mutations(
id INTEGER PRIMARY KEY AUTOINCREMENT,
collection_name TEXT NOT NULL,
record_key TEXT NOT NULL,
op_kind TEXT NOT NULL CHECK(op_kind = 'append'),
payload_json TEXT NOT NULL,
schema_id TEXT,
write_cursor INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS operational_state(
collection_name TEXT NOT NULL,
record_key TEXT NOT NULL,
payload_json TEXT NOT NULL,
schema_id TEXT,
write_cursor INTEGER NOT NULL,
PRIMARY KEY(collection_name, record_key)
);
CREATE TABLE IF NOT EXISTS _fathomdb_open_state(key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT OR IGNORE INTO operational_collections(
name, kind, schema_json, retention_json, format_version, created_at
) VALUES (
'projection_failures',
'append_only_log',
'{\"type\":\"object\"}',
'{}',
1,
0
);",
},
Migration {
step_id: 5,
sql: "CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
body,
kind UNINDEXED,
write_cursor UNINDEXED
);",
},
Migration {
step_id: 6,
sql: "CREATE TABLE IF NOT EXISTS _fathomdb_projection_state(
kind TEXT PRIMARY KEY,
last_enqueued_cursor INTEGER NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS _fathomdb_vector_kinds(
kind TEXT PRIMARY KEY,
profile TEXT NOT NULL,
created_at INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS _fathomdb_vector_rows(
rowid INTEGER PRIMARY KEY,
kind TEXT NOT NULL,
write_cursor INTEGER NOT NULL UNIQUE
);",
},
Migration {
step_id: 7,
sql: "CREATE TABLE IF NOT EXISTS _fathomdb_projection_terminal(
write_cursor INTEGER PRIMARY KEY,
state TEXT NOT NULL CHECK(state IN ('failed', 'up_to_date'))
);",
},
Migration {
step_id: 8,
sql: "ALTER TABLE canonical_nodes ADD COLUMN source_id TEXT;
ALTER TABLE canonical_edges ADD COLUMN source_id TEXT;
CREATE INDEX IF NOT EXISTS canonical_nodes_source_id_idx
ON canonical_nodes(source_id);
CREATE INDEX IF NOT EXISTS canonical_edges_source_id_idx
ON canonical_edges(source_id);",
},
Migration {
step_id: 9,
sql: "CREATE TEMP TABLE _vec0_migration_assertion(
check_passes INTEGER NOT NULL CHECK(check_passes = 1)
);
INSERT INTO _vec0_migration_assertion(check_passes)
SELECT CASE WHEN EXISTS (
SELECT 1 FROM _fathomdb_vector_rows
WHERE kind NOT IN ('email','article','paper','meeting','note','todo','doc')
) THEN 0 ELSE 1 END;
DROP TABLE _vec0_migration_assertion;",
},
Migration {
step_id: 10,
sql: "ALTER TABLE _fathomdb_embedder_profiles ADD COLUMN mean_vec BLOB",
},
Migration {
step_id: 11,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: tokenizer-default upgrade (drop+recreate FTS5 projection; no source-record migration)
DROP TABLE IF EXISTS search_index;
CREATE VIRTUAL TABLE search_index USING fts5(
body,
kind UNINDEXED,
write_cursor UNINDEXED,
tokenize = 'porter unicode61 remove_diacritics 2'
);",
},
Migration {
step_id: 12,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: G0 transaction-time identity substrate
ALTER TABLE canonical_nodes ADD COLUMN logical_id TEXT;
ALTER TABLE canonical_nodes ADD COLUMN superseded_at INTEGER;
ALTER TABLE canonical_edges ADD COLUMN logical_id TEXT;
ALTER TABLE canonical_edges ADD COLUMN superseded_at INTEGER;
CREATE UNIQUE INDEX IF NOT EXISTS canonical_nodes_logical_active_idx
ON canonical_nodes(logical_id) WHERE superseded_at IS NULL;
CREATE UNIQUE INDEX IF NOT EXISTS canonical_edges_logical_active_idx
ON canonical_edges(logical_id) WHERE superseded_at IS NULL;
CREATE INDEX IF NOT EXISTS canonical_nodes_kind_idx
ON canonical_nodes(kind);
CREATE INDEX IF NOT EXISTS canonical_edges_from_id_idx
ON canonical_edges(from_id);
CREATE INDEX IF NOT EXISTS canonical_edges_to_id_idx
ON canonical_edges(to_id);",
},
Migration {
step_id: 13,
sql: "CREATE INDEX IF NOT EXISTS operational_mutations_collection_id_idx
ON operational_mutations(collection_name, id);",
},
Migration {
step_id: 14,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: G11 edge enrichment (5 additive nullable columns + edge FTS table)
ALTER TABLE canonical_edges ADD COLUMN body TEXT;
ALTER TABLE canonical_edges ADD COLUMN t_valid TEXT;
ALTER TABLE canonical_edges ADD COLUMN t_invalid TEXT;
ALTER TABLE canonical_edges ADD COLUMN confidence REAL;
ALTER TABLE canonical_edges ADD COLUMN extractor_model_id TEXT;
CREATE VIRTUAL TABLE IF NOT EXISTS search_index_edges USING fts5(
body,
kind UNINDEXED,
write_cursor UNINDEXED,
tokenize = 'porter unicode61 remove_diacritics 2'
);",
},
Migration {
step_id: 15,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: R3 temporal_fallback provenance flag (additive nullable BOOLEAN column)
ALTER TABLE canonical_edges ADD COLUMN temporal_fallback INTEGER;",
},
Migration {
step_id: 16,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: EXP-S row_kind structural-role tag (additive NOT NULL DEFAULT 'leaf' column; separate axis from doc-type kind)
ALTER TABLE canonical_nodes ADD COLUMN row_kind TEXT NOT NULL DEFAULT 'leaf';",
},
Migration {
step_id: 17,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: F5 fielded FTS (new multi-column search_index_v2 FTS5 table + O(N) re-index; search_index retained)
CREATE VIRTUAL TABLE IF NOT EXISTS search_index_v2 USING fts5(
kind,
body,
status,
write_cursor UNINDEXED,
tokenize = 'porter unicode61 remove_diacritics 2'
);
INSERT INTO search_index_v2(kind, body, status, write_cursor)
SELECT
kind,
body,
CASE WHEN json_valid(body)
THEN COALESCE(json_extract(body, '$.status'), '')
ELSE '' END,
write_cursor
FROM canonical_nodes;",
},
Migration {
step_id: 18,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: F9 importance ranking scalar (additive nullable REAL; 3-way sentinel, NULL=graceful-absent)
ALTER TABLE canonical_nodes ADD COLUMN importance REAL;",
},
Migration {
step_id: 19,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: #5 vector-equivalence probe substrate (new internal _fathomdb_embed_probe table; UN-centered f32 references only, NEVER persists P1 bits)
CREATE TABLE IF NOT EXISTS _fathomdb_embed_probe(
probe_ordinal INTEGER PRIMARY KEY,
probe_text TEXT NOT NULL,
reference_vec BLOB NOT NULL,
embedder_name TEXT NOT NULL,
embedder_revision TEXT NOT NULL,
dim INTEGER NOT NULL
);",
},
Migration {
step_id: 20,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: OPP-12 Phase-1 existence axis (state NOT NULL DEFAULT 'active' + nullable reason on canonical_nodes + active-only partial index; no surrogate backfill — F-23 ruling 1a)
ALTER TABLE canonical_nodes ADD COLUMN state TEXT NOT NULL DEFAULT 'active';
ALTER TABLE canonical_nodes ADD COLUMN reason TEXT;
CREATE INDEX IF NOT EXISTS canonical_nodes_state_active_idx
ON canonical_nodes(write_cursor) WHERE state = 'active';",
},
Migration {
step_id: 21,
sql: "UPDATE canonical_nodes
SET source_id = '_legacy:pre-0.8.20'
WHERE source_id IS NULL AND logical_id IS NULL;
UPDATE canonical_edges
SET source_id = '_legacy:pre-0.8.20'
WHERE source_id IS NULL;",
},
Migration {
step_id: 22,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: R-20-NV node validity window (valid_from/valid_until INTEGER epoch-seconds on canonical_nodes; NULL = unbounded; half-open [valid_from, valid_until); deliberately INTEGER, diverging from the ISO-8601 TEXT canonical_edges.t_valid/t_invalid, which are unchanged)
ALTER TABLE canonical_nodes ADD COLUMN valid_from INTEGER;
ALTER TABLE canonical_nodes ADD COLUMN valid_until INTEGER;
CREATE INDEX IF NOT EXISTS canonical_nodes_validity_idx
ON canonical_nodes(valid_from, valid_until)
WHERE superseded_at IS NULL AND state = 'active';",
},
Migration {
step_id: 23,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: TC-33 edge temporal representation → INTEGER epoch seconds (recreate canonical_edges with INTEGER t_valid/t_invalid + typeof CHECKs so junk is UNSTORABLE; NULL still means \"still valid\"). NO DATA MIGRATION (HITL 2026-07-21): existing edge rows do NOT survive and no stored ISO-8601 value is converted.
INSERT OR REPLACE INTO _fathomdb_open_state(key, value)
SELECT 'tc33_reserved_write_cursor',
CAST(MAX(write_cursor) AS TEXT)
FROM canonical_edges
HAVING MAX(write_cursor) IS NOT NULL;
DELETE FROM search_index_edges;
-- fix-4 (TC-33): mark every edge cursor terminal BEFORE the DROP so
-- the SHARED projection cursor can walk past rows this recreate
-- removes; a pending edge (no terminal) would otherwise strand the
-- cursor and freeze surviving node projections too. 'up_to_date' is
-- the CHECK-valid token ('superseded' would be swallowed by
-- OR IGNORE). Complementary to the reserved-high-water fix above.
INSERT OR IGNORE INTO _fathomdb_projection_terminal(write_cursor, state)
SELECT write_cursor, 'up_to_date' FROM canonical_edges;
-- fix-6 (TC-33): delete the dropped edges' VECTOR sidecar rows
-- BEFORE the DROP, while canonical_edges still lists the edge
-- cursors. Scoped to edge cursors — _fathomdb_vector_rows also
-- holds NODE sidecar rows, which must survive. The vec0 table
-- vector_default is engine-created (dim-aware) and may not exist
-- here, so the engine prunes it to match right after
-- ensure_vector_partition. This is the third row-owned-projection
-- facet step 23 clears for every dropped edge (with the reserved
-- high-water mark and the terminal backfill above). NO DATA
-- MIGRATION: it deletes derived rows for already-dropped edges.
DELETE FROM _fathomdb_vector_rows
WHERE write_cursor IN (SELECT write_cursor FROM canonical_edges);
DROP TABLE canonical_edges;
CREATE TABLE canonical_edges(
write_cursor INTEGER NOT NULL,
kind TEXT NOT NULL,
from_id TEXT NOT NULL,
to_id TEXT NOT NULL,
source_id TEXT,
logical_id TEXT,
superseded_at INTEGER,
body TEXT,
t_valid INTEGER CHECK (t_valid IS NULL OR typeof(t_valid) = 'integer'),
t_invalid INTEGER CHECK (t_invalid IS NULL OR typeof(t_invalid) = 'integer'),
confidence REAL,
extractor_model_id TEXT,
temporal_fallback INTEGER
);
CREATE INDEX IF NOT EXISTS canonical_edges_source_id_idx
ON canonical_edges(source_id);
CREATE UNIQUE INDEX IF NOT EXISTS canonical_edges_logical_active_idx
ON canonical_edges(logical_id) WHERE superseded_at IS NULL;
CREATE INDEX IF NOT EXISTS canonical_edges_from_id_idx
ON canonical_edges(from_id);
CREATE INDEX IF NOT EXISTS canonical_edges_to_id_idx
ON canonical_edges(to_id);",
},
Migration {
step_id: 24,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: R-20-PR/R-20-EAV projection-registry EAV + property-FTS substrate (net-new _fathomdb_projection_registry durable derived-cache + canonical_attributes row-owned EAV projection + property_search_index FTS5 property-FTS). NO DATA MIGRATION (HITL 2026-07-21): shape only, no backfill.
CREATE TABLE _fathomdb_projection_registry(
name TEXT PRIMARY KEY,
roles TEXT NOT NULL,
fts_tokenizer TEXT,
vector_embedder TEXT,
vector_declared INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE canonical_attributes(
write_cursor INTEGER NOT NULL,
attr_name TEXT NOT NULL,
attr_value TEXT
);
CREATE INDEX canonical_attributes_name_value_idx
ON canonical_attributes(attr_name, attr_value);
CREATE INDEX canonical_attributes_cursor_idx
ON canonical_attributes(write_cursor);
CREATE VIRTUAL TABLE property_search_index USING fts5(
attr_value,
attr_name UNINDEXED,
write_cursor UNINDEXED,
tokenize = 'porter unicode61 remove_diacritics 2'
);",
},
Migration {
step_id: 25,
sql: "-- MIGRATION-ACCRETION-EXEMPTION: Slice-45 nested projection source declaration; additive registry column only, no data migration or canonical-body rewrite.
ALTER TABLE _fathomdb_projection_registry ADD COLUMN source TEXT;",
},
Migration {
step_id: 26,
sql: "CREATE INDEX IF NOT EXISTS canonical_nodes_write_cursor_idx ON canonical_nodes(write_cursor);
CREATE INDEX IF NOT EXISTS canonical_edges_write_cursor_idx ON canonical_edges(write_cursor);",
},
];
pub const RESERVED_WRITE_CURSOR_KEY: &str = "tc33_reserved_write_cursor";
pub fn migrate(conn: &Connection) -> Result<MigrationReport, MigrationError> {
migrate_with_steps(conn, MIGRATIONS)
}
pub fn migrate_with_steps(
conn: &Connection,
migrations: &[Migration],
) -> Result<MigrationReport, MigrationError> {
migrate_with_event_sink(conn, migrations, |_| {})
}
pub fn migrate_with_event_sink(
conn: &Connection,
migrations: &[Migration],
mut emit: impl FnMut(&MigrationStepReport),
) -> Result<MigrationReport, MigrationError> {
let before = user_version(conn)?;
if before > SCHEMA_VERSION {
return Err(MigrationError::IncompatibleSchemaVersion {
seen: before,
supported: SCHEMA_VERSION,
});
}
let mut current = before;
let mut reports = Vec::new();
for migration in migrations.iter().filter(|migration| migration.step_id > before) {
if migration.step_id != current.saturating_add(1) {
return Err(MigrationError::Storage {
message: "migration registry is not contiguous",
});
}
let started = Instant::now();
if let Err(_err) = apply_one(conn, migration) {
reports.push(MigrationStepReport {
step_id: migration.step_id,
duration_ms: Some(duration_ms(started)),
failed: true,
});
emit(reports.last().expect("failed step report was just pushed"));
let schema_version_current = user_version(conn).unwrap_or(current);
return Err(MigrationError::MigrationError(MigrationFailureReport {
schema_version_before: before,
schema_version_current,
migration_steps: reports,
}));
}
current = migration.step_id;
reports.push(MigrationStepReport {
step_id: migration.step_id,
duration_ms: Some(duration_ms(started)),
failed: false,
});
emit(reports.last().expect("successful step report was just pushed"));
}
Ok(MigrationReport {
schema_version_before: before,
schema_version_after: user_version(conn)?,
migration_steps: reports,
})
}
fn apply_one(conn: &Connection, migration: &Migration) -> rusqlite::Result<()> {
conn.execute_batch("BEGIN IMMEDIATE")?;
let result = (|| {
conn.execute_batch(migration.sql)?;
conn.pragma_update(None, PRAGMA_USER_VERSION, migration.step_id)?;
Ok(())
})();
match result {
Ok(()) => conn.execute_batch("COMMIT"),
Err(err) => {
let _ = conn.execute_batch("ROLLBACK");
Err(err)
}
}
}
fn user_version(conn: &Connection) -> Result<u32, MigrationError> {
conn.query_row("PRAGMA user_version", [], |row| row.get::<_, u32>(0))
.map_err(|_| MigrationError::Storage { message: "could not read schema version" })
}
fn duration_ms(started: Instant) -> u64 {
u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MigrationAccretionError {
pub offender: String,
}
impl Display for MigrationAccretionError {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "migration accretion guard rejected {}", self.offender)
}
}
impl std::error::Error for MigrationAccretionError {}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MigrationLogicalIdPinError {
pub offender: String,
pub statement: String,
}
impl Display for MigrationLogicalIdPinError {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
"TC-11 logical_id pin rejected {}: a migration may never populate `logical_id` on an \
existing canonical row — offending statement: {}",
self.offender, self.statement
)
}
}
impl std::error::Error for MigrationLogicalIdPinError {}
const PINNED_IDENTITY_TABLES: [&str; 2] = ["CANONICAL_NODES", "CANONICAL_EDGES"];
const PINNED_IDENTITY_COLUMN: &str = "LOGICAL_ID";
pub fn check_migration_logical_id_pin(
name: &str,
sql: &str,
) -> Result<(), MigrationLogicalIdPinError> {
for statement in normalized_statements(sql) {
if statement_violates_logical_id_pin(&statement) {
return Err(MigrationLogicalIdPinError { offender: name.to_string(), statement });
}
}
Ok(())
}
fn normalized_statements(sql: &str) -> Vec<String> {
let mut statements = Vec::new();
let mut current = String::new();
let chars: Vec<char> = sql.chars().collect();
let mut i = 0;
while i < chars.len() {
let c = chars[i];
match c {
'-' if chars.get(i + 1) == Some(&'-') => {
while i < chars.len() && chars[i] != '\n' {
i += 1;
}
}
'/' if chars.get(i + 1) == Some(&'*') => {
i += 2;
while i < chars.len() && !(chars[i] == '*' && chars.get(i + 1) == Some(&'/')) {
i += 1;
}
i = (i + 2).min(chars.len());
current.push(' ');
}
'\'' => {
i += 1;
while i < chars.len() {
if chars[i] == '\'' {
if chars.get(i + 1) == Some(&'\'') {
i += 2;
continue;
}
i += 1;
break;
}
i += 1;
}
current.push_str("''");
}
'"' | '`' | '[' => {
let close = if c == '[' { ']' } else { c };
i += 1;
while i < chars.len() && chars[i] != close {
current.push(chars[i].to_ascii_uppercase());
i += 1;
}
i += 1;
}
';' => {
statements.push(normalize_statement(¤t));
current.clear();
i += 1;
}
_ => {
current.push(c.to_ascii_uppercase());
i += 1;
}
}
}
statements.push(normalize_statement(¤t));
statements.retain(|s| !s.is_empty());
statements
}
fn normalize_statement(raw: &str) -> String {
tighten_qualified_names(&collapse_whitespace(raw))
}
fn collapse_whitespace(raw: &str) -> String {
raw.split_whitespace().collect::<Vec<_>>().join(" ")
}
fn tighten_qualified_names(statement: &str) -> String {
if !statement.contains('.') {
return statement.to_string();
}
statement.split('.').map(str::trim).collect::<Vec<_>>().join(".")
}
fn statement_violates_logical_id_pin(statement: &str) -> bool {
let names_column = statement.contains(PINNED_IDENTITY_COLUMN);
if writes_pinned_identity_anywhere(statement, names_column) {
return true;
}
if statement.starts_with("CREATE") && statement.contains(" TRIGGER ") && names_column {
return mentions_pinned_table(statement);
}
if statement.starts_with("UPDATE") {
let Some(table) = update_target_table(statement) else { return false };
if !is_pinned_table(&table) {
return false;
}
return set_clause(statement).is_some_and(|clause| clause.contains(PINNED_IDENTITY_COLUMN));
}
if statement.starts_with("INSERT") || statement.starts_with("REPLACE") {
let Some((table, rest)) = table_after_into(statement) else { return false };
if !is_pinned_table(&table) {
return false;
}
return names_column || !rest.trim_start().starts_with('(');
}
if statement.starts_with("ALTER") {
let Some(table) = token_after(statement, "ALTER TABLE ") else { return false };
if !is_pinned_table(&table) || !names_column {
return false;
}
return statement.contains("DEFAULT") || statement.contains("RENAME");
}
if statement.starts_with("CREATE") && statement.contains(" TABLE ") {
let Some(table) = token_after(statement, " TABLE ") else { return false };
return is_pinned_table(&table) && names_column && statement.contains("DEFAULT");
}
false
}
fn writes_pinned_identity_anywhere(statement: &str, names_column: bool) -> bool {
if names_column
&& mentions_pinned_table(statement)
&& set_clause(statement).is_some_and(|clause| clause.contains(PINNED_IDENTITY_COLUMN))
{
return true;
}
table_after_into(statement).is_some_and(|(table, rest)| {
is_pinned_table(&table) && (names_column || !rest.trim_start().starts_with('('))
})
}
fn bare_table_name(token: &str) -> &str {
token.rsplit_once('.').map_or(token, |(_, table)| table)
}
fn is_pinned_table(token: &str) -> bool {
PINNED_IDENTITY_TABLES.contains(&bare_table_name(token))
}
fn mentions_pinned_table(statement: &str) -> bool {
statement
.split(|c: char| !(c.is_ascii_alphanumeric() || c == '_' || c == '.'))
.any(is_pinned_table)
}
const SQLITE_CONFLICT_ACTIONS: [&str; 5] = ["ROLLBACK", "ABORT", "FAIL", "IGNORE", "REPLACE"];
fn update_target_table(statement: &str) -> Option<String> {
let mut rest = statement.strip_prefix("UPDATE ")?;
if let Some(after_or) = rest.strip_prefix("OR ") {
if let Some(after_action) = SQLITE_CONFLICT_ACTIONS
.iter()
.find_map(|action| after_or.strip_prefix(*action)?.strip_prefix(' '))
{
rest = after_action;
}
}
first_token(rest)
}
fn token_after(statement: &str, marker: &str) -> Option<String> {
first_token(statement.split_once(marker)?.1)
}
fn first_token(rest: &str) -> Option<String> {
let token: String =
rest.chars().take_while(|c| !c.is_whitespace() && *c != '(' && *c != ',').collect();
(!token.is_empty()).then_some(token)
}
fn table_after_into(statement: &str) -> Option<(String, String)> {
let rest = statement.split_once(" INTO ")?.1;
let table: String =
rest.chars().take_while(|c| !c.is_whitespace() && *c != '(' && *c != ',').collect();
(!table.is_empty()).then(|| (table.clone(), rest[table.len()..].to_string()))
}
fn set_clause(statement: &str) -> Option<&str> {
let after_set = statement.split_once(" SET ")?.1;
let end = [" WHERE ", " FROM ", " RETURNING "]
.iter()
.filter_map(|kw| after_set.find(kw))
.min()
.unwrap_or(after_set.len());
Some(&after_set[..end])
}
pub fn check_migration_accretion(name: &str, sql: &str) -> Result<(), MigrationAccretionError> {
let upper = sql.to_ascii_uppercase();
let adds_schema = upper.contains("CREATE TABLE") || upper.contains("ADD COLUMN");
let names_removal = upper.contains("DROP TABLE") || upper.contains("DROP COLUMN");
let has_exemption = sql.contains("-- MIGRATION-ACCRETION-EXEMPTION: ");
if adds_schema && !names_removal && !has_exemption {
return Err(MigrationAccretionError { offender: name.to_string() });
}
Ok(())
}