xmtp_mls_core/
events.rs

1//! Facts used by this client's internal subscribers.
2
3use xmtp_db::{
4    ConnectionError, DbQuery,
5    consent_record::{
6        ConsentChange, ConsentKey, ConsentState as StoredConsentState, ConsentType, ConsentValue,
7        StoredConsentRecord,
8    },
9    group_message::StoredGroupMessage,
10    user_preferences::StoredUserPreferences,
11};
12use xmtp_mls_group_state::group_metadata::DmMembers;
13use xmtp_proto::types::GroupId;
14
15use xmtp_cryptography::secret::Secret;
16use xmtp_events::{
17    ClientEvent, ConsentChanged, ConsentEntityKind, ConsentState, DeletionCause, EventContext,
18    EventWriter, HmacKeysUpdated, MessageDeleted, MessageRef,
19};
20
21/// A stored preference change, for the legacy in-process streams. `Debug`
22/// shows the HMAC root key only as its length: `Secret` redacts it.
23#[derive(Clone, Debug)]
24pub enum PreferenceUpdate {
25    Consent(StoredConsentRecord),
26    Hmac { key: Secret, cycled_at_ns: i64 },
27}
28
29#[derive(Clone, Debug)]
30pub enum InternalEvent {
31    GroupJoined {
32        group_id: GroupId,
33        is_sync: bool,
34    },
35    MessageStored {
36        group_id: GroupId,
37        message_id: Vec<u8>,
38        expires_at_ns: Option<i64>,
39        is_sync: bool,
40    },
41    /// A consent record or the root key was marked owed.
42    OutboxChanged,
43    TaskScheduled,
44    NotificationSettingsChanged,
45    /// Stored consent and root-key changes, for the legacy in-process streams.
46    PreferencesChanged {
47        updates: Vec<PreferenceUpdate>,
48    },
49    /// Deleted rows whose expiry had not passed when the event was emitted.
50    /// A row can expire before delivery, so a reader checks it again.
51    MessagesDeleted(DeletedMessages),
52    /// Deleted rows whose expiry had passed: expiry cleanup, or a delete of
53    /// an already expired message. Their bodies are cleared.
54    MessagesExpired(DeletedMessages),
55}
56
57/// Rows removed by one delete, and the inbox that deleted them.
58#[derive(Clone, Debug)]
59pub struct DeletedMessages {
60    pub messages: Vec<StoredGroupMessage>,
61    /// The inbox that sent the delete. `None` when no inbox did: expiry
62    /// cleanup, or a local removal of the row.
63    pub deleted_by_inbox_id: Option<String>,
64}
65
66impl InternalEvent {
67    /// Whether the event reports deleted messages, by any cause.
68    pub fn is_message_deletion(&self) -> bool {
69        matches!(self, Self::MessagesDeleted(_) | Self::MessagesExpired(_))
70    }
71}
72
73/// Report stored consent changes: one event per changed key and entity the
74/// app names it by, the legacy stream, and a wake for the outbox when a change
75/// is owed. This is the only consent event emitter.
76// implements: CONS-059, EVENT-001, EVENT-059
77pub fn emit_consent_changes(
78    ctx: &impl crate::context::CoreContext,
79    events: &impl EventWriter<InternalEvent>,
80    db: &impl DbQuery,
81    changes: &[ConsentChange],
82) -> Result<(), ConnectionError> {
83    // One entry per key: the reported state before the first change, the value after the last.
84    let mut net: Vec<(&ConsentKey, StoredConsentState, ConsentValue)> = vec![];
85    for change in changes {
86        match net.iter_mut().find(|(key, ..)| **key == change.key) {
87            Some(entry) => entry.2 = change.value,
88            None => net.push((&change.key, change.before, change.value)),
89        }
90    }
91    let mut legacy = vec![];
92    let mut state_changed = false;
93    for (key, before, value) in net {
94        let changed = before != value.state;
95        state_changed |= changed;
96        for (entity_type, entity, context) in app_entities(db, ctx.inbox_id(), key)? {
97            if changed {
98                let kind = match entity_type {
99                    ConsentType::ConversationId => ConsentEntityKind::Conversation,
100                    ConsentType::InboxId => ConsentEntityKind::Inbox,
101                };
102                let event = ClientEvent::ConsentChanged(ConsentChanged {
103                    entity_kind: kind,
104                    entity: entity.clone(),
105                    state: match value.state {
106                        StoredConsentState::Unknown => ConsentState::Unknown,
107                        StoredConsentState::Allowed => ConsentState::Allowed,
108                        StoredConsentState::Denied => ConsentState::Denied,
109                    },
110                });
111                events.emit_with_context(Some(event), None, context);
112            }
113            legacy.push(PreferenceUpdate::Consent(StoredConsentRecord {
114                entity_type,
115                state: value.state,
116                entity,
117                consented_at_ns: value.consented_at_ns,
118                owed_seq: 0,
119            }));
120        }
121    }
122    if state_changed {
123        // implements: PUSH-260
124        ctx.task_channels().mark_notification_changed();
125    }
126    if !legacy.is_empty() {
127        events.emit(
128            None,
129            Some(InternalEvent::PreferencesChanged { updates: legacy }),
130        );
131    }
132    if changes.iter().any(|change| change.marked) {
133        events.emit(None, Some(InternalEvent::OutboxChanged));
134    }
135    Ok(())
136}
137
138/// The entities an app names for one changed key: an inbox, and
139/// the DM the app lists with that inbox; a conversation only when
140/// no held group reads another key. Never a virtual group.
141fn app_entities(
142    db: &impl DbQuery,
143    own_inbox: &str,
144    key: &ConsentKey,
145) -> Result<Vec<(ConsentType, String, EventContext)>, ConnectionError> {
146    let mut entities = vec![];
147    let held = match key.entity_type {
148        ConsentType::InboxId => {
149            entities.push((
150                ConsentType::InboxId,
151                key.entity.clone(),
152                EventContext::default(),
153            ));
154            let dm_id = DmMembers {
155                member_one_inbox_id: own_inbox,
156                member_two_inbox_id: key.entity.as_str(),
157            };
158            db.find_dm_by_dm_id(&dm_id.to_string())?
159        }
160        ConsentType::ConversationId => {
161            // A conversation entity that names no group id is reported as named.
162            let held = match key.group_id() {
163                Some(id) => db.find_group(&id)?,
164                None => None,
165            };
166            if held.is_none() {
167                entities.push((
168                    ConsentType::ConversationId,
169                    key.entity.clone(),
170                    EventContext::default(),
171                ));
172            }
173            held
174        }
175    };
176    if let Some(group) = held
177        && !group.conversation_type.is_virtual()
178        && ConsentKey::of_group(&group, own_inbox).as_ref() == Some(key)
179    {
180        let context = dm_event_context(group.dm_id.as_deref());
181        entities.push((ConsentType::ConversationId, hex::encode(group.id), context));
182    }
183    Ok(entities)
184}
185
186/// Report a stored root-key change: push re-registers, the legacy stream
187/// receives the key, and the outbox wakes when the key is owed.
188pub fn emit_root_key_changed(
189    events: &impl EventWriter<InternalEvent>,
190    db: &impl DbQuery,
191    marked: bool,
192) -> Result<(), xmtp_db::StorageError> {
193    let preferences = StoredUserPreferences::load(db)?;
194    let updates = preferences
195        .hmac_key
196        .map(|key| PreferenceUpdate::Hmac {
197            key: key.as_slice().to_vec().into(),
198            cycled_at_ns: preferences.hmac_key_cycled_at_ns.unwrap_or_default(),
199        })
200        .into_iter()
201        .collect();
202    events.emit(
203        Some(ClientEvent::HmacKeysUpdated(HmacKeysUpdated)),
204        Some(InternalEvent::PreferencesChanged { updates }),
205    );
206    if marked {
207        events.emit(None, Some(InternalEvent::OutboxChanged));
208    }
209    Ok(())
210}
211
212/// An event context naming the DM `dm_id`, if any.
213pub fn dm_event_context(dm_id: Option<&str>) -> EventContext {
214    EventContext {
215        dm_identifier: dm_id.map(|id| id.as_bytes().to_vec()),
216        ..Default::default()
217    }
218}
219
220/// Emit the deletion of `messages` by the inbox `deleted_by_inbox_id`, or by
221/// no inbox (`None`) for a local removal of the rows.
222pub fn emit_deleted_messages(
223    writer: &impl EventWriter<InternalEvent>,
224    messages: Vec<StoredGroupMessage>,
225    cause: DeletionCause,
226    deleted_by_inbox_id: Option<String>,
227    db: &impl DbQuery,
228) -> Result<(), ConnectionError> {
229    if messages.is_empty() {
230        return Ok(());
231    }
232    for message in &messages {
233        let group = db.find_group(&message.group_id)?;
234        if group
235            .as_ref()
236            .is_some_and(|group| group.conversation_type.is_virtual())
237        {
238            continue;
239        }
240        let reference = MessageRef {
241            group_id: message.group_id.to_vec(),
242            message_id: message.id.clone(),
243        };
244        let event = ClientEvent::MessageDeleted(MessageDeleted {
245            group_id: reference.group_id,
246            message_id: reference.message_id,
247            cause,
248        });
249        let context = dm_event_context(group.as_ref().and_then(|group| group.dm_id.as_deref()));
250        writer.emit_with_context(Some(event), None, context);
251    }
252    // A delete of an already expired message must not deliver its body.
253    // implements: META-051
254    let now = xmtp_common::time::now_ns();
255    let (expired, live): (Vec<_>, Vec<_>) = messages
256        .into_iter()
257        .partition(|message| message.expire_at_ns.is_some_and(|at| at <= now));
258    if !live.is_empty() {
259        writer.emit(
260            None,
261            Some(InternalEvent::MessagesDeleted(DeletedMessages {
262                messages: live,
263                deleted_by_inbox_id: deleted_by_inbox_id.clone(),
264            })),
265        );
266    }
267    if !expired.is_empty() {
268        let expired = expired
269            .into_iter()
270            .map(|message| StoredGroupMessage {
271                decrypted_message_bytes: Vec::new(),
272                ..message
273            })
274            .collect();
275        writer.emit(
276            None,
277            Some(InternalEvent::MessagesExpired(DeletedMessages {
278                messages: expired,
279                deleted_by_inbox_id,
280            })),
281        );
282    }
283    Ok(())
284}
285
286pub(crate) fn emit_expired_messages(
287    writer: &impl EventWriter<InternalEvent>,
288    messages: Vec<StoredGroupMessage>,
289    db: &impl DbQuery,
290) -> Result<(), ConnectionError> {
291    if messages.is_empty() {
292        return Ok(());
293    }
294    for message in &messages {
295        let group = db.find_group(&message.group_id)?;
296        if group
297            .as_ref()
298            .is_some_and(|group| group.conversation_type.is_virtual())
299        {
300            continue;
301        }
302        let context = dm_event_context(group.as_ref().and_then(|group| group.dm_id.as_deref()));
303        writer.emit_with_context(
304            Some(ClientEvent::MessageExpired(MessageRef {
305                group_id: message.group_id.to_vec(),
306                message_id: message.id.clone(),
307            })),
308            None,
309            context,
310        );
311    }
312    writer.emit(
313        None,
314        Some(InternalEvent::MessagesExpired(DeletedMessages {
315            messages,
316            deleted_by_inbox_id: None,
317        })),
318    );
319    Ok(())
320}
321
322#[cfg(test)]
323mod tests {
324    use super::*;
325    use crate::{
326        context::CoreContext,
327        test_context::{real_context, stored_msg},
328    };
329    use xmtp_db::{
330        group::{GroupMembershipState, StoredGroup},
331        prelude::*,
332    };
333    use xmtp_events::{EventBus, EventFilter, EventKind};
334    use xmtp_proto::types::ConversationType;
335
336    // verifies: EVENT-020
337    #[xmtp_common::test(unwrap_try = true)]
338    async fn stitched_dm_delete_and_expiry_reach_group_filter() {
339        use xmtp_proto::types::Cursor;
340
341        let context = real_context().await;
342        let db = context.db();
343        let [ga, gb, gc] = [71, 72, 73].map(|id| GroupId::from([id; 16]));
344        for (id, dm_id) in [(ga, "stitched"), (gb, "stitched"), (gc, "other")] {
345            StoredGroup::builder()
346                .id(id)
347                .created_at_ns(xmtp_common::time::now_ns())
348                .membership_state(GroupMembershipState::Allowed)
349                .added_by_inbox_id(context.inbox_id())
350                .conversation_type(ConversationType::Dm)
351                .dm_id(Some(dm_id.to_string()))
352                .build()?
353                .store(&db)?;
354        }
355        let messages = [stored_msg(Cursor(1), gb), stored_msg(Cursor(2), gc)];
356        let bus = EventBus::<InternalEvent>::new();
357        let mut filter = EventFilter::new([EventKind::MessageDeleted, EventKind::MessageExpired]);
358        filter.group_ids = Some(vec![ga.to_vec()]);
359        filter.dm_identifiers = vec![db.find_group(&ga)?.unwrap().dm_id.unwrap().into_bytes()];
360        let public = bus.subscribe(filter, Some(10));
361        emit_deleted_messages(
362            &bus,
363            messages.to_vec(),
364            DeletionCause::DeletedLocally,
365            None,
366            &db,
367        )?;
368        emit_expired_messages(&bus, messages.to_vec(), &db)?;
369        let events: Vec<_> = public
370            .drain()
371            .into_iter()
372            .filter_map(|e| e.client)
373            .collect();
374        assert_eq!(events.len(), 2);
375        assert!(matches!(&events[0], ClientEvent::MessageDeleted(deleted)
376            if deleted.group_id == gb.to_vec()));
377        assert!(matches!(&events[1], ClientEvent::MessageExpired(expired)
378            if expired.group_id == gb.to_vec()));
379    }
380
381    /// One event per changed key and app entity: an inbox change is also
382    /// reported under the DM the app lists with that inbox, Restored too; a
383    /// group record that a held DM does not read, a virtual group, and a new
384    /// unknown record report nothing; the legacy stream gets every entity.
385    // verifies: CONS-059, EVENT-001, EVENT-059
386    #[xmtp_common::test(unwrap_try = true)]
387    async fn consent_events() {
388        use xmtp_db::consent_record::{ConsentChange, ConsentKey, ConsentValue};
389        let context = real_context().await;
390        let db = context.db();
391        let own = context.inbox_id();
392        let store = |id: u8, peer: Option<&str>, kind, state| {
393            let mut pair = peer.map(|peer| [own.to_string(), peer.to_string()]);
394            if let Some(pair) = &mut pair {
395                pair.sort();
396            }
397            StoredGroup::builder()
398                .id(GroupId::from([id; 16]))
399                .created_at_ns(1)
400                .membership_state(state)
401                .added_by_inbox_id(own)
402                .conversation_type(kind)
403                .dm_id(pair.map(|pair| format!("dm:{}", pair.join(":"))))
404                .build()
405                .unwrap()
406                .store(&db)
407                .unwrap();
408            GroupId::from([id; 16])
409        };
410        let restored = store(
411            1,
412            Some("peer-two"),
413            ConversationType::Dm,
414            GroupMembershipState::Restored,
415        );
416        let held_dm = store(
417            2,
418            Some("peer-three"),
419            ConversationType::Dm,
420            GroupMembershipState::Allowed,
421        );
422        let sync = store(
423            3,
424            None,
425            ConversationType::Sync,
426            GroupMembershipState::Allowed,
427        );
428        let change = |key: ConsentKey, before, state| ConsentChange {
429            key,
430            before,
431            value: ConsentValue {
432                state,
433                consented_at_ns: 1,
434            },
435            marked: false,
436        };
437        let bus = EventBus::<InternalEvent>::new();
438        let public = bus.subscribe(EventFilter::new([EventKind::ConsentChanged]), Some(16));
439        let legacy = bus.subscribe(
440            EventFilter::default()
441                .with_internal(|event| matches!(event, InternalEvent::PreferencesChanged { .. })),
442            Some(16),
443        );
444        let named = |events: Vec<xmtp_events::EventEnvelope<InternalEvent>>| -> Vec<String> {
445            events
446                .into_iter()
447                .filter_map(|event| match event.client {
448                    Some(ClientEvent::ConsentChanged(change)) => Some(change.entity),
449                    _ => None,
450                })
451                .collect()
452        };
453        use StoredConsentState::*;
454        for (changes, expected) in [
455            (
456                vec![change(ConsentKey::inbox("peer-one"), Unknown, Denied)],
457                vec!["peer-one".to_string()],
458            ),
459            (
460                vec![change(ConsentKey::inbox("peer-two"), Unknown, Denied)],
461                vec!["peer-two".to_string(), hex::encode(restored)],
462            ),
463            (
464                vec![change(ConsentKey::group(held_dm.as_ref()), Unknown, Denied)],
465                vec![],
466            ),
467            (
468                vec![change(ConsentKey::group(sync.as_ref()), Unknown, Denied)],
469                vec![],
470            ),
471            (
472                vec![change(ConsentKey::inbox("new"), Unknown, Unknown)],
473                vec![],
474            ),
475            (
476                vec![change(
477                    ConsentKey::parse(ConsentType::ConversationId, "opaque").unwrap(),
478                    Unknown,
479                    Denied,
480                )],
481                vec!["opaque".to_string()],
482            ),
483        ] {
484            emit_consent_changes(&context, &bus, &db, &changes)?;
485            assert_eq!(named(public.drain()), expected, "{changes:?}");
486        }
487        let entities: Vec<Vec<String>> = legacy
488            .drain()
489            .into_iter()
490            .filter_map(|event| match event.internal {
491                Some(InternalEvent::PreferencesChanged { updates }) => Some(
492                    updates
493                        .into_iter()
494                        .filter_map(|update| match update {
495                            PreferenceUpdate::Consent(record) => Some(record.entity),
496                            PreferenceUpdate::Hmac { .. } => None,
497                        })
498                        .collect(),
499                ),
500                _ => None,
501            })
502            .collect();
503        assert_eq!(
504            entities,
505            vec![
506                vec!["peer-one".to_string()],
507                vec!["peer-two".to_string(), hex::encode(restored)],
508                vec!["new".to_string()],
509                vec!["opaque".to_string()],
510            ]
511        );
512    }
513}
xmtp_mls_core/events.rs

xmtp_mls_core/
events.rs
