Merge 9c2dee8ee2481065478db0b87096ab484b3750cb into 6d80e6ec1221f2b3650cda19c7642262530d08df

fix: add rpc interface permission check

config/webhooks.yml

@@ -7,11 +7,11 @@ beforeSendSingleMsg:
   # If not set, all contentType messages will through this filter.
   deniedTypes: []
 beforeUpdateUserInfoEx:
-  enable:  false
+  enable: false
   timeout: 5
   failedContinue: true
 afterUpdateUserInfoEx:
-  enable:  false
+  enable: false
   timeout: 5
 afterSendSingleMsg:
   enable: false
@@ -181,3 +181,19 @@ afterImportFriends:
 afterRemoveBlack:
   enable: false
   timeout: 5
+beforeCreateSingleChatConversations:
+  enable: false
+  timeout: 5
+  failedContinue: false
+afterCreateSingleChatConversations:
+  enable: false
+  timeout: 5
+  failedContinue: false
+beforeCreateGroupChatConversations:
+  enable: false
+  timeout: 5
+  failedContinue: false
+afterCreateGroupChatConversations:
+  enable: false
+  timeout: 5
+  failedContinue: false

internal/api/conversation.go

@@ -16,6 +16,7 @@ package api
 
 import (
 	"github.com/gin-gonic/gin"
+
 	"github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/tools/a2r"
 )
@@ -71,3 +72,7 @@ func (o *ConversationApi) GetNotNotifyConversationIDs(c *gin.Context) {
 func (o *ConversationApi) GetPinnedConversationIDs(c *gin.Context) {
 	a2r.Call(c, conversation.ConversationClient.GetPinnedConversationIDs, o.Client)
 }
+
+func (o *ConversationApi) UpdateConversationsByUser(c *gin.Context) {
+	a2r.Call(c, conversation.ConversationClient.UpdateConversationsByUser, o.Client)
+}

internal/api/router.go

@@ -9,6 +9,8 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin/binding"
 	"github.com/go-playground/validator/v10"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
 	"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -28,7 +30,6 @@ import (
 	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mw"
-	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 const (
@@ -268,6 +269,7 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin
 		conversationGroup.POST("/get_owner_conversation", c.GetOwnerConversation)
 		conversationGroup.POST("/get_not_notify_conversation_ids", c.GetNotNotifyConversationIDs)
 		conversationGroup.POST("/get_pinned_conversation_ids", c.GetPinnedConversationIDs)
+		conversationGroup.POST("/update_conversations_by_user", c.UpdateConversationsByUser)
 	}
 
 	{

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -62,22 +62,3 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont
 	//		msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
 	//}
 }
-
-func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (*OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
-
-func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // an instance in the consumer group
-	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
-		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
-	for msg := range claim.Messages() {
-		ctx := mc.historyConsumerGroup.GetContextFromMsg(msg)
-		if len(msg.Value) != 0 {
-			mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess)
-		} else {
-			log.ZError(ctx, "mongo msg get from kafka but is nil", nil, "conversationID", msg.Key)
-		}
-		sess.MarkMessage(msg, "")
-	}
-	return nil
-}

