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    pub ahead: Vec<u32>,
85}
86
87#[derive(Debug, Default, Clone, Copy)]
88pub struct MigrationResult {
89    pub migrations_run: usize,
90    pub migrations_skipped: usize,
91}
92
93#[derive(Debug)]
94pub struct MigrationStatus {
95    pub extension_id: ExtensionId,
96    pub total_defined: usize,
97    pub total_applied: usize,
98    pub pending_count: usize,
99    pub pending: Vec<Migration>,
100    pub applied: Vec<AppliedMigration>,
101}
102
103impl MigrationService<'_> {
104    pub async fn plan_pending(
105        &self,
106        extension: &dyn Extension,
107    ) -> Result<Vec<PendingMigration>, LoaderError> {
108        let ext_id = &ExtensionId::new(extension.metadata().id);
109        let defined = extension.migrations();
110
111        if defined.is_empty() {
112            return Ok(Vec::new());
113        }
114
115        self.ensure_migrations_table_exists().await?;
116        let applied_versions: HashSet<u32> = self
117            .get_applied_migrations(ext_id)
118            .await?
119            .into_iter()
120            .map(|m| m.version)
121            .collect();
122
123        Ok(defined
124            .into_iter()
125            .filter(|m| !m.tombstone && !applied_versions.contains(&m.version))
126            .map(|m| PendingMigration {
127                extension_id: ext_id.clone(),
128                version: m.version,
129                name: m.name.clone(),
130                sql: m.sql,
131                checksum: m.checksum(),
132                no_tx: m.no_transaction,
133                breaks_older: m.breaks_older(),
134            })
135            .collect())
136    }
137
138    pub async fn status(
139        &self,
140        extension: &dyn Extension,
141    ) -> Result<ExtensionMigrationStatus, LoaderError> {
142        let ext_id = &ExtensionId::new(extension.metadata().id);
143        let defined = extension.migrations();
144
145        self.ensure_migrations_table_exists().await?;
146        let applied = self.get_applied_migrations(ext_id).await?;
147
148        let applied_versions: HashSet<u32> = applied.iter().map(|m| m.version).collect();
149        let applied_rows: std::collections::HashMap<u32, &AppliedMigration> =
150            applied.iter().map(|m| (m.version, m)).collect();
151
152        let mut slots = SlotClassification::default();
153        for m in &defined {
154            let row = applied_rows.get(&m.version).copied();
155            slots.classify(ext_id, m, row, &applied_versions);
156        }
157        let SlotClassification {
158            pending,
159            drift,
160            slot_collisions,
161            tombstoned,
162        } = slots;
163
164        let ahead = super::ahead_versions(&applied, &defined);
165        let orphaned = super::orphaned_versions(&applied, &defined)
166            .into_iter()
167            .map(|version| OrphanedMigration {
168                extension_id: ext_id.clone(),
169                version,
170                name: applied
171                    .iter()
172                    .find(|m| m.version == version)
173                    .map_or_else(String::new, |m| m.name.clone()),
174            })
175            .collect();
176
177        Ok(ExtensionMigrationStatus {
178            extension_id: ext_id.clone(),
179            applied,
180            pending,
181            drift,
182            slot_collisions,
183            orphaned,
184            tombstoned,
185            ahead,
186        })
187    }
188
189    pub async fn get_migration_status(
190        &self,
191        extension: &dyn Extension,
192    ) -> Result<MigrationStatus, LoaderError> {
193        self.ensure_migrations_table_exists().await?;
194
195        let ext_id = &ExtensionId::new(extension.metadata().id);
196        let defined_migrations = extension.migrations();
197        let applied = self.get_applied_migrations(ext_id).await?;
198
199        let applied_versions: HashSet<u32> = applied.iter().map(|m| m.version).collect();
200
201        let pending: Vec<_> = defined_migrations
202            .iter()
203            .filter(|m| !m.tombstone && !applied_versions.contains(&m.version))
204            .cloned()
205            .collect();
206
207        Ok(MigrationStatus {
208            extension_id: ext_id.clone(),
209            total_defined: defined_migrations.len(),
210            total_applied: applied.len(),
211            pending_count: pending.len(),
212            pending,
213            applied,
214        })
215    }
216}
217
218#[derive(Default)]
219struct SlotClassification {
220    pending: Vec<PendingMigration>,
221    drift: Vec<ChecksumDrift>,
222    slot_collisions: Vec<SlotCollision>,
223    tombstoned: Vec<TombstonedSlot>,
224}
225
226impl SlotClassification {
227    fn classify(
228        &mut self,
229        ext_id: &ExtensionId,
230        m: &Migration,
231        row: Option<&AppliedMigration>,
232        applied_versions: &HashSet<u32>,
233    ) {
234        if m.tombstone {
235            self.tombstoned.push(TombstonedSlot {
236                extension_id: ext_id.clone(),
237                version: m.version,
238                name: m.name.clone(),
239                tracked: applied_versions.contains(&m.version),
240            });
241            return;
242        }
243        let current_checksum = m.checksum();
244        let Some(row) = row else {
245            self.pending.push(PendingMigration {
246                extension_id: ext_id.clone(),
247                version: m.version,
248                name: m.name.clone(),
249                sql: m.sql,
250                checksum: current_checksum,
251                no_tx: m.no_transaction,
252                breaks_older: m.breaks_older(),
253            });
254            return;
255        };
256        if row.name != m.name {
257            self.slot_collisions.push(SlotCollision {
258                extension_id: ext_id.clone(),
259                version: m.version,
260                stored_name: row.name.clone(),
261                current_name: m.name.clone(),
262            });
263        } else if !super::checksum_transition::matches_checksum(m, &row.checksum) {
264            self.drift.push(ChecksumDrift {
265                extension_id: ext_id.clone(),
266                version: m.version,
267                name: m.name.clone(),
268                stored_checksum: row.checksum.clone(),
269                current_checksum,
270            });
271        }
272    }
273}
systemprompt_database/lifecycle/migrations/status.rs

systemprompt_database/lifecycle/migrations/
status.rs
