xmtp_api/
notification.rs

1use crate::{ApiClientWrapper, Result, dyn_err};
2use xmtp_common::time::{Instant, timeout};
3use xmtp_proto::{
4    api_client::XmtpBackendClient,
5    backend_v1::{
6        RecipientState, RegisterRequest, UnregisterRequest, UnregisterResponse,
7        UpdateSubscriptionsRequest,
8    },
9};
10
11/// Each request ends by `deadline`, which the caller shares across the
12/// requests of one notification turn. A rate-limit wait that would pass it
13/// returns the rejection at once, so the notification worker schedules the
14/// next attempt from its delay.
15impl<C: XmtpBackendClient> ApiClientWrapper<C> {
16    /// Create or renew a notification recipient.
17    #[xmtp_common::rpc_span]
18    pub async fn register(
19        &self,
20        request: RegisterRequest,
21        deadline: Instant,
22    ) -> Result<RecipientState> {
23        timeout(
24            deadline.saturating_duration_since(Instant::now()),
25            self.retry_call_until(
26                || self.api_client.register(request.clone()),
27                false,
28                Some(deadline),
29            ),
30        )
31        .await
32        .map_err(xmtp_proto::api::ApiClientError::from)
33        .map_err(dyn_err)?
34        .map_err(dyn_err)
35    }
36
37    /// Delete a notification recipient and its subscriptions.
38    #[xmtp_common::rpc_span]
39    pub async fn unregister(
40        &self,
41        request: UnregisterRequest,
42        deadline: Instant,
43    ) -> Result<UnregisterResponse> {
44        timeout(
45            deadline.saturating_duration_since(Instant::now()),
46            self.retry_call_until(
47                || self.api_client.unregister(request.clone()),
48                false,
49                Some(deadline),
50            ),
51        )
52        .await
53        .map_err(xmtp_proto::api::ApiClientError::from)
54        .map_err(dyn_err)?
55        .map_err(dyn_err)
56    }
57
58    /// Atomically update notification subscriptions and renew the recipient.
59    #[xmtp_common::rpc_span]
60    pub async fn update_subscriptions(
61        &self,
62        request: UpdateSubscriptionsRequest,
63        deadline: Instant,
64    ) -> Result<RecipientState> {
65        timeout(
66            deadline.saturating_duration_since(Instant::now()),
67            self.retry_call_until(
68                || self.api_client.update_subscriptions(request.clone()),
69                false,
70                Some(deadline),
71            ),
72        )
73        .await
74        .map_err(xmtp_proto::api::ApiClientError::from)
75        .map_err(dyn_err)?
76        .map_err(dyn_err)
77    }
78}
xmtp_api/notification.rs

xmtp_api/
notification.rs
