IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-14 05:56:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

msg database

Browse Source
This commit is contained in:
withchao
2023-02-23 18:48:14 +08:00
parent 2f1c064413
commit c8004279a4
5 changed files with 159 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/db/cache/redis.go
Vendored
+49 -70
View File
@@ -18,27 +18,6 @@ import (
"github.com/golang/protobuf/proto"
)
const (
userIncrSeq = "REDIS_USER_INCR_SEQ:" // user incr seq
appleDeviceToken = "DEVICE_TOKEN"
userMinSeq = "REDIS_USER_MIN_SEQ:"
getuiToken = "GETUI_TOKEN"
getuiTaskID = "GETUI_TASK_ID"
messageCache = "MESSAGE_CACHE:"
signalCache = "SIGNAL_CACHE:"
signalListCache = "SIGNAL_LIST_CACHE:"
FcmToken = "FCM_TOKEN:"
groupUserMinSeq = "GROUP_USER_MIN_SEQ:"
groupMaxSeq = "GROUP_MAX_SEQ:"
groupMinSeq = "GROUP_MIN_SEQ:"
sendMsgFailedFlag = "SEND_MSG_FAILED_FLAG:"
userBadgeUnreadCountSum = "USER_BADGE_UNREAD_COUNT_SUM:"
exTypeKeyLocker = "EX_LOCK:"
uidPidToken = "UID_PID_TOKEN_STATUS:"
)
type Cache interface {
IncrUserSeq(ctx context.Context, userID string) (int64, error)
GetUserMaxSeq(ctx context.Context, userID string) (int64, error)
@@ -150,78 +129,78 @@ func (r *RedisClient) GetClient() redis.UniversalClient {
// Perform seq auto-increment operation of user messages
func (r *RedisClient) IncrUserSeq(ctx context.Context, uid string) (int64, error) {
key := userIncrSeq + uid
seq, err := r.rdb.Incr(context.Background(), key).Result()
seq, err := r.rdb.Incr(ctx, key).Result()
return seq, err
}
// Get the largest Seq
func (r *RedisClient) GetUserMaxSeq(ctx context.Context, uid string) (int64, error) {
key := userIncrSeq + uid
seq, err := r.rdb.Get(context.Background(), key).Result()
seq, err := r.rdb.Get(ctx, key).Result()
return int64(utils.StringToInt(seq)), err
}
// set the largest Seq
func (r *RedisClient) SetUserMaxSeq(ctx context.Context, uid string, maxSeq int64) error {
key := userIncrSeq + uid
return r.rdb.Set(context.Background(), key, maxSeq, 0).Err()
return r.rdb.Set(ctx, key, maxSeq, 0).Err()
}
// Set the user's minimum seq
func (r *RedisClient) SetUserMinSeq(ctx context.Context, uid string, minSeq int64) (err error) {
key := userMinSeq + uid
return r.rdb.Set(context.Background(), key, minSeq, 0).Err()
return r.rdb.Set(ctx, key, minSeq, 0).Err()
}
// Get the smallest Seq
func (r *RedisClient) GetUserMinSeq(ctx context.Context, uid string) (int64, error) {
key := userMinSeq + uid
seq, err := r.rdb.Get(context.Background(), key).Result()
seq, err := r.rdb.Get(ctx, key).Result()
return int64(utils.StringToInt(seq)), err
}
func (r *RedisClient) SetGroupUserMinSeq(ctx context.Context, groupID, userID string, minSeq int64) (err error) {
key := groupUserMinSeq + "g:" + groupID + "u:" + userID
return r.rdb.Set(context.Background(), key, minSeq, 0).Err()
return r.rdb.Set(ctx, key, minSeq, 0).Err()
}
func (r *RedisClient) GetGroupUserMinSeq(ctx context.Context, groupID, userID string) (int64, error) {
key := groupUserMinSeq + "g:" + groupID + "u:" + userID
seq, err := r.rdb.Get(context.Background(), key).Result()
seq, err := r.rdb.Get(ctx, key).Result()
return int64(utils.StringToInt(seq)), err
}
func (r *RedisClient) GetGroupMaxSeq(ctx context.Context, groupID string) (int64, error) {
key := groupMaxSeq + groupID
seq, err := r.rdb.Get(context.Background(), key).Result()
seq, err := r.rdb.Get(ctx, key).Result()
return int64(utils.StringToInt(seq)), err
}
func (r *RedisClient) IncrGroupMaxSeq(ctx context.Context, groupID string) (int64, error) {
key := groupMaxSeq + groupID
seq, err := r.rdb.Incr(context.Background(), key).Result()
seq, err := r.rdb.Incr(ctx, key).Result()
return seq, err
}
func (r *RedisClient) SetGroupMaxSeq(ctx context.Context, groupID string, maxSeq int64) error {
key := groupMaxSeq + groupID
return r.rdb.Set(context.Background(), key, maxSeq, 0).Err()
return r.rdb.Set(ctx, key, maxSeq, 0).Err()
}
func (r *RedisClient) SetGroupMinSeq(ctx context.Context, groupID string, minSeq int64) error {
key := groupMinSeq + groupID
return r.rdb.Set(context.Background(), key, minSeq, 0).Err()
return r.rdb.Set(ctx, key, minSeq, 0).Err()
}
// Store userid and platform class to redis
func (r *RedisClient) AddTokenFlag(ctx context.Context, userID string, platformID int, token string, flag int) error {
key := uidPidToken + userID + ":" + constant.PlatformIDToName(platformID)
return r.rdb.HSet(context.Background(), key, token, flag).Err()
return r.rdb.HSet(ctx, key, token, flag).Err()
}
//key:userID+platform-> <token, flag>
func (r *RedisClient) GetTokenMapByUidPid(ctx context.Context, userID, platformID int) (map[string]int, error) {
key := uidPidToken + userID + ":" + platformID
m, err := r.rdb.HGetAll(context.Background(), key).Result()
func (r *RedisClient) GetTokenMapByUidPid(ctx context.Context, userID string, platformID int) (map[string]int, error) {
key := uidPidToken + userID + ":" + strconv.Itoa(platformID)
m, err := r.rdb.HGetAll(ctx, key).Result()
mm := make(map[string]int)
for k, v := range m {
mm[k] = utils.StringToInt(v)
@@ -231,7 +210,7 @@ func (r *RedisClient) GetTokenMapByUidPid(ctx context.Context, userID, platformI
func (r *RedisClient) GetTokensWithoutError(ctx context.Context, userID, platform string) (map[string]int, error) {
key := uidPidToken + userID + ":" + platform
m, err := r.rdb.HGetAll(context.Background(), key).Result()
m, err := r.rdb.HGetAll(ctx, key).Result()
if err != nil && err == redis.Nil {
return nil, nil
}
@@ -248,19 +227,19 @@ func (r *RedisClient) SetTokenMapByUidPid(ctx context.Context, userID string, pl
for k, v := range m {
mm[k] = v
}
return r.rdb.HSet(context.Background(), key, mm).Err()
return r.rdb.HSet(ctx, key, mm).Err()
}
func (r *RedisClient) DeleteTokenByUidPid(ctx context.Context, userID string, platform string, fields []string) error {
key := uidPidToken + userID + ":" + platform
return r.rdb.HDel(context.Background(), key, fields...).Err()
return r.rdb.HDel(ctx, key, fields...).Err()
}
func (r *RedisClient) GetMessagesBySeq(ctx context.Context, userID string, seqList []int64, operationID string) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err2 error) {
for _, v := range seqList {
//MESSAGE_CACHE:169.254.225.224_reliability1653387820_0_1
key := messageCache + userID + "_" + strconv.Itoa(int(v))
result, err := r.rdb.Get(context.Background(), key).Result()
result, err := r.rdb.Get(ctx, key).Result()
if err != nil {
if err != redis.Nil {
err2 = err
@@ -290,13 +269,13 @@ func (r *RedisClient) SetMessageToCache(ctx context.Context, userID string, msgs
continue
}
err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err()
//err = r.rdb.HMSet(context.Background(), "12", map[string]interface{}{"1": 2, "343": false}).Err()
//err = r.rdb.HMSet(ctx, "12", map[string]interface{}{"1": 2, "343": false}).Err()
if err != nil {
failedMsgs = append(failedMsgs, *msg)
}
}
if len(failedMsgs) != 0 {
return len(failedMsgs), errors.New(fmt.Sprintf("set msg to cache failed, failed lists: %q,%s", failedList))
return len(failedMsgs), errors.New(fmt.Sprintf("set msg to cache failed, failed lists: %v", failedMsgs))
}
_, err := pipe.Exec(ctx)
return 0, err
@@ -355,16 +334,16 @@ func (r *RedisClient) HandleSignalInfo(ctx context.Context, operationID string,
return false, err
}
keyList := signalListCache + userID
err = r.rdb.LPush(context.Background(), keyList, msg.ClientMsgID).Err()
err = r.rdb.LPush(ctx, keyList, msg.ClientMsgID).Err()
if err != nil {
return false, err
}
err = r.rdb.Expire(context.Background(), keyList, time.Duration(timeout)*time.Second).Err()
err = r.rdb.Expire(ctx, keyList, time.Duration(timeout)*time.Second).Err()
if err != nil {
return false, err
}
key := signalCache + msg.ClientMsgID
err = r.rdb.Set(context.Background(), key, msg.Content, time.Duration(timeout)*time.Second).Err()
err = r.rdb.Set(ctx, key, msg.Content, time.Duration(timeout)*time.Second).Err()
if err != nil {
return false, err
}
@@ -376,7 +355,7 @@ func (r *RedisClient) HandleSignalInfo(ctx context.Context, operationID string,
func (r *RedisClient) GetSignalInfoFromCacheByClientMsgID(ctx context.Context, clientMsgID string) (invitationInfo *pbRtc.SignalInviteReq, err error) {
key := signalCache + clientMsgID
invitationInfo = &pbRtc.SignalInviteReq{}
bytes, err := r.rdb.Get(context.Background(), key).Bytes()
bytes, err := r.rdb.Get(ctx, key).Bytes()
if err != nil {
return nil, err
}
@@ -397,7 +376,7 @@ func (r *RedisClient) GetSignalInfoFromCacheByClientMsgID(ctx context.Context, c
func (r *RedisClient) GetAvailableSignalInvitationInfo(ctx context.Context, userID string) (invitationInfo *pbRtc.SignalInviteReq, err error) {
keyList := signalListCache + userID
result := r.rdb.LPop(context.Background(), keyList)
result := r.rdb.LPop(ctx, keyList)
if err = result.Err(); err != nil {
return nil, utils.Wrap(err, "GetAvailableSignalInvitationInfo failed")
}
@@ -418,14 +397,14 @@ func (r *RedisClient) GetAvailableSignalInvitationInfo(ctx context.Context, user
func (r *RedisClient) DelUserSignalList(ctx context.Context, userID string) error {
keyList := signalListCache + userID
err := r.rdb.Del(context.Background(), keyList).Err()
err := r.rdb.Del(ctx, keyList).Err()
return err
}
func (r *RedisClient) DelMsgFromCache(ctx context.Context, uid string, seqList []int64, operationID string) {
for _, seq := range seqList {
key := messageCache + uid + "_" + strconv.Itoa(int(seq))
result, err := r.rdb.Get(context.Background(), key).Result()
result, err := r.rdb.Get(ctx, key).Result()
if err != nil {
if err == redis.Nil {
} else {
@@ -441,35 +420,35 @@ func (r *RedisClient) DelMsgFromCache(ctx context.Context, uid string, seqList [
if err != nil {
continue
}
if err := r.rdb.Set(context.Background(), key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil {
if err := r.rdb.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil {
}
}
}
func (r *RedisClient) SetGetuiToken(ctx context.Context, token string, expireTime int64) error {
return r.rdb.Set(context.Background(), getuiToken, token, time.Duration(expireTime)*time.Second).Err()
return r.rdb.Set(ctx, getuiToken, token, time.Duration(expireTime)*time.Second).Err()
}
func (r *RedisClient) GetGetuiToken(ctx context.Context) (string, error) {
result, err := r.rdb.Get(context.Background(), getuiToken).Result()
result, err := r.rdb.Get(ctx, getuiToken).Result()
return result, err
}
func (r *RedisClient) SetGetuiTaskID(ctx context.Context, taskID string, expireTime int64) error {
return r.rdb.Set(context.Background(), getuiTaskID, taskID, time.Duration(expireTime)*time.Second).Err()
return r.rdb.Set(ctx, getuiTaskID, taskID, time.Duration(expireTime)*time.Second).Err()
}
func (r *RedisClient) GetGetuiTaskID(ctx context.Context) (string, error) {
result, err := r.rdb.Get(context.Background(), getuiTaskID).Result()
result, err := r.rdb.Get(ctx, getuiTaskID).Result()
return result, err
}
func (r *RedisClient) SetSendMsgStatus(ctx context.Context, status int32, operationID string) error {
return r.rdb.Set(context.Background(), sendMsgFailedFlag+operationID, status, time.Hour*24).Err()
return r.rdb.Set(ctx, sendMsgFailedFlag+operationID, status, time.Hour*24).Err()
}
func (r *RedisClient) GetSendMsgStatus(ctx context.Context, operationID string) (int, error) {
result, err := r.rdb.Get(context.Background(), sendMsgFailedFlag+operationID).Result()
result, err := r.rdb.Get(ctx, sendMsgFailedFlag+operationID).Result()
if err != nil {
return 0, err
}
@@ -479,71 +458,71 @@ func (r *RedisClient) GetSendMsgStatus(ctx context.Context, operationID string)
func (r *RedisClient) SetFcmToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) (err error) {
key := FcmToken + account + ":" + strconv.Itoa(platformID)
return r.rdb.Set(context.Background(), key, fcmToken, time.Duration(expireTime)*time.Second).Err()
return r.rdb.Set(ctx, key, fcmToken, time.Duration(expireTime)*time.Second).Err()
}
func (r *RedisClient) GetFcmToken(ctx context.Context, account string, platformID int) (string, error) {
key := FcmToken + account + ":" + strconv.Itoa(platformID)
return r.rdb.Get(context.Background(), key).Result()
return r.rdb.Get(ctx, key).Result()
}
func (r *RedisClient) DelFcmToken(ctx context.Context, account string, platformID int) error {
key := FcmToken + account + ":" + strconv.Itoa(platformID)
return r.rdb.Del(context.Background(), key).Err()
return r.rdb.Del(ctx, key).Err()
}
func (r *RedisClient) IncrUserBadgeUnreadCountSum(ctx context.Context, uid string) (int, error) {
key := userBadgeUnreadCountSum + uid
seq, err := r.rdb.Incr(context.Background(), key).Result()
seq, err := r.rdb.Incr(ctx, key).Result()
return int(seq), err
}
func (r *RedisClient) SetUserBadgeUnreadCountSum(ctx context.Context, uid string, value int) error {
key := userBadgeUnreadCountSum + uid
return r.rdb.Set(context.Background(), key, value, 0).Err()
return r.rdb.Set(ctx, key, value, 0).Err()
}
func (r *RedisClient) GetUserBadgeUnreadCountSum(ctx context.Context, uid string) (int, error) {
key := userBadgeUnreadCountSum + uid
seq, err := r.rdb.Get(context.Background(), key).Result()
seq, err := r.rdb.Get(ctx, key).Result()
return utils.StringToInt(seq), err
}
func (r *RedisClient) JudgeMessageReactionEXISTS(ctx context.Context, clientMsgID string, sessionType int32) (bool, error) {
key := r.getMessageReactionExPrefix(clientMsgID, sessionType)
n, err := r.rdb.Exists(context.Background(), key).Result()
n, err := r.rdb.Exists(ctx, key).Result()
return n > 0, err
}
func (r *RedisClient) GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error) {
key := r.getMessageReactionExPrefix(clientMsgID, sessionType)
return r.rdb.HGetAll(context.Background(), key).Result()
return r.rdb.HGetAll(ctx, key).Result()
}
func (r *RedisClient) DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error {
key := r.getMessageReactionExPrefix(clientMsgID, sessionType)
return r.rdb.HDel(context.Background(), key, subKey).Err()
return r.rdb.HDel(ctx, key, subKey).Err()
}
func (r *RedisClient) SetMessageReactionExpire(ctx context.Context, clientMsgID string, sessionType int32, expiration time.Duration) (bool, error) {
key := r.getMessageReactionExPrefix(clientMsgID, sessionType)
return r.rdb.Expire(context.Background(), key, expiration).Result()
return r.rdb.Expire(ctx, key, expiration).Result()
}
func (r *RedisClient) GetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey string) (string, error) {
key := r.getMessageReactionExPrefix(clientMsgID, sessionType)
result, err := r.rdb.HGet(context.Background(), key, typeKey).Result()
result, err := r.rdb.HGet(ctx, key, typeKey).Result()
return result, err
}
func (r *RedisClient) SetMessageTypeKeyValue(ctx context.Context, clientMsgID string, sessionType int32, typeKey, value string) error {
key := r.getMessageReactionExPrefix(clientMsgID, sessionType)
return r.rdb.HSet(context.Background(), key, typeKey, value).Err()
return r.rdb.HSet(ctx, key, typeKey, value).Err()
}
func (r *RedisClient) LockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error {
key := exTypeKeyLocker + clientMsgID + "_" + TypeKey
return r.rdb.SetNX(context.Background(), key, 1, time.Minute).Err()
return r.rdb.SetNX(ctx, key, 1, time.Minute).Err()
}
func (r *RedisClient) UnLockMessageTypeKey(ctx context.Context, clientMsgID string, TypeKey string) error {
key := exTypeKeyLocker + clientMsgID + "_" + TypeKey
return r.rdb.Del(context.Background(), key).Err()
return r.rdb.Del(ctx, key).Err()
}