feat(groups): add a worker to delete expired messages (#1503)

bindings_node/src/conversation.rs

@@ -28,8 +28,8 @@ use crate::{
   streams::StreamCloser,
   ErrorWrapper,
 };
-
 use prost::Message as ProstMessage;
+use xmtp_mls::groups::group_mutable_metadata::MessageDisappearingSettings as XmtpConversationMessageDisappearingSettings;
 
 use napi_derive::napi;
 
@@ -38,6 +38,33 @@ pub struct GroupMetadata {
   inner: XmtpGroupMetadata,
 }
 
+/// Settings for disappearing messages in a conversation.
+///
+/// # Fields
+///
+/// * `from_ns` - The timestamp (in nanoseconds) from when messages should be tracked for deletion.
+/// * `in_ns` - The duration (in nanoseconds) after which tracked messages will be deleted.
+#[napi(object)]
+#[derive(Clone)]
+pub struct MessageDisappearingSettings {
+  pub from_ns: i64,
+  pub in_ns: i64,
+}
+
+#[napi]
+impl MessageDisappearingSettings {
+  #[napi]
+  pub fn new(from_ns: i64, in_ns: i64) -> Self {
+    Self { from_ns, in_ns }
+  }
+}
+
+impl From<MessageDisappearingSettings> for XmtpConversationMessageDisappearingSettings {
+  fn from(value: MessageDisappearingSettings) -> Self {
+    XmtpConversationMessageDisappearingSettings::new(value.from_ns, value.in_ns)
+  }
+}
+
 #[napi]
 impl GroupMetadata {
   #[napi]

bindings_node/src/conversations.rs

@@ -12,6 +12,7 @@ use xmtp_mls::storage::group::ConversationType as XmtpConversationType;
 use xmtp_mls::storage::group::GroupMembershipState as XmtpGroupMembershipState;
 use xmtp_mls::storage::group::GroupQueryArgs;
 
+use crate::conversation::MessageDisappearingSettings;
 use crate::message::Message;
 use crate::permissions::{GroupPermissionsOptions, PermissionPolicySet};
 use crate::ErrorWrapper;
@@ -122,8 +123,7 @@ pub struct CreateGroupOptions {
   pub group_description: Option<String>,
   pub group_pinned_frame_url: Option<String>,
   pub custom_permission_policy_set: Option<PermissionPolicySet>,
-  pub message_expiration_from_ms: Option<i64>,
-  pub message_expiration_ms: Option<i64>,
+  pub message_disappearing_settings: Option<MessageDisappearingSettings>,
 }
 
 impl CreateGroupOptions {
@@ -133,8 +133,9 @@ impl CreateGroupOptions {
       image_url_square: self.group_image_url_square,
       description: self.group_description,
       pinned_frame_url: self.group_pinned_frame_url,
-      message_expiration_from_ms: self.message_expiration_from_ms,
-      message_expiration_ms: self.message_expiration_ms,
+      message_disappearing_settings: self
+        .message_disappearing_settings
+        .map(|settings| settings.into()),
     }
   }
 }
@@ -163,8 +164,7 @@ impl Conversations {
       group_description: None,
       group_pinned_frame_url: None,
       custom_permission_policy_set: None,
-      message_expiration_from_ms: None,
-      message_expiration_ms: None,
+      message_disappearing_settings: None,
     });
 
     if let Some(GroupPermissionsOptions::CustomPolicy) = options.permissions {

bindings_node/src/permissions.rs

@@ -161,7 +161,7 @@ pub struct PermissionPolicySet {
   pub update_group_description_policy: PermissionPolicy,
   pub update_group_image_url_square_policy: PermissionPolicy,
   pub update_group_pinned_frame_url_policy: PermissionPolicy,
-  pub update_message_expiration_ms_policy: PermissionPolicy,
+  pub update_message_disappearing_policy: PermissionPolicy,
 }
 
 impl From<PreconfiguredPolicies> for GroupPermissionsOptions {
@@ -216,8 +216,8 @@ impl GroupPermissions {
       update_group_pinned_frame_url_policy: get_policy(
         XmtpMetadataField::GroupPinnedFrameUrl.as_str(),
       ),
-      update_message_expiration_ms_policy: get_policy(
-        XmtpMetadataField::MessageExpirationMillis.as_str(),
+      update_message_disappearing_policy: get_policy(
+        XmtpMetadataField::MessageDisappearInNS.as_str(),
       ),
     })
   }
@@ -246,8 +246,8 @@ impl TryFrom<PermissionPolicySet> for PolicySet {
       policy_set.update_group_pinned_frame_url_policy.try_into()?,
     );
     metadata_permissions_map.insert(
-      XmtpMetadataField::MessageExpirationMillis.to_string(),
-      policy_set.update_message_expiration_ms_policy.try_into()?,
+      XmtpMetadataField::MessageDisappearInNS.to_string(),
+      policy_set.update_message_disappearing_policy.try_into()?,
     );
 
     Ok(PolicySet {

bindings_node/test/Conversations.test.ts

@@ -52,7 +52,7 @@ describe('Conversations', () => {
       updateGroupDescriptionPolicy: 0,
       updateGroupImageUrlSquarePolicy: 0,
       updateGroupPinnedFrameUrlPolicy: 0,
-      updateMessageExpirationMsPolicy: 2,
+      updateMessageDisappearingPolicy: 2,
     })
     expect(group.addedByInboxId()).toBe(client1.inboxId())
     expect((await group.findMessages()).length).toBe(1)
@@ -104,7 +104,7 @@ describe('Conversations', () => {
           updateGroupDescriptionPolicy: 1,
           updateGroupImageUrlSquarePolicy: 0,
           updateGroupPinnedFrameUrlPolicy: 3,
-          updateMessageExpirationMsPolicy: 2,
+          updateMessageDisappearingPolicy: 2,
         },
       })
     expect(group).toBeDefined()
@@ -120,7 +120,7 @@ describe('Conversations', () => {
       updateGroupDescriptionPolicy: 1,
       updateGroupImageUrlSquarePolicy: 0,
       updateGroupPinnedFrameUrlPolicy: 3,
-      updateMessageExpirationMsPolicy: 2,
+      updateMessageDisappearingPolicy: 2,
     })
   })
 
