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

minco_sqlx_postgres/
lib.rs
