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#[derive(Debug, Clone)]
11pub struct AppliedMigration {
12    pub extension_id: String,
13    pub version: u32,
14    pub name: String,
15    pub checksum: String,
16    pub applied_at: Option<String>,
17}
18
19#[derive(Debug, Clone)]
20pub struct PendingMigration {
21    pub extension_id: String,
22    pub version: u32,
23    pub name: String,
24    pub sql: &'static str,
25    pub checksum: String,
26    pub no_tx: bool,
27}
28
29#[derive(Debug, Clone)]
30pub struct ChecksumDrift {
31    pub extension_id: String,
32    pub version: u32,
33    pub name: String,
34    pub stored_checksum: String,
35    pub current_checksum: String,
36}
37
38#[derive(Debug, Clone, Default)]
39pub struct ExtensionMigrationStatus {
40    pub extension_id: String,
41    pub applied: Vec<AppliedMigration>,
42    pub pending: Vec<PendingMigration>,
43    pub drift: Vec<ChecksumDrift>,
44}
45
46#[derive(Debug, Default, Clone, Copy)]
47pub struct MigrationResult {
48    pub migrations_run: usize,
49    pub migrations_skipped: usize,
50}
51
52#[derive(Debug)]
53pub struct MigrationStatus {
54    pub extension_id: String,
55    pub total_defined: usize,
56    pub total_applied: usize,
57    pub pending_count: usize,
58    pub pending: Vec<Migration>,
59    pub applied: Vec<AppliedMigration>,
60}
61
62impl MigrationService<'_> {
63    pub async fn plan_pending(
64        &self,
65        extension: &dyn Extension,
66    ) -> Result<Vec<PendingMigration>, LoaderError> {
67        let ext_id = extension.metadata().id;
68        let defined = extension.migrations();
69
70        if defined.is_empty() {
71            return Ok(Vec::new());
72        }
73
74        self.ensure_migrations_table_exists().await?;
75        let applied_versions: HashSet<u32> = self
76            .get_applied_migrations(ext_id)
77            .await?
78            .into_iter()
79            .map(|m| m.version)
80            .collect();
81
82        Ok(defined
83            .into_iter()
84            .filter(|m| !applied_versions.contains(&m.version))
85            .map(|m| PendingMigration {
86                extension_id: ext_id.to_owned(),
87                version: m.version,
88                name: m.name.clone(),
89                sql: m.sql,
90                checksum: m.checksum(),
91                no_tx: false,
92            })
93            .collect())
94    }
95
96    pub async fn status(
97        &self,
98        extension: &dyn Extension,
99    ) -> Result<ExtensionMigrationStatus, LoaderError> {
100        let ext_id = extension.metadata().id;
101        let defined = extension.migrations();
102
103        self.ensure_migrations_table_exists().await?;
104        let applied = self.get_applied_migrations(ext_id).await?;
105
106        let applied_versions: HashSet<u32> = applied.iter().map(|m| m.version).collect();
107        let applied_checksums: std::collections::HashMap<u32, &str> = applied
108            .iter()
109            .map(|m| (m.version, m.checksum.as_str()))
110            .collect();
111
112        let mut pending = Vec::new();
113        let mut drift = Vec::new();
114
115        for m in &defined {
116            let current_checksum = m.checksum();
117            if applied_versions.contains(&m.version) {
118                if let Some(&stored_checksum) = applied_checksums.get(&m.version)
119                    && stored_checksum != current_checksum
120                {
121                    drift.push(ChecksumDrift {
122                        extension_id: ext_id.to_owned(),
123                        version: m.version,
124                        name: m.name.clone(),
125                        stored_checksum: stored_checksum.to_owned(),
126                        current_checksum,
127                    });
128                }
129            } else {
130                pending.push(PendingMigration {
131                    extension_id: ext_id.to_owned(),
132                    version: m.version,
133                    name: m.name.clone(),
134                    sql: m.sql,
135                    checksum: current_checksum,
136                    no_tx: false,
137                });
138            }
139        }
140
141        Ok(ExtensionMigrationStatus {
142            extension_id: ext_id.to_owned(),
143            applied,
144            pending,
145            drift,
146        })
147    }
148
149    pub async fn get_migration_status(
150        &self,
151        extension: &dyn Extension,
152    ) -> Result<MigrationStatus, LoaderError> {
153        self.ensure_migrations_table_exists().await?;
154
155        let ext_id = extension.metadata().id;
156        let defined_migrations = extension.migrations();
157        let applied = self.get_applied_migrations(ext_id).await?;
158
159        let applied_versions: HashSet<u32> = applied.iter().map(|m| m.version).collect();
160
161        let pending: Vec<_> = defined_migrations
162            .iter()
163            .filter(|m| !applied_versions.contains(&m.version))
164            .cloned()
165            .collect();
166
167        Ok(MigrationStatus {
168            extension_id: ext_id.to_owned(),
169            total_defined: defined_migrations.len(),
170            total_applied: applied.len(),
171            pending_count: pending.len(),
172            pending,
173            applied,
174        })
175    }
176}
systemprompt_database/lifecycle/migrations/status.rs

systemprompt_database/lifecycle/migrations/
status.rs
