IM/open-im-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

sourceID2conversationID

Browse Source
This commit is contained in:
wangchuxiao
2023-05-04 15:06:23 +08:00
parent d855beb354
commit 5c8208e0f8
34 changed files with 1627 additions and 1550 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/db/controller/msg.go
+84 -51
View File
@@ -30,18 +30,19 @@ import (
type MsgDatabase interface {
// 批量插入消息
BatchInsertChat2DB(ctx context.Context, sourceID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) error
BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) error
// 刪除redis中消息缓存
DeleteMessageFromCache(ctx context.Context, sourceID string, msgList []*pbMsg.MsgDataToMQ) error
DeleteMessageFromCache(ctx context.Context, conversationID string, msgList []*pbMsg.MsgDataToMQ) error
// incrSeq然后批量插入缓存
BatchInsertChat2Cache(ctx context.Context, sourceID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) (int64, error)
BatchInsertChat2Cache(ctx context.Context, conversationID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) (int64, error)
// incrSeq通知seq然后批量插入缓存
NotificationBatchInsertChat2Cache(ctx context.Context, sourceID string, msgs []*pbMsg.MsgDataToMQ) (int64, error)
NotificationBatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*pbMsg.MsgDataToMQ) (int64, error)
// 删除消息 返回不存在的seqList
DelMsgBySeqs(ctx context.Context, userID string, seqs []int64) (totalUnExistSeqs []int64, err error)
// 获取群ID或者UserID最新一条在mongo里面的消息
// 通过seqList获取mongo中写扩散消息
GetMsgBySeqs(ctx context.Context, userID string, seqs []int64) (seqMsg []*sdkws.MsgData, err error)
GetMsgBySeqsRange(ctx context.Context, conversationID string, begin, end, num int64) (seqMsg []*sdkws.MsgData, err error)
// 通过seqList获取大群在 mongo里面的消息
GetSuperGroupMsgBySeqs(ctx context.Context, groupID string, seqs []int64) (seqMsg []*sdkws.MsgData, err error)
// 删除用户所有消息/redis/mongo然后重置seq
@@ -66,12 +67,12 @@ type MsgDatabase interface {
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, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (*pbMsg.ExtendMsg, error)
InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) 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, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
SetSendMsgStatus(ctx context.Context, id string, status int32) error
GetSendMsgStatus(ctx context.Context, id string) (int32, error)
GetUserMaxSeq(ctx context.Context, userID string) (int64, error)
@@ -80,9 +81,9 @@ type MsgDatabase interface {
GetGroupMinSeq(ctx context.Context, groupID string) (int64, error)
MsgToMQ(ctx context.Context, key string, msg2mq *pbMsg.MsgDataToMQ) error
MsgToModifyMQ(ctx context.Context, sourceID string, messages []*pbMsg.MsgDataToMQ) error
MsgToPushMQ(ctx context.Context, sourceID string, msg2mq *pbMsg.MsgDataToMQ) (int32, int64, error)
MsgToMongoMQ(ctx context.Context, sourceID string, messages []*pbMsg.MsgDataToMQ, lastSeq int64) error
MsgToModifyMQ(ctx context.Context, conversationID string, messages []*pbMsg.MsgDataToMQ) error
MsgToPushMQ(ctx context.Context, conversationID string, msg2mq *pbMsg.MsgDataToMQ) (int32, int64, error)
MsgToMongoMQ(ctx context.Context, conversationID string, messages []*pbMsg.MsgDataToMQ, lastSeq int64) error
}
func NewMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.Model) MsgDatabase {
@@ -140,12 +141,12 @@ func (db *msgDatabase) DeleteOneMessageKey(ctx context.Context, clientMsgID stri
return db.cache.DeleteOneMessageKey(ctx, clientMsgID, sessionType, subKey)
}
func (db *msgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
return db.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, db.extendMsgSetModel.Pb2Model(reactionExtensions))
func (db *msgDatabase) 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 *msgDatabase) GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (*pbMsg.ExtendMsg, error) {
extendMsgSet, err := db.extendMsgDatabase.GetExtendMsgSet(ctx, sourceID, sessionType, maxMsgUpdateTime)
func (db *msgDatabase) 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
}
@@ -172,8 +173,8 @@ func (db *msgDatabase) GetExtendMsg(ctx context.Context, sourceID string, sessio
}, nil
}
func (db *msgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
return db.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, db.extendMsgSetModel.Pb2Model(reactionExtensions))
func (db *msgDatabase) 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))
}
func (db *msgDatabase) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
@@ -189,26 +190,26 @@ func (db *msgDatabase) MsgToMQ(ctx context.Context, key string, msg2mq *pbMsg.Ms
return err
}
func (db *msgDatabase) MsgToModifyMQ(ctx context.Context, sourceID string, messages []*pbMsg.MsgDataToMQ) error {
func (db *msgDatabase) MsgToModifyMQ(ctx context.Context, conversationID string, messages []*pbMsg.MsgDataToMQ) error {
if len(messages) > 0 {
_, _, err := db.producerToModify.SendMessage(ctx, sourceID, &pbMsg.MsgDataToModifyByMQ{SourceID: sourceID, Messages: messages})
_, _, err := db.producerToModify.SendMessage(ctx, conversationID, &pbMsg.MsgDataToModifyByMQ{ConversationID: conversationID, Messages: messages})
return err
}
return nil
}
func (db *msgDatabase) MsgToPushMQ(ctx context.Context, key string, msg2mq *pbMsg.MsgDataToMQ) (int32, int64, error) {
mqPushMsg := pbMsg.PushMsgDataToMQ{MsgData: msg2mq.MsgData, SourceID: key}
partition, offset, err := db.producerToPush.SendMessage(ctx, key, &mqPushMsg)
func (db *msgDatabase) MsgToPushMQ(ctx context.Context, conversationID string, msg2mq *pbMsg.MsgDataToMQ) (int32, int64, error) {
mqPushMsg := pbMsg.PushMsgDataToMQ{MsgData: msg2mq.MsgData, ConversationID: conversationID}
partition, offset, err := db.producerToPush.SendMessage(ctx, conversationID, &mqPushMsg)
if err != nil {
log.ZError(ctx, "MsgToPushMQ", err, "key", key, "msg2mq", msg2mq)
log.ZError(ctx, "MsgToPushMQ", err, "key", conversationID, "msg2mq", msg2mq)
}
return partition, offset, err
}
func (db *msgDatabase) MsgToMongoMQ(ctx context.Context, sourceID string, messages []*pbMsg.MsgDataToMQ, lastSeq int64) error {
func (db *msgDatabase) MsgToMongoMQ(ctx context.Context, conversationID string, messages []*pbMsg.MsgDataToMQ, lastSeq int64) error {
if len(messages) > 0 {
_, _, err := db.producerToModify.SendMessage(ctx, sourceID, &pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, SourceID: sourceID, Messages: messages})
_, _, err := db.producerToModify.SendMessage(ctx, conversationID, &pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, ConversationID: conversationID, Messages: messages})
return err
}
return nil
@@ -229,7 +230,7 @@ func (db *msgDatabase) GetGroupMinSeq(ctx context.Context, groupID string) (int6
return db.cache.GetGroupMinSeq(ctx, groupID)
}
func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, sourceID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) error {
func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, conversationID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) error {
//newTime := utils.GetCurrentTimestampByMill()
if int64(len(msgList)) > db.msg.GetSingleGocMsgNum() {
return errors.New("too large")
@@ -263,11 +264,11 @@ func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, sourceID string,
if insertCounter < remain {
msgsToMongo = append(msgsToMongo, sMsg)
insertCounter++
docID = db.msg.GetDocID(sourceID, currentMaxSeq)
docID = db.msg.GetDocID(conversationID, currentMaxSeq)
//log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID)
} else {
msgsToMongoNext = append(msgsToMongoNext, sMsg)
docIDNext = db.msg.GetDocID(sourceID, currentMaxSeq)
docIDNext = db.msg.GetDocID(conversationID, currentMaxSeq)
//log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain, "userID: ", userID)
}
}
@@ -317,11 +318,18 @@ func (db *msgDatabase) DeleteMessageFromCache(ctx context.Context, userID string
return db.cache.DeleteMessageFromCache(ctx, userID, msgs)
}
func (db *msgDatabase) NotificationBatchInsertChat2Cache(ctx context.Context, sourceID string, msgs []*pbMsg.MsgDataToMQ) (int64, error) {
func (db *msgDatabase) NotificationBatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*pbMsg.MsgDataToMQ) (int64, error) {
lenList := len(msgs)
if int64(lenList) > db.msg.GetSingleGocMsgNum() {
return 0, errors.New("too large")
}
if lenList < 1 {
return 0, errors.New("too short as 0")
}
return 0, nil
}
func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, sourceID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) (int64, error) {
func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgList []*pbMsg.MsgDataToMQ, currentMaxSeq int64) (int64, error) {
//newTime := utils.GetCurrentTimestampByMill()
lenList := len(msgList)
if int64(lenList) > db.msg.GetSingleGocMsgNum() {
@@ -335,21 +343,21 @@ func (db *msgDatabase) BatchInsertChat2Cache(ctx context.Context, sourceID strin
for _, m := range msgList {
currentMaxSeq++
m.MsgData.Seq = currentMaxSeq
//log.Debug(operationID, "cache msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", sourceID, "seq: ", currentMaxSeq)
//log.Debug(operationID, "cache msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", conversationID, "seq: ", currentMaxSeq)
}
//log.Debug(operationID, "SetMessageToCache ", sourceID, len(msgList))
failedNum, err := db.cache.SetMessageToCache(ctx, sourceID, msgList)
//log.Debug(operationID, "SetMessageToCache ", conversationID, len(msgList))
failedNum, err := db.cache.SetMessageToCache(ctx, conversationID, msgList)
if err != nil {
prome.Add(prome.MsgInsertRedisFailedCounter, failedNum)
//log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), sourceID)
//log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), conversationID)
} else {
prome.Inc(prome.MsgInsertRedisSuccessCounter)
}
//log.Debug(operationID, "batch to redis cost time ", mongo2.getCurrentTimestampByMill()-newTime, sourceID, len(msgList))
//log.Debug(operationID, "batch to redis cost time ", mongo2.getCurrentTimestampByMill()-newTime, conversationID, len(msgList))
if msgList[0].MsgData.SessionType == constant.SuperGroupChatType {
err = db.cache.SetGroupMaxSeq(ctx, sourceID, currentMaxSeq)
err = db.cache.SetGroupMaxSeq(ctx, conversationID, currentMaxSeq)
} else {
err = db.cache.SetUserMaxSeq(ctx, sourceID, currentMaxSeq)
err = db.cache.SetUserMaxSeq(ctx, conversationID, currentMaxSeq)
}
if err != nil {
prome.Inc(prome.SeqSetFailedCounter)
@@ -424,16 +432,16 @@ func (db *msgDatabase) GetMsgAndIndexBySeqsInOneDoc(ctx context.Context, docID s
return seqMsgs, indexes, unExistSeqs, nil
}
func (db *msgDatabase) GetNewestMsg(ctx context.Context, sourceID string) (msgPb *sdkws.MsgData, err error) {
msgInfo, err := db.msgDocDatabase.GetNewestMsg(ctx, sourceID)
func (db *msgDatabase) GetNewestMsg(ctx context.Context, conversationID string) (msgPb *sdkws.MsgData, err error) {
msgInfo, err := db.msgDocDatabase.GetNewestMsg(ctx, conversationID)
if err != nil {
return nil, err
}
return db.unmarshalMsg(msgInfo)
}
func (db *msgDatabase) GetOldestMsg(ctx context.Context, sourceID string) (msgPb *sdkws.MsgData, err error) {
msgInfo, err := db.msgDocDatabase.GetOldestMsg(ctx, sourceID)
func (db *msgDatabase) GetOldestMsg(ctx context.Context, conversationID string) (msgPb *sdkws.MsgData, err error) {
msgInfo, err := db.msgDocDatabase.GetOldestMsg(ctx, conversationID)
if err != nil {
return nil, err
}
@@ -449,10 +457,10 @@ func (db *msgDatabase) unmarshalMsg(msgInfo *unRelationTb.MsgInfoModel) (msgPb *
return msgPb, nil
}
func (db *msgDatabase) getMsgBySeqs(ctx context.Context, sourceID string, seqs []int64, diffusionType int) (seqMsgs []*sdkws.MsgData, err error) {
func (db *msgDatabase) getMsgBySeqs(ctx context.Context, conversationID string, seqs []int64, diffusionType int) (seqMsgs []*sdkws.MsgData, err error) {
var hasSeqs []int64
singleCount := 0
m := db.msg.GetDocIDSeqsMap(sourceID, seqs)
m := db.msg.GetDocIDSeqsMap(conversationID, seqs)
for docID, value := range m {
doc, err := db.msgDocDatabase.FindOneByDocID(ctx, docID)
if err != nil {
@@ -483,7 +491,7 @@ func (db *msgDatabase) getMsgBySeqs(ctx context.Context, sourceID string, seqs [
if diffusionType == constant.WriteDiffusion {
exceptionMsg = db.msg.GenExceptionMessageBySeqs(diff)
} else if diffusionType == constant.ReadDiffusion {
exceptionMsg = db.msg.GenExceptionSuperGroupMessageBySeqs(diff, sourceID)
exceptionMsg = db.msg.GenExceptionSuperGroupMessageBySeqs(diff, conversationID)
}
seqMsgs = append(seqMsgs, exceptionMsg...)
}
@@ -511,6 +519,31 @@ func (db *msgDatabase) GetMsgBySeqs(ctx context.Context, userID string, seqs []i
return successMsgs, nil
}
func (db *msgDatabase) GetMsgBySeqsRange(ctx context.Context, conversationID string, begin, end, num int64) (seqMsg []*sdkws.MsgData, err error) {
var seqs []int64
for i := begin; i <= end; i++ {
seqs = append(seqs, i)
}
successMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs)
if err != nil {
if err != redis.Nil {
prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs))
log.Error(mcontext.GetOperationID(ctx), "get message from redis exception", err.Error(), failedSeqs)
}
}
prome.Add(prome.MsgPullFromRedisSuccessCounter, len(successMsgs))
if len(failedSeqs) > 0 {
mongoMsgs, err := db.getMsgBySeqs(ctx, conversationID, seqs, constant.ReadDiffusion)
if err != nil {
prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs))
return nil, err
}
prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs))
successMsgs = append(successMsgs, mongoMsgs...)
}
return successMsgs, nil
}
func (db *msgDatabase) GetSuperGroupMsgBySeqs(ctx context.Context, groupID string, seqs []int64) (successMsgs []*sdkws.MsgData, err error) {
successMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, groupID, seqs)
if err != nil {
@@ -595,13 +628,13 @@ func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
// seq 70
// set minSeq 21
// recursion 删除list并且返回设置的最小seq
func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string, index int64, delStruct *delMsgRecursionStruct, remainTime int64) (int64, error) {
func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, conversationID string, index int64, delStruct *delMsgRecursionStruct, remainTime int64) (int64, error) {
// find from oldest list
msgs, err := db.msgDocDatabase.GetMsgsByIndex(ctx, sourceID, index)
msgs, err := db.msgDocDatabase.GetMsgsByIndex(ctx, conversationID, index)
if err != nil || msgs.DocID == "" {
if err != nil {
if err == unrelation.ErrMsgListNotExist {
log.NewDebug(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), "ID:", sourceID, "index:", index, err.Error())
log.NewDebug(mcontext.GetOperationID(ctx), utils.GetSelfFuncName(), "ID:", conversationID, "index:", index, err.Error())
} else {
//log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID)
}
@@ -613,7 +646,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
}
return delStruct.getSetMinSeq() + 1, nil
}
//log.NewDebug(operationID, "ID:", sourceID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg))
//log.NewDebug(operationID, "ID:", conversationID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg))
if int64(len(msgs.Msg)) > db.msg.GetSingleGocMsgNum() {
log.ZWarn(ctx, "msgs too large", nil, "lenth", len(msgs.Msg), "docID:", msgs.DocID)
}
@@ -656,7 +689,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
}
}
// 继续递归 index+1
seq, err := db.deleteMsgRecursion(ctx, sourceID, index+1, delStruct, remainTime)
seq, err := db.deleteMsgRecursion(ctx, conversationID, index+1, delStruct, remainTime)
return seq, utils.Wrap(err, "deleteMsg failed")
}
@@ -689,8 +722,8 @@ func (db *msgDatabase) GetSuperGroupMinMaxSeqInMongoAndCache(ctx context.Context
return
}
func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, sourceID string) (minSeqMongo, maxSeqMongo int64, err error) {
oldestMsgMongo, err := db.msgDocDatabase.GetOldestMsg(ctx, sourceID)
func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, conversationID string) (minSeqMongo, maxSeqMongo int64, err error) {
oldestMsgMongo, err := db.msgDocDatabase.GetOldestMsg(ctx, conversationID)
if err != nil {
return 0, 0, err
}
@@ -699,7 +732,7 @@ func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, sourceID string) (
return 0, 0, err
}
minSeqMongo = msgPb.Seq
newestMsgMongo, err := db.msgDocDatabase.GetNewestMsg(ctx, sourceID)
newestMsgMongo, err := db.msgDocDatabase.GetNewestMsg(ctx, conversationID)
if err != nil {
return 0, 0, err
}