Files
open-im-server/pkg/common/db/mongoModel.go
T

1217 lines
41 KiB
Go
Raw Normal View History

2021-05-26 19:17:51 +08:00
package db
import (
"Open_IM/pkg/common/config"
2022-03-17 19:00:05 +08:00
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
2022-07-20 21:05:12 +08:00
pbMsg "Open_IM/pkg/proto/msg"
2022-01-18 09:25:41 +08:00
open_im_sdk "Open_IM/pkg/proto/sdk_ws"
2022-01-20 11:36:43 +08:00
"Open_IM/pkg/utils"
2022-02-21 15:01:06 +08:00
"context"
2021-05-26 19:17:51 +08:00
"errors"
2022-03-17 19:00:05 +08:00
"fmt"
2022-05-31 19:16:29 +08:00
"github.com/go-redis/redis/v8"
2022-03-17 19:00:05 +08:00
"github.com/gogo/protobuf/sortkeys"
2022-05-30 11:55:27 +08:00
"go.mongodb.org/mongo-driver/mongo"
2022-03-28 12:17:07 +08:00
"go.mongodb.org/mongo-driver/mongo/options"
2022-03-25 18:46:27 +08:00
"math/rand"
2022-05-04 18:02:24 +08:00
"sync"
2022-03-17 19:00:05 +08:00
2022-02-21 15:01:06 +08:00
//"github.com/garyburd/redigo/redis"
2021-05-26 19:17:51 +08:00
"github.com/golang/protobuf/proto"
2022-04-24 17:42:41 +08:00
"go.mongodb.org/mongo-driver/bson"
2022-02-21 15:01:06 +08:00
"strconv"
2021-05-26 19:17:51 +08:00
"time"
)
2021-12-23 17:34:32 +08:00
const cChat = "msg"
2021-07-15 11:14:07 +08:00
const cGroup = "group"
2022-03-25 18:46:27 +08:00
const cTag = "tag"
2022-03-28 19:18:30 +08:00
const cSendLog = "send_log"
2022-04-18 19:24:36 +08:00
const cWorkMoment = "work_moment"
const cCommentMsg = "comment_msg"
2022-05-27 18:40:24 +08:00
const cSuperGroup = "super_group"
2022-05-30 11:55:27 +08:00
const cUserToSuperGroup = "user_to_super_group"
const singleGocMsgNum = 5000
2021-06-28 15:32:26 +08:00
2022-05-20 11:00:11 +08:00
func GetSingleGocMsgNum() int {
return singleGocMsgNum
}
2021-06-28 15:32:26 +08:00
type MsgInfo struct {
SendTime int64
Msg []byte
}
2021-05-26 19:17:51 +08:00
type UserChat struct {
UID string
2021-06-28 15:32:26 +08:00
Msg []MsgInfo
2021-05-26 19:17:51 +08:00
}
2021-12-31 10:40:32 +08:00
type GroupMember_x struct {
2021-07-15 11:14:07 +08:00
GroupID string
UIDList []string
}
2022-01-20 11:36:43 +08:00
func (d *DataBases) GetMinSeqFromMongo(uid string) (MinSeq uint32, err error) {
2022-02-21 15:01:06 +08:00
return 1, nil
//var i, NB uint32
//var seqUid string
//session := d.mgoSession.Clone()
//if session == nil {
// return MinSeq, errors.New("session == nil")
//}
//defer session.Close()
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//MaxSeq, err := d.GetUserMaxSeq(uid)
//if err != nil && err != redis.ErrNil {
// return MinSeq, err
//}
//NB = uint32(MaxSeq / singleGocMsgNum)
//for i = 0; i <= NB; i++ {
// seqUid = indexGen(uid, i)
// n, err := c.Find(bson.M{"uid": seqUid}).Count()
// if err == nil && n != 0 {
// if i == 0 {
// MinSeq = 1
// } else {
// MinSeq = uint32(i * singleGocMsgNum)
// }
// break
// }
//}
//return MinSeq, nil
}
2022-02-23 12:00:02 +08:00
func (d *DataBases) GetMinSeqFromMongo2(uid string) (MinSeq uint32, err error) {
return 1, nil
}
2022-05-04 18:02:24 +08:00
// deleteMsgByLogic
2022-05-25 15:56:39 +08:00
func (d *DataBases) DelMsgBySeqList(userID string, seqList []uint32, operationID string) (totalUnexistSeqList []uint32, err error) {
2022-05-04 18:35:13 +08:00
log.Debug(operationID, utils.GetSelfFuncName(), "args ", userID, seqList)
2022-05-04 18:02:24 +08:00
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)
2022-05-25 15:56:39 +08:00
lock := sync.Mutex{}
2022-05-04 18:02:24 +08:00
var wg sync.WaitGroup
wg.Add(len(suffixUserID2SubSeqList))
for k, v := range suffixUserID2SubSeqList {
go func(suffixUserID string, subSeqList []uint32, operationID string) {
2022-05-25 15:56:39 +08:00
defer wg.Done()
unexistSeqList, err := d.DelMsgBySeqListInOneDoc(suffixUserID, subSeqList, operationID)
if err != nil {
2022-05-25 16:16:23 +08:00
log.Error(operationID, "DelMsgBySeqListInOneDoc failed ", err.Error(), suffixUserID, subSeqList)
2022-05-25 15:56:39 +08:00
return
2022-05-04 18:02:24 +08:00
}
2022-05-25 15:56:39 +08:00
lock.Lock()
totalUnexistSeqList = append(totalUnexistSeqList, unexistSeqList...)
lock.Unlock()
2022-05-04 18:02:24 +08:00
}(k, v, operationID)
}
2022-05-25 15:56:39 +08:00
return totalUnexistSeqList, err
2022-05-04 18:02:24 +08:00
}
2022-05-25 15:56:39 +08:00
func (d *DataBases) DelMsgBySeqListInOneDoc(suffixUserID string, seqList []uint32, operationID string) ([]uint32, error) {
2022-05-04 18:35:13 +08:00
log.Debug(operationID, utils.GetSelfFuncName(), "args ", suffixUserID, seqList)
2022-05-25 15:56:39 +08:00
seqMsgList, indexList, unexistSeqList, err := d.GetMsgAndIndexBySeqListInOneMongo2(suffixUserID, seqList, operationID)
2022-05-04 18:02:24 +08:00
if err != nil {
2022-05-25 15:56:39 +08:00
return nil, utils.Wrap(err, "")
2022-05-04 18:02:24 +08:00
}
for i, v := range seqMsgList {
if err := d.ReplaceMsgByIndex(suffixUserID, v, operationID, indexList[i]); err != nil {
2022-05-25 15:56:39 +08:00
return nil, utils.Wrap(err, "")
2022-05-04 18:02:24 +08:00
}
}
2022-05-25 15:56:39 +08:00
return unexistSeqList, nil
2022-05-04 18:02:24 +08:00
}
2022-03-17 19:00:05 +08:00
// deleteMsgByLogic
func (d *DataBases) DelMsgLogic(uid string, seqList []uint32, operationID string) error {
sortkeys.Uint32s(seqList)
seqMsgs, err := d.GetMsgBySeqListMongo2(uid, seqList, operationID)
if err != nil {
2022-03-21 16:56:52 +08:00
return utils.Wrap(err, "")
2022-03-17 19:00:05 +08:00
}
for _, seqMsg := range seqMsgs {
2022-03-21 16:56:52 +08:00
log.NewDebug(operationID, utils.GetSelfFuncName(), *seqMsg)
2022-03-17 19:00:05 +08:00
seqMsg.Status = constant.MsgDeleted
if err = d.ReplaceMsgBySeq(uid, seqMsg, operationID); err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "ReplaceMsgListBySeq error", err.Error())
}
}
return nil
}
2022-05-04 18:02:24 +08:00
func (d *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
}
2022-03-17 19:00:05 +08:00
func (d *DataBases) ReplaceMsgBySeq(uid string, msg *open_im_sdk.MsgData, operationID string) error {
2022-03-21 16:56:52 +08:00
log.NewInfo(operationID, utils.GetSelfFuncName(), uid, *msg)
2022-03-17 19:00:05 +08:00
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)
2022-03-21 16:56:52 +08:00
bytes, err := proto.Marshal(msg)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "proto marshal", err.Error())
return utils.Wrap(err, "")
}
2022-05-04 18:02:24 +08:00
2022-03-17 19:00:05 +08:00
updateResult, err := c.UpdateOne(
ctx, bson.M{"uid": uid},
2022-03-21 16:56:52 +08:00
bson.M{"$set": bson.M{s: bytes}})
2022-03-17 19:00:05 +08:00
log.NewInfo(operationID, utils.GetSelfFuncName(), updateResult)
if err != nil {
log.NewError(operationID, utils.GetSelfFuncName(), "UpdateOne", err.Error())
2022-03-21 16:56:52 +08:00
return utils.Wrap(err, "")
2022-03-17 19:00:05 +08:00
}
return nil
}
2022-02-21 15:01:06 +08:00
func (d *DataBases) GetMsgBySeqList(uid string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
2022-02-23 12:00:02 +08:00
log.NewInfo(operationID, utils.GetSelfFuncName(), uid, seqList)
2022-02-21 15:01:06 +08:00
var hasSeqList []uint32
singleCount := 0
2021-11-10 18:13:04 +08:00
session := d.mgoSession.Clone()
if session == nil {
2022-02-21 15:01:06 +08:00
return nil, errors.New("session == nil")
2021-11-10 18:13:04 +08:00
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
2022-02-21 15:01:06 +08:00
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])
2021-11-10 18:13:04 +08:00
} else {
2022-02-21 15:01:06 +08:00
t[seqUid] = append(value, seqList[i])
2021-11-10 18:13:04 +08:00
}
}
2022-02-21 15:01:06 +08:00
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...)
2021-11-10 18:13:04 +08:00
}
2022-02-21 15:01:06 +08:00
return seqMsg, nil
2021-11-10 18:13:04 +08:00
}
2022-02-21 15:01:06 +08:00
func (d *DataBases) GetMsgBySeqListMongo2(uid string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, err error) {
2022-01-20 11:36:43 +08:00
var hasSeqList []uint32
singleCount := 0
2022-02-21 15:01:06 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
2022-01-20 11:36:43 +08:00
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 {
2022-01-20 11:36:43 +08:00
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 {
2022-02-21 15:01:06 +08:00
if err = c.FindOne(ctx, bson.M{"uid": seqUid}).Decode(&sChat); err != nil {
2022-01-20 18:36:30 +08:00
log.NewError(operationID, "not find seqUid", seqUid, value, uid, seqList, err.Error())
continue
}
singleCount = 0
for i := 0; i < len(sChat.Msg); i++ {
2021-12-23 17:34:32 +08:00
msg := new(open_im_sdk.MsgData)
if err = proto.Unmarshal(sChat.Msg[i].Msg, msg); err != nil {
2022-01-20 18:36:30 +08:00
log.NewError(operationID, "Unmarshal err", seqUid, value, uid, seqList, err.Error())
2022-01-20 11:36:43 +08:00
return nil, err
}
2022-01-20 11:36:43 +08:00
if isContainInt32(msg.Seq, value) {
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(value) {
break
}
}
}
}
2022-01-20 11:36:43 +08:00
if len(hasSeqList) != len(seqList) {
var diff []uint32
diff = utils.Difference(hasSeqList, seqList)
exceptionMSg := genExceptionMessageBySeqList(diff)
seqMsg = append(seqMsg, exceptionMSg...)
}
return seqMsg, nil
}
2022-05-30 11:56:32 +08:00
func (d *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
}
2022-02-21 15:01:06 +08:00
2022-05-25 15:56:39 +08:00
func (d *DataBases) GetMsgAndIndexBySeqListInOneMongo2(suffixUserID string, seqList []uint32, operationID string) (seqMsg []*open_im_sdk.MsgData, indexList []int, unexistSeqList []uint32, err error) {
2022-05-04 18:02:24 +08:00
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())
2022-05-25 15:56:39 +08:00
return nil, nil, nil, utils.Wrap(err, "")
2022-05-04 18:02:24 +08:00
}
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())
2022-05-25 15:56:39 +08:00
return nil, nil, nil, err
2022-05-04 18:02:24 +08:00
}
if isContainInt32(msg.Seq, seqList) {
indexList = append(indexList, i)
seqMsg = append(seqMsg, msg)
hasSeqList = append(hasSeqList, msg.Seq)
singleCount++
if singleCount == len(seqList) {
break
}
}
}
2022-05-25 15:56:39 +08:00
for _, i := range seqList {
if isContainInt32(i, hasSeqList) {
continue
}
unexistSeqList = append(unexistSeqList, i)
}
return seqMsg, indexList, unexistSeqList, nil
2022-05-04 18:02:24 +08:00
}
2022-01-20 11:36:43 +08:00
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
}
2022-05-30 11:56:32 +08:00
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
}
2022-02-21 15:01:06 +08:00
func (d *DataBases) SaveUserChatMongo2(uid string, sendTime int64, m *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 := 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 {
2022-03-16 18:02:26 +08:00
return utils.Wrap(err, "")
2022-02-21 15:01:06 +08:00
}
err = c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": sMsg}}).Err()
2022-05-20 11:00:11 +08:00
log.NewWarn(operationID, "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
2022-02-21 15:01:06 +08:00
if err != nil {
sChat := UserChat{}
sChat.UID = seqUid
sChat.Msg = append(sChat.Msg, sMsg)
2022-03-16 18:02:26 +08:00
if _, err = c.InsertOne(ctx, &sChat); err != nil {
2022-02-21 15:01:06 +08:00
log.NewDebug(operationID, "InsertOne failed", filter)
return utils.Wrap(err, "")
}
2022-03-16 18:02:26 +08:00
} else {
2022-02-21 15:01:06 +08:00
log.NewDebug(operationID, "FindOneAndUpdate ok", filter)
}
log.NewDebug(operationID, "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
return nil
}
2022-05-20 11:00:11 +08:00
//
//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
//}
2021-12-23 17:34:32 +08:00
func (d *DataBases) SaveUserChat(uid string, sendTime int64, m *pbMsg.MsgDataToDB) error {
var seqUid string
newTime := getCurrentTimestampByMill()
2021-06-28 15:32:26 +08:00
session := d.mgoSession.Clone()
2021-05-26 19:17:51 +08:00
if session == nil {
return errors.New("session == nil")
}
defer session.Close()
2021-12-10 10:49:49 +08:00
log.NewDebug("", "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
2021-06-28 15:32:26 +08:00
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
2021-12-23 17:34:32 +08:00
seqUid = getSeqUid(uid, m.MsgData.Seq)
n, err := c.Find(bson.M{"uid": seqUid}).Count()
2021-05-26 19:17:51 +08:00
if err != nil {
return err
}
2021-12-10 10:49:49 +08:00
log.NewDebug("", "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
2021-06-28 15:32:26 +08:00
sMsg := MsgInfo{}
sMsg.SendTime = sendTime
2022-01-20 18:38:20 +08:00
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
2021-06-28 15:32:26 +08:00
return err
}
2021-05-26 19:17:51 +08:00
if n == 0 {
sChat := UserChat{}
sChat.UID = seqUid
2021-06-28 15:32:26 +08:00
sChat.Msg = append(sChat.Msg, sMsg)
2021-05-26 19:17:51 +08:00
err = c.Insert(&sChat)
if err != nil {
return err
}
} else {
err = c.Update(bson.M{"uid": seqUid}, bson.M{"$push": bson.M{"msg": sMsg}})
2021-05-26 19:17:51 +08:00
if err != nil {
return err
}
}
2021-12-10 10:49:49 +08:00
log.NewDebug("", "insert mgo data cost time", getCurrentTimestampByMill()-newTime)
2021-05-26 19:17:51 +08:00
return nil
}
2021-06-28 15:32:26 +08:00
func (d *DataBases) DelUserChat(uid string) error {
2022-02-21 15:01:06 +08:00
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)
//
//delTime := time.Now().Unix() - int64(config.Config.Mongo.DBRetainChatRecords)*24*3600
//if err := c.Update(bson.M{"uid": uid}, bson.M{"$pull": bson.M{"msg": bson.M{"sendtime": bson.M{"$lte": delTime}}}}); err != nil {
// return err
//}
//
//return nil
}
2021-05-26 19:17:51 +08:00
2022-02-21 15:01:06 +08:00
func (d *DataBases) DelUserChatMongo2(uid string) error {
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}
2021-05-26 19:17:51 +08:00
2021-06-28 15:32:26 +08:00
delTime := time.Now().Unix() - int64(config.Config.Mongo.DBRetainChatRecords)*24*3600
2022-02-21 15:01:06 +08:00
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, "")
2021-05-26 19:17:51 +08:00
}
return nil
}
2022-02-21 15:01:06 +08:00
func (d *DataBases) MgoUserCount() (int, error) {
return 0, nil
//session := d.mgoSession.Clone()
//if session == nil {
// return 0, errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
//
//return c.Find(nil).Count()
2021-06-28 15:32:26 +08:00
}
2021-05-26 19:17:51 +08:00
2021-06-28 15:32:26 +08:00
func (d *DataBases) MgoSkipUID(count int) (string, error) {
2022-02-21 15:01:06 +08:00
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
2021-05-26 19:17:51 +08:00
}
2021-07-15 11:14:07 +08:00
func (d *DataBases) GetGroupMember(groupID string) []string {
2022-02-21 15:01:06 +08:00
return nil
//groupInfo := GroupMember_x{}
//groupInfo.GroupID = groupID
//groupInfo.UIDList = make([]string, 0)
//
//session := d.mgoSession.Clone()
//if session == nil {
// return groupInfo.UIDList
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
//
//if err := c.Find(bson.M{"groupid": groupInfo.GroupID}).One(&groupInfo); err != nil {
// return groupInfo.UIDList
//}
//
//return groupInfo.UIDList
2021-07-15 11:14:07 +08:00
}
func (d *DataBases) AddGroupMember(groupID, uid string) error {
return nil
2022-02-21 15:01:06 +08:00
//session := d.mgoSession.Clone()
//if session == nil {
// return errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
//
//n, err := c.Find(bson.M{"groupid": groupID}).Count()
//if err != nil {
// return err
//}
//
//if n == 0 {
// groupInfo := GroupMember_x{}
// groupInfo.GroupID = groupID
// groupInfo.UIDList = append(groupInfo.UIDList, uid)
// err = c.Insert(&groupInfo)
// if err != nil {
// return err
// }
//} else {
// err = c.Update(bson.M{"groupid": groupID}, bson.M{"$addToSet": bson.M{"uidlist": uid}})
// if err != nil {
// return err
// }
//}
//
//return nil
2021-07-15 11:14:07 +08:00
}
func (d *DataBases) DelGroupMember(groupID, uid string) error {
return nil
2022-02-21 15:01:06 +08:00
//session := d.mgoSession.Clone()
//if session == nil {
// return errors.New("session == nil")
//}
//defer session.Close()
//
//c := session.DB(config.Config.Mongo.DBDatabase).C(cGroup)
//
//if err := c.Update(bson.M{"groupid": groupID}, bson.M{"$pull": bson.M{"uidlist": uid}}); err != nil {
// return err
//}
//
//return nil
2021-07-15 11:14:07 +08:00
}
2022-03-25 18:46:27 +08:00
type Tag struct {
2022-03-28 19:05:58 +08:00
UserID string `bson:"user_id"`
TagID string `bson:"tag_id"`
TagName string `bson:"tag_name"`
UserList []string `bson:"user_list"`
2022-03-25 18:46:27 +08:00
}
func (d *DataBases) GetUserTags(userID string) ([]Tag, error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cTag)
var tags []Tag
2022-03-28 19:05:58 +08:00
cursor, err := c.Find(ctx, bson.M{"user_id": userID})
2022-03-28 12:17:07 +08:00
if err != nil {
return tags, err
}
2022-03-28 16:20:20 +08:00
if err = cursor.All(ctx, &tags); err != nil {
2022-03-28 12:17:07 +08:00
return tags, err
2022-03-25 18:46:27 +08:00
}
return tags, nil
}
func (d *DataBases) CreateTag(userID, tagName string, userList []string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cTag)
tagID := generateTagID(tagName, userID)
tag := Tag{
2022-03-28 12:17:07 +08:00
UserID: userID,
2022-03-25 18:46:27 +08:00
TagID: tagID,
TagName: tagName,
UserList: userList,
}
2022-03-28 12:17:07 +08:00
_, err := c.InsertOne(ctx, tag)
2022-03-25 18:46:27 +08:00
return err
}
2022-03-31 14:50:02 +08:00
func (d *DataBases) GetTagByID(userID, tagID string) (Tag, error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cTag)
var tag Tag
err := c.FindOne(ctx, bson.M{"user_id": userID, "tag_id": tagID}).Decode(&tag)
return tag, err
}
2022-03-25 18:46:27 +08:00
func (d *DataBases) DeleteTag(userID, tagID string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cTag)
2022-03-28 19:05:58 +08:00
_, err := c.DeleteOne(ctx, bson.M{"user_id": userID, "tag_id": tagID})
2022-03-25 18:46:27 +08:00
return err
}
2022-03-28 12:17:07 +08:00
func (d *DataBases) SetTag(userID, tagID, newName string, increaseUserIDList []string, reduceUserIDList []string) error {
2022-03-25 18:46:27 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cTag)
2022-03-28 12:17:07 +08:00
var tag Tag
2022-03-28 19:05:58 +08:00
if err := c.FindOne(ctx, bson.M{"tag_id": tagID, "user_id": userID}).Decode(&tag); err != nil {
2022-03-28 16:20:20 +08:00
return err
}
2022-03-28 12:17:07 +08:00
if newName != "" {
2022-03-28 19:05:58 +08:00
_, err := c.UpdateOne(ctx, bson.M{"user_id": userID, "tag_id": tagID}, bson.M{"$set": bson.M{"tag_name": newName}})
2022-03-28 12:17:07 +08:00
if err != nil {
return err
}
2022-03-25 18:46:27 +08:00
}
2022-03-28 12:17:07 +08:00
tag.UserList = append(tag.UserList, increaseUserIDList...)
2022-03-31 11:43:14 +08:00
tag.UserList = utils.RemoveRepeatedStringInList(tag.UserList)
2022-03-28 12:17:07 +08:00
for _, v := range reduceUserIDList {
for i2, v2 := range tag.UserList {
if v == v2 {
2022-03-28 16:20:20 +08:00
tag.UserList[i2] = ""
2022-03-28 12:17:07 +08:00
}
}
}
2022-03-28 16:20:20 +08:00
var newUserList []string
for _, v := range tag.UserList {
if v != "" {
newUserList = append(newUserList, v)
}
}
2022-03-28 19:05:58 +08:00
_, err := c.UpdateOne(ctx, bson.M{"user_id": userID, "tag_id": tagID}, bson.M{"$set": bson.M{"user_list": newUserList}})
2022-03-25 18:46:27 +08:00
if err != nil {
return err
}
return nil
}
func (d *DataBases) GetUserIDListByTagID(userID, tagID string) ([]string, error) {
2022-03-28 12:17:07 +08:00
var tag Tag
2022-03-25 18:46:27 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cTag)
2022-03-28 19:05:58 +08:00
_ = c.FindOne(ctx, bson.M{"user_id": userID, "tag_id": tagID}).Decode(&tag)
2022-03-28 12:17:07 +08:00
return tag.UserList, nil
2022-03-25 18:46:27 +08:00
}
2022-03-28 18:13:01 +08:00
type TagUser struct {
2022-03-28 19:05:58 +08:00
UserID string `bson:"user_id"`
UserName string `bson:"user_name"`
2022-03-28 18:13:01 +08:00
}
2022-03-25 18:46:27 +08:00
type TagSendLog struct {
2022-03-31 19:49:04 +08:00
UserList []TagUser `bson:"tag_list"`
2022-03-31 16:50:46 +08:00
SendID string `bson:"send_id"`
SenderPlatformID int32 `bson:"sender_platform_id"`
Content string `bson:"content"`
SendTime int64 `bson:"send_time"`
2022-03-25 18:46:27 +08:00
}
2022-03-31 16:50:46 +08:00
func (d *DataBases) SaveTagSendLog(tagSendLog *TagSendLog) error {
2022-03-25 18:46:27 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
2022-03-29 15:35:03 +08:00
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSendLog)
2022-03-25 18:46:27 +08:00
_, err := c.InsertOne(ctx, tagSendLog)
return err
}
2022-03-28 16:20:20 +08:00
func (d *DataBases) GetTagSendLogs(userID string, showNumber, pageNumber int32) ([]TagSendLog, error) {
var tagSendLogs []TagSendLog
2022-03-25 18:46:27 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSendLog)
2022-03-28 19:15:59 +08:00
findOpts := options.Find().SetLimit(int64(showNumber)).SetSkip(int64(showNumber) * (int64(pageNumber) - 1)).SetSort(bson.M{"send_time": -1})
2022-03-28 19:05:58 +08:00
cursor, err := c.Find(ctx, bson.M{"send_id": userID}, findOpts)
2022-03-25 18:46:27 +08:00
if err != nil {
return tagSendLogs, err
}
2022-03-28 17:10:45 +08:00
err = cursor.All(ctx, &tagSendLogs)
2022-03-25 18:46:27 +08:00
if err != nil {
return tagSendLogs, err
}
return tagSendLogs, nil
}
2022-04-15 12:03:23 +08:00
type WorkMoment struct {
2022-05-06 15:58:50 +08:00
WorkMomentID string `bson:"work_moment_id"`
UserID string `bson:"user_id"`
UserName string `bson:"user_name"`
FaceURL string `bson:"face_url"`
Content string `bson:"content"`
LikeUserList []*WorkMomentUser `bson:"like_user_list"`
AtUserList []*WorkMomentUser `bson:"at_user_list"`
PermissionUserList []*WorkMomentUser `bson:"permission_user_list"`
Comments []*Comment `bson:"comments"`
PermissionUserIDList []string `bson:"permission_user_id_list"`
Permission int32 `bson:"permission"`
CreateTime int32 `bson:"create_time"`
}
type WorkMomentUser struct {
2022-04-18 19:24:36 +08:00
UserID string `bson:"user_id"`
UserName string `bson:"user_name"`
2022-04-15 17:43:04 +08:00
}
type Comment struct {
2022-04-18 19:24:36 +08:00
UserID string `bson:"user_id" json:"user_id"`
2022-04-19 14:33:48 +08:00
UserName string `bson:"user_name" json:"user_name"`
2022-04-18 19:24:36 +08:00
ReplyUserID string `bson:"reply_user_id" json:"reply_user_id"`
ReplyUserName string `bson:"reply_user_name" json:"reply_user_name"`
ContentID string `bson:"content_id" json:"content_id"`
Content string `bson:"content" json:"content"`
CreateTime int32 `bson:"create_time" json:"create_time"`
2022-04-15 12:03:23 +08:00
}
2022-04-15 12:10:33 +08:00
func (d *DataBases) CreateOneWorkMoment(workMoment *WorkMoment) error {
2022-04-18 19:24:36 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
workMomentID := generateWorkMomentID(workMoment.UserID)
workMoment.WorkMomentID = workMomentID
workMoment.CreateTime = int32(time.Now().Unix())
_, err := c.InsertOne(ctx, workMoment)
return err
2022-04-15 12:03:23 +08:00
}
2022-04-15 17:43:04 +08:00
func (d *DataBases) DeleteOneWorkMoment(workMomentID string) error {
2022-04-18 19:24:36 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
_, err := c.DeleteOne(ctx, bson.M{"work_moment_id": workMomentID})
return err
2022-04-15 17:43:04 +08:00
}
2022-05-05 11:58:48 +08:00
func (d *DataBases) DeleteComment(workMomentID, contentID, opUserID string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
_, err := c.UpdateOne(ctx, bson.D{{"work_moment_id", workMomentID},
{"$or", bson.A{
bson.D{{"user_id", opUserID}},
bson.D{{"comments", bson.M{"$elemMatch": bson.M{"user_id": opUserID}}}},
},
}}, bson.M{"$pull": bson.M{"comments": bson.M{"content_id": contentID}}})
return err
}
2022-04-15 17:43:04 +08:00
func (d *DataBases) GetWorkMomentByID(workMomentID string) (*WorkMoment, error) {
2022-04-18 19:24:36 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
workMoment := &WorkMoment{}
err := c.FindOne(ctx, bson.M{"work_moment_id": workMomentID}).Decode(workMoment)
return workMoment, err
2022-04-15 17:43:04 +08:00
}
2022-04-22 16:41:41 +08:00
func (d *DataBases) LikeOneWorkMoment(likeUserID, userName, workMomentID string) (*WorkMoment, bool, error) {
2022-04-18 19:24:36 +08:00
workMoment, err := d.GetWorkMomentByID(workMomentID)
if err != nil {
2022-04-22 16:41:41 +08:00
return nil, false, err
2022-04-18 19:24:36 +08:00
}
2022-04-19 13:05:54 +08:00
var isAlreadyLike bool
2022-04-20 16:25:33 +08:00
for i, user := range workMoment.LikeUserList {
2022-04-18 19:24:36 +08:00
if likeUserID == user.UserID {
2022-04-19 13:05:54 +08:00
isAlreadyLike = true
2022-04-20 16:25:33 +08:00
workMoment.LikeUserList = append(workMoment.LikeUserList[0:i], workMoment.LikeUserList[i+1:]...)
2022-04-18 19:24:36 +08:00
}
}
2022-04-19 13:05:54 +08:00
if !isAlreadyLike {
2022-05-06 15:58:50 +08:00
workMoment.LikeUserList = append(workMoment.LikeUserList, &WorkMomentUser{UserID: likeUserID, UserName: userName})
2022-04-19 13:05:54 +08:00
}
2022-04-20 16:25:33 +08:00
log.NewDebug("", utils.GetSelfFuncName(), workMoment)
2022-04-18 19:24:36 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
2022-04-21 10:43:31 +08:00
_, err = c.UpdateOne(ctx, bson.M{"work_moment_id": workMomentID}, bson.M{"$set": bson.M{"like_user_list": workMoment.LikeUserList}})
2022-04-22 16:41:41 +08:00
return workMoment, !isAlreadyLike, err
2022-04-15 17:43:04 +08:00
}
func (d *DataBases) SetUserWorkMomentsLevel(userID string, level int32) error {
return nil
}
2022-04-20 16:25:33 +08:00
func (d *DataBases) CommentOneWorkMoment(comment *Comment, workMomentID string) (WorkMoment, error) {
2022-04-18 19:24:36 +08:00
comment.ContentID = generateWorkMomentCommentID(workMomentID)
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
var workMoment WorkMoment
err := c.FindOneAndUpdate(ctx, bson.M{"work_moment_id": workMomentID}, bson.M{"$push": bson.M{"comments": comment}}).Decode(&workMoment)
return workMoment, err
2022-04-15 17:43:04 +08:00
}
2022-04-24 17:42:41 +08:00
func (d *DataBases) GetUserSelfWorkMoments(userID string, showNumber, pageNumber int32) ([]WorkMoment, error) {
2022-04-18 19:24:36 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
var workMomentList []WorkMoment
findOpts := options.Find().SetLimit(int64(showNumber)).SetSkip(int64(showNumber) * (int64(pageNumber) - 1)).SetSort(bson.M{"create_time": -1})
result, err := c.Find(ctx, bson.M{"user_id": userID}, findOpts)
if err != nil {
return workMomentList, nil
}
err = result.All(ctx, &workMomentList)
return workMomentList, err
2022-04-15 17:43:04 +08:00
}
2022-04-24 17:42:41 +08:00
func (d *DataBases) GetUserWorkMoments(opUserID, userID string, showNumber, pageNumber int32) ([]WorkMoment, error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
var workMomentList []WorkMoment
findOpts := options.Find().SetLimit(int64(showNumber)).SetSkip(int64(showNumber) * (int64(pageNumber) - 1)).SetSort(bson.M{"create_time": -1})
result, err := c.Find(ctx, bson.D{ // 等价条件: select * from
{"user_id", userID},
{"$or", bson.A{
2022-07-29 14:36:07 +08:00
bson.D{{"permission", constant.WorkMomentPermissionCantSee}, {"permission_user_id_list", bson.D{{"$nin", bson.A{opUserID}}}}},
bson.D{{"permission", constant.WorkMomentPermissionCanSee}, {"permission_user_id_list", bson.D{{"$in", bson.A{opUserID}}}}},
2022-04-24 17:42:41 +08:00
bson.D{{"permission", constant.WorkMomentPublic}},
}},
}, findOpts)
if err != nil {
return workMomentList, nil
}
err = result.All(ctx, &workMomentList)
return workMomentList, err
}
2022-04-28 19:05:04 +08:00
func (d *DataBases) GetUserFriendWorkMoments(showNumber, pageNumber int32, userID string) ([]WorkMoment, error) {
2022-04-18 19:24:36 +08:00
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cWorkMoment)
var workMomentList []WorkMoment
findOpts := options.Find().SetLimit(int64(showNumber)).SetSkip(int64(showNumber) * (int64(pageNumber) - 1)).SetSort(bson.M{"create_time": -1})
2022-04-24 18:49:03 +08:00
result, err := c.Find(ctx, bson.D{
2022-04-24 17:42:41 +08:00
{"$or", bson.A{
2022-04-24 18:49:03 +08:00
bson.D{{"user_id", userID}}, //self
2022-04-28 19:05:04 +08:00
bson.D{
2022-04-24 18:49:03 +08:00
{"$or", bson.A{
bson.D{{"permission", constant.WorkMomentPermissionCantSee}, {"permission_user_id_list", bson.D{{"$nin", bson.A{userID}}}}},
bson.D{{"permission", constant.WorkMomentPermissionCanSee}, {"permission_user_id_list", bson.D{{"$in", bson.A{userID}}}}},
bson.D{{"permission", constant.WorkMomentPublic}},
}}},
},
},
2022-04-24 17:42:41 +08:00
}, findOpts)
2022-04-18 19:24:36 +08:00
if err != nil {
2022-04-22 16:41:41 +08:00
return workMomentList, err
2022-04-18 19:24:36 +08:00
}
err = result.All(ctx, &workMomentList)
return workMomentList, err
2022-04-15 17:43:04 +08:00
}
2022-05-27 18:40:24 +08:00
type SuperGroup struct {
2022-05-30 11:55:27 +08:00
GroupID string `bson:"group_id"`
//MemberNumCount int `bson:"member_num_count"`
MemberIDList []string `bson:"member_id_list"`
}
type UserToSuperGroup struct {
UserID string `bson:"user_id"`
GroupIDList []string `bson:"group_id_list"`
2022-05-27 18:40:24 +08:00
}
func (d *DataBases) CreateSuperGroup(groupID string, initMemberIDList []string, memberNumCount int) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSuperGroup)
2022-05-30 15:10:26 +08:00
session, err := d.mongoClient.StartSession()
if err != nil {
return utils.Wrap(err, "start session failed")
}
defer session.EndSession(ctx)
sCtx := mongo.NewSessionContext(ctx, session)
2022-05-27 18:40:24 +08:00
superGroup := SuperGroup{
2022-05-30 11:55:27 +08:00
GroupID: groupID,
MemberIDList: initMemberIDList,
2022-05-27 18:40:24 +08:00
}
2022-05-30 15:10:26 +08:00
_, err = c.InsertOne(sCtx, superGroup)
if err != nil {
2022-07-29 15:01:06 +08:00
_ = session.AbortTransaction(ctx)
2022-05-30 15:10:26 +08:00
return utils.Wrap(err, "transaction failed")
}
var users []UserToSuperGroup
for _, v := range initMemberIDList {
users = append(users, UserToSuperGroup{
UserID: v,
})
}
2022-05-30 15:45:19 +08:00
upsert := true
opts := &options.UpdateOptions{
Upsert: &upsert,
}
2022-05-30 16:29:37 +08:00
c = d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cUserToSuperGroup)
2022-06-21 17:37:22 +08:00
//_, err = c.UpdateMany(sCtx, bson.M{"user_id": bson.M{"$in": initMemberIDList}}, bson.M{"$addToSet": bson.M{"group_id_list": groupID}}, opts)
//if err != nil {
// session.AbortTransaction(ctx)
// return utils.Wrap(err, "transaction failed")
//}
for _, userID := range initMemberIDList {
_, err = c.UpdateOne(sCtx, bson.M{"user_id": userID}, bson.M{"$addToSet": bson.M{"group_id_list": groupID}}, opts)
if err != nil {
2022-07-29 15:01:06 +08:00
_ = session.AbortTransaction(ctx)
2022-06-21 17:37:22 +08:00
return utils.Wrap(err, "transaction failed")
}
2022-05-30 15:10:26 +08:00
}
2022-05-27 18:40:24 +08:00
return err
}
2022-05-27 16:13:44 +08:00
2022-05-27 18:40:24 +08:00
func (d *DataBases) GetSuperGroup(groupID string) (SuperGroup, error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSuperGroup)
superGroup := SuperGroup{}
err := c.FindOne(ctx, bson.M{"group_id": groupID}).Decode(&superGroup)
return superGroup, err
}
func (d *DataBases) AddUserToSuperGroup(groupID string, userIDList []string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSuperGroup)
2022-05-30 11:55:27 +08:00
session, err := d.mongoClient.StartSession()
if err != nil {
return utils.Wrap(err, "start session failed")
}
defer session.EndSession(ctx)
sCtx := mongo.NewSessionContext(ctx, session)
if err != nil {
return utils.Wrap(err, "start transaction failed")
}
_, err = c.UpdateOne(sCtx, bson.M{"group_id": groupID}, bson.M{"$addToSet": bson.M{"member_id_list": bson.M{"$each": userIDList}}})
if err != nil {
2022-07-29 15:01:06 +08:00
_ = session.AbortTransaction(ctx)
2022-05-30 11:55:27 +08:00
return utils.Wrap(err, "transaction failed")
}
c = d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cUserToSuperGroup)
var users []UserToSuperGroup
for _, v := range userIDList {
users = append(users, UserToSuperGroup{
UserID: v,
})
}
2022-05-30 16:23:15 +08:00
upsert := true
opts := &options.UpdateOptions{
Upsert: &upsert,
}
2022-05-30 17:12:08 +08:00
for _, userID := range userIDList {
2022-05-30 17:30:55 +08:00
_, err = c.UpdateOne(sCtx, bson.M{"user_id": userID}, bson.M{"$addToSet": bson.M{"group_id_list": groupID}}, opts)
2022-05-30 17:12:08 +08:00
if err != nil {
2022-07-29 15:01:06 +08:00
_ = session.AbortTransaction(ctx)
2022-05-30 17:12:08 +08:00
return utils.Wrap(err, "transaction failed")
}
2022-05-30 11:55:27 +08:00
}
2022-07-29 15:01:06 +08:00
_ = session.CommitTransaction(ctx)
2022-05-27 18:40:24 +08:00
return err
}
func (d *DataBases) RemoverUserFromSuperGroup(groupID string, userIDList []string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSuperGroup)
2022-05-30 11:55:27 +08:00
session, err := d.mongoClient.StartSession()
if err != nil {
return utils.Wrap(err, "start session failed")
}
defer session.EndSession(ctx)
sCtx := mongo.NewSessionContext(ctx, session)
_, err = c.UpdateOne(ctx, bson.M{"group_id": groupID}, bson.M{"$pull": bson.M{"member_id_list": bson.M{"$in": userIDList}}})
2022-05-30 15:10:26 +08:00
if err != nil {
2022-07-29 15:01:06 +08:00
_ = session.AbortTransaction(ctx)
2022-05-30 15:10:26 +08:00
return utils.Wrap(err, "transaction failed")
}
2022-05-30 11:55:27 +08:00
err = d.RemoveGroupFromUser(ctx, sCtx, groupID, userIDList)
if err != nil {
2022-07-29 15:01:06 +08:00
_ = session.AbortTransaction(ctx)
2022-05-30 11:55:27 +08:00
return utils.Wrap(err, "transaction failed")
}
2022-07-29 15:01:06 +08:00
_ = session.CommitTransaction(ctx)
2022-05-27 18:40:24 +08:00
return err
}
2022-05-30 11:55:27 +08:00
func (d *DataBases) GetSuperGroupByUserID(userID string) (UserToSuperGroup, error) {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cUserToSuperGroup)
var user UserToSuperGroup
return user, c.FindOne(ctx, bson.M{"user_id": userID}).Decode(&user)
}
2022-05-27 18:40:24 +08:00
func (d *DataBases) DeleteSuperGroup(groupID string) error {
ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cSuperGroup)
2022-05-30 11:55:27 +08:00
session, err := d.mongoClient.StartSession()
if err != nil {
return utils.Wrap(err, "start session failed")
}
defer session.EndSession(ctx)
sCtx := mongo.NewSessionContext(ctx, session)
superGroup := &SuperGroup{}
result := c.FindOneAndDelete(sCtx, bson.M{"group_id": groupID})
err = result.Decode(superGroup)
if err != nil {
session.AbortTransaction(ctx)
return utils.Wrap(err, "transaction failed")
}
if err = d.RemoveGroupFromUser(ctx, sCtx, groupID, superGroup.MemberIDList); err != nil {
session.AbortTransaction(ctx)
return utils.Wrap(err, "transaction failed")
}
session.CommitTransaction(ctx)
return nil
}
2022-06-08 16:46:21 +08:00
func (d *DataBases) RemoveGroupFromUser(ctx, sCtx context.Context, groupID string, userIDList []string) error {
2022-05-30 11:55:27 +08:00
var users []UserToSuperGroup
2022-06-08 16:46:21 +08:00
for _, v := range userIDList {
2022-05-30 11:55:27 +08:00
users = append(users, UserToSuperGroup{
UserID: v,
})
}
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cUserToSuperGroup)
2022-06-08 16:46:21 +08:00
_, err := c.UpdateOne(sCtx, bson.M{"user_id": bson.M{"$in": userIDList}}, bson.M{"$pull": bson.M{"group_id_list": groupID}})
2022-05-30 11:55:27 +08:00
if err != nil {
return utils.Wrap(err, "UpdateOne transaction failed")
}
2022-05-27 18:40:24 +08:00
return err
2022-05-27 16:13:44 +08:00
}
2022-03-25 18:46:27 +08:00
func generateTagID(tagName, userID string) string {
2022-03-29 15:50:20 +08:00
return utils.Md5(tagName + userID + strconv.Itoa(rand.Int()) + time.Now().String())
2022-03-25 18:46:27 +08:00
}
2022-04-15 17:43:04 +08:00
func generateWorkMomentID(userID string) string {
return utils.Md5(userID + strconv.Itoa(rand.Int()) + time.Now().String())
}
2022-04-18 19:24:36 +08:00
func generateWorkMomentCommentID(workMomentID string) string {
return utils.Md5(workMomentID + strconv.Itoa(rand.Int()) + time.Now().String())
}
func getCurrentTimestampByMill() int64 {
return time.Now().UnixNano() / 1e6
}
2022-05-22 15:09:21 +08:00
func GetCurrentTimestampByMill() int64 {
return time.Now().UnixNano() / 1e6
}
2022-03-25 18:46:27 +08:00
2022-01-20 11:36:43 +08:00
func getSeqUid(uid string, seq uint32) string {
seqSuffix := seq / singleGocMsgNum
2021-11-10 18:13:04 +08:00
return indexGen(uid, seqSuffix)
}
2022-05-31 19:16:29 +08:00
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
}
2022-05-28 18:10:08 +08:00
func getSeqSuperGroupID(groupID string, seq uint32) string {
seqSuffix := seq / singleGocMsgNum
return superGroupIndexGen(groupID, seqSuffix)
}
2022-03-17 19:00:05 +08:00
2022-05-24 21:34:02 +08:00
func GetSeqUid(uid string, seq uint32) string {
return getSeqUid(uid, seq)
}
2022-03-17 19:00:05 +08:00
func getMsgIndex(seq uint32) int {
seqSuffix := seq / singleGocMsgNum
var index uint32
if seqSuffix == 0 {
2022-05-04 18:02:24 +08:00
index = (seq - seqSuffix*singleGocMsgNum) - 1
2022-03-17 19:00:05 +08:00
} else {
index = seq - seqSuffix*singleGocMsgNum
}
return int(index)
}
2022-01-20 11:36:43 +08:00
func isContainInt32(target uint32, List []uint32) bool {
for _, element := range List {
if target == element {
return true
}
}
return false
2022-05-25 15:56:39 +08:00
}
2022-05-25 15:56:39 +08:00
func isNotContainInt32(target uint32, List []uint32) bool {
for _, i := range List {
if i == target {
return false
}
}
return true
}
2022-05-25 15:56:39 +08:00
2022-01-20 11:36:43 +08:00
func indexGen(uid string, seqSuffix uint32) string {
return uid + ":" + strconv.FormatInt(int64(seqSuffix), 10)
2021-11-10 18:13:04 +08:00
}
2022-05-28 18:10:08 +08:00
func superGroupIndexGen(groupID string, seqSuffix uint32) string {
return "super_group_" + groupID + ":" + strconv.FormatInt(int64(seqSuffix), 10)
}
2022-05-31 19:16:29 +08:00
2022-06-01 15:53:14 +08:00
func (d *DataBases) CleanUpUserMsgFromMongo(userID string, operationID string) error {
2022-05-31 19:16:29 +08:00
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))
2022-06-01 15:53:14 +08:00
log.Error(operationID, "getSeqUserIDList", seqUsers)
2022-05-31 19:16:29 +08:00
_, err = c.DeleteMany(ctx, bson.M{"uid": bson.M{"$in": seqUsers}})
if err == mongo.ErrNoDocuments {
return nil
}
return utils.Wrap(err, "")
}