boson_backend_postgres/
lib.rs

1//! `PostgreSQL` [`QueueBackend`](boson_core::QueueBackend) for Boson.
2//!
3//! Choose at worker boot. Enable via the `boson` crate `postgres` feature.
4//!
5//! ## Entry points
6//!
7//! - [`PostgresQueueBackend::connect`] — open a pool and bootstrap schema
8//! - [`install_default_postgres_backend`] — register on the global [`QueueRouter`](boson_core::QueueRouter)
9
10mod bootstrap;
11
12use boson_backend_sql_common::SqlQueueBackend;
13use boson_core::Result;
14use sqlx::PgPool;
15
16pub use bootstrap::{
17    install_default_postgres_backend, install_isolated_postgres_backend, postgres_test_url,
18};
19
20/// PostgreSQL-backed queue backend.
21pub struct PostgresQueueBackend {
22    inner: SqlQueueBackend,
23}
24
25impl PostgresQueueBackend {
26    /// Connect to `PostgreSQL` at `url`.
27    ///
28    /// # Errors
29    ///
30    /// Returns an error when the pool cannot connect or schema bootstrap fails.
31    pub async fn new(url: &str) -> Result<Self> {
32        Self::connect(url).await
33    }
34
35    /// Connect using a `PostgreSQL` connection URL.
36    ///
37    /// # Examples
38    ///
39    /// ```rust,no_run
40    /// use boson_backend_postgres::PostgresQueueBackend;
41    ///
42    /// # async fn connect() -> boson_core::Result<()> {
43    /// let backend = PostgresQueueBackend::connect("postgres://localhost/boson").await?;
44    /// # Ok(())
45    /// # }
46    /// ```
47    ///
48    /// # Errors
49    ///
50    /// Returns an error when the pool cannot connect or schema bootstrap fails.
51    pub async fn connect(url: &str) -> Result<Self> {
52        let inner = SqlQueueBackend::connect_postgres(url).await?;
53        Ok(Self { inner })
54    }
55
56    /// Connect with an isolated schema (for parallel tests).
57    ///
58    /// # Errors
59    ///
60    /// Returns an error when schema creation, pool connect, or bootstrap fails.
61    pub async fn connect_isolated(url: &str, schema: &str) -> Result<Self> {
62        let inner = SqlQueueBackend::connect_postgres_isolated(url, schema).await?;
63        Ok(Self { inner })
64    }
65
66    /// Wrap an existing pool (schema bootstrap runs).
67    ///
68    /// # Errors
69    ///
70    /// Returns an error when schema bootstrap fails.
71    pub async fn from_pool(pool: PgPool) -> Result<Self> {
72        let inner = SqlQueueBackend::from_postgres_pool(pool).await?;
73        Ok(Self { inner })
74    }
75
76    /// Underlying connection pool.
77    ///
78    /// # Panics
79    ///
80    /// Panics if the inner pool is not `PostgreSQL` (internal invariant violation).
81    #[must_use]
82    pub fn pool(&self) -> &PgPool {
83        match self.inner.pool() {
84            boson_backend_sql_common::SqlPool::Postgres(pool) => pool,
85            boson_backend_sql_common::SqlPool::Sqlite(_) => {
86                panic!("postgres backend has non-postgres pool")
87            }
88        }
89    }
90}
91
92impl std::fmt::Debug for PostgresQueueBackend {
93    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
94        f.debug_struct("PostgresQueueBackend").finish_non_exhaustive()
95    }
96}
97
98boson_backend_sql_common::delegate_queue_backend!(PostgresQueueBackend, inner);
boson_backend_postgres/lib.rs

boson_backend_postgres/
lib.rs
