xmtp_mls/client/
notifications.rs

1//! Local notification settings and the native notification API.
2
3pub use crate::core::notifications::{
4    NotificationChannel, NotificationConfig, NotificationOverride,
5};
6pub(crate) use crate::core::notifications::{
7    decode, effective, enabled_from_record, encode, encode_config, notification_settings_changed,
8    recipient_secret, stored_config,
9};
10use crate::{
11    context::XmtpSharedContext,
12    worker::{WorkerKind, notifications as worker},
13};
14use serde::{Deserialize, Serialize};
15use xmtp_common::{ErrorCode, RetryableError};
16use xmtp_db::{
17    StorageError, TransactionOutcome::Continue, XmtpMlsStorageProvider,
18    notifications::StoredNotification, prelude::*,
19};
20
21/// The locally stored notification state. This getter does not make a request.
22#[derive(Debug)]
23pub enum NotificationState {
24    Disabled,
25    Enabled,
26    Failed(NotificationError),
27}
28
29/// Errors use fixed messages and never include notification credentials.
30#[derive(thiserror::Error, ErrorCode)]
31pub enum NotificationError {
32    /// The task runner is disabled. Not retryable.
33    #[error("notification task runner is disabled")]
34    TaskRunnerDisabled,
35    /// The recipient credential was rejected. Not retryable.
36    #[error("notification permission denied")]
37    PermissionDenied,
38    /// The delivery configuration is invalid. Not retryable.
39    #[error("notification configuration is invalid")]
40    InvalidArgument,
41    /// A notification value is outside the allowed range. Not retryable.
42    #[error("notification value is out of range")]
43    OutOfRange,
44    /// The backend does not support notifications. Not retryable.
45    #[error("notifications are not implemented")]
46    Unimplemented,
47    /// The delivery channel is not configured. Not retryable.
48    #[error("notification channel is not configured")]
49    ChannelNotConfigured,
50    /// The recipient topic limit was reached. Retry after the desired set changes.
51    #[error("notification topic limit reached")]
52    ResourceExhausted,
53    /// The notification request exceeded its time limit. Retryable.
54    #[error("notification request timed out")]
55    RequestTimeout,
56    /// The recipient must register again. Retryable.
57    #[error("notification recipient is not registered")]
58    NotFound,
59    /// A notification request failed. Retryable by the notification task.
60    #[error("notification request failed")]
61    #[error_code(inherit)]
62    Api(#[source] xmtp_api::ApiError),
63    #[error(transparent)]
64    #[error_code(inherit)]
65    Storage(#[from] StorageError),
66    #[error(transparent)]
67    #[error_code(inherit)]
68    Group(#[from] crate::groups::GroupError),
69}
70
71impl std::fmt::Debug for NotificationError {
72    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
73        f.write_str(self.error_code())
74    }
75}
76
77impl RetryableError for NotificationError {
78    fn is_retryable(&self) -> bool {
79        match self {
80            Self::TaskRunnerDisabled | Self::ResourceExhausted => false,
81            Self::Storage(e) => e.is_retryable(),
82            Self::Group(e) => e.is_retryable(),
83            _ => self.failure().is_none(),
84        }
85    }
86
87    fn rate_limit_delay(&self) -> Option<xmtp_common::time::Duration> {
88        match self {
89            Self::Api(e) => e.rate_limit_delay(),
90            Self::Group(e) => e.rate_limit_delay(),
91            _ => None,
92        }
93    }
94}
95
96impl From<xmtp_api::ApiError> for NotificationError {
97    fn from(error: xmtp_api::ApiError) -> Self {
98        use tonic::Code;
99        // A rate limit is not the recipient topic limit: the request succeeds
100        // after its delay, so the state stays enabled and adds keep going.
101        // implements: PUSH-267
102        if error.rate_limit_delay().is_some() {
103            return Self::Api(error);
104        }
105        let mut source: Option<&(dyn std::error::Error + 'static)> = Some(&error);
106        while let Some(current) = source {
107            if matches!(
108                current.downcast_ref::<xmtp_proto::api::ApiClientError>(),
109                Some(xmtp_proto::api::ApiClientError::Expired(_))
110            ) {
111                return Self::RequestTimeout;
112            }
113            source = current.source();
114        }
115        match xmtp_proto::api::grpc_status(&error).map(|status| status.code()) {
116            Some(Code::PermissionDenied) => Self::PermissionDenied,
117            Some(Code::InvalidArgument) => Self::InvalidArgument,
118            Some(Code::OutOfRange) => Self::OutOfRange,
119            Some(Code::Unimplemented) => Self::Unimplemented,
120            Some(Code::FailedPrecondition) => Self::ChannelNotConfigured,
121            Some(Code::ResourceExhausted) => Self::ResourceExhausted,
122            Some(Code::NotFound) => Self::NotFound,
123            _ => Self::Api(error),
124        }
125    }
126}
127
128#[derive(Serialize, Deserialize)]
129pub(crate) enum Failure {
130    PermissionDenied,
131    InvalidArgument,
132    OutOfRange,
133    Unimplemented,
134    ChannelNotConfigured,
135}
136
137impl NotificationError {
138    pub(crate) fn failure(&self) -> Option<Failure> {
139        match self {
140            Self::PermissionDenied => Some(Failure::PermissionDenied),
141            Self::InvalidArgument => Some(Failure::InvalidArgument),
142            Self::OutOfRange => Some(Failure::OutOfRange),
143            Self::Unimplemented => Some(Failure::Unimplemented),
144            Self::ChannelNotConfigured => Some(Failure::ChannelNotConfigured),
145            _ => None,
146        }
147    }
148}
149
150impl From<Failure> for NotificationError {
151    fn from(failure: Failure) -> Self {
152        match failure {
153            Failure::PermissionDenied => Self::PermissionDenied,
154            Failure::InvalidArgument => Self::InvalidArgument,
155            Failure::OutOfRange => Self::OutOfRange,
156            Failure::Unimplemented => Self::Unimplemented,
157            Failure::ChannelNotConfigured => Self::ChannelNotConfigured,
158        }
159    }
160}
161
162pub(crate) fn state(record: &StoredNotification) -> Result<NotificationState, StorageError> {
163    match record.push_state {
164        0 => Ok(NotificationState::Disabled),
165        1 => Ok(NotificationState::Enabled),
166        2 => Ok(NotificationState::Failed(
167            decode::<Failure>(
168                record
169                    .push_failed_error
170                    .as_deref()
171                    .ok_or(StorageError::DbDeserialize)?,
172            )?
173            .into(),
174        )),
175        _ => Err(StorageError::DbDeserialize),
176    }
177}
178
179/// Store configuration, then register inline. The task retries transient errors.
180// implements: PUSH-266
181#[xmtp_common::rpc_span]
182pub(crate) async fn enable_notifications<Context: XmtpSharedContext>(
183    context: &Context,
184    config: NotificationConfig,
185) -> Result<NotificationState, NotificationError> {
186    if !context
187        .worker_config()
188        .worker_enabled(WorkerKind::TaskRunner)
189    {
190        return Err(NotificationError::TaskRunnerDisabled);
191    }
192    let valid_channel = match &config.channel {
193        NotificationChannel::Apns { .. } | NotificationChannel::Fcm { .. } => true,
194        NotificationChannel::Http { signing_key, .. } => {
195            (16..=64).contains(&signing_key.as_slice().len())
196        }
197    };
198    if !valid_channel {
199        return Err(NotificationError::InvalidArgument);
200    }
201    let generation = crate::state_tx::state_write(context.mls_storage(), |tx| {
202        let storage = tx.storage();
203        let db = storage.db();
204        let mut record = db.notification_record()?;
205        if record.push_recipient_id.is_none() {
206            record.push_recipient_id = Some(xmtp_common::rand_vec::<32>());
207            record.push_recipient_secret =
208                Some(xmtp_cryptography::rand::rand_secret::<32>().into());
209        }
210        record.push_generation = record
211            .push_generation
212            .checked_add(1)
213            .ok_or(StorageError::DbSerialize)?;
214        record.push_config = Some(encode_config(&config)?);
215        record.push_state = 1;
216        record.push_failed_error = None;
217        // A new configuration does not end a pending rate-limit delay.
218        // Deadlines that cannot be decoded hold nothing, so they never
219        // stop an app from enabling.
220        let hold = worker::rate_limit_hold(&record).unwrap_or_default();
221        record.push_deadlines = Some(encode(&hold)?);
222        record.push_suppressed = None;
223        db.save_notification_record(&record)?;
224        Ok::<_, StorageError>(Continue(record.push_generation))
225    })?
226    .into_continued();
227    notification_settings_changed(context);
228    let _guard = context.task_channels().notification_request.lock().await;
229    let record = context.db().notification_record()?;
230    if record.push_generation == generation && record.push_state == 1 {
231        let result = worker::register(context, &record, &config).await;
232        if let Ok(true) = &result {
233            worker::resume_after_registration(context, generation)?;
234        }
235        drop(_guard);
236        notification_settings_changed(context);
237        result?;
238    }
239    Ok(notification_state(context)?)
240}
241
242/// Disable locally before unregistering. The recipient identity and overrides stay.
243/// A failed unregister leaves the client disabled with its recipient reset owed.
244#[xmtp_common::rpc_span]
245pub(crate) async fn disable_notifications<Context: XmtpSharedContext>(
246    context: &Context,
247) -> Result<(), NotificationError> {
248    let record = crate::state_tx::state_write(context.mls_storage(), |tx| {
249        let storage = tx.storage();
250        let db = storage.db();
251        // Deadlines that cannot be decoded hold nothing, so they never
252        // stop an app from disabling.
253        let hold = worker::rate_limit_hold(&db.notification_record()?).unwrap_or_default();
254        let mut record = db.disable_notifications()?;
255        // The task must not send inside a pending rate-limit delay after a
256        // later enable. App calls themselves are not held (RATE-041 binds
257        // the client's own retries).
258        if hold.rate_limited(xmtp_common::time::now_ns()) {
259            record.push_deadlines = Some(encode(&hold)?);
260            db.save_notification_record(&record)?;
261        }
262        Ok::<_, StorageError>(Continue(record))
263    })?
264    .into_continued();
265    notification_settings_changed(context);
266    if record.push_recipient_id.is_none() {
267        return Ok(());
268    }
269    let _guard = context.task_channels().notification_request.lock().await;
270    // A later disable still wants this recipient removed. The request
271    // lock orders any subsequent enable's Register after Unregister.
272    if context.db().notification_record()?.push_state != 0 {
273        notification_settings_changed(context);
274        return Ok(());
275    }
276    worker::unregister(context, &record, worker::request_deadline()).await
277}
278
279/// Read the locally stored state without a backend call.
280pub(crate) fn notification_state<Context: XmtpSharedContext>(
281    context: &Context,
282) -> Result<NotificationState, StorageError> {
283    state(&context.db().notification_record()?)
284}
285
286#[cfg(any(test, feature = "test-utils"))]
287#[doc(hidden)]
288pub mod test_support;
289#[cfg(test)]
290mod tests;
xmtp_mls/client/notifications.rs

xmtp_mls/client/
notifications.rs
