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        if migration.tombstone {
80            return Err(LoaderError::MigrationFailed {
81                extension: ext_id.to_owned(),
82                message: format!(
83                    "Cannot revert migration {version} ('{}'): the slot is tombstoned — its file \
84                     was deleted, so there is no down SQL to run",
85                    migration.name
86                ),
87            });
88        }
89
90        let down_sql = migration
91            .down
92            .ok_or_else(|| LoaderError::MigrationNotReversible {
93                extension: ext_id.to_owned(),
94                version,
95            })?;
96
97        info!(
98            extension = %ext_id,
99            version = migration.version,
100            name = %migration.name,
101            "Reverting migration"
102        );
103
104        let statements = SqlExecutor::parse_sql_statements(down_sql).map_err(|e| {
105            LoaderError::MigrationFailed {
106                extension: ext_id.to_owned(),
107                message: format!(
108                    "Failed to parse down migration {} ({}): {e}",
109                    migration.version, migration.name
110                ),
111            }
112        })?;
113        let delete_params: [&dyn ToDbValue; 2] = [&ext_id, &version];
114        execute_statements_transactional(
115            self.db,
116            &statements,
117            ext_id,
118            migration,
119            Some(TrackingWrite {
120                sql: "DELETE FROM extension_migrations WHERE extension_id = $1 AND version = $2",
121                params: &delete_params,
122            }),
123        )
124        .await
125    }
126}
systemprompt_database/lifecycle/migrations/down.rs

systemprompt_database/lifecycle/migrations/
down.rs
