feat: add scripts (#525)

This commit is contained in:
Xinwei Xiong
2023-07-13 11:37:23 +08:00
committed by GitHub
parent eb7953cacb
commit 7bf8a898e2
56 changed files with 2049 additions and 442 deletions
+345 -52
View File
@@ -18,10 +18,11 @@ import (
"context"
"errors"
"go.mongodb.org/mongo-driver/mongo"
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 (
@@ -40,16 +41,36 @@ type CommonMsgDatabase interface {
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)
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)
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)
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列表
UserMsgsDestruct(cte context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error)
UserMsgsDestruct(
cte context.Context,
userID string,
conversationID string,
destructTime int64,
lastMsgDestructTime time.Time,
) (seqs []int64, err error)
// 用户根据seq删除消息
DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error
@@ -75,7 +96,10 @@ type CommonMsgDatabase interface {
UserSetHasReadSeqs(ctx context.Context, userID string, hasReadSeqs map[string]int64) error
GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error)
GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache 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)
@@ -85,8 +109,23 @@ type CommonMsgDatabase interface {
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
RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error)
RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error)
RangeUserSendCount(
ctx context.Context,
start time.Time,
end time.Time,
group bool,
ase bool,
pageNumber int32,
showNumber int32,
) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error)
RangeGroupSendCount(
ctx context.Context,
start time.Time,
end time.Time,
ase bool,
pageNumber int32,
showNumber int32,
) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error)
}
func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase {
@@ -121,16 +160,32 @@ func (db *commonMsgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *sd
return err
}
func (db *commonMsgDatabase) MsgToModifyMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData) error {
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})
_, _, 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})
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
@@ -138,15 +193,30 @@ func (db *commonMsgDatabase) MsgToPushMQ(ctx context.Context, key, conversationI
return partition, offset, nil
}
func (db *commonMsgDatabase) MsgToMongoMQ(ctx context.Context, key, conversationID string, messages []*sdkws.MsgData, lastSeq int64) error {
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})
_, _, 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 {
func (db *commonMsgDatabase) BatchInsertBlock(
ctx context.Context,
conversationID string,
fields []any,
key int8,
firstSeq int64,
) error {
if len(fields) == 0 {
return nil
}
@@ -247,7 +317,12 @@ func (db *commonMsgDatabase) BatchInsertBlock(ctx context.Context, conversationI
return nil
}
func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*sdkws.MsgData, currentMaxSeq int64) error {
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")
}
@@ -293,11 +368,21 @@ func (db *commonMsgDatabase) BatchInsertChat2DB(ctx context.Context, conversatio
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 {
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 {
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 {
@@ -320,7 +405,11 @@ func (db *commonMsgDatabase) DelUserDeleteMsgsList(ctx context.Context, conversa
db.cache.DelUserDeleteMsgsList(ctx, conversationID, seqs)
}
func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
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)
@@ -367,7 +456,11 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
return lastMaxSeq, isNew, utils.Wrap(err, "")
}
func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
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)
@@ -381,7 +474,11 @@ func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversat
return totalMsgs, nil
}
func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID string, seqs []int64) (totalMsgs []*unRelationTb.MsgInfoModel, err error) {
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 {
@@ -391,8 +488,25 @@ func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID
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)
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)
@@ -409,7 +523,12 @@ func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID strin
return seqMsgs, nil
}
func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num, userMaxSeq int64) (int64, int64, []*sdkws.MsgData, error) {
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
@@ -429,7 +548,18 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
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)
log.ZDebug(
ctx,
"GetMsgBySeqsRange",
"userMinSeq",
userMinSeq,
"conMinSeq",
minSeq,
"conMaxSeq",
maxSeq,
"userMaxSeq",
userMaxSeq,
)
if userMaxSeq != 0 {
if userMaxSeq < maxSeq {
maxSeq = userMaxSeq
@@ -479,7 +609,18 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
cacheDelNum += 1
}
}
log.ZDebug(ctx, "get delSeqs from redis", "delSeqs", delSeqs, "userID", userID, "conversationID", conversationID, "cacheDelNum", cacheDelNum)
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)
@@ -499,7 +640,15 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
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)
log.ZError(
ctx,
"get message from redis exception",
err,
"conversationID",
conversationID,
"seqs",
reGetSeqsCache,
)
}
}
failedSeqs = append(failedSeqs, failedSeqs2...)
@@ -525,7 +674,12 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
return minSeq, maxSeq, successMsgs, nil
}
func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
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
@@ -551,10 +705,33 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
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.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)
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)
@@ -568,7 +745,11 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
return minSeq, maxSeq, successMsgs, nil
}
func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {
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)
@@ -588,7 +769,13 @@ func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Cont
return db.cache.SetMinSeq(ctx, conversationID, minSeq)
}
func (db *commonMsgDatabase) UserMsgsDestruct(ctx context.Context, userID string, conversationID string, destructTime int64, lastMsgDestructTime time.Time) (seqs []int64, err error) {
func (db *commonMsgDatabase) UserMsgsDestruct(
ctx context.Context,
userID string,
conversationID string,
destructTime int64,
lastMsgDestructTime time.Time,
) (seqs []int64, err error) {
var index int64
for {
// from oldest 2 newest
@@ -596,7 +783,16 @@ func (db *commonMsgDatabase) UserMsgsDestruct(ctx context.Context, userID string
if err != nil || msgDocModel.DocID == "" {
if err != nil {
if err == unrelation.ErrMsgListNotExist {
log.ZDebug(ctx, "deleteMsgRecursion finished", "conversationID", conversationID, "userID", userID, "index", index)
log.ZDebug(
ctx,
"deleteMsgRecursion finished",
"conversationID",
conversationID,
"userID",
userID,
"index",
index,
)
} else {
log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
}
@@ -645,13 +841,26 @@ func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
// 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) {
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)
log.ZDebug(
ctx,
"deleteMsgRecursion ErrMsgListNotExist",
"conversationID",
conversationID,
"index:",
index,
)
} else {
log.ZError(ctx, "deleteMsgRecursion GetUserMsgListByIndex failed", err, "conversationID", conversationID, "index", index)
}
@@ -663,11 +872,23 @@ func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversatio
}
return delStruct.getSetMinSeq() + 1, nil
}
log.ZDebug(ctx, "doc info", "conversationID", conversationID, "index", index, "docID", msgDocModel.DocID, "len", len(msgDocModel.Msg))
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() {
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
@@ -704,7 +925,11 @@ func (db *commonMsgDatabase) deleteMsgRecursion(ctx context.Context, conversatio
return seq, err
}
func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conversationID string, allSeqs []int64) error {
func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(
ctx context.Context,
conversationID string,
allSeqs []int64,
) error {
if err := db.cache.DeleteMessages(ctx, conversationID, allSeqs); err != nil {
return err
}
@@ -720,7 +945,12 @@ func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conve
return nil
}
func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error {
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)
@@ -789,31 +1019,70 @@ func (db *commonMsgDatabase) GetMinSeqs(ctx context.Context, conversationIDs []s
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) {
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) {
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 {
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) {
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 {
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 {
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 {
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) {
func (db *commonMsgDatabase) GetHasReadSeqs(
ctx context.Context,
userID string,
conversationIDs []string,
) (map[string]int64, error) {
return db.cache.GetHasReadSeqs(ctx, userID, conversationIDs)
}
@@ -829,7 +1098,10 @@ func (db *commonMsgDatabase) GetSendMsgStatus(ctx context.Context, id string) (i
return db.cache.GetSendMsgStatus(ctx, id)
}
func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo, minSeqCache, maxSeqCache int64, err error) {
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
@@ -845,11 +1117,17 @@ func (db *commonMsgDatabase) GetConversationMinMaxSeqInMongoAndCache(ctx context
return
}
func (db *commonMsgDatabase) GetMongoMaxAndMinSeq(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
func (db *commonMsgDatabase) GetMongoMaxAndMinSeq(
ctx context.Context,
conversationID string,
) (minSeqMongo, maxSeqMongo int64, err error) {
return db.GetMinMaxSeqMongo(ctx, conversationID)
}
func (db *commonMsgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
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
@@ -863,10 +1141,25 @@ func (db *commonMsgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversation
return
}
func (db *commonMsgDatabase) RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error) {
func (db *commonMsgDatabase) RangeUserSendCount(
ctx context.Context,
start time.Time,
end time.Time,
group bool,
ase bool,
pageNumber int32,
showNumber int32,
) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, dateCount map[string]int64, err error) {
return db.msgDocDatabase.RangeUserSendCount(ctx, start, end, group, ase, pageNumber, showNumber)
}
func (db *commonMsgDatabase) RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error) {
func (db *commonMsgDatabase) RangeGroupSendCount(
ctx context.Context,
start time.Time,
end time.Time,
ase bool,
pageNumber int32,
showNumber int32,
) (msgCount int64, userCount int64, groups []*unRelationTb.GroupCount, dateCount map[string]int64, err error) {
return db.msgDocDatabase.RangeGroupSendCount(ctx, start, end, ase, pageNumber, showNumber)
}