persona_wire/migrations/
mod.rs

1//! Schema migration framework — Diesel / sqlx style, scoped to persona-wire's
2//! SQLite store. Each numbered migration declares an idempotent `up` step
3//! that the [`Runner`] tracks in a `schema_migrations` table so re-running
4//! is a no-op once applied.
5//!
6//! Migrations are listed in [`ALL`] in **execution order**. A new schema
7//! change adds one module under this directory plus one entry at the end of
8//! `ALL`. Down migrations are intentionally not modelled in v1 — the
9//! framework leaves room for a future `Migration::down` extension once a
10//! real rollback need shows up.
11
12use anyhow::{anyhow, Context, Result};
13use rusqlite::{params, Connection};
14pub use ulid::Ulid;
15
16pub mod m001_node_id_ulid;
17pub mod m002_registry_id_ulid;
18pub mod m003_bundle_installs_fk_relax;
19
20/// One immutable, monotonic schema change. `id` must be unique across the
21/// whole [`ALL`] list and never change once shipped — it is the persistent
22/// key the `schema_migrations` table records.
23pub trait Migration: Sync {
24    /// Stable identifier (e.g. `"001_node_id_ulid"`). Persisted to
25    /// `schema_migrations.version`; renaming after release breaks
26    /// idempotency for already-migrated stores.
27    fn id(&self) -> &'static str;
28
29    /// One-line human description (surfaced by `pw-migrate status` etc).
30    fn description(&self) -> &'static str;
31
32    /// Apply this migration to `conn`. The framework runs `up` inside an
33    /// outer `BEGIN IMMEDIATE` transaction managed by [`Runner`]; the
34    /// migration body is free to issue further `PRAGMA` / DDL / DML as
35    /// long as it leaves the schema in the post-migration shape.
36    fn up(&self, conn: &Connection) -> Result<()>;
37}
38
39/// Registry of all known migrations, in **execution order**. Append-only.
40pub static ALL: &[&'static dyn Migration] = &[
41    &m001_node_id_ulid::MIGRATION,
42    &m002_registry_id_ulid::MIGRATION,
43    &m003_bundle_installs_fk_relax::MIGRATION,
44];
45
46/// Driver around a `Connection` that knows how to read / write the
47/// `schema_migrations` ledger and apply pending migrations.
48pub struct Runner<'a> {
49    conn: &'a Connection,
50}
51
52impl<'a> Runner<'a> {
53    pub fn new(conn: &'a Connection) -> Self {
54        Self { conn }
55    }
56
57    /// Ensure the bookkeeping table exists. Idempotent — safe to call on
58    /// every CLI invocation.
59    pub fn ensure_table(&self) -> Result<()> {
60        self.conn
61            .execute_batch(
62                "CREATE TABLE IF NOT EXISTS schema_migrations (
63                     version     TEXT PRIMARY KEY,
64                     description TEXT NOT NULL,
65                     applied_at  INTEGER NOT NULL
66                 );",
67            )
68            .context("create schema_migrations")?;
69        Ok(())
70    }
71
72    /// Snapshot of (applied, pending) migrations vs the [`ALL`] registry.
73    pub fn status(&self) -> Result<Status> {
74        self.ensure_table()?;
75        let mut stmt = self.conn.prepare(
76            "SELECT version, description, applied_at FROM schema_migrations \
77             ORDER BY applied_at",
78        )?;
79        let applied: Vec<AppliedRow> = stmt
80            .query_map([], |r| {
81                Ok(AppliedRow {
82                    version: r.get(0)?,
83                    description: r.get(1)?,
84                    applied_at: r.get(2)?,
85                })
86            })?
87            .collect::<rusqlite::Result<_>>()?;
88        drop(stmt);
89        let applied_ids: std::collections::HashSet<String> =
90            applied.iter().map(|r| r.version.clone()).collect();
91        let pending: Vec<&'static dyn Migration> = ALL
92            .iter()
93            .copied()
94            .filter(|m| !applied_ids.contains(m.id()))
95            .collect();
96        Ok(Status { applied, pending })
97    }
98
99    /// Apply every pending migration in [`ALL`] order, up to and including
100    /// `target` (or all pending when `target` is `None`). Returns the list
101    /// of migrations actually applied this call (skipped ones are not
102    /// included).
103    pub fn up(&self, target: Option<&str>) -> Result<Vec<AppliedNow>> {
104        let status = self.status()?;
105        if let Some(t) = target {
106            if !ALL.iter().any(|m| m.id() == t) {
107                return Err(anyhow!("unknown migration: {t}"));
108            }
109        }
110        let mut applied_now = Vec::new();
111        for m in status.pending {
112            self.apply_one(m)?;
113            applied_now.push(AppliedNow {
114                version: m.id().to_string(),
115                description: m.description().to_string(),
116            });
117            if target.is_some_and(|t| t == m.id()) {
118                break;
119            }
120        }
121        Ok(applied_now)
122    }
123
124    /// Apply exactly one migration by id (errors if already applied OR not
125    /// known). Useful when an operator wants strict 1-step control.
126    pub fn apply(&self, id: &str) -> Result<()> {
127        let m = ALL
128            .iter()
129            .copied()
130            .find(|m| m.id() == id)
131            .ok_or_else(|| anyhow!("unknown migration: {id}"))?;
132        let already = self.status()?.applied.iter().any(|r| r.version == id);
133        if already {
134            return Err(anyhow!("migration already applied: {id}"));
135        }
136        self.apply_one(m)
137    }
138
139    fn apply_one(&self, m: &'static dyn Migration) -> Result<()> {
140        // FK guard off + outer transaction. Migration body may toggle further
141        // pragmas but the commit / rollback is the framework's responsibility.
142        self.conn
143            .execute_batch("PRAGMA foreign_keys = OFF; BEGIN IMMEDIATE;")
144            .context("open migration transaction")?;
145        let res: Result<()> = (|| {
146            m.up(self.conn)
147                .with_context(|| format!("migration {} body", m.id()))?;
148            self.conn
149                .execute(
150                    "INSERT INTO schema_migrations(version, description, applied_at) \
151                     VALUES (?1, ?2, ?3)",
152                    params![m.id(), m.description(), epoch_ms()?],
153                )
154                .with_context(|| format!("record schema_migrations row for {}", m.id()))?;
155            // Re-enable FK + validate.
156            self.conn
157                .execute_batch("PRAGMA foreign_keys = ON;")
158                .context("re-enable foreign_keys")?;
159            let mut stmt = self.conn.prepare("PRAGMA foreign_key_check;")?;
160            let violations: Vec<(String, i64, String, i64)> = stmt
161                .query_map([], |r| {
162                    Ok((
163                        r.get::<_, String>(0)?,
164                        r.get::<_, i64>(1)?,
165                        r.get::<_, String>(2)?,
166                        r.get::<_, i64>(3)?,
167                    ))
168                })?
169                .collect::<rusqlite::Result<_>>()?;
170            if !violations.is_empty() {
171                return Err(anyhow!(
172                    "foreign_key_check violations after {}: {:?}",
173                    m.id(),
174                    violations
175                ));
176            }
177            Ok(())
178        })();
179        match res {
180            Ok(()) => {
181                self.conn.execute("COMMIT", [])?;
182                Ok(())
183            }
184            Err(e) => {
185                let _ = self.conn.execute("ROLLBACK", []);
186                Err(e.context(format!("migration {} aborted; rolled back", m.id())))
187            }
188        }
189    }
190}
191
192fn epoch_ms() -> Result<i64> {
193    use std::time::{SystemTime, UNIX_EPOCH};
194    let d = SystemTime::now()
195        .duration_since(UNIX_EPOCH)
196        .context("system time")?;
197    Ok(d.as_millis() as i64)
198}
199
200#[derive(Clone)]
201pub struct Status {
202    pub applied: Vec<AppliedRow>,
203    pub pending: Vec<&'static dyn Migration>,
204}
205
206impl std::fmt::Debug for Status {
207    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
208        f.debug_struct("Status")
209            .field("applied", &self.applied)
210            .field(
211                "pending_ids",
212                &self.pending.iter().map(|m| m.id()).collect::<Vec<_>>(),
213            )
214            .finish()
215    }
216}
217
218#[derive(Debug, Clone)]
219pub struct AppliedRow {
220    pub version: String,
221    pub description: String,
222    pub applied_at: i64,
223}
224
225#[derive(Debug, Clone)]
226pub struct AppliedNow {
227    pub version: String,
228    pub description: String,
229}
230
231// Re-exported so individual migration modules and the bin agree on the
232// type for ULID minting + mapping tables.
233pub(crate) use Ulid as MigrationUlid;
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238
239    fn seed_v0_6(conn: &Connection) {
240        conn.execute_batch(
241            r#"
242            CREATE TABLE type_registry (
243                name TEXT PRIMARY KEY,
244                kind TEXT NOT NULL,
245                schema_json TEXT,
246                severity_allowed TEXT
247            );
248            CREATE TABLE nodes (
249                id TEXT PRIMARY KEY,
250                type TEXT NOT NULL REFERENCES type_registry(name),
251                sot_ref TEXT, confidence REAL, applicability TEXT,
252                last_verified_at INTEGER, review_due INTEGER,
253                version INTEGER NOT NULL DEFAULT 1,
254                prev_id TEXT,
255                metadata TEXT NOT NULL DEFAULT '{}'
256            );
257            CREATE TABLE edges (
258                id TEXT PRIMARY KEY,
259                src_node TEXT NOT NULL REFERENCES nodes(id),
260                tgt_node TEXT NOT NULL REFERENCES nodes(id),
261                kind TEXT NOT NULL REFERENCES type_registry(name),
262                severity TEXT,
263                metadata TEXT NOT NULL DEFAULT '{}',
264                version INTEGER NOT NULL DEFAULT 1,
265                prev_id TEXT
266            );
267            CREATE TABLE versions (
268                target_kind TEXT NOT NULL,
269                target_id TEXT NOT NULL,
270                version INTEGER NOT NULL,
271                diff TEXT NOT NULL DEFAULT '{}',
272                ts INTEGER NOT NULL,
273                author TEXT,
274                PRIMARY KEY (target_kind, target_id, version)
275            );
276            CREATE TABLE specifications (
277                name TEXT PRIMARY KEY,
278                expr_json TEXT NOT NULL,
279                created_at INTEGER NOT NULL DEFAULT 0
280            );
281            CREATE TABLE projections (
282                name TEXT PRIMARY KEY,
283                spec_ref TEXT NOT NULL,
284                template TEXT NOT NULL,
285                target_form TEXT NOT NULL,
286                created_at INTEGER NOT NULL DEFAULT 0,
287                template_engine TEXT,
288                projection_kind TEXT,
289                projection_config TEXT
290            );
291            INSERT INTO type_registry(name, kind) VALUES
292                ('persona', 'node'), ('outline_node', 'node'), ('routes_to', 'edge');
293            INSERT INTO nodes(id, type, version) VALUES
294                ('alpha', 'persona', 1),
295                ('alpha.active', 'outline_node', 1);
296            INSERT INTO edges(id, src_node, tgt_node, kind, version) VALUES
297                ('e.alpha.active', 'alpha', 'alpha.active', 'routes_to', 1);
298            INSERT INTO specifications(name, expr_json) VALUES
299                ('active_personas', '{"TypeIs":"persona"}');
300            INSERT INTO projections(name, spec_ref, template, target_form) VALUES
301                ('alpha.section.active', 'active_personas', '## {{name}}', 'markdown');
302            "#,
303        )
304        .unwrap();
305    }
306
307    #[test]
308    fn runner_up_applies_all_pending_then_skips_on_rerun() {
309        let conn = Connection::open_in_memory().unwrap();
310        seed_v0_6(&conn);
311        let runner = Runner::new(&conn);
312
313        let s = runner.status().unwrap();
314        assert_eq!(s.applied.len(), 0);
315        assert_eq!(s.pending.len(), 3);
316
317        let now = runner.up(None).unwrap();
318        assert_eq!(now.len(), 3);
319
320        let s2 = runner.status().unwrap();
321        assert_eq!(s2.applied.len(), 3);
322        assert!(s2.pending.is_empty());
323
324        // Re-run is a no-op.
325        let now2 = runner.up(None).unwrap();
326        assert!(now2.is_empty());
327
328        // Post-state sanity.
329        let bad_nodes: i64 = conn
330            .query_row(
331                "SELECT COUNT(*) FROM nodes WHERE length(id) != 26",
332                [],
333                |r| r.get(0),
334            )
335            .unwrap();
336        assert_eq!(bad_nodes, 0);
337        let bad_specs: i64 = conn
338            .query_row(
339                "SELECT COUNT(*) FROM specifications WHERE length(id) != 26",
340                [],
341                |r| r.get(0),
342            )
343            .unwrap();
344        assert_eq!(bad_specs, 0);
345    }
346
347    #[test]
348    fn runner_apply_specific_id_then_repeat_errors() {
349        let conn = Connection::open_in_memory().unwrap();
350        seed_v0_6(&conn);
351        let runner = Runner::new(&conn);
352
353        runner.apply("001_node_id_ulid").unwrap();
354        let s = runner.status().unwrap();
355        assert_eq!(s.applied.len(), 1);
356        assert_eq!(s.pending.len(), 2);
357
358        // Second time errors.
359        assert!(runner.apply("001_node_id_ulid").is_err());
360
361        // Unknown id errors.
362        assert!(runner.apply("999_nope").is_err());
363    }
364
365    #[test]
366    fn runner_target_stops_at_named_migration() {
367        let conn = Connection::open_in_memory().unwrap();
368        seed_v0_6(&conn);
369        let runner = Runner::new(&conn);
370        let now = runner.up(Some("001_node_id_ulid")).unwrap();
371        assert_eq!(now.len(), 1);
372        assert_eq!(now[0].version, "001_node_id_ulid");
373        let s = runner.status().unwrap();
374        assert_eq!(s.pending.len(), 2);
375        assert_eq!(s.pending[0].id(), "002_registry_id_ulid");
376        assert_eq!(s.pending[1].id(), "003_bundle_installs_fk_relax");
377    }
378}
persona_wire/migrations/mod.rs

persona_wire/migrations/
mod.rs
