fathomdb_schema/
lib.rs

1use std::fmt::{Display, Formatter};
2use std::time::Instant;
3
4use rusqlite::Connection;
5
6pub const SCHEMA_VERSION: u32 = 8;
7
8/// SQLite `PRAGMA` name carrying the on-disk schema-version sentinel.
9///
10/// Public on-disk surface per `dev/interfaces/wire.md` § Schema-version
11/// sentinel; advanced by successful migrations per `dev/design/migrations.md`.
12pub const PRAGMA_USER_VERSION: &str = "user_version";
13
14/// Suffix of the canonical SQLite database file (`<db-name>.sqlite`).
15pub const SQLITE_SUFFIX: &str = ".sqlite";
16
17/// Suffix of the SQLite write-ahead log file (`<db-name>.sqlite-wal`).
18pub const WAL_SUFFIX: &str = "-wal";
19
20/// Suffix of the sidecar lock file (`<db-name>.sqlite.lock`).
21///
22/// Per `dev/design/bindings.md` § 7, this sidecar flock is the load-bearing
23/// cross-process exclusion layer; it surfaces lock contention before SQLite
24/// I/O begins.
25pub const LOCK_SUFFIX: &str = ".lock";
26
27/// Suffix of the optional SQLite rollback journal file
28/// (`<db-name>.sqlite-journal`).
29pub const JOURNAL_SUFFIX: &str = "-journal";
30
31#[must_use]
32pub fn bootstrap_steps() -> &'static [&'static str] {
33    &["create canonical tables", "register projection metadata", "seed rewrite-era configuration"]
34}
35
36/// Canonical tables owned by the rewrite-era schema, in stable display
37/// order. Excludes FTS, vec0, and projection shadow tables (re-derivable
38/// from canonical state) and internal `_fathomdb_*` metadata.
39///
40/// `doctor dump-row-counts` enumerates this set; `doctor dump-schema`
41/// uses it to order canonical tables ahead of derived/internal ones.
42pub const CANONICAL_TABLES: &[&str] = &[
43    "canonical_nodes",
44    "canonical_edges",
45    "operational_collections",
46    "operational_mutations",
47    "operational_state",
48];
49
50#[derive(Clone, Copy, Debug, Eq, PartialEq)]
51pub struct Migration {
52    pub step_id: u32,
53    pub sql: &'static str,
54}
55
56#[derive(Clone, Debug, Eq, PartialEq)]
57pub struct MigrationStepReport {
58    pub step_id: u32,
59    pub duration_ms: Option<u64>,
60    pub failed: bool,
61}
62
63#[derive(Clone, Debug, Eq, PartialEq)]
64pub struct MigrationReport {
65    pub schema_version_before: u32,
66    pub schema_version_after: u32,
67    pub migration_steps: Vec<MigrationStepReport>,
68}
69
70#[derive(Clone, Debug, Eq, PartialEq)]
71pub struct MigrationFailureReport {
72    pub schema_version_before: u32,
73    pub schema_version_current: u32,
74    pub migration_steps: Vec<MigrationStepReport>,
75}
76
77#[derive(Clone, Debug, Eq, PartialEq)]
78pub enum MigrationError {
79    IncompatibleSchemaVersion { seen: u32, supported: u32 },
80    MigrationError(MigrationFailureReport),
81    Storage { message: &'static str },
82}
83
84impl Display for MigrationError {
85    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
86        match self {
87            Self::IncompatibleSchemaVersion { seen, supported } => {
88                write!(f, "database schema version {seen} is incompatible with supported version {supported}")
89            }
90            Self::MigrationError(report) => write!(
91                f,
92                "schema migration failed at step {}",
93                report.migration_steps.last().map_or(0, |step| step.step_id)
94            ),
95            Self::Storage { message } => write!(f, "schema storage error: {message}"),
96        }
97    }
98}
99
100impl std::error::Error for MigrationError {}
101
102pub const MIGRATIONS: &[Migration] = &[
103    Migration {
104        step_id: 1,
105        sql: "CREATE TABLE IF NOT EXISTS _fathomdb_schema_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL)",
106    },
107    Migration {
108        step_id: 2,
109        sql: "CREATE TABLE IF NOT EXISTS _fathomdb_migrations(step_id INTEGER PRIMARY KEY, applied_at_ms INTEGER NOT NULL);
110              CREATE TABLE IF NOT EXISTS canonical_nodes(write_cursor INTEGER NOT NULL, kind TEXT NOT NULL, body TEXT NOT NULL);
111              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);",
112    },
113    Migration {
114        step_id: 3,
115        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)",
116    },
117    Migration {
118        step_id: 4,
119        sql: "CREATE TABLE IF NOT EXISTS operational_collections(
120                  name TEXT PRIMARY KEY,
121                  kind TEXT NOT NULL CHECK(kind IN ('append_only_log', 'latest_state')),
122                  schema_json TEXT NOT NULL,
123                  retention_json TEXT NOT NULL,
124                  format_version INTEGER NOT NULL,
125                  created_at INTEGER NOT NULL
126              );
127              CREATE TABLE IF NOT EXISTS operational_mutations(
128                  id INTEGER PRIMARY KEY AUTOINCREMENT,
129                  collection_name TEXT NOT NULL,
130                  record_key TEXT NOT NULL,
131                  op_kind TEXT NOT NULL CHECK(op_kind = 'append'),
132                  payload_json TEXT NOT NULL,
133                  schema_id TEXT,
134                  write_cursor INTEGER NOT NULL
135              );
136              CREATE TABLE IF NOT EXISTS operational_state(
137                  collection_name TEXT NOT NULL,
138                  record_key TEXT NOT NULL,
139                  payload_json TEXT NOT NULL,
140                  schema_id TEXT,
141                  write_cursor INTEGER NOT NULL,
142                  PRIMARY KEY(collection_name, record_key)
143              );
144              CREATE TABLE IF NOT EXISTS _fathomdb_open_state(key TEXT PRIMARY KEY, value TEXT NOT NULL);
145              INSERT OR IGNORE INTO operational_collections(
146                  name, kind, schema_json, retention_json, format_version, created_at
147              ) VALUES (
148                  'projection_failures',
149                  'append_only_log',
150                  '{\"type\":\"object\"}',
151                  '{}',
152                  1,
153                  0
154              );",
155    },
156    Migration {
157        step_id: 5,
158        sql: "CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
159                  body,
160                  kind UNINDEXED,
161                  write_cursor UNINDEXED
162              );",
163    },
164    Migration {
165        step_id: 6,
166        sql: "CREATE TABLE IF NOT EXISTS _fathomdb_projection_state(
167                  kind TEXT PRIMARY KEY,
168                  last_enqueued_cursor INTEGER NOT NULL DEFAULT 0,
169                  updated_at INTEGER NOT NULL DEFAULT 0
170              );
171              CREATE TABLE IF NOT EXISTS _fathomdb_vector_kinds(
172                  kind TEXT PRIMARY KEY,
173                  profile TEXT NOT NULL,
174                  created_at INTEGER NOT NULL DEFAULT 0
175              );
176              CREATE TABLE IF NOT EXISTS _fathomdb_vector_rows(
177                  rowid INTEGER PRIMARY KEY,
178                  kind TEXT NOT NULL,
179                  write_cursor INTEGER NOT NULL UNIQUE
180              );",
181    },
182    Migration {
183        step_id: 7,
184        sql: "CREATE TABLE IF NOT EXISTS _fathomdb_projection_terminal(
185                  write_cursor INTEGER PRIMARY KEY,
186                  state TEXT NOT NULL CHECK(state IN ('failed', 'up_to_date'))
187              );",
188    },
189    // Phase 9 Pack B — REQ-026 / AC-028a/b/c / AC-042 recovery seam.
190    // `source_id` is nullable; existing canonical rows back-fill to NULL,
191    // so reads from older callers stay schema-stable. REQ-045 accretion
192    // offset is documented in `migrations/008_source_id.sql` as inherently
193    // impossible for this slice (every existing canonical column is
194    // load-bearing for replay / projections / recovery locators); the
195    // next schema-touching pack carries the offset budget for two adds.
196    Migration {
197        step_id: 8,
198        sql: "ALTER TABLE canonical_nodes ADD COLUMN source_id TEXT;
199              ALTER TABLE canonical_edges ADD COLUMN source_id TEXT;
200              CREATE INDEX IF NOT EXISTS canonical_nodes_source_id_idx
201                  ON canonical_nodes(source_id);
202              CREATE INDEX IF NOT EXISTS canonical_edges_source_id_idx
203                  ON canonical_edges(source_id);",
204    },
205];
206
207pub fn migrate(conn: &Connection) -> Result<MigrationReport, MigrationError> {
208    migrate_with_steps(conn, MIGRATIONS)
209}
210
211pub fn migrate_with_steps(
212    conn: &Connection,
213    migrations: &[Migration],
214) -> Result<MigrationReport, MigrationError> {
215    migrate_with_event_sink(conn, migrations, |_| {})
216}
217
218pub fn migrate_with_event_sink(
219    conn: &Connection,
220    migrations: &[Migration],
221    mut emit: impl FnMut(&MigrationStepReport),
222) -> Result<MigrationReport, MigrationError> {
223    let before = user_version(conn)?;
224    if before > SCHEMA_VERSION {
225        return Err(MigrationError::IncompatibleSchemaVersion {
226            seen: before,
227            supported: SCHEMA_VERSION,
228        });
229    }
230
231    let mut current = before;
232    let mut reports = Vec::new();
233
234    for migration in migrations.iter().filter(|migration| migration.step_id > before) {
235        if migration.step_id != current.saturating_add(1) {
236            return Err(MigrationError::Storage {
237                message: "migration registry is not contiguous",
238            });
239        }
240
241        let started = Instant::now();
242        if let Err(_err) = apply_one(conn, migration) {
243            reports.push(MigrationStepReport {
244                step_id: migration.step_id,
245                duration_ms: Some(duration_ms(started)),
246                failed: true,
247            });
248            emit(reports.last().expect("failed step report was just pushed"));
249            let schema_version_current = user_version(conn).unwrap_or(current);
250            return Err(MigrationError::MigrationError(MigrationFailureReport {
251                schema_version_before: before,
252                schema_version_current,
253                migration_steps: reports,
254            }));
255        }
256
257        current = migration.step_id;
258        reports.push(MigrationStepReport {
259            step_id: migration.step_id,
260            duration_ms: Some(duration_ms(started)),
261            failed: false,
262        });
263        emit(reports.last().expect("successful step report was just pushed"));
264    }
265
266    Ok(MigrationReport {
267        schema_version_before: before,
268        schema_version_after: user_version(conn)?,
269        migration_steps: reports,
270    })
271}
272
273fn apply_one(conn: &Connection, migration: &Migration) -> rusqlite::Result<()> {
274    conn.execute_batch("BEGIN IMMEDIATE")?;
275    let result = (|| {
276        conn.execute_batch(migration.sql)?;
277        conn.pragma_update(None, PRAGMA_USER_VERSION, migration.step_id)?;
278        Ok(())
279    })();
280
281    match result {
282        Ok(()) => conn.execute_batch("COMMIT"),
283        Err(err) => {
284            let _ = conn.execute_batch("ROLLBACK");
285            Err(err)
286        }
287    }
288}
289
290fn user_version(conn: &Connection) -> Result<u32, MigrationError> {
291    conn.query_row("PRAGMA user_version", [], |row| row.get::<_, u32>(0))
292        .map_err(|_| MigrationError::Storage { message: "could not read schema version" })
293}
294
295fn duration_ms(started: Instant) -> u64 {
296    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
297}
298
299#[derive(Clone, Debug, Eq, PartialEq)]
300pub struct MigrationAccretionError {
301    pub offender: String,
302}
303
304impl Display for MigrationAccretionError {
305    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
306        write!(f, "migration accretion guard rejected {}", self.offender)
307    }
308}
309
310impl std::error::Error for MigrationAccretionError {}
311
312pub fn check_migration_accretion(name: &str, sql: &str) -> Result<(), MigrationAccretionError> {
313    let upper = sql.to_ascii_uppercase();
314    let adds_schema = upper.contains("CREATE TABLE") || upper.contains("ADD COLUMN");
315    let names_removal = upper.contains("DROP TABLE") || upper.contains("DROP COLUMN");
316    let has_exemption = sql.contains("-- MIGRATION-ACCRETION-EXEMPTION: ");
317
318    if adds_schema && !names_removal && !has_exemption {
319        return Err(MigrationAccretionError { offender: name.to_string() });
320    }
321
322    Ok(())
323}
fathomdb_schema/lib.rs

fathomdb_schema/
lib.rs
