feat: use robot to migrate code

Signed-off-by: kubbot & kubecub <3293172751ysy@gmail.com>

pkg/common/db/controller/auth.go

@@ -0,0 +1,62 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/tokenverify"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"github.com/golang-jwt/jwt/v4"
+)
+
+type AuthDatabase interface {
+	//结果为空 不返回错误
+	GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error)
+	//创建token
+	CreateToken(ctx context.Context, userID string, platformID int) (string, error)
+}
+
+type authDatabase struct {
+	cache cache.MsgModel
+
+	accessSecret string
+	accessExpire int64
+}
+
+func NewAuthDatabase(cache cache.MsgModel, accessSecret string, accessExpire int64) AuthDatabase {
+	return &authDatabase{cache: cache, accessSecret: accessSecret, accessExpire: accessExpire}
+}
+
+// 结果为空 不返回错误
+func (a *authDatabase) GetTokensWithoutError(ctx context.Context, userID string, platformID int) (map[string]int, error) {
+	return a.cache.GetTokensWithoutError(ctx, userID, platformID)
+}
+
+// 创建token
+func (a *authDatabase) CreateToken(ctx context.Context, userID string, platformID int) (string, error) {
+	tokens, err := a.cache.GetTokensWithoutError(ctx, userID, platformID)
+	if err != nil {
+		return "", err
+	}
+	var deleteTokenKey []string
+	for k, v := range tokens {
+		_, err = tokenverify.GetClaimFromToken(k)
+		if err != nil || v != constant.NormalToken {
+			deleteTokenKey = append(deleteTokenKey, k)
+		}
+	}
+	if len(deleteTokenKey) != 0 {
+		err := a.cache.DeleteTokenByUidPid(ctx, userID, platformID, deleteTokenKey)
+		if err != nil {
+			return "", err
+		}
+	}
+	claims := tokenverify.BuildClaims(userID, platformID, a.accessExpire)
+	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
+	tokenString, err := token.SignedString([]byte(a.accessSecret))
+	if err != nil {
+		return "", utils.Wrap(err, "")
+	}
+	return tokenString, a.cache.AddTokenFlag(ctx, userID, platformID, tokenString, constant.NormalToken)
+}

pkg/common/db/controller/black.go

@@ -0,0 +1,78 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+)
+
+type BlackDatabase interface {
+	// Create 增加黑名单
+	Create(ctx context.Context, blacks []*relation.BlackModel) (err error)
+	// Delete 删除黑名单
+	Delete(ctx context.Context, blacks []*relation.BlackModel) (err error)
+	// FindOwnerBlacks 获取黑名单列表
+	FindOwnerBlacks(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (blacks []*relation.BlackModel, total int64, err error)
+	FindBlackIDs(ctx context.Context, ownerUserID string) (blackIDs []string, err error)
+	// CheckIn 检查user2是否在user1的黑名单列表中(inUser1Blacks==true) 检查user1是否在user2的黑名单列表中(inUser2Blacks==true)
+	CheckIn(ctx context.Context, userID1, userID2 string) (inUser1Blacks bool, inUser2Blacks bool, err error)
+}
+
+type blackDatabase struct {
+	black relation.BlackModelInterface
+	cache cache.BlackCache
+}
+
+func NewBlackDatabase(black relation.BlackModelInterface, cache cache.BlackCache) BlackDatabase {
+	return &blackDatabase{black, cache}
+}
+
+// Create 增加黑名单
+func (b *blackDatabase) Create(ctx context.Context, blacks []*relation.BlackModel) (err error) {
+	if err := b.black.Create(ctx, blacks); err != nil {
+		return err
+	}
+	return b.deleteBlackIDsCache(ctx, blacks)
+}
+
+// Delete 删除黑名单
+func (b *blackDatabase) Delete(ctx context.Context, blacks []*relation.BlackModel) (err error) {
+	if err := b.black.Delete(ctx, blacks); err != nil {
+		return err
+	}
+	return b.deleteBlackIDsCache(ctx, blacks)
+}
+
+func (b *blackDatabase) deleteBlackIDsCache(ctx context.Context, blacks []*relation.BlackModel) (err error) {
+	cache := b.cache.NewCache()
+	for _, black := range blacks {
+		cache = cache.DelBlackIDs(ctx, black.OwnerUserID)
+	}
+	return cache.ExecDel(ctx)
+}
+
+// FindOwnerBlacks 获取黑名单列表
+func (b *blackDatabase) FindOwnerBlacks(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (blacks []*relation.BlackModel, total int64, err error) {
+	return b.black.FindOwnerBlacks(ctx, ownerUserID, pageNumber, showNumber)
+}
+
+// CheckIn 检查user2是否在user1的黑名单列表中(inUser1Blacks==true) 检查user1是否在user2的黑名单列表中(inUser2Blacks==true)
+func (b *blackDatabase) CheckIn(ctx context.Context, userID1, userID2 string) (inUser1Blacks bool, inUser2Blacks bool, err error) {
+	userID1BlackIDs, err := b.cache.GetBlackIDs(ctx, userID1)
+	if err != nil {
+		return
+	}
+	userID2BlackIDs, err := b.cache.GetBlackIDs(ctx, userID2)
+	if err != nil {
+		return
+	}
+	log.ZDebug(ctx, "blackIDs", "user1BlackIDs", userID1BlackIDs, "user2BlackIDs", userID2BlackIDs)
+	return utils.IsContain(userID2, userID1BlackIDs), utils.IsContain(userID1, userID2BlackIDs), nil
+}
+
+func (b *blackDatabase) FindBlackIDs(ctx context.Context, ownerUserID string) (blackIDs []string, err error) {
+	return b.cache.GetBlackIDs(ctx, ownerUserID)
+}

pkg/common/db/controller/chatlog.go

@@ -0,0 +1,27 @@
+package controller
+
+import (
+	relationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
+)
+
+type ChatLogDatabase interface {
+	CreateChatLog(msg *pbMsg.MsgDataToMQ) error
+	GetChatLog(chatLog *relationTb.ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []relationTb.ChatLogModel, error)
+}
+
+func NewChatLogDatabase(chatLogModelInterface relationTb.ChatLogModelInterface) ChatLogDatabase {
+	return &chatLogDatabase{chatLogModel: chatLogModelInterface}
+}
+
+type chatLogDatabase struct {
+	chatLogModel relationTb.ChatLogModelInterface
+}
+
+func (c *chatLogDatabase) CreateChatLog(msg *pbMsg.MsgDataToMQ) error {
+	return c.chatLogModel.Create(msg)
+}
+
+func (c *chatLogDatabase) GetChatLog(chatLog *relationTb.ChatLogModel, pageNumber, showNumber int32, contentTypes []int32) (int64, []relationTb.ChatLogModel, error) {
+	return c.chatLogModel.GetChatLog(chatLog, pageNumber, showNumber, contentTypes)
+}

pkg/common/db/controller/conversation.go

@@ -0,0 +1,271 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	relationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+)
+
+type ConversationDatabase interface {
+	//UpdateUserConversationFiled 更新用户该会话的属性信息
+	UpdateUsersConversationFiled(ctx context.Context, userIDs []string, conversationID string, args map[string]interface{}) error
+	//CreateConversation 创建一批新的会话
+	CreateConversation(ctx context.Context, conversations []*relationTb.ConversationModel) error
+	//SyncPeerUserPrivateConversation 同步对端私聊会话内部保证事务操作
+	SyncPeerUserPrivateConversationTx(ctx context.Context, conversation []*relationTb.ConversationModel) error
+	//FindConversations 根据会话ID获取某个用户的多个会话
+	FindConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*relationTb.ConversationModel, error)
+	//FindRecvMsgNotNotifyUserIDs 获取超级大群开启免打扰的用户ID
+	FindRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error)
+	//GetUserAllConversation 获取一个用户在服务器上所有的会话
+	GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*relationTb.ConversationModel, error)
+	//SetUserConversations 设置用户多个会话属性，如果会话不存在则创建，否则更新,内部保证原子性
+	SetUserConversations(ctx context.Context, ownerUserID string, conversations []*relationTb.ConversationModel) error
+	//SetUsersConversationFiledTx 设置多个用户会话关于某个字段的更新操作，如果会话不存在则创建，否则更新，内部保证事务操作
+	SetUsersConversationFiledTx(ctx context.Context, userIDs []string, conversation *relationTb.ConversationModel, filedMap map[string]interface{}) error
+	CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error
+	GetConversationIDs(ctx context.Context, userID string) ([]string, error)
+	GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error)
+	GetAllConversationIDs(ctx context.Context) ([]string, error)
+	GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
+	GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationTb.ConversationModel, error)
+}
+
+func NewConversationDatabase(conversation relationTb.ConversationModelInterface, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
+	return &conversationDatabase{
+		conversationDB: conversation,
+		cache:          cache,
+		tx:             tx,
+	}
+}
+
+type conversationDatabase struct {
+	conversationDB relationTb.ConversationModelInterface
+	cache          cache.ConversationCache
+	tx             tx.Tx
+}
+
+func (c *conversationDatabase) SetUsersConversationFiledTx(ctx context.Context, userIDs []string, conversation *relationTb.ConversationModel, filedMap map[string]interface{}) (err error) {
+	cache := c.cache.NewCache()
+	if err := c.tx.Transaction(func(tx any) error {
+		conversationTx := c.conversationDB.NewTx(tx)
+		haveUserIDs, err := conversationTx.FindUserID(ctx, userIDs, []string{conversation.ConversationID})
+		if err != nil {
+			return err
+		}
+		if len(haveUserIDs) > 0 {
+			_, err = conversationTx.UpdateByMap(ctx, haveUserIDs, conversation.ConversationID, filedMap)
+			if err != nil {
+				return err
+			}
+			cache = cache.DelUsersConversation(conversation.ConversationID, haveUserIDs...)
+			if _, ok := filedMap["has_read_seq"]; ok {
+				for _, userID := range haveUserIDs {
+					cache = cache.DelUserAllHasReadSeqs(userID, conversation.ConversationID)
+				}
+			}
+		}
+		NotUserIDs := utils.DifferenceString(haveUserIDs, userIDs)
+		log.ZDebug(ctx, "SetUsersConversationFiledTx", "NotUserIDs", NotUserIDs, "haveUserIDs", haveUserIDs, "userIDs", userIDs)
+		var conversations []*relationTb.ConversationModel
+		for _, v := range NotUserIDs {
+			temp := new(relationTb.ConversationModel)
+			if err := utils.CopyStructFields(temp, conversation); err != nil {
+				return err
+			}
+			temp.OwnerUserID = v
+			conversations = append(conversations, temp)
+
+		}
+		if len(conversations) > 0 {
+			err = conversationTx.Create(ctx, conversations)
+			if err != nil {
+				return err
+			}
+			cache = cache.DelConversationIDs(NotUserIDs...).DelUserConversationIDsHash(NotUserIDs...)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (c *conversationDatabase) UpdateUsersConversationFiled(ctx context.Context, userIDs []string, conversationID string, args map[string]interface{}) error {
+	_, err := c.conversationDB.UpdateByMap(ctx, userIDs, conversationID, args)
+	if err != nil {
+		return err
+	}
+	return c.cache.DelUsersConversation(conversationID, userIDs...).ExecDel(ctx)
+}
+
+func (c *conversationDatabase) CreateConversation(ctx context.Context, conversations []*relationTb.ConversationModel) error {
+	if err := c.conversationDB.Create(ctx, conversations); err != nil {
+		return err
+	}
+	var userIDs []string
+	cache := c.cache.NewCache()
+	for _, conversation := range conversations {
+		cache = cache.DelConvsersations(conversation.OwnerUserID, conversation.ConversationID)
+		userIDs = append(userIDs, conversation.OwnerUserID)
+	}
+	return cache.DelConversationIDs(userIDs...).DelUserConversationIDsHash(userIDs...).ExecDel(ctx)
+}
+
+func (c *conversationDatabase) SyncPeerUserPrivateConversationTx(ctx context.Context, conversations []*relationTb.ConversationModel) error {
+	cache := c.cache.NewCache()
+	if err := c.tx.Transaction(func(tx any) error {
+		conversationTx := c.conversationDB.NewTx(tx)
+		for _, conversation := range conversations {
+			for _, v := range [][2]string{{conversation.OwnerUserID, conversation.UserID}, {conversation.UserID, conversation.OwnerUserID}} {
+				haveUserIDs, err := conversationTx.FindUserID(ctx, []string{v[0]}, []string{conversation.ConversationID})
+				if err != nil {
+					return err
+				}
+				if len(haveUserIDs) > 0 {
+					_, err := conversationTx.UpdateByMap(ctx, []string{v[0]}, conversation.ConversationID, map[string]interface{}{"is_private_chat": conversation.IsPrivateChat})
+					if err != nil {
+						return err
+					}
+					cache = cache.DelUsersConversation(conversation.ConversationID, v[0])
+				} else {
+					newConversation := *conversation
+					newConversation.OwnerUserID = v[0]
+					newConversation.UserID = v[1]
+					newConversation.ConversationID = conversation.ConversationID
+					newConversation.IsPrivateChat = conversation.IsPrivateChat
+					if err := conversationTx.Create(ctx, []*relationTb.ConversationModel{&newConversation}); err != nil {
+						return err
+					}
+					cache = cache.DelConversationIDs(v[0]).DelUserConversationIDsHash(v[0])
+				}
+			}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return c.cache.ExecDel(ctx)
+}
+
+func (c *conversationDatabase) FindConversations(ctx context.Context, ownerUserID string, conversationIDs []string) ([]*relationTb.ConversationModel, error) {
+	return c.cache.GetConversations(ctx, ownerUserID, conversationIDs)
+}
+
+func (c *conversationDatabase) GetConversation(ctx context.Context, ownerUserID string, conversationID string) (*relationTb.ConversationModel, error) {
+	return c.cache.GetConversation(ctx, ownerUserID, conversationID)
+}
+
+func (c *conversationDatabase) GetUserAllConversation(ctx context.Context, ownerUserID string) ([]*relationTb.ConversationModel, error) {
+	return c.cache.GetUserAllConversations(ctx, ownerUserID)
+}
+
+func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUserID string, conversations []*relationTb.ConversationModel) error {
+	cache := c.cache.NewCache()
+	if err := c.tx.Transaction(func(tx any) error {
+		var conversationIDs []string
+		for _, conversation := range conversations {
+			conversationIDs = append(conversationIDs, conversation.ConversationID)
+		}
+		conversationTx := c.conversationDB.NewTx(tx)
+		existConversations, err := conversationTx.Find(ctx, ownerUserID, conversationIDs)
+		if err != nil {
+			return err
+		}
+		if len(existConversations) > 0 {
+			for _, conversation := range conversations {
+				err = conversationTx.Update(ctx, conversation)
+				if err != nil {
+					return err
+				}
+			}
+		}
+		var existConversationIDs []string
+		for _, conversation := range existConversations {
+			existConversationIDs = append(existConversationIDs, conversation.ConversationID)
+		}
+
+		var notExistConversations []*relationTb.ConversationModel
+		for _, conversation := range conversations {
+			if !utils.IsContain(conversation.ConversationID, existConversationIDs) {
+				notExistConversations = append(notExistConversations, conversation)
+			}
+		}
+		if len(notExistConversations) > 0 {
+			err = c.conversationDB.Create(ctx, notExistConversations)
+			if err != nil {
+				return err
+			}
+			cache = cache.DelConversationIDs(ownerUserID).DelUserConversationIDsHash(ownerUserID)
+		}
+		cache = cache.DelConvsersations(ownerUserID, existConversationIDs...)
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (c *conversationDatabase) FindRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error) {
+	return c.cache.GetSuperGroupRecvMsgNotNotifyUserIDs(ctx, groupID)
+}
+
+func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context, groupID string, userIDs []string) error {
+	cache := c.cache.NewCache()
+	conversationID := utils.GetConversationIDBySessionType(constant.SuperGroupChatType, groupID)
+	if err := c.tx.Transaction(func(tx any) error {
+		existConversationUserIDs, err := c.conversationDB.FindUserID(ctx, userIDs, []string{conversationID})
+		if err != nil {
+			return err
+		}
+		notExistUserIDs := utils.DifferenceString(userIDs, existConversationUserIDs)
+		var conversations []*relationTb.ConversationModel
+		for _, v := range notExistUserIDs {
+			conversation := relationTb.ConversationModel{ConversationType: constant.SuperGroupChatType, GroupID: groupID, OwnerUserID: v, ConversationID: conversationID}
+			conversations = append(conversations, &conversation)
+		}
+		cache = cache.DelConversationIDs(notExistUserIDs...).DelUserConversationIDsHash(notExistUserIDs...)
+		if len(conversations) > 0 {
+			err = c.conversationDB.Create(ctx, conversations)
+			if err != nil {
+				return err
+			}
+		}
+		_, err = c.conversationDB.UpdateByMap(ctx, existConversationUserIDs, conversationID, map[string]interface{}{"max_seq": 0})
+		if err != nil {
+			return err
+		}
+		for _, v := range existConversationUserIDs {
+			cache = cache.DelConvsersations(v, conversationID)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (c *conversationDatabase) GetConversationIDs(ctx context.Context, userID string) ([]string, error) {
+	return c.cache.GetUserConversationIDs(ctx, userID)
+}
+
+func (c *conversationDatabase) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) {
+	return c.cache.GetUserConversationIDsHash(ctx, ownerUserID)
+}
+
+func (c *conversationDatabase) GetAllConversationIDs(ctx context.Context) ([]string, error) {
+	return c.conversationDB.GetAllConversationIDs(ctx)
+}
+
+func (c *conversationDatabase) GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error) {
+	return c.cache.GetUserAllHasReadSeqs(ctx, ownerUserID)
+}
+
+func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationTb.ConversationModel, error) {
+	return c.conversationDB.GetConversationsByConversationID(ctx, conversationIDs)
+}

pkg/common/db/controller/extend_msg.go

@@ -0,0 +1,58 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
+)
+
+// for mongoDB
+type ExtendMsgDatabase interface {
+	CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error
+	GetAllExtendMsgSet(ctx context.Context, ID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error)
+	GetExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error)
+	InsertExtendMsg(ctx context.Context, conversationID string, sessionType int32, msg *unRelationTb.ExtendMsgModel) error
+	InsertOrUpdateReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*unRelationTb.KeyValueModel) error
+	DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*unRelationTb.KeyValueModel) error
+	GetExtendMsg(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error)
+}
+
+type extendMsgDatabase struct {
+	database unRelationTb.ExtendMsgSetModelInterface
+	cache    cache.ExtendMsgSetCache
+	ctxTx    tx.CtxTx
+}
+
+func NewExtendMsgDatabase(extendMsgModel unRelationTb.ExtendMsgSetModelInterface, cache cache.ExtendMsgSetCache, ctxTx tx.CtxTx) ExtendMsgDatabase {
+	return &extendMsgDatabase{database: extendMsgModel, cache: cache, ctxTx: ctxTx}
+}
+
+func (e *extendMsgDatabase) CreateExtendMsgSet(ctx context.Context, set *unRelationTb.ExtendMsgSetModel) error {
+	return e.database.CreateExtendMsgSet(ctx, set)
+}
+
+func (e *extendMsgDatabase) GetAllExtendMsgSet(ctx context.Context, conversationID string, opts *unRelationTb.GetAllExtendMsgSetOpts) (sets []*unRelationTb.ExtendMsgSetModel, err error) {
+	return e.database.GetAllExtendMsgSet(ctx, conversationID, opts)
+}
+
+func (e *extendMsgDatabase) GetExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, maxMsgUpdateTime int64) (*unRelationTb.ExtendMsgSetModel, error) {
+	return e.database.GetExtendMsgSet(ctx, conversationID, sessionType, maxMsgUpdateTime)
+}
+
+func (e *extendMsgDatabase) InsertExtendMsg(ctx context.Context, conversationID string, sessionType int32, msg *unRelationTb.ExtendMsgModel) error {
+	return e.database.InsertExtendMsg(ctx, conversationID, sessionType, msg)
+}
+
+func (e *extendMsgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*unRelationTb.KeyValueModel) error {
+	return e.database.InsertOrUpdateReactionExtendMsgSet(ctx, conversationID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
+}
+
+func (e *extendMsgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*unRelationTb.KeyValueModel) error {
+	return e.database.DeleteReactionExtendMsgSet(ctx, conversationID, sessionType, clientMsgID, msgFirstModifyTime, reactionExtensionList)
+}
+
+func (e *extendMsgDatabase) GetExtendMsg(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (extendMsg *unRelationTb.ExtendMsgModel, err error) {
+	return e.cache.GetExtendMsg(ctx, conversationID, sessionType, clientMsgID, maxMsgUpdateTime)
+}

pkg/common/db/controller/friend.go

@@ -0,0 +1,252 @@
+package controller
+
+import (
+	"context"
+	"time"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"gorm.io/gorm"
+)
+
+type FriendDatabase interface {
+	// 检查user2是否在user1的好友列表中(inUser1Friends==true) 检查user1是否在user2的好友列表中(inUser2Friends==true)
+	CheckIn(ctx context.Context, user1, user2 string) (inUser1Friends bool, inUser2Friends bool, err error)
+	// 增加或者更新好友申请
+	AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error)
+	// 先判断是否在好友表，如果在则不插入
+	BecomeFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, addSource int32) (err error)
+	// 拒绝好友申请
+	RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error)
+	// 同意好友申请
+	AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error)
+	// 删除好友
+	Delete(ctx context.Context, ownerUserID string, friendUserIDs []string) (err error)
+	// 更新好友备注
+	UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error)
+	// 获取ownerUserID的好友列表
+	PageOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error)
+	// friendUserID在哪些人的好友列表中
+	PageInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error)
+	// 获取我发出去的好友申请
+	PageFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error)
+	// 获取我收到的的好友申请
+	PageFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error)
+	// 获取某人指定好友的信息
+	FindFriendsWithError(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error)
+	FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error)
+}
+
+type friendDatabase struct {
+	friend        relation.FriendModelInterface
+	friendRequest relation.FriendRequestModelInterface
+	tx            tx.Tx
+	cache         cache.FriendCache
+}
+
+func NewFriendDatabase(friend relation.FriendModelInterface, friendRequest relation.FriendRequestModelInterface, cache cache.FriendCache, tx tx.Tx) FriendDatabase {
+	return &friendDatabase{friend: friend, friendRequest: friendRequest, cache: cache, tx: tx}
+}
+
+// ok 检查user2是否在user1的好友列表中(inUser1Friends==true) 检查user1是否在user2的好友列表中(inUser2Friends==true)
+func (f *friendDatabase) CheckIn(ctx context.Context, userID1, userID2 string) (inUser1Friends bool, inUser2Friends bool, err error) {
+	userID1FriendIDs, err := f.cache.GetFriendIDs(ctx, userID1)
+	if err != nil {
+		return
+	}
+	userID2FriendIDs, err := f.cache.GetFriendIDs(ctx, userID2)
+	if err != nil {
+		return
+	}
+	return utils.IsContain(userID2, userID1FriendIDs), utils.IsContain(userID1, userID2FriendIDs), nil
+}
+
+// 增加或者更新好友申请 如果之前有记录则更新，没有记录则新增
+func (f *friendDatabase) AddFriendRequest(ctx context.Context, fromUserID, toUserID string, reqMsg string, ex string) (err error) {
+	return f.tx.Transaction(func(tx any) error {
+		_, err := f.friendRequest.NewTx(tx).Take(ctx, fromUserID, toUserID)
+		//有db错误
+		if err != nil && errs.Unwrap(err) != gorm.ErrRecordNotFound {
+			return err
+		}
+		//无错误 则更新
+		if err == nil {
+			m := make(map[string]interface{}, 1)
+			m["handle_result"] = 0
+			m["handle_msg"] = ""
+			m["req_msg"] = reqMsg
+			m["ex"] = ex
+			m["create_time"] = time.Now()
+			if err := f.friendRequest.NewTx(tx).UpdateByMap(ctx, fromUserID, toUserID, m); err != nil {
+				return err
+			}
+			return nil
+		}
+		//gorm.ErrRecordNotFound 错误，则新增
+		if err := f.friendRequest.NewTx(tx).Create(ctx, []*relation.FriendRequestModel{{FromUserID: fromUserID, ToUserID: toUserID, ReqMsg: reqMsg, Ex: ex, CreateTime: time.Now(), HandleTime: time.Unix(0, 0)}}); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
+// (1)先判断是否在好友表 （在不在都不返回错误） (2)对于不在好友列表的 插入即可
+func (f *friendDatabase) BecomeFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, addSource int32) (err error) {
+	cache := f.cache.NewCache()
+	if err := f.tx.Transaction(func(tx any) error {
+		//先find 找出重复的 去掉重复的
+		fs1, err := f.friend.NewTx(tx).FindFriends(ctx, ownerUserID, friendUserIDs)
+		if err != nil {
+			return err
+		}
+		opUserID := mcontext.GetOperationID(ctx)
+		for _, v := range friendUserIDs {
+			fs1 = append(fs1, &relation.FriendModel{OwnerUserID: ownerUserID, FriendUserID: v, AddSource: addSource, OperatorUserID: opUserID})
+		}
+		fs11 := utils.DistinctAny(fs1, func(e *relation.FriendModel) string {
+			return e.FriendUserID
+		})
+
+		err = f.friend.NewTx(tx).Create(ctx, fs11)
+		if err != nil {
+			return err
+		}
+		fs2, err := f.friend.NewTx(tx).FindReversalFriends(ctx, ownerUserID, friendUserIDs)
+		if err != nil {
+			return err
+		}
+		var newFriendIDs []string
+		for _, v := range friendUserIDs {
+			fs2 = append(fs2, &relation.FriendModel{OwnerUserID: v, FriendUserID: ownerUserID, AddSource: addSource, OperatorUserID: opUserID})
+			newFriendIDs = append(newFriendIDs, v)
+		}
+		fs22 := utils.DistinctAny(fs2, func(e *relation.FriendModel) string {
+			return e.OwnerUserID
+		})
+		err = f.friend.NewTx(tx).Create(ctx, fs22)
+		if err != nil {
+			return err
+		}
+		newFriendIDs = append(newFriendIDs, ownerUserID)
+		cache = cache.DelFriendIDs(newFriendIDs...)
+		return nil
+	}); err != nil {
+		return nil
+	}
+	return cache.ExecDel(ctx)
+}
+
+// 拒绝好友申请 (1)检查是否有申请记录且为未处理状态 （没有记录返回错误） (2)修改申请记录 已拒绝
+func (f *friendDatabase) RefuseFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) {
+	fr, err := f.friendRequest.Take(ctx, friendRequest.FromUserID, friendRequest.ToUserID)
+	if err != nil {
+		return err
+	}
+	if fr.HandleResult != 0 {
+		return errs.ErrArgs.Wrap("the friend request has been processed")
+	}
+	friendRequest.HandleResult = constant.FriendResponseRefuse
+	friendRequest.HandleTime = time.Now()
+	err = f.friendRequest.Update(ctx, friendRequest)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// AgreeFriendRequest 同意好友申请  (1)检查是否有申请记录且为未处理状态 （没有记录返回错误） (2)检查是否好友（不返回错误）   (3) 建立双向好友关系（存在的忽略）
+func (f *friendDatabase) AgreeFriendRequest(ctx context.Context, friendRequest *relation.FriendRequestModel) (err error) {
+	return f.tx.Transaction(func(tx any) error {
+		fr, err := f.friendRequest.NewTx(tx).Take(ctx, friendRequest.FromUserID, friendRequest.ToUserID)
+		if err != nil {
+			return err
+		}
+		if fr.HandleResult != 0 {
+			return errs.ErrArgs.Wrap("the friend request has been processed")
+		}
+		friendRequest.HandlerUserID = mcontext.GetOpUserID(ctx)
+		friendRequest.HandleResult = constant.FriendResponseAgree
+		friendRequest.HandleTime = time.Now()
+		err = f.friendRequest.NewTx(tx).Update(ctx, friendRequest)
+		if err != nil {
+			return err
+		}
+		exists, err := f.friend.NewTx(tx).FindUserState(ctx, friendRequest.FromUserID, friendRequest.ToUserID)
+		if err != nil {
+			return err
+		}
+		existsMap := utils.SliceSet(utils.Slice(exists, func(friend *relation.FriendModel) [2]string {
+			return [...]string{friend.OwnerUserID, friend.FriendUserID} // 自己 - 好友
+		}))
+		var adds []*relation.FriendModel
+		if _, ok := existsMap[[...]string{friendRequest.ToUserID, friendRequest.FromUserID}]; !ok { // 自己 - 好友
+			adds = append(adds, &relation.FriendModel{OwnerUserID: friendRequest.ToUserID, FriendUserID: friendRequest.FromUserID, AddSource: int32(constant.BecomeFriendByApply), OperatorUserID: friendRequest.FromUserID})
+		}
+		if _, ok := existsMap[[...]string{friendRequest.FromUserID, friendRequest.ToUserID}]; !ok { // 好友 - 自己
+			adds = append(adds, &relation.FriendModel{OwnerUserID: friendRequest.FromUserID, FriendUserID: friendRequest.ToUserID, AddSource: int32(constant.BecomeFriendByApply), OperatorUserID: friendRequest.FromUserID})
+		}
+		if len(adds) > 0 {
+			if err := f.friend.NewTx(tx).Create(ctx, adds); err != nil {
+				return err
+			}
+		}
+		return f.cache.DelFriendIDs(friendRequest.ToUserID, friendRequest.FromUserID).ExecDel(ctx)
+	})
+}
+
+// 删除好友  外部判断是否好友关系
+func (f *friendDatabase) Delete(ctx context.Context, ownerUserID string, friendUserIDs []string) (err error) {
+	if err := f.friend.Delete(ctx, ownerUserID, friendUserIDs); err != nil {
+		return err
+	}
+	return f.cache.DelFriendIDs(append(friendUserIDs, ownerUserID)...).ExecDel(ctx)
+}
+
+// 更新好友备注 零值也支持
+func (f *friendDatabase) UpdateRemark(ctx context.Context, ownerUserID, friendUserID, remark string) (err error) {
+	if err := f.friend.UpdateRemark(ctx, ownerUserID, friendUserID, remark); err != nil {
+		return err
+	}
+	return f.cache.DelFriend(ownerUserID, friendUserID).ExecDel(ctx)
+}
+
+// 获取ownerUserID的好友列表 无结果不返回错误
+func (f *friendDatabase) PageOwnerFriends(ctx context.Context, ownerUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) {
+	return f.friend.FindOwnerFriends(ctx, ownerUserID, pageNumber, showNumber)
+}
+
+// friendUserID在哪些人的好友列表中
+func (f *friendDatabase) PageInWhoseFriends(ctx context.Context, friendUserID string, pageNumber, showNumber int32) (friends []*relation.FriendModel, total int64, err error) {
+	return f.friend.FindInWhoseFriends(ctx, friendUserID, pageNumber, showNumber)
+}
+
+// 获取我发出去的好友申请  无结果不返回错误
+func (f *friendDatabase) PageFriendRequestFromMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) {
+	return f.friendRequest.FindFromUserID(ctx, userID, pageNumber, showNumber)
+}
+
+// 获取我收到的的好友申请 无结果不返回错误
+func (f *friendDatabase) PageFriendRequestToMe(ctx context.Context, userID string, pageNumber, showNumber int32) (friends []*relation.FriendRequestModel, total int64, err error) {
+	return f.friendRequest.FindToUserID(ctx, userID, pageNumber, showNumber)
+}
+
+// 获取某人指定好友的信息 如果有好友不存在，也返回错误
+func (f *friendDatabase) FindFriendsWithError(ctx context.Context, ownerUserID string, friendUserIDs []string) (friends []*relation.FriendModel, err error) {
+	friends, err = f.friend.FindFriends(ctx, ownerUserID, friendUserIDs)
+	if err != nil {
+		return
+	}
+	if len(friends) != len(friendUserIDs) {
+		err = errs.ErrRecordNotFound.Wrap()
+	}
+	return
+}
+
+func (f *friendDatabase) FindFriendUserIDs(ctx context.Context, ownerUserID string) (friendUserIDs []string, err error) {
+	return f.cache.GetFriendIDs(ctx, ownerUserID)
+}

pkg/common/db/controller/group.go

@@ -0,0 +1,421 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/relation"
+	relationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"github.com/dtm-labs/rockscache"
+	"github.com/redis/go-redis/v9"
+	"go.mongodb.org/mongo-driver/mongo"
+	"gorm.io/gorm"
+)
+
+type GroupDatabase interface {
+	// Group
+	CreateGroup(ctx context.Context, groups []*relationTb.GroupModel, groupMembers []*relationTb.GroupMemberModel) error
+	TakeGroup(ctx context.Context, groupID string) (group *relationTb.GroupModel, err error)
+	FindGroup(ctx context.Context, groupIDs []string) (groups []*relationTb.GroupModel, err error)
+	SearchGroup(ctx context.Context, keyword string, pageNumber, showNumber int32) (uint32, []*relationTb.GroupModel, error)
+	UpdateGroup(ctx context.Context, groupID string, data map[string]any) error
+	DismissGroup(ctx context.Context, groupID string, deleteMember bool) error // 解散群，并删除群成员
+	GetGroupIDsByGroupType(ctx context.Context, groupType int) (groupIDs []string, err error)
+	// GroupMember
+	TakeGroupMember(ctx context.Context, groupID string, userID string) (groupMember *relationTb.GroupMemberModel, err error)
+	TakeGroupOwner(ctx context.Context, groupID string) (*relationTb.GroupMemberModel, error)
+	FindGroupMember(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32) ([]*relationTb.GroupMemberModel, error)
+	FindGroupMemberUserID(ctx context.Context, groupID string) ([]string, error)
+	FindGroupMemberNum(ctx context.Context, groupID string) (uint32, error)
+	FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error)
+	PageGroupRequest(ctx context.Context, groupIDs []string, pageNumber, showNumber int32) (uint32, []*relationTb.GroupRequestModel, error)
+	//PageGroupMember(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32, pageNumber, showNumber int32) (uint32, []*relationTb.GroupMemberModel, error)
+	PageGetJoinGroup(ctx context.Context, userID string, pageNumber, showNumber int32) (total uint32, totalGroupMembers []*relationTb.GroupMemberModel, err error)
+	PageGetGroupMember(ctx context.Context, groupID string, pageNumber, showNumber int32) (total uint32, totalGroupMembers []*relationTb.GroupMemberModel, err error)
+	SearchGroupMember(ctx context.Context, keyword string, groupIDs []string, userIDs []string, roleLevels []int32, pageNumber, showNumber int32) (uint32, []*relationTb.GroupMemberModel, error)
+	HandlerGroupRequest(ctx context.Context, groupID string, userID string, handledMsg string, handleResult int32, member *relationTb.GroupMemberModel) error
+	DeleteGroupMember(ctx context.Context, groupID string, userIDs []string) error
+	MapGroupMemberUserID(ctx context.Context, groupIDs []string) (map[string]*relationTb.GroupSimpleUserID, error)
+	MapGroupMemberNum(ctx context.Context, groupIDs []string) (map[string]uint32, error)
+	TransferGroupOwner(ctx context.Context, groupID string, oldOwnerUserID, newOwnerUserID string, roleLevel int32) error // 转让群
+	UpdateGroupMember(ctx context.Context, groupID string, userID string, data map[string]any) error
+	UpdateGroupMembers(ctx context.Context, data []*relationTb.BatchUpdateGroupMember) error
+	// GroupRequest
+	CreateGroupRequest(ctx context.Context, requests []*relationTb.GroupRequestModel) error
+	TakeGroupRequest(ctx context.Context, groupID string, userID string) (*relationTb.GroupRequestModel, error)
+	PageGroupRequestUser(ctx context.Context, userID string, pageNumber, showNumber int32) (uint32, []*relationTb.GroupRequestModel, error)
+	// SuperGroupModelInterface
+	FindSuperGroup(ctx context.Context, groupIDs []string) ([]*unRelationTb.SuperGroupModel, error)
+	FindJoinSuperGroup(ctx context.Context, userID string) ([]string, error)
+	CreateSuperGroup(ctx context.Context, groupID string, initMemberIDList []string) error
+	DeleteSuperGroup(ctx context.Context, groupID string) error
+	DeleteSuperGroupMember(ctx context.Context, groupID string, userIDs []string) error
+	CreateSuperGroupMember(ctx context.Context, groupID string, userIDs []string) error
+}
+
+func NewGroupDatabase(
+	group relationTb.GroupModelInterface,
+	member relationTb.GroupMemberModelInterface,
+	request relationTb.GroupRequestModelInterface,
+	tx tx.Tx,
+	ctxTx tx.CtxTx,
+	superGroup unRelationTb.SuperGroupModelInterface,
+	cache cache.GroupCache,
+) GroupDatabase {
+	database := &groupDatabase{
+		groupDB:        group,
+		groupMemberDB:  member,
+		groupRequestDB: request,
+		tx:             tx,
+		ctxTx:          ctxTx,
+		cache:          cache,
+		mongoDB:        superGroup,
+	}
+	return database
+}
+
+func InitGroupDatabase(db *gorm.DB, rdb redis.UniversalClient, database *mongo.Database) GroupDatabase {
+	rcOptions := rockscache.NewDefaultOptions()
+	rcOptions.StrongConsistency = true
+	rcOptions.RandomExpireAdjustment = 0.2
+	return NewGroupDatabase(
+		relation.NewGroupDB(db),
+		relation.NewGroupMemberDB(db),
+		relation.NewGroupRequest(db),
+		tx.NewGorm(db),
+		tx.NewMongo(database.Client()),
+		unrelation.NewSuperGroupMongoDriver(database),
+		cache.NewGroupCacheRedis(rdb, relation.NewGroupDB(db), relation.NewGroupMemberDB(db), relation.NewGroupRequest(db), unrelation.NewSuperGroupMongoDriver(database), rcOptions),
+	)
+}
+
+type groupDatabase struct {
+	groupDB        relationTb.GroupModelInterface
+	groupMemberDB  relationTb.GroupMemberModelInterface
+	groupRequestDB relationTb.GroupRequestModelInterface
+	tx             tx.Tx
+	ctxTx          tx.CtxTx
+	cache          cache.GroupCache
+	mongoDB        unRelationTb.SuperGroupModelInterface
+}
+
+func (g *groupDatabase) GetGroupIDsByGroupType(ctx context.Context, groupType int) (groupIDs []string, err error) {
+	return g.groupDB.GetGroupIDsByGroupType(ctx, groupType)
+}
+
+func (g *groupDatabase) FindGroupMemberUserID(ctx context.Context, groupID string) ([]string, error) {
+	return g.cache.GetGroupMemberIDs(ctx, groupID)
+}
+
+func (g *groupDatabase) FindGroupMemberNum(ctx context.Context, groupID string) (uint32, error) {
+	num, err := g.cache.GetGroupMemberNum(ctx, groupID)
+	if err != nil {
+		return 0, err
+	}
+	return uint32(num), nil
+}
+
+func (g *groupDatabase) CreateGroup(ctx context.Context, groups []*relationTb.GroupModel, groupMembers []*relationTb.GroupMemberModel) error {
+	var cache = g.cache.NewCache()
+	if err := g.tx.Transaction(func(tx any) error {
+		if len(groups) > 0 {
+			if err := g.groupDB.NewTx(tx).Create(ctx, groups); err != nil {
+				return err
+			}
+		}
+		if len(groupMembers) > 0 {
+			if err := g.groupMemberDB.NewTx(tx).Create(ctx, groupMembers); err != nil {
+				return err
+			}
+		}
+		createGroupIDs := utils.DistinctAnyGetComparable(groups, func(group *relationTb.GroupModel) string {
+			return group.GroupID
+		})
+		m := make(map[string]struct{})
+
+		for _, groupMember := range groupMembers {
+			if _, ok := m[groupMember.GroupID]; !ok {
+				m[groupMember.GroupID] = struct{}{}
+				cache = cache.DelGroupMemberIDs(groupMember.GroupID).DelGroupMembersHash(groupMember.GroupID).DelGroupsMemberNum(groupMember.GroupID)
+			}
+			cache = cache.DelJoinedGroupID(groupMember.UserID).DelGroupMembersInfo(groupMember.GroupID, groupMember.UserID)
+		}
+		cache = cache.DelGroupsInfo(createGroupIDs...)
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (g *groupDatabase) TakeGroup(ctx context.Context, groupID string) (group *relationTb.GroupModel, err error) {
+	return g.cache.GetGroupInfo(ctx, groupID)
+}
+
+func (g *groupDatabase) FindGroup(ctx context.Context, groupIDs []string) (groups []*relationTb.GroupModel, err error) {
+	return g.cache.GetGroupsInfo(ctx, groupIDs)
+}
+
+func (g *groupDatabase) SearchGroup(ctx context.Context, keyword string, pageNumber, showNumber int32) (uint32, []*relationTb.GroupModel, error) {
+	return g.groupDB.Search(ctx, keyword, pageNumber, showNumber)
+}
+
+func (g *groupDatabase) UpdateGroup(ctx context.Context, groupID string, data map[string]any) error {
+	if err := g.groupDB.UpdateMap(ctx, groupID, data); err != nil {
+		return err
+	}
+	return g.cache.DelGroupsInfo(groupID).ExecDel(ctx)
+}
+
+func (g *groupDatabase) DismissGroup(ctx context.Context, groupID string, deleteMember bool) error {
+	cache := g.cache.NewCache()
+	if err := g.tx.Transaction(func(tx any) error {
+		if err := g.groupDB.NewTx(tx).UpdateStatus(ctx, groupID, constant.GroupStatusDismissed); err != nil {
+			return err
+		}
+		if deleteMember {
+			if err := g.groupMemberDB.NewTx(tx).DeleteGroup(ctx, []string{groupID}); err != nil {
+				return err
+			}
+			userIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID)
+			if err != nil {
+				return err
+			}
+			cache = cache.DelJoinedGroupID(userIDs...).DelGroupMemberIDs(groupID).DelGroupsMemberNum(groupID).DelGroupMembersHash(groupID)
+		}
+		cache = cache.DelGroupsInfo(groupID)
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (g *groupDatabase) TakeGroupMember(ctx context.Context, groupID string, userID string) (groupMember *relationTb.GroupMemberModel, err error) {
+	return g.cache.GetGroupMemberInfo(ctx, groupID, userID)
+}
+
+func (g *groupDatabase) TakeGroupOwner(ctx context.Context, groupID string) (*relationTb.GroupMemberModel, error) {
+	return g.groupMemberDB.TakeOwner(ctx, groupID) // todo cache group owner
+}
+
+func (g *groupDatabase) FindUserManagedGroupID(ctx context.Context, userID string) (groupIDs []string, err error) {
+	return g.groupMemberDB.FindUserManagedGroupID(ctx, userID)
+}
+
+func (g *groupDatabase) PageGroupRequest(ctx context.Context, groupIDs []string, pageNumber, showNumber int32) (uint32, []*relationTb.GroupRequestModel, error) {
+	return g.groupRequestDB.PageGroup(ctx, groupIDs, pageNumber, showNumber)
+}
+
+func (g *groupDatabase) FindGroupMember(ctx context.Context, groupIDs []string, userIDs []string, roleLevels []int32) (totalGroupMembers []*relationTb.GroupMemberModel, err error) {
+	if roleLevels == nil {
+		for _, groupID := range groupIDs {
+			groupMembers, err := g.cache.GetGroupMembersInfo(ctx, groupID, userIDs)
+			if err != nil {
+				return nil, err
+			}
+			totalGroupMembers = append(totalGroupMembers, groupMembers...)
+		}
+		return totalGroupMembers, nil
+	}
+	return g.groupMemberDB.Find(ctx, groupIDs, userIDs, roleLevels)
+}
+
+func (g *groupDatabase) PageGetJoinGroup(ctx context.Context, userID string, pageNumber, showNumber int32) (total uint32, totalGroupMembers []*relationTb.GroupMemberModel, err error) {
+	groupIDs, err := g.cache.GetJoinedGroupIDs(ctx, userID)
+	if err != nil {
+		return 0, nil, err
+	}
+	for _, groupID := range utils.Paginate(groupIDs, int(pageNumber), int(showNumber)) {
+		groupMembers, err := g.cache.GetGroupMembersInfo(ctx, groupID, []string{userID})
+		if err != nil {
+			return 0, nil, err
+		}
+		totalGroupMembers = append(totalGroupMembers, groupMembers...)
+	}
+	return uint32(len(groupIDs)), totalGroupMembers, nil
+}
+
+func (g *groupDatabase) PageGetGroupMember(ctx context.Context, groupID string, pageNumber, showNumber int32) (total uint32, totalGroupMembers []*relationTb.GroupMemberModel, err error) {
+	groupMemberIDs, err := g.cache.GetGroupMemberIDs(ctx, groupID)
+	if err != nil {
+		return 0, nil, err
+	}
+	pageIDs := utils.Paginate(groupMemberIDs, int(pageNumber), int(showNumber))
+	if len(pageIDs) == 0 {
+		return uint32(len(groupMemberIDs)), nil, nil
+	}
+	members, err := g.cache.GetGroupMembersInfo(ctx, groupID, pageIDs)
+	if err != nil {
+		return 0, nil, err
+	}
+	return uint32(len(groupMemberIDs)), members, nil
+}
+
+func (g *groupDatabase) SearchGroupMember(ctx context.Context, keyword string, groupIDs []string, userIDs []string, roleLevels []int32, pageNumber, showNumber int32) (uint32, []*relationTb.GroupMemberModel, error) {
+	return g.groupMemberDB.SearchMember(ctx, keyword, groupIDs, userIDs, roleLevels, pageNumber, showNumber)
+}
+
+func (g *groupDatabase) HandlerGroupRequest(ctx context.Context, groupID string, userID string, handledMsg string, handleResult int32, member *relationTb.GroupMemberModel) error {
+	cache := g.cache.NewCache()
+	if err := g.tx.Transaction(func(tx any) error {
+		if err := g.groupRequestDB.NewTx(tx).UpdateHandler(ctx, groupID, userID, handledMsg, handleResult); err != nil {
+			return err
+		}
+		if member != nil {
+			if err := g.groupMemberDB.NewTx(tx).Create(ctx, []*relationTb.GroupMemberModel{member}); err != nil {
+				return err
+			}
+			cache = cache.DelGroupMembersHash(groupID).DelGroupMemberIDs(groupID).DelGroupsMemberNum(groupID).DelJoinedGroupID(member.UserID)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (g *groupDatabase) DeleteGroupMember(ctx context.Context, groupID string, userIDs []string) error {
+	if err := g.groupMemberDB.Delete(ctx, groupID, userIDs); err != nil {
+		return err
+	}
+	return g.cache.DelGroupMembersHash(groupID).DelGroupMemberIDs(groupID).DelGroupsMemberNum(groupID).DelJoinedGroupID(userIDs...).DelGroupMembersInfo(groupID, userIDs...).ExecDel(ctx)
+}
+
+func (g *groupDatabase) MapGroupMemberUserID(ctx context.Context, groupIDs []string) (map[string]*relationTb.GroupSimpleUserID, error) {
+	return g.cache.GetGroupMemberHashMap(ctx, groupIDs)
+}
+
+func (g *groupDatabase) MapGroupMemberNum(ctx context.Context, groupIDs []string) (m map[string]uint32, err error) {
+	m = make(map[string]uint32)
+	for _, groupID := range groupIDs {
+		num, err := g.cache.GetGroupMemberNum(ctx, groupID)
+		if err != nil {
+			return nil, err
+		}
+		m[groupID] = uint32(num)
+	}
+	return m, nil
+}
+
+func (g *groupDatabase) TransferGroupOwner(ctx context.Context, groupID string, oldOwnerUserID, newOwnerUserID string, roleLevel int32) error {
+	if err := g.tx.Transaction(func(tx any) error {
+		rowsAffected, err := g.groupMemberDB.NewTx(tx).UpdateRoleLevel(ctx, groupID, oldOwnerUserID, roleLevel)
+		if err != nil {
+			return err
+		}
+		if rowsAffected != 1 {
+			return utils.Wrap(fmt.Errorf("oldOwnerUserID %s rowsAffected = %d", oldOwnerUserID, rowsAffected), "")
+		}
+		rowsAffected, err = g.groupMemberDB.NewTx(tx).UpdateRoleLevel(ctx, groupID, newOwnerUserID, constant.GroupOwner)
+		if err != nil {
+			return err
+		}
+		if rowsAffected != 1 {
+			return utils.Wrap(fmt.Errorf("newOwnerUserID %s rowsAffected = %d", newOwnerUserID, rowsAffected), "")
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return g.cache.DelGroupMembersInfo(groupID, oldOwnerUserID, newOwnerUserID).ExecDel(ctx)
+}
+
+func (g *groupDatabase) UpdateGroupMember(ctx context.Context, groupID string, userID string, data map[string]any) error {
+	if err := g.groupMemberDB.Update(ctx, groupID, userID, data); err != nil {
+		return err
+	}
+	return g.cache.DelGroupMembersInfo(groupID, userID).ExecDel(ctx)
+}
+
+func (g *groupDatabase) UpdateGroupMembers(ctx context.Context, data []*relationTb.BatchUpdateGroupMember) error {
+	var cache = g.cache.NewCache()
+	if err := g.tx.Transaction(func(tx any) error {
+		for _, item := range data {
+			if err := g.groupMemberDB.NewTx(tx).Update(ctx, item.GroupID, item.UserID, item.Map); err != nil {
+				return err
+			}
+			cache = cache.DelGroupMembersInfo(item.GroupID, item.UserID)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (g *groupDatabase) CreateGroupRequest(ctx context.Context, requests []*relationTb.GroupRequestModel) error {
+	return g.tx.Transaction(func(tx any) error {
+		db := g.groupRequestDB.NewTx(tx)
+		for _, request := range requests {
+			if err := db.Delete(ctx, request.GroupID, request.UserID); err != nil {
+				return err
+			}
+		}
+		return db.Create(ctx, requests)
+	})
+}
+
+func (g *groupDatabase) TakeGroupRequest(ctx context.Context, groupID string, userID string) (*relationTb.GroupRequestModel, error) {
+	return g.groupRequestDB.Take(ctx, groupID, userID)
+}
+
+func (g *groupDatabase) PageGroupRequestUser(ctx context.Context, userID string, pageNumber, showNumber int32) (uint32, []*relationTb.GroupRequestModel, error) {
+	return g.groupRequestDB.Page(ctx, userID, pageNumber, showNumber)
+}
+
+func (g *groupDatabase) FindSuperGroup(ctx context.Context, groupIDs []string) (models []*unRelationTb.SuperGroupModel, err error) {
+	return g.cache.GetSuperGroupMemberIDs(ctx, groupIDs...)
+}
+
+func (g *groupDatabase) FindJoinSuperGroup(ctx context.Context, userID string) ([]string, error) {
+	return g.cache.GetJoinedSuperGroupIDs(ctx, userID)
+}
+
+func (g *groupDatabase) CreateSuperGroup(ctx context.Context, groupID string, initMemberIDs []string) error {
+	if err := g.mongoDB.CreateSuperGroup(ctx, groupID, initMemberIDs); err != nil {
+		return err
+	}
+	return g.cache.DelSuperGroupMemberIDs(groupID).DelJoinedSuperGroupIDs(initMemberIDs...).ExecDel(ctx)
+}
+
+func (g *groupDatabase) DeleteSuperGroup(ctx context.Context, groupID string) error {
+	cache := g.cache.NewCache()
+	if err := g.ctxTx.Transaction(ctx, func(ctx context.Context) error {
+		if err := g.mongoDB.DeleteSuperGroup(ctx, groupID); err != nil {
+			return err
+		}
+		models, err := g.cache.GetSuperGroupMemberIDs(ctx, groupID)
+		if err != nil {
+			return err
+		}
+		cache = cache.DelSuperGroupMemberIDs(groupID)
+		if len(models) > 0 {
+			cache = cache.DelJoinedSuperGroupIDs(models[0].MemberIDs...)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	return cache.ExecDel(ctx)
+}
+
+func (g *groupDatabase) DeleteSuperGroupMember(ctx context.Context, groupID string, userIDs []string) error {
+	if err := g.mongoDB.RemoverUserFromSuperGroup(ctx, groupID, userIDs); err != nil {
+		return err
+	}
+	return g.cache.DelSuperGroupMemberIDs(groupID).DelJoinedSuperGroupIDs(userIDs...).ExecDel(ctx)
+}
+
+func (g *groupDatabase) CreateSuperGroupMember(ctx context.Context, groupID string, userIDs []string) error {
+	if err := g.mongoDB.AddUserToSuperGroup(ctx, groupID, userIDs); err != nil {
+		return err
+	}
+	return g.cache.DelSuperGroupMemberIDs(groupID).DelJoinedSuperGroupIDs(userIDs...).ExecDel(ctx)
+}

pkg/common/db/controller/msg.go

@@ -0,0 +1,927 @@
+package controller
+
+import (
+	"fmt"
+	"github.com/redis/go-redis/v9"
+	"time"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/convert"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/kafka"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
+
+	"context"
+	"errors"
+
+	pbMsg "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/msg"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+const (
+	updateKeyMsg = iota
+	updateKeyRevoke
+)
+
+type CommonMsgDatabase interface {
+	// 批量插入消息
+	BatchInsertChat2DB(ctx context.Context, conversationID string, msgs []*sdkws.MsgData, currentMaxSeq int64) error
+	// 撤回消息
+	RevokeMsg(ctx context.Context, conversationID string, seq int64, revoke *unRelationTb.RevokeModel) error
+	// mark as read
+	MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, seqs []int64) error
+	// 刪除redis中消息缓存
+	DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
+	DelUserDeleteMsgsList(ctx context.Context, conversationID string, seqs []int64)
+	// incrSeq然后批量插入缓存
+	BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error)
+
+	//  通过seqList获取mongo中写扩散消息
+	GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
+	// 通过seqList获取大群在 mongo里面的消息
+	GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
+	// 删除会话消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache)
+	DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
+	// 用户根据seq删除消息
+	DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error
+	// 物理删除消息置空
+	DeleteMsgsPhysicalBySeqs(ctx context.Context, conversationID string, seqs []int64) error
+
+	SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error
+	GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMaxSeq(ctx context.Context, conversationID string) (int64, error)
+	SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error
+	SetMinSeqs(ctx context.Context, seqs map[string]int64) error
+
+	GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
+	GetMinSeq(ctx context.Context, conversationID string) (int64, error)
+	GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error)
+	GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error)
+	SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error
+	SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error)
+	SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) (err error)
+	SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error
+	GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error)
+	GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error)
+	UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error
+
+	GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (maxSeq, minSeq int64, err error)
+	GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error)
+	SetSendMsgStatus(ctx context.Context, id string, status int32) error
+	GetSendMsgStatus(ctx context.Context, id string) (int32, error)
+
+	// to mq
+	MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error
+	MsgToModifyMQ(ctx context.Context, key, conversarionID string, msgs []*sdkws.MsgData) error
+	MsgToPushMQ(ctx context.Context, key, conversarionID string, msg2mq *sdkws.MsgData) (int32, int64, error)
+	MsgToMongoMQ(ctx context.Context, key, conversarionID string, msgs []*sdkws.MsgData, lastSeq int64) error
+
+	// modify
+	JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error)
+	SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error
+	SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error)
+	GetExtendMsg(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (*pbMsg.ExtendMsg, error)
+	InsertOrUpdateReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
+	GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error)
+	GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error)
+	DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error
+	DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
+}
+
+func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase {
+	return &commonMsgDatabase{
+		msgDocDatabase:   msgDocModel,
+		cache:            cacheModel,
+		producer:         kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.LatestMsgToRedis.Topic),
+		producerToMongo:  kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToMongo.Topic),
+		producerToPush:   kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToPush.Topic),
+		producerToModify: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToModify.Topic),
+	}
+}
+
+func InitCommonMsgDatabase(rdb redis.UniversalClient, database *mongo.Database) CommonMsgDatabase {
+	cacheModel := cache.NewMsgCacheModel(rdb)
+	msgDocModel := unrelation.NewMsgMongoDriver(database)
+	CommonMsgDatabase := NewCommonMsgDatabase(msgDocModel, cacheModel)
+	return CommonMsgDatabase
+}
+
+type commonMsgDatabase struct {
+	msgDocDatabase    unRelationTb.MsgDocModelInterface
+	extendMsgDatabase unRelationTb.ExtendMsgSetModelInterface
+	extendMsgSetModel unRelationTb.ExtendMsgSetModel
+	msg               unRelationTb.MsgDocModel
+	cache             cache.MsgModel
+	producer          *kafka.Producer
+	producerToMongo   *kafka.Producer
+	producerToModify  *kafka.Producer
+	producerToPush    *kafka.Producer
+}
+
+func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sdkws.MsgData) error {
+	_, _, err := db.producer.SendMessage(ctx, key, msg2mq)
+	return err
+}
+
+func (db *commonMsgDatabase) MsgToModifyMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData) error {
+	if len(messages) > 0 {
+		_, _, err := db.producerToModify.SendMessage(ctx, key, &pbMsg.MsgDataToModifyByMQ{ConversationID: conversationID, Messages: messages})
+		return err
+	}
+	return nil
+}
+
+func (db *commonMsgDatabase) MsgToPushMQ(ctx context.Context, key, conversationID string, msg2mq *sdkws.MsgData) (int32, int64, error) {
+	partition, offset, err := db.producerToPush.SendMessage(ctx, key, &pbMsg.PushMsgDataToMQ{MsgData: msg2mq, ConversationID: conversationID})
+	if err != nil {
+		log.ZError(ctx, "MsgToPushMQ", err, "key", key, "msg2mq", msg2mq)
+		return 0, 0, err
+	}
+	return partition, offset, nil
+}
+
+func (db *commonMsgDatabase) MsgToMongoMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData, lastSeq int64) error {
+	if len(messages) > 0 {
+		_, _, err := db.producerToMongo.SendMessage(ctx, key, &pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, ConversationID: conversationID, MsgData: messages})
+		return err
+	}
+	return nil
+}
+
+func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationID string, fields []any, key int8, firstSeq int64) error {
+	if len(fields) == 0 {
+		return nil
+	}
+	num := db.msg.GetSingleGocMsgNum()
+	//num = 100
+	for i, field := range fields { // 检查类型
+		var ok bool
+		switch key {
+		case updateKeyMsg:
+			var msg *unRelationTb.MsgDataModel
+			msg, ok = field.(*unRelationTb.MsgDataModel)
+			if msg != nil && msg.Seq != firstSeq+int64(i) {
+				return errs.ErrInternalServer.Wrap("seq is invalid")
+			}
+		case updateKeyRevoke:
+			_, ok = field.(*unRelationTb.RevokeModel)
+		default:
+			return errs.ErrInternalServer.Wrap("key is invalid")
+		}
+		if !ok {
+			return errs.ErrInternalServer.Wrap("field type is invalid")
+		}
+	}
+	// 返回值为true表示数据库存在该文档，false表示数据库不存在该文档
+	updateMsgModel := func(seq int64, i int) (bool, error) {
+		var (
+			res *mongo.UpdateResult
+			err error
+		)
+		docID := db.msg.GetDocID(conversationID, seq)
+		index := db.msg.GetMsgIndex(seq)
+		field := fields[i]
+		switch key {
+		case updateKeyMsg:
+			res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "msg", field)
+		case updateKeyRevoke:
+			res, err = db.msgDocDatabase.UpdateMsg(ctx, docID, index, "revoke", field)
+		}
+		if err != nil {
+			return false, err
+		}
+		return res.MatchedCount > 0, nil
+	}
+	tryUpdate := true
+	for i := 0; i < len(fields); i++ {
+		seq := firstSeq + int64(i) // 当前seq
+		if tryUpdate {
+			matched, err := updateMsgModel(seq, i)
+			if err != nil {
+				return err
+			}
+			if matched {
+				continue // 匹配到了，继续下一个(不一定修改)
+			}
+		}
+		doc := unRelationTb.MsgDocModel{
+			DocID: db.msg.GetDocID(conversationID, seq),
+			Msg:   make([]*unRelationTb.MsgInfoModel, num),
+		}
+		var insert int // 插入的数量
+		for j := i; j < len(fields); j++ {
+			seq = firstSeq + int64(j)
+			if db.msg.GetDocID(conversationID, seq) != doc.DocID {
+				break
+			}
+			insert++
+			switch key {
+			case updateKeyMsg:
+				doc.Msg[db.msg.GetMsgIndex(seq)] = &unRelationTb.MsgInfoModel{
+					Msg: fields[j].(*unRelationTb.MsgDataModel),
+				}
+			case updateKeyRevoke:
+				doc.Msg[db.msg.GetMsgIndex(seq)] = &unRelationTb.MsgInfoModel{
+					Revoke: fields[j].(*unRelationTb.RevokeModel),
+				}
+			}
+		}
+		for i, model := range doc.Msg {
+			if model == nil {
+				model = &unRelationTb.MsgInfoModel{}
+				doc.Msg[i] = model
+			}
+			if model.DelList == nil {
+				doc.Msg[i].DelList = []string{}
+			}
+		}
+		if err := db.msgDocDatabase.Create(ctx, &doc); err != nil {
+			if mongo.IsDuplicateKeyError(err) {
+				i--              // 存在并发,重试当前数据
+				tryUpdate = true // 以修改模式
+				continue
+			}
+			return err
+		}
+		tryUpdate = false // 当前以插入成功,下一块优先插入模式
+		i += insert - 1   // 跳过已插入的数据
+	}
+	return nil
+}
+
+func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
+	if len(msgList) == 0 {
+		return errs.ErrArgs.Wrap("msgList is empty")
+	}
+	msgs := make([]any, len(msgList))
+	for i, msg := range msgList {
+		if msg == nil {
+			continue
+		}
+		var offlinePushModel *unRelationTb.OfflinePushModel
+		if msg.OfflinePushInfo != nil {
+			offlinePushModel = &unRelationTb.OfflinePushModel{
+				Title:         msg.OfflinePushInfo.Title,
+				Desc:          msg.OfflinePushInfo.Desc,
+				Ex:            msg.OfflinePushInfo.Ex,
+				IOSPushSound:  msg.OfflinePushInfo.IOSPushSound,
+				IOSBadgeCount: msg.OfflinePushInfo.IOSBadgeCount,
+			}
+		}
+		msgs[i] = &unRelationTb.MsgDataModel{
+			SendID:           msg.SendID,
+			RecvID:           msg.RecvID,
+			GroupID:          msg.GroupID,
+			ClientMsgID:      msg.ClientMsgID,
+			ServerMsgID:      msg.ServerMsgID,
+			SenderPlatformID: msg.SenderPlatformID,
+			SenderNickname:   msg.SenderNickname,
+			SenderFaceURL:    msg.SenderFaceURL,
+			SessionType:      msg.SessionType,
+			MsgFrom:          msg.MsgFrom,
+			ContentType:      msg.ContentType,
+			Content:          string(msg.Content),
+			Seq:              msg.Seq,
+			SendTime:         msg.SendTime,
+			CreateTime:       msg.CreateTime,
+			Status:           msg.Status,
+			Options:          msg.Options,
+			OfflinePush:      offlinePushModel,
+			AtUserIDList:     msg.AtUserIDList,
+			AttachedInfo:     msg.AttachedInfo,
+			Ex:               msg.Ex,
+		}
+	}
+	return db.BatchInsertBlock(ctx, conversationID, msgs, updateKeyMsg, msgList[0].Seq)
+}
+
+func (db *commonMsgDatabase) RevokeMsg(ctx context.Context, conversationID string, seq int64, revoke *unRelationTb.RevokeModel) error {
+	return db.BatchInsertBlock(ctx, conversationID, []any{revoke}, updateKeyRevoke, seq)
+}
+
+func (db *commonMsgDatabase) MarkSingleChatMsgsAsRead(ctx context.Context, userID string, conversationID string, totalSeqs []int64) error {
+	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, totalSeqs) {
+		var indexes []int64
+		for _, seq := range seqs {
+			indexes = append(indexes, db.msg.GetMsgIndex(seq))
+		}
+		log.ZDebug(ctx, "MarkSingleChatMsgsAsRead", "userID", userID, "docID", docID, "indexes", indexes)
+		if err := db.msgDocDatabase.MarkSingleChatMsgsAsRead(ctx, userID, docID, indexes); err != nil {
+			log.ZError(ctx, "MarkSingleChatMsgsAsRead", err, "userID", userID, "docID", docID, "indexes", indexes)
+			return err
+		}
+	}
+	return nil
+}
+
+func (db *commonMsgDatabase) DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error {
+	return db.cache.DeleteMessages(ctx, conversationID, seqs)
+}
+
+func (db *commonMsgDatabase) DelUserDeleteMsgsList(ctx context.Context, conversationID string, seqs []int64) {
+	db.cache.DelUserDeleteMsgsList(ctx, conversationID, seqs)
+}
+
+func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
+	currentMaxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		prome.Inc(prome.SeqGetFailedCounter)
+		return 0, false, err
+	}
+	prome.Inc(prome.SeqGetSuccessCounter)
+	lenList := len(msgs)
+	if int64(lenList) > db.msg.GetSingleGocMsgNum() {
+		return 0, false, errors.New("too large")
+	}
+	if lenList < 1 {
+		return 0, false, errors.New("too short as 0")
+	}
+	if errs.Unwrap(err) == redis.Nil {
+		isNew = true
+	}
+	lastMaxSeq := currentMaxSeq
+	userSeqMap := make(map[string]int64)
+	for _, m := range msgs {
+		currentMaxSeq++
+		m.Seq = currentMaxSeq
+		userSeqMap[m.SendID] = m.Seq
+	}
+	failedNum, err := db.cache.SetMessageToCache(ctx, conversationID, msgs)
+	if err != nil {
+		prome.Add(prome.MsgInsertRedisFailedCounter, failedNum)
+		log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
+	} else {
+		prome.Inc(prome.MsgInsertRedisSuccessCounter)
+	}
+	err = db.cache.SetMaxSeq(ctx, conversationID, currentMaxSeq)
+	if err != nil {
+		prome.Inc(prome.SeqSetFailedCounter)
+	} else {
+		prome.Inc(prome.SeqSetSuccessCounter)
+	}
+	err2 := db.cache.SetHasReadSeqs(ctx, conversationID, userSeqMap)
+	if err != nil {
+		log.ZError(ctx, "SetHasReadSeqs error", err2, "userSeqMap", userSeqMap, "conversationID", conversationID)
+		prome.Inc(prome.SeqSetFailedCounter)
+	} else {
+		prome.Inc(prome.SeqSetSuccessCounter)
+	}
+	return lastMaxSeq, isNew, utils.Wrap(err, "")
+}
+
+func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
+	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, seqs) {
+		//log.ZDebug(ctx, "getMsgBySeqs", "docID", docID, "seqs", seqs)
+		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, seqs)
+		if err != nil {
+			return nil, err
+		}
+		for _, msg := range msgs {
+			totalMsgs = append(totalMsgs, convert.MsgDB2Pb(msg.Msg))
+		}
+	}
+	return totalMsgs, nil
+}
+
+// func (db *commonMsgDatabase) refetchDelSeqsMsgs(ctx context.Context, conversationID string, delNums, rangeBegin, begin int64) (seqMsgs []*unRelationTb.MsgDataModel, err error) {
+// 	var reFetchSeqs []int64
+// 	if delNums > 0 {
+// 		newBeginSeq := rangeBegin - delNums
+// 		if newBeginSeq >= begin {
+// 			newEndSeq := rangeBegin - 1
+// 			for i := newBeginSeq; i <= newEndSeq; i++ {
+// 				reFetchSeqs = append(reFetchSeqs, i)
+// 			}
+// 		}
+// 	}
+// 	if len(reFetchSeqs) == 0 {
+// 		return
+// 	}
+// 	if len(reFetchSeqs) > 0 {
+// 		m := db.msg.GetDocIDSeqsMap(conversationID, reFetchSeqs)
+// 		for docID, seqs := range m {
+// 			msgs, _, err := db.findMsgInfoBySeq(ctx, docID, seqs)
+// 			if err != nil {
+// 				return nil, err
+// 			}
+// 			for _, msg := range msgs {
+// 				if msg.Status != constant.MsgDeleted {
+// 					seqMsgs = append(seqMsgs, msg)
+// 				}
+// 			}
+// 		}
+// 	}
+// 	if len(seqMsgs) < int(delNums) {
+// 		seqMsgs2, err := db.refetchDelSeqsMsgs(ctx, conversationID, delNums-int64(len(seqMsgs)), rangeBegin-1, begin)
+// 		if err != nil {
+// 			return seqMsgs, err
+// 		}
+// 		seqMsgs = append(seqMsgs, seqMsgs2...)
+// 	}
+// 	return seqMsgs, nil
+// }
+
+func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID string, seqs []int64) (totalMsgs []*unRelationTb.MsgInfoModel, err error) {
+	msgs, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, userID, docID, seqs)
+	for _, msg := range msgs {
+		if msg.IsRead {
+			msg.Msg.IsRead = true
+		}
+	}
+	return msgs, err
+}
+
+func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID string, conversationID string, allSeqs []int64, begin, end int64) (seqMsgs []*sdkws.MsgData, err error) {
+	log.ZDebug(ctx, "getMsgBySeqsRange", "conversationID", conversationID, "allSeqs", allSeqs, "begin", begin, "end", end)
+	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, allSeqs) {
+		log.ZDebug(ctx, "getMsgBySeqsRange", "docID", docID, "seqs", seqs)
+		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, seqs)
+		if err != nil {
+			return nil, err
+		}
+		for _, msg := range msgs {
+			if msg.IsRead {
+				msg.Msg.IsRead = true
+			}
+			seqMsgs = append(seqMsgs, convert.MsgDB2Pb(msg.Msg))
+		}
+	}
+	return seqMsgs, nil
+}
+
+func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (int64, int64, []*sdkws.MsgData, error) {
+	userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		return 0, 0, nil, err
+	}
+	minSeq, err := db.cache.GetMinSeq(ctx, conversationID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		return 0, 0, nil, err
+	}
+	if userMinSeq > minSeq {
+		minSeq = userMinSeq
+	}
+	if minSeq > end {
+		log.ZInfo(ctx, "minSeq > end", "minSeq", minSeq, "end", end)
+		return 0, 0, nil, nil
+	}
+	maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		return 0, 0, nil, err
+	}
+	log.ZDebug(ctx, "GetMsgBySeqsRange", "userMinSeq", userMinSeq, "conMinSeq", minSeq, "conMaxSeq", maxSeq, "userMaxSeq", userMaxSeq)
+	if userMaxSeq != 0 {
+		if userMaxSeq < maxSeq {
+			maxSeq = userMaxSeq
+		}
+	}
+	if begin < minSeq {
+		begin = minSeq
+	}
+	if end > maxSeq {
+		end = maxSeq
+	}
+	if end < begin {
+		return 0, 0, nil, errs.ErrArgs.Wrap("seq end < begin")
+	}
+	var seqs []int64
+	for i := end; i > end-num; i-- {
+		if i >= begin {
+			seqs = append([]int64{i}, seqs...)
+		} else {
+			break
+		}
+	}
+	if len(seqs) == 0 {
+		return 0, 0, nil, nil
+	}
+	newBegin := seqs[0]
+	newEnd := seqs[len(seqs)-1]
+	log.ZDebug(ctx, "GetMsgBySeqsRange", "first seqs", seqs, "newBegin", newBegin, "newEnd", newEnd)
+	cachedMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs)
+	if err != nil {
+		if err != redis.Nil {
+			prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs))
+			log.ZError(ctx, "get message from redis exception", err, "conversationID", conversationID, "seqs", seqs)
+		}
+	}
+	var successMsgs []*sdkws.MsgData
+	if len(cachedMsgs) > 0 {
+		delSeqs, err := db.cache.GetUserDelList(ctx, userID, conversationID)
+		if err != nil && errs.Unwrap(err) != redis.Nil {
+			return 0, 0, nil, err
+		}
+		var cacheDelNum int
+		for _, msg := range cachedMsgs {
+			if !utils.Contain(msg.Seq, delSeqs...) {
+				successMsgs = append(successMsgs, msg)
+			} else {
+				cacheDelNum += 1
+			}
+		}
+		log.ZDebug(ctx, "get delSeqs from redis", "delSeqs", delSeqs, "userID", userID, "conversationID", conversationID, "cacheDelNum", cacheDelNum)
+		var reGetSeqsCache []int64
+		for i := 1; i <= cacheDelNum; {
+			newSeq := newBegin - int64(i)
+			if newSeq >= begin {
+				if !utils.Contain(newSeq, delSeqs...) {
+					log.ZDebug(ctx, "seq del in cache, a new seq in range append", "new seq", newSeq)
+					reGetSeqsCache = append(reGetSeqsCache, newSeq)
+					i++
+				}
+			} else {
+				break
+			}
+		}
+		if len(reGetSeqsCache) > 0 {
+			log.ZDebug(ctx, "reGetSeqsCache", "reGetSeqsCache", reGetSeqsCache)
+			cachedMsgs, failedSeqs2, err := db.cache.GetMessagesBySeq(ctx, conversationID, reGetSeqsCache)
+			if err != nil {
+				if err != redis.Nil {
+					prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs2))
+					log.ZError(ctx, "get message from redis exception", err, "conversationID", conversationID, "seqs", reGetSeqsCache)
+				}
+			}
+			failedSeqs = append(failedSeqs, failedSeqs2...)
+			successMsgs = append(successMsgs, cachedMsgs...)
+		}
+	}
+	log.ZDebug(ctx, "get msgs from cache", "successMsgs", successMsgs)
+	if len(failedSeqs) != 0 {
+		log.ZDebug(ctx, "msgs not exist in redis", "seqs", failedSeqs)
+	}
+	// get from cache or db
+	prome.Add(prome.MsgPullFromRedisSuccessCounter, len(successMsgs))
+	if len(failedSeqs) > 0 {
+		mongoMsgs, err := db.getMsgBySeqsRange(ctx, userID, conversationID, failedSeqs, begin, end)
+		if err != nil {
+			prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs))
+			return 0, 0, nil, err
+		}
+		prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs))
+		successMsgs = append(successMsgs, mongoMsgs...)
+	}
+
+	return minSeq, maxSeq, successMsgs, nil
+}
+
+func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
+	userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		return 0, 0, nil, err
+	}
+	minSeq, err := db.cache.GetMinSeq(ctx, conversationID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		return 0, 0, nil, err
+	}
+	maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		return 0, 0, nil, err
+	}
+	if userMinSeq < minSeq {
+		minSeq = userMinSeq
+	}
+	var newSeqs []int64
+	for _, seq := range seqs {
+		if seq >= minSeq && seq <= maxSeq {
+			newSeqs = append(newSeqs, seq)
+		}
+	}
+	successMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, newSeqs)
+	if err != nil {
+		if err != redis.Nil {
+			prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs))
+			log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
+		}
+	}
+	log.ZInfo(ctx, "db.cache.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs", seqs, "successMsgs", len(successMsgs), "failedSeqs", failedSeqs, "conversationID", conversationID)
+	prome.Add(prome.MsgPullFromRedisSuccessCounter, len(successMsgs))
+	if len(failedSeqs) > 0 {
+		mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
+		if err != nil {
+			prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs))
+			return 0, 0, nil, err
+		}
+		prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs))
+		successMsgs = append(successMsgs, mongoMsgs...)
+	}
+	return minSeq, maxSeq, successMsgs, nil
+}
+
+func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
+	var delStruct delMsgRecursionStruct
+	var skip int64
+	minSeq, err := db.deleteMsgRecursion(ctx, conversationID, skip, &delStruct, remainTime)
+	if err != nil {
+		return err
+	}
+	log.ZInfo(ctx, "DeleteConversationMsgsAndSetMinSeq", "conversationID", conversationID, "minSeq", minSeq)
+	if minSeq == 0 {
+		return nil
+	}
+	if remainTime == 0 {
+		err = db.cache.CleanUpOneConversationAllMsg(ctx, conversationID)
+		if err != nil {
+			log.ZWarn(ctx, "CleanUpOneUserAllMsg", err, "conversationID", conversationID)
+		}
+	}
+	return db.cache.SetMinSeq(ctx, conversationID, minSeq)
+}
+
+// this is struct for recursion
+type delMsgRecursionStruct struct {
+	minSeq    int64
+	delDocIDs []string
+}
+
+func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
+	return d.minSeq
+}
+
+// index 0....19(del) 20...69
+// seq 70
+// set minSeq 21
+// recursion 删除list并且返回设置的最小seq
+func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversationID string, index int64, delStruct *delMsgRecursionStruct, remainTime int64) (int64, error) {
+	// find from oldest list
+	msgDocModel, err := db.msgDocDatabase.GetMsgDocModelByIndex(ctx, conversationID, index, 1)
+	if err != nil || msgDocModel.DocID == "" {
+		if err != nil {
+			if err == unrelation.ErrMsgListNotExist {
+				log.ZDebug(ctx, "deleteMsgRecursion ErrMsgListNotExist", "conversationID", conversationID, "index:", index)
+			} else {
+				log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
+			}
+		}
+		// 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList), 结束递归
+		err = db.msgDocDatabase.DeleteDocs(ctx, delStruct.delDocIDs)
+		if err != nil {
+			return 0, err
+		}
+		return delStruct.getSetMinSeq() + 1, nil
+	}
+	log.ZDebug(ctx, "doc info", "conversationID", conversationID, "index", index, "docID", msgDocModel.DocID, "len", len(msgDocModel.Msg))
+	if int64(len(msgDocModel.Msg)) > db.msg.GetSingleGocMsgNum() {
+		log.ZWarn(ctx, "msgs too large", nil, "lenth", len(msgDocModel.Msg), "docID:", msgDocModel.DocID)
+	}
+	if msgDocModel.IsFull() && msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.SendTime+(remainTime*1000) < utils.GetCurrentTimestampByMill() {
+		log.ZDebug(ctx, "doc is full and all msg is expired", "docID", msgDocModel.DocID)
+		delStruct.delDocIDs = append(delStruct.delDocIDs, msgDocModel.DocID)
+		delStruct.minSeq = msgDocModel.Msg[len(msgDocModel.Msg)-1].Msg.Seq
+	} else {
+		var hasMarkDelFlag bool
+		var delMsgIndexs []int
+		for i, MsgInfoModel := range msgDocModel.Msg {
+			if MsgInfoModel != nil && MsgInfoModel.Msg != nil {
+				if utils.GetCurrentTimestampByMill() > MsgInfoModel.Msg.SendTime+(remainTime*1000) {
+					delMsgIndexs = append(delMsgIndexs, i)
+					hasMarkDelFlag = true
+				} else {
+					// 到本条消息不需要删除, minSeq置为这条消息的seq
+					if len(delStruct.delDocIDs) > 0 {
+						log.ZDebug(ctx, "delete docs", "delDocIDs", delStruct.delDocIDs)
+					}
+					if err := db.msgDocDatabase.DeleteDocs(ctx, delStruct.delDocIDs); err != nil {
+						return 0, err
+					}
+					if hasMarkDelFlag {
+						log.ZDebug(ctx, "delete msg by index", "delMsgIndexs", delMsgIndexs, "docID", msgDocModel.DocID)
+						// mark del all delMsgIndexs
+						if err := db.msgDocDatabase.DeleteMsgsInOneDocByIndex(ctx, msgDocModel.DocID, delMsgIndexs); err != nil {
+							return delStruct.getSetMinSeq(), err
+						}
+					}
+					return MsgInfoModel.Msg.Seq, nil
+				}
+			}
+		}
+	}
+	//  继续递归 index+1
+	seq, err := db.deleteMsgRecursion(ctx, conversationID, index+1, delStruct, remainTime)
+	return seq, err
+}
+
+func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conversationID string, allSeqs []int64) error {
+	if err := db.cache.DeleteMessages(ctx, conversationID, allSeqs); err != nil {
+		return err
+	}
+	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, allSeqs) {
+		var indexes []int
+		for _, seq := range seqs {
+			indexes = append(indexes, int(db.msg.GetMsgIndex(seq)))
+		}
+		if err := db.msgDocDatabase.DeleteMsgsInOneDocByIndex(ctx, docID, indexes); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error {
+	cachedMsgs, _, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs)
+	if err != nil && errs.Unwrap(err) != redis.Nil {
+		log.ZWarn(ctx, "DeleteUserMsgsBySeqs", err, "conversationID", conversationID, "seqs", seqs)
+		return err
+	}
+	if len(cachedMsgs) > 0 {
+		var cacheSeqs []int64
+		for _, msg := range cachedMsgs {
+			cacheSeqs = append(cacheSeqs, msg.Seq)
+		}
+		if err := db.cache.UserDeleteMsgs(ctx, conversationID, cacheSeqs, userID); err != nil {
+			return err
+		}
+	}
+
+	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, seqs) {
+		for _, seq := range seqs {
+			if _, err := db.msgDocDatabase.PushUnique(ctx, docID, db.msg.GetMsgIndex(seq), "del_list", []string{userID}); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (db *commonMsgDatabase) DeleteMsgsBySeqs(ctx context.Context, conversationID string, seqs []int64) error {
+	return nil
+}
+
+func (db *commonMsgDatabase) CleanUpUserConversationsMsgs(ctx context.Context, user string, conversationIDs []string) {
+	for _, conversationID := range conversationIDs {
+		maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
+		if err != nil {
+			if err == redis.Nil {
+				log.ZInfo(ctx, "max seq is nil", "conversationID", conversationID)
+			} else {
+				log.ZError(ctx, "get max seq failed", err, "conversationID", conversationID)
+			}
+			continue
+		}
+		if err := db.cache.SetMinSeq(ctx, conversationID, maxSeq+1); err != nil {
+			log.ZError(ctx, "set min seq failed", err, "conversationID", conversationID, "minSeq", maxSeq+1)
+		}
+	}
+}
+
+func (db *commonMsgDatabase) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error {
+	return db.cache.SetMaxSeq(ctx, conversationID, maxSeq)
+}
+func (db *commonMsgDatabase) GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
+	return db.cache.GetMaxSeqs(ctx, conversationIDs)
+}
+func (db *commonMsgDatabase) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) {
+	return db.cache.GetMaxSeq(ctx, conversationID)
+}
+func (db *commonMsgDatabase) SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error {
+	return db.cache.SetMinSeq(ctx, conversationID, minSeq)
+}
+func (db *commonMsgDatabase) SetMinSeqs(ctx context.Context, seqs map[string]int64) error {
+	return db.cache.SetMinSeqs(ctx, seqs)
+}
+
+func (db *commonMsgDatabase) GetMinSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error) {
+	return db.cache.GetMinSeqs(ctx, conversationIDs)
+}
+func (db *commonMsgDatabase) GetMinSeq(ctx context.Context, conversationID string) (int64, error) {
+	return db.cache.GetMinSeq(ctx, conversationID)
+}
+func (db *commonMsgDatabase) GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error) {
+	return db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
+}
+func (db *commonMsgDatabase) GetConversationUserMinSeqs(ctx context.Context, conversationID string, userIDs []string) (map[string]int64, error) {
+	return db.cache.GetConversationUserMinSeqs(ctx, conversationID, userIDs)
+}
+func (db *commonMsgDatabase) SetConversationUserMinSeq(ctx context.Context, conversationID string, userID string, minSeq int64) error {
+	return db.cache.SetConversationUserMinSeq(ctx, conversationID, userID, minSeq)
+}
+func (db *commonMsgDatabase) SetConversationUserMinSeqs(ctx context.Context, conversationID string, seqs map[string]int64) (err error) {
+	return db.cache.SetConversationUserMinSeqs(ctx, conversationID, seqs)
+}
+
+func (db *commonMsgDatabase) SetUserConversationsMinSeqs(ctx context.Context, userID string, seqs map[string]int64) error {
+	return db.cache.SetUserConversationsMinSeqs(ctx, userID, seqs)
+}
+
+func (db *commonMsgDatabase) UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error {
+	return db.cache.UserSetHasReadSeqs(ctx, userID, hasReadSeqs)
+}
+
+func (db *commonMsgDatabase) SetHasReadSeq(ctx context.Context, userID string, conversationID string, hasReadSeq int64) error {
+	return db.cache.SetHasReadSeq(ctx, userID, conversationID, hasReadSeq)
+}
+func (db *commonMsgDatabase) GetHasReadSeqs(ctx context.Context, userID string, conversationIDs []string) (map[string]int64, error) {
+	return db.cache.GetHasReadSeqs(ctx, userID, conversationIDs)
+}
+
+func (db *commonMsgDatabase) GetHasReadSeq(ctx context.Context, userID string, conversationID string) (int64, error) {
+	return db.cache.GetHasReadSeq(ctx, userID, conversationID)
+}
+
+func (db *commonMsgDatabase) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
+	return db.cache.SetSendMsgStatus(ctx, id, status)
+}
+
+func (db *commonMsgDatabase) GetSendMsgStatus(ctx context.Context, id string) (int32, error) {
+	return db.cache.GetSendMsgStatus(ctx, id)
+}
+
+func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) {
+	minSeqMongo, maxSeqMongo, err = db.GetMinMaxSeqMongo(ctx, conversationID)
+	if err != nil {
+		return
+	}
+	minSeqCache, err = db.cache.GetMinSeq(ctx, conversationID)
+	if err != nil {
+		return
+	}
+	maxSeqCache, err = db.cache.GetMaxSeq(ctx, conversationID)
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (db *commonMsgDatabase) GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (maxSeq, minSeq int64, err error) {
+	return db.GetMinMaxSeqMongo(ctx, conversationID)
+}
+
+func (db *commonMsgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
+	oldestMsgMongo, err := db.msgDocDatabase.GetOldestMsg(ctx, conversationID)
+	if err != nil {
+		return
+	}
+	minSeqMongo = oldestMsgMongo.Msg.Seq
+	newestMsgMongo, err := db.msgDocDatabase.GetNewestMsg(ctx, conversationID)
+	if err != nil {
+		return
+	}
+	maxSeqMongo = newestMsgMongo.Msg.Seq
+	return
+}
+
+func (db *commonMsgDatabase) JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error) {
+	return db.cache.JudgeMessageReactionExist(ctx, clientMsgID, sessionType)
+}
+
+func (db *commonMsgDatabase) SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error {
+	return db.cache.SetMessageTypeKeyValue(ctx, clientMsgID, sessionType, typeKey, value)
+}
+
+func (db *commonMsgDatabase) SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error) {
+	return db.cache.SetMessageReactionExpire(ctx, clientMsgID, sessionType, expiration)
+}
+
+func (db *commonMsgDatabase) GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error) {
+	return db.cache.GetMessageTypeKeyValue(ctx, clientMsgID, sessionType, typeKey)
+}
+
+func (db *commonMsgDatabase) GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error) {
+	return db.cache.GetOneMessageAllReactionList(ctx, clientMsgID, sessionType)
+}
+
+func (db *commonMsgDatabase) DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error {
+	return db.cache.DeleteOneMessageKey(ctx, clientMsgID, sessionType, subKey)
+}
+
+func (db *commonMsgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
+	return db.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, conversationID, sessionType, clientMsgID, msgFirstModifyTime, db.extendMsgSetModel.Pb2Model(reactionExtensions))
+}
+
+func (db *commonMsgDatabase) GetExtendMsg(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (*pbMsg.ExtendMsg, error) {
+	extendMsgSet, err := db.extendMsgDatabase.GetExtendMsgSet(ctx, conversationID, sessionType, maxMsgUpdateTime)
+	if err != nil {
+		return nil, err
+	}
+	extendMsg, ok := extendMsgSet.ExtendMsgs[clientMsgID]
+	if !ok {
+		return nil, errs.ErrRecordNotFound.Wrap(fmt.Sprintf("cant find client msg id: %s", clientMsgID))
+	}
+	reactionExtensionList := make(map[string]*pbMsg.KeyValueResp)
+	for key, model := range extendMsg.ReactionExtensionList {
+		reactionExtensionList[key] = &pbMsg.KeyValueResp{
+			KeyValue: &sdkws.KeyValue{
+				TypeKey:          model.TypeKey,
+				Value:            model.Value,
+				LatestUpdateTime: model.LatestUpdateTime,
+			},
+		}
+	}
+	return &pbMsg.ExtendMsg{
+		ReactionExtensions: reactionExtensionList,
+		ClientMsgID:        extendMsg.ClientMsgID,
+		MsgFirstModifyTime: extendMsg.MsgFirstModifyTime,
+		AttachedInfo:       extendMsg.AttachedInfo,
+		Ex:                 extendMsg.Ex,
+	}, nil
+}
+
+func (db *commonMsgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
+	return db.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, conversationID, sessionType, clientMsgID, msgFirstModifyTime, db.extendMsgSetModel.Pb2Model(reactionExtensions))
+}

pkg/common/db/controller/msg_test.go

@@ -0,0 +1,251 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
+	unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/unrelation"
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+func Test_BatchInsertChat2DB(t *testing.T) {
+	config.Config.Mongo.Address = []string{"192.168.44.128:37017"}
+	config.Config.Mongo.Timeout = 60
+	config.Config.Mongo.Database = "openIM"
+	config.Config.Mongo.Source = "admin"
+	config.Config.Mongo.Username = "root"
+	config.Config.Mongo.Password = "openIM123"
+	config.Config.Mongo.MaxPoolSize = 100
+	config.Config.RetainChatRecords = 3650
+	config.Config.ChatRecordsClearTime = "0 2 * * 3"
+
+	mongo, err := unrelation.NewMongo()
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = mongo.GetDatabase().Client().Ping(context.Background(), nil)
+	if err != nil {
+		panic(err)
+	}
+
+	db := &commonMsgDatabase{
+		msgDocDatabase: unrelation.NewMsgMongoDriver(mongo.GetDatabase()),
+	}
+
+	//ctx := context.Background()
+	//msgs := make([]*sdkws.MsgData, 0, 1)
+	//for i := 0; i < cap(msgs); i++ {
+	//	msgs = append(msgs, &sdkws.MsgData{
+	//		Content:  []byte(fmt.Sprintf("test-%d", i)),
+	//		SendTime: time.Now().UnixMilli(),
+	//	})
+	//}
+	//err = db.BatchInsertChat2DB(ctx, "test", msgs, 0)
+	//if err != nil {
+	//	panic(err)
+	//}
+
+	_ = db.BatchInsertChat2DB
+	c := mongo.GetDatabase().Collection("msg")
+
+	ch := make(chan int)
+	rand.Seed(time.Now().UnixNano())
+
+	index := 10
+
+	var wg sync.WaitGroup
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		go func(channelID int) {
+			defer wg.Done()
+			<-ch
+			var arr []string
+			for i := 0; i < 500; i++ {
+				arr = append(arr, strconv.Itoa(i+1))
+			}
+			rand.Shuffle(len(arr), func(i, j int) {
+				arr[i], arr[j] = arr[j], arr[i]
+			})
+			for j, s := range arr {
+				if j == 0 {
+					fmt.Printf("channnelID: %d, arr[0]: %s\n", channelID, arr[j])
+				}
+				filter := bson.M{"doc_id": "test:0"}
+				update := bson.M{
+					"$addToSet": bson.M{
+						fmt.Sprintf("msgs.%d.del_list", index): bson.M{"$each": []string{s}},
+					},
+				}
+				_, err := c.UpdateOne(context.Background(), filter, update)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}(i)
+	}
+
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-ch
+			var arr []string
+			for i := 0; i < 500; i++ {
+				arr = append(arr, strconv.Itoa(1001+i))
+			}
+			rand.Shuffle(len(arr), func(i, j int) {
+				arr[i], arr[j] = arr[j], arr[i]
+			})
+			for _, s := range arr {
+				filter := bson.M{"doc_id": "test:0"}
+				update := bson.M{
+					"$addToSet": bson.M{
+						fmt.Sprintf("msgs.%d.read_list", index): bson.M{"$each": []string{s}},
+					},
+				}
+				_, err := c.UpdateOne(context.Background(), filter, update)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}()
+	}
+
+	time.Sleep(time.Second * 2)
+
+	close(ch)
+
+	wg.Wait()
+
+}
+
+func GetDB() *commonMsgDatabase {
+	config.Config.Mongo.Address = []string{"192.168.44.128:37017"}
+	config.Config.Mongo.Timeout = 60
+	config.Config.Mongo.Database = "openIM"
+	config.Config.Mongo.Source = "admin"
+	config.Config.Mongo.Username = "root"
+	config.Config.Mongo.Password = "openIM123"
+	config.Config.Mongo.MaxPoolSize = 100
+	config.Config.RetainChatRecords = 3650
+	config.Config.ChatRecordsClearTime = "0 2 * * 3"
+
+	mongo, err := unrelation.NewMongo()
+	if err != nil {
+		panic(err)
+	}
+	err = mongo.GetDatabase().Client().Ping(context.Background(), nil)
+	if err != nil {
+		panic(err)
+	}
+	return &commonMsgDatabase{
+		msgDocDatabase: unrelation.NewMsgMongoDriver(mongo.GetDatabase()),
+	}
+}
+
+func Test_Insert(t *testing.T) {
+	db := GetDB()
+	ctx := context.Background()
+	var arr []any
+	for i := 0; i < 345; i++ {
+		if i%2 == 0 {
+			arr = append(arr, (*unRelationTb.MsgDataModel)(nil))
+			continue
+		}
+		arr = append(arr, &unRelationTb.MsgDataModel{
+			Seq:     int64(i),
+			Content: fmt.Sprintf("test-%d", i),
+		})
+	}
+	if err := db.BatchInsertBlock(ctx, "test", arr, updateKeyMsg, 1); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func Test_Revoke(t *testing.T) {
+	db := GetDB()
+	ctx := context.Background()
+	var arr []any
+	for i := 0; i < 456; i++ {
+		arr = append(arr, &unRelationTb.RevokeModel{
+			UserID:   "uid_" + strconv.Itoa(i),
+			Nickname: "uname_" + strconv.Itoa(i),
+			Time:     time.Now().UnixMilli(),
+		})
+	}
+	if err := db.BatchInsertBlock(ctx, "test", arr, updateKeyRevoke, 123); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func Test_FindBySeq(t *testing.T) {
+	if err := log.InitFromConfig("", "", 6, true, false, "", 2); err != nil {
+		t.Fatal(err)
+	}
+	db := GetDB()
+	ctx := context.Background()
+	fmt.Println(db.msgDocDatabase.(*unrelation.MsgMongoDriver).GetMsgBySeqIndexIn1Doc(ctx, "100", "si_100_101:0", []int64{1}))
+	//res, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, "123456", "test:0", []int64{1, 2, 3})
+	//if err != nil {
+	//	t.Fatal(err)
+	//}
+	//db.GetMsgBySeqs(ctx, "100", "si_100_101:0", []int64{6})
+	//data, _ := json.Marshal(res)
+	//fmt.Println(string(data))
+}
+
+//func Test_Delete(t *testing.T) {
+//	db := GetDB()
+//	ctx := context.Background()
+//	var arr []any
+//	for i := 0; i < 123; i++ {
+//		arr = append(arr, []string{"uid_1", "uid_2"})
+//	}
+//	if err := db.BatchInsertBlock(ctx, "test", arr, updateKeyDel, 210); err != nil {
+//		t.Fatal(err)
+//	}
+//}
+
+//func Test_Delete1(t *testing.T) {
+//	config.Config.Mongo.DBAddress = []string{"192.168.44.128:37017"}
+//	config.Config.Mongo.DBTimeout = 60
+//	config.Config.Mongo.DBDatabase = "openIM"
+//	config.Config.Mongo.DBSource = "admin"
+//	config.Config.Mongo.DBUserName = "root"
+//	config.Config.Mongo.DBPassword = "openIM123"
+//	config.Config.Mongo.DBMaxPoolSize = 100
+//	config.Config.Mongo.DBRetainChatRecords = 3650
+//	config.Config.Mongo.ChatRecordsClearTime = "0 2 * * 3"
+//
+//	mongo, err := unrelation.NewMongo()
+//	if err != nil {
+//		panic(err)
+//	}
+//	err = mongo.GetDatabase().Client().Ping(context.Background(), nil)
+//	if err != nil {
+//		panic(err)
+//	}
+//
+//	c := mongo.GetClient().Database("openIM").Collection("msg")
+//
+//	var o unRelationTb.MsgDocModel
+//
+//	err = c.FindOne(context.Background(), bson.M{"doc_id": "test:0"}).Decode(&o)
+//	if err != nil {
+//		panic(err)
+//	}
+//
+//	for i, model := range o.Msg {
+//		fmt.Println(i, model == nil)
+//	}
+//
+//}

pkg/common/db/controller/push.go

@@ -0,0 +1,23 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+)
+
+type PushDatabase interface {
+	DelFcmToken(ctx context.Context, userID string, platformID int) error
+}
+
+type pushDataBase struct {
+	cache cache.MsgModel
+}
+
+func NewPushDatabase(cache cache.MsgModel) PushDatabase {
+	return &pushDataBase{cache: cache}
+}
+
+func (p *pushDataBase) DelFcmToken(ctx context.Context, userID string, platformID int) error {
+	return p.cache.DelFcmToken(ctx, userID, platformID)
+}

pkg/common/db/controller/storage.go

@@ -0,0 +1,539 @@
+package controller
+
+import "C"
+import (
+	"bytes"
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/obj"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/third"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+	"github.com/google/uuid"
+	"io"
+	"net/url"
+	"path"
+	"strconv"
+	"time"
+)
+
+const (
+	hashPrefix     = "hash"
+	tempPrefix     = "temp"
+	fragmentPrefix = "fragment_"
+	urlsName       = "urls.json"
+)
+
+type S3Database interface {
+	ApplyPut(ctx context.Context, req *third.ApplyPutReq) (*third.ApplyPutResp, error)
+	GetPut(ctx context.Context, req *third.GetPutReq) (*third.GetPutResp, error)
+	ConfirmPut(ctx context.Context, req *third.ConfirmPutReq) (*third.ConfirmPutResp, error)
+	GetUrl(ctx context.Context, req *third.GetUrlReq) (*third.GetUrlResp, error)
+	GetHashInfo(ctx context.Context, req *third.GetHashInfoReq) (*third.GetHashInfoResp, error)
+	CleanExpirationObject(ctx context.Context, t time.Time)
+}
+
+func NewS3Database(obj obj.Interface, hash relation.ObjectHashModelInterface, info relation.ObjectInfoModelInterface, put relation.ObjectPutModelInterface, url *url.URL) S3Database {
+	return &s3Database{
+		url:  url,
+		obj:  obj,
+		hash: hash,
+		info: info,
+		put:  put,
+	}
+}
+
+type s3Database struct {
+	url  *url.URL
+	obj  obj.Interface
+	hash relation.ObjectHashModelInterface
+	info relation.ObjectInfoModelInterface
+	put  relation.ObjectPutModelInterface
+}
+
+// today 今天的日期
+func (c *s3Database) today() string {
+	return time.Now().Format("20060102")
+}
+
+// fragmentName 根据序号生成文件名
+func (c *s3Database) fragmentName(index int) string {
+	return fragmentPrefix + strconv.Itoa(index+1)
+}
+
+// getFragmentNum 获取分片大小和分片数量
+func (c *s3Database) getFragmentNum(fragmentSize int64, objectSize int64) (int64, int) {
+	if size := c.obj.MinFragmentSize(); fragmentSize < size {
+		fragmentSize = size
+	}
+	if fragmentSize <= 0 || objectSize <= fragmentSize {
+		return objectSize, 1
+	} else {
+		num := int(objectSize / fragmentSize)
+		if objectSize%fragmentSize > 0 {
+			num++
+		}
+		if n := c.obj.MaxFragmentNum(); num > n {
+			num = n
+		}
+		return fragmentSize, num
+	}
+}
+
+func (c *s3Database) CheckHash(hash string) error {
+	val, err := hex.DecodeString(hash)
+	if err != nil {
+		return err
+	}
+	if len(val) != md5.Size {
+		return errs.ErrArgs.Wrap("invalid hash")
+	}
+	return nil
+}
+
+func (c *s3Database) urlName(name string) string {
+	u := url.URL{
+		Scheme:      c.url.Scheme,
+		Opaque:      c.url.Opaque,
+		User:        c.url.User,
+		Host:        c.url.Host,
+		Path:        c.url.Path,
+		RawPath:     c.url.RawPath,
+		OmitHost:    c.url.OmitHost,
+		ForceQuery:  c.url.ForceQuery,
+		RawQuery:    c.url.RawQuery,
+		Fragment:    c.url.Fragment,
+		RawFragment: c.url.RawFragment,
+	}
+	v := make(url.Values, 1)
+	v.Set("name", name)
+	u.RawQuery = v.Encode()
+	return u.String()
+}
+
+func (c *s3Database) UUID() string {
+	return uuid.New().String()
+}
+
+func (c *s3Database) HashName(hash string) string {
+	return path.Join(hashPrefix, hash+"_"+c.today()+"_"+c.UUID())
+}
+
+func (c *s3Database) isNotFound(err error) bool {
+	return relation.IsNotFound(err)
+}
+
+func (c *s3Database) ApplyPut(ctx context.Context, req *third.ApplyPutReq) (*third.ApplyPutResp, error) {
+	if err := c.CheckHash(req.Hash); err != nil {
+		return nil, err
+	}
+	if err := c.obj.CheckName(req.Name); err != nil {
+		return nil, err
+	}
+	if req.ValidTime != 0 && req.ValidTime <= time.Now().UnixMilli() {
+		return nil, errors.New("invalid ValidTime")
+	}
+	var expirationTime *time.Time
+	if req.ValidTime != 0 {
+		expirationTime = utils.ToPtr(time.UnixMilli(req.ValidTime))
+	}
+	if hash, err := c.hash.Take(ctx, req.Hash, c.obj.Name()); err == nil {
+		o := relation.ObjectInfoModel{
+			Name:        req.Name,
+			Hash:        hash.Hash,
+			ValidTime:   expirationTime,
+			ContentType: req.ContentType,
+			CreateTime:  time.Now(),
+		}
+		if err := c.info.SetObject(ctx, &o); err != nil {
+			return nil, err
+		}
+		return &third.ApplyPutResp{Url: c.urlName(o.Name)}, nil // 服务器已存在
+	} else if !c.isNotFound(err) {
+		return nil, err
+	}
+	// 新上传
+	var fragmentNum int
+	const effective = time.Hour * 24 * 2
+	req.FragmentSize, fragmentNum = c.getFragmentNum(req.FragmentSize, req.Size)
+	put := relation.ObjectPutModel{
+		PutID:         req.PutID,
+		Hash:          req.Hash,
+		Name:          req.Name,
+		ObjectSize:    req.Size,
+		ContentType:   req.ContentType,
+		FragmentSize:  req.FragmentSize,
+		ValidTime:     expirationTime,
+		EffectiveTime: time.Now().Add(effective),
+	}
+	if put.PutID == "" {
+		put.PutID = c.UUID()
+	}
+	if v, err := c.put.Take(ctx, put.PutID); err == nil {
+		now := time.Now().UnixMilli()
+		if v.EffectiveTime.UnixMilli() <= now {
+			if err := c.put.DelPut(ctx, []string{v.PutID}); err != nil {
+				return nil, err
+			}
+		} else {
+			return nil, errs.ErrDuplicateKey.Wrap(fmt.Sprintf("duplicate put id %s", put.PutID))
+		}
+	} else if !c.isNotFound(err) {
+		return nil, err
+	}
+	put.Path = path.Join(tempPrefix, c.today(), req.Hash, put.PutID)
+	putURLs := make([]string, 0, fragmentNum)
+	for i := 0; i < fragmentNum; i++ {
+		url, err := c.obj.PresignedPutURL(ctx, &obj.ApplyPutArgs{
+			Bucket:        c.obj.TempBucket(),
+			Name:          path.Join(put.Path, c.fragmentName(i)),
+			Effective:     effective,
+			MaxObjectSize: req.FragmentSize,
+		})
+		if err != nil {
+			return nil, err
+		}
+		putURLs = append(putURLs, url)
+	}
+	urlsJsonData, err := json.Marshal(putURLs)
+	if err != nil {
+		return nil, err
+	}
+	t := md5.Sum(urlsJsonData)
+	put.PutURLsHash = hex.EncodeToString(t[:])
+	_, err = c.obj.PutObject(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: path.Join(put.Path, urlsName)}, bytes.NewReader(urlsJsonData), int64(len(urlsJsonData)))
+	if err != nil {
+		return nil, err
+	}
+	put.CreateTime = time.Now()
+	if err := c.put.Create(ctx, []*relation.ObjectPutModel{&put}); err != nil {
+		return nil, err
+	}
+	return &third.ApplyPutResp{
+		PutID:        put.PutID,
+		FragmentSize: put.FragmentSize,
+		PutURLs:      putURLs,
+		ValidTime:    put.EffectiveTime.UnixMilli(),
+	}, nil
+}
+
+func (c *s3Database) GetPut(ctx context.Context, req *third.GetPutReq) (*third.GetPutResp, error) {
+	up, err := c.put.Take(ctx, req.PutID)
+	if err != nil {
+		return nil, err
+	}
+	reader, err := c.obj.GetObject(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: path.Join(up.Path, urlsName)})
+	if err != nil {
+		return nil, err
+	}
+	urlsData, err := io.ReadAll(reader)
+	if err != nil {
+		return nil, err
+	}
+	t := md5.Sum(urlsData)
+	if h := hex.EncodeToString(t[:]); h != up.PutURLsHash {
+		return nil, fmt.Errorf("invalid put urls hash %s %s", h, up.PutURLsHash)
+	}
+	var urls []string
+	if err := json.Unmarshal(urlsData, &urls); err != nil {
+		return nil, err
+	}
+	_, fragmentNum := c.getFragmentNum(up.FragmentSize, up.ObjectSize)
+	if len(urls) != fragmentNum {
+		return nil, fmt.Errorf("invalid urls length %d fragment %d", len(urls), fragmentNum)
+	}
+	fragments := make([]*third.GetPutFragment, fragmentNum)
+	for i := 0; i < fragmentNum; i++ {
+		name := path.Join(up.Path, c.fragmentName(i))
+		o, err := c.obj.GetObjectInfo(ctx, &obj.BucketObject{
+			Bucket: c.obj.TempBucket(),
+			Name:   name,
+		})
+		if err != nil {
+			if c.obj.IsNotFound(err) {
+				fragments[i] = &third.GetPutFragment{Url: urls[i]}
+				continue
+			}
+			return nil, err
+		}
+		fragments[i] = &third.GetPutFragment{Size: o.Size, Hash: o.Hash, Url: urls[i]}
+	}
+	var validTime int64
+	if up.ValidTime != nil {
+		validTime = up.ValidTime.UnixMilli()
+	}
+	return &third.GetPutResp{
+		FragmentSize: up.FragmentSize,
+		Size:         up.ObjectSize,
+		Name:         up.Name,
+		Hash:         up.Hash,
+		Fragments:    fragments,
+		PutURLsHash:  up.PutURLsHash,
+		ContentType:  up.ContentType,
+		ValidTime:    validTime,
+	}, nil
+}
+
+func (c *s3Database) ConfirmPut(ctx context.Context, req *third.ConfirmPutReq) (_ *third.ConfirmPutResp, _err error) {
+	put, err := c.put.Take(ctx, req.PutID)
+	if err != nil {
+		return nil, err
+	}
+	_, pack := c.getFragmentNum(put.FragmentSize, put.ObjectSize)
+	defer func() {
+		if _err == nil {
+			// 清理上传的碎片
+			err := c.obj.DeleteObject(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: put.Path})
+			if err != nil {
+				log.ZError(ctx, "deleteObject failed", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
+			}
+		}
+	}()
+	now := time.Now().UnixMilli()
+	if put.EffectiveTime.UnixMilli() < now {
+		return nil, errs.ErrFileUploadedExpired.Wrap("put expired")
+	}
+	if put.ValidTime != nil && put.ValidTime.UnixMilli() < now {
+		return nil, errs.ErrFileUploadedExpired.Wrap("object expired")
+	}
+	if hash, err := c.hash.Take(ctx, put.Hash, c.obj.Name()); err == nil {
+		o := relation.ObjectInfoModel{
+			Name:        put.Name,
+			Hash:        hash.Hash,
+			ValidTime:   put.ValidTime,
+			ContentType: put.ContentType,
+			CreateTime:  time.Now(),
+		}
+		if err := c.info.SetObject(ctx, &o); err != nil {
+			return nil, err
+		}
+		defer func() {
+			err := c.obj.DeleteObject(ctx, &obj.BucketObject{
+				Bucket: c.obj.TempBucket(),
+				Name:   put.Path,
+			})
+			if err != nil {
+				log.ZError(ctx, "DeleteObject", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
+			}
+		}()
+		// 服务端已存在
+		return &third.ConfirmPutResp{
+			Url: c.urlName(o.Name),
+		}, nil
+	} else if !c.isNotFound(err) {
+		return nil, err
+	}
+	src := make([]obj.BucketObject, pack)
+	for i := 0; i < pack; i++ {
+		name := path.Join(put.Path, c.fragmentName(i))
+		o, err := c.obj.GetObjectInfo(ctx, &obj.BucketObject{
+			Bucket: c.obj.TempBucket(),
+			Name:   name,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if i+1 == pack { // 最后一个
+			size := put.ObjectSize - put.FragmentSize*int64(i)
+			if size != o.Size {
+				return nil, fmt.Errorf("last fragment %d size %d not equal to %d hash %s", i, o.Size, size, o.Hash)
+			}
+		} else {
+			if o.Size != put.FragmentSize {
+				return nil, fmt.Errorf("fragment %d size %d not equal to %d hash %s", i, o.Size, put.FragmentSize, o.Hash)
+			}
+		}
+		src[i] = obj.BucketObject{
+			Bucket: c.obj.TempBucket(),
+			Name:   name,
+		}
+	}
+	dst := &obj.BucketObject{
+		Bucket: c.obj.DataBucket(),
+		Name:   c.HashName(put.Hash),
+	}
+	if len(src) == 1 { // 未分片直接触发copy
+		// 检查数据完整性,避免脏数据
+		o, err := c.obj.GetObjectInfo(ctx, &src[0])
+		if err != nil {
+			return nil, err
+		}
+		if put.ObjectSize != o.Size {
+			return nil, fmt.Errorf("size mismatching should %d reality %d", put.ObjectSize, o.Size)
+		}
+		if put.Hash != o.Hash {
+			return nil, fmt.Errorf("hash mismatching should %s reality %s", put.Hash, o.Hash)
+		}
+		if err := c.obj.CopyObject(ctx, &src[0], dst); err != nil {
+			return nil, err
+		}
+	} else {
+		tempBucket := &obj.BucketObject{
+			Bucket: c.obj.TempBucket(),
+			Name:   path.Join(put.Path, "merge_"+c.UUID()),
+		}
+		defer func() { // 清理合成的文件
+			if err := c.obj.DeleteObject(ctx, tempBucket); err != nil {
+				log.ZError(ctx, "DeleteObject", err, "Bucket", tempBucket.Bucket, "Path", tempBucket.Name)
+			}
+		}()
+		err := c.obj.ComposeObject(ctx, src, tempBucket)
+		if err != nil {
+			return nil, err
+		}
+		info, err := c.obj.GetObjectInfo(ctx, tempBucket)
+		if err != nil {
+			return nil, err
+		}
+		if put.ObjectSize != info.Size {
+			return nil, fmt.Errorf("size mismatch should %d reality %d", put.ObjectSize, info.Size)
+		}
+		if put.Hash != info.Hash {
+			return nil, fmt.Errorf("hash mismatch should %s reality %s", put.Hash, info.Hash)
+		}
+		if err := c.obj.CopyObject(ctx, tempBucket, dst); err != nil {
+			return nil, err
+		}
+	}
+	h := &relation.ObjectHashModel{
+		Hash:       put.Hash,
+		Engine:     c.obj.Name(),
+		Size:       put.ObjectSize,
+		Bucket:     c.obj.DataBucket(),
+		Name:       dst.Name,
+		CreateTime: time.Now(),
+	}
+	if err := c.hash.Create(ctx, []*relation.ObjectHashModel{h}); err != nil {
+		return nil, err
+	}
+	o := &relation.ObjectInfoModel{
+		Name:        put.Name,
+		Hash:        put.Hash,
+		ContentType: put.ContentType,
+		ValidTime:   put.ValidTime,
+		CreateTime:  time.Now(),
+	}
+	if err := c.info.SetObject(ctx, o); err != nil {
+		return nil, err
+	}
+	if err := c.put.DelPut(ctx, []string{put.PutID}); err != nil {
+		log.ZError(ctx, "DelPut", err, "PutID", put.PutID)
+	}
+	return &third.ConfirmPutResp{
+		Url: c.urlName(o.Name),
+	}, nil
+}
+
+func (c *s3Database) GetUrl(ctx context.Context, req *third.GetUrlReq) (*third.GetUrlResp, error) {
+	info, err := c.info.Take(ctx, req.Name)
+	if err != nil {
+		return nil, err
+	}
+	if info.ValidTime != nil && info.ValidTime.Before(time.Now()) {
+		return nil, errs.ErrRecordNotFound.Wrap("object expired")
+	}
+	hash, err := c.hash.Take(ctx, info.Hash, c.obj.Name())
+	if err != nil {
+		return nil, err
+	}
+	opt := obj.HeaderOption{ContentType: info.ContentType}
+	if req.Attachment {
+		opt.Filename = info.Name
+	}
+	u, err := c.obj.PresignedGetURL(ctx, hash.Bucket, hash.Name, time.Duration(req.Expires)*time.Millisecond, &opt)
+	if err != nil {
+		return nil, err
+	}
+	return &third.GetUrlResp{
+		Url:  u,
+		Size: hash.Size,
+		Hash: hash.Hash,
+	}, nil
+}
+
+func (c *s3Database) CleanExpirationObject(ctx context.Context, t time.Time) {
+	// 清理上传产生的临时文件
+	c.cleanPutTemp(ctx, t, 10)
+	// 清理hash引用全过期的文件
+	c.cleanExpirationObject(ctx, t)
+	// 清理没有引用的hash对象
+	c.clearNoCitation(ctx, c.obj.Name(), 10)
+}
+
+func (c *s3Database) cleanPutTemp(ctx context.Context, t time.Time, num int) {
+	for {
+		puts, err := c.put.FindExpirationPut(ctx, t, num)
+		if err != nil {
+			log.ZError(ctx, "FindExpirationPut", err, "Time", t, "Num", num)
+			return
+		}
+		if len(puts) == 0 {
+			return
+		}
+		for _, put := range puts {
+			err := c.obj.DeleteObject(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: put.Path})
+			if err != nil {
+				log.ZError(ctx, "DeleteObject", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
+				return
+			}
+		}
+		ids := utils.Slice(puts, func(e *relation.ObjectPutModel) string { return e.PutID })
+		err = c.put.DelPut(ctx, ids)
+		if err != nil {
+			log.ZError(ctx, "DelPut", err, "PutID", ids)
+			return
+		}
+	}
+}
+
+func (c *s3Database) cleanExpirationObject(ctx context.Context, t time.Time) {
+	err := c.info.DeleteExpiration(ctx, t)
+	if err != nil {
+		log.ZError(ctx, "DeleteExpiration", err, "Time", t)
+	}
+}
+
+func (c *s3Database) clearNoCitation(ctx context.Context, engine string, limit int) {
+	for {
+		list, err := c.hash.DeleteNoCitation(ctx, engine, limit)
+		if err != nil {
+			log.ZError(ctx, "DeleteNoCitation", err, "Engine", engine, "Limit", limit)
+			return
+		}
+		if len(list) == 0 {
+			return
+		}
+		var hasErr bool
+		for _, h := range list {
+			err := c.obj.DeleteObject(ctx, &obj.BucketObject{Bucket: h.Bucket, Name: h.Name})
+			if err != nil {
+				hasErr = true
+				log.ZError(ctx, "DeleteObject", err, "Bucket", h.Bucket, "Path", h.Name)
+				continue
+			}
+		}
+		if hasErr {
+			return
+		}
+	}
+}
+
+func (c *s3Database) GetHashInfo(ctx context.Context, req *third.GetHashInfoReq) (*third.GetHashInfoResp, error) {
+	if err := c.CheckHash(req.Hash); err != nil {
+		return nil, err
+	}
+	o, err := c.hash.Take(ctx, req.Hash, c.obj.Name())
+	if err != nil {
+		return nil, err
+	}
+	return &third.GetHashInfoResp{
+		Hash: o.Hash,
+		Size: o.Size,
+	}, nil
+}

pkg/common/db/controller/third.go

@@ -0,0 +1,28 @@
+package controller
+
+import (
+	"context"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+)
+
+type ThirdDatabase interface {
+	FcmUpdateToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) error
+	SetAppBadge(ctx context.Context, userID string, value int) error
+}
+
+type thirdDatabase struct {
+	cache cache.MsgModel
+}
+
+func NewThirdDatabase(cache cache.MsgModel) ThirdDatabase {
+	return &thirdDatabase{cache: cache}
+}
+
+func (t *thirdDatabase) FcmUpdateToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) error {
+	return t.cache.SetFcmToken(ctx, account, platformID, fcmToken, expireTime)
+}
+
+func (t *thirdDatabase) SetAppBadge(ctx context.Context, userID string, value int) error {
+	return t.cache.SetUserBadgeUnreadCountSum(ctx, userID, value)
+}

pkg/common/db/controller/user.go

@@ -0,0 +1,143 @@
+package controller
+
+import (
+	"context"
+	"time"
+
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/relation"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/tx"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
+)
+
+type UserDatabase interface {
+	//获取指定用户的信息 如有userID未找到 也返回错误
+	FindWithError(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error)
+	//获取指定用户的信息 如有userID未找到 不返回错误
+	Find(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error)
+	//插入多条 外部保证userID 不重复 且在db中不存在
+	Create(ctx context.Context, users []*relation.UserModel) (err error)
+	//更新（非零值） 外部保证userID存在
+	Update(ctx context.Context, user *relation.UserModel) (err error)
+	//更新（零值） 外部保证userID存在
+	UpdateByMap(ctx context.Context, userID string, args map[string]interface{}) (err error)
+	//如果没找到，不返回错误
+	Page(ctx context.Context, pageNumber, showNumber int32) (users []*relation.UserModel, count int64, err error)
+	//只要有一个存在就为true
+	IsExist(ctx context.Context, userIDs []string) (exist bool, err error)
+	//获取所有用户ID
+	GetAllUserID(ctx context.Context) ([]string, error)
+	//函数内部先查询db中是否存在，存在则什么都不做；不存在则插入
+	InitOnce(ctx context.Context, users []*relation.UserModel) (err error)
+	// 获取用户总数
+	CountTotal(ctx context.Context) (int64, error)
+	// 获取范围内用户增量
+	CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error)
+}
+
+type userDatabase struct {
+	userDB relation.UserModelInterface
+	cache  cache.UserCache
+	tx     tx.Tx
+}
+
+func NewUserDatabase(userDB relation.UserModelInterface, cache cache.UserCache, tx tx.Tx) UserDatabase {
+	return &userDatabase{userDB: userDB, cache: cache, tx: tx}
+}
+
+func (u *userDatabase) InitOnce(ctx context.Context, users []*relation.UserModel) (err error) {
+	userIDs := utils.Slice(users, func(e *relation.UserModel) string {
+		return e.UserID
+	})
+	result, err := u.userDB.Find(ctx, userIDs)
+	if err != nil {
+		return err
+	}
+	miss := utils.SliceAnySub(users, result, func(e *relation.UserModel) string { return e.UserID })
+	if len(miss) > 0 {
+		_ = u.userDB.Create(ctx, miss)
+	}
+	return nil
+}
+
+// 获取指定用户的信息 如有userID未找到 也返回错误
+func (u *userDatabase) FindWithError(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error) {
+	users, err = u.cache.GetUsersInfo(ctx, userIDs)
+	if err != nil {
+		return
+	}
+	if len(users) != len(userIDs) {
+		err = errs.ErrRecordNotFound.Wrap("userID not found")
+	}
+	return
+}
+
+// 获取指定用户的信息 如有userID未找到 不返回错误
+func (u *userDatabase) Find(ctx context.Context, userIDs []string) (users []*relation.UserModel, err error) {
+	users, err = u.cache.GetUsersInfo(ctx, userIDs)
+	return
+}
+
+// 插入多条 外部保证userID 不重复 且在db中不存在
+func (u *userDatabase) Create(ctx context.Context, users []*relation.UserModel) (err error) {
+	if err := u.tx.Transaction(func(tx any) error {
+		err = u.userDB.Create(ctx, users)
+		if err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	var userIDs []string
+	for _, user := range users {
+		userIDs = append(userIDs, user.UserID)
+	}
+	return u.cache.DelUsersInfo(userIDs...).ExecDel(ctx)
+}
+
+// 更新（非零值） 外部保证userID存在
+func (u *userDatabase) Update(ctx context.Context, user *relation.UserModel) (err error) {
+	if err := u.userDB.Update(ctx, user); err != nil {
+		return err
+	}
+	return u.cache.DelUsersInfo(user.UserID).ExecDel(ctx)
+}
+
+// 更新（零值） 外部保证userID存在
+func (u *userDatabase) UpdateByMap(ctx context.Context, userID string, args map[string]interface{}) (err error) {
+	if err := u.userDB.UpdateByMap(ctx, userID, args); err != nil {
+		return err
+	}
+	return u.cache.DelUsersInfo(userID).ExecDel(ctx)
+}
+
+// 获取，如果没找到，不返回错误
+func (u *userDatabase) Page(ctx context.Context, pageNumber, showNumber int32) (users []*relation.UserModel, count int64, err error) {
+	return u.userDB.Page(ctx, pageNumber, showNumber)
+}
+
+// userIDs是否存在 只要有一个存在就为true
+func (u *userDatabase) IsExist(ctx context.Context, userIDs []string) (exist bool, err error) {
+	users, err := u.userDB.Find(ctx, userIDs)
+	if err != nil {
+		return false, err
+	}
+	if len(users) > 0 {
+		return true, nil
+	}
+	return false, nil
+}
+
+func (u *userDatabase) GetAllUserID(ctx context.Context) (userIDs []string, err error) {
+	return u.userDB.GetAllUserID(ctx)
+}
+
+func (u *userDatabase) CountTotal(ctx context.Context) (count int64, err error) {
+	return u.userDB.CountTotal(ctx)
+}
+
+func (u *userDatabase) CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error) {
+	return u.userDB.CountRangeEverydayTotal(ctx, start, end)
+}