xmtp_mls/client/
notifications.rs

1//! Local notification settings and the native notification API.
2
3use crate::{
4    client::Client,
5    context::XmtpSharedContext,
6    groups::MlsGroup,
7    worker::{WorkerKind, notifications as worker},
8};
9use serde::{Deserialize, Serialize};
10use xmtp_common::{ErrorCode, RetryableError};
11use xmtp_db::consent_record::{ConsentState, ConsentType};
12use xmtp_db::{
13    StorageError, TransactionOutcome::Continue, XmtpMlsStorageProvider,
14    notifications::StoredNotification, prelude::*,
15};
16use xmtp_proto::backend_v1::{self, register_request::Delivery};
17
18/// The delivery endpoint. Credentials are omitted from debug output.
19#[derive(Clone, Serialize, Deserialize)]
20pub enum NotificationChannel {
21    Apns { token: String },
22    Fcm { token: String },
23    Http { url: String, signing_key: Vec<u8> },
24}
25
26impl std::fmt::Debug for NotificationChannel {
27    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28        f.write_str(match self {
29            Self::Apns { .. } => "Apns",
30            Self::Fcm { .. } => "Fcm",
31            Self::Http { .. } => "Http",
32        })
33    }
34}
35
36/// Delivery details and rules used to compute the desired subscriptions.
37#[derive(Clone, Serialize, Deserialize)]
38pub struct NotificationConfig {
39    pub channel: NotificationChannel,
40    pub consent_states: Vec<ConsentState>,
41    pub include_welcomes: bool,
42    pub include_sync_groups: bool,
43    pub include_commits: bool,
44}
45
46impl NotificationConfig {
47    /// Use the standard rules with the supplied delivery endpoint.
48    pub fn new(channel: NotificationChannel) -> Self {
49        Self {
50            channel,
51            consent_states: vec![ConsentState::Allowed],
52            include_welcomes: true,
53            include_sync_groups: false,
54            include_commits: false,
55        }
56    }
57
58    pub(crate) fn channel_id(&self) -> i32 {
59        match self.channel {
60            NotificationChannel::Apns { .. } => backend_v1::Channel::Apns as i32,
61            NotificationChannel::Fcm { .. } => backend_v1::Channel::Fcm as i32,
62            NotificationChannel::Http { .. } => backend_v1::Channel::Http as i32,
63        }
64    }
65
66    pub(crate) fn registration(&self, record: &StoredNotification) -> backend_v1::RegisterRequest {
67        let delivery = match &self.channel {
68            NotificationChannel::Apns { token } => Delivery::Apns(backend_v1::ApnsDelivery {
69                token: token.clone(),
70            }),
71            NotificationChannel::Fcm { token } => Delivery::Fcm(backend_v1::FcmDelivery {
72                token: token.clone(),
73            }),
74            NotificationChannel::Http { url, signing_key } => {
75                Delivery::Http(backend_v1::HttpDelivery {
76                    url: url.clone(),
77                    signing_key: signing_key.clone(),
78                })
79            }
80        };
81        backend_v1::RegisterRequest {
82            recipient_id: record.push_recipient_id.clone().unwrap_or_default(),
83            recipient_secret: record.push_recipient_secret.clone().unwrap_or_default(),
84            delivery: Some(delivery),
85        }
86    }
87}
88
89/// The locally stored notification state. This getter does not make a request.
90#[derive(Debug)]
91pub enum NotificationState {
92    Disabled,
93    Enabled,
94    Failed(NotificationError),
95}
96
97/// A conversation rule. Sync groups use the client rule only.
98#[derive(Clone, Copy, Debug, PartialEq, Eq)]
99pub enum NotificationOverride {
100    Enabled,
101    Disabled,
102    Default,
103}
104
105/// Errors use fixed messages and never include notification credentials.
106#[derive(thiserror::Error, ErrorCode)]
107pub enum NotificationError {
108    /// The task runner is disabled. Not retryable.
109    #[error("notification task runner is disabled")]
110    TaskRunnerDisabled,
111    /// The recipient credential was rejected. Not retryable.
112    #[error("notification permission denied")]
113    PermissionDenied,
114    /// The delivery configuration is invalid. Not retryable.
115    #[error("notification configuration is invalid")]
116    InvalidArgument,
117    /// A notification value is outside the allowed range. Not retryable.
118    #[error("notification value is out of range")]
119    OutOfRange,
120    /// The backend does not support notifications. Not retryable.
121    #[error("notifications are not implemented")]
122    Unimplemented,
123    /// The delivery channel is not configured. Not retryable.
124    #[error("notification channel is not configured")]
125    ChannelNotConfigured,
126    /// The recipient topic limit was reached. Retry after the desired set changes.
127    #[error("notification topic limit reached")]
128    ResourceExhausted,
129    /// The notification request exceeded its time limit. Retryable.
130    #[error("notification request timed out")]
131    RequestTimeout,
132    /// The recipient must register again. Retryable.
133    #[error("notification recipient is not registered")]
134    NotFound,
135    /// A notification request failed. Retryable by the notification task.
136    #[error("notification request failed")]
137    #[error_code(inherit)]
138    Api(#[source] xmtp_api::ApiError),
139    #[error(transparent)]
140    #[error_code(inherit)]
141    Storage(#[from] StorageError),
142    #[error(transparent)]
143    #[error_code(inherit)]
144    Group(#[from] crate::groups::GroupError),
145}
146
147impl std::fmt::Debug for NotificationError {
148    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
149        f.write_str(self.error_code())
150    }
151}
152
153impl RetryableError for NotificationError {
154    fn is_retryable(&self) -> bool {
155        match self {
156            Self::TaskRunnerDisabled | Self::ResourceExhausted => false,
157            Self::Storage(e) => e.is_retryable(),
158            Self::Group(e) => e.is_retryable(),
159            _ => self.failure().is_none(),
160        }
161    }
162}
163
164impl From<xmtp_api::ApiError> for NotificationError {
165    fn from(error: xmtp_api::ApiError) -> Self {
166        use tonic::Code;
167        let mut source: Option<&(dyn std::error::Error + 'static)> = Some(&error);
168        while let Some(current) = source {
169            if matches!(
170                current.downcast_ref::<xmtp_proto::api::ApiClientError>(),
171                Some(xmtp_proto::api::ApiClientError::Expired(_))
172            ) {
173                return Self::RequestTimeout;
174            }
175            source = current.source();
176        }
177        match xmtp_proto::api::grpc_status(&error).map(|status| status.code()) {
178            Some(Code::PermissionDenied) => Self::PermissionDenied,
179            Some(Code::InvalidArgument) => Self::InvalidArgument,
180            Some(Code::OutOfRange) => Self::OutOfRange,
181            Some(Code::Unimplemented) => Self::Unimplemented,
182            Some(Code::FailedPrecondition) => Self::ChannelNotConfigured,
183            Some(Code::ResourceExhausted) => Self::ResourceExhausted,
184            Some(Code::NotFound) => Self::NotFound,
185            _ => Self::Api(error),
186        }
187    }
188}
189
190#[derive(Serialize, Deserialize)]
191pub(crate) enum Failure {
192    PermissionDenied,
193    InvalidArgument,
194    OutOfRange,
195    Unimplemented,
196    ChannelNotConfigured,
197}
198
199impl NotificationError {
200    pub(crate) fn failure(&self) -> Option<Failure> {
201        match self {
202            Self::PermissionDenied => Some(Failure::PermissionDenied),
203            Self::InvalidArgument => Some(Failure::InvalidArgument),
204            Self::OutOfRange => Some(Failure::OutOfRange),
205            Self::Unimplemented => Some(Failure::Unimplemented),
206            Self::ChannelNotConfigured => Some(Failure::ChannelNotConfigured),
207            _ => None,
208        }
209    }
210}
211
212impl From<Failure> for NotificationError {
213    fn from(failure: Failure) -> Self {
214        match failure {
215            Failure::PermissionDenied => Self::PermissionDenied,
216            Failure::InvalidArgument => Self::InvalidArgument,
217            Failure::OutOfRange => Self::OutOfRange,
218            Failure::Unimplemented => Self::Unimplemented,
219            Failure::ChannelNotConfigured => Self::ChannelNotConfigured,
220        }
221    }
222}
223
224pub(crate) fn encode<T: Serialize>(value: &T) -> Result<Vec<u8>, StorageError> {
225    serde_json::to_vec(value).map_err(|_| StorageError::DbSerialize)
226}
227
228pub(crate) fn decode<T: serde::de::DeserializeOwned>(value: &[u8]) -> Result<T, StorageError> {
229    serde_json::from_slice(value).map_err(|_| StorageError::DbDeserialize)
230}
231
232pub(crate) fn state(record: &StoredNotification) -> Result<NotificationState, StorageError> {
233    match record.push_state {
234        0 => Ok(NotificationState::Disabled),
235        1 => Ok(NotificationState::Enabled),
236        2 => Ok(NotificationState::Failed(
237            decode::<Failure>(
238                record
239                    .push_failed_error
240                    .as_deref()
241                    .ok_or(StorageError::DbDeserialize)?,
242            )?
243            .into(),
244        )),
245        _ => Err(StorageError::DbDeserialize),
246    }
247}
248
249impl<Context: XmtpSharedContext> Client<Context> {
250    /// Store configuration, then register inline. The task retries transient errors.
251    #[xmtp_common::rpc_span]
252    pub async fn enable_notifications(
253        &self,
254        config: NotificationConfig,
255    ) -> Result<NotificationState, NotificationError> {
256        if !self
257            .context
258            .worker_config()
259            .worker_enabled(WorkerKind::TaskRunner)
260        {
261            return Err(NotificationError::TaskRunnerDisabled);
262        }
263        if let NotificationChannel::Http { signing_key, .. } = &config.channel
264            && !(16..=64).contains(&signing_key.len())
265        {
266            return Err(NotificationError::InvalidArgument);
267        }
268        let generation = {
269            let mut pending = self
270                .context
271                .task_channels()
272                .notification_pending_topics
273                .lock();
274            let generation = crate::state_tx::state_write(self.context.mls_storage(), |tx| {
275                let storage = tx.storage();
276                let db = storage.db();
277                let mut record = db.notification_record()?;
278                if record.push_recipient_id.is_none() {
279                    record.push_recipient_id = Some(xmtp_common::rand_vec::<32>());
280                    record.push_recipient_secret = Some(xmtp_common::rand_vec::<32>());
281                }
282                record.push_generation = record
283                    .push_generation
284                    .checked_add(1)
285                    .ok_or(StorageError::DbSerialize)?;
286                record.push_config = Some(encode(&config)?);
287                record.push_state = 1;
288                record.push_failed_error = None;
289                record.push_deadlines = Some(encode(&worker::Deadlines::default())?);
290                record.push_suppressed = None;
291                // A disabled client keeps confirmed backend topics in memory.
292                // Restore them as stale so this configuration can reconcile them.
293                let present: std::collections::BTreeSet<_> = db
294                    .uploaded_topics()?
295                    .into_iter()
296                    .map(|row| row.topic)
297                    .collect();
298                let restore: Vec<_> = pending
299                    .values()
300                    .filter(|row| !present.contains(&row.topic))
301                    .cloned()
302                    .map(|mut row| {
303                        row.stale = true;
304                        row
305                    })
306                    .collect();
307                db.confirm_uploaded_topics(&restore, &[])?;
308                db.save_notification_record(&record)?;
309                Ok::<_, StorageError>(Continue(record.push_generation))
310            })?
311            .into_continued();
312            pending.clear();
313            generation
314        };
315        self.context.task_channels().wake_notifications();
316        let _guard = self
317            .context
318            .task_channels()
319            .notification_request
320            .lock()
321            .await;
322        let record = self.context.db().notification_record()?;
323        if record.push_generation == generation && record.push_state == 1 {
324            let result = worker::register(&self.context, &record, &config).await;
325            if let Ok(true) = &result {
326                worker::resume_after_registration(&self.context, generation)?;
327            }
328            drop(_guard);
329            self.context.task_channels().wake_notifications();
330            result?;
331        }
332        Ok(self.notification_state()?)
333    }
334
335    /// Disable locally before unregistering. The recipient identity and overrides stay.
336    /// A failed unregister leaves the client disabled; the backend recipient expires.
337    #[xmtp_common::rpc_span]
338    pub async fn disable_notifications(&self) -> Result<(), NotificationError> {
339        let record = {
340            let mut pending = self
341                .context
342                .task_channels()
343                .notification_pending_topics
344                .lock();
345            let (record, cleared) =
346                crate::state_tx::state_write(self.context.mls_storage(), |tx| {
347                    Ok::<_, StorageError>(Continue(tx.storage().db().disable_notifications()?))
348                })?
349                .into_continued();
350            pending.extend(cleared.into_iter().map(|row| (row.topic.clone(), row)));
351            record
352        };
353        self.context.task_channels().wake_notifications();
354        if record.push_recipient_id.is_none() {
355            return Ok(());
356        }
357        let _guard = self
358            .context
359            .task_channels()
360            .notification_request
361            .lock()
362            .await;
363        // A later disable still wants this recipient removed. The request
364        // lock orders any subsequent enable's Register after Unregister.
365        if self.context.db().notification_record()?.push_state != 0 {
366            self.context.task_channels().wake_notifications();
367            return Ok(());
368        }
369        let request = backend_v1::UnregisterRequest {
370            recipient_id: record.push_recipient_id.unwrap_or_default(),
371            recipient_secret: record.push_recipient_secret.unwrap_or_default(),
372        };
373        match worker::bounded(self.context.api().unregister(request)).await {
374            Ok(_) | Err(NotificationError::NotFound) => {
375                self.context
376                    .task_channels()
377                    .notification_pending_topics
378                    .lock()
379                    .clear();
380                Ok(())
381            }
382            Err(error) => Err(error),
383        }
384    }
385
386    /// Read the locally stored state without a backend call.
387    pub fn notification_state(&self) -> Result<NotificationState, StorageError> {
388        state(&self.context.db().notification_record()?)
389    }
390}
391
392/// Resolve only local rules. Active membership is checked separately by the task.
393pub(crate) fn effective(
394    config: &NotificationConfig,
395    group: &xmtp_db::group::StoredGroup,
396    consent: ConsentState,
397) -> bool {
398    use xmtp_proto::types::ConversationType;
399    if group.conversation_type == ConversationType::Sync {
400        return config.include_sync_groups;
401    }
402    if !matches!(
403        group.conversation_type,
404        ConversationType::Group | ConversationType::Dm
405    ) {
406        return false;
407    }
408    match group.push_override {
409        Some(1) => true,
410        Some(0) => false,
411        _ => config.consent_states.contains(&consent),
412    }
413}
414
415impl<Context: XmtpSharedContext> MlsGroup<Context> {
416    /// Set an override after which the task recomputes the desired set.
417    pub fn set_notifications(&self, value: NotificationOverride) -> Result<(), StorageError> {
418        crate::state_tx::state_write(self.context.mls_storage(), |tx| {
419            tx.storage().db().set_notification_override(
420                &self.group_id,
421                match value {
422                    NotificationOverride::Enabled => Some(1),
423                    NotificationOverride::Disabled => Some(0),
424                    NotificationOverride::Default => None,
425                },
426            )?;
427            Ok::<_, StorageError>(Continue(()))
428        })?;
429        self.context.task_channels().wake_notifications();
430        Ok(())
431    }
432
433    /// Return the effective local rule for this conversation.
434    pub fn notifications_enabled(&self) -> Result<bool, StorageError> {
435        let db = self.context.db();
436        let record = db.notification_record()?;
437        if record.push_state != 1 {
438            return Ok(false);
439        }
440        let config: NotificationConfig = decode(
441            record
442                .push_config
443                .as_deref()
444                .ok_or(StorageError::DbDeserialize)?,
445        )?;
446        let group = db
447            .find_group(&self.group_id)?
448            .ok_or(xmtp_db::NotFound::GroupById(self.group_id))?;
449        let consent = db
450            .get_consent_record(hex::encode(self.group_id), ConsentType::ConversationId)?
451            .map(|row| row.state)
452            .unwrap_or(ConsentState::Unknown);
453        Ok(effective(&config, &group, consent))
454    }
455}
456
457#[cfg(test)]
458mod tests;
xmtp_mls/client/notifications.rs

xmtp_mls/client/
notifications.rs
