fathomdb_schema/
lib.rs

1use std::fmt::{Display, Formatter};
2use std::time::Instant;
3
4use rusqlite::Connection;
5
6pub const SCHEMA_VERSION: u32 = 10;
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    // 0.7.0 Pack 1 — Vector binary-quantization data encoding.
206    // Per `dev/design/0.7.0-vector-quant-pack1.md` D4 (fix-3). Stages
207    // the existing f32 corpus + kind mapping into a regular SQL table,
208    // drops + recreates `vector_default` with the new schema (sibling
209    // `embedding_bin bit[768]`, `source_type TEXT partition key`,
210    // `kind TEXT`, `created_at INTEGER`), then repopulates with
211    // SQL-side `vec_quantize_binary` and the D3 CASE mapping. A
212    // prefix CHECK-constraint preflight aborts the migration if any
213    // `_fathomdb_vector_rows.kind` is outside the locked vocabulary.
214    //
215    // `<dim>=768` is hardcoded against the default profile
216    // (`load_default_profile` -> `DEFAULT_EMBEDDER_DIMENSION` in
217    // fathomdb-engine). The design notes this constraint and defers
218    // a runtime-dim migration to 0.7.1.
219    Migration {
220        step_id: 9,
221        // SQL-side: D4 fix-3.1 preflight only. The vec0 reshape itself is
222        // dim-aware and lives in the engine crate's
223        // `ensure_vector_partition_pack1` (called by `ensure_vector_partition`
224        // immediately after `migrate_with_event_sink` returns). Splitting the
225        // preflight (SQL, in-tx with `apply_one`) from the reshape (Rust,
226        // dim-driven by `_fathomdb_embedder_profiles.dimension`) is required
227        // because `fathomdb-schema::Migration` is a `&'static str` with no
228        // runtime parameterization, and the existing dim=8 / dim=384 test
229        // suite must stay GREEN. The reshape is idempotent across crashes:
230        // if open fails between this step's commit (user_version=9) and the
231        // Rust reshape, the next open re-detects the old shape and replays
232        // the reshape. See dev/plans/runs/0.7.0-PVQ-P1-IMPL-output.json
233        // for the design-memo deviation note.
234        sql: "CREATE TEMP TABLE _vec0_migration_assertion(
235                  check_passes INTEGER NOT NULL CHECK(check_passes = 1)
236              );
237              INSERT INTO _vec0_migration_assertion(check_passes)
238                  SELECT CASE WHEN EXISTS (
239                      SELECT 1 FROM _fathomdb_vector_rows
240                      WHERE kind NOT IN ('email','article','paper','meeting','note','todo','doc')
241                  ) THEN 0 ELSE 1 END;
242              DROP TABLE _vec0_migration_assertion;",
243    },
244    // 0.7.1 EU-5a2 — mean-centering schema column.
245    // Per `dev/design/embedder.md` §0.2: nullable BLOB holding the
246    // pinned per-workspace mean vector for the default profile. Byte
247    // length, when non-NULL, MUST equal `4 * dimension` (f32 little-endian).
248    // Pure additive ALTER; SQLite stores NULL for the pre-existing row.
249    // Lifecycle (compute-once-on-first-ingest threshold-pin) is in the
250    // engine crate, not the schema layer.
251    Migration {
252        step_id: 10,
253        sql: "ALTER TABLE _fathomdb_embedder_profiles ADD COLUMN mean_vec BLOB",
254    },
255];
256
257pub fn migrate(conn: &Connection) -> Result<MigrationReport, MigrationError> {
258    migrate_with_steps(conn, MIGRATIONS)
259}
260
261pub fn migrate_with_steps(
262    conn: &Connection,
263    migrations: &[Migration],
264) -> Result<MigrationReport, MigrationError> {
265    migrate_with_event_sink(conn, migrations, |_| {})
266}
267
268pub fn migrate_with_event_sink(
269    conn: &Connection,
270    migrations: &[Migration],
271    mut emit: impl FnMut(&MigrationStepReport),
272) -> Result<MigrationReport, MigrationError> {
273    let before = user_version(conn)?;
274    if before > SCHEMA_VERSION {
275        return Err(MigrationError::IncompatibleSchemaVersion {
276            seen: before,
277            supported: SCHEMA_VERSION,
278        });
279    }
280
281    let mut current = before;
282    let mut reports = Vec::new();
283
284    for migration in migrations.iter().filter(|migration| migration.step_id > before) {
285        if migration.step_id != current.saturating_add(1) {
286            return Err(MigrationError::Storage {
287                message: "migration registry is not contiguous",
288            });
289        }
290
291        let started = Instant::now();
292        if let Err(_err) = apply_one(conn, migration) {
293            reports.push(MigrationStepReport {
294                step_id: migration.step_id,
295                duration_ms: Some(duration_ms(started)),
296                failed: true,
297            });
298            emit(reports.last().expect("failed step report was just pushed"));
299            let schema_version_current = user_version(conn).unwrap_or(current);
300            return Err(MigrationError::MigrationError(MigrationFailureReport {
301                schema_version_before: before,
302                schema_version_current,
303                migration_steps: reports,
304            }));
305        }
306
307        current = migration.step_id;
308        reports.push(MigrationStepReport {
309            step_id: migration.step_id,
310            duration_ms: Some(duration_ms(started)),
311            failed: false,
312        });
313        emit(reports.last().expect("successful step report was just pushed"));
314    }
315
316    Ok(MigrationReport {
317        schema_version_before: before,
318        schema_version_after: user_version(conn)?,
319        migration_steps: reports,
320    })
321}
322
323fn apply_one(conn: &Connection, migration: &Migration) -> rusqlite::Result<()> {
324    conn.execute_batch("BEGIN IMMEDIATE")?;
325    let result = (|| {
326        conn.execute_batch(migration.sql)?;
327        conn.pragma_update(None, PRAGMA_USER_VERSION, migration.step_id)?;
328        Ok(())
329    })();
330
331    match result {
332        Ok(()) => conn.execute_batch("COMMIT"),
333        Err(err) => {
334            let _ = conn.execute_batch("ROLLBACK");
335            Err(err)
336        }
337    }
338}
339
340fn user_version(conn: &Connection) -> Result<u32, MigrationError> {
341    conn.query_row("PRAGMA user_version", [], |row| row.get::<_, u32>(0))
342        .map_err(|_| MigrationError::Storage { message: "could not read schema version" })
343}
344
345fn duration_ms(started: Instant) -> u64 {
346    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
347}
348
349#[derive(Clone, Debug, Eq, PartialEq)]
350pub struct MigrationAccretionError {
351    pub offender: String,
352}
353
354impl Display for MigrationAccretionError {
355    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
356        write!(f, "migration accretion guard rejected {}", self.offender)
357    }
358}
359
360impl std::error::Error for MigrationAccretionError {}
361
362pub fn check_migration_accretion(name: &str, sql: &str) -> Result<(), MigrationAccretionError> {
363    let upper = sql.to_ascii_uppercase();
364    let adds_schema = upper.contains("CREATE TABLE") || upper.contains("ADD COLUMN");
365    let names_removal = upper.contains("DROP TABLE") || upper.contains("DROP COLUMN");
366    let has_exemption = sql.contains("-- MIGRATION-ACCRETION-EXEMPTION: ");
367
368    if adds_schema && !names_removal && !has_exemption {
369        return Err(MigrationAccretionError { offender: name.to_string() });
370    }
371
372    Ok(())
373}
fathomdb_schema/lib.rs

fathomdb_schema/
lib.rs