@@ -142,7 +142,7 @@ describe('Conversations', () => {
       updateGroupDescriptionPolicy: 0,
       updateGroupImageUrlSquarePolicy: 0,
       updateGroupPinnedFrameUrlPolicy: 0,
-      updateMessageExpirationMsPolicy: 2,
+      updateMessageDisappearingPolicy: 2,
     })
 
     await group.updatePermissionPolicy(
@@ -159,7 +159,7 @@ describe('Conversations', () => {
       updateGroupDescriptionPolicy: 0,
       updateGroupImageUrlSquarePolicy: 0,
       updateGroupPinnedFrameUrlPolicy: 0,
-      updateMessageExpirationMsPolicy: 2,
+      updateMessageDisappearingPolicy: 2,
     })
 
     await group.updatePermissionPolicy(
@@ -177,7 +177,7 @@ describe('Conversations', () => {
       updateGroupDescriptionPolicy: 0,
       updateGroupImageUrlSquarePolicy: 0,
       updateGroupPinnedFrameUrlPolicy: 0,
-      updateMessageExpirationMsPolicy: 2,
+      updateMessageDisappearingPolicy: 2,
     })
   })
 
@@ -204,7 +204,7 @@ describe('Conversations', () => {
       updateGroupImageUrlSquarePolicy: 0,
       updateGroupNamePolicy: 0,
       updateGroupPinnedFrameUrlPolicy: 0,
-      updateMessageExpirationMsPolicy: 0,
+      updateMessageDisappearingPolicy: 0,
     })
     expect(group.addedByInboxId()).toBe(client1.inboxId())
     expect((await group.findMessages()).length).toBe(0)
@@ -342,7 +342,7 @@ describe('Conversations', () => {
       updateGroupDescriptionPolicy: 2,
       updateGroupImageUrlSquarePolicy: 2,
       updateGroupPinnedFrameUrlPolicy: 2,
-      updateMessageExpirationMsPolicy: 2,
+      updateMessageDisappearingPolicy: 2,
     })
 
     const groupWithDescription = await client1

bindings_wasm/src/conversation.rs

@@ -18,12 +18,29 @@ use xmtp_mls::storage::group_message::{GroupMessageKind as XmtpGroupMessageKind,
 use xmtp_proto::xmtp::mls::message_contents::EncodedContent as XmtpEncodedContent;
 
 use prost::Message as ProstMessage;
+use xmtp_mls::groups::group_mutable_metadata::MessageDisappearingSettings as XmtpMessageDisappearingSettings;
 
 #[wasm_bindgen]
 pub struct GroupMetadata {
   inner: XmtpGroupMetadata,
 }
 
+#[wasm_bindgen]
+#[derive(Clone)]
+pub struct MessageDisappearingSettings {
+  #[allow(dead_code)]
+  inner: XmtpMessageDisappearingSettings,
+}
+
+impl From<MessageDisappearingSettings> for XmtpMessageDisappearingSettings {
+  fn from(value: MessageDisappearingSettings) -> Self {
+    Self {
+      from_ns: value.inner.from_ns,
+      in_ns: value.inner.in_ns,
+    }
+  }
+}
+
 #[wasm_bindgen]
 impl GroupMetadata {
   #[wasm_bindgen(js_name = creatorInboxId)]

bindings_wasm/src/conversations.rs

@@ -7,6 +7,7 @@ use xmtp_mls::storage::group::ConversationType as XmtpConversationType;
 use xmtp_mls::storage::group::GroupMembershipState as XmtpGroupMembershipState;
 use xmtp_mls::storage::group::GroupQueryArgs;
 
+use crate::conversation::MessageDisappearingSettings;
 use crate::messages::Message;
 use crate::permissions::{GroupPermissionsOptions, PermissionPolicySet};
 use crate::{client::RustXmtpClient, conversation::Conversation};
@@ -130,10 +131,8 @@ pub struct CreateGroupOptions {
   pub group_pinned_frame_url: Option<String>,
   #[wasm_bindgen(js_name = customPermissionPolicySet)]
   pub custom_permission_policy_set: Option<PermissionPolicySet>,
-  #[wasm_bindgen(js_name = messageExpirationFromMillis)]
-  pub message_expiration_from_ms: Option<i64>,
-  #[wasm_bindgen(js_name = messageExpirationMillis)]
-  pub message_expiration_ms: Option<i64>,
+  #[wasm_bindgen(js_name = messageDisappearingSettings)]
+  pub message_disappearing_settings: Option<MessageDisappearingSettings>,
 }
 
 #[wasm_bindgen]
@@ -147,8 +146,7 @@ impl CreateGroupOptions {
     group_description: Option<String>,
     group_pinned_frame_url: Option<String>,
     custom_permission_policy_set: Option<PermissionPolicySet>,
-    message_expiration_from_ms: Option<i64>,
-    message_expiration_ms: Option<i64>,
+    message_disappearing_settings: Option<MessageDisappearingSettings>,
   ) -> Self {
     Self {
       permissions,
@@ -157,8 +155,7 @@ impl CreateGroupOptions {
       group_description,
       group_pinned_frame_url,
       custom_permission_policy_set,
-      message_expiration_from_ms,
-      message_expiration_ms,
+      message_disappearing_settings,
     }
   }
 }
@@ -170,8 +167,9 @@ impl CreateGroupOptions {
       image_url_square: self.group_image_url_square,
       description: self.group_description,
       pinned_frame_url: self.group_pinned_frame_url,
-      message_expiration_from_ms: self.message_expiration_from_ms,
-      message_expiration_ms: self.message_expiration_ms,
+      message_disappearing_settings: self
+        .message_disappearing_settings
+        .map(|settings| settings.into()),
     }
   }
 }
