xmtp_mls/groups/mls_sync/
receive.rs

1//! Receiving envelopes and storing transcript messages.
2
3use super::*;
4
5impl<Context> MlsGroup<Context>
6where
7    Context: XmtpSharedContext,
8{
9    /// Wait for a fixed network prefix. The summary is local history, not proof of completion.
10    #[tracing::instrument(skip_all, level = "trace")]
11    pub async fn receive(&self) -> Result<ProcessSummary, GroupError> {
12        use xmtp_db::delivery::{DeliveryScope, QueryDelivery};
13        let db = self.context.db();
14        let mut position = db.current_delivery_cursor()?;
15        crate::subscriptions::barrier::receive_through_current(
16            &self.context,
17            vec![xmtp_proto::types::Topic::new_group_message(self.group_id)],
18        )
19        .await?;
20        let upper = db.current_delivery_cursor()?;
21        let settings = self.context.incoming_runtime().policy();
22        let mut summary = ProcessSummary::default();
23        loop {
24            let rows = db.replay_delivery_messages_bounded(
25                position,
26                &DeliveryScope::Groups(vec![self.group_id]),
27                xmtp_common::time::now_ns(),
28                settings.max_local_read_rows,
29                settings.max_local_read_bytes,
30            )?;
31            if rows.is_empty() {
32                break;
33            }
34            let mut reached_upper = false;
35            for row in rows {
36                if row.cursor.delivery_sequence > upper.delivery_sequence {
37                    reached_upper = true;
38                    break;
39                }
40                position = row.cursor;
41                let message = row.message;
42                summary.add_id(message.cursor());
43                summary.add(MessageIdentifier {
44                    cursor: message.cursor(),
45                    group_id: message.group_id,
46                    created_ns: chrono::DateTime::from_timestamp_nanos(message.sent_at_ns),
47                    previously_processed: false,
48                    internal_id: Some(message.id),
49                    group_context: None,
50                    intent_kind: None,
51                });
52            }
53            if reached_upper || position.delivery_sequence >= upper.delivery_sequence {
54                break;
55            }
56        }
57        Ok(summary)
58    }
59
60    #[tracing::instrument(skip_all, level = "trace")]
61    pub(super) fn maybe_update_cursor(
62        &self,
63        db: &impl DbQuery,
64        message: &xmtp_proto::types::GroupMessage,
65    ) -> Result<bool, StorageError> {
66        let updated = db.update_cursor(
67            message.group_id,
68            xmtp_db::refresh_state::EntityKind::ApplicationMessage,
69            message.cursor,
70        )?;
71        if updated {
72            log_event!(
73                Event::GroupCursorUpdate,
74                self.context.installation_id(),
75                group_id = message.group_id.as_slice(),
76                cursor = message.cursor.0,
77            );
78        } else {
79            tracing::debug!("no cursor update required");
80        }
81        Ok(updated)
82    }
83
84    pub(super) fn save_transcript_message(
85        &self,
86        validated_commit: ValidatedCommit,
87        timestamp_ns: u64,
88        cursor: Cursor,
89        storage: &impl XmtpMlsStorageProvider,
90    ) -> Result<Option<(StoredGroupMessage, GroupUpdated)>, GroupMessageProcessingError> {
91        if validated_commit.is_empty() {
92            return Ok(None);
93        }
94        let sender_installation_id = validated_commit.actor_installation_id();
95        let sender_inbox_id = validated_commit.actor_inbox_id();
96
97        let pending_remove_users = &storage.db().get_pending_remove_users(&self.group_id)?;
98        let payload: GroupUpdated = validated_commit.into_with(pending_remove_users);
99        tracing::info!("Storing transcript message");
100        let encoded_payload = GroupUpdatedCodec::encode(payload.clone())?;
101        let mut encoded_payload_bytes = Vec::new();
102        encoded_payload.encode(&mut encoded_payload_bytes)?;
103
104        let message_id = calculate_message_id(
105            self.group_id,
106            encoded_payload_bytes.as_slice(),
107            &timestamp_ns.to_string(),
108        );
109        let content_type = encoded_payload.r#type.unwrap_or_else(|| {
110            tracing::warn!("Missing content type in encoded payload, using default values");
111            // Default content type values
112            xmtp_proto::xmtp::mls::message_contents::ContentTypeId {
113                authority_id: "unknown".to_string(),
114                type_id: "unknown".to_string(),
115                version_major: 0,
116                version_minor: 0,
117            }
118        });
119
120        self.handle_metadata_update_from_commit(&payload.metadata_field_changes, storage)?;
121
122        // When a DM is stitched, it can repeat group updates. We want to prevent saving those messages.
123        if self.update_already_exists(&payload, storage)? {
124            return Ok(None);
125        }
126
127        let msg = StoredGroupMessage {
128            id: message_id,
129            group_id: self.group_id,
130            decrypted_message_bytes: encoded_payload_bytes,
131            sent_at_ns: timestamp_ns as i64,
132            kind: GroupMessageKind::MembershipChange,
133            sender_installation_id,
134            sender_inbox_id,
135            delivery_status: DeliveryStatus::Published,
136            content_type: content_type.type_id.into(),
137            version_major: content_type.version_major as i32,
138            version_minor: content_type.version_minor as i32,
139            authority_id: content_type.authority_id.to_string(),
140            reference_id: None,
141            sequence_id: cursor.0 as i64,
142            envelope_hash: None,
143            expiry_ns: None,
144            expire_at_ns: None,
145            inserted_at_ns: 0, // Will be set by database
146            should_push: true,
147            // Matches the key used to derive `message_id` above.
148            idempotency_key: timestamp_ns.to_string(),
149        };
150
151        msg.store_or_ignore(&storage.db())?;
152        Ok(Some((msg, payload)))
153    }
154
155    fn update_already_exists(
156        &self,
157        payload: &GroupUpdated,
158        storage: &impl XmtpMlsStorageProvider,
159    ) -> Result<bool, GroupMessageProcessingError> {
160        if self.dm_id.is_none() || payload.added_inboxes.is_empty() {
161            // Only dedupe for DMs.
162            // Only dedupe for group adds.
163            return Ok(false);
164        }
165
166        let mut deduper = GroupUpdateDeduper::default();
167        let mut inserted_after_ns = None;
168        let mut msgs;
169        loop {
170            // DMs are stitched, so we don't want to have the same
171            // group updates from multiple DMs being saved to the database.
172            msgs = self.find_messages_v2_with_conn(
173                &MsgQueryArgs {
174                    content_types: Some(vec![ContentType::GroupUpdated]),
175                    inserted_after_ns,
176                    limit: Some(100),
177                    ..Default::default()
178                },
179                storage.db(),
180            )?;
181
182            let Some(msg) = msgs.last() else {
183                break;
184            };
185            inserted_after_ns = Some(msg.metadata.inserted_at_ns);
186
187            for msg in msgs {
188                let MessageBody::GroupUpdated(update) = msg.content else {
189                    continue;
190                };
191
192                deduper.consume(&update);
193            }
194        }
195
196        Ok(deduper.is_dupe(payload))
197    }
198}
xmtp_mls/groups/mls_sync/receive.rs

xmtp_mls/groups/mls_sync/
receive.rs
