blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/errcode' into errcode

Browse Source
This commit is contained in:
withchao 2023-06-09 14:21:10 +08:00
commit 8d6bd95c62
3 changed files with 45 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
pkg/common/db/cache/msg.go vendored
View File

@ -383,13 +383,23 @@ func (c *msgCache) DelUserDeleteMsgsList(ctx context.Context, conversationID str
log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq)
continue
}
for _, userID := range delUsers {
if err := c.rdb.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err(); err != nil {
log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq, "userID", userID)
if len(delUsers) > 0 {
pipe := c.rdb.Pipeline()
var failedFlag bool
for _, userID := range delUsers {
err = pipe.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err()
if err != nil {
failedFlag = true
log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq, "userID", userID)
}
}
if err := c.rdb.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil {
log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq)
if !failedFlag {
if err := pipe.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil {
log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq)
}
}
if _, err := pipe.Exec(ctx); err != nil {
log.ZError(ctx, "pipe exec failed", err, "conversationID", conversationID, "seq", seq)
}
}
}

38
pkg/common/db/controller/msg.go
View File

@ -465,12 +465,16 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
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 {
@ -479,11 +483,14 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
break
}
}
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, seqs)
log.ZError(ctx, "get message from redis exception", err, "conversationID", conversationID, "seqs", seqs)
}
}
var successMsgs []*sdkws.MsgData
@ -500,10 +507,19 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
successMsgs = append(successMsgs, msg)
}
}
for i := 1; i <= len(delSeqs); i++ {
newSeq := newBegin - int64(i)
if newSeq >= begin {
log.ZDebug(ctx, "seq del in cache, a new seq in range append", "new seq", newSeq)
failedSeqs = append(failedSeqs, newSeq)
} else {
break
}
}
}
log.ZDebug(ctx, "get msgs from cache", "successMsgs", successMsgs)
if len(failedSeqs) != 0 {
log.ZDebug(ctx, "msgs not exist in redis", err, "seqs", seqs)
log.ZDebug(ctx, "msgs not exist in redis", err, "seqs", failedSeqs)
}
// get from cache or db
prome.Add(prome.MsgPullFromRedisSuccessCounter, len(successMsgs))
@ -672,17 +688,19 @@ func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conve
}
func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error {
msgs, _, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs)
if err != nil {
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
}
var cacheSeqs []int64
for _, msg := range msgs {
cacheSeqs = append(cacheSeqs, msg.Seq)
}
if err := db.cache.UserDeleteMsgs(ctx, conversationID, cacheSeqs, userID); err != nil {
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) {

2
pkg/common/log/zap.go
View File

@ -119,7 +119,7 @@ func (l *ZapLogger) cores(isStdout bool, isJson bool, logLocation string, rotate
return nil, err
}
var cores []zapcore.Core
if logLocation != "" && !isStdout {
if logLocation != "" {
cores = []zapcore.Core{
zapcore.NewCore(fileEncoder, writer, zap.NewAtomicLevelAt(l.level)),
}