systemprompt_database/lifecycle/migrations/
status.rs

1//! Migration status and plan queries, plus the value types they return.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use super::MigrationService;
7use std::collections::HashSet;
8use systemprompt_extension::{Extension, LoaderError, Migration};
9use systemprompt_identifiers::ExtensionId;
10
11/// A recorded migration.
12#[derive(Debug, Clone)]
13pub struct AppliedMigration {
14    pub extension_id: ExtensionId,
15    pub version: u32,
16    pub name: String,
17    pub checksum: String,
18    pub applied_at: Option<String>,
19}
20
21#[derive(Debug, Clone)]
22pub struct PendingMigration {
23    pub extension_id: ExtensionId,
24    pub version: u32,
25    pub name: String,
26    pub sql: &'static str,
27    pub checksum: String,
28    pub no_tx: bool,
29    pub breaks_older: bool,
30}
31
32/// An applied migration whose slot the extension no longer declares at all.
33///
34/// The file was deleted without leaving a `.tombstone`, so the number reads as
35/// free in the tree while every established database has spent it.
36#[derive(Debug, Clone)]
37pub struct OrphanedMigration {
38    pub extension_id: ExtensionId,
39    pub version: u32,
40    pub name: String,
41}
42
43/// A slot declared spent by a `.tombstone` file.
44#[derive(Debug, Clone)]
45pub struct TombstonedSlot {
46    pub extension_id: ExtensionId,
47    pub version: u32,
48    pub name: String,
49    pub tracked: bool,
50}
51
52/// An applied migration whose slot is now occupied by a differently-named file.
53///
54/// This is not drift: drift means the same migration was edited in place. A
55/// name mismatch means the number was reused by a different migration, so the
56/// recorded row and the file on disk describe two different things and neither
57/// checksum tells the truth about the database.
58#[derive(Debug, Clone)]
59pub struct SlotCollision {
60    pub extension_id: ExtensionId,
61    pub version: u32,
62    pub stored_name: String,
63    pub current_name: String,
64}
65
66#[derive(Debug, Clone)]
67pub struct ChecksumDrift {
68    pub extension_id: ExtensionId,
69    pub version: u32,
70    pub name: String,
71    pub stored_checksum: String,
72    pub current_checksum: String,
73}
74
75#[derive(Debug, Clone)]
76pub struct ExtensionMigrationStatus {
77    pub extension_id: ExtensionId,
78    pub applied: Vec<AppliedMigration>,
79    pub pending: Vec<PendingMigration>,
80    pub drift: Vec<ChecksumDrift>,
81    pub slot_collisions: Vec<SlotCollision>,
82    pub orphaned: Vec<OrphanedMigration>,
83    pub tombstoned: Vec<TombstonedSlot>,
84}
85
86#[derive(Debug, Default, Clone, Copy)]
87pub struct MigrationResult {
88    pub migrations_run: usize,
89    pub migrations_skipped: usize,
90}
91
92#[derive(Debug)]
93pub struct MigrationStatus {
94    pub extension_id: ExtensionId,
95    pub total_defined: usize,
96    pub total_applied: usize,
97    pub pending_count: usize,
98    pub pending: Vec<Migration>,
99    pub applied: Vec<AppliedMigration>,
100}
101
102impl MigrationService<'_> {
103    pub async fn plan_pending(
104        &self,
105        extension: &dyn Extension,
106    ) -> Result<Vec<PendingMigration>, LoaderError> {
107        let ext_id = &ExtensionId::new(extension.metadata().id);
108        let defined = extension.migrations();
109
110        if defined.is_empty() {
111            return Ok(Vec::new());
112        }
113
114        self.ensure_migrations_table_exists().await?;
115        let applied_versions: HashSet<u32> = self
116            .get_applied_migrations(ext_id)
117            .await?
118            .into_iter()
119            .map(|m| m.version)
120            .collect();
121
122        Ok(defined
123            .into_iter()
124            .filter(|m| !m.tombstone && !applied_versions.contains(&m.version))
125            .map(|m| PendingMigration {
126                extension_id: ext_id.clone(),
127                version: m.version,
128                name: m.name.clone(),
129                sql: m.sql,
130                checksum: m.checksum(),
131                no_tx: m.no_transaction,
132                breaks_older: m.breaks_older(),
133            })
134            .collect())
135    }
136
137    pub async fn status(
138        &self,
139        extension: &dyn Extension,
140    ) -> Result<ExtensionMigrationStatus, LoaderError> {
141        let ext_id = &ExtensionId::new(extension.metadata().id);
142        let defined = extension.migrations();
143
144        self.ensure_migrations_table_exists().await?;
145        let applied = self.get_applied_migrations(ext_id).await?;
146
147        let applied_versions: HashSet<u32> = applied.iter().map(|m| m.version).collect();
148        let applied_rows: std::collections::HashMap<u32, &AppliedMigration> =
149            applied.iter().map(|m| (m.version, m)).collect();
150
151        let mut slots = SlotClassification::default();
152        for m in &defined {
153            let row = applied_rows.get(&m.version).copied();
154            slots.classify(ext_id, m, row, &applied_versions);
155        }
156        let SlotClassification {
157            pending,
158            drift,
159            slot_collisions,
160            tombstoned,
161        } = slots;
162
163        let orphaned = super::orphaned_versions(&applied, &defined)
164            .into_iter()
165            .map(|version| OrphanedMigration {
166                extension_id: ext_id.clone(),
167                version,
168                name: applied
169                    .iter()
170                    .find(|m| m.version == version)
171                    .map_or_else(String::new, |m| m.name.clone()),
172            })
173            .collect();
174
175        Ok(ExtensionMigrationStatus {
176            extension_id: ext_id.clone(),
177            applied,
178            pending,
179            drift,
180            slot_collisions,
181            orphaned,
182            tombstoned,
183        })
184    }
185
186    pub async fn get_migration_status(
187        &self,
188        extension: &dyn Extension,
189    ) -> Result<MigrationStatus, LoaderError> {
190        self.ensure_migrations_table_exists().await?;
191
192        let ext_id = &ExtensionId::new(extension.metadata().id);
193        let defined_migrations = extension.migrations();
194        let applied = self.get_applied_migrations(ext_id).await?;
195
196        let applied_versions: HashSet<u32> = applied.iter().map(|m| m.version).collect();
197
198        let pending: Vec<_> = defined_migrations
199            .iter()
200            .filter(|m| !m.tombstone && !applied_versions.contains(&m.version))
201            .cloned()
202            .collect();
203
204        Ok(MigrationStatus {
205            extension_id: ext_id.clone(),
206            total_defined: defined_migrations.len(),
207            total_applied: applied.len(),
208            pending_count: pending.len(),
209            pending,
210            applied,
211        })
212    }
213}
214
215#[derive(Default)]
216struct SlotClassification {
217    pending: Vec<PendingMigration>,
218    drift: Vec<ChecksumDrift>,
219    slot_collisions: Vec<SlotCollision>,
220    tombstoned: Vec<TombstonedSlot>,
221}
222
223impl SlotClassification {
224    fn classify(
225        &mut self,
226        ext_id: &ExtensionId,
227        m: &Migration,
228        row: Option<&AppliedMigration>,
229        applied_versions: &HashSet<u32>,
230    ) {
231        if m.tombstone {
232            self.tombstoned.push(TombstonedSlot {
233                extension_id: ext_id.clone(),
234                version: m.version,
235                name: m.name.clone(),
236                tracked: applied_versions.contains(&m.version),
237            });
238            return;
239        }
240        let current_checksum = m.checksum();
241        let Some(row) = row else {
242            self.pending.push(PendingMigration {
243                extension_id: ext_id.clone(),
244                version: m.version,
245                name: m.name.clone(),
246                sql: m.sql,
247                checksum: current_checksum,
248                no_tx: m.no_transaction,
249                breaks_older: m.breaks_older(),
250            });
251            return;
252        };
253        if row.name != m.name {
254            self.slot_collisions.push(SlotCollision {
255                extension_id: ext_id.clone(),
256                version: m.version,
257                stored_name: row.name.clone(),
258                current_name: m.name.clone(),
259            });
260        } else if !super::checksum_transition::matches_checksum(m, &row.checksum) {
261            self.drift.push(ChecksumDrift {
262                extension_id: ext_id.clone(),
263                version: m.version,
264                name: m.name.clone(),
265                stored_checksum: row.checksum.clone(),
266                current_checksum,
267            });
268        }
269    }
270}
systemprompt_database/lifecycle/migrations/status.rs

systemprompt_database/lifecycle/migrations/
status.rs
