systemprompt_database/lifecycle/migrations/
mark_applied.rs

1//! Record an extension migration as applied without running its SQL.
2//!
3//! Recovers the partial-state case where a migration's schema changes are
4//! present in the database but no row exists in `extension_migrations` to
5//! track them. Distinct from checksum-drift repair, which reconciles rows
6//! that already exist: here, the operator asserts the migration is already
7//! applied; the service only computes the current checksum and writes the
8//! tracking row.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13use super::MigrationService;
14use systemprompt_extension::{Extension, LoaderError};
15use systemprompt_identifiers::ExtensionId;
16
17#[derive(Debug, Clone)]
18pub struct MarkAppliedOutcome {
19    pub extension_id: ExtensionId,
20    pub version: u32,
21    pub name: String,
22    pub checksum: String,
23}
24
25impl MigrationService<'_> {
26    pub async fn mark_applied(
27        &self,
28        extension: &dyn Extension,
29        version: u32,
30    ) -> Result<MarkAppliedOutcome, LoaderError> {
31        let ext_id = &ExtensionId::new(extension.metadata().id);
32
33        let migration = extension
34            .migrations()
35            .into_iter()
36            .find(|m| m.version == version)
37            .ok_or_else(|| LoaderError::MigrationFailed {
38                extension: ext_id.clone(),
39                message: format!(
40                    "Migration version {version} is not defined for extension '{ext_id}'"
41                ),
42            })?;
43
44        if migration.tombstone {
45            return Err(LoaderError::MigrationFailed {
46                extension: ext_id.clone(),
47                message: format!(
48                    "Migration {version} ('{}') is a tombstone: the slot is recorded as spent and \
49                     has no SQL, so there is nothing to mark applied",
50                    migration.name
51                ),
52            });
53        }
54
55        self.ensure_migrations_table_exists().await?;
56
57        let applied = self.get_applied_migrations(ext_id).await?;
58        if applied.iter().any(|m| m.version == version) {
59            return Err(LoaderError::MigrationFailed {
60                extension: ext_id.clone(),
61                message: format!(
62                    "Migration {version} ('{}') is already tracked as applied for extension \
63                     '{ext_id}'; nothing to do",
64                    migration.name
65                ),
66            });
67        }
68
69        let id = format!("{ext_id}_{:03}", migration.version);
70        let checksum = migration.checksum();
71
72        self.db
73            .execute(
74                &"INSERT INTO extension_migrations (id, extension_id, version, name, checksum) \
75                  VALUES ($1, $2, $3, $4, $5)",
76                &[&id, &ext_id, &migration.version, &migration.name, &checksum],
77            )
78            .await
79            .map_err(|e| LoaderError::MigrationStepFailed {
80                extension: ext_id.clone(),
81                context: "Failed to record migration as applied".to_owned(),
82                source: Box::new(e),
83            })?;
84
85        Ok(MarkAppliedOutcome {
86            extension_id: ext_id.clone(),
87            version: migration.version,
88            name: migration.name.clone(),
89            checksum,
90        })
91    }
92}
systemprompt_database/lifecycle/migrations/mark_applied.rs

systemprompt_database/lifecycle/migrations/
mark_applied.rs
