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

minco_sqlx_postgres/
lib.rs
