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

systemprompt_database/lifecycle/migrations/
status.rs
