blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-05-07 02:56:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

modify msg

Browse Source
This commit is contained in:
wangchuxiao 2022-12-09 18:03:40 +08:00
parent 66f49feebb
commit 24288aa597
4 changed files with 85 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
internal/msg_transfer/logic/online_msg_to_mongo_handler.go
View File

@ -61,7 +61,52 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
}
//if v.MsgData.ContentType == ? {}
} else if v.MsgData.ContentType == constant.ReactionMessageModifierNotification {
var req pbMsg.ModifyMessageReactionExtensionsReq
if req.IsExternalExtensions {
log.NewInfo(req.OperationID, "msg:", req.String(), "this is external extensions")
continue
}
if !req.IsReact {
// first time to modify
var reactionExtensionList = make(map[string]db.KeyValue)
for k, v := range req.ReactionExtensionList {
reactionExtensionList[k] = db.KeyValue{
TypeKey: v.TypeKey,
Value: v.Value,
LatestUpdateTime: v.LatestUpdateTime,
}
}
extendMsg := db.ExtendMsg{
ReactionExtensionList: reactionExtensionList,
ClientMsgID: req.ClientMsgID,
MsgFirstModifyTime: req.MsgFirstModifyTime,
}
if req.AttachedInfo != nil {
extendMsg.AttachedInfo = req.AttachedInfo.Value
}
if req.Ex != nil {
extendMsg.Ex = req.Ex.Value
}
if err := db.DB.InsertExtendMsg(req.SourceID, req.SessionType, &extendMsg); err != nil {
log.NewError(req.OperationID, "MsgFirstModify InsertExtendMsg failed", req.SourceID, req.SessionType, extendMsg, err.Error())
continue
}
} else {
// is already modify
if err := db.DB.InsertOrUpdateReactionExtendMsgSet(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime, req.ReactionExtensionList); err != nil {
log.NewError(req.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
}
}
} else if v.MsgData.ContentType == 2301 {
var req pbMsg.OperateMessageListReactionExtensionsReq
var clientMsgIDList []string
for _, v := range req.MessageReactionKeyList {
clientMsgIDList = append(clientMsgIDList, v.ClientMsgID)
}
if err := db.DB.DeleteReactionExtendMsgSet(req.SourceID, req.SessionType, clientMsgIDList, req.OpUserID); err != nil {
log.NewError(req.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
}
}
}
}

6
internal/rpc/msg/extend_msg_callback.go
View File

@ -1,9 +1,11 @@
package msg
func callbackSetMessageReactionExtensions() {
import "Open_IM/pkg/proto/msg"
func callbackSetMessageReactionExtensions(req *msg.ModifyMessageReactionExtensionsReq) {
}
func callbackDeleteMessageReactionExtensions() {
func callbackDeleteMessageReactionExtensions(req *msg.OperateMessageListReactionExtensionsReq) {
}

8
pkg/base_info/msg.go
View File

@ -69,8 +69,8 @@ type ModifyMessageReactionExtensionsReq struct {
type ModifyMessageReactionExtensionsResp struct {
CommResp
Data struct {
SuccessList []*sdk_ws.ExtendMsg `json:"successList"`
FailedList []*sdk_ws.ExtendMsg `json:"failedList"`
SuccessList []*msg.ExtendMsgResp `json:"successList"`
FailedList []*msg.ExtendMsgResp `json:"failedList"`
} `json:"data"`
}
@ -84,8 +84,8 @@ type OperateMessageListReactionExtensionsReq struct {
type OperateMessageListReactionExtensionsResp struct {
CommResp
Data struct {
SuccessList []*sdk_ws.ExtendMsg `json:"successList"`
FailedList []*sdk_ws.ExtendMsg `json:"failedList"`
SuccessList []*msg.ExtendMsgResp `json:"successList"`
FailedList []*msg.ExtendMsgResp `json:"failedList"`
} `json:"data"`
}

65
pkg/common/db/extend_msg_mongo_model.go
View File

@ -14,10 +14,11 @@ import (
)
const cExtendMsgSet = "extend_msgs"
const MaxNum = 100
type ExtendMsgSet struct {
SourceID string `bson:"id" json:"ID"`
SessionType string `bson:"session_type" json:"sessionType"`
SourceID string `bson:"source_id" json:"ID"`
SessionType int32 `bson:"session_type" json:"sessionType"`
ExtendMsgs map[string]ExtendMsg `bson:"extend_msgs" json:"extendMsgs"`
ExtendMsgNum int32 `bson:"extend_msg_num" json:"extendMsgNum"`
CreateTime int64 `bson:"create_time" json:"createTime"` // this block's create time
@ -31,11 +32,11 @@ type KeyValue struct {
}
type ExtendMsg struct {
ReactionExtensionList map[string]KeyValue `bson:"content" json:"content"`
ReactionExtensionList map[string]KeyValue `bson:"reaction_extension_list" json:"reactionExtensionList"`
ClientMsgID string `bson:"client_msg_id" json:"clientMsgID"`
MsgFirstModifyTime int64 `bson:"create_time" json:"createTime"` // this extendMsg create time
AttachedInfo *string `bson:"attached_info" json:"attachedInfo"`
Ex *string `bson:"ex" json:"ex"`
MsgFirstModifyTime int64 `bson:"msg_first_modify_time" json:"msgFirstModifyTime"` // this extendMsg create time
AttachedInfo string `bson:"attached_info" json:"attachedInfo"`
Ex string `bson:"ex" json:"ex"`
}
func GetExtendMsgSetID(ID string, index int32) string {
@ -79,35 +80,35 @@ type GetExtendMsgSetOpts struct {
ExcludeExtendMsgs bool
}
func (d *DataBases) GetExtendMsgSet(ID string, index int32, opts *GetExtendMsgSetOpts) (*ExtendMsgSet, error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
var set ExtendMsgSet
var findOneOpt *options.FindOneOptions
if opts != nil {
if opts.ExcludeExtendMsgs {
findOneOpt = &options.FindOneOptions{}
findOneOpt.SetProjection(bson.M{"extend_msgs": 0})
}
}
err := c.FindOne(ctx, bson.M{"uid": GetExtendMsgSetID(ID, index)}, findOneOpt).Decode(&set)
return &set, err
}
// first modify msg
func (d *DataBases) InsertExtendMsgAndGetIndex(ID string, index int32, msg *ExtendMsg) error {
func (d *DataBases) InsertExtendMsg(sourceID string, sessionType int32, msg *ExtendMsg) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
_, err := c.UpdateOne(ctx, bson.M{"uid": GetExtendMsgSetID(ID, index)}, bson.M{"$set": bson.M{"latest_update_time": utils.GetCurrentTimestampBySecond(), "$inc": bson.M{"extend_msg_num": 1}, fmt.Sprintf("extend_msgs.%s", msg.ClientMsgID): msg}})
result, err := c.UpdateOne(ctx, bson.M{"source_id": sourceID, "session_type": sessionType}, bson.M{"$set": bson.M{"max_msg_update_time": msg.MsgFirstModifyTime, "$inc": bson.M{"extend_msg_num": 1}, fmt.Sprintf("extend_msgs.%s", msg.ClientMsgID): msg}})
if err != nil {
return utils.Wrap(err, "")
}
if result.UpsertedCount == 0 {
if err := d.CreateExtendMsgSet(&ExtendMsgSet{
SourceID: sourceID,
SessionType: sessionType,
ExtendMsgs: map[string]ExtendMsg{msg.ClientMsgID: *msg},
ExtendMsgNum: 1,
CreateTime: msg.MsgFirstModifyTime,
MaxMsgUpdateTime: msg.MsgFirstModifyTime,
}); err != nil {
return err
}
}
return nil
}
// insert or update
func (d *DataBases) InsertOrUpdateReactionExtendMsgSet(ID string, index int32, clientMsgID, userID, value string) error {
func (d *DataBases) InsertOrUpdateReactionExtendMsgSet(sourceID string, sessionType int32, clientMsgID, typeKey, value string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
reactionExtendMsgSet := KeyValue{
TypeKey: userID,
TypeKey: typeKey,
Value: value,
LatestUpdateTime: utils.GetCurrentTimestampBySecond(),
}
@ -115,21 +116,17 @@ func (d *DataBases) InsertOrUpdateReactionExtendMsgSet(ID string, index int32, c
opt := &options.UpdateOptions{
Upsert: &upsert,
}
_, err := c.UpdateOne(ctx, bson.M{"uid": GetExtendMsgSetID(ID, index)}, bson.M{"$set": bson.M{"latest_update_time": utils.GetCurrentTimestampBySecond()}, fmt.Sprintf("extend_msgs.%s.%s", clientMsgID, userID): reactionExtendMsgSet}, opt)
_, err := c.UpdateOne(ctx, bson.M{"source_id": sourceID, "session_type": sessionType}, bson.M{"$set": bson.M{fmt.Sprintf("extend_msgs.%s.%s", clientMsgID, typeKey): reactionExtendMsgSet}}, opt)
return err
}
func (d *DataBases) DeleteReactionExtendMsgSet(ID string, index int32, clientMsgID, userID string) error {
// delete TypeKey
func (d *DataBases) DeleteReactionExtendMsgSet(sourceID string, sessionType int32, clientMsgID []string, userID string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
_, err := c.UpdateOne(ctx, bson.M{"uid": GetExtendMsgSetID(ID, index)}, bson.M{"$unset": bson.M{fmt.Sprintf("extend_msgs.%s.%s", clientMsgID, userID): ""}})
_, err := c.UpdateOne(ctx, bson.M{"source_id": sourceID, "session_type": sessionType}, bson.M{"$unset": bson.M{fmt.Sprintf("extend_msgs.%s.%s", clientMsgID, userID): ""}})
return err
}
// by index start end
func (d *DataBases) GetExtendMsgList(ID string, index int32, clientMsgID string) (extendMsg *ExtendMsg, err error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cExtendMsgSet)
err = c.FindOne(ctx, bson.M{"uid": GetExtendMsgSetID(ID, index), "extend_msgs": bson.M{}}).Decode(&extendMsg)
return extendMsg, err
func (d *DataBases) GetExtendMsg(sourceID string, sessionType int32, clientMsgID string, maxMsgUpdateTime int64) error {
}