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

xmtp_mls/client/
notifications.rs
