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

systemprompt_database/lifecycle/migrations/
down.rs
