xmtp_mls/groups/
message_list.rs

1use crate::context::XmtpSharedContext;
2use crate::groups::MlsGroup;
3use crate::messages::decoded_message::DecodedMessage;
4use crate::messages::enrichment::{
5    EnrichMessageError, EnrichedStoredMessage, enrich_messages, enrich_messages_with_stored,
6};
7use xmtp_db::DbQuery;
8use xmtp_db::group_message::{ContentType as DbContentType, MsgQueryArgs};
9use xmtp_db::prelude::QueryGroupMessage;
10
11mod page;
12pub use page::{EnrichedHistoryPage, EnrichedRecoveryPage};
13
14impl<Context> MlsGroup<Context>
15where
16    Context: XmtpSharedContext,
17{
18    #[xmtp_common::mls_span]
19    pub fn find_messages_v2(
20        &self,
21        query: &MsgQueryArgs,
22    ) -> Result<Vec<DecodedMessage>, EnrichMessageError> {
23        let conn = self.context.db();
24        self.find_messages_v2_with_conn(query, conn)
25    }
26
27    #[xmtp_common::mls_span]
28    pub fn find_messages_v2_with_stored(
29        &self,
30        query: &MsgQueryArgs,
31    ) -> Result<Vec<EnrichedStoredMessage>, EnrichMessageError> {
32        let conn = self.context.db();
33        use xmtp_db::delivery::QueryDelivery;
34        let rows = conn
35            .app_visible_message_rows(
36                &self.group_id,
37                &filter_out_hidden_message_types_from_query(query),
38            )
39            .map_err(EnrichMessageError::Storage)?;
40        let cursors: std::collections::HashMap<_, _> = rows
41            .iter()
42            .map(|row| (row.stored.id.clone(), row.cursor))
43            .collect();
44        let mut enriched = enrich_messages_with_stored(
45            conn,
46            &self.group_id,
47            rows.into_iter().map(|row| row.stored).collect(),
48        )?;
49        for message in &mut enriched {
50            message.delivery_cursor = cursors.get(&message.stored.id).copied().flatten();
51        }
52        Ok(enriched)
53    }
54
55    #[xmtp_common::mls_span]
56    pub fn find_messages_v2_with_conn<C>(
57        &self,
58        query: &MsgQueryArgs,
59        conn: C,
60    ) -> Result<Vec<DecodedMessage>, EnrichMessageError>
61    where
62        C: QueryGroupMessage + DbQuery,
63    {
64        let initial_messages = conn.get_group_messages(
65            &self.group_id,
66            &filter_out_hidden_message_types_from_query(query),
67        )?;
68
69        enrich_messages(conn, &self.group_id, initial_messages)
70    }
71}
72
73fn filter_out_hidden_message_types_from_query(query: &MsgQueryArgs) -> MsgQueryArgs {
74    let mut new_query = query.clone();
75    let hidden_message_types = vec![
76        DbContentType::Reaction,
77        DbContentType::ReadReceipt,
78        DbContentType::DeleteMessage,
79    ];
80
81    let excluded_content_types = match &query.exclude_content_types {
82        Some(excluded) => {
83            let mut content_types = excluded.clone();
84            content_types.extend(
85                hidden_message_types
86                    .into_iter()
87                    .filter(|t| !excluded.contains(t)),
88            );
89            content_types
90        }
91        None => hidden_message_types,
92    };
93    new_query.exclude_content_types = Some(excluded_content_types);
94    new_query
95}
96
97#[cfg(test)]
98mod tests {
99    use super::*;
100    use crate::builder::ClientBuilder;
101    use crate::groups::QueryableContentFields;
102    use crate::messages::decoded_message::{ContentDecodeFailureKind, MessageBody};
103    use crate::prelude::*;
104    use hex::ToHexExt;
105    use xmtp_common::time::now_ns;
106    use xmtp_content_types::ContentCodec;
107    use xmtp_content_types::reaction::ReactionCodec;
108    use xmtp_content_types::test_utils::TestContentGenerator;
109    use xmtp_content_types::text::TextCodec;
110    use xmtp_cryptography::utils::generate_local_wallet;
111    use xmtp_db::Store;
112    use xmtp_db::group_message::{
113        ContentType as DbContentType, DeliveryStatus, GroupMessageKind, StoredGroupMessage,
114    };
115    use xmtp_proto::types::GroupId;
116    use xmtp_proto::xmtp::mls::message_contents::content_types::ReactionAction;
117    use xmtp_proto::xmtp::mls::message_contents::{ContentTypeId, EncodedContent};
118
119    async fn setup_test_group() -> (MlsGroup<impl XmtpSharedContext>, impl XmtpSharedContext) {
120        let client = ClientBuilder::new_test_client(&generate_local_wallet()).await;
121        let group = client.create_group(None, Default::default()).unwrap();
122
123        (group, client.context.clone())
124    }
125
126    fn create_test_message(
127        group_id: &GroupId,
128        message_id: Vec<u8>,
129        encoded_content: EncodedContent,
130        sent_at_ns: i64,
131    ) -> StoredGroupMessage {
132        let content_bytes = xmtp_content_types::encoded_content_to_bytes(encoded_content.clone());
133        let queryable_fields =
134            QueryableContentFields::try_from(encoded_content).unwrap_or_default();
135
136        StoredGroupMessage {
137            id: message_id,
138            group_id: *group_id,
139            decrypted_message_bytes: content_bytes,
140            sent_at_ns,
141            kind: GroupMessageKind::Application,
142            sender_installation_id: vec![1, 2, 3],
143            sender_inbox_id: "sender".into(),
144            delivery_status: DeliveryStatus::Published,
145            content_type: queryable_fields.content_type,
146            version_major: queryable_fields.version_major,
147            version_minor: queryable_fields.version_minor,
148            authority_id: queryable_fields.authority_id,
149            reference_id: queryable_fields.reference_id,
150            sequence_id: 0,
151            envelope_hash: None,
152            expiry_ns: None,
153            expiry_unknown: false,
154            deleted_by_inbox_id: None,
155            expire_at_ns: None,
156            inserted_at_ns: 0,
157            should_push: true,
158            idempotency_key: sent_at_ns.to_string(),
159        }
160    }
161
162    // For tests that need malformed content
163    fn create_test_message_raw(
164        group_id: &GroupId,
165        message_id: Vec<u8>,
166        content: Vec<u8>,
167        sent_at_ns: i64,
168        content_type_id: Option<ContentTypeId>,
169        reference_id: Option<Vec<u8>>,
170    ) -> StoredGroupMessage {
171        let queryable_fields = content_type_id
172            .as_ref()
173            .map(|ct| {
174                (
175                    DbContentType::from_identifier(&ct.authority_id, &ct.type_id, ct.version_major),
176                    ct.version_major as i32,
177                    ct.version_minor as i32,
178                    ct.authority_id.clone(),
179                )
180            })
181            .unwrap_or((DbContentType::Text, 1, 0, "xmtp.org".to_string()));
182
183        StoredGroupMessage {
184            id: message_id,
185            group_id: *group_id,
186            decrypted_message_bytes: content,
187            sent_at_ns,
188            kind: GroupMessageKind::Application,
189            sender_installation_id: vec![1, 2, 3],
190            sender_inbox_id: "sender".into(),
191            delivery_status: DeliveryStatus::Published,
192            content_type: queryable_fields.0,
193            version_major: queryable_fields.1,
194            version_minor: queryable_fields.2,
195            authority_id: queryable_fields.3,
196            reference_id,
197            sequence_id: 0,
198            envelope_hash: None,
199            expiry_ns: None,
200            expiry_unknown: false,
201            deleted_by_inbox_id: None,
202            expire_at_ns: None,
203            inserted_at_ns: 0,
204            should_push: true,
205            idempotency_key: sent_at_ns.to_string(),
206        }
207    }
208
209    /// Stores message `n` with id `[n; 32]`, sent `n` microseconds from now.
210    fn create_and_store_message<S>(conn: &S, group_id: &GroupId, n: u8, content: EncodedContent)
211    where
212        StoredGroupMessage: Store<S>,
213    {
214        let sent_at_ns = now_ns() + i64::from(n) * 1000;
215        create_test_message(group_id, vec![n; 32], content, sent_at_ns)
216            .store(conn)
217            .unwrap();
218    }
219
220    /// The hex reference to message `n`.
221    fn ref_to(n: u8) -> String {
222        hex::encode([n; 32])
223    }
224
225    /// The `n` of each message stored with id `[n; 32]`.
226    fn ids(messages: &[DecodedMessage]) -> Vec<u8> {
227        messages
228            .iter()
229            .map(|message| message.metadata.id[0])
230            .collect()
231    }
232
233    fn emojis(message: &DecodedMessage) -> Vec<&str> {
234        message
235            .reactions
236            .iter()
237            .filter_map(|reaction| match &reaction.content {
238                MessageBody::Reaction(reaction) => Some(reaction.content.as_str()),
239                _ => None,
240            })
241            .collect()
242    }
243
244    // ===== Tests =====
245
246    #[xmtp_common::test(unwrap_try = true)]
247    async fn test_exclude_content_types_with_custom_exclusions() {
248        let (group, context) = setup_test_group().await;
249        let conn = context.db();
250        create_and_store_message(
251            &conn,
252            &group.group_id,
253            1,
254            TestContentGenerator::text_content("Hello World"),
255        );
256
257        // Store a locally generated GroupUpdated transcript.
258        let mut transcript = create_test_message(
259            &group.group_id,
260            vec![2; 32],
261            TestContentGenerator::group_updated_content(vec!["inbox1".to_string()]),
262            now_ns() + 2000,
263        );
264        transcript.kind = GroupMessageKind::MembershipChange;
265        transcript.store(&conn)?;
266
267        let messages = group.find_messages_v2(&MsgQueryArgs::default())?;
268        assert_eq!(ids(&messages), [1, 2]);
269
270        // Excluding Text returns only the GroupUpdated message
271        let query = MsgQueryArgs {
272            exclude_content_types: Some(vec![DbContentType::Text]),
273            ..Default::default()
274        };
275        let messages = group.find_messages_v2(&query)?;
276        assert_eq!(ids(&messages), [2]);
277        assert!(
278            matches!(messages[0].content, MessageBody::GroupUpdated(_)),
279            "{:?}",
280            messages[0].content
281        );
282    }
283
284    /// Reactions attach to their target and are hidden from the list, as are
285    /// read receipts and deletions. A malformed reaction row is not attached.
286    #[xmtp_common::test(unwrap_try = true)]
287    async fn reactions_attach_to_their_target_and_hidden_types_are_filtered() {
288        let (group, context) = setup_test_group().await;
289        let conn = context.db();
290        let put = |n, content| create_and_store_message(&conn, &group.group_id, n, content);
291
292        put(1, TestContentGenerator::text_content("Hello World"));
293        let reaction = |emoji| {
294            TestContentGenerator::reaction_content(&ref_to(1), emoji, ReactionAction::Added)
295        };
296        put(2, reaction("👍"));
297        put(3, reaction("❤️"));
298        create_test_message_raw(
299            &group.group_id,
300            vec![4; 32],
301            b"invalid reaction bytes".to_vec(),
302            now_ns() + 4000,
303            Some(ReactionCodec::content_type()),
304            Some(vec![1; 32]),
305        )
306        .store(&conn)?;
307        put(5, TestContentGenerator::read_receipt_content());
308        put(6, TestContentGenerator::delete_message_content(&ref_to(1)));
309        put(7, TestContentGenerator::text_content("Second message"));
310
311        let messages = group.find_messages_v2(&MsgQueryArgs::default())?;
312        assert_eq!(ids(&messages), [1, 7], "hidden types are filtered");
313        assert!(
314            matches!(&messages[0].content, MessageBody::Text(text) if text.content == "Hello World")
315        );
316        assert!(
317            matches!(&messages[1].content, MessageBody::Text(text) if text.content == "Second message")
318        );
319        assert_eq!(emojis(&messages[0]), ["👍", "❤️"]);
320        assert_eq!(messages[0].reactions.len(), 2, "malformed reaction");
321        assert!(messages[1].reactions.is_empty());
322    }
323
324    /// Each message carries its 50 latest reaction messages, and a count of
325    /// all of its unexpired ones, on every read path. A parent with fewer
326    /// reactions keeps all of them.
327    // verifies: PROC-053, META-051
328    #[xmtp_common::test(unwrap_try = true)]
329    async fn history_reaction_summary_is_bounded_per_message() {
330        const REACTION_COUNT: i64 = 256;
331        const SUMMARY: usize = 50;
332
333        let client = ClientBuilder::new_test_client(&generate_local_wallet()).await;
334        let group = client.create_group(None, Default::default())?;
335        let conn = client.context.db();
336        let base = now_ns();
337        let busy_parent = vec![0xa5; 32];
338        let quiet_parent = vec![0xb6; 32];
339        for parent in [&busy_parent, &quiet_parent] {
340            create_test_message(
341                &group.group_id,
342                parent.clone(),
343                TestContentGenerator::text_content("target"),
344                base,
345            )
346            .store(&conn)?;
347        }
348        let reaction = |parent: &[u8], id: Vec<u8>, offset: i64| {
349            create_test_message(
350                &group.group_id,
351                id,
352                TestContentGenerator::reaction_content(
353                    &parent.encode_hex(),
354                    "👍",
355                    ReactionAction::Added,
356                ),
357                base + offset,
358            )
359        };
360        // Each row is a distinct, accepted reaction from a participant.
361        for offset in 1..=REACTION_COUNT {
362            reaction(&busy_parent, offset.to_be_bytes().to_vec(), offset).store(&conn)?;
363        }
364        // The newest reaction has expired, so it neither shows nor counts.
365        let mut expired = reaction(&busy_parent, vec![0xee; 8], REACTION_COUNT + 1);
366        expired.expire_at_ns = Some(now_ns() - 1);
367        expired.store(&conn)?;
368        for offset in 1..=3 {
369            reaction(&quiet_parent, vec![0xb6, offset as u8], offset).store(&conn)?;
370        }
371
372        let assert_summary = |message: &DecodedMessage| {
373            let offsets: Vec<_> = message
374                .reactions
375                .iter()
376                .map(|reaction| reaction.metadata.sent_at_ns - base)
377                .collect();
378            if message.metadata.id == busy_parent {
379                let latest: Vec<_> =
380                    (REACTION_COUNT - SUMMARY as i64 + 1..=REACTION_COUNT).collect();
381                assert_eq!(offsets, latest);
382                assert_eq!(message.num_reaction_messages, REACTION_COUNT as usize);
383            } else {
384                assert_eq!(offsets, [1, 2, 3]);
385                assert_eq!(message.num_reaction_messages, 3);
386            }
387        };
388
389        let history = group.find_messages_v2(&MsgQueryArgs::default())?;
390        assert_eq!(history.len(), 2);
391        history.iter().for_each(assert_summary);
392        for parent in [&busy_parent, &quiet_parent] {
393            assert_summary(&client.message_v2(parent.clone())?);
394        }
395        let sdk_history = group.find_messages_v2_with_stored(&MsgQueryArgs::default())?;
396        assert_eq!(sdk_history.len(), 2);
397        sdk_history
398            .iter()
399            .for_each(|message| assert_summary(&message.decoded));
400    }
401
402    /// A reply carries its parent one level deep. A reference to a missing
403    /// message, or one that is not a message id, leaves the reply without one.
404    #[xmtp_common::test(unwrap_try = true)]
405    async fn replies_resolve_one_level_of_parent() {
406        let (group, context) = setup_test_group().await;
407        let conn = context.db();
408        let put = |n, content| create_and_store_message(&conn, &group.group_id, n, content);
409        let reply_to = |reference: &str| {
410            let content = b"reply".to_vec();
411            TestContentGenerator::reply_content(reference, TextCodec::content_type(), content)
412        };
413
414        put(1, TestContentGenerator::text_content("Original"));
415        put(2, reply_to(&ref_to(1)));
416        put(3, reply_to(&ref_to(2)));
417        put(4, reply_to(&ref_to(99)));
418        put(5, reply_to("not-valid-hex!@#"));
419
420        let messages = group.find_messages_v2(&MsgQueryArgs::default())?;
421        assert_eq!(ids(&messages), [1, 2, 3, 4, 5]);
422        let reply = |n: usize| match &messages[n - 1].content {
423            MessageBody::Reply(reply) => reply,
424            other => panic!("message {n}: expected reply, got {other:?}"),
425        };
426        for (n, reference, parent) in [
427            (2, ref_to(1), Some(1)),
428            (3, ref_to(2), Some(2)),
429            (4, ref_to(99), None),
430            (5, "not-valid-hex!@#".into(), None),
431        ] {
432            let in_reply_to = reply(n).in_reply_to.as_ref();
433            assert_eq!(reply(n).reference_id, reference, "reply {n}");
434            assert_eq!(in_reply_to.map(|p| p.metadata.id[0]), parent, "reply {n}");
435        }
436        let parent = |n| reply(n).in_reply_to.as_deref().expect("parent");
437        assert!(
438            matches!(&parent(2).content, MessageBody::Text(text) if text.content == "Original")
439        );
440        assert!(
441            matches!(&parent(3).content, MessageBody::Reply(inner) if inner.in_reply_to.is_none()),
442            "parents are not resolved recursively"
443        );
444    }
445
446    /// In a stitched DM, a deletion applies to its target, and to a reply's
447    /// parent, only when it is stored in the target's group.
448    // verifies: PROC-037
449    #[xmtp_common::test(unwrap_try = true)]
450    async fn stitched_deletions_apply_in_their_own_group() {
451        use xmtp_common::Generate;
452        use xmtp_db::ConnectionExt;
453        use xmtp_db::diesel::{ExpressionMethods, QueryDsl, RunQueryDsl};
454        use xmtp_db::group::{ConversationType, QueryGroup};
455        use xmtp_db::schema::groups::dsl;
456
457        let (group, context) = setup_test_group().await;
458        let conn = context.db();
459        let home = group.group_id;
460        conn.raw_query(|sqlite| {
461            xmtp_db::diesel::update(dsl::groups.find(home))
462                .set((
463                    dsl::conversation_type.eq(ConversationType::Dm),
464                    dsl::dm_id.eq("dm:sender:peer"),
465                ))
466                .execute(sqlite)
467        })?;
468        let mut sibling = conn.find_group(&home)??;
469        sibling.id = GroupId::generate();
470        sibling.store(&conn)?;
471        let sibling = sibling.id;
472        let put = |group_id, n, content| create_and_store_message(&conn, &group_id, n, content);
473        let reply_to = |n| {
474            let content = b"reply".to_vec();
475            TestContentGenerator::reply_content(&ref_to(n), TextCodec::content_type(), content)
476        };
477        put(sibling, 1, TestContentGenerator::text_content("deleted"));
478        put(sibling, 2, TestContentGenerator::text_content("kept"));
479        put(home, 3, reply_to(1));
480        put(home, 4, reply_to(2));
481        // Message 1 is deleted in its own group, message 2 from the other one.
482        for (group_id, n, target) in [(sibling, 5, 1), (home, 6, 2)] {
483            let deletion = TestContentGenerator::delete_message_content(&ref_to(target));
484            put(group_id, n, deletion);
485            let mut row = conn.get_group_message(vec![n; 32])??;
486            row.reference_id = Some(vec![target; 32]);
487            crate::groups::deletion::apply_deletion(
488                &conn,
489                &crate::groups::deletion::DiscardEvents,
490                &row,
491                crate::groups::deletion::Source::Topic {
492                    super_admin: false,
493                    cause: xmtp_events::DeletionCause::Deleted,
494                },
495            )?;
496        }
497
498        let messages = group.find_messages_v2_with_stored(&MsgQueryArgs::default())?;
499        let messages: Vec<_> = messages.into_iter().map(|m| m.decoded).collect();
500        assert_eq!(ids(&messages), [1, 2, 3, 4]);
501        let parent = |n: usize| match &messages[n - 1].content {
502            MessageBody::Reply(reply) => reply.in_reply_to.as_deref().expect("parent"),
503            other => panic!("message {n}: expected reply, got {other:?}"),
504        };
505        // The text shown, or None for a deletion placeholder.
506        let text = |message: &DecodedMessage| match &message.content {
507            MessageBody::Text(text) => Some(text.content.clone()),
508            MessageBody::DeletedMessage { .. } => None,
509            other => panic!("expected text or a deletion, got {other:?}"),
510        };
511        assert_eq!(text(&messages[0]), None, "target 1");
512        assert_eq!(text(parent(3)), None, "parent of 3");
513        assert_eq!(text(&messages[1]), Some("kept".into()), "target 2");
514        assert_eq!(text(parent(4)), Some("kept".into()), "parent of 4");
515    }
516
517    // verifies: CTYPE-008, CTYPE-009
518    #[xmtp_common::test(unwrap_try = true)]
519    async fn test_find_messages_undecodable_messages() {
520        let (group, context) = setup_test_group().await;
521        let conn = context.db();
522        let put = |n, content| create_and_store_message(&conn, &group.group_id, n, content);
523
524        put(1, TestContentGenerator::text_content("Valid message"));
525        // An old row: bytes that are not an EncodedContent, cached as Text
526        // the way a client stored it before classification was checked.
527        let malformed = vec![0xff, 0x00, 0x80];
528        create_test_message_raw(
529            &group.group_id,
530            vec![2; 32],
531            malformed.clone(),
532            now_ns() + 2000,
533            None,
534            None,
535        )
536        .store(&conn)?;
537        // A complete typed envelope with no core codec
538        put(3, TestContentGenerator::invalid_content());
539
540        let messages = group.find_messages_v2(&MsgQueryArgs::default())?;
541        assert_eq!(ids(&messages), [1, 2, 3]);
542        assert!(
543            matches!(&messages[0].content, MessageBody::Text(text) if text.content == "Valid message")
544        );
545
546        assert_eq!(messages[1].metadata.content_type, None);
547        let MessageBody::Undecodable(undecodable) = &messages[1].content else {
548            panic!(
549                "expected undecodable content, got {:?}",
550                messages[1].content
551            );
552        };
553        assert_eq!(undecodable.raw_bytes, malformed);
554        assert_eq!(
555            undecodable.failure.kind,
556            ContentDecodeFailureKind::MalformedEnvelope
557        );
558
559        let MessageBody::Custom(content) = &messages[2].content else {
560            panic!("expected custom content for unknown type message");
561        };
562        assert_eq!(
563            content.encoded.fallback,
564            Some("Invalid message".to_string())
565        );
566        assert_eq!(
567            messages[2].fallback_text,
568            Some("Invalid message".to_string())
569        );
570    }
571
572    /// History, lookup by id, and a reply's parent all keep a malformed row
573    /// with its id and exact bytes, and later rows stay visible.
574    // verifies: CTYPE-008
575    #[xmtp_common::test(unwrap_try = true)]
576    async fn malformed_rows_are_kept_on_history_lookup_and_parent_paths() {
577        let client = ClientBuilder::new_test_client(&generate_local_wallet()).await;
578        let group = client.create_group(None, Default::default())?;
579        let conn = client.context.db();
580        let put = |n, content| create_and_store_message(&conn, &group.group_id, n, content);
581
582        let malformed = vec![0xff, 0x00, 0x80];
583        let malformed_id = vec![7; 32];
584        create_test_message_raw(
585            &group.group_id,
586            malformed_id.clone(),
587            malformed.clone(),
588            now_ns(),
589            None,
590            None,
591        )
592        .store(&conn)?;
593        let reply = TestContentGenerator::reply_content(
594            &ref_to(7),
595            TextCodec::content_type(),
596            b"replying to bytes".to_vec(),
597        );
598        put(8, reply);
599        put(9, TestContentGenerator::text_content("later"));
600
601        let messages = group.find_messages_v2(&MsgQueryArgs::default())?;
602        assert_eq!(ids(&messages), [7, 8, 9]);
603        let is_kept = |message: &DecodedMessage| {
604            matches!(&message.content, MessageBody::Undecodable(undecodable)
605                if undecodable.raw_bytes == malformed
606                    && undecodable.failure.kind == ContentDecodeFailureKind::MalformedEnvelope)
607        };
608        assert!(is_kept(&messages[0]), "history: {:?}", messages[0].content);
609        assert!(matches!(&messages[2].content, MessageBody::Text(text) if text.content == "later"));
610
611        let MessageBody::Reply(reply) = &messages[1].content else {
612            panic!("expected reply, got {:?}", messages[1].content);
613        };
614        let parent = reply.in_reply_to.as_ref().expect("reply parent");
615        assert_eq!(parent.metadata.id, malformed_id);
616        assert!(is_kept(parent), "parent: {:?}", parent.content);
617
618        let looked_up = client.message_v2(malformed_id.clone())?;
619        assert_eq!(looked_up.metadata.id, malformed_id);
620        assert!(is_kept(&looked_up), "lookup: {:?}", looked_up.content);
621    }
622
623    /// A read with no limit enriches every row, past SQLite's bound-value limit.
624    #[xmtp_common::test(unwrap_try = true)]
625    async fn unlimited_history_read_spans_the_bind_limit() {
626        use xmtp_db::ConnectionExt;
627        use xmtp_db::diesel::RunQueryDsl;
628        use xmtp_db::diesel::sql_types::{BigInt, Binary, Integer};
629        const ROWS: i64 = 33_000;
630
631        let (group, context) = setup_test_group().await;
632        let conn = context.db();
633        let text =
634            xmtp_content_types::encoded_content_to_bytes(TestContentGenerator::text_content("row"));
635        conn.raw_query(|sqlite| {
636            xmtp_db::diesel::sql_query(
637                "WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n WHERE x<?) \
638                 INSERT INTO group_messages (id,group_id,decrypted_message_bytes,sent_at_ns,kind,\
639                 sender_installation_id,sender_inbox_id,delivery_status,content_type,version_major,\
640                 version_minor,authority_id,sequence_id,idempotency_key) \
641                 SELECT CAST(printf('%032d',x) AS BLOB),?,?,x,?,x'01','sender',?,?,1,0,\
642                 'xmtp.org',0,CAST(x AS TEXT) FROM n",
643            )
644            .bind::<BigInt, _>(ROWS)
645            .bind::<Binary, _>(group.group_id.as_ref())
646            .bind::<Binary, _>(&text)
647            .bind::<Integer, _>(GroupMessageKind::Application as i32)
648            .bind::<Integer, _>(DeliveryStatus::Published as i32)
649            .bind::<Integer, _>(DbContentType::Text as i32)
650            .execute(sqlite)
651        })?;
652        let newest = format!("{ROWS:032}").into_bytes();
653        create_test_message(
654            &group.group_id,
655            vec![0xee; 32],
656            TestContentGenerator::reaction_content(
657                &newest.encode_hex(),
658                "👍",
659                ReactionAction::Added,
660            ),
661            ROWS + 1,
662        )
663        .store(&conn)?;
664
665        let ascending: Vec<i64> = (1..=ROWS).collect();
666        let stored = group.find_messages_v2_with_stored(&MsgQueryArgs::default())?;
667        let sent_at: Vec<_> = stored.iter().map(|m| m.stored.sent_at_ns).collect();
668        assert_eq!(sent_at, ascending);
669        assert_eq!(emojis(&stored.last()?.decoded), ["👍"]);
670        let decoded = group.find_messages_v2(&MsgQueryArgs::default())?;
671        let sent_at: Vec<_> = decoded.iter().map(|m| m.metadata.sent_at_ns).collect();
672        assert_eq!(sent_at, ascending);
673        assert_eq!(emojis(decoded.last()?), ["👍"]);
674    }
675}
xmtp_mls/groups/message_list.rs

xmtp_mls/groups/
message_list.rs
