baukit_push/
pending_receipt.rs

1//! Deferred receipt polling for notifications the provider accepted but had not settled.
2
3use std::{collections::BTreeMap, num::NonZeroU32};
4
5use chrono::{DateTime, TimeDelta, Utc};
6use thiserror::Error;
7use uuid::Uuid;
8
9use crate::{
10    DeviceRegistry, DeviceToken, PushDeliveryStatus, PushError, PushOutcome, PushReceiptSource,
11    PushStoreError, PushStoreFuture, PushTicketId,
12};
13
14/// How long a recorded ticket waits before its first poll, and between polls.
15///
16/// Expo recommends checking receipts 15 minutes after the send.
17pub const RECEIPT_POLL_DELAY: TimeDelta = TimeDelta::minutes(15);
18
19/// How long the provider keeps a receipt after the send.
20///
21/// Expo clears receipts after 24 hours. A ticket older than this never
22/// settles; [`PendingReceiptStore::purge`] removes it.
23pub const RECEIPT_RETENTION: TimeDelta = TimeDelta::hours(24);
24
25/// An accepted notification whose receipt has not been read yet.
26#[derive(Clone, Debug, Eq, PartialEq)]
27pub struct PendingReceipt {
28    /// The ticket the provider issued for the notification.
29    pub ticket: PushTicketId,
30    /// The account the notification went to; erasing it removes the ticket.
31    pub owner_id: Uuid,
32    /// The device token the notification went to.
33    pub token: DeviceToken,
34    /// The instant the send started; invalidation compares it.
35    pub sent_at: DateTime<Utc>,
36}
37
38/// Storage port for tickets whose receipts are still outstanding.
39///
40/// Each ticket carries a due instant. [`PendingReceiptStore::take_due`] moves
41/// the tickets it returns to a later due instant, so concurrent pollers take
42/// disjoint batches and a ticket that stays unsettled goes to the back of the
43/// queue instead of blocking newer ones.
44pub trait PendingReceiptStore: Send + Sync {
45    /// Records tickets first due at `due_at`; returns how many were new.
46    ///
47    /// A ticket that is already recorded is left unchanged.
48    fn record(
49        &self,
50        receipts: Vec<PendingReceipt>,
51        due_at: DateTime<Utc>,
52    ) -> PushStoreFuture<'_, Result<u64, PushStoreError>>;
53
54    /// Takes up to `limit` tickets due at or before `now`, oldest due first,
55    /// and makes them due again at `retry_at`.
56    fn take_due(
57        &self,
58        now: DateTime<Utc>,
59        retry_at: DateTime<Utc>,
60        limit: NonZeroU32,
61    ) -> PushStoreFuture<'_, Result<Vec<PendingReceipt>, PushStoreError>>;
62
63    /// Deletes tickets whose receipts settled; returns how many went.
64    fn delete(
65        &self,
66        tickets: Vec<PushTicketId>,
67    ) -> PushStoreFuture<'_, Result<u64, PushStoreError>>;
68
69    /// Deletes up to `limit` tickets sent before `sent_before`; returns how many went.
70    ///
71    /// Pass `now - RECEIPT_RETENTION`. Call it again until it returns less
72    /// than `limit`.
73    fn purge(
74        &self,
75        sent_before: DateTime<Utc>,
76        limit: NonZeroU32,
77    ) -> PushStoreFuture<'_, Result<u64, PushStoreError>>;
78
79    /// Deletes every ticket of one owner; returns how many went.
80    fn erase_owner(&self, owner_id: Uuid) -> PushStoreFuture<'_, Result<u64, PushStoreError>>;
81
82    /// Records every [`PushDeliveryStatus::Accepted`] outcome of one send to `owner_id`.
83    ///
84    /// Call it after [`PushSender::send`](crate::PushSender::send) with the
85    /// instant the send started. The tickets are first due
86    /// [`RECEIPT_POLL_DELAY`] later. Outcomes without an accepted ticket cost
87    /// no store round trip.
88    fn record_accepted(
89        &self,
90        owner_id: Uuid,
91        outcomes: &[PushOutcome],
92        sent_at: DateTime<Utc>,
93    ) -> PushStoreFuture<'_, Result<u64, PushStoreError>> {
94        let receipts = accepted_receipts(owner_id, outcomes, sent_at);
95        if receipts.is_empty() {
96            return Box::pin(std::future::ready(Ok(0)));
97        }
98        self.record(receipts, sent_at + RECEIPT_POLL_DELAY)
99    }
100}
101
102/// Returns a pending receipt for each accepted outcome of one send to `owner_id`.
103#[must_use]
104pub fn accepted_receipts(
105    owner_id: Uuid,
106    outcomes: &[PushOutcome],
107    sent_at: DateTime<Utc>,
108) -> Vec<PendingReceipt> {
109    outcomes
110        .iter()
111        .filter_map(|outcome| match &outcome.status {
112            PushDeliveryStatus::Accepted(ticket) => Some(PendingReceipt {
113                ticket: ticket.clone(),
114                owner_id,
115                token: outcome.token.clone(),
116                sent_at,
117            }),
118            PushDeliveryStatus::Delivered | PushDeliveryStatus::Rejected(_) => None,
119        })
120        .collect()
121}
122
123/// What one [`poll_pending_receipts`] run did.
124#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
125pub struct ReceiptPoll {
126    /// Tickets taken from the store. Fewer than the limit means the queue is drained.
127    pub taken: u64,
128    /// Tickets whose receipt settled and that were deleted.
129    pub settled: u64,
130    /// Device tokens removed from the registry.
131    pub invalidated: u64,
132}
133
134/// A failure that stopped a receipt poll.
135///
136/// Taken tickets stay recorded and come due again after [`RECEIPT_POLL_DELAY`].
137#[derive(Clone, Debug, Error, Eq, PartialEq)]
138pub enum ReceiptPollError {
139    /// The pending receipt store or the registry failed.
140    #[error(transparent)]
141    Store(#[from] PushStoreError),
142    /// The provider's receipt request failed.
143    #[error(transparent)]
144    Push(#[from] PushError),
145}
146
147/// Polls one batch of due receipts and applies what settled.
148///
149/// Takes up to `limit` due tickets, fetches their receipts, invalidates every
150/// `DeviceNotRegistered` token with the `sent_at` of its own send, and deletes
151/// the settled tickets. Tickets without a receipt yet stay recorded for a
152/// later run. Invalidation runs before deletion, so a failure in between
153/// repeats the invalidation on the next run, which removes nothing new.
154///
155/// # Errors
156///
157/// Returns [`ReceiptPollError`] when the store, the registry, or the receipt
158/// request fails.
159pub async fn poll_pending_receipts<S, P, R>(
160    source: &S,
161    pending: &P,
162    registry: &R,
163    now: DateTime<Utc>,
164    limit: NonZeroU32,
165) -> Result<ReceiptPoll, ReceiptPollError>
166where
167    S: PushReceiptSource + ?Sized,
168    P: PendingReceiptStore + ?Sized,
169    R: DeviceRegistry + ?Sized,
170{
171    let due = pending
172        .take_due(now, now + RECEIPT_POLL_DELAY, limit)
173        .await?;
174    let taken = count(due.len());
175    if due.is_empty() {
176        return Ok(ReceiptPoll::default());
177    }
178    let receipts = source
179        .receipts(due.iter().map(|receipt| receipt.ticket.clone()).collect())
180        .await?;
181
182    let mut dead_by_send = BTreeMap::<DateTime<Utc>, Vec<DeviceToken>>::new();
183    let mut settled = Vec::new();
184    for pending_receipt in due {
185        let Some(receipt) = receipts.get(&pending_receipt.ticket) else {
186            continue;
187        };
188        if receipt.is_token_dead() {
189            dead_by_send
190                .entry(pending_receipt.sent_at)
191                .or_default()
192                .push(pending_receipt.token);
193        }
194        settled.push(pending_receipt.ticket);
195    }
196
197    let mut invalidated = 0;
198    for (sent_at, mut tokens) in dead_by_send {
199        tokens.sort();
200        tokens.dedup();
201        invalidated += registry.invalidate(tokens, sent_at).await?;
202    }
203    let settled = if settled.is_empty() {
204        0
205    } else {
206        pending.delete(settled).await?
207    };
208    Ok(ReceiptPoll {
209        taken,
210        settled,
211        invalidated,
212    })
213}
214
215fn count(len: usize) -> u64 {
216    u64::try_from(len).unwrap_or(u64::MAX)
217}
218
219#[cfg(all(test, feature = "test-support"))]
220mod tests {
221    use super::*;
222    use crate::{
223        DevicePlatform, DeviceRegistration, FakePushSender, MemoryDeviceRegistry,
224        MemoryPendingReceiptStore, PushMessage, PushReceipt, PushRejection, PushSender,
225    };
226
227    type TestResult = Result<(), Box<dyn std::error::Error>>;
228
229    const LIMIT: NonZeroU32 = NonZeroU32::new(10).expect("ten is not zero");
230
231    fn at(minutes: i64) -> DateTime<Utc> {
232        DateTime::UNIX_EPOCH + TimeDelta::minutes(minutes)
233    }
234
235    fn token(value: &str) -> DeviceToken {
236        DeviceToken::new(value).expect("valid test token")
237    }
238
239    async fn register(registry: &MemoryDeviceRegistry, owner_id: Uuid, value: &str, minutes: i64) {
240        registry
241            .register(DeviceRegistration::new(
242                owner_id,
243                token(value),
244                DevicePlatform::Ios,
245                at(minutes),
246            ))
247            .await
248            .expect("memory register");
249    }
250
251    async fn registered(registry: &MemoryDeviceRegistry, owner_id: Uuid) -> Vec<String> {
252        let mut tokens = registry
253            .list_for_owner(owner_id)
254            .await
255            .expect("memory list")
256            .into_iter()
257            .map(|device| device.token.expose().to_owned())
258            .collect::<Vec<_>>();
259        tokens.sort();
260        tokens
261    }
262
263    #[tokio::test]
264    async fn a_late_device_not_registered_receipt_removes_the_token() -> TestResult {
265        let sender = FakePushSender::new();
266        let pending = MemoryPendingReceiptStore::new();
267        let registry = MemoryDeviceRegistry::new();
268        let owner = Uuid::now_v7();
269        let names = ["fine", "gone", "reinstalled", "slow"];
270        for name in names {
271            register(&registry, owner, name, 0).await;
272            sender.accept_without_receipt(token(name)).await;
273        }
274
275        let sent_at = at(1);
276        let outcomes = sender
277            .send(
278                names
279                    .iter()
280                    .map(|name| PushMessage::new(token(name), "t", "b"))
281                    .collect(),
282            )
283            .await?;
284        assert_eq!(pending.record_accepted(owner, &outcomes, sent_at).await?, 4);
285        assert_eq!(
286            poll_pending_receipts(&sender, &pending, &registry, at(5), LIMIT).await?,
287            ReceiptPoll::default(),
288            "nothing is due before the poll delay"
289        );
290
291        let dead = PushReceipt::Rejected(PushRejection::DeviceNotRegistered);
292        sender
293            .settle_receipt(token("fine"), PushReceipt::Delivered)
294            .await;
295        sender.settle_receipt(token("gone"), dead.clone()).await;
296        sender.settle_receipt(token("reinstalled"), dead).await;
297        register(&registry, owner, "reinstalled", 10).await;
298
299        let due_at = sent_at + RECEIPT_POLL_DELAY;
300        let poll = poll_pending_receipts(&sender, &pending, &registry, due_at, LIMIT).await?;
301        assert_eq!(
302            poll,
303            ReceiptPoll {
304                taken: 4,
305                settled: 3,
306                invalidated: 1
307            }
308        );
309        assert_eq!(
310            registered(&registry, owner).await,
311            ["fine", "reinstalled", "slow"],
312            "a registration after the send outlives the receipt"
313        );
314        let still_pending = pending.pending();
315        assert_eq!(still_pending.len(), 1);
316        assert_eq!(still_pending[0].token, token("slow"));
317
318        assert_eq!(
319            poll_pending_receipts(&sender, &pending, &registry, due_at, LIMIT)
320                .await?
321                .taken,
322            0,
323            "a still-pending ticket waits for the next delay"
324        );
325        Ok(())
326    }
327
328    #[tokio::test]
329    async fn a_failed_receipt_request_keeps_the_tickets() -> TestResult {
330        let sender = FakePushSender::new();
331        let pending = MemoryPendingReceiptStore::new();
332        let registry = MemoryDeviceRegistry::new();
333        sender.accept_without_receipt(token("slow")).await;
334        let outcomes = sender
335            .send(vec![PushMessage::new(token("slow"), "t", "b")])
336            .await?;
337        pending
338            .record_accepted(Uuid::now_v7(), &outcomes, at(0))
339            .await?;
340
341        sender
342            .fail_with(PushError::Transport {
343                class: baukit_http::RetryClass::Unavailable,
344            })
345            .await;
346        let error = poll_pending_receipts(&sender, &pending, &registry, at(20), LIMIT)
347            .await
348            .expect_err("the receipt request fails");
349        assert!(matches!(error, ReceiptPollError::Push(_)));
350        assert_eq!(pending.pending().len(), 1);
351        Ok(())
352    }
353
354    #[tokio::test]
355    async fn settled_outcomes_record_nothing() -> TestResult {
356        let pending = MemoryPendingReceiptStore::new();
357        let outcomes = [PushOutcome {
358            token: token("fine"),
359            status: PushDeliveryStatus::Delivered,
360        }];
361        let owner = Uuid::now_v7();
362        assert_eq!(pending.record_accepted(owner, &outcomes, at(0)).await?, 0);
363        assert!(accepted_receipts(owner, &outcomes, at(0)).is_empty());
364        Ok(())
365    }
366}
baukit_push/pending_receipt.rs

baukit_push/
pending_receipt.rs
