feat: support message cache

internal/msgtransfer/init.go

@@ -108,11 +108,11 @@ func Start(ctx context.Context, index int, config *Config) error {
 		cm.Watch(ctx)
 	}
 
-	msgModel := redis.NewMsgCache(rdb)
 	msgDocModel, err := mgo.NewMsgMongo(mgocli.GetDB())
 	if err != nil {
 		return err
 	}
+	msgModel := redis.NewMsgCache(rdb, msgDocModel)
 	seqConversation, err := mgo.NewSeqConversationMongo(mgocli.GetDB())
 	if err != nil {
 		return err

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -77,27 +77,13 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
 	for _, msg := range msgFromMQ.MsgData {
 		seqs = append(seqs, msg.Seq)
 	}
-	err = mc.msgTransferDatabase.DeleteMessagesFromCache(ctx, msgFromMQ.ConversationID, seqs)
-	if err != nil {
-		log.ZError(
-			ctx,
-			"remove cache msg from redis err",
-			err,
-			"msg",
-			msgFromMQ.MsgData,
-			"conversationID",
-			msgFromMQ.ConversationID,
-		)
-	}
 }
 
-func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
+
 func (*OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
 
-func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(
-	sess sarama.ConsumerGroupSession,
-	claim sarama.ConsumerGroupClaim,
-) error { // an instance in the consumer group
+func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // an instance in the consumer group
 	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
 		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
 	for msg := range claim.Messages() {

internal/rpc/msg/delete.go

@@ -106,10 +106,8 @@ func (m *msgServer) DeleteMsgPhysical(ctx context.Context, req *msg.DeleteMsgPhy
 		return nil, err
 	}
 	remainTime := timeutil.GetCurrentTimestampBySecond() - req.Timestamp
-	for _, conversationID := range req.ConversationIDs {
-		if err := m.MsgDatabase.DeleteConversationMsgsAndSetMinSeq(ctx, conversationID, remainTime); err != nil {
-			log.ZWarn(ctx, "DeleteConversationMsgsAndSetMinSeq error", err, "conversationID", conversationID, "err", err)
-		}
+	if _, err := m.DestructMsgs(ctx, &msg.DestructMsgsReq{Timestamp: remainTime, Limit: 9999}); err != nil {
+		return nil, err
 	}
 	return &msg.DeleteMsgPhysicalResp{}, nil
 }

internal/rpc/msg/revoke.go

@@ -63,7 +63,8 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 	log.ZDebug(ctx, "GetMsgBySeqs", "conversationID", req.ConversationID, "seq", req.Seq, "msg", string(data))
 	var role int32
 	if !authverify.IsAppManagerUid(ctx, m.config.Share.IMAdminUserID) {
-		switch msgs[0].SessionType {
+		sessionType := msgs[0].SessionType
+		switch sessionType {
 		case constant.SingleChatType:
 			if err := authverify.CheckAccessV3(ctx, msgs[0].SendID, m.config.Share.IMAdminUserID); err != nil {
 				return nil, err
@@ -89,7 +90,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
 				role = member.RoleLevel
 			}
 		default:
-			return nil, errs.ErrInternalServer.WrapMsg("msg sessionType not supported")
+			return nil, errs.ErrInternalServer.WrapMsg("msg sessionType not supported", "sessionType", sessionType)
 		}
 	}
 	now := time.Now().UnixMilli()

internal/rpc/msg/server.go

@@ -89,7 +89,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 	if err != nil {
 		return err
 	}
-	msgModel := redis.NewMsgCache(rdb)
+	msgModel := redis.NewMsgCache(rdb, msgDocModel)
 	seqConversation, err := mgo.NewSeqConversationMongo(mgocli.GetDB())
 	if err != nil {
 		return err

pkg/common/storage/cache/cachekey/msg.go

@@ -15,50 +15,16 @@
 package cachekey
 
 import (
-	"github.com/openimsdk/protocol/constant"
 	"strconv"
 )
 
 const (
-	messageCache         = "MESSAGE_CACHE:"
-	messageDelUserList   = "MESSAGE_DEL_USER_LIST:"
-	userDelMessagesList  = "USER_DEL_MESSAGES_LIST:"
-	sendMsgFailedFlag    = "SEND_MSG_FAILED_FLAG:"
-	exTypeKeyLocker      = "EX_LOCK:"
-	reactionExSingle     = "EX_SINGLE_"
-	reactionWriteGroup   = "EX_GROUP_"
-	reactionReadGroup    = "EX_SUPER_GROUP_"
-	reactionNotification = "EX_NOTIFICATION_"
+	sendMsgFailedFlag = "SEND_MSG_FAILED_FLAG:"
+	messageCache      = "MSG_CACHE:"
 )
 
-func GetMessageCacheKey(conversationID string, seq int64) string {
-	return messageCache + conversationID + "_" + strconv.Itoa(int(seq))
-}
-
-func GetMessageDelUserListKey(conversationID string, seq int64) string {
-	return messageDelUserList + conversationID + ":" + strconv.Itoa(int(seq))
-}
-
-func GetUserDelListKey(conversationID, userID string) string {
-	return userDelMessagesList + conversationID + ":" + userID
-}
-
-func GetMessageReactionExKey(clientMsgID string, sessionType int32) string {
-	switch sessionType {
-	case constant.SingleChatType:
-		return reactionExSingle + clientMsgID
-	case constant.WriteGroupChatType:
-		return reactionWriteGroup + clientMsgID
-	case constant.ReadGroupChatType:
-		return reactionReadGroup + clientMsgID
-	case constant.NotificationChatType:
-		return reactionNotification + clientMsgID
-	}
-
-	return ""
-}
-func GetLockMessageTypeKey(clientMsgID string, TypeKey string) string {
-	return exTypeKeyLocker + clientMsgID + "_" + TypeKey
+func GetMsgCacheKey(conversationID string, seq int64) string {
+	return messageCache + conversationID + ":" + strconv.Itoa(int(seq))
 }
 
 func GetSendMsgKey(id string) string {

pkg/common/storage/cache/msg.go

@@ -16,23 +16,14 @@ package cache
 
 import (
 	"context"
-	"time"
-
-	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 )
 
 type MsgCache interface {
-	GetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsg []*sdkws.MsgData, failedSeqList []int64, err error)
-	SetMessagesToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error)
-	DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error
 	SetSendMsgStatus(ctx context.Context, id string, status int32) error
 	GetSendMsgStatus(ctx context.Context, id string) (int32, error)
-	JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error)
-	GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error)
-	DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error
-	SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error)
-	GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error)
-	SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error
-	LockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error
-	UnLockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error
+
+	GetMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error)
+	DelMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) error
+	SetMessageBySeqs(ctx context.Context, conversationID string, msgs []*model.MsgInfoModel) error
 }

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

