xmtp_mls/groups/

membership.rs

//! Adding, removing, and re-adding members.

use super::*;
use prost::Message;
use xmtp_mls_group_state::app_data::{
    components::tls_map_components::GroupMembershipComponent, typed_facade::MlsGroupAppData,
};
use xmtp_proto::ShortHex;
use xmtp_proto::xmtp::mls::message_contents::{GroupMembershipEntry, group_membership_entry};

impl<Context> MlsGroup<Context>
where
    Context: XmtpSharedContext,
{
    ///
    /// Add members to the group by account address
    ///
    /// An account resolves to an inbox only when the newest state of the inbox
    /// the backend names holds it. The add fails with
    /// [`GroupError::AddressNotFound`], before any commit, when an account has
    /// no such inbox.
    ///
    /// If any existing members have new installations that have not been added or removed, the
    /// group membership will be updated to include those changes as well.
    /// # Returns
    /// - `Ok(UpdateGroupMembershipResult)`: Contains details about the membership changes, including:
    ///   - `added_members`: list of added installations
    ///   - `removed_members`: A list of installations that were removed.
    ///   - `members_with_errors`: A list of members that encountered errors during the update.
    /// - `Err(GroupError)`: If the operation fails due to an error.
    #[tracing::instrument(level = "trace", skip_all)]
    pub async fn add_members_by_identity(
        &self,
        account_identifiers: &[Identifier],
    ) -> Result<UpdateGroupMembershipResult, GroupError> {
        // The member limit is enforced in `add_members`, the one method every
        // entry point reaches.
        self.add_members_by_targets(&crate::inbox_handles::account_targets(account_identifiers))
            .await
    }

    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", skip_all, fields(inbox_id = %self.context.inbox_id(), inbox_ids = ?inbox_ids.as_ref().iter().map(|i| i.as_ref()).collect::<Vec<_>>())))]
    #[cfg_attr(
        not(any(test, feature = "test-utils")),
        tracing::instrument(level = "trace", skip_all)
    )]
    pub async fn add_members<S: AsIdRef>(
        &self,
        inbox_ids: impl AsRef<[S]>,
    ) -> Result<UpdateGroupMembershipResult, GroupError> {
        self.ensure_not_paused().await?;

        let ids = inbox_ids
            .as_ref()
            .iter()
            .map(AsIdRef::as_ref)
            .collect::<Vec<&str>>();
        let intent_data = self
            .get_membership_update_intent(ids.as_slice(), &[])
            .await?;

        // TODO:nm this isn't the best test for whether the request is valid
        // If some existing group member has an update, this will return an intent with changes
        // when we really should return an error
        let mut result = UpdateGroupMembershipResult::from(intent_data.clone());

        if intent_data.is_empty() {
            tracing::warn!("Member already added");
            return Ok(result);
        }

        // The deployment sets the ceiling, checked here — the one
        // method both entry points reach — before the commit is built and
        // before anything is published. The count comes from the group's own
        // membership extension unioned with the inboxes this commit would add,
        // not from `members()`: that helper skips an inbox still at sequence id
        // zero, which is exactly where a just-created group's creator sits.
        let max_members = self
            .context
            .server_configuration()
            .configuration()
            .mls
            .max_group_members;
        let existing = self.with_group_snapshot(|group| {
            Ok(xmtp_mls_validation::commit::extract_group_membership(
                group.extensions(),
            )?)
        })?;
        let resulting: HashSet<&str> = existing
            .inbox_ids()
            .into_iter()
            .chain(intent_data.membership_updates.keys().map(String::as_str))
            .collect();
        if resulting.len() > max_members {
            return Err(GroupError::UserLimitExceeded);
        }

        let intent = QueueIntent::update_group_membership()
            .data(intent_data)
            .queue(self)?;

        self.sync_until_intent_resolved(intent.id).await?;
        let epoch = self.epoch().await?;

        // Publication can discover failures after the request was built,
        // including additions suppressed to fit an envelope. Read the final
        // entries for this request so split-child failures reach its caller.
        result.failed_installations = self.with_group_snapshot(|group| {
            let entries = MlsGroupAppData::new(group.extensions())
                .get::<GroupMembershipComponent>()?
                .ok_or(GroupError::InvalidGroupMembership)?;
            let mut failed = Vec::new();
            for (inbox, bytes) in entries.iter() {
                if !result.added_members.contains_key(&inbox.to_hex()) {
                    continue;
                }
                let entry = GroupMembershipEntry::decode(bytes.as_slice())?;
                let Some(group_membership_entry::Version::V1(entry)) = entry.version else {
                    return Err(GroupError::InvalidGroupMembership);
                };
                failed.extend(entry.failed_installations);
            }
            failed.sort_unstable();
            failed.dedup();
            Ok(failed)
        })?;

        log_event!(
            Event::AddedMembers,
            self.context.installation_id(),
            group_id = self.group_id,
            members = ?ids,
            epoch
        );

        Ok(result)
    }

    /// Add the DM peer when this DM's `GROUP_MEMBERSHIP` lacks it, before a
    /// message is queued in the DM.
    ///
    /// A creation that stops between storing the DM and adding the peer
    /// leaves a DM that holds only its creator. A DM peer can never leave or
    /// be removed, so a missing peer always means such an interruption.
    /// Await this before queueing a message in a DM: intents publish in queue
    /// order, so a message queued before the add is encrypted at an epoch
    /// the peer never joins. Does nothing for a group that is not a DM, an
    /// inactive DM (such as a restored placeholder), or a DM whose pair does
    /// not include this inbox.
    ///
    /// Returns Ok when the add is published with a receipt but its processing
    /// is not confirmed when the wait ends. That outcome belongs to the add,
    /// not to the caller's message, which is not queued yet. The backend holds
    /// the add, and nothing is prepared behind a published state change until
    /// its echo is applied (SEND-022), so a message queued next is built at an
    /// epoch that holds the peer. Every other failure of the add, including a
    /// wait that ends before the add has a receipt, is returned.
    // implements: DMS-020
    pub async fn ensure_dm_peer(&self) -> Result<(), GroupError> {
        match self.add_missing_dm_peer().await {
            Err(error @ GroupError::PublishedButUnconfirmed { .. }) => {
                tracing::warn!(
                    group_id = self.group_id.short_hex(),
                    ?error,
                    "the DM peer add is published but unconfirmed; later intents wait behind it"
                );
                Ok(())
            }
            result => result,
        }
    }

    /// Add the DM peer when this DM's `GROUP_MEMBERSHIP` lacks it, and return
    /// every failure of the add, as a new DM's creation does.
    // implements: DMS-020
    pub(crate) async fn add_missing_dm_peer(&self) -> Result<(), GroupError> {
        let Some(peer) = self.missing_dm_peer()? else {
            return Ok(());
        };
        tracing::info!(
            group_id = self.group_id.short_hex(),
            "adding the missing peer of an interrupted DM creation"
        );
        self.add_members(&[peer]).await?;
        // An add can resolve as a no-op, such as one that would remove this
        // installation's own leaf after its inbox revoked it.
        if self.missing_dm_peer()?.is_some() {
            return Err(GroupError::InvalidGroupMembership);
        }
        Ok(())
    }

    /// The peer of this DM when its `GROUP_MEMBERSHIP` lacks it. None for a
    /// group that is not an active DM of this inbox.
    fn missing_dm_peer(&self) -> Result<Option<String>, GroupError> {
        if self.conversation_type != ConversationType::Dm || !self.is_active()? {
            return Ok(None);
        }
        // Read the membership extension, not `members()`: that helper skips
        // an inbox at sequence id zero, where a joiner sees the creator.
        self.with_group_snapshot(|group| {
            let Some(pair) = extract_group_metadata(group.extensions())
                .map_err(MetadataPermissionsError::from)?
                .dm_members
            else {
                return Ok(None);
            };
            let own = self.context.inbox_id();
            let peer = if pair.member_one_inbox_id == own {
                pair.member_two_inbox_id
            } else if pair.member_two_inbox_id == own {
                pair.member_one_inbox_id
            } else {
                return Ok(None);
            };
            let membership =
                xmtp_mls_validation::commit::extract_group_membership(group.extensions())?;
            Ok::<_, GroupError>(membership.get(&peer).is_none().then_some(peer))
        })
    }

    /// Removes members from the group by their account addresses. An account
    /// resolves to an inbox only when the newest state of the inbox the
    /// backend names holds it; an account with no such inbox is skipped.
    ///
    /// # Arguments
    /// * `client` - The XMTP client.
    /// * `account_addresses_to_remove` - A vector of account addresses to remove from the group.
    ///
    /// # Returns
    /// A `Result` indicating success or failure of the operation.
    pub async fn remove_members_by_identity(
        &self,
        account_addresses_to_remove: &[Identifier],
    ) -> Result<(), GroupError> {
        self.remove_members_by_targets(&crate::inbox_handles::account_targets(
            account_addresses_to_remove,
        ))
        .await
    }

    /// Removes members from the group by their inbox IDs.
    ///
    /// # Arguments
    /// * `client` - The XMTP client.
    /// * `inbox_ids` - A vector of inbox IDs to remove from the group.
    ///
    /// # Returns
    /// A `Result` indicating success or failure of the operation.
    #[cfg_attr(any(test, feature = "test-utils"), tracing::instrument(level = "info", skip_all, fields(inbox_id = %self.context.inbox_id(), inbox_ids = ?inbox_ids)))]
    #[cfg_attr(
        not(any(test, feature = "test-utils")),
        tracing::instrument(level = "trace", skip_all)
    )]
    pub async fn remove_members(&self, inbox_ids: &[InboxIdRef<'_>]) -> Result<(), GroupError> {
        self.remove_inboxes(inbox_ids, false).await
    }

    async fn remove_inboxes(&self, ids: &[InboxIdRef<'_>], leave: bool) -> Result<(), GroupError> {
        self.ensure_not_paused().await?;
        let mut intent_data = self.get_membership_update_intent(&[], ids).await?;
        if leave {
            intent_data =
                intent_data.with_leave_removals(ids.iter().map(|s| s.to_string()).collect());
        }
        let intent = QueueIntent::update_group_membership()
            .data(intent_data)
            .queue(self)?;

        let _ = self.sync_until_intent_resolved(intent.id).await?;

        Ok(())
    }

    /// Removes and readds installations from the MLS tree.
    ///
    /// The installation list should be validated beforehand - invalid installations
    /// will simply be omitted at the time that the intent's publish data is computed.
    ///
    /// # Arguments
    /// * `installations` - A vector of installations to readd.
    ///
    /// # Returns
    /// A `Result` indicating success or failure of the operation.
    #[allow(dead_code)]
    pub(crate) async fn readd_installations(
        &self,
        installations: Vec<Vec<u8>>,
    ) -> Result<(), GroupError> {
        self.ensure_not_paused().await?;

        let readd_min_version =
            LibXMTPVersion::parse(xmtp_configuration::MIN_RECOVERY_REQUEST_VERSION)
                .map_err(super::validated_commit::CommitValidationError::from)?;
        let metadata = self.live_mutable_metadata()?;
        let group_version = metadata
            .attributes()
            .get(MetadataField::MinimumSupportedProtocolVersion.as_str());
        let group_min_version =
            LibXMTPVersion::parse(group_version.unwrap_or(&"0.0.0".to_string()))
                .map_err(super::validated_commit::CommitValidationError::from)?;

        if readd_min_version > group_min_version {
            self.update_group_min_version(xmtp_configuration::MIN_RECOVERY_REQUEST_VERSION)
                .await?;
        }

        let intent_data: Vec<u8> = ReaddInstallationsIntentData::new(installations.clone()).into();
        let intent = QueueIntent::readd_installations()
            .data(intent_data)
            .queue(self)?;

        let _ = self.sync_until_intent_resolved(intent.id).await?;

        Ok(())
    }

    /// Process this group's pending self-remove requests end-to-end: remove the
    /// members still in the group that requested removal, then clean up stale
    /// pending-remove rows. Idempotent and a no-op when this client is not a
    /// super-admin, so it is safe to call from both the inline message-processing
    /// fast-path and the durable `TaskRunner` retry path.
    pub(crate) async fn process_pending_self_removals(&self) -> Result<(), GroupError> {
        // Both helpers early-return on an empty pending list; cleanup owns the
        // flag-clear. Keeping the empty-check inside cleanup (rather than a
        // separate clear here) avoids racing a concurrent LeaveRequest insert and
        // wrongly clearing the flag.
        self.remove_members_pending_removal().await?;
        self.cleanup_pending_removal_list().await?;
        Ok(())
    }

    /// Removes all members from the group who are currently in the pending removal list.
    ///
    /// Only admins and super admins can call this function. Validates permissions, filters
    /// out invalid removal requests and performs batch removal of valid pending members.
    ///
    /// # Returns
    /// * `Ok(())` - All valid pending members were successfully removed
    /// * `Err(GroupError)` - Failed to retrieve metadata, validate permissions or execute removals
    pub async fn remove_members_pending_removal(&self) -> Result<(), GroupError> {
        let pending_removal_list = self.pending_remove_list()?;

        if pending_removal_list.is_empty() {
            tracing::debug!(
                group_id = %self.group_id,
                inbox_id = %self.context.inbox_id(),
                "Group has no pending removal members"
            );
            return Ok(());
        }

        let is_super_admin = self.live_is_super_admin(self.context.inbox_id())?;
        if !is_super_admin {
            tracing::debug!(
                group_id = %self.group_id,
                inbox_id = %self.context.inbox_id(),
                "Current inbox ID is not in admin or super admin list, skipping pending removal processing"
            );
            return Ok(());
        }

        // Get current group members to validate which ones actually exist
        let members = self.members().await?;
        let member_inbox_ids: HashSet<String> =
            members.iter().map(|m| m.inbox_id.clone()).collect();

        // Filter pending removals to only include actual group members
        let valid_removals: Vec<&str> = pending_removal_list
            .iter()
            .filter(|inbox_id| member_inbox_ids.contains(*inbox_id))
            .map(|s| s.as_str())
            .collect();

        if valid_removals.is_empty() {
            tracing::warn!(
                group_id = %self.group_id,
                pending_count = pending_removal_list.len(),
                "No valid members found in pending removal list"
            );
            return Ok(());
        }
        // Log members that are in pending list but not in group
        let invalid_removals: Vec<&String> = pending_removal_list
            .iter()
            .filter(|inbox_id| !member_inbox_ids.contains(*inbox_id))
            .collect();

        if !invalid_removals.is_empty() {
            tracing::warn!(
                group_id = %self.group_id,
                invalid_members = ?invalid_removals,
                "Some members in pending removal list are not in the group"
            );
        }

        // Remove all valid members at once
        tracing::info!(
            group_id = %self.group_id,
            removing_count = valid_removals.len(),
            members_to_remove = ?valid_removals,
            "Removing pending members from group"
        );

        match self.remove_inboxes(&valid_removals, true).await {
            Ok(_) => {
                tracing::info!(
                    group_id = %self.group_id,
                    removed_count = valid_removals.len(),
                    removed_members = ?valid_removals,
                    "Successfully removed all pending members from group"
                );
            }
            Err(e) => {
                tracing::error!(
                    group_id = %self.group_id,
                    removed_members = ?valid_removals,
                    error = %e,
                    "Failed to remove pending members from group"
                );
                return Err(e);
            }
        }

        Ok(())
    }

    /// Removes members from the pending removal list who are no longer in the group.
    ///
    /// Iterates through all members in the pending removal list, checking each one to see
    /// if they're still in the group. If a member is no longer in the group, they are
    /// removed from the pending list. The pending list is refreshed after each removal
    /// to ensure we're working with the most current data.
    ///
    /// # Returns
    /// * `Ok(())` - Successfully processed all pending removal members
    /// * `Err(GroupError)` - Failed to retrieve data or update the pending list
    pub async fn cleanup_pending_removal_list(&self) -> Result<(), GroupError> {
        tracing::debug!(
            group_id = %self.group_id,
            "Starting pending removal list cleanup"
        );

        // Get both lists upfront
        let pending_removal_list = self.pending_remove_list()?;

        if pending_removal_list.is_empty() {
            tracing::debug!(
                group_id = %self.group_id,
                "No pending removals to clean up"
            );
            // Clear the pending leave request status
            self.context
                .db()
                .set_group_has_pending_leave_request_status(&self.group_id, Some(false))?;
            return Ok(());
        }

        // Get current group members
        let current_members = self.members().await?;
        let current_member_ids: Vec<String> = current_members
            .iter()
            .map(|member| member.inbox_id.clone())
            .collect();

        // Calculate removed members: users in pending list but not in current group
        let removed_members: Vec<String> = pending_removal_list
            .iter()
            .filter(|pending_user| !current_member_ids.contains(pending_user))
            .cloned()
            .collect();

        if !removed_members.is_empty() {
            tracing::info!(
                group_id = %self.group_id,
                removed_count = removed_members.len(),
                removed_members = ?removed_members,
                "Removing members from pending removal list - they are no longer in the group"
            );

            // Remove all users who are no longer in the group from pending list
            self.context
                .db()
                .delete_pending_remove_users(&self.group_id, removed_members)?;
        }

        // After cleanup, check if there are any pending removals left
        let remaining_pending_list = self.pending_remove_list()?;
        if remaining_pending_list.is_empty() {
            // Clear the pending leave request status if no pending removals remain
            self.context
                .db()
                .set_group_has_pending_leave_request_status(&self.group_id, Some(false))?;
        }

        tracing::info!(
            group_id = %self.group_id,
            remaining_pending = remaining_pending_list.len(),
            "Finished cleaning up pending removal list"
        );

        Ok(())
    }

    // implements: GMOD-031
    pub async fn leave_group(&self) -> Result<(), GroupError> {
        self.ensure_not_paused().await?;

        // Check if user is a member
        let is_member = self.is_member().await?;
        if !is_member {
            return Err(GroupLeaveValidationError::NotAGroupMember.into());
        }

        //check member size
        let members = self.members().await?;

        // check if the group has other members
        if members.len() == 1 {
            return Err(GroupLeaveValidationError::SingleMemberLeaveRejected.into());
        }

        // check if the conversation is not a DM
        if self.metadata().await?.conversation_type == ConversationType::Dm {
            return Err(GroupLeaveValidationError::DmLeaveForbidden.into());
        }

        let is_super_admin = self.live_is_super_admin(self.context.inbox_id())?;

        // super-admin cannot leave a group; must be demoted first
        // since SuperAdmins can't remove other SuperAdmins they need to be demoted first
        if is_super_admin {
            return Err(GroupLeaveValidationError::SuperAdminLeaveForbidden.into());
        }

        if !self.is_in_pending_remove(self.context.inbox_id())? {
            let content = LeaveRequestCodec::encode(LeaveRequest {
                authenticated_note: None,
            })?;
            self.send_message(
                &encoded_content_to_bytes(content),
                SendMessageOpts::default(),
            )
            .await?;
        };
        Ok(())
    }

    /// Checks if the current user is a member of the group.
    /// Returns true if the user is a member, false otherwise.
    #[tracing::instrument(level = "debug", skip(self))]
    async fn is_member(&self) -> Result<bool, GroupError> {
        let members = self.members().await?;
        Ok(members
            .iter()
            .any(|m| m.inbox_id == self.context.inbox_id()))
    }
}