Conversation Sitching

bindings_ffi/src/mls.rs

@@ -592,15 +592,19 @@ pub struct FfiListConversationsOptions {
     pub created_before_ns: Option<i64>,
     pub limit: Option<i64>,
     pub consent_state: Option<FfiConsentState>,
+    pub include_duplicate_dms: bool,
 }
 
 impl From<FfiListConversationsOptions> for GroupQueryArgs {
     fn from(opts: FfiListConversationsOptions) -> GroupQueryArgs {
-        GroupQueryArgs::default()
-            .maybe_created_before_ns(opts.created_before_ns)
-            .maybe_created_after_ns(opts.created_after_ns)
-            .maybe_limit(opts.limit)
-            .maybe_consent_state(opts.consent_state.map(Into::into))
+        GroupQueryArgs {
+            created_before_ns: opts.created_before_ns,
+            created_after_ns: opts.created_after_ns,
+            limit: opts.limit,
+            consent_state: opts.consent_state.map(Into::into),
+            include_duplicate_dms: opts.include_duplicate_dms,
+            ..Default::default()
+        }
     }
 }
 
@@ -1351,15 +1355,15 @@ impl FfiConversation {
 
         let messages: Vec<FfiMessage> = self
             .inner
-            .find_messages(
-                &MsgQueryArgs::default()
-                    .maybe_sent_before_ns(opts.sent_before_ns)
-                    .maybe_sent_after_ns(opts.sent_after_ns)
-                    .maybe_kind(kind)
-                    .maybe_delivery_status(delivery_status)
-                    .maybe_limit(opts.limit)
-                    .maybe_direction(direction),
-            )?
+            .find_messages(&MsgQueryArgs {
+                sent_before_ns: opts.sent_before_ns,
+                sent_after_ns: opts.sent_after_ns,
+                limit: opts.limit,
+                kind,
+                delivery_status,
+                direction,
+                ..Default::default()
+            })?
             .into_iter()
             .map(|msg| msg.into())
             .collect();

bindings_node/src/conversation.rs

@@ -178,9 +178,12 @@ impl Conversation {
       ConversationType::Dm => Some(XmtpGroupMessageKind::Application),
       ConversationType::Sync => None,
     };
-    let opts: MsgQueryArgs = opts.into();
+    let opts = MsgQueryArgs {
+      kind,
+      ..opts.into()
+    };
     let messages: Vec<Message> = group
-      .find_messages(&opts.maybe_kind(kind))
+      .find_messages(&opts)
       .map_err(ErrorWrapper::from)?
       .into_iter()
       .map(|msg| msg.into())

bindings_node/src/message.rs

@@ -82,12 +82,14 @@ impl From<ListMessagesOptions> for MsgQueryArgs {
     let delivery_status = opts.delivery_status.map(Into::into);
     let direction = opts.direction.map(Into::into);
 
-    MsgQueryArgs::default()
-      .maybe_sent_before_ns(opts.sent_before_ns)
-      .maybe_sent_after_ns(opts.sent_after_ns)
-      .maybe_delivery_status(delivery_status)
-      .maybe_limit(opts.limit)
-      .maybe_direction(direction)
+    MsgQueryArgs {
+      sent_before_ns: opts.sent_before_ns,
+      sent_after_ns: opts.sent_after_ns,
+      delivery_status,
+      limit: opts.limit,
+      direction,
+      ..Default::default()
+    }
   }
 }
 

bindings_wasm/src/conversation.rs

@@ -201,9 +201,13 @@ impl Conversation {
       ConversationType::Dm => Some(XmtpGroupMessageKind::Application),
       ConversationType::Sync => None,
     };
-    let opts: MsgQueryArgs = opts.into();
+
+    let opts = MsgQueryArgs {
+      kind,
+      ..opts.into()
+    };
     let messages: Vec<Message> = group
-      .find_messages(&opts.maybe_kind(kind))
+      .find_messages(&opts)
       .map_err(|e| JsError::new(&format!("{e}")))?
       .into_iter()
       .map(Into::into)

bindings_wasm/src/messages.rs

@@ -87,12 +87,14 @@ impl From<ListMessagesOptions> for MsgQueryArgs {
     let delivery_status = opts.delivery_status.map(Into::into);
     let direction = opts.direction.map(Into::into);
 
-    MsgQueryArgs::default()
-      .maybe_sent_before_ns(opts.sent_before_ns)
-      .maybe_sent_after_ns(opts.sent_after_ns)
-      .maybe_delivery_status(delivery_status)
-      .maybe_limit(opts.limit)
-      .maybe_direction(direction)
+    MsgQueryArgs {
+      sent_before_ns: opts.sent_before_ns,
+      sent_after_ns: opts.sent_after_ns,
+      delivery_status,
+      limit: opts.limit,
+      direction,
+      ..Default::default()
+    }
   }
 }
 

examples/cli/cli-client.rs

@@ -455,8 +455,10 @@ async fn main() -> color_eyre::eyre::Result<()> {
             let group = client.get_sync_group(provider.conn_ref())?;
             let group_id_str = hex::encode(group.group_id.clone());
             group.sync().await?;
-            let messages = group
-                .find_messages(&MsgQueryArgs::default().kind(GroupMessageKind::Application))?;
+            let messages = group.find_messages(&MsgQueryArgs {
+                kind: Some(GroupMessageKind::Application),
+                ..Default::default()
+            })?;
             info!(
                 group_id = group_id_str,
                 messages = messages.len(),

xmtp_mls/diesel.toml

@@ -2,7 +2,7 @@
 # see https://diesel.rs/guides/configuring-diesel-cli
 
 [print_schema]
-file = "src/storage/encrypted_store/schema.rs"
+file = "src/storage/encrypted_store/schema_gen.rs"
 
 [migrations_directory]
 dir = "migrations"

xmtp_mls/migrations/2024-12-20-143210_create_conversation_list_view/down.sql

@@ -1 +1 @@
-DROP VIEW IF EXISTS conversation_list;
+DROP VIEW IF EXISTS conversation_list;

xmtp_mls/migrations/2024-12-20-143210_create_conversation_list_view/up.sql

@@ -42,6 +42,6 @@ SELECT
     rm.authority_id
 FROM
     groups g
-        LEFT JOIN ranked_messages rm
-                  ON g.id = rm.group_id AND rm.row_num = 1
-ORDER BY COALESCE(rm.sent_at_ns, g.created_at_ns) DESC;
+    LEFT JOIN ranked_messages rm
+    ON g.id = rm.group_id AND rm.row_num = 1
+ORDER BY COALESCE(rm.sent_at_ns, g.created_at_ns) DESC;

xmtp_mls/migrations/2024-12-20-214747_add_dm_id/down.sql

@@ -0,0 +1,5 @@
+ALTER TABLE groups DROP COLUMN dm_id;
+ALTER TABLE groups DROP COLUMN last_message_ns;
+ALTER TABLE groups ADD COLUMN dm_inbox_id TEXT;
+
+DROP TRIGGER IF EXISTS msg_inserted;

xmtp_mls/migrations/2024-12-20-214747_add_dm_id/up.sql

@@ -0,0 +1,76 @@
+DROP VIEW IF EXISTS conversation_list;
+ALTER TABLE groups ADD COLUMN dm_id TEXT;
+ALTER TABLE groups ADD COLUMN last_message_ns BIGINT;
+
+-- Fill the dm_id column
+UPDATE groups
+SET dm_id = 'dm:' ||
+    LOWER(
+        CASE
+            WHEN LOWER((SELECT inbox_id FROM identity)) < LOWER(dm_inbox_id)
+            THEN (SELECT inbox_id FROM identity) || ':' || dm_inbox_id
+            ELSE dm_inbox_id || ':' || (SELECT inbox_id FROM identity)
+        END
+    )
+WHERE dm_inbox_id IS NOT NULL;
+
+DROP INDEX IF EXISTS idx_dm_target;
+ALTER TABLE groups DROP COLUMN dm_inbox_id;
+
+-- Create a trigger to auto-update group table on insert
+CREATE TRIGGER msg_inserted
+AFTER INSERT ON group_messages
+BEGIN
+  UPDATE groups
+  SET last_message_ns = (strftime('%s', 'now') * 1000000000) + (strftime('%f', 'now') * 1000000)
+  WHERE id = NEW.group_id;
+END;
+
+
+CREATE VIEW conversation_list AS
+WITH ranked_messages AS (
+    SELECT
+        gm.group_id,
+        gm.id AS message_id,
+        gm.decrypted_message_bytes,
+        gm.sent_at_ns,
+        gm.kind AS message_kind,
+        gm.sender_installation_id,
+        gm.sender_inbox_id,
+        gm.delivery_status,
+        gm.content_type,
+        gm.version_major,
+        gm.version_minor,
+        gm.authority_id,
+        ROW_NUMBER() OVER (PARTITION BY gm.group_id ORDER BY gm.sent_at_ns DESC) AS row_num
+    FROM
+        group_messages gm
+    WHERE
+        gm.kind = 1
+)
+SELECT
+    g.id AS id,
+    g.created_at_ns,
+    g.membership_state,
+    g.installations_last_checked,
+    g.added_by_inbox_id,
+    g.welcome_id,
+    g.dm_id,
+    g.rotated_at_ns,
+    g.conversation_type,
+    rm.message_id,
+    rm.decrypted_message_bytes,
+    rm.sent_at_ns,
+    rm.message_kind,
+    rm.sender_installation_id,
+    rm.sender_inbox_id,
+    rm.delivery_status,
+    rm.content_type,
+    rm.version_major,
+    rm.version_minor,
+    rm.authority_id
+FROM
+    groups g
+    LEFT JOIN ranked_messages rm
+    ON g.id = rm.group_id AND rm.row_num = 1
+ORDER BY COALESCE(rm.sent_at_ns, g.created_at_ns) DESC;

xmtp_mls/src/client.rs

@@ -37,8 +37,8 @@ use crate::groups::ConversationListItem;
 use crate::{
     api::ApiClientWrapper,
     groups::{
-        device_sync::preference_sync::UserPreferenceUpdate, group_permissions::PolicySet,
-        GroupError, GroupMetadataOptions, MlsGroup,
+        device_sync::preference_sync::UserPreferenceUpdate, group_metadata::DmMembers,
+        group_permissions::PolicySet, GroupError, GroupMetadataOptions, MlsGroup,
     },
     identity::{parse_credential, Identity, IdentityError},
     identity_updates::{load_identity_updates, IdentityUpdateError},
@@ -637,8 +637,12 @@ where
         target_inbox_id: String,
     ) -> Result<MlsGroup<Self>, ClientError> {
         let conn = self.store().conn()?;
+
         let group = conn
-            .find_dm_group(&target_inbox_id)?
+            .find_dm_group(&DmMembers {
+                member_one_inbox_id: self.inbox_id(),
+                member_two_inbox_id: &target_inbox_id,
+            })?
             .ok_or(NotFound::DmByInbox(target_inbox_id))?;
         Ok(MlsGroup::new(self.clone(), group.id, group.created_at_ns))
     }
@@ -943,6 +947,7 @@ where
         let query_args = GroupQueryArgs {
             consent_state,
             include_sync_groups: true,
+            include_duplicate_dms: true,
             ..GroupQueryArgs::default()
         };
         let groups = provider

xmtp_mls/src/groups/device_sync.rs

@@ -496,7 +496,10 @@ where
 
         let messages = provider.conn_ref().get_group_messages(
             &sync_group.group_id,
-            &MsgQueryArgs::default().kind(GroupMessageKind::Application),
+            &MsgQueryArgs {
+                kind: Some(GroupMessageKind::Application),
+                ..Default::default()
+            },
         )?;
 
         for msg in messages.into_iter().rev() {
@@ -529,8 +532,10 @@ where
         let sync_group = self.get_sync_group(provider.conn_ref())?;
         sync_group.sync_with_conn(provider).await?;
 
-        let messages = sync_group
-            .find_messages(&MsgQueryArgs::default().kind(GroupMessageKind::Application))?;
+        let messages = sync_group.find_messages(&MsgQueryArgs {
+            kind: Some(GroupMessageKind::Application),
+            ..Default::default()
+        })?;
 
         for msg in messages.into_iter().rev() {
             let Ok(msg_content) =