systemprompt_database/lifecycle/migrations/
stamp.rs

1//! Fresh-install baseline stamping.
2//!
3//! The declarative schema (`schema/*.sql`) is the baseline: a fresh database
4//! reaches target shape from the structural/dependent DDL alone, so its
5//! migrations carry no information and must not execute. [`MigrationService::
6//! assess_freshness`] decides, before any DDL has run, whether an extension is
7//! landing on a fresh database; [`MigrationService::stamp_all_migrations`]
8//! then records every defined migration in `extension_migrations` without
9//! executing its SQL. Established databases (any tracking history, or any
10//! owned table already present) take the normal incremental path.
11//!
12//! Copyright (c) systemprompt.io — Business Source License 1.1.
13//! See <https://systemprompt.io> for licensing details.
14
15use super::MigrationService;
16use systemprompt_extension::{Extension, LoaderError};
17use tracing::{info, warn};
18
19#[derive(Debug, Clone, Copy)]
20pub struct FreshnessCheck {
21    pub no_history: bool,
22    pub tables_present: usize,
23    pub tables_total: usize,
24}
25
26impl FreshnessCheck {
27    #[must_use]
28    pub const fn is_fresh(&self) -> bool {
29        self.no_history && self.tables_present == 0
30    }
31}
32
33impl MigrationService<'_> {
34    pub async fn assess_freshness(
35        &self,
36        extension_id: &str,
37        owned_tables: &[String],
38    ) -> Result<FreshnessCheck, LoaderError> {
39        self.ensure_migrations_table_exists().await?;
40
41        let no_history = self.get_applied_migrations(extension_id).await?.is_empty();
42
43        let mut tables_present = 0usize;
44        for table in owned_tables {
45            let (schema, name) = table.split_once('.').unwrap_or(("public", table.as_str()));
46            let result = self
47                .db
48                .query_raw_with(
49                    &"SELECT 1 AS present FROM information_schema.tables WHERE table_schema = $1 \
50                      AND table_name = $2",
51                    &[&schema, &name],
52                )
53                .await
54                .map_err(|e| LoaderError::MigrationFailed {
55                    extension: extension_id.to_owned(),
56                    message: format!("Failed to check for existing table '{table}': {e}"),
57                })?;
58            if !result.rows.is_empty() {
59                tables_present += 1;
60            }
61        }
62
63        let check = FreshnessCheck {
64            no_history,
65            tables_present,
66            tables_total: owned_tables.len(),
67        };
68
69        if check.no_history && check.tables_present > 0 && check.tables_present < check.tables_total
70        {
71            warn!(
72                extension = %extension_id,
73                tables_present = check.tables_present,
74                tables_total = check.tables_total,
75                "Extension has no migration history but some owned tables already exist; \
76                 treating as an established database and executing migrations normally"
77            );
78        }
79
80        Ok(check)
81    }
82
83    pub async fn stamp_all_migrations(
84        &self,
85        extension: &dyn Extension,
86    ) -> Result<u32, LoaderError> {
87        let ext_id = extension.metadata().id;
88        let migrations = extension.migrations();
89
90        if migrations.is_empty() {
91            return Ok(0);
92        }
93
94        let mut tx =
95            self.db
96                .begin_transaction()
97                .await
98                .map_err(|e| LoaderError::MigrationFailed {
99                    extension: ext_id.to_owned(),
100                    message: format!("Failed to begin baseline stamp transaction: {e}"),
101                })?;
102
103        let mut stamped = 0u32;
104        for migration in &migrations {
105            let id = format!("{}_{:03}", ext_id, migration.version);
106            let checksum = migration.checksum();
107            if let Err(e) = tx
108                .execute(
109                    &"INSERT INTO extension_migrations (id, extension_id, version, name, \
110                      checksum) VALUES ($1, $2, $3, $4, $5)",
111                    &[&id, &ext_id, &migration.version, &migration.name, &checksum],
112                )
113                .await
114            {
115                let rollback_note = match tx.rollback().await {
116                    Ok(()) => String::new(),
117                    Err(rb) => format!(" (rollback also failed: {rb})"),
118                };
119                return Err(LoaderError::MigrationFailed {
120                    extension: ext_id.to_owned(),
121                    message: format!(
122                        "Failed to stamp migration {} ({}) as applied: {e}{rollback_note}",
123                        migration.version, migration.name
124                    ),
125                });
126            }
127            stamped += 1;
128        }
129
130        tx.commit()
131            .await
132            .map_err(|e| LoaderError::MigrationFailed {
133                extension: ext_id.to_owned(),
134                message: format!("Failed to commit baseline stamp: {e}"),
135            })?;
136
137        info!(
138            extension = %ext_id,
139            migrations_stamped = stamped,
140            "Fresh install: stamped migrations as baseline without executing them"
141        );
142
143        Ok(stamped)
144    }
145}
systemprompt_database/lifecycle/migrations/stamp.rs

systemprompt_database/lifecycle/migrations/
stamp.rs