internal/push/push.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/mcache"
@@ -106,7 +107,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	go func() {
 		pushHandler.WaitCache()
 		fn := func(ctx context.Context, key string, value []byte) error {
-			pushHandler.HandleMs2PsChat(ctx, value)
+			pushHandler.HandleMs2PsChat(authverify.WithTempAdmin(ctx), value)
 			return nil
 		}
 		consumerCtx := mcontext.SetOperationID(context.Background(), "push_"+strconv.Itoa(int(rand.Uint32())))

internal/rpc/conversation/callback.go

@@ -0,0 +1,117 @@
+package conversation
+
+import (
+	"context"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	dbModel "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
+	"github.com/openimsdk/tools/utils/datautil"
+)
+
+func (c *conversationServer) webhookBeforeCreateSingleChatConversations(ctx context.Context, before *config.BeforeConfig, req *dbModel.Conversation) error {
+	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
+		cbReq := &callbackstruct.CallbackBeforeCreateSingleChatConversationsReq{
+			CallbackCommand:  callbackstruct.CallbackBeforeCreateSingleChatConversationsCommand,
+			OwnerUserID:      req.OwnerUserID,
+			ConversationID:   req.ConversationID,
+			ConversationType: req.ConversationType,
+			UserID:           req.UserID,
+			RecvMsgOpt:       req.RecvMsgOpt,
+			IsPinned:         req.IsPinned,
+			IsPrivateChat:    req.IsPrivateChat,
+			BurnDuration:     req.BurnDuration,
+			GroupAtType:      req.GroupAtType,
+			AttachedInfo:     req.AttachedInfo,
+			Ex:               req.Ex,
+		}
+
+		resp := &callbackstruct.CallbackBeforeCreateSingleChatConversationsResp{}
+
+		if err := c.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+			return err
+		}
+
+		datautil.NotNilReplace(&req.RecvMsgOpt, resp.RecvMsgOpt)
+		datautil.NotNilReplace(&req.IsPinned, resp.IsPinned)
+		datautil.NotNilReplace(&req.IsPrivateChat, resp.IsPrivateChat)
+		datautil.NotNilReplace(&req.BurnDuration, resp.BurnDuration)
+		datautil.NotNilReplace(&req.GroupAtType, resp.GroupAtType)
+		datautil.NotNilReplace(&req.AttachedInfo, resp.AttachedInfo)
+		datautil.NotNilReplace(&req.Ex, resp.Ex)
+		return nil
+	})
+}
+
+func (c *conversationServer) webhookAfterCreateSingleChatConversations(ctx context.Context, after *config.AfterConfig, req *dbModel.Conversation) error {
+	cbReq := &callbackstruct.CallbackAfterCreateSingleChatConversationsReq{
+		CallbackCommand:  callbackstruct.CallbackAfterCreateSingleChatConversationsCommand,
+		OwnerUserID:      req.OwnerUserID,
+		ConversationID:   req.ConversationID,
+		ConversationType: req.ConversationType,
+		UserID:           req.UserID,
+		RecvMsgOpt:       req.RecvMsgOpt,
+		IsPinned:         req.IsPinned,
+		IsPrivateChat:    req.IsPrivateChat,
+		BurnDuration:     req.BurnDuration,
+		GroupAtType:      req.GroupAtType,
+		AttachedInfo:     req.AttachedInfo,
+		Ex:               req.Ex,
+	}
+
+	c.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterCreateSingleChatConversationsResp{}, after)
+	return nil
+}
+
+func (c *conversationServer) webhookBeforeCreateGroupChatConversations(ctx context.Context, before *config.BeforeConfig, req *dbModel.Conversation) error {
+	return webhook.WithCondition(ctx, before, func(ctx context.Context) error {
+		cbReq := &callbackstruct.CallbackBeforeCreateGroupChatConversationsReq{
+			CallbackCommand:  callbackstruct.CallbackBeforeCreateGroupChatConversationsCommand,
+			ConversationID:   req.ConversationID,
+			ConversationType: req.ConversationType,
+			GroupID:          req.GroupID,
+			RecvMsgOpt:       req.RecvMsgOpt,
+			IsPinned:         req.IsPinned,
+			IsPrivateChat:    req.IsPrivateChat,
+			BurnDuration:     req.BurnDuration,
+			GroupAtType:      req.GroupAtType,
+			AttachedInfo:     req.AttachedInfo,
+			Ex:               req.Ex,
+		}
+
+		resp := &callbackstruct.CallbackBeforeCreateGroupChatConversationsResp{}
+
+		if err := c.webhookClient.SyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, resp, before); err != nil {
+			return err
+		}
+
+		datautil.NotNilReplace(&req.RecvMsgOpt, resp.RecvMsgOpt)
+		datautil.NotNilReplace(&req.IsPinned, resp.IsPinned)
+		datautil.NotNilReplace(&req.IsPrivateChat, resp.IsPrivateChat)
+		datautil.NotNilReplace(&req.BurnDuration, resp.BurnDuration)
+		datautil.NotNilReplace(&req.GroupAtType, resp.GroupAtType)
+		datautil.NotNilReplace(&req.AttachedInfo, resp.AttachedInfo)
+		datautil.NotNilReplace(&req.Ex, resp.Ex)
+		return nil
+	})
+}
+
+func (c *conversationServer) webhookAfterCreateGroupChatConversations(ctx context.Context, after *config.AfterConfig, req *dbModel.Conversation) error {
+	cbReq := &callbackstruct.CallbackAfterCreateGroupChatConversationsReq{
+		CallbackCommand:  callbackstruct.CallbackAfterCreateGroupChatConversationsCommand,
+		ConversationID:   req.ConversationID,
+		ConversationType: req.ConversationType,
+		GroupID:          req.GroupID,
+		RecvMsgOpt:       req.RecvMsgOpt,
+		IsPinned:         req.IsPinned,
+		IsPrivateChat:    req.IsPrivateChat,
+		BurnDuration:     req.BurnDuration,
+		GroupAtType:      req.GroupAtType,
+		AttachedInfo:     req.AttachedInfo,
+		Ex:               req.Ex,
+	}
+
+	c.webhookClient.AsyncPost(ctx, cbReq.GetCallbackCommand(), cbReq, &callbackstruct.CallbackAfterCreateGroupChatConversationsResp{}, after)
+	return nil
+}

internal/rpc/conversation/conversation.go

@@ -23,6 +23,8 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/dbbuild"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 
+	"google.golang.org/grpc"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
@@ -31,6 +33,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	dbModel "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
 	"github.com/openimsdk/open-im-server/v3/pkg/localcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/protocol/constant"
