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

makefile

Browse Source
This commit is contained in:
wangchuxiao 2023-03-06 16:23:16 +08:00
parent 8143d589aa
commit 23c24308f7
7 changed files with 70 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/common/notification/extend_msg.go
View File

@ -25,7 +25,7 @@ func (c *Check) ExtendMessageUpdatedNotification(ctx context.Context, sendID str
if len(keyMap) == 0 { if len(keyMap) == 0 {
return return
} }
m.SuccessReactionExtensionList = keyMap m.SuccessReactionExtensions = keyMap
m.ClientMsgID = req.ClientMsgID m.ClientMsgID = req.ClientMsgID
m.IsReact = resp.IsReact m.IsReact = resp.IsReact
m.IsExternalExtensions = req.IsExternalExtensions m.IsExternalExtensions = req.IsExternalExtensions
@ -47,7 +47,7 @@ func (c *Check) ExtendMessageDeleteNotification(ctx context.Context, sendID stri
if len(keyMap) == 0 { if len(keyMap) == 0 {
return return
} }
m.SuccessReactionExtensionList = keyMap m.SuccessReactionExtensions = keyMap
m.ClientMsgID = req.ClientMsgID m.ClientMsgID = req.ClientMsgID
m.MsgFirstModifyTime = req.MsgFirstModifyTime m.MsgFirstModifyTime = req.MsgFirstModifyTime

18
internal/msgtransfer/modify_msg_handler.go
View File

@ -10,7 +10,6 @@ import (
"OpenIM/pkg/common/log" "OpenIM/pkg/common/log"
"OpenIM/pkg/common/tracelog" "OpenIM/pkg/common/tracelog"
pbMsg "OpenIM/pkg/proto/msg" pbMsg "OpenIM/pkg/proto/msg"
sdkws "OpenIM/pkg/proto/sdkws"
"OpenIM/pkg/utils" "OpenIM/pkg/utils"
"context" "context"
"encoding/json" "encoding/json"
@ -23,6 +22,7 @@ type ModifyMsgConsumerHandler struct {
modifyMsgConsumerGroup *kfk.MConsumerGroup modifyMsgConsumerGroup *kfk.MConsumerGroup
extendMsgDatabase controller.ExtendMsgDatabase extendMsgDatabase controller.ExtendMsgDatabase
extendSetMsgModel unRelationTb.ExtendMsgSetModel
} }
func NewModifyMsgConsumerHandler(database controller.ExtendMsgDatabase) *ModifyMsgConsumerHandler { func NewModifyMsgConsumerHandler(database controller.ExtendMsgDatabase) *ModifyMsgConsumerHandler {
@ -66,7 +66,6 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
if !isReactionFromCache { if !isReactionFromCache {
continue continue
} }
ctx := context.Background()
tracelog.SetOperationID(ctx, operationID) tracelog.SetOperationID(ctx, operationID)
if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageModifier { if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageModifier {
notification := &apistruct.ReactionMessageModifierNotification{} notification := &apistruct.ReactionMessageModifierNotification{}
@ -85,7 +84,7 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
ClientMsgID: notification.ClientMsgID, ClientMsgID: notification.ClientMsgID,
MsgFirstModifyTime: notification.MsgFirstModifyTime, MsgFirstModifyTime: notification.MsgFirstModifyTime,
} }
for _, v := range notification.SuccessReactionExtensionList { for _, v := range notification.SuccessReactionExtensions {
reactionExtensionList[v.TypeKey] = unRelationTb.KeyValueModel{ reactionExtensionList[v.TypeKey] = unRelationTb.KeyValueModel{
TypeKey: v.TypeKey, TypeKey: v.TypeKey,
Value: v.Value, Value: v.Value,
@ -98,16 +97,7 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
continue continue
} }
} else { } else {
var reactionExtensionList = make(map[string]*sdkws.KeyValue) if err := mmc.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
for _, v := range notification.SuccessReactionExtensionList {
reactionExtensionList[v.TypeKey] = &sdkws.KeyValue{
TypeKey: v.TypeKey,
Value: v.Value,
LatestUpdateTime: v.LatestUpdateTime,
}
}
// is already modify
if err := mmc.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, reactionExtensionList); err != nil {
log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed") log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
} }
} }
@ -116,7 +106,7 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil { if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
continue continue
} }
if err := mmc.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, notification.SuccessReactionExtensionList); err != nil { if err := mmc.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed") log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
} }
} }

2
internal/msgtransfer/persistent_msg_handler.go
View File

@ -65,7 +65,7 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(ctx context.Context, cMs
} }
if tag { if tag {
log.NewInfo(operationID, "msg_transfer msg persisting", string(msg)) log.NewInfo(operationID, "msg_transfer msg persisting", string(msg))
if err = pc.chatLogInterface.CreateChatLog(msgFromMQ); err != nil { if err = pc.chatLogDatabase.CreateChatLog(msgFromMQ); err != nil {
log.NewError(operationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String()) log.NewError(operationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String())
return return
} }

24
internal/rpc/msg/extend_msg.go
View File

@ -93,7 +93,7 @@ func (m *msgServer) SetMessageReactionExtensions(ctx context.Context, req *msg.S
//} else { //} else {
// log.Debug(req.OperationID, "redis handle secondly", req.String()) // log.Debug(req.OperationID, "redis handle secondly", req.String())
// //
// for k, v := range req.ReactionExtensionList { // for k, v := range req.Pb2Model {
// err := m.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k) // err := m.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
// if err != nil { // if err != nil {
// setKeyResultInfo(&resp, 100, err.Error(), req.ClientMsgID, k, v) // setKeyResultInfo(&resp, 100, err.Error(), req.ClientMsgID, k, v)
@ -179,7 +179,7 @@ func (m *msgServer) GetMessagesReactionExtensions(ctx context.Context, req *msg.
// utils.JsonStringToStruct(v, temp) // utils.JsonStringToStruct(v, temp)
// keyMap[k] = temp // keyMap[k] = temp
// } // }
// oneMessage.ReactionExtensionList = keyMap // oneMessage.Pb2Model = keyMap
// //
// } else { // } else {
// mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, messageValue.ClientMsgID, messageValue.MsgFirstModifyTime) // mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, messageValue.ClientMsgID, messageValue.MsgFirstModifyTime)
@ -191,14 +191,14 @@ func (m *msgServer) GetMessagesReactionExtensions(ctx context.Context, req *msg.
// } // }
// keyMap := make(map[string]*sdkws.KeyValue) // keyMap := make(map[string]*sdkws.KeyValue)
// //
// for k, v := range mongoValue.ReactionExtensionList { // for k, v := range mongoValue.Pb2Model {
// temp := new(sdkws.KeyValue) // temp := new(sdkws.KeyValue)
// temp.TypeKey = v.TypeKey // temp.TypeKey = v.TypeKey
// temp.Value = v.Value // temp.Value = v.Value
// temp.LatestUpdateTime = v.LatestUpdateTime // temp.LatestUpdateTime = v.LatestUpdateTime
// keyMap[k] = temp // keyMap[k] = temp
// } // }
// oneMessage.ReactionExtensionList = keyMap // oneMessage.Pb2Model = keyMap
// } // }
// rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage) // rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
//} //}
@ -218,7 +218,7 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
//if callbackResp.ActionCode != constant.ActionAllow || callbackResp.ErrCode != 0 { //if callbackResp.ActionCode != constant.ActionAllow || callbackResp.ErrCode != 0 {
// rResp.ErrCode = int32(callbackResp.ErrCode) // rResp.ErrCode = int32(callbackResp.ErrCode)
// rResp.ErrMsg = callbackResp.ErrMsg // rResp.ErrMsg = callbackResp.ErrMsg
// for _, value := range req.ReactionExtensionList { // for _, value := range req.Pb2Model {
// temp := new(msg.KeyValueResp) // temp := new(msg.KeyValueResp)
// temp.KeyValue = value // temp.KeyValue = value
// temp.ErrMsg = callbackResp.ErrMsg // temp.ErrMsg = callbackResp.ErrMsg
@ -242,7 +242,7 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
// *req = append((*req)[:i], (*req)[i+1:]...) // *req = append((*req)[:i], (*req)[i+1:]...)
// } // }
// } // }
// }(&req.ReactionExtensionList, v.KeyValue.TypeKey) // }(&req.Pb2Model, v.KeyValue.TypeKey)
// rResp.Result = append(rResp.Result, v) // rResp.Result = append(rResp.Result, v)
// } // }
//} //}
@ -250,7 +250,7 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
//if err != nil { //if err != nil {
// rResp.ErrCode = 100 // rResp.ErrCode = 100
// rResp.ErrMsg = err.Error() // rResp.ErrMsg = err.Error()
// for _, value := range req.ReactionExtensionList { // for _, value := range req.Pb2Model {
// temp := new(msg.KeyValueResp) // temp := new(msg.KeyValueResp)
// temp.KeyValue = value // temp.KeyValue = value
// temp.ErrMsg = err.Error() // temp.ErrMsg = err.Error()
@ -262,7 +262,7 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
// //
//if isExists { //if isExists {
// log.Debug(req.OperationID, "redis handle this delete", req.String()) // log.Debug(req.OperationID, "redis handle this delete", req.String())
// for _, v := range req.ReactionExtensionList { // for _, v := range req.Pb2Model {
// err := m.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, v.TypeKey) // err := m.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, v.TypeKey)
// if err != nil { // if err != nil {
// setDeleteKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, v.TypeKey, v) // setDeleteKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, v.TypeKey, v)
@ -293,7 +293,7 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
// if err != nil { // if err != nil {
// rResp.ErrCode = 100 // rResp.ErrCode = 100
// rResp.ErrMsg = err.Error() // rResp.ErrMsg = err.Error()
// for _, value := range req.ReactionExtensionList { // for _, value := range req.Pb2Model {
// temp := new(msg.KeyValueResp) // temp := new(msg.KeyValueResp)
// temp.KeyValue = value // temp.KeyValue = value
// temp.ErrMsg = err.Error() // temp.ErrMsg = err.Error()
@ -306,7 +306,7 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
// if err != nil { // if err != nil {
// rResp.ErrCode = 200 // rResp.ErrCode = 200
// rResp.ErrMsg = err.Error() // rResp.ErrMsg = err.Error()
// for _, value := range req.ReactionExtensionList { // for _, value := range req.Pb2Model {
// temp := new(msg.KeyValueResp) // temp := new(msg.KeyValueResp)
// temp.KeyValue = value // temp.KeyValue = value
// temp.ErrMsg = err.Error() // temp.ErrMsg = err.Error()
@ -316,10 +316,10 @@ func (m *msgServer) DeleteMessageReactionExtensions(ctx context.Context, req *ms
// return &rResp, nil // return &rResp, nil
// } // }
// setValue := make(map[string]*sdkws.KeyValue) // setValue := make(map[string]*sdkws.KeyValue)
// for _, v := range req.ReactionExtensionList { // for _, v := range req.Pb2Model {
// //
// temp := new(sdkws.KeyValue) // temp := new(sdkws.KeyValue)
// if vv, ok := mongoValue.ReactionExtensionList[v.TypeKey]; ok { // if vv, ok := mongoValue.Pb2Model[v.TypeKey]; ok {
// utils.CopyStructFields(temp, &vv) // utils.CopyStructFields(temp, &vv)
// if v.LatestUpdateTime != vv.LatestUpdateTime { // if v.LatestUpdateTime != vv.LatestUpdateTime {
// setDeleteKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, v.TypeKey, temp) // setDeleteKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, v.TypeKey, temp)

28
pkg/apistruct/msg.go
View File

@ -121,21 +121,21 @@ type DeleteMessageReactionExtensionsResp struct {
} }
type ReactionMessageModifierNotification struct { type ReactionMessageModifierNotification struct {
SourceID string `json:"sourceID" binding:"required"` SourceID string `json:"sourceID" binding:"required"`
OpUserID string `json:"opUserID" binding:"required"` OpUserID string `json:"opUserID" binding:"required"`
SessionType int32 `json:"sessionType" binding:"required"` SessionType int32 `json:"sessionType" binding:"required"`
SuccessReactionExtensionList map[string]*sdkws.KeyValue `json:"reactionExtensionList,omitempty" binding:"required"` SuccessReactionExtensions map[string]*sdkws.KeyValue `json:"reactionExtensionList,omitempty" binding:"required"`
ClientMsgID string `json:"clientMsgID" binding:"required"` ClientMsgID string `json:"clientMsgID" binding:"required"`
IsReact bool `json:"isReact"` IsReact bool `json:"isReact"`
IsExternalExtensions bool `json:"isExternalExtensions"` IsExternalExtensions bool `json:"isExternalExtensions"`
MsgFirstModifyTime int64 `json:"msgFirstModifyTime"` MsgFirstModifyTime int64 `json:"msgFirstModifyTime"`
} }
type ReactionMessageDeleteNotification struct { type ReactionMessageDeleteNotification struct {
SourceID string `json:"sourceID" binding:"required"` SourceID string `json:"sourceID" binding:"required"`
OpUserID string `json:"opUserID" binding:"required"` OpUserID string `json:"opUserID" binding:"required"`
SessionType int32 `json:"sessionType" binding:"required"` SessionType int32 `json:"sessionType" binding:"required"`
SuccessReactionExtensionList map[string]*sdkws.KeyValue `json:"reactionExtensionList,omitempty" binding:"required"` SuccessReactionExtensions map[string]*sdkws.KeyValue `json:"reactionExtensionList,omitempty" binding:"required"`
ClientMsgID string `json:"clientMsgID" binding:"required"` ClientMsgID string `json:"clientMsgID" binding:"required"`
MsgFirstModifyTime int64 `json:"msgFirstModifyTime"` MsgFirstModifyTime int64 `json:"msgFirstModifyTime"`
} }

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

@ -77,8 +77,8 @@ type MsgDatabase interface {
func NewMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.Model) MsgDatabase { func NewMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.Model) MsgDatabase {
return &msgDatabase{ return &msgDatabase{
msgDocModel: msgDocModel, msgDocDatabase: msgDocModel,
cache: cacheModel, cache: cacheModel,
} }
} }
@ -90,22 +90,11 @@ func InitMsgDatabase(rdb redis.UniversalClient, database *mongo.Database) MsgDat
} }
type msgDatabase struct { type msgDatabase struct {
msgDocModel unRelationTb.MsgDocModelInterface msgDocDatabase unRelationTb.MsgDocModelInterface
cache cache.Model extendMsgDatabase unRelationTb.ExtendMsgSetModelInterface
msg unRelationTb.MsgDocModel cache cache.Model
extendMsgModel unRelationTb.ExtendMsgSetModelInterface msg unRelationTb.MsgDocModel
} extendMsgSetModel unRelationTb.ExtendMsgSetModel
func (db *msgDatabase) reactionExtensionList(reactionExtensionList map[string]*sdkws.KeyValue) map[string]*unRelationTb.KeyValueModel {
r := make(map[string]*unRelationTb.KeyValueModel)
for key, value := range reactionExtensionList {
r[key] = &unRelationTb.KeyValueModel{
TypeKey: value.TypeKey,
Value: value.Value,
LatestUpdateTime: value.LatestUpdateTime,
}
}
return r
} }
func (db *msgDatabase) JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error) { func (db *msgDatabase) JudgeMessageReactionExist(ctx context.Context, clientMsgID string, sessionType int32) (bool, error) {
@ -133,11 +122,11 @@ func (db *msgDatabase) DeleteOneMessageKey(ctx context.Context, clientMsgID stri
} }
func (db *msgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error { func (db *msgDatabase) InsertOrUpdateReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
return db.extendMsgModel.InsertOrUpdateReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, db.reactionExtensionList(reactionExtensions)) return db.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, sourceID, 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) { func (db *msgDatabase) GetExtendMsg(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) (*pbMsg.ExtendMsg, error) {
extendMsgSet, err := db.extendMsgModel.GetExtendMsgSet(ctx, sourceID, sessionType, maxMsgUpdateTime) extendMsgSet, err := db.extendMsgDatabase.GetExtendMsgSet(ctx, sourceID, sessionType, maxMsgUpdateTime)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -165,7 +154,7 @@ func (db *msgDatabase) GetExtendMsg(ctx context.Context, sourceID string, sessio
} }
func (db *msgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error { func (db *msgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, sourceID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
return db.extendMsgModel.DeleteReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, db.reactionExtensionList(reactionExtensions)) return db.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, sourceID, sessionType, clientMsgID, msgFirstModifyTime, db.extendMsgSetModel.Pb2Model(reactionExtensions))
} }
func (db *msgDatabase) SetSendMsgStatus(ctx context.Context, id string, status int32) error { func (db *msgDatabase) SetSendMsgStatus(ctx context.Context, id string, status int32) error {
@ -244,13 +233,13 @@ func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, sourceID string,
//filter := bson.M{"uid": seqUid} //filter := bson.M{"uid": seqUid}
//log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo, "userID: ", userID) //log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo, "userID: ", userID)
//err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err() //err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgsToMongo}}}).Err()
err = db.msgDocModel.PushMsgsToDoc(ctx, docID, msgsToMongo) err = db.msgDocDatabase.PushMsgsToDoc(ctx, docID, msgsToMongo)
if err != nil { if err != nil {
if err == mongo.ErrNoDocuments { if err == mongo.ErrNoDocuments {
doc := &unRelationTb.MsgDocModel{} doc := &unRelationTb.MsgDocModel{}
doc.DocID = docID doc.DocID = docID
doc.Msg = msgsToMongo doc.Msg = msgsToMongo
if err = db.msgDocModel.Create(ctx, doc); err != nil { if err = db.msgDocDatabase.Create(ctx, doc); err != nil {
prome.Inc(prome.MsgInsertMongoFailedCounter) prome.Inc(prome.MsgInsertMongoFailedCounter)
//log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) //log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "") return utils.Wrap(err, "")
@ -270,7 +259,7 @@ func (db *msgDatabase) BatchInsertChat2DB(ctx context.Context, sourceID string,
nextDoc.DocID = docIDNext nextDoc.DocID = docIDNext
nextDoc.Msg = msgsToMongoNext nextDoc.Msg = msgsToMongoNext
//log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext, "userID: ", userID) //log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext, "userID: ", userID)
if err = db.msgDocModel.Create(ctx, nextDoc); err != nil { if err = db.msgDocDatabase.Create(ctx, nextDoc); err != nil {
prome.Inc(prome.MsgInsertMongoFailedCounter) prome.Inc(prome.MsgInsertMongoFailedCounter)
//log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) //log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "") return utils.Wrap(err, "")
@ -364,7 +353,7 @@ func (db *msgDatabase) DelMsgBySeqsInOneDoc(ctx context.Context, docID string, s
return nil, err return nil, err
} }
for i, v := range seqMsgs { for i, v := range seqMsgs {
if err = db.msgDocModel.UpdateMsgStatusByIndexInOneDoc(ctx, docID, v, indexes[i], constant.MsgDeleted); err != nil { if err = db.msgDocDatabase.UpdateMsgStatusByIndexInOneDoc(ctx, docID, v, indexes[i], constant.MsgDeleted); err != nil {
return nil, err return nil, err
} }
} }
@ -372,7 +361,7 @@ func (db *msgDatabase) DelMsgBySeqsInOneDoc(ctx context.Context, docID string, s
} }
func (db *msgDatabase) GetMsgAndIndexBySeqsInOneDoc(ctx context.Context, docID string, seqs []int64) (seqMsgs []*sdkws.MsgData, indexes []int, unExistSeqs []int64, err error) { func (db *msgDatabase) GetMsgAndIndexBySeqsInOneDoc(ctx context.Context, docID string, seqs []int64) (seqMsgs []*sdkws.MsgData, indexes []int, unExistSeqs []int64, err error) {
doc, err := db.msgDocModel.FindOneByDocID(ctx, docID) doc, err := db.msgDocDatabase.FindOneByDocID(ctx, docID)
if err != nil { if err != nil {
return nil, nil, nil, err return nil, nil, nil, err
} }
@ -403,7 +392,7 @@ func (db *msgDatabase) GetMsgAndIndexBySeqsInOneDoc(ctx context.Context, docID s
} }
func (db *msgDatabase) GetNewestMsg(ctx context.Context, sourceID string) (msgPb *sdkws.MsgData, err error) { func (db *msgDatabase) GetNewestMsg(ctx context.Context, sourceID string) (msgPb *sdkws.MsgData, err error) {
msgInfo, err := db.msgDocModel.GetNewestMsg(ctx, sourceID) msgInfo, err := db.msgDocDatabase.GetNewestMsg(ctx, sourceID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -411,7 +400,7 @@ func (db *msgDatabase) GetNewestMsg(ctx context.Context, sourceID string) (msgPb
} }
func (db *msgDatabase) GetOldestMsg(ctx context.Context, sourceID string) (msgPb *sdkws.MsgData, err error) { func (db *msgDatabase) GetOldestMsg(ctx context.Context, sourceID string) (msgPb *sdkws.MsgData, err error) {
msgInfo, err := db.msgDocModel.GetOldestMsg(ctx, sourceID) msgInfo, err := db.msgDocDatabase.GetOldestMsg(ctx, sourceID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -432,7 +421,7 @@ func (db *msgDatabase) getMsgBySeqs(ctx context.Context, sourceID string, seqs [
singleCount := 0 singleCount := 0
m := db.msg.GetDocIDSeqsMap(sourceID, seqs) m := db.msg.GetDocIDSeqsMap(sourceID, seqs)
for docID, value := range m { for docID, value := range m {
doc, err := db.msgDocModel.FindOneByDocID(ctx, docID) doc, err := db.msgDocDatabase.FindOneByDocID(ctx, docID)
if err != nil { if err != nil {
//log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error()) //log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error())
continue continue
@ -575,7 +564,7 @@ func (d *delMsgRecursionStruct) getSetMinSeq() int64 {
// recursion 删除list并且返回设置的最小seq // 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, sourceID string, index int64, delStruct *delMsgRecursionStruct, remainTime int64) (int64, error) {
// find from oldest list // find from oldest list
msgs, err := db.msgDocModel.GetMsgsByIndex(ctx, sourceID, index) msgs, err := db.msgDocDatabase.GetMsgsByIndex(ctx, sourceID, index)
if err != nil || msgs.DocID == "" { if err != nil || msgs.DocID == "" {
if err != nil { if err != nil {
if err == unrelation.ErrMsgListNotExist { if err == unrelation.ErrMsgListNotExist {
@ -585,7 +574,7 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
} }
} }
// 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList) // 获取报错，或者获取不到了，物理删除并且返回seq delMongoMsgsPhysical(delStruct.delDocIDList)
err = db.msgDocModel.Delete(ctx, delStruct.delDocIDs) err = db.msgDocDatabase.Delete(ctx, delStruct.delDocIDs)
if err != nil { if err != nil {
return 0, err return 0, err
} }
@ -620,11 +609,11 @@ func (db *msgDatabase) deleteMsgRecursion(ctx context.Context, sourceID string,
msg.SendTime = 0 msg.SendTime = 0
hasMarkDelFlag = true hasMarkDelFlag = true
} else { } else {
if err := db.msgDocModel.Delete(ctx, delStruct.delDocIDs); err != nil { if err := db.msgDocDatabase.Delete(ctx, delStruct.delDocIDs); err != nil {
return 0, err return 0, err
} }
if hasMarkDelFlag { if hasMarkDelFlag {
if err := db.msgDocModel.UpdateOneDoc(ctx, msgs); err != nil { if err := db.msgDocDatabase.UpdateOneDoc(ctx, msgs); err != nil {
return delStruct.getSetMinSeq(), utils.Wrap(err, "") return delStruct.getSetMinSeq(), utils.Wrap(err, "")
} }
} }
@ -668,7 +657,7 @@ func (db *msgDatabase) GetSuperGroupMinMaxSeqInMongoAndCache(ctx context.Context
} }
func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, sourceID string) (minSeqMongo, maxSeqMongo int64, err error) { func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, sourceID string) (minSeqMongo, maxSeqMongo int64, err error) {
oldestMsgMongo, err := db.msgDocModel.GetOldestMsg(ctx, sourceID) oldestMsgMongo, err := db.msgDocDatabase.GetOldestMsg(ctx, sourceID)
if err != nil { if err != nil {
return 0, 0, err return 0, 0, err
} }
@ -677,7 +666,7 @@ func (db *msgDatabase) GetMinMaxSeqMongo(ctx context.Context, sourceID string) (
return 0, 0, err return 0, 0, err
} }
minSeqMongo = msgPb.Seq minSeqMongo = msgPb.Seq
newestMsgMongo, err := db.msgDocModel.GetNewestMsg(ctx, sourceID) newestMsgMongo, err := db.msgDocDatabase.GetNewestMsg(ctx, sourceID)
if err != nil { if err != nil {
return 0, 0, err return 0, 0, err
} }

13
pkg/common/db/table/unrelation/extend_msg_set.go
View File

@ -1,6 +1,7 @@
package unrelation package unrelation
import ( import (
"OpenIM/pkg/proto/sdkws"
"context" "context"
"strconv" "strconv"
"strings" "strings"
@ -66,3 +67,15 @@ func (e *ExtendMsgSetModel) SplitSourceIDAndGetIndex() int32 {
type GetAllExtendMsgSetOpts struct { type GetAllExtendMsgSetOpts struct {
ExcludeExtendMsgs bool ExcludeExtendMsgs bool
} }
func (ExtendMsgSetModel) Pb2Model(reactionExtensionList map[string]*sdkws.KeyValue) map[string]*KeyValueModel {
r := make(map[string]*KeyValueModel)
for key, value := range reactionExtensionList {
r[key] = &KeyValueModel{
TypeKey: value.TypeKey,
Value: value.Value,
LatestUpdateTime: value.LatestUpdateTime,
}
}
return r
}