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 = result
40            .rows
41            .iter()
42            .map(|row| {
43                row.get("version")
44                    .and_then(serde_json::Value::as_i64)
45                    .and_then(|v| u32::try_from(v).ok())
46                    .ok_or_else(|| LoaderError::MigrationFailed {
47                        extension: ext_id.to_owned(),
48                        message: "extension_migrations row has a malformed `version` column"
49                            .to_owned(),
50                    })
51            })
52            .collect::<Result<Vec<u32>, LoaderError>>()?;
53
54        if versions.is_empty() {
55            return Ok(MigrationResult::default());
56        }
57
58        let migrations = extension.migrations();
59        let mut migrations_run = 0;
60
61        for version in versions {
62            self.revert_version(ext_id, version, &migrations).await?;
63            migrations_run += 1;
64        }
65
66        Ok(MigrationResult {
67            migrations_run,
68            migrations_skipped: 0,
69        })
70    }
71
72    async fn revert_version(
73        &self,
74        ext_id: &str,
75        version: u32,
76        migrations: &[Migration],
77    ) -> Result<(), LoaderError> {
78        let migration = migrations
79            .iter()
80            .find(|m| m.version == version)
81            .ok_or_else(|| LoaderError::MigrationFailed {
82                extension: ext_id.to_owned(),
83                message: format!(
84                    "Cannot revert migration {version}: not declared in Extension::migrations()"
85                ),
86            })?;
87
88        if migration.tombstone {
89            return Err(LoaderError::MigrationFailed {
90                extension: ext_id.to_owned(),
91                message: format!(
92                    "Cannot revert migration {version} ('{}'): the slot is tombstoned — its file \
93                     was deleted, so there is no down SQL to run",
94                    migration.name
95                ),
96            });
97        }
98
99        let down_sql = migration
100            .down
101            .ok_or_else(|| LoaderError::MigrationNotReversible {
102                extension: ext_id.to_owned(),
103                version,
104            })?;
105
106        info!(
107            extension = %ext_id,
108            version = migration.version,
109            name = %migration.name,
110            "Reverting migration"
111        );
112
113        let statements = SqlExecutor::parse_sql_statements(down_sql).map_err(|e| {
114            LoaderError::MigrationFailed {
115                extension: ext_id.to_owned(),
116                message: format!(
117                    "Failed to parse down migration {} ({}): {e}",
118                    migration.version, migration.name
119                ),
120            }
121        })?;
122        let delete_params: [&dyn ToDbValue; 2] = [&ext_id, &version];
123        execute_statements_transactional(
124            self.db,
125            &statements,
126            ext_id,
127            migration,
128            Some(TrackingWrite {
129                sql: "DELETE FROM extension_migrations WHERE extension_id = $1 AND version = $2",
130                params: &delete_params,
131            }),
132        )
133        .await
134    }
135}
systemprompt_database/lifecycle/migrations/down.rs

systemprompt_database/lifecycle/migrations/
down.rs