@@ -40,7 +43,6 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
-	"google.golang.org/grpc"
 )
 
 type conversationServer struct {
@@ -50,9 +52,10 @@ type conversationServer struct {
 	conversationNotificationSender *ConversationNotificationSender
 	config                         *Config
 
-	userClient  *rpcli.UserClient
-	msgClient   *rpcli.MsgClient
-	groupClient *rpcli.GroupClient
+	webhookClient *webhook.Client
+	userClient    *rpcli.UserClient
+	msgClient     *rpcli.MsgClient
+	groupClient   *rpcli.GroupClient
 }
 
 type Config struct {
@@ -61,6 +64,7 @@ type Config struct {
 	MongodbConfig      config.Mongo
 	NotificationConfig config.Notification
 	Share              config.Share
+	WebhooksConfig     config.Webhooks
 	LocalCacheConfig   config.LocalCache
 	Discovery          config.Discovery
 }
@@ -91,16 +95,25 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
 	if err != nil {
 		return err
 	}
+
 	msgClient := rpcli.NewMsgClient(msgConn)
+
+	cs := conversationServer{
+		config:        config,
+		webhookClient: webhook.NewWebhookClient(config.WebhooksConfig.URL),
+		userClient:    rpcli.NewUserClient(userConn),
+		groupClient:   rpcli.NewGroupClient(groupConn),
+		msgClient:     msgClient,
+	}
+
+	cs.conversationNotificationSender = NewConversationNotificationSender(&config.NotificationConfig, msgClient)
+	cs.conversationDatabase = controller.NewConversationDatabase(
+		conversationDB,
+		redis.NewConversationRedis(rdb, &config.LocalCacheConfig, conversationDB),
+		mgocli.GetTx())
+
 	localcache.InitLocalCache(&config.LocalCacheConfig)
-	pbconversation.RegisterConversationServer(server, &conversationServer{
-		conversationNotificationSender: NewConversationNotificationSender(&config.NotificationConfig, msgClient),
-		conversationDatabase: controller.NewConversationDatabase(conversationDB,
-			redis.NewConversationRedis(rdb, &config.LocalCacheConfig, conversationDB), mgocli.GetTx()),
-		userClient:  rpcli.NewUserClient(userConn),
-		groupClient: rpcli.NewGroupClient(groupConn),
-		msgClient:   msgClient,
-	})
+	pbconversation.RegisterConversationServer(server, &cs)
 	return nil
 }
 
@@ -355,47 +368,76 @@ func (c *conversationServer) CreateSingleChatConversations(ctx context.Context,
 	var conversation dbModel.Conversation
 	switch req.ConversationType {
 	case constant.SingleChatType:
-		var conversation dbModel.Conversation
+		// sendUser create
 		conversation.ConversationID = req.ConversationID
 		conversation.ConversationType = req.ConversationType
 		conversation.OwnerUserID = req.SendID
 		conversation.UserID = req.RecvID
+		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
+			return nil, err
+		}
+
 		err := c.conversationDatabase.CreateConversation(ctx, []*dbModel.Conversation{&conversation})
 		if err != nil {
 			log.ZWarn(ctx, "create conversation failed", err, "conversation", conversation)
 		}
 
+		c.webhookAfterCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.AfterCreateSingleChatConversations, &conversation)
+
+		// recvUser create
 		conversation2 := conversation
 		conversation2.OwnerUserID = req.RecvID
 		conversation2.UserID = req.SendID
+		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
+			return nil, err
+		}
+
 		err = c.conversationDatabase.CreateConversation(ctx, []*dbModel.Conversation{&conversation2})
 		if err != nil {
 			log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation)
 		}
+
+		c.webhookAfterCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.AfterCreateSingleChatConversations, &conversation2)
 	case constant.NotificationChatType:
-		var conversation dbModel.Conversation
 		conversation.ConversationID = req.ConversationID
 		conversation.ConversationType = req.ConversationType
 		conversation.OwnerUserID = req.RecvID
 		conversation.UserID = req.SendID
+		if err := c.webhookBeforeCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateSingleChatConversations, &conversation); err != nil && err != servererrs.ErrCallbackContinue {
+			return nil, err
+		}
+
 		err := c.conversationDatabase.CreateConversation(ctx, []*dbModel.Conversation{&conversation})
 		if err != nil {
 			log.ZWarn(ctx, "create conversation failed", err, "conversation2", conversation)
 		}
+
+		c.webhookAfterCreateSingleChatConversations(ctx, &c.config.WebhooksConfig.AfterCreateSingleChatConversations, &conversation)
 	}
 
 	return &pbconversation.CreateSingleChatConversationsResp{}, nil
 }
 
 func (c *conversationServer) CreateGroupChatConversations(ctx context.Context, req *pbconversation.CreateGroupChatConversationsReq) (*pbconversation.CreateGroupChatConversationsResp, error) {
-	err := c.conversationDatabase.CreateGroupChatConversation(ctx, req.GroupID, req.UserIDs)
+	var conversation dbModel.Conversation
+
+	conversation.ConversationID = msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
+	conversation.GroupID = req.GroupID
+	conversation.ConversationType = constant.ReadGroupChatType
+
+	if err := c.webhookBeforeCreateGroupChatConversations(ctx, &c.config.WebhooksConfig.BeforeCreateGroupChatConversations, &conversation); err != nil {
+		return nil, err
+	}
+
+	err := c.conversationDatabase.CreateGroupChatConversation(ctx, req.GroupID, req.UserIDs, &conversation)
 	if err != nil {
 		return nil, err
 	}
-	conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID)
-	if err := c.msgClient.SetUserConversationMaxSeq(ctx, conversationID, req.UserIDs, 0); err != nil {
+	if err := c.msgClient.SetUserConversationMaxSeq(ctx, conversation.ConversationID, req.UserIDs, 0); err != nil {
 		return nil, err
 	}
+
+	c.webhookAfterCreateGroupChatConversations(ctx, &c.config.WebhooksConfig.AfterCreateGroupChatConversations, &conversation)
 	return &pbconversation.CreateGroupChatConversationsResp{}, nil
 }
 

internal/rpc/conversation/notification.go

@@ -17,11 +17,11 @@ package conversation
 import (
 	"context"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/notification"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	"github.com/openimsdk/protocol/msg"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/notification"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/sdkws"
 )

internal/rpc/group/cache.go

@@ -17,14 +17,13 @@ package group
 import (
 	"context"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
 	pbgroup "github.com/openimsdk/protocol/group"
 )
 
 // GetGroupInfoCache get group info from cache.