@@ -218,8 +216,7 @@ impl Conversations {
       group_description: None,
       group_pinned_frame_url: None,
       custom_permission_policy_set: None,
-      message_expiration_from_ms: None,
-      message_expiration_ms: None,
+      message_disappearing_settings: None,
     });
 
     if let Some(GroupPermissionsOptions::CustomPolicy) = options.permissions {

bindings_wasm/src/permissions.rs

@@ -170,8 +170,8 @@ pub struct PermissionPolicySet {
   pub update_group_image_url_square_policy: PermissionPolicy,
   #[wasm_bindgen(js_name = updateGroupPinnedFrameUrlPolicy)]
   pub update_group_pinned_frame_url_policy: PermissionPolicy,
-  #[wasm_bindgen(js_name = updateMessageExpirationPolicy)]
-  pub update_message_expiration_ms_policy: PermissionPolicy,
+  #[wasm_bindgen(js_name = updateMessageDisappearingPolicy)]
+  pub update_message_disappearing_policy: PermissionPolicy,
 }
 
 #[wasm_bindgen]
@@ -187,7 +187,7 @@ impl PermissionPolicySet {
     update_group_description_policy: PermissionPolicy,
     update_group_image_url_square_policy: PermissionPolicy,
     update_group_pinned_frame_url_policy: PermissionPolicy,
-    update_message_expiration_ms_policy: PermissionPolicy,
+    update_message_disappearing_policy: PermissionPolicy,
   ) -> Self {
     Self {
       add_member_policy,
@@ -198,7 +198,7 @@ impl PermissionPolicySet {
       update_group_description_policy,
       update_group_image_url_square_policy,
       update_group_pinned_frame_url_policy,
-      update_message_expiration_ms_policy,
+      update_message_disappearing_policy,
     }
   }
 }
@@ -257,8 +257,8 @@ impl GroupPermissions {
       update_group_pinned_frame_url_policy: get_policy(
         XmtpMetadataField::GroupPinnedFrameUrl.as_str(),
       ),
-      update_message_expiration_ms_policy: get_policy(
-        XmtpMetadataField::MessageExpirationMillis.as_str(),
+      update_message_disappearing_policy: get_policy(
+        XmtpMetadataField::MessageDisappearInNS.as_str(),
       ),
     })
   }
@@ -285,8 +285,8 @@ impl TryFrom<PermissionPolicySet> for PolicySet {
       policy_set.update_group_pinned_frame_url_policy.try_into()?,
     );
     metadata_permissions_map.insert(
-      XmtpMetadataField::MessageExpirationMillis.to_string(),
-      policy_set.update_message_expiration_ms_policy.try_into()?,
+      XmtpMetadataField::MessageDisappearInNS.to_string(),
+      policy_set.update_message_disappearing_policy.try_into()?,
     );
 
     Ok(PolicySet {
@@ -306,6 +306,8 @@ pub enum MetadataField {
   Description,
   ImageUrlSquare,
   PinnedFrameUrl,
+  MessageExpirationFromMS,
+  MessageExpirationMS,
 }
 
 impl From<&MetadataField> for XmtpMetadataField {
@@ -315,6 +317,8 @@ impl From<&MetadataField> for XmtpMetadataField {
       MetadataField::Description => XmtpMetadataField::Description,
       MetadataField::ImageUrlSquare => XmtpMetadataField::GroupImageUrlSquare,
       MetadataField::PinnedFrameUrl => XmtpMetadataField::GroupPinnedFrameUrl,
+      MetadataField::MessageExpirationFromMS => XmtpMetadataField::MessageDisappearFromNS,
+      MetadataField::MessageExpirationMS => XmtpMetadataField::MessageDisappearInNS,
     }
   }
 }

xmtp_mls/migrations/2025-01-16-143131_add_message_expiration_to_groups/down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE GROUPS DROP COLUMN message_disappear_from_ns;
+ALTER TABLE GROUPS DROP COLUMN message_disappear_in_ns;

xmtp_mls/migrations/2025-01-16-143131_add_message_expiration_to_groups/up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE GROUPS ADD COLUMN message_disappear_from_ns  BIGINT;
+ALTER TABLE GROUPS ADD COLUMN message_disappear_in_ns BIGINT;

xmtp_mls/src/groups/device_sync.rs

@@ -1,4 +1,5 @@
 use super::{GroupError, MlsGroup};
+use crate::groups::disappearing_messages::DisappearingMessagesCleanerWorker;
 #[cfg(any(test, feature = "test-utils"))]
 pub use crate::utils::WorkerHandle;
 use crate::{
@@ -138,6 +139,19 @@ where
         self.set_sync_worker_handle(worker.handle.clone());
         worker.spawn_worker();
     }
+
+    #[instrument(level = "trace", skip_all)]
+    pub fn start_disappearing_messages_cleaner_worker(&self) {
+        let client = self.clone();
+        tracing::trace!(
+            inbox_id = client.inbox_id(),
+            installation_id = hex::encode(client.installation_public_key()),
+            "starting expired messages cleaner worker"
+        );
+
+        let worker = DisappearingMessagesCleanerWorker::new(client);
+        worker.spawn_worker();
+    }
 }
 
 pub struct SyncWorker<ApiClient, V> {