feat: implement DeleteConversations interface. (#3549)

* feat: implement DeleteConversations interface.

* remove unused comment.

* update logic and rename method.

pkg/common/storage/cache/conversation.go

@@ -16,6 +16,7 @@ package cache
 
 import (
 	"context"
+
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 )
 
@@ -57,7 +58,7 @@ type ConversationCache interface {
 	GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
 	DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache
 	DelConversationNotNotifyMessageUserIDs(userIDs ...string) ConversationCache
-	DelConversationPinnedMessageUserIDs(userIDs ...string) ConversationCache
+	DelUserPinnedConversations(userIDs ...string) ConversationCache
 	DelConversationVersionUserIDs(userIDs ...string) ConversationCache
 
 	FindMaxConversationUserVersion(ctx context.Context, userID string) (*relationtb.VersionLog, error)

pkg/common/storage/cache/redis/conversation.go

@@ -253,7 +253,7 @@ func (c *ConversationRedisCache) DelConversationNotNotifyMessageUserIDs(userIDs
 	return cache
 }
 
-func (c *ConversationRedisCache) DelConversationPinnedMessageUserIDs(userIDs ...string) cache.ConversationCache {
+func (c *ConversationRedisCache) DelUserPinnedConversations(userIDs ...string) cache.ConversationCache {
 	cache := c.CloneConversationCache()
 	for _, userID := range userIDs {
 		cache.AddKeys(c.getPinnedConversationIDsKey(userID))

pkg/common/storage/controller/conversation.go

@@ -78,6 +78,8 @@ type ConversationDatabase interface {
 	GetPinnedConversationIDs(ctx context.Context, userID string) ([]string, error)
 	// FindRandConversation finds random conversations based on the specified timestamp and limit.
 	FindRandConversation(ctx context.Context, ts int64, limit int) ([]*relationtb.Conversation, error)
+
+	DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error)
 }
 
 func NewConversationDatabase(conversation database.Conversation, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
@@ -120,7 +122,7 @@ func (c *conversationDatabase) SetUsersConversationFieldTx(ctx context.Context,
 				cache = cache.DelConversationNotNotifyMessageUserIDs(userIDs...)
 			}
 			if _, ok := fieldMap["is_pinned"]; ok {
-				cache = cache.DelConversationPinnedMessageUserIDs(userIDs...)
+				cache = cache.DelUserPinnedConversations(userIDs...)
 			}
 			cache = cache.DelConversationVersionUserIDs(haveUserIDs...)
 		}
@@ -172,7 +174,7 @@ func (c *conversationDatabase) UpdateUsersConversationField(ctx context.Context,
 		cache = cache.DelConversationNotNotifyMessageUserIDs(userIDs...)
 	}
 	if _, ok := args["is_pinned"]; ok {
-		cache = cache.DelConversationPinnedMessageUserIDs(userIDs...)
+		cache = cache.DelUserPinnedConversations(userIDs...)
 	}
 	return cache.ChainExecDel(ctx)
 }
@@ -203,7 +205,7 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat
 		DelUserConversationIDsHash(userIDs...).
 		DelConversationVersionUserIDs(userIDs...).
 		DelConversationNotNotifyMessageUserIDs(notNotifyUserIDs...).
-		DelConversationPinnedMessageUserIDs(pinnedUserIDs...).
+		DelUserPinnedConversations(pinnedUserIDs...).
 		ChainExecDel(ctx)
 }
 
@@ -259,7 +261,7 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs
 		cache := c.cache.CloneConversationCache()
 		cache = cache.DelConversationVersionUserIDs(ownerUserID).
 			DelConversationNotNotifyMessageUserIDs(ownerUserID).
-			DelConversationPinnedMessageUserIDs(ownerUserID)
+			DelUserPinnedConversations(ownerUserID)
 
 		groupIDs := datautil.Distinct(datautil.Filter(conversations, func(e *relationtb.Conversation) (string, bool) {
 			return e.GroupID, e.GroupID != ""
@@ -429,3 +431,21 @@ func (c *conversationDatabase) GetPinnedConversationIDs(ctx context.Context, use
 func (c *conversationDatabase) FindRandConversation(ctx context.Context, ts int64, limit int) ([]*relationtb.Conversation, error) {
 	return c.conversationDB.FindRandConversation(ctx, ts, limit)
 }
+
+func (c *conversationDatabase) DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) {
+	return c.tx.Transaction(ctx, func(ctx context.Context) error {
+		err = c.conversationDB.DeleteUsersConversations(ctx, userID, conversationIDs)
+		if err != nil {
+			return err
+		}
+		cache := c.cache.CloneConversationCache()
+		cache = cache.DelConversations(userID, conversationIDs...).
+			DelConversationVersionUserIDs(userID).
+			DelConversationIDs(userID).
+			DelUserConversationIDsHash(userID).
+			DelConversationNotNotifyMessageUserIDs(userID).
+			DelUserPinnedConversations(userID)
+
+		return cache.ChainExecDel(ctx)
+	})
+}

pkg/common/storage/database/conversation.go

@@ -44,4 +44,5 @@ type Conversation interface {
 	GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
 	FindConversationUserVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error)
 	FindRandConversation(ctx context.Context, ts int64, limit int) ([]*model.Conversation, error)
+	DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error)
 }

pkg/common/storage/database/mgo/conversation.go

@@ -308,3 +308,20 @@ func (c *ConversationMgo) FindRandConversation(ctx context.Context, ts int64, li
 	}
 	return mongoutil.Aggregate[*model.Conversation](ctx, c.coll, pipeline)
 }
+
+func (c *ConversationMgo) DeleteUsersConversations(ctx context.Context, userID string, conversationIDs []string) (err error) {
+	if len(conversationIDs) == 0 {
+		return nil
+	}
+	return mongoutil.IncrVersion(func() error {
+		err := mongoutil.DeleteMany(ctx, c.coll, bson.M{"owner_user_id": userID, "conversation_id": bson.M{"$in": conversationIDs}})
+		return err
+	}, func() error {
+		for _, conversationID := range conversationIDs {
+			if err := c.version.IncrVersion(ctx, userID, []string{conversationID}, model.VersionStateDelete); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}