minco_sqlx_postgres/
lib.rs

1//! Bounded `SQLx` `PostgreSQL` pools for local servers and serverless runtimes.
2#![forbid(unsafe_code)]
3
4use minco_db::{
5    AppliedMigration, DatabaseBackend, MigrationSet, SeedPlan, SeedTransaction, SeedVerification,
6    TargetState, resolve_seed_source, validate_seed_plan as validate_seed_model_plan,
7};
8use serde::{Deserialize, Serialize};
9pub use sqlx::PgPool;
10use sqlx::postgres::PgPoolOptions;
11use std::{
12    path::{Path, PathBuf},
13    time::Duration,
14};
15use thiserror::Error;
16
17pub mod audit_v2;
18pub mod jobs;
19pub mod plugin_adapters;
20
21const MINCO_PLAN_LOCK_ID: i64 = 0x4d49_4e43_4f5f_504c;
22
23#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
24pub struct PostgresPoolConfig {
25    pub url: String,
26    pub max_connections: u32,
27    pub acquire_timeout_seconds: u64,
28    pub idle_timeout_seconds: u64,
29}
30
31impl std::fmt::Debug for PostgresPoolConfig {
32    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33        formatter
34            .debug_struct("PostgresPoolConfig")
35            .field("url", &"[REDACTED DATABASE URL]")
36            .field("max_connections", &self.max_connections)
37            .field("acquire_timeout_seconds", &self.acquire_timeout_seconds)
38            .field("idle_timeout_seconds", &self.idle_timeout_seconds)
39            .finish()
40    }
41}
42
43impl PostgresPoolConfig {
44    pub fn serverless(url: impl Into<String>) -> Self {
45        Self {
46            url: url.into(),
47            max_connections: 2,
48            acquire_timeout_seconds: 5,
49            idle_timeout_seconds: 60,
50        }
51    }
52    pub fn validate(&self) -> Result<(), PostgresError> {
53        if self.url.trim().is_empty() {
54            return Err(PostgresError::InvalidConfig("database URL is empty".into()));
55        }
56        if self.max_connections == 0 {
57            return Err(PostgresError::InvalidConfig(
58                "max_connections must be at least 1".into(),
59            ));
60        }
61        Ok(())
62    }
63}
64
65pub async fn connect(config: &PostgresPoolConfig) -> Result<PgPool, PostgresError> {
66    config.validate()?;
67    Ok(PgPoolOptions::new()
68        .min_connections(0)
69        .max_connections(config.max_connections)
70        .acquire_timeout(Duration::from_secs(config.acquire_timeout_seconds))
71        .idle_timeout(Some(Duration::from_secs(config.idle_timeout_seconds)))
72        .connect(&config.url)
73        .await?)
74}
75
76pub fn connect_lazy(config: &PostgresPoolConfig) -> Result<PgPool, PostgresError> {
77    config.validate()?;
78    Ok(PgPoolOptions::new()
79        .min_connections(0)
80        .max_connections(config.max_connections)
81        .acquire_timeout(Duration::from_secs(config.acquire_timeout_seconds))
82        .idle_timeout(Some(Duration::from_secs(config.idle_timeout_seconds)))
83        .connect_lazy(&config.url)?)
84}
85
86pub async fn migrate(pool: &PgPool, path: impl AsRef<Path>) -> Result<(), PostgresError> {
87    let migrator = sqlx::migrate::Migrator::new(path.as_ref()).await?;
88    migrator.run(pool).await?;
89    Ok(())
90}
91
92pub async fn migrate_with_history_table(
93    pool: &PgPool,
94    path: impl AsRef<Path>,
95    history_table: &'static str,
96) -> Result<(), PostgresError> {
97    validate_identifier(history_table, "migration history table")?;
98    let mut migrator = sqlx::migrate::Migrator::new(path.as_ref()).await?;
99    migrator.dangerous_set_table_name(history_table);
100    migrator.run(pool).await?;
101    Ok(())
102}
103
104pub async fn migration_target_state(
105    pool: &PgPool,
106    set: &MigrationSet,
107) -> Result<TargetState, PostgresError> {
108    validate_set(set)?;
109    if !table_exists(pool, &set.history_table).await? {
110        return Ok(TargetState::default());
111    }
112    let dirty_query = format!(
113        "SELECT version FROM {} WHERE success = false ORDER BY version LIMIT 1",
114        set.history_table
115    );
116    // The only interpolated token was accepted by `validate_identifier`.
117    let dirty_version = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(dirty_query))
118        .fetch_optional(pool)
119        .await?;
120    let applied_query = format!(
121        "SELECT version, checksum FROM {} WHERE success = true ORDER BY version",
122        set.history_table
123    );
124    // The only interpolated token was accepted by `validate_identifier`.
125    let applied = sqlx::query_as::<_, (i64, Vec<u8>)>(sqlx::AssertSqlSafe(applied_query))
126        .fetch_all(pool)
127        .await?
128        .into_iter()
129        .map(|(version, checksum)| AppliedMigration {
130            version,
131            sqlx_checksum_sha384: hex(&checksum),
132        })
133        .collect();
134    Ok(TargetState {
135        dirty_version,
136        applied,
137    })
138}
139
140pub async fn verify_migration_tables(
141    pool: &PgPool,
142    set: &MigrationSet,
143) -> Result<Vec<String>, PostgresError> {
144    validate_set(set)?;
145    let mut missing = Vec::new();
146    for table in &set.verify_tables {
147        if !table_exists(pool, table).await? {
148            missing.push(table.clone());
149        }
150    }
151    Ok(missing)
152}
153
154pub async fn apply_migration_set(
155    pool: &PgPool,
156    project_root: &Path,
157    set: &MigrationSet,
158) -> Result<(), PostgresError> {
159    apply_migration_plan(pool, project_root, std::slice::from_ref(set)).await
160}
161
162pub async fn apply_migration_plan(
163    pool: &PgPool,
164    project_root: &Path,
165    sets: &[MigrationSet],
166) -> Result<(), PostgresError> {
167    if sets.is_empty() {
168        return Err(PostgresError::InvalidConfig(
169            "migration plan contains no sets".into(),
170        ));
171    }
172    let mut migrators = Vec::with_capacity(sets.len());
173    for set in sets {
174        validate_set(set)?;
175        let root = migration_root(project_root, set)?;
176        let mut migrator = sqlx::migrate::Migrator::new(root).await?;
177        verify_resolved_migrations(&migrator, set)?;
178        migrator.dangerous_set_table_name(set.history_table.clone());
179        migrators.push(migrator);
180    }
181
182    let mut connection = pool.acquire().await?;
183    // SQLx's migrator returns early on dirty/checksum/execution failures before
184    // its normal unlock call. Closing this session on every exit guarantees
185    // that PostgreSQL releases both the plan and SQLx advisory locks.
186    connection.close_on_drop();
187    sqlx::query("SELECT pg_advisory_lock($1)")
188        .bind(MINCO_PLAN_LOCK_ID)
189        .execute(&mut *connection)
190        .await?;
191    for migrator in migrators {
192        migrator.run_direct(None, &mut *connection, false).await?;
193    }
194    let unlocked = sqlx::query_scalar::<_, bool>("SELECT pg_advisory_unlock($1)")
195        .bind(MINCO_PLAN_LOCK_ID)
196        .fetch_one(&mut *connection)
197        .await?;
198    if !unlocked {
199        return Err(PostgresError::PlanLockLost);
200    }
201    Ok(())
202}
203
204pub async fn apply_seed_plan(
205    pool: &PgPool,
206    project_root: &Path,
207    plan: &SeedPlan,
208) -> Result<(), PostgresError> {
209    validate_seed_plan(plan)?;
210    let sources = plan
211        .seeds
212        .iter()
213        .map(|seed| resolve_seed_source(project_root, seed))
214        .collect::<Result<Vec<_>, _>>()
215        .map_err(|error| PostgresError::SeedSource(error.to_string()))?;
216    match plan.seeds[0].transaction {
217        SeedTransaction::Required => {
218            let mut transaction = pool.begin().await?;
219            for source in sources {
220                sqlx::raw_sql(sqlx::AssertSqlSafe(source.apply_sql))
221                    .execute(&mut *transaction)
222                    .await?;
223            }
224            transaction.commit().await?;
225        }
226        SeedTransaction::Autocommit => {
227            for source in sources {
228                sqlx::raw_sql(sqlx::AssertSqlSafe(source.apply_sql))
229                    .execute(pool)
230                    .await?;
231            }
232        }
233    }
234    Ok(())
235}
236
237pub async fn verify_seed_plan(
238    pool: &PgPool,
239    project_root: &Path,
240    plan: &SeedPlan,
241) -> Result<Vec<SeedVerification>, PostgresError> {
242    validate_seed_plan(plan)?;
243    let mut transaction = pool.begin().await?;
244    sqlx::query("SET TRANSACTION READ ONLY")
245        .execute(&mut *transaction)
246        .await?;
247    let mut verification = Vec::with_capacity(plan.seeds.len());
248    for seed in &plan.seeds {
249        let source = resolve_seed_source(project_root, seed)
250            .map_err(|error| PostgresError::SeedSource(error.to_string()))?;
251        let rows = sqlx::query_scalar::<_, bool>(sqlx::AssertSqlSafe(source.verify_sql))
252            .fetch_all(&mut *transaction)
253            .await?;
254        if rows.len() != 1 {
255            return Err(PostgresError::InvalidConfig(format!(
256                "seed {} verification must return exactly one boolean row",
257                seed.id
258            )));
259        }
260        verification.push(SeedVerification {
261            seed_id: seed.id.clone(),
262            verified: rows[0],
263        });
264    }
265    transaction.rollback().await?;
266    Ok(verification)
267}
268
269pub async fn ready(pool: &PgPool) -> bool {
270    matches!(
271        sqlx::query_scalar::<_, i32>("SELECT 1")
272            .fetch_one(pool)
273            .await,
274        Ok(1)
275    )
276}
277
278fn validate_seed_plan(plan: &SeedPlan) -> Result<(), PostgresError> {
279    validate_seed_model_plan(plan).map_err(|error| PostgresError::SeedSource(error.to_string()))?;
280    if plan.seeds.is_empty() {
281        return Err(PostgresError::InvalidConfig(
282            "seed plan contains no seeds".into(),
283        ));
284    }
285    if plan
286        .seeds
287        .iter()
288        .any(|seed| seed.backend != DatabaseBackend::Postgres)
289    {
290        return Err(PostgresError::InvalidConfig(
291            "seed plan contains a non-PostgreSQL seed".into(),
292        ));
293    }
294    if plan
295        .seeds
296        .iter()
297        .any(|seed| seed.transaction != plan.seeds[0].transaction)
298    {
299        return Err(PostgresError::InvalidConfig(
300            "seed plan mixes transaction behaviors".into(),
301        ));
302    }
303    Ok(())
304}
305
306async fn table_exists(pool: &PgPool, table: &str) -> Result<bool, PostgresError> {
307    validate_identifier(table, "table")?;
308    Ok(sqlx::query_scalar::<_, bool>(
309        "SELECT EXISTS (
310            SELECT 1
311            FROM pg_catalog.pg_class
312            WHERE oid = to_regclass($1)
313              AND relkind IN ('r', 'p')
314        )",
315    )
316    .bind(table)
317    .fetch_one(pool)
318    .await?)
319}
320
321fn validate_set(set: &MigrationSet) -> Result<(), PostgresError> {
322    if set.backend != DatabaseBackend::Postgres {
323        return Err(PostgresError::InvalidConfig(format!(
324            "migration set {} targets a different database backend",
325            set.id
326        )));
327    }
328    validate_identifier(&set.history_table, "migration history table")?;
329    for table in &set.verify_tables {
330        validate_identifier(table, "verification table")?;
331    }
332    Ok(())
333}
334
335fn migration_root(project_root: &Path, set: &MigrationSet) -> Result<PathBuf, PostgresError> {
336    let project_root = project_root.canonicalize().map_err(PostgresError::Io)?;
337    if set.root.is_absolute() {
338        return Err(PostgresError::InvalidConfig(format!(
339            "migration set {} has an absolute source root",
340            set.id
341        )));
342    }
343    let root = project_root
344        .join(&set.root)
345        .canonicalize()
346        .map_err(PostgresError::Io)?;
347    if !root.starts_with(&project_root) {
348        return Err(PostgresError::InvalidConfig(format!(
349            "migration set {} source root escapes the project",
350            set.id
351        )));
352    }
353    Ok(root)
354}
355
356fn verify_resolved_migrations(
357    migrator: &sqlx::migrate::Migrator,
358    set: &MigrationSet,
359) -> Result<(), PostgresError> {
360    let resolved = migrator.iter().collect::<Vec<_>>();
361    if resolved.len() != set.migrations.len() {
362        return Err(PostgresError::SourceDrift(set.id.clone()));
363    }
364    for (resolved, expected) in resolved.iter().zip(&set.migrations) {
365        if resolved.version != expected.version
366            || hex(resolved.checksum.as_ref()) != expected.sqlx_checksum_sha384
367        {
368            return Err(PostgresError::SourceDrift(set.id.clone()));
369        }
370    }
371    Ok(())
372}
373
374fn hex(bytes: &[u8]) -> String {
375    const DIGITS: &[u8; 16] = b"0123456789abcdef";
376    let mut output = String::with_capacity(bytes.len() * 2);
377    for byte in bytes {
378        output.push(char::from(DIGITS[usize::from(byte >> 4)]));
379        output.push(char::from(DIGITS[usize::from(byte & 0x0f)]));
380    }
381    output
382}
383
384fn validate_identifier(value: &str, description: &str) -> Result<(), PostgresError> {
385    let mut bytes = value.bytes();
386    let valid_start = bytes
387        .next()
388        .is_some_and(|byte| byte.is_ascii_alphabetic() || byte == b'_');
389    if !valid_start
390        || value.len() > 63
391        || !bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
392    {
393        return Err(PostgresError::InvalidConfig(format!(
394            "{description} must be a PostgreSQL identifier of at most 63 ASCII characters"
395        )));
396    }
397    Ok(())
398}
399
400#[derive(Debug, Error)]
401pub enum PostgresError {
402    #[error("invalid PostgreSQL configuration: {0}")]
403    InvalidConfig(String),
404    #[error("PostgreSQL error: {0}")]
405    Sqlx(#[from] sqlx::Error),
406    #[error("PostgreSQL migration error: {0}")]
407    Migration(#[from] sqlx::migrate::MigrateError),
408    #[error("PostgreSQL migration source changed after planning for set {0}")]
409    SourceDrift(String),
410    #[error("PostgreSQL migration plan advisory lock was not held at unlock")]
411    PlanLockLost,
412    #[error("PostgreSQL migration filesystem operation failed: {0}")]
413    Io(#[from] std::io::Error),
414    #[error("PostgreSQL seed source validation failed: {0}")]
415    SeedSource(String),
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421    use minco_db::{MigrationState, compare_target, load_catalog};
422    use std::fs;
423    use tempfile::TempDir;
424    use uuid::Uuid;
425
426    #[test]
427    fn serverless_defaults_bound_connection_pressure() {
428        let config = PostgresPoolConfig::serverless("postgres://example.invalid/db");
429        assert_eq!(config.max_connections, 2);
430        assert_eq!(config.acquire_timeout_seconds, 5);
431    }
432
433    #[test]
434    fn pool_configuration_debug_redacts_database_credentials() {
435        let config =
436            PostgresPoolConfig::serverless("postgres://minco:secret-password@example.invalid/db");
437        let debug = format!("{config:?}");
438        assert!(!debug.contains("secret-password"));
439        assert!(!debug.contains("postgres://"));
440    }
441
442    #[tokio::test]
443    async fn migration_history_table_rejects_dynamic_sql_tokens() {
444        let config = PostgresPoolConfig::serverless("postgres://example.invalid/db");
445        let pool = connect_lazy(&config).expect("lazy pool");
446        let result =
447            migrate_with_history_table(&pool, Path::new("missing"), "_migrations;DROP").await;
448        assert!(matches!(result, Err(PostgresError::InvalidConfig(_))));
449    }
450
451    #[tokio::test]
452    async fn lifecycle_rejects_a_set_for_another_backend_before_connecting() {
453        let config = PostgresPoolConfig::serverless("postgres://example.invalid/db");
454        let pool = connect_lazy(&config).expect("lazy pool");
455        let set = MigrationSet {
456            id: "wrong-backend".into(),
457            owner: "application:test".into(),
458            backend: DatabaseBackend::Sqlite,
459            root: "migrations".into(),
460            history_table: "_minco_test_migrations".into(),
461            depends_on: Vec::new(),
462            verify_tables: vec!["example".into()],
463            digest: "digest".into(),
464            migrations: Vec::new(),
465        };
466
467        let error = migration_target_state(&pool, &set)
468            .await
469            .expect_err("backend mismatch must fail before database access");
470        assert!(matches!(error, PostgresError::InvalidConfig(_)));
471    }
472
473    #[tokio::test]
474    async fn lifecycle_migration_is_behavioral_when_postgres_is_configured() {
475        let Ok(url) = std::env::var("MINCO_TEST_POSTGRES_URL") else {
476            eprintln!("MINCO_TEST_POSTGRES_URL not set; PostgreSQL lifecycle proof skipped");
477            return;
478        };
479        let suffix = Uuid::new_v4().simple().to_string();
480        let table = format!("minco_lifecycle_{suffix}");
481        let history = format!("_minco_lifecycle_{suffix}");
482        let project = TempDir::new().expect("temporary migration project");
483        let migrations = project.path().join("migrations");
484        fs::create_dir(&migrations).expect("create migration directory");
485        fs::write(
486            migrations.join("0001_example.sql"),
487            format!("CREATE TABLE {table} (id BIGINT PRIMARY KEY);\n"),
488        )
489        .expect("write migration");
490        fs::write(
491            migrations.join(minco_db::MIGRATION_SET_MANIFEST),
492            format!(
493                concat!(
494                    "schema = 1\n",
495                    "id = \"test-postgres\"\n",
496                    "owner = \"application:test\"\n",
497                    "backend = \"postgres\"\n",
498                    "history_table = \"{}\"\n",
499                    "verify_tables = [\"{}\"]\n",
500                    "\n",
501                    "[[migration]]\n",
502                    "version = 1\n",
503                    "risk = \"additive\"\n",
504                    "reversible = false\n",
505                ),
506                history, table
507            ),
508        )
509        .expect("write lifecycle manifest");
510        let set = load_catalog(project.path(), &[Path::new("migrations").to_path_buf()])
511            .expect("load lifecycle catalog")
512            .sets
513            .into_iter()
514            .next()
515            .expect("migration set");
516        let pool = connect(&PostgresPoolConfig::serverless(url))
517            .await
518            .expect("connect PostgreSQL");
519
520        let before = migration_target_state(&pool, &set)
521            .await
522            .expect("read empty target state");
523        assert!(before.applied.is_empty());
524        let (first, second) = tokio::join!(
525            apply_migration_set(&pool, project.path(), &set),
526            apply_migration_set(&pool, project.path(), &set)
527        );
528        first.expect("apply first concurrent migration plan");
529        second.expect("apply second concurrent migration plan");
530        let after = migration_target_state(&pool, &set)
531            .await
532            .expect("read applied target state");
533        assert_eq!(
534            compare_target(&set, &after).entries[0].state,
535            MigrationState::Applied
536        );
537        assert!(
538            verify_migration_tables(&pool, &set)
539                .await
540                .expect("verify tables")
541                .is_empty()
542        );
543
544        sqlx::query(sqlx::AssertSqlSafe(format!("DROP TABLE IF EXISTS {table}")))
545            .execute(&pool)
546            .await
547            .expect("clean up verified table");
548        sqlx::query(sqlx::AssertSqlSafe(format!(
549            "CREATE VIEW {table} AS SELECT 1::BIGINT AS id"
550        )))
551        .execute(&pool)
552        .await
553        .expect("replace verified table with a view");
554        let missing_tables = verify_migration_tables(&pool, &set)
555            .await
556            .expect("reject view as a verification table");
557        assert_eq!(missing_tables.as_slice(), std::slice::from_ref(&table));
558        sqlx::query(sqlx::AssertSqlSafe(format!("DROP VIEW {table}")))
559            .execute(&pool)
560            .await
561            .expect("clean up verified view");
562        sqlx::query(sqlx::AssertSqlSafe(format!(
563            "DROP TABLE IF EXISTS {history}"
564        )))
565        .execute(&pool)
566        .await
567        .expect("clean up migration history");
568    }
569}
minco_sqlx_postgres/lib.rs

minco_sqlx_postgres/
lib.rs