@@ -2,87 +2,39 @@ package redis
 
 import (
 	"context"
+	"encoding/json"
+	"github.com/dtm-labs/rockscache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
-	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/utils/datautil"
 	"github.com/redis/go-redis/v9"
 	"time"
 ) //
 
 // msgCacheTimeout is  expiration time of message cache, 86400 seconds
-const msgCacheTimeout = 86400
+const msgCacheTimeout = time.Hour * 24
 
-func NewMsgCache(client redis.UniversalClient) cache.MsgCache {
-	return &msgCache{rdb: client}
+func NewMsgCache(client redis.UniversalClient, db database.Msg) cache.MsgCache {
+	return &msgCache{
+		rdb:            client,
+		rcClient:       rockscache.NewClient(client, *GetRocksCacheOptions()),
+		msgDocDatabase: db,
+	}
 }
 
 type msgCache struct {
-	rdb redis.UniversalClient
-}
-
-func (c *msgCache) getMessageCacheKey(conversationID string, seq int64) string {
-	return cachekey.GetMessageCacheKey(conversationID, seq)
-}
-func (c *msgCache) getMessageDelUserListKey(conversationID string, seq int64) string {
-	return cachekey.GetMessageDelUserListKey(conversationID, seq)
-}
-
-func (c *msgCache) getUserDelList(conversationID, userID string) string {
-	return cachekey.GetUserDelListKey(conversationID, userID)
+	rdb            redis.UniversalClient
+	rcClient       *rockscache.Client
+	msgDocDatabase database.Msg
 }
 
 func (c *msgCache) getSendMsgKey(id string) string {
 	return cachekey.GetSendMsgKey(id)
 }
 
-func (c *msgCache) getLockMessageTypeKey(clientMsgID string, TypeKey string) string {
-	return cachekey.GetLockMessageTypeKey(clientMsgID, TypeKey)
-}
-
-func (c *msgCache) getMessageReactionExPrefix(clientMsgID string, sessionType int32) string {
-	return cachekey.GetMessageReactionExKey(clientMsgID, sessionType)
-}
-
-func (c *msgCache) SetMessagesToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) {
-	msgMap := datautil.SliceToMap(msgs, func(msg *sdkws.MsgData) string {
-		return c.getMessageCacheKey(conversationID, msg.Seq)
-	})
-	keys := datautil.Slice(msgs, func(msg *sdkws.MsgData) string {
-		return c.getMessageCacheKey(conversationID, msg.Seq)
-	})
-	err := ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
-		var values []string
-		for _, key := range keys {
-			if msg, ok := msgMap[key]; ok {
-				s, err := msgprocessor.Pb2String(msg)
-				if err != nil {
-					return err
-				}
-				values = append(values, s)
-			}
-		}
-		return LuaSetBatchWithCommonExpire(ctx, c.rdb, keys, values, msgCacheTimeout)
-	})
-	if err != nil {
-		return 0, err
-	}
-	return len(msgs), nil
-}
-
-func (c *msgCache) DeleteMessagesFromCache(ctx context.Context, conversationID string, seqs []int64) error {
-	var keys []string
-	for _, seq := range seqs {
-		keys = append(keys, c.getMessageCacheKey(conversationID, seq))
-	}
-
-	return ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
-		return LuaDeleteBatch(ctx, c.rdb, keys)
-	})
-}
-
 func (c *msgCache) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
 	return errs.Wrap(c.rdb.Set(ctx, c.getSendMsgKey(id), status, time.Hour*24).Err())
 }
