systemprompt_database/lifecycle/migrations/
down.rs

1//! Reverting applied migrations via their declared `down` SQL.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use super::exec::{TrackingWrite, execute_statements_transactional};
7use super::{MigrationResult, MigrationService};
8use crate::services::SqlExecutor;
9use systemprompt_extension::{Extension, LoaderError, Migration};
10use systemprompt_identifiers::ToDbValue;
11use tracing::info;
12
13impl MigrationService<'_> {
14    pub async fn run_down_migrations(
15        &self,
16        extension: &dyn Extension,
17        count: u32,
18    ) -> Result<MigrationResult, LoaderError> {
19        if count == 0 {
20            return Ok(MigrationResult::default());
21        }
22
23        let ext_id = extension.metadata().id;
24        self.ensure_migrations_table_exists().await?;
25
26        let result = self
27            .db
28            .query_raw_with(
29                &"SELECT version FROM extension_migrations WHERE extension_id = $1 ORDER BY \
30                  version DESC LIMIT $2",
31                &[&ext_id, &count],
32            )
33            .await
34            .map_err(|e| LoaderError::MigrationFailed {
35                extension: ext_id.to_owned(),
36                message: format!("Failed to query applied migrations for revert: {e}"),
37            })?;
38
39        let versions: Vec<u32> = result
40            .rows
41            .iter()
42            .filter_map(|row| row.get("version")?.as_i64().map(|v| v as u32))
43            .collect();
44
45        if versions.is_empty() {
46            return Ok(MigrationResult::default());
47        }
48
49        let migrations = extension.migrations();
50        let mut migrations_run = 0;
51
52        for version in versions {
53            self.revert_version(ext_id, version, &migrations).await?;
54            migrations_run += 1;
55        }
56
57        Ok(MigrationResult {
58            migrations_run,
59            migrations_skipped: 0,
60        })
61    }
62
63    async fn revert_version(
64        &self,
65        ext_id: &str,
66        version: u32,
67        migrations: &[Migration],
68    ) -> Result<(), LoaderError> {
69        let migration = migrations
70            .iter()
71            .find(|m| m.version == version)
72            .ok_or_else(|| LoaderError::MigrationFailed {
73                extension: ext_id.to_owned(),
74                message: format!(
75                    "Cannot revert migration {version}: not declared in Extension::migrations()"
76                ),
77            })?;
78
79        let down_sql = migration
80            .down
81            .ok_or_else(|| LoaderError::MigrationNotReversible {
82                extension: ext_id.to_owned(),
83                version,
84            })?;
85
86        info!(
87            extension = %ext_id,
88            version = migration.version,
89            name = %migration.name,
90            "Reverting migration"
91        );
92
93        let statements = SqlExecutor::parse_sql_statements(down_sql).map_err(|e| {
94            LoaderError::MigrationFailed {
95                extension: ext_id.to_owned(),
96                message: format!(
97                    "Failed to parse down migration {} ({}): {e}",
98                    migration.version, migration.name
99                ),
100            }
101        })?;
102        let delete_params: [&dyn ToDbValue; 2] = [&ext_id, &version];
103        execute_statements_transactional(
104            self.db,
105            &statements,
106            ext_id,
107            migration,
108            Some(TrackingWrite {
109                sql: "DELETE FROM extension_migrations WHERE extension_id = $1 AND version = $2",
110                params: &delete_params,
111            }),
112        )
113        .await
114    }
115}
systemprompt_database/lifecycle/migrations/down.rs

systemprompt_database/lifecycle/migrations/
down.rs
