dovecote_sqlx_postgres/
lib.rs

1//! PostgreSQL schema and SQLx boundary for Dovecote.
2//!
3//! This crate publishes versioned PostgreSQL migration artifacts and implements
4//! the caller-transaction-bound enqueue, schema verification, leased lifecycle
5//! operations, and live and finite snapshot paging for Dovecote. The locking,
6//! database-time, fencing, and rollback contracts are covered by repository
7//! tests; release advertisement remains subject to the published support matrix
8//! and release gates.
9
10mod enqueue;
11mod error;
12mod finalize;
13mod import;
14mod lifecycle;
15mod migration;
16mod page;
17mod schema;
18
19pub use enqueue::enqueue;
20pub use error::{
21    ClaimError, EnqueueError, FinalizeError, ImportError, MutationError, PageError, SchemaError,
22    TransientKind,
23};
24pub use finalize::finalize_pending_delivery_for_migration;
25pub use import::import_for_migration;
26pub use lifecycle::{ack, claim, quarantine, release, renew, retry};
27pub use migration::{
28    CrateVersion, MIGRATIONS, Migration, MigrationCompatibility, MigrationCompatibilityError,
29    SCHEMA_VERSION,
30};
31pub use page::{SnapshotPager, begin_snapshot, page};
32pub use schema::check_schema;
33
34use dovecote::{EnqueueOutcome, FinalizeOutcome, ImportOutcome, ImportedDeliveryState, NewEvent};
35use sqlx::{PgPool, Postgres, Transaction};
36
37/// PostgreSQL adapter for Dovecote's durable event and delivery schema.
38#[derive(Clone)]
39pub struct PostgresDovecote {
40    pool: PgPool,
41}
42
43impl PostgresDovecote {
44    /// Creates an adapter using the supplied SQLx pool.
45    pub fn new(pool: PgPool) -> Self {
46        Self { pool }
47    }
48
49    /// Borrows the pool used by this adapter.
50    pub fn pool(&self) -> &PgPool {
51        &self.pool
52    }
53
54    /// Enqueues an event in the caller-owned transaction.
55    pub async fn enqueue<'c>(
56        &self,
57        transaction: &mut Transaction<'c, Postgres>,
58        event: NewEvent,
59    ) -> Result<EnqueueOutcome, EnqueueError> {
60        enqueue(transaction, event).await
61    }
62
63    /// Imports one already-validated event and its legacy delivery state in
64    /// the caller-owned transaction. This is migration infrastructure, not a
65    /// replacement for [`Self::enqueue`].
66    pub async fn import_for_migration<'c>(
67        &self,
68        transaction: &mut Transaction<'c, Postgres>,
69        event: NewEvent,
70        state: ImportedDeliveryState,
71    ) -> Result<ImportOutcome, ImportError> {
72        import_for_migration(transaction, event, state).await
73    }
74
75    /// Records the legacy publisher's authoritative delivery time for a
76    /// canonical pending migration import. This operation is migration
77    /// infrastructure, not an ordinary acknowledgement shortcut.
78    pub async fn finalize_pending_delivery_for_migration<'c>(
79        &self,
80        transaction: &mut Transaction<'c, Postgres>,
81        row_id: dovecote::RowId,
82        delivered_at: time::OffsetDateTime,
83    ) -> Result<FinalizeOutcome, FinalizeError> {
84        finalize_pending_delivery_for_migration(transaction, row_id, delivered_at).await
85    }
86
87    /// Verifies that the pool's current PostgreSQL schema satisfies Dovecote
88    /// migration version 1.
89    pub async fn check_schema(&self) -> Result<(), SchemaError> {
90        check_schema(&self.pool).await
91    }
92
93    /// Reads a bounded live page after `after_row_id`.
94    pub async fn page(
95        &self,
96        after_row_id: Option<dovecote::RowId>,
97        limit: dovecote::Limit,
98    ) -> Result<Vec<dovecote::PagedEvent>, PageError> {
99        page(&self.pool, after_row_id, limit).await
100    }
101
102    /// Begins a finite, repeatable-read snapshot pager.
103    pub async fn begin_snapshot(&self) -> Result<SnapshotPager, PageError> {
104        begin_snapshot(&self.pool).await
105    }
106
107    /// Claims eligible events under short PostgreSQL transactions.
108    pub async fn claim(
109        &self,
110        worker: dovecote::WorkerId,
111        lease_for: dovecote::Lease,
112        limit: dovecote::Limit,
113    ) -> Result<Vec<dovecote::ClaimedEvent>, ClaimError> {
114        claim(&self.pool, worker, lease_for, limit).await
115    }
116
117    /// Renews one current, unexpired claim using PostgreSQL time.
118    pub async fn renew(
119        &self,
120        row_id: dovecote::RowId,
121        claim_token: &dovecote::ClaimToken,
122        lease_for: dovecote::Lease,
123    ) -> Result<(), MutationError> {
124        renew(&self.pool, row_id, claim_token, lease_for).await
125    }
126
127    /// Acknowledges one current, unexpired claim.
128    pub async fn ack(
129        &self,
130        row_id: dovecote::RowId,
131        claim_token: &dovecote::ClaimToken,
132    ) -> Result<(), MutationError> {
133        ack(&self.pool, row_id, claim_token).await
134    }
135
136    /// Returns one current, unexpired claim to pending with a failure.
137    pub async fn retry(
138        &self,
139        row_id: dovecote::RowId,
140        claim_token: &dovecote::ClaimToken,
141        failure: &dovecote::Failure,
142        backoff: dovecote::Delay,
143    ) -> Result<(), MutationError> {
144        retry(&self.pool, row_id, claim_token, failure, backoff).await
145    }
146
147    /// Returns one current, unexpired claim to pending after a delay.
148    pub async fn release(
149        &self,
150        row_id: dovecote::RowId,
151        claim_token: &dovecote::ClaimToken,
152        delay: dovecote::Delay,
153    ) -> Result<(), MutationError> {
154        release(&self.pool, row_id, claim_token, delay).await
155    }
156
157    /// Quarantines one current, unexpired claim with an operator reason.
158    pub async fn quarantine(
159        &self,
160        row_id: dovecote::RowId,
161        claim_token: &dovecote::ClaimToken,
162        reason: &dovecote::QuarantineReason,
163    ) -> Result<(), MutationError> {
164        quarantine(&self.pool, row_id, claim_token, reason).await
165    }
166}
dovecote_sqlx_postgres/lib.rs

dovecote_sqlx_postgres/
lib.rs