@@ -92,81 +44,53 @@ func (c *msgCache) GetSendMsgStatus(ctx context.Context, id string) (int32, erro
 	return int32(result), errs.Wrap(err)
 }
 
-func (c *msgCache) LockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error {
-	key := c.getLockMessageTypeKey(clientMsgID, TypeKey)
-	return errs.Wrap(c.rdb.SetNX(ctx, key, 1, time.Minute).Err())
-}
-
-func (c *msgCache) UnLockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error {
-	key := c.getLockMessageTypeKey(clientMsgID, TypeKey)
-	return errs.Wrap(c.rdb.Del(ctx, key).Err())
+func (c *msgCache) GetMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) ([]*model.MsgInfoModel, error) {
+	if len(seqs) == 0 {
+		return nil, nil
+	}
+	getKey := func(seq int64) string {
+		return cachekey.GetMsgCacheKey(conversationID, seq)
+	}
+	getMsgID := func(msg *model.MsgInfoModel) int64 {
+		return msg.Msg.Seq
+	}
+	find := func(ctx context.Context, seqs []int64) ([]*model.MsgInfoModel, error) {
+		return c.msgDocDatabase.FindSeqs(ctx, conversationID, seqs)
+	}
+	return batchGetCache2(ctx, c.rcClient, msgCacheTimeout, seqs, getKey, getMsgID, find)
 }
 
-func (c *msgCache) JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error) {
-	n, err := c.rdb.Exists(ctx, c.getMessageReactionExPrefix(clientMsgID, sessionType)).Result()
+func (c *msgCache) DelMessageBySeqs(ctx context.Context, conversationID string, seqs []int64) error {
+	if len(seqs) == 0 {
+		return nil
+	}
+	keys := datautil.Slice(seqs, func(seq int64) string {
+		return cachekey.GetMsgCacheKey(conversationID, seq)
+	})
+	slotKeys, err := groupKeysBySlot(ctx, getRocksCacheRedisClient(c.rcClient), keys)
 	if err != nil {
-		return false, errs.Wrap(err)
+		return err
 	}
-
-	return n > 0, nil
-}
-
-func (c *msgCache) SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error {
-	return errs.Wrap(c.rdb.HSet(ctx, c.getMessageReactionExPrefix(clientMsgID, sessionType), typeKey, value).Err())
-}
-
-func (c *msgCache) SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error) {
-	val, err := c.rdb.Expire(ctx, c.getMessageReactionExPrefix(clientMsgID, sessionType), expiration).Result()
-	return val, errs.Wrap(err)
-}
-
-func (c *msgCache) GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error) {
-	val, err := c.rdb.HGet(ctx, c.getMessageReactionExPrefix(clientMsgID, sessionType), typeKey).Result()
-	return val, errs.Wrap(err)
-}
-
-func (c *msgCache) GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error) {
-	val, err := c.rdb.HGetAll(ctx, c.getMessageReactionExPrefix(clientMsgID, sessionType)).Result()
-	return val, errs.Wrap(err)
-}
-
-func (c *msgCache) DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error {
-	return errs.Wrap(c.rdb.HDel(ctx, c.getMessageReactionExPrefix(clientMsgID, sessionType), subKey).Err())
+	for _, keys := range slotKeys {
+		if err := c.rcClient.TagAsDeletedBatch2(ctx, keys); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
-func (c *msgCache) GetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err error) {
-	var keys []string
-	keySeqMap := make(map[string]int64, 10)
-	for _, seq := range seqs {
-		key := c.getMessageCacheKey(conversationID, seq)
-		keys = append(keys, key)
-		keySeqMap[key] = seq
-	}
-	err = ProcessKeysBySlot(ctx, c.rdb, keys, func(ctx context.Context, slot int64, keys []string) error {
-		result, err := LuaGetBatch(ctx, c.rdb, keys)
+func (c *msgCache) SetMessageBySeqs(ctx context.Context, conversationID string, msgs []*model.MsgInfoModel) error {
+	for _, msg := range msgs {
+		if msg == nil || msg.Msg == nil || msg.Msg.Seq <= 0 {
+			continue
+		}
+		data, err := json.Marshal(msg)
 		if err != nil {
 			return err
 		}
-		for i, value := range result {
-			seq := keySeqMap[keys[i]]
-			if value == nil {
-				failedSeqs = append(failedSeqs, seq)
-				continue
-			}
-
-			msg := &sdkws.MsgData{}
-			msgString, ok := value.(string)
-			if !ok || msgprocessor.String2Pb(msgString, msg) != nil {
-				failedSeqs = append(failedSeqs, seq)
-				continue
-			}
-			seqMsgs = append(seqMsgs, msg)
-
+		if err := c.rcClient.RawSet(ctx, cachekey.GetMsgCacheKey(conversationID, msg.Msg.Seq), string(data), msgCacheTimeout); err != nil {
+			return err
 		}
-		return nil
-	})
-	if err != nil {
-		return nil, nil, err
 	}
-	return seqMsgs, failedSeqs, nil
+	return nil
 }