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

systemprompt_database/lifecycle/migrations/
status.rs
