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

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::ConversationMessageDisappearingSettings as XmtpConversationMessageDisappearingSettings;
 
 use napi_derive::napi;
 
@@ -38,6 +38,28 @@ pub struct GroupMetadata {
   inner: XmtpGroupMetadata,
 }
 
+#[napi(object)]
+#[derive(Clone)]
+pub struct ConversationMessageDisappearingSettings {
+  #[napi]
+  pub inner: XmtpConversationMessageDisappearingSettings,
+}
+
+#[napi]
+impl ConversationMessageDisappearingSettings {
+  #[napi]
+  pub fn new(from_ns: i64, in_ns: i64) -> Self {
+    let inner = XmtpConversationMessageDisappearingSettings { from_ns, in_ns };
+    Self { inner }
+  }
+}
+
+impl From<ConversationMessageDisappearingSettings> for XmtpConversationMessageDisappearingSettings {
+  fn from(value: ConversationMessageDisappearingSettings) -> Self {
+    XmtpConversationMessageDisappearingSettings::new(value.inner.from_ns, value.inner.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::ConversationMessageDisappearingSettings;
 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<ConversationMessageDisappearingSettings>,
 }
 
 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_wasm/src/conversation.rs

@@ -18,12 +18,19 @@ 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::ConversationMessageDisappearingSettings as XmtpGroupMessageDisappearingSettings;
 
 #[wasm_bindgen]
 pub struct GroupMetadata {
   inner: XmtpGroupMetadata,
 }
 
+#[wasm_bindgen]
+#[derive(Clone)]
+pub struct ConversationMessageDisappearingSettings {
+  inner: XmtpGroupMessageDisappearingSettings,
+}
+
 #[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::ConversationMessageDisappearingSettings;
 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<ConversationMessageDisappearingSettings>,
 }
 
 #[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<ConversationMessageDisappearingSettings>,
   ) -> 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,7 @@ 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: None, // todo: fix mapping,
     }
   }
 }
@@ -218,8 +214,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> {

xmtp_mls/src/groups/disappearing_messages.rs

@@ -0,0 +1,85 @@
+use crate::client::ClientError;
+use crate::storage::StorageError;
+use crate::Client;
+use std::time::Duration;
+use thiserror::Error;
+use tokio::sync::OnceCell;
+use xmtp_id::scw_verifier::SmartContractSignatureVerifier;
+use xmtp_proto::api_client::trait_impls::XmtpApi;
+
+#[derive(Debug, Error)]
+pub enum DisappearingMessagesCleanerError {
+    #[error("storage error: {0}")]
+    Storage(#[from] StorageError),
+    #[error("client error: {0}")]
+    Client(#[from] ClientError),
+}
+
+pub struct DisappearingMessagesCleanerWorker<ApiClient, V> {
+    client: Client<ApiClient, V>,
+    init: OnceCell<()>,
+}
+impl<ApiClient, V> DisappearingMessagesCleanerWorker<ApiClient, V>
+where
+    ApiClient: XmtpApi + Send + Sync + 'static,
+    V: SmartContractSignatureVerifier + Send + Sync + 'static,
+{
+    pub fn new(client: Client<ApiClient, V>) -> Self {
+        Self {
+            client,
+            init: OnceCell::new(),
+        }
+    }
+    pub(crate) fn spawn_worker(mut self) {
+        crate::spawn(None, async move {
+            let inbox_id = self.client.inbox_id().to_string();
+            let installation_id = hex::encode(self.client.installation_public_key());
+            while let Err(err) = self.run().await {
+                tracing::info!("Running worker..");
+                match err {
+                    DisappearingMessagesCleanerError::Client(ClientError::Storage(
+                        StorageError::PoolNeedsConnection,
+                    )) => {
+                        tracing::warn!(
+                            inbox_id,
+                            installation_id,
+                            "Pool disconnected. task will restart on reconnect"
+                        );
+                        break;
+                    }
+                    _ => {
+                        tracing::error!(inbox_id, installation_id, "sync worker error {err}");
+                        // Wait 2 seconds before restarting.
+                        xmtp_common::time::sleep(Duration::from_secs(2)).await;
+                    }
+                }
+            }
+        });
+    }
+}
+
+impl<ApiClient, V> DisappearingMessagesCleanerWorker<ApiClient, V>
+where
+    ApiClient: XmtpApi + Send + Sync + 'static,
+    V: SmartContractSignatureVerifier + Send + Sync + 'static,
+{
+    /// Iterate on the list of groups and delete expired messages
+    async fn delete_expired_messages(&mut self) -> Result<(), DisappearingMessagesCleanerError> {
+        let provider = self.client.mls_provider()?;
+        match provider.conn_ref().delete_expired_messages() {
+            Ok(deleted_count) => {
+                tracing::info!("Successfully deleted {} expired messages", deleted_count);
+            }
+            Err(e) => {
+                tracing::error!("Failed to delete expired messages, error: {:?}", e);
+            }
+        }
+        Ok(())
+    }
+    async fn run(&mut self) -> Result<(), DisappearingMessagesCleanerError> {
+        if let Err(err) = self.delete_expired_messages().await {
+            tracing::error!("Error during deletion of expired messages: {:?}", err);
+        }
+        Ok(())
+    }
+}