dovecote_sqlx_postgres/
scope.rs

1//! Explicit tenant-scoped and administrative PostgreSQL handles.
2
3use dovecote::{
4    ClaimedEvent, EnqueueOutcome, FinalizeOutcome, ImportOutcome, ImportedDeliveryState, NewEvent,
5    TenantId,
6};
7use sqlx::{Postgres, Transaction};
8use time::OffsetDateTime;
9
10use crate::{
11    ClaimError, EnqueueError, FinalizeError, ImportError, MutationError, PageError, SnapshotPager,
12    enqueue, finalize, import, lifecycle, page, rls,
13};
14
15/// PostgreSQL Dovecote operations restricted to one validated tenant.
16#[derive(Clone)]
17pub struct TenantDovecote {
18    pool: sqlx::PgPool,
19    tenant_id: TenantId,
20}
21
22impl TenantDovecote {
23    pub(crate) fn new(pool: sqlx::PgPool, tenant_id: TenantId) -> Self {
24        Self { pool, tenant_id }
25    }
26
27    /// Returns this handle's validated tenant identifier.
28    pub fn tenant_id(&self) -> &TenantId {
29        &self.tenant_id
30    }
31
32    /// Borrows the pool used by this handle.
33    pub fn pool(&self) -> &sqlx::PgPool {
34        &self.pool
35    }
36
37    /// Binds this tenant to a transaction for the optional PostgreSQL RLS
38    /// profile. Adapter predicates remain active even without RLS.
39    pub async fn bind_tenant<'c>(
40        &self,
41        transaction: &mut Transaction<'c, Postgres>,
42    ) -> Result<(), sqlx::Error> {
43        rls::bind_tenant(transaction, &self.tenant_id).await
44    }
45
46    /// Enqueues an event for this tenant in the caller-owned transaction.
47    pub async fn enqueue<'c>(
48        &self,
49        transaction: &mut Transaction<'c, Postgres>,
50        event: NewEvent,
51    ) -> Result<EnqueueOutcome, EnqueueError> {
52        self.bind_tenant(transaction)
53            .await
54            .map_err(|source| EnqueueError::sql("bind tenant", source))?;
55        enqueue::enqueue_for_scope(transaction, &self.tenant_id, event).await
56    }
57
58    /// Imports one event and legacy state for this tenant.
59    pub async fn import_for_migration<'c>(
60        &self,
61        transaction: &mut Transaction<'c, Postgres>,
62        event: NewEvent,
63        state: ImportedDeliveryState,
64    ) -> Result<ImportOutcome, ImportError> {
65        self.bind_tenant(transaction)
66            .await
67            .map_err(|source| ImportError::sql("bind tenant", source))?;
68        import::import_for_scope(transaction, &self.tenant_id, event, state).await
69    }
70
71    /// Finalizes one canonical pending migration row for this tenant.
72    pub async fn finalize_pending_delivery_for_migration<'c>(
73        &self,
74        transaction: &mut Transaction<'c, Postgres>,
75        row_id: dovecote::RowId,
76        delivered_at: OffsetDateTime,
77    ) -> Result<FinalizeOutcome, FinalizeError> {
78        self.bind_tenant(transaction)
79            .await
80            .map_err(|source| FinalizeError::sql("bind tenant", source))?;
81        finalize::finalize_for_scope(transaction, &self.tenant_id, row_id, delivered_at).await
82    }
83
84    /// Reads a live page restricted to this tenant.
85    pub async fn page(
86        &self,
87        after_row_id: Option<dovecote::RowId>,
88        limit: dovecote::Limit,
89    ) -> Result<Vec<dovecote::PagedEvent>, PageError> {
90        page::page_for_scope(&self.pool, Some(&self.tenant_id), after_row_id, limit).await
91    }
92
93    /// Begins a finite snapshot pager restricted to this tenant.
94    pub async fn begin_snapshot(&self) -> Result<SnapshotPager, PageError> {
95        page::begin_snapshot_for_scope(&self.pool, Some(&self.tenant_id)).await
96    }
97
98    /// Claims pending and expired deliveries for this tenant.
99    pub async fn claim(
100        &self,
101        worker: dovecote::WorkerId,
102        lease_for: dovecote::Lease,
103        limit: dovecote::Limit,
104    ) -> Result<Vec<ClaimedEvent>, ClaimError> {
105        lifecycle::claim_for_scope(&self.pool, Some(&self.tenant_id), worker, lease_for, limit)
106            .await
107    }
108
109    /// Renews one current claim for this tenant.
110    pub async fn renew(
111        &self,
112        row_id: dovecote::RowId,
113        claim_token: &dovecote::ClaimToken,
114        lease_for: dovecote::Lease,
115    ) -> Result<(), MutationError> {
116        lifecycle::renew_for_scope(
117            &self.pool,
118            Some(&self.tenant_id),
119            row_id,
120            claim_token,
121            lease_for,
122        )
123        .await
124    }
125
126    /// Acknowledges one current claim for this tenant.
127    pub async fn ack(
128        &self,
129        row_id: dovecote::RowId,
130        claim_token: &dovecote::ClaimToken,
131    ) -> Result<(), MutationError> {
132        lifecycle::ack_for_scope(&self.pool, Some(&self.tenant_id), row_id, claim_token).await
133    }
134
135    /// Returns one current claim to pending for this tenant.
136    pub async fn retry(
137        &self,
138        row_id: dovecote::RowId,
139        claim_token: &dovecote::ClaimToken,
140        failure: &dovecote::Failure,
141        backoff: dovecote::Delay,
142    ) -> Result<(), MutationError> {
143        lifecycle::retry_for_scope(
144            &self.pool,
145            Some(&self.tenant_id),
146            row_id,
147            claim_token,
148            failure,
149            backoff,
150        )
151        .await
152    }
153
154    /// Releases one current claim for this tenant.
155    pub async fn release(
156        &self,
157        row_id: dovecote::RowId,
158        claim_token: &dovecote::ClaimToken,
159        delay: dovecote::Delay,
160    ) -> Result<(), MutationError> {
161        lifecycle::release_for_scope(
162            &self.pool,
163            Some(&self.tenant_id),
164            row_id,
165            claim_token,
166            delay,
167        )
168        .await
169    }
170
171    /// Quarantines one current claim for this tenant.
172    pub async fn quarantine(
173        &self,
174        row_id: dovecote::RowId,
175        claim_token: &dovecote::ClaimToken,
176        reason: &dovecote::QuarantineReason,
177    ) -> Result<(), MutationError> {
178        lifecycle::quarantine_for_scope(
179            &self.pool,
180            Some(&self.tenant_id),
181            row_id,
182            claim_token,
183            reason,
184        )
185        .await
186    }
187}
188
189/// Explicit all-tenant PostgreSQL Dovecote operations.
190#[derive(Clone)]
191pub struct AdminDovecote {
192    pool: sqlx::PgPool,
193}
194
195impl AdminDovecote {
196    pub(crate) fn new(pool: sqlx::PgPool) -> Self {
197        Self { pool }
198    }
199
200    /// Borrows the pool used by this handle.
201    pub fn pool(&self) -> &sqlx::PgPool {
202        &self.pool
203    }
204
205    /// Enqueues an event for an explicitly named tenant.
206    pub async fn enqueue<'c>(
207        &self,
208        transaction: &mut Transaction<'c, Postgres>,
209        tenant_id: TenantId,
210        event: NewEvent,
211    ) -> Result<EnqueueOutcome, EnqueueError> {
212        enqueue::enqueue_for_scope(transaction, &tenant_id, event).await
213    }
214
215    /// Imports one event and legacy state for an explicitly named tenant.
216    pub async fn import_for_migration<'c>(
217        &self,
218        transaction: &mut Transaction<'c, Postgres>,
219        tenant_id: TenantId,
220        event: NewEvent,
221        state: ImportedDeliveryState,
222    ) -> Result<ImportOutcome, ImportError> {
223        import::import_for_scope(transaction, &tenant_id, event, state).await
224    }
225
226    /// Finalizes one migration row for an explicitly named tenant.
227    pub async fn finalize_pending_delivery_for_migration<'c>(
228        &self,
229        transaction: &mut Transaction<'c, Postgres>,
230        tenant_id: TenantId,
231        row_id: dovecote::RowId,
232        delivered_at: OffsetDateTime,
233    ) -> Result<FinalizeOutcome, FinalizeError> {
234        finalize::finalize_for_scope(transaction, &tenant_id, row_id, delivered_at).await
235    }
236
237    /// Reads a live page across all tenants.
238    pub async fn page(
239        &self,
240        after_row_id: Option<dovecote::RowId>,
241        limit: dovecote::Limit,
242    ) -> Result<Vec<dovecote::PagedEvent>, PageError> {
243        page::page_for_scope(&self.pool, None, after_row_id, limit).await
244    }
245
246    /// Begins a finite snapshot pager across all tenants.
247    pub async fn begin_snapshot(&self) -> Result<SnapshotPager, PageError> {
248        page::begin_snapshot_for_scope(&self.pool, None).await
249    }
250
251    /// Claims pending and expired deliveries across all tenants.
252    pub async fn claim(
253        &self,
254        worker: dovecote::WorkerId,
255        lease_for: dovecote::Lease,
256        limit: dovecote::Limit,
257    ) -> Result<Vec<ClaimedEvent>, ClaimError> {
258        lifecycle::claim_for_scope(&self.pool, None, worker, lease_for, limit).await
259    }
260
261    /// Renews one claim for an explicitly named tenant.
262    pub async fn renew(
263        &self,
264        tenant_id: TenantId,
265        row_id: dovecote::RowId,
266        claim_token: &dovecote::ClaimToken,
267        lease_for: dovecote::Lease,
268    ) -> Result<(), MutationError> {
269        lifecycle::renew_for_scope(&self.pool, Some(&tenant_id), row_id, claim_token, lease_for)
270            .await
271    }
272
273    /// Acknowledges one claim for an explicitly named tenant.
274    pub async fn ack(
275        &self,
276        tenant_id: TenantId,
277        row_id: dovecote::RowId,
278        claim_token: &dovecote::ClaimToken,
279    ) -> Result<(), MutationError> {
280        lifecycle::ack_for_scope(&self.pool, Some(&tenant_id), row_id, claim_token).await
281    }
282
283    /// Retries one claim for an explicitly named tenant.
284    pub async fn retry(
285        &self,
286        tenant_id: TenantId,
287        row_id: dovecote::RowId,
288        claim_token: &dovecote::ClaimToken,
289        failure: &dovecote::Failure,
290        backoff: dovecote::Delay,
291    ) -> Result<(), MutationError> {
292        lifecycle::retry_for_scope(
293            &self.pool,
294            Some(&tenant_id),
295            row_id,
296            claim_token,
297            failure,
298            backoff,
299        )
300        .await
301    }
302
303    /// Releases one claim for an explicitly named tenant.
304    pub async fn release(
305        &self,
306        tenant_id: TenantId,
307        row_id: dovecote::RowId,
308        claim_token: &dovecote::ClaimToken,
309        delay: dovecote::Delay,
310    ) -> Result<(), MutationError> {
311        lifecycle::release_for_scope(&self.pool, Some(&tenant_id), row_id, claim_token, delay).await
312    }
313
314    /// Quarantines one claim for an explicitly named tenant.
315    pub async fn quarantine(
316        &self,
317        tenant_id: TenantId,
318        row_id: dovecote::RowId,
319        claim_token: &dovecote::ClaimToken,
320        reason: &dovecote::QuarantineReason,
321    ) -> Result<(), MutationError> {
322        lifecycle::quarantine_for_scope(&self.pool, Some(&tenant_id), row_id, claim_token, reason)
323            .await
324    }
325}
dovecote_sqlx_postgres/scope.rs

dovecote_sqlx_postgres/
scope.rs