-func (s *groupServer) GetGroupInfoCache(ctx context.Context, req *pbgroup.GetGroupInfoCacheReq) (*pbgroup.GetGroupInfoCacheResp, error) {
-	group, err := s.db.TakeGroup(ctx, req.GroupID)
+func (g *groupServer) GetGroupInfoCache(ctx context.Context, req *pbgroup.GetGroupInfoCacheReq) (*pbgroup.GetGroupInfoCacheResp, error) {
+	group, err := g.db.TakeGroup(ctx, req.GroupID)
 	if err != nil {
 		return nil, err
 	}
@@ -34,7 +33,7 @@ func (s *groupServer) GetGroupInfoCache(ctx context.Context, req *pbgroup.GetGro
 }
 
 func (g *groupServer) GetGroupMemberCache(ctx context.Context, req *pbgroup.GetGroupMemberCacheReq) (*pbgroup.GetGroupMemberCacheResp, error) {
-	if err := authverify.CheckAccess(ctx, req.GroupMemberID); err != nil {
+	if err := g.checkAdminOrInGroup(ctx, req.GroupID); err != nil {
 		return nil, err
 	}
 	members, err := g.db.TakeGroupMember(ctx, req.GroupID, req.GroupMemberID)

internal/rpc/group/group.go

@@ -23,8 +23,11 @@ import (
 	"strings"
 	"time"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/dbbuild"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 
+	"google.golang.org/grpc"
+
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -1305,6 +1308,9 @@ func (g *groupServer) GetGroups(ctx context.Context, req *pbgroup.GetGroupsReq)
 }
 
 func (g *groupServer) GetGroupMembersCMS(ctx context.Context, req *pbgroup.GetGroupMembersCMSReq) (*pbgroup.GetGroupMembersCMSResp, error) {
+	if err := g.checkAdminOrInGroup(ctx, req.GroupID); err != nil {
+		return nil, err
+	}
 	total, members, err := g.db.SearchGroupMember(ctx, req.UserName, req.GroupID, req.Pagination)
 	if err != nil {
 		return nil, err
@@ -1714,6 +1720,9 @@ func (g *groupServer) GetUserInGroupMembers(ctx context.Context, req *pbgroup.Ge
 	if len(req.GroupIDs) == 0 {
 		return nil, errs.ErrArgs.WrapMsg("groupIDs empty")
 	}
+	if err := authverify.CheckAccess(ctx, req.UserID); err != nil {
+		return nil, err
+	}
 	members, err := g.db.FindGroupMemberUser(ctx, req.GroupIDs, req.UserID)
 	if err != nil {
 		return nil, err
@@ -1745,6 +1754,9 @@ func (g *groupServer) GetGroupMemberRoleLevel(ctx context.Context, req *pbgroup.
 	if len(req.RoleLevels) == 0 {
 		return nil, errs.ErrArgs.WrapMsg("RoleLevels empty")
 	}
+	if err := g.checkAdminOrInGroup(ctx, req.GroupID); err != nil {
+		return nil, err
+	}
 	members, err := g.db.FindGroupMemberRoleLevels(ctx, req.GroupID, req.RoleLevels)
 	if err != nil {
 		return nil, err

internal/rpc/group/statistics.go

@@ -23,7 +23,7 @@ import (
 	"github.com/openimsdk/tools/errs"
 )
 
-func (s *groupServer) GroupCreateCount(ctx context.Context, req *group.GroupCreateCountReq) (*group.GroupCreateCountResp, error) {
+func (g *groupServer) GroupCreateCount(ctx context.Context, req *group.GroupCreateCountReq) (*group.GroupCreateCountResp, error) {
 	if req.Start > req.End {
 		return nil, errs.ErrArgs.WrapMsg("start > end: %d > %d", req.Start, req.End)
 	}
@@ -35,11 +35,11 @@ func (s *groupServer) GroupCreateCount(ctx context.Context, req *group.GroupCrea
 		return nil, err
 	}
 	start := time.UnixMilli(req.Start)
-	before, err := s.db.CountTotal(ctx, &start)
+	before, err := g.db.CountTotal(ctx, &start)
 	if err != nil {
 		return nil, err
 	}
-	count, err := s.db.CountRangeEverydayTotal(ctx, start, time.UnixMilli(req.End))
+	count, err := g.db.CountRangeEverydayTotal(ctx, start, time.UnixMilli(req.End))
 	if err != nil {
 		return nil, err
 	}

internal/rpc/msg/send.go

@@ -24,7 +24,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/open-im-server/v3/pkg/util/conversationutil"
 	"github.com/openimsdk/protocol/constant"
-	pbconversation "github.com/openimsdk/protocol/conversation"
+	pbconv "github.com/openimsdk/protocol/conversation"
 	pbmsg "github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/protocol/wrapperspb"
@@ -32,7 +32,6 @@ import (
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
-	"google.golang.org/protobuf/proto"
 )
 
 func (m *msgServer) SendMsg(ctx context.Context, req *pbmsg.SendMsgReq) (*pbmsg.SendMsgResp, error) {
@@ -110,7 +109,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
 
 	var atUserID []string
 
-	conversation := &pbconversation.ConversationReq{
+	conversation := &pbconv.ConversationReq{
 		ConversationID:   msgprocessor.GetConversationIDByMsg(msg),
 		ConversationType: msg.SessionType,
 		GroupID:          msg.GroupID,

internal/tools/cron/cron_test.go

@@ -46,7 +46,7 @@ func TestName(t *testing.T) {
 
 	srv := &cronServer{
 		ctx: ctx,
-		config: &CronTaskConfig{
+		config: &Config{
 			CronTask: config.CronTask{
 				RetainChatRecords: 1,
 				FileExpireTime:    1,

pkg/callbackstruct/constant.go

@@ -15,51 +15,55 @@
 package callbackstruct
 
 const (
-	CallbackBeforeInviteJoinGroupCommand    = "callbackBeforeInviteJoinGroupCommand"
-	CallbackAfterJoinGroupCommand           = "callbackAfterJoinGroupCommand"
-	CallbackAfterSetGroupInfoCommand        = "callbackAfterSetGroupInfoCommand"
-	CallbackAfterSetGroupInfoExCommand      = "callbackAfterSetGroupInfoExCommand"
-	CallbackBeforeSetGroupInfoCommand       = "callbackBeforeSetGroupInfoCommand"
-	CallbackBeforeSetGroupInfoExCommand     = "callbackBeforeSetGroupInfoExCommand"
-	CallbackAfterRevokeMsgCommand           = "callbackBeforeAfterMsgCommand"
-	CallbackBeforeAddBlackCommand           = "callbackBeforeAddBlackCommand"
-	CallbackAfterAddFriendCommand           = "callbackAfterAddFriendCommand"
-	CallbackBeforeAddFriendAgreeCommand     = "callbackBeforeAddFriendAgreeCommand"
-	CallbackAfterAddFriendAgreeCommand      = "callbackAfterAddFriendAgreeCommand"
-	CallbackAfterDeleteFriendCommand        = "callbackAfterDeleteFriendCommand"
-	CallbackBeforeImportFriendsCommand      = "callbackBeforeImportFriendsCommand"
-	CallbackAfterImportFriendsCommand       = "callbackAfterImportFriendsCommand"
-	CallbackAfterRemoveBlackCommand         = "callbackAfterRemoveBlackCommand"
-	CallbackAfterQuitGroupCommand           = "callbackAfterQuitGroupCommand"
-	CallbackAfterKickGroupCommand           = "callbackAfterKickGroupCommand"
-	CallbackAfterDisMissGroupCommand        = "callbackAfterDisMissGroupCommand"
-	CallbackBeforeJoinGroupCommand          = "callbackBeforeJoinGroupCommand"
-	CallbackAfterGroupMsgReadCommand        = "callbackAfterGroupMsgReadCommand"
-	CallbackBeforeMsgModifyCommand          = "callbackBeforeMsgModifyCommand"
-	CallbackAfterUpdateUserInfoCommand      = "callbackAfterUpdateUserInfoCommand"
-	CallbackAfterUpdateUserInfoExCommand    = "callbackAfterUpdateUserInfoExCommand"
-	CallbackBeforeUpdateUserInfoExCommand   = "callbackBeforeUpdateUserInfoExCommand"
-	CallbackBeforeUserRegisterCommand       = "callbackBeforeUserRegisterCommand"
-	CallbackAfterUserRegisterCommand        = "callbackAfterUserRegisterCommand"
-	CallbackAfterTransferGroupOwnerCommand  = "callbackAfterTransferGroupOwnerCommand"
-	CallbackBeforeSetFriendRemarkCommand    = "callbackBeforeSetFriendRemarkCommand"
-	CallbackAfterSetFriendRemarkCommand     = "callbackAfterSetFriendRemarkCommand"
-	CallbackAfterSingleMsgReadCommand       = "callbackAfterSingleMsgReadCommand"
-	CallbackBeforeSendSingleMsgCommand      = "callbackBeforeSendSingleMsgCommand"
-	CallbackAfterSendSingleMsgCommand       = "callbackAfterSendSingleMsgCommand"
-	CallbackBeforeSendGroupMsgCommand       = "callbackBeforeSendGroupMsgCommand"
-	CallbackAfterSendGroupMsgCommand        = "callbackAfterSendGroupMsgCommand"
-	CallbackAfterUserOnlineCommand          = "callbackAfterUserOnlineCommand"
-	CallbackAfterUserOfflineCommand         = "callbackAfterUserOfflineCommand"
-	CallbackAfterUserKickOffCommand         = "callbackAfterUserKickOffCommand"
-	CallbackBeforeOfflinePushCommand        = "callbackBeforeOfflinePushCommand"
-	CallbackBeforeOnlinePushCommand         = "callbackBeforeOnlinePushCommand"
-	CallbackBeforeGroupOnlinePushCommand    = "callbackBeforeGroupOnlinePushCommand"
-	CallbackBeforeAddFriendCommand          = "callbackBeforeAddFriendCommand"
-	CallbackBeforeUpdateUserInfoCommand     = "callbackBeforeUpdateUserInfoCommand"
-	CallbackBeforeCreateGroupCommand        = "callbackBeforeCreateGroupCommand"
-	CallbackAfterCreateGroupCommand         = "callbackAfterCreateGroupCommand"
-	CallbackBeforeMembersJoinGroupCommand   = "callbackBeforeMembersJoinGroupCommand"
-	CallbackBeforeSetGroupMemberInfoCommand = "callbackBeforeSetGroupMemberInfoCommand"
-	CallbackAfterSetGroupMemberInfoCommand  = "callbackAfterSetGroupMemberInfoCommand"
+	CallbackBeforeInviteJoinGroupCommand               = "callbackBeforeInviteJoinGroupCommand"
+	CallbackAfterJoinGroupCommand                      = "callbackAfterJoinGroupCommand"
+	CallbackAfterSetGroupInfoCommand                   = "callbackAfterSetGroupInfoCommand"
+	CallbackAfterSetGroupInfoExCommand                 = "callbackAfterSetGroupInfoExCommand"
+	CallbackBeforeSetGroupInfoCommand                  = "callbackBeforeSetGroupInfoCommand"
+	CallbackBeforeSetGroupInfoExCommand                = "callbackBeforeSetGroupInfoExCommand"
+	CallbackAfterRevokeMsgCommand                      = "callbackBeforeAfterMsgCommand"
+	CallbackBeforeAddBlackCommand                      = "callbackBeforeAddBlackCommand"
+	CallbackAfterAddFriendCommand                      = "callbackAfterAddFriendCommand"
+	CallbackBeforeAddFriendAgreeCommand                = "callbackBeforeAddFriendAgreeCommand"
+	CallbackAfterAddFriendAgreeCommand                 = "callbackAfterAddFriendAgreeCommand"
+	CallbackAfterDeleteFriendCommand                   = "callbackAfterDeleteFriendCommand"
+	CallbackBeforeImportFriendsCommand                 = "callbackBeforeImportFriendsCommand"
+	CallbackAfterImportFriendsCommand                  = "callbackAfterImportFriendsCommand"
+	CallbackAfterRemoveBlackCommand                    = "callbackAfterRemoveBlackCommand"
+	CallbackAfterQuitGroupCommand                      = "callbackAfterQuitGroupCommand"
+	CallbackAfterKickGroupCommand                      = "callbackAfterKickGroupCommand"
+	CallbackAfterDisMissGroupCommand                   = "callbackAfterDisMissGroupCommand"
+	CallbackBeforeJoinGroupCommand                     = "callbackBeforeJoinGroupCommand"
+	CallbackAfterGroupMsgReadCommand                   = "callbackAfterGroupMsgReadCommand"
+	CallbackBeforeMsgModifyCommand                     = "callbackBeforeMsgModifyCommand"
+	CallbackAfterUpdateUserInfoCommand                 = "callbackAfterUpdateUserInfoCommand"
+	CallbackAfterUpdateUserInfoExCommand               = "callbackAfterUpdateUserInfoExCommand"
+	CallbackBeforeUpdateUserInfoExCommand              = "callbackBeforeUpdateUserInfoExCommand"
+	CallbackBeforeUserRegisterCommand                  = "callbackBeforeUserRegisterCommand"
+	CallbackAfterUserRegisterCommand                   = "callbackAfterUserRegisterCommand"
+	CallbackAfterTransferGroupOwnerCommand             = "callbackAfterTransferGroupOwnerCommand"
+	CallbackBeforeSetFriendRemarkCommand               = "callbackBeforeSetFriendRemarkCommand"
+	CallbackAfterSetFriendRemarkCommand                = "callbackAfterSetFriendRemarkCommand"
+	CallbackAfterSingleMsgReadCommand                  = "callbackAfterSingleMsgReadCommand"
+	CallbackBeforeSendSingleMsgCommand                 = "callbackBeforeSendSingleMsgCommand"
+	CallbackAfterSendSingleMsgCommand                  = "callbackAfterSendSingleMsgCommand"
+	CallbackBeforeSendGroupMsgCommand                  = "callbackBeforeSendGroupMsgCommand"
+	CallbackAfterSendGroupMsgCommand                   = "callbackAfterSendGroupMsgCommand"
+	CallbackAfterUserOnlineCommand                     = "callbackAfterUserOnlineCommand"
+	CallbackAfterUserOfflineCommand                    = "callbackAfterUserOfflineCommand"
+	CallbackAfterUserKickOffCommand                    = "callbackAfterUserKickOffCommand"
+	CallbackBeforeOfflinePushCommand                   = "callbackBeforeOfflinePushCommand"
+	CallbackBeforeOnlinePushCommand                    = "callbackBeforeOnlinePushCommand"
+	CallbackBeforeGroupOnlinePushCommand               = "callbackBeforeGroupOnlinePushCommand"
+	CallbackBeforeAddFriendCommand                     = "callbackBeforeAddFriendCommand"
+	CallbackBeforeUpdateUserInfoCommand                = "callbackBeforeUpdateUserInfoCommand"
+	CallbackBeforeCreateGroupCommand                   = "callbackBeforeCreateGroupCommand"
+	CallbackAfterCreateGroupCommand                    = "callbackAfterCreateGroupCommand"
+	CallbackBeforeMembersJoinGroupCommand              = "callbackBeforeMembersJoinGroupCommand"
+	CallbackBeforeSetGroupMemberInfoCommand            = "callbackBeforeSetGroupMemberInfoCommand"
+	CallbackAfterSetGroupMemberInfoCommand             = "callbackAfterSetGroupMemberInfoCommand"
+	CallbackBeforeCreateSingleChatConversationsCommand = "callbackBeforeCreateSingleChatConversationsCommand"
+	CallbackAfterCreateSingleChatConversationsCommand  = "callbackAfterCreateSingleChatConversationsCommand"
+	CallbackBeforeCreateGroupChatConversationsCommand  = "callbackBeforeCreateGroupChatConversationsCommand"
+	CallbackAfterCreateGroupChatConversationsCommand   = "callbackAfterCreateGroupChatConversationsCommand"
 )

pkg/callbackstruct/conversation.go

@@ -0,0 +1,91 @@
+package callbackstruct
+
+type CallbackBeforeCreateSingleChatConversationsReq struct {
+	CallbackCommand  `json:"callbackCommand"`
+	OwnerUserID      string `json:"owner_user_id"`
+	ConversationID   string `json:"conversation_id"`
+	ConversationType int32  `json:"conversation_type"`
+	UserID           string `json:"user_id"`
+	RecvMsgOpt       int32  `json:"recv_msg_opt"`
+	IsPinned         bool   `json:"is_pinned"`
+	IsPrivateChat    bool   `json:"is_private_chat"`
+	BurnDuration     int32  `json:"burn_duration"`
+	GroupAtType      int32  `json:"group_at_type"`
+	AttachedInfo     string `json:"attached_info"`
+	Ex               string `json:"ex"`
+}
+
+type CallbackBeforeCreateSingleChatConversationsResp struct {
+	CommonCallbackResp
+	RecvMsgOpt    *int32  `json:"recv_msg_opt"`
+	IsPinned      *bool   `json:"is_pinned"`
+	IsPrivateChat *bool   `json:"is_private_chat"`
+	BurnDuration  *int32  `json:"burn_duration"`
+	GroupAtType   *int32  `json:"group_at_type"`
+	AttachedInfo  *string `json:"attached_info"`
+	Ex            *string `json:"ex"`
+}
+
+type CallbackAfterCreateSingleChatConversationsReq struct {
+	CallbackCommand  `json:"callbackCommand"`
+	OwnerUserID      string `json:"owner_user_id"`
+	ConversationID   string `json:"conversation_id"`
+	ConversationType int32  `json:"conversation_type"`
+	UserID           string `json:"user_id"`
+	RecvMsgOpt       int32  `json:"recv_msg_opt"`
+	IsPinned         bool   `json:"is_pinned"`
+	IsPrivateChat    bool   `json:"is_private_chat"`
+	BurnDuration     int32  `json:"burn_duration"`
+	GroupAtType      int32  `json:"group_at_type"`
+	AttachedInfo     string `json:"attached_info"`
+	Ex               string `json:"ex"`
+}
+
+type CallbackAfterCreateSingleChatConversationsResp struct {
+	CommonCallbackResp
+}
+
+type CallbackBeforeCreateGroupChatConversationsReq struct {
+	CallbackCommand  `json:"callbackCommand"`
+	OwnerUserID      string `json:"owner_user_id"`
+	ConversationID   string `json:"conversation_id"`
+	ConversationType int32  `json:"conversation_type"`
+	GroupID          string `json:"group_id"`
+	RecvMsgOpt       int32  `json:"recv_msg_opt"`
+	IsPinned         bool   `json:"is_pinned"`
+	IsPrivateChat    bool   `json:"is_private_chat"`
+	BurnDuration     int32  `json:"burn_duration"`
+	GroupAtType      int32  `json:"group_at_type"`
+	AttachedInfo     string `json:"attached_info"`
+	Ex               string `json:"ex"`
+}
+
+type CallbackBeforeCreateGroupChatConversationsResp struct {
+	CommonCallbackResp
+	RecvMsgOpt    *int32  `json:"recv_msg_opt"`
+	IsPinned      *bool   `json:"is_pinned"`
+	IsPrivateChat *bool   `json:"is_private_chat"`
+	BurnDuration  *int32  `json:"burn_duration"`
+	GroupAtType   *int32  `json:"group_at_type"`
+	AttachedInfo  *string `json:"attached_info"`
+	Ex            *string `json:"ex"`
+}
+
+type CallbackAfterCreateGroupChatConversationsReq struct {
+	CallbackCommand  `json:"callbackCommand"`
+	OwnerUserID      string `json:"owner_user_id"`
+	ConversationID   string `json:"conversation_id"`
+	ConversationType int32  `json:"conversation_type"`
+	GroupID          string `json:"group_id"`
+	RecvMsgOpt       int32  `json:"recv_msg_opt"`
+	IsPinned         bool   `json:"is_pinned"`
+	IsPrivateChat    bool   `json:"is_private_chat"`
+	BurnDuration     int32  `json:"burn_duration"`
+	GroupAtType      int32  `json:"group_at_type"`
+	AttachedInfo     string `json:"attached_info"`
+	Ex               string `json:"ex"`
+}
+
+type CallbackAfterCreateGroupChatConversationsResp struct {
+	CommonCallbackResp
+}

pkg/common/cmd/conversation.go

@@ -41,6 +41,7 @@ func NewConversationRpcCmd() *ConversationRpcCmd {
 		config.MongodbConfigFileName:            &conversationConfig.MongodbConfig,
 		config.ShareFileName:                    &conversationConfig.Share,
 		config.NotificationFileName:             &conversationConfig.NotificationConfig,
+		config.WebhooksConfigFileName:           &conversationConfig.WebhooksConfig,
 		config.LocalCacheConfigFileName:         &conversationConfig.LocalCacheConfig,
 		config.DiscoveryConfigFilename:          &conversationConfig.Discovery,
 	}
@@ -67,6 +68,7 @@ func (a *ConversationRpcCmd) runE() error {
 			a.conversationConfig.NotificationConfig.GetConfigFileName(),
 			a.conversationConfig.Share.GetConfigFileName(),
 			a.conversationConfig.LocalCacheConfig.GetConfigFileName(),
+			a.conversationConfig.WebhooksConfig.GetConfigFileName(),
 			a.conversationConfig.Discovery.GetConfigFileName(),
 		}, nil,
 		conversation.Start)

pkg/common/storage/controller/conversation.go

@@ -22,7 +22,6 @@ import (
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/db/pagination"
 	"github.com/openimsdk/tools/db/tx"
@@ -47,8 +46,11 @@ type ConversationDatabase interface {
 	// SetUsersConversationFieldTx updates a specific field for multiple users' conversations, creating new conversations if they do not exist, or updates them otherwise. This operation is
 	// transactional.
 	SetUsersConversationFieldTx(ctx context.Context, userIDs []string, conversation *relationtb.Conversation, fieldMap map[string]any) error
+	// UpdateUserConversations updates all conversations related to a specified user.
+	// This function does NOT update the user's own conversations but rather the conversations where this user is involved (e.g., other users' conversations referencing this user).
+	UpdateUserConversations(ctx context.Context, userID string, args map[string]any) error
 	// CreateGroupChatConversation creates a group chat conversation for the specified group ID and user IDs.
-	CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error
+	CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string, conversations *relationtb.Conversation) error
 	// GetConversationIDs retrieves conversation IDs for a given user.
 	GetConversationIDs(ctx context.Context, userID string) ([]string, error)
 	// GetUserConversationIDsHash gets the hash of conversation IDs for a given user.
@@ -146,6 +148,18 @@ func (c *conversationDatabase) SetUsersConversationFieldTx(ctx context.Context,
 	})
 }
 
+func (c *conversationDatabase) UpdateUserConversations(ctx context.Context, userID string, args map[string]any) error {
+	conversations, err := c.conversationDB.UpdateUserConversations(ctx, userID, args)
+	if err != nil {
+		return err
+	}
+	cache := c.cache.CloneConversationCache()
+	for _, conversation := range conversations {
+		cache = cache.DelUsersConversation(conversation.ConversationID, conversation.OwnerUserID).DelConversationVersionUserIDs(conversation.OwnerUserID)
+	}
+	return cache.ChainExecDel(ctx)
+}
+
 func (c *conversationDatabase) UpdateUsersConversationField(ctx context.Context, userIDs []string, conversationID string, args map[string]any) error {
 	_, err := c.conversationDB.UpdateByMap(ctx, userIDs, conversationID, args)
 	if err != nil {
@@ -298,10 +312,10 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs
 //	return c.cache.GetSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID)
 //}
 
-func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error {
+func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string, conversation *relationtb.Conversation) error {
 	return c.tx.Transaction(ctx, func(ctx context.Context) error {
 		cache := c.cache.CloneConversationCache()
-		conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID)
+		conversationID := conversation.ConversationID
 		existConversationUserIDs, err := c.conversationDB.FindUserID(ctx, userIDs, []string{conversationID})
 		if err != nil {
 			return err
@@ -309,7 +323,15 @@ func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context,
 		notExistUserIDs := stringutil.DifferenceString(userIDs, existConversationUserIDs)
 		var conversations []*relationtb.Conversation
 		for _, v := range notExistUserIDs {
-			conversation := relationtb.Conversation{ConversationType: constant.ReadGroupChatType, GroupID: groupID, OwnerUserID: v, ConversationID: conversationID}
+			conversation := relationtb.Conversation{
+				ConversationType: conversation.ConversationType, GroupID: groupID, OwnerUserID: v, ConversationID: conversationID,
+				// the parameters have default value
+				RecvMsgOpt: conversation.RecvMsgOpt, IsPinned: conversation.IsPinned, IsPrivateChat: conversation.IsPrivateChat,
+				BurnDuration: conversation.BurnDuration, GroupAtType: conversation.GroupAtType, AttachedInfo: conversation.AttachedInfo,
+				Ex: conversation.Ex, MaxSeq: conversation.MaxSeq, MinSeq: conversation.MinSeq, CreateTime: conversation.CreateTime,
+				MsgDestructTime: conversation.MsgDestructTime, IsMsgDestruct: conversation.IsMsgDestruct, LatestMsgDestructTime: conversation.LatestMsgDestructTime,
+			}
+
 			conversations = append(conversations, &conversation)
 			cache = cache.DelConversations(v, conversationID).DelConversationNotReceiveMessageUserIDs(conversationID)
 		}
@@ -320,7 +342,7 @@ func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context,
 				return err
 			}
 		}
-		_, err = c.conversationDB.UpdateByMap(ctx, existConversationUserIDs, conversationID, map[string]any{"max_seq": 0})
+		_, err = c.conversationDB.UpdateByMap(ctx, existConversationUserIDs, conversationID, map[string]any{"max_seq": conversation.MaxSeq})
 		if err != nil {
 			return err
 		}

pkg/common/storage/database/conversation.go

@@ -24,6 +24,7 @@ import (
 type Conversation interface {
 	Create(ctx context.Context, conversations []*model.Conversation) (err error)
 	UpdateByMap(ctx context.Context, userIDs []string, conversationID string, args map[string]any) (rows int64, err error)
+	UpdateUserConversations(ctx context.Context, userID string, args map[string]any) ([]*model.Conversation, error)
 	Update(ctx context.Context, conversation *model.Conversation) (err error)
 	Find(ctx context.Context, ownerUserID string, conversationIDs []string) (conversations []*model.Conversation, err error)
 	FindUserID(ctx context.Context, userIDs []string, conversationIDs []string) ([]string, error)

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

@@ -21,23 +21,32 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+
 	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/db/mongoutil"
 	"github.com/openimsdk/tools/db/pagination"
 	"github.com/openimsdk/tools/errs"
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
 func NewConversationMongo(db *mongo.Database) (*ConversationMgo, error) {
 	coll := db.Collection(database.ConversationName)
-	_, err := coll.Indexes().CreateOne(context.Background(), mongo.IndexModel{
-		Keys: bson.D{
-			{Key: "owner_user_id", Value: 1},
-			{Key: "conversation_id", Value: 1},
+	_, err := coll.Indexes().CreateMany(context.Background(), []mongo.IndexModel{
+		{
+			Keys: bson.D{
+				{Key: "owner_user_id", Value: 1},
+				{Key: "conversation_id", Value: 1},
+			},
+			Options: options.Index().SetUnique(true),
+		},
+		{
+			Keys: bson.D{
+				{Key: "user_id", Value: 1},
+			},
+			Options: options.Index(),
 		},
-		Options: options.Index().SetUnique(true),
 	})
 	if err != nil {
 		return nil, errs.Wrap(err)
@@ -101,6 +110,38 @@ func (c *ConversationMgo) UpdateByMap(ctx context.Context, userIDs []string, con
 	return rows, nil
 }
 
+func (c *ConversationMgo) UpdateUserConversations(ctx context.Context, userID string, args map[string]any) ([]*model.Conversation, error) {
+	if len(args) == 0 {
+		return nil, nil
+	}
+	filter := bson.M{
+		"user_id": userID,
+	}
+
+	conversations, err := mongoutil.Find[*model.Conversation](ctx, c.coll, filter, options.Find().SetProjection(bson.M{"_id": 0, "owner_user_id": 1, "conversation_id": 1}))
+	if err != nil {
+		return nil, err
+	}
+	err = mongoutil.IncrVersion(func() error {
+		_, err := mongoutil.UpdateMany(ctx, c.coll, filter, bson.M{"$set": args})
+		if err != nil {
+			return err
+		}
+		return nil
+	}, func() error {
+		for _, conversation := range conversations {
+			if err := c.version.IncrVersion(ctx, conversation.OwnerUserID, []string{conversation.ConversationID}, model.VersionStateUpdate); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return conversations, nil
+}
+
 func (c *ConversationMgo) Update(ctx context.Context, conversation *model.Conversation) (err error) {
 	return mongoutil.IncrVersion(func() error {
 		return mongoutil.UpdateOne(ctx, c.coll, bson.M{"owner_user_id": conversation.OwnerUserID, "conversation_id": conversation.ConversationID}, bson.M{"$set": conversation}, true)