You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Open-IM-Server/pkg/common/db/unrelation/mongo_model.go

752 lines
24 KiB

2 years ago
package unrelation
import (
"Open_IM/pkg/common/config"
3 years ago
"Open_IM/pkg/common/constant"
2 years ago
"Open_IM/pkg/common/db"
"Open_IM/pkg/common/log"
2 years ago
pbMsg "Open_IM/pkg/proto/msg"
3 years ago
open_im_sdk "Open_IM/pkg/proto/sdk_ws"
3 years ago
"Open_IM/pkg/utils"
3 years ago
"context"
"errors"
3 years ago
"fmt"
2 years ago
"math/rand"
"sync"
"github.com/go-redis/redis/v8"
3 years ago
"github.com/gogo/protobuf/sortkeys"
2 years ago
"go.mongodb.org/mongo-driver/bson/primitive"
3 years ago
"go.mongodb.org/mongo-driver/mongo"
3 years ago
"go.mongodb.org/mongo-driver/mongo/options"
3 years ago
3 years ago
//"github.com/garyburd/redigo/redis"
"github.com/golang/protobuf/proto"
3 years ago
"go.mongodb.org/mongo-driver/bson"
3 years ago
"strconv"
"time"
)
const cChat = "msg"
3 years ago
const cGroup = "group"
2 years ago
3 years ago
const cCommentMsg = "comment_msg"
2 years ago
const singleGocMsgNum = 5000
3 years ago
3 years ago
func GetSingleGocMsgNum() int {
return singleGocMsgNum
}
3 years ago
type MsgInfo struct {
SendTime int64
Msg []byte
}
type UserChat struct {
UID string
2 years ago
//ListIndex int `bson:"index"`
3 years ago
Msg []MsgInfo
}
3 years ago
type GroupMember_x struct {
3 years ago
GroupID string
UIDList []string
}
2 years ago
var ErrMsgListNotExist = errors.New("user not have msg in mongoDB")
// deleteMsgByLogic
2 years ago
func (d *db.DataBases) DelMsgBySeqList(userID string, seqList []uint32, operationID string) (totalUnexistSeqList []uint32, err error) {
log.Debug(operationID, utils.GetSelfFuncName(), "args ", userID, seqList)
sortkeys.Uint32s(seqList)
suffixUserID2SubSeqList := func(uid string, seqList []uint32) map[string][]uint32 {
t := make(map[string][]uint32)
for i := 0; i < len(seqList); i++ {
seqUid := getSeqUid(uid, seqList[i])
if value, ok := t[seqUid]; !ok {
var temp []uint32
t[seqUid] = append(temp, seqList[i])
} else {
t[seqUid] = append(value, seqList[i])
}
}
return t
}(userID, seqList)
3 years ago
lock := sync.Mutex{}
var wg sync.WaitGroup
wg.Add(len(suffixUserID2SubSeqList))
for k, v := range suffixUserID2SubSeqList {
go func(suffixUserID string, subSeqList []uint32, operationID string) {
3 years ago
defer wg.Done()
unexistSeqList, err := d.DelMsgBySeqListInOneDoc(suffixUserID, subSeqList, operationID)
if err != nil {
3 years ago
log.Error(operationID, "DelMsgBySeqListInOneDoc failed ", err.Error(), suffixUserID, subSeqList)
3 years ago
return
}
3 years ago
lock.Lock()
totalUnexistSeqList = append(totalUnexistSeqList, unexistSeqList...)
lock.Unlock()
}(k, v, operationID)
}
3 years ago
return totalUnexistSeqList, err
}
2 years ago
func (d *db.DataBases) DelMsgBySeqListInOneDoc(suffixUserID string, seqList []uint32, operationID string) ([]uint32, error) {
log.Debug(operationID, utils.GetSelfFuncName(), "args ", suffixUserID, seqList)
3 years ago
seqMsgList, indexList, unexistSeqList, err := d.GetMsgAndIndexBySeqListInOneMongo2(suffixUserID, seqList, operationID)
if err != nil {
3 years ago
return nil, utils.Wrap(err, "")
}
for i, v := range seqMsgList {
if err := d.ReplaceMsgByIndex(suffixUserID, v, operationID, indexList[i]); err != nil {
3 years ago
return nil, utils.Wrap(err, "")
}
}
3 years ago
return unexistSeqList, nil
}
3 years ago
// deleteMsgByLogic
2 years ago
func (d *db.DataBases) DelMsgLogic(uid string, seqList []uint32, operationID string) error {
3 years ago
sortkeys.Uint32s(seqList)
seqMsgs, err := d.GetMsgBySeqListMongo2(uid, seqList, operationID)
if err != nil {
3 years ago
return utils.Wrap(err, "")
3 years ago
}
for _, seqMsg := range seqMsgs {
3 years ago
log.NewDebug(operationID, utils.GetSelfFuncName(), *seqMsg)
3 years ago
seqMsg.Status = constant.MsgDeleted
if err = d.ReplaceMsgBySeq(uid, seqMsg, operationID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "ReplaceMsgListBySeq error", err.Error())
}
}
return nil
}
2 years ago
func (d *db.DataBases) ReplaceMsgByIndex(suffixUserID string, msg *open_im_sdk.MsgData, operationID string, seqIndex int) error {
log.NewInfo(operationID, utils.GetSelfFuncName(), suffixUserID, *msg)
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
s := fmt.Sprintf("msg.%d.msg", seqIndex)
log.NewDebug(operationID, utils.GetSelfFuncName(), seqIndex, s)
msg.Status = constant.MsgDeleted
bytes, err := proto.Marshal(msg)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "proto marshal failed ", err.Error(), msg.String())
return utils.Wrap(err, "")
}
updateResult, err := c.UpdateOne(ctx, bson.M{"uid": suffixUserID}, bson.M{"$set": bson.M{s: bytes}})
log.NewInfo(operationID, utils.GetSelfFuncName(), updateResult)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "UpdateOne", err.Error())
return utils.Wrap(err, "")
}
return nil
}
2 years ago
func (d *db.DataBases) ReplaceMsgBySeq(uid string, msg *open_im_sdk.MsgData, operationID string) error {
3 years ago
log.NewInfo(operationID, utils.GetSelfFuncName(), uid, *msg)
3 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
uid = getSeqUid(uid, msg.Seq)
seqIndex := getMsgIndex(msg.Seq)
s := fmt.Sprintf("msg.%d.msg", seqIndex)
log.NewDebug(operationID, utils.GetSelfFuncName(), seqIndex, s)
3 years ago
bytes, err := proto.Marshal(msg)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "proto marshal", err.Error())
return utils.Wrap(err, "")
}
3 years ago
updateResult, err := c.UpdateOne(
ctx, bson.M{"uid": uid},
3 years ago
bson.M{"$set": bson.M{s: bytes}})
3 years ago
log.NewInfo(operationID, utils.GetSelfFuncName(), updateResult)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "UpdateOne", err.Error())
3 years ago
return utils.Wrap(err, "")
3 years ago
}
return nil
2 years ago
}
2 years ago
func (d *db.DataBases) UpdateOneMsgList(msg *UserChat) error {
2 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
_, err := c.UpdateOne(ctx, bson.M{"uid": msg.UID}, bson.M{"$set": bson.M{"msg": msg.Msg}})
return err
3 years ago
}
2 years ago
func (d *db.DataBases) GetMsgBySeqList(uid string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
3 years ago
log.NewInfo(operationID, utils.GetSelfFuncName(), uid, seqList)
3 years ago
var hasSeqList []uint32
singleCount := 0
session := d.mgoSession.Clone()
if session == nil {
3 years ago
return nil, errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
3 years ago
m := func(uid string, seqList []uint32) map[string][]uint32 {
t := make(map[string][]uint32)
for i := 0; i < len(seqList); i++ {
seqUid := getSeqUid(uid, seqList[i])
if value, ok := t[seqUid]; !ok {
var temp []uint32
t[seqUid] = append(temp, seqList[i])
} else {
3 years ago
t[seqUid] = append(value, seqList[i])
}
}
3 years ago
return t
}(uid, seqList)
sChat := UserChat{}
for seqUid, value := range m {
if err = c.Find(bson.M{"uid": seqUid}).One(&sChat); err != nil {
log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error())
continue
}
singleCount = 0
for i := 0; i < len(sChat.Msg); i++ {
msg := new(open_im_sdk.MsgData)
if err = proto.Unmarshal(sChat.Msg[i].Msg, msg); err != nil {
log.NewError(operationID, "Unmarshal err", seqUid, value, uid, seqList, err.Error())
return nil, err
}
if isContainInt32(msg.Seq, value) {
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(value) {
break
}
}
}
}
if len(hasSeqList) != len(seqList) {
var diff []uint32
diff = utils.Difference(hasSeqList, seqList)
exceptionMSg := genExceptionMessageBySeqList(diff)
seqMsg = append(seqMsg, exceptionMSg...)
}
3 years ago
return seqMsg, nil
}
3 years ago
2 years ago
func (d *db.DataBases) GetUserMsgListByIndex(ID string, index int64) (*UserChat, error) {
2 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
2 years ago
regex := fmt.Sprintf("^%s", ID)
2 years ago
findOpts := options.Find().SetLimit(1).SetSkip(index).SetSort(bson.M{"uid": 1})
2 years ago
var msgs []UserChat
2 years ago
//primitive.Regex{Pattern: regex}
2 years ago
cursor, err := c.Find(ctx, bson.M{"uid": primitive.Regex{Pattern: regex}}, findOpts)
2 years ago
if err != nil {
2 years ago
return nil, utils.Wrap(err, "")
2 years ago
}
2 years ago
err = cursor.All(context.Background(), &msgs)
2 years ago
if err != nil {
2 years ago
return nil, utils.Wrap(err, fmt.Sprintf("cursor is %s", cursor.Current.String()))
2 years ago
}
2 years ago
if len(msgs) > 0 {
2 years ago
return &msgs[0], nil
2 years ago
} else {
2 years ago
return nil, ErrMsgListNotExist
2 years ago
}
2 years ago
}
2 years ago
func (d *db.DataBases) DelMongoMsgs(IDList []string) error {
2 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
_, err := c.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": IDList}})
return err
}
2 years ago
func (d *db.DataBases) ReplaceMsgToBlankByIndex(suffixID string, index int) (replaceMaxSeq uint32, err error) {
2 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
userChat := &UserChat{}
2 years ago
err = c.FindOne(ctx, bson.M{"uid": suffixID}).Decode(&userChat)
2 years ago
if err != nil {
2 years ago
return 0, err
2 years ago
}
for i, msg := range userChat.Msg {
if i <= index {
2 years ago
msgPb := &open_im_sdk.MsgData{}
if err = proto.Unmarshal(msg.Msg, msgPb); err != nil {
continue
}
newMsgPb := &open_im_sdk.MsgData{Seq: msgPb.Seq}
bytes, err := proto.Marshal(newMsgPb)
if err != nil {
continue
}
msg.Msg = bytes
2 years ago
msg.SendTime = 0
2 years ago
replaceMaxSeq = msgPb.Seq
2 years ago
}
}
2 years ago
_, err = c.UpdateOne(ctx, bson.M{"uid": suffixID}, bson.M{"$set": bson.M{"msg": userChat.Msg}})
2 years ago
return replaceMaxSeq, err
2 years ago
}
2 years ago
func (d *db.DataBases) GetNewestMsg(ID string) (msg *open_im_sdk.MsgData, err error) {
2 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
regex := fmt.Sprintf("^%s", ID)
2 years ago
findOpts := options.Find().SetLimit(1).SetSort(bson.M{"uid": -1})
2 years ago
var userChats []UserChat
cursor, err := c.Find(ctx, bson.M{"uid": bson.M{"$regex": regex}}, findOpts)
if err != nil {
return nil, err
}
2 years ago
err = cursor.All(ctx, &userChats)
2 years ago
if err != nil {
return nil, utils.Wrap(err, "")
}
2 years ago
if len(userChats) > 0 {
if len(userChats[0].Msg) > 0 {
msgPb := &open_im_sdk.MsgData{}
err = proto.Unmarshal(userChats[0].Msg[len(userChats[0].Msg)-1].Msg, msgPb)
if err != nil {
return nil, utils.Wrap(err, "")
}
return msgPb, nil
2 years ago
}
return nil, errors.New("len(userChats[0].Msg) < 0")
}
2 years ago
return nil, nil
2 years ago
}
2 years ago
func (d *db.DataBases) GetOldestMsg(ID string) (msg *open_im_sdk.MsgData, err error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
regex := fmt.Sprintf("^%s", ID)
findOpts := options.Find().SetLimit(1).SetSort(bson.M{"uid": 1})
var userChats []UserChat
cursor, err := c.Find(ctx, bson.M{"uid": bson.M{"$regex": regex}}, findOpts)
if err != nil {
return nil, err
}
err = cursor.All(ctx, &userChats)
if err != nil {
return nil, utils.Wrap(err, "")
}
var oldestMsg []byte
if len(userChats) > 0 {
for _, v := range userChats[0].Msg {
if v.SendTime != 0 {
oldestMsg = v.Msg
break
}
}
if len(oldestMsg) == 0 {
oldestMsg = userChats[0].Msg[len(userChats[0].Msg)-1].Msg
}
msgPb := &open_im_sdk.MsgData{}
err = proto.Unmarshal(oldestMsg, msgPb)
if err != nil {
return nil, utils.Wrap(err, "")
}
return msgPb, nil
}
return nil, nil
}
2 years ago
func (d *db.DataBases) GetMsgBySeqListMongo2(uid string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
3 years ago
var hasSeqList []uint32
singleCount := 0
3 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
3 years ago
m := func(uid string, seqList []uint32) map[string][]uint32 {
t := make(map[string][]uint32)
for i := 0; i < len(seqList); i++ {
seqUid := getSeqUid(uid, seqList[i])
if value, ok := t[seqUid]; !ok {
3 years ago
var temp []uint32
t[seqUid] = append(temp, seqList[i])
} else {
t[seqUid] = append(value, seqList[i])
}
}
return t
}(uid, seqList)
sChat := UserChat{}
for seqUid, value := range m {
3 years ago
if err = c.FindOne(ctx, bson.M{"uid": seqUid}).Decode(&sChat); err != nil {
3 years ago
log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error())
continue
}
singleCount = 0
for i := 0; i < len(sChat.Msg); i++ {
msg := new(open_im_sdk.MsgData)
if err = proto.Unmarshal(sChat.Msg[i].Msg, msg); err != nil {
3 years ago
log.NewError(operationID, "Unmarshal err", seqUid, value, uid, seqList, err.Error())
3 years ago
return nil, err
}
3 years ago
if isContainInt32(msg.Seq, value) {
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(value) {
break
}
}
}
}
3 years ago
if len(hasSeqList) != len(seqList) {
var diff []uint32
diff = utils.Difference(hasSeqList, seqList)
exceptionMSg := genExceptionMessageBySeqList(diff)
seqMsg = append(seqMsg, exceptionMSg...)
}
return seqMsg, nil
}
2 years ago
func (d *db.DataBases) GetSuperGroupMsgBySeqListMongo(groupID string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
var hasSeqList []uint32
singleCount := 0
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
m := func(uid string, seqList []uint32) map[string][]uint32 {
t := make(map[string][]uint32)
for i := 0; i < len(seqList); i++ {
seqUid := getSeqUid(uid, seqList[i])
if value, ok := t[seqUid]; !ok {
var temp []uint32
t[seqUid] = append(temp, seqList[i])
} else {
t[seqUid] = append(value, seqList[i])
}
}
return t
}(groupID, seqList)
sChat := UserChat{}
for seqUid, value := range m {
if err = c.FindOne(ctx, bson.M{"uid": seqUid}).Decode(&sChat); err != nil {
log.NewError(operationID, "not find seqGroupID", seqUid, value, groupID, seqList, err.Error())
continue
}
singleCount = 0
for i := 0; i < len(sChat.Msg); i++ {
msg := new(open_im_sdk.MsgData)
if err = proto.Unmarshal(sChat.Msg[i].Msg, msg); err != nil {
log.NewError(operationID, "Unmarshal err", seqUid, value, groupID, seqList, err.Error())
return nil, err
}
if isContainInt32(msg.Seq, value) {
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(value) {
break
}
}
}
}
if len(hasSeqList) != len(seqList) {
var diff []uint32
diff = utils.Difference(hasSeqList, seqList)
exceptionMSg := genExceptionSuperGroupMessageBySeqList(diff, groupID)
seqMsg = append(seqMsg, exceptionMSg...)
}
return seqMsg, nil
}
3 years ago
2 years ago
func (d *db.DataBases) GetMsgAndIndexBySeqListInOneMongo2(suffixUserID string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, indexList []int, unexistSeqList []uint32, err error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
sChat := UserChat{}
if err = c.FindOne(ctx, bson.M{"uid": suffixUserID}).Decode(&sChat); err != nil {
log.NewError(operationID, "not find seqUid", suffixUserID, err.Error())
3 years ago
return nil, nil, nil, utils.Wrap(err, "")
}
singleCount := 0
var hasSeqList []uint32
for i := 0; i < len(sChat.Msg); i++ {
msg := new(open_im_sdk.MsgData)
if err = proto.Unmarshal(sChat.Msg[i].Msg, msg); err != nil {
log.NewError(operationID, "Unmarshal err", msg.String(), err.Error())
3 years ago
return nil, nil, nil, err
}
if isContainInt32(msg.Seq, seqList) {
indexList = append(indexList, i)
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(seqList) {
break
}
}
}
3 years ago
for _, i := range seqList {
if isContainInt32(i, hasSeqList) {
continue
}
unexistSeqList = append(unexistSeqList, i)
}
return seqMsg, indexList, unexistSeqList, nil
}
3 years ago
func genExceptionMessageBySeqList(seqList []uint32) (exceptionMsg []*open_im_sdk.MsgData) {
for _, v := range seqList {
msg := new(open_im_sdk.MsgData)
msg.Seq = v
exceptionMsg = append(exceptionMsg, msg)
}
return exceptionMsg
}
func genExceptionSuperGroupMessageBySeqList(seqList []uint32, groupID string) (exceptionMsg []*open_im_sdk.MsgData) {
for _, v := range seqList {
msg := new(open_im_sdk.MsgData)
msg.Seq = v
msg.GroupID = groupID
msg.SessionType = constant.SuperGroupChatType
exceptionMsg = append(exceptionMsg, msg)
}
return exceptionMsg
}
2 years ago
func (d *db.DataBases) SaveUserChatMongo2(uid string, sendTime int64, m *pbMsg.MsgDataToDB) error {
3 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
newTime := getCurrentTimestampByMill()
operationID := ""
seqUid := getSeqUid(uid, m.MsgData.Seq)
filter := bson.M{"uid": seqUid}
var err error
sMsg := MsgInfo{}
sMsg.SendTime = sendTime
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
3 years ago
return utils.Wrap(err, "")
3 years ago
}
err = c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": sMsg}}).Err()
3 years ago
log.NewWarn(operationID, "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
3 years ago
if err != nil {
sChat := UserChat{}
sChat.UID = seqUid
sChat.Msg = append(sChat.Msg, sMsg)
3 years ago
if _, err = c.InsertOne(ctx, &sChat); err != nil {
3 years ago
log.NewDebug(operationID, "InsertOne failed", filter)
return utils.Wrap(err, "")
}
3 years ago
} else {
3 years ago
log.NewDebug(operationID, "FindOneAndUpdate ok", filter)
}
log.NewDebug(operationID, "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
return nil
}
3 years ago
//
//func (d *DataBases) SaveUserChatListMongo2(uid string, sendTime int64, msgList []*pbMsg.MsgDataToDB) error {
// ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
// c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
// newTime := getCurrentTimestampByMill()
// operationID := ""
// seqUid := ""
// msgListToMongo := make([]MsgInfo, 0)
//
// for _, m := range msgList {
// seqUid = getSeqUid(uid, m.MsgData.Seq)
// var err error
// sMsg := MsgInfo{}
// sMsg.SendTime = sendTime
// if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
// return utils.Wrap(err, "")
// }
// msgListToMongo = append(msgListToMongo, sMsg)
// }
//
// filter := bson.M{"uid": seqUid}
// log.NewDebug(operationID, "filter ", seqUid)
// err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err()
// log.NewWarn(operationID, "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
// if err != nil {
// sChat := UserChat{}
// sChat.UID = seqUid
// sChat.Msg = msgListToMongo
//
// if _, err = c.InsertOne(ctx, &sChat); err != nil {
// log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
// return utils.Wrap(err, "")
// }
// } else {
// log.NewDebug(operationID, "FindOneAndUpdate ok", filter)
// }
//
// log.NewDebug(operationID, "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
// return nil
//}
2 years ago
func (d *db.DataBases) SaveUserChat(uid string, sendTime int64, m *pbMsg.MsgDataToDB) error {
var seqUid string
newTime := getCurrentTimestampByMill()
3 years ago
session := d.mgoSession.Clone()
if session == nil {
return errors.New("session == nil")
}
defer session.Close()
3 years ago
log.NewDebug("", "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
3 years ago
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
seqUid = getSeqUid(uid, m.MsgData.Seq)
n, err := c.Find(bson.M{"uid": seqUid}).Count()
if err != nil {
return err
}
3 years ago
log.NewDebug("", "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
3 years ago
sMsg := MsgInfo{}
sMsg.SendTime = sendTime
3 years ago
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
3 years ago
return err
}
if n == 0 {
sChat := UserChat{}
sChat.UID = seqUid
3 years ago
sChat.Msg = append(sChat.Msg, sMsg)
err = c.Insert(&sChat)
if err != nil {
return err
}
} else {
err = c.Update(bson.M{"uid": seqUid}, bson.M{"$push": bson.M{"msg": sMsg}})
if err != nil {
return err
}
}
3 years ago
log.NewDebug("", "insert mgo data cost time", getCurrentTimestampByMill()-newTime)
return nil
}
2 years ago
func (d *db.DataBases) DelUserChatMongo2(uid string) error {
3 years ago
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
filter := bson.M{"uid": uid}
3 years ago
delTime := time.Now().Unix() - int64(config.Config.Mongo.DBRetainChatRecords)*24*3600
3 years ago
if _, err := c.UpdateOne(ctx, filter, bson.M{"$pull": bson.M{"msg": bson.M{"sendtime": bson.M{"$lte": delTime}}}}); err != nil {
return utils.Wrap(err, "")
}
return nil
}
2 years ago
func (d *db.DataBases) MgoSkipUID(count int) (string, error) {
3 years ago
return "", nil
//session := d.mgoSession.Clone()
//if session == nil {
// return "", errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//
//sChat := UserChat{}
//c.Find(nil).Skip(count).Limit(1).One(&sChat)
//return sChat.UID, nil
}
3 years ago
3 years ago
func generateTagID(tagName, userID string) string {
3 years ago
return utils.Md5(tagName + userID + strconv.Itoa(rand.Int()) + time.Now().String())
3 years ago
}
3 years ago
func generateWorkMomentID(userID string) string {
return utils.Md5(userID + strconv.Itoa(rand.Int()) + time.Now().String())
}
3 years ago
func generateWorkMomentCommentID(workMomentID string) string {
return utils.Md5(workMomentID + strconv.Itoa(rand.Int()) + time.Now().String())
}
func getCurrentTimestampByMill() int64 {
return time.Now().UnixNano() / 1e6
}
3 years ago
func GetCurrentTimestampByMill() int64 {
return time.Now().UnixNano() / 1e6
}
3 years ago
3 years ago
func getSeqUid(uid string, seq uint32) string {
seqSuffix := seq / singleGocMsgNum
return indexGen(uid, seqSuffix)
}
func getSeqUserIDList(userID string, maxSeq uint32) []string {
seqMaxSuffix := maxSeq / singleGocMsgNum
var seqUserIDList []string
for i := 0; i <= int(seqMaxSuffix); i++ {
seqUserID := indexGen(userID, uint32(i))
seqUserIDList = append(seqUserIDList, seqUserID)
}
return seqUserIDList
}
3 years ago
func getSeqSuperGroupID(groupID string, seq uint32) string {
seqSuffix := seq / singleGocMsgNum
return superGroupIndexGen(groupID, seqSuffix)
}
3 years ago
3 years ago
func GetSeqUid(uid string, seq uint32) string {
return getSeqUid(uid, seq)
}
3 years ago
func getMsgIndex(seq uint32) int {
seqSuffix := seq / singleGocMsgNum
var index uint32
if seqSuffix == 0 {
index = (seq - seqSuffix*singleGocMsgNum) - 1
3 years ago
} else {
index = seq - seqSuffix*singleGocMsgNum
}
return int(index)
}
3 years ago
func isContainInt32(target uint32, List []uint32) bool {
for _, element := range List {
if target == element {
return true
}
}
return false
3 years ago
}
3 years ago
func isNotContainInt32(target uint32, List []uint32) bool {
for _, i := range List {
if i == target {
return false
}
}
return true
}
3 years ago
3 years ago
func indexGen(uid string, seqSuffix uint32) string {
return uid + ":" + strconv.FormatInt(int64(seqSuffix), 10)
}
2 years ago
3 years ago
func superGroupIndexGen(groupID string, seqSuffix uint32) string {
return "super_group_" + groupID + ":" + strconv.FormatInt(int64(seqSuffix), 10)
}
2 years ago
func (d *db.DataBases) CleanUpUserMsgFromMongo(userID string, operationID string) error {
ctx := context.Background()
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
maxSeq, err := d.GetUserMaxSeq(userID)
if err == redis.Nil {
return nil
}
if err != nil {
return utils.Wrap(err, "")
}
seqUsers := getSeqUserIDList(userID, uint32(maxSeq))
3 years ago
log.Error(operationID, "getSeqUserIDList", seqUsers)
_, err = c.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": seqUsers}})
if err == mongo.ErrNoDocuments {
return nil
}
return utils.Wrap(err, "")
}