mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-04 00:55:59 +08:00
fix: reset branch. (#1277)
* fix: to start im or chat, ZooKeeper must be started first. * fix: msg gateway start output err info Signed-off-by: Gordon <1432970085@qq.com> * fix: msg gateway start output err info Signed-off-by: Gordon <1432970085@qq.com> * chore: package path changes Signed-off-by: withchao <993506633@qq.com> * fix: go mod update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * chore: package path changes Signed-off-by: withchao <993506633@qq.com> * chore: package path changes Signed-off-by: withchao <993506633@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: token update Signed-off-by: Gordon <1432970085@qq.com> * fix: get all userID Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: msggateway add online status call Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: log change Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: log change Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * chore: network mode change Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * feat: add api of get server time Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * feat: remove go work sum Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: pull message add isRead field Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: check msg-transfer script Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: script update Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: start don't kill old process Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: check component Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: pull message set isRead only message come from single. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> * fix: multiple gateway kick user each other. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: add ex field to update group info. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * cicd: robot automated Change * refactor: change project module name. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: change project module name. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * refactor: change project module name. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * test: for pressure test. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: for pressure test. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: for pressure test. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: message log. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * fxi: component check output valid info. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fxi: component check output valid info. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * test: send message test log. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * cicd: robot automated Change * cicd: robot automated Change * test: remove info log. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * feat: api of send message add sendTime field. Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> * fix: add callback for update user's info. * cicd: robot automated Change * fix: change callback command name. * cicd: robot automated Change * fix: single chat unread status change. * fix: single chat unread status change. * fix: single chat unread status change. * fix: user status change. * cicd: robot automated Change * fix: user status change. * fix: user status change. * fix: user status change. * cicd: robot automated Change * fix: ws close when user logout. * fix: remove repeat platform on online status. * cicd: robot automated Change * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation . * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. * re: remove router of unsubscribeStatus. * re: remove router of unsubscribeStatus. * re: remove router of unsubscribeStatus. * re: remove router of unsubscribeStatus. * fix: reset branch * fix: not support redis cluster. CROSSSLOT Keys in request don't hash to the same slot * fix: update user.FaceURL do not trigger GroupMemberInfoSetNotification * cicd: robot automated Change * fix: api send messages for notification conversation. * fix: api send messages for notification conversation. --------- Signed-off-by: Gordon <1432970085@qq.com> Signed-off-by: withchao <993506633@qq.com> Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com> Signed-off-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: withchao <993506633@qq.com> Co-authored-by: Xinwei Xiong <3293172751NSS@gmail.com> Co-authored-by: FGadvancer <FGadvancer@users.noreply.github.com> Co-authored-by: withchao <withchao@users.noreply.github.com>
This commit is contained in:
Gordon
@@ -16,7 +16,6 @@ package unrelation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -45,12 +44,27 @@ type Mongo struct {
|
||||
// NewMongo Initialize MongoDB connection.
|
||||
func NewMongo() (*Mongo, error) {
|
||||
specialerror.AddReplace(mongo.ErrNoDocuments, errs.ErrRecordNotFound)
|
||||
// uri := "mongodb://sample.host:27017/?maxPoolSize=20&w=majority"
|
||||
var uri string
|
||||
uri := "mongodb://sample.host:27017/?maxPoolSize=20&w=majority"
|
||||
if config.Config.Mongo.Uri != "" {
|
||||
uri = config.Config.Mongo.Uri
|
||||
} else {
|
||||
uri = defaultMongoUriForNewMongo()
|
||||
mongodbHosts := ""
|
||||
for i, v := range config.Config.Mongo.Address {
|
||||
if i == len(config.Config.Mongo.Address)-1 {
|
||||
mongodbHosts += v
|
||||
} else {
|
||||
mongodbHosts += v + ","
|
||||
}
|
||||
}
|
||||
if config.Config.Mongo.Password != "" && config.Config.Mongo.Username != "" {
|
||||
uri = fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d&authSource=admin",
|
||||
config.Config.Mongo.Username, config.Config.Mongo.Password, mongodbHosts,
|
||||
config.Config.Mongo.Database, config.Config.Mongo.MaxPoolSize)
|
||||
} else {
|
||||
uri = fmt.Sprintf("mongodb://%s/%s/?maxPoolSize=%d&authSource=admin",
|
||||
mongodbHosts, config.Config.Mongo.Database,
|
||||
config.Config.Mongo.MaxPoolSize)
|
||||
}
|
||||
}
|
||||
fmt.Println("mongo:", uri)
|
||||
var mongoClient *mongo.Client
|
||||
@@ -62,41 +76,17 @@ func NewMongo() (*Mongo, error) {
|
||||
if err == nil {
|
||||
return &Mongo{db: mongoClient}, nil
|
||||
}
|
||||
var cmdErr mongo.CommandError
|
||||
if errors.As(err, &cmdErr) {
|
||||
if cmdErr, ok := err.(mongo.CommandError); ok {
|
||||
if cmdErr.Code == 13 || cmdErr.Code == 18 {
|
||||
return nil, err
|
||||
} else {
|
||||
fmt.Printf("Failed to connect to MongoDB: %s\n", err)
|
||||
}
|
||||
fmt.Printf("Failed to connect to MongoDB: %s\n", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
func defaultMongoUriForNewMongo() string {
|
||||
var uri string
|
||||
mongodbHosts := ""
|
||||
for i, v := range config.Config.Mongo.Address {
|
||||
if i == len(config.Config.Mongo.Address)-1 {
|
||||
mongodbHosts += v
|
||||
} else {
|
||||
mongodbHosts += v + ","
|
||||
}
|
||||
}
|
||||
if config.Config.Mongo.Password != "" && config.Config.Mongo.Username != "" {
|
||||
uri = fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d&authSource=admin",
|
||||
config.Config.Mongo.Username, config.Config.Mongo.Password, mongodbHosts,
|
||||
config.Config.Mongo.Database, config.Config.Mongo.MaxPoolSize)
|
||||
} else {
|
||||
uri = fmt.Sprintf("mongodb://%s/%s/?maxPoolSize=%d&authSource=admin",
|
||||
mongodbHosts, config.Config.Mongo.Database,
|
||||
config.Config.Mongo.MaxPoolSize)
|
||||
}
|
||||
|
||||
return uri
|
||||
}
|
||||
|
||||
func (m *Mongo) GetClient() *mongo.Client {
|
||||
return m.db
|
||||
}
|
||||
@@ -116,7 +106,6 @@ func (m *Mongo) CreateSuperGroupIndex() error {
|
||||
if err := m.createMongoIndex(unrelation.CUserToSuperGroup, true, "user_id"); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -150,6 +139,5 @@ func (m *Mongo) createMongoIndex(collection string, isUnique bool, keys ...strin
|
||||
if err != nil {
|
||||
return utils.Wrap(err, result)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -49,7 +49,6 @@ type MsgMongoDriver struct {
|
||||
|
||||
func NewMsgMongoDriver(database *mongo.Database) table.MsgDocModelInterface {
|
||||
collection := database.Collection(table.MsgDocModel{}.TableName())
|
||||
|
||||
return &MsgMongoDriver{MsgCollection: collection}
|
||||
}
|
||||
|
||||
@@ -60,7 +59,6 @@ func (m *MsgMongoDriver) PushMsgsToDoc(ctx context.Context, docID string, msgsTo
|
||||
|
||||
func (m *MsgMongoDriver) Create(ctx context.Context, model *table.MsgDocModel) error {
|
||||
_, err := m.MsgCollection.InsertOne(ctx, model)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -83,7 +81,6 @@ func (m *MsgMongoDriver) UpdateMsg(
|
||||
if err != nil {
|
||||
return nil, utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
@@ -111,7 +108,6 @@ func (m *MsgMongoDriver) PushUnique(
|
||||
if err != nil {
|
||||
return nil, utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
@@ -124,7 +120,6 @@ func (m *MsgMongoDriver) UpdateMsgContent(ctx context.Context, docID string, ind
|
||||
if err != nil {
|
||||
return utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -148,14 +143,12 @@ func (m *MsgMongoDriver) UpdateMsgStatusByIndexInOneDoc(
|
||||
if err != nil {
|
||||
return utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MsgMongoDriver) FindOneByDocID(ctx context.Context, docID string) (*table.MsgDocModel, error) {
|
||||
doc := &table.MsgDocModel{}
|
||||
err := m.MsgCollection.FindOne(ctx, bson.M{"doc_id": docID}).Decode(doc)
|
||||
|
||||
return doc, err
|
||||
}
|
||||
|
||||
@@ -184,7 +177,6 @@ func (m *MsgMongoDriver) GetMsgDocModelByIndex(
|
||||
if len(msgs) > 0 {
|
||||
return &msgs[0], nil
|
||||
}
|
||||
|
||||
return nil, ErrMsgListNotExist
|
||||
}
|
||||
|
||||
@@ -233,7 +225,6 @@ func (m *MsgMongoDriver) DeleteMsgsInOneDocByIndex(ctx context.Context, docID st
|
||||
if err != nil {
|
||||
return utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -242,7 +233,6 @@ func (m *MsgMongoDriver) DeleteDocs(ctx context.Context, docIDs []string) error
|
||||
return nil
|
||||
}
|
||||
_, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": bson.M{"$in": docIDs}})
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -256,7 +246,6 @@ func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(
|
||||
for _, seq := range seqs {
|
||||
indexs = append(indexs, m.model.GetMsgIndex(seq))
|
||||
}
|
||||
//nolint:govet //This is already the officially recommended standard practice.
|
||||
pipeline := mongo.Pipeline{
|
||||
{
|
||||
{"$match", bson.D{
|
||||
@@ -347,7 +336,6 @@ func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(
|
||||
}
|
||||
msgs = append(msgs, msg)
|
||||
}
|
||||
|
||||
return msgs, nil
|
||||
}
|
||||
|
||||
@@ -356,7 +344,6 @@ func (m *MsgMongoDriver) IsExistDocID(ctx context.Context, docID string) (bool,
|
||||
if err != nil {
|
||||
return false, errs.Wrap(err)
|
||||
}
|
||||
|
||||
return count > 0, nil
|
||||
}
|
||||
|
||||
@@ -385,7 +372,6 @@ func (m *MsgMongoDriver) MarkSingleChatMsgsAsRead(
|
||||
updates = append(updates, updateModel)
|
||||
}
|
||||
_, err := m.MsgCollection.BulkWrite(ctx, updates)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -625,39 +611,7 @@ func (m *MsgMongoDriver) RangeUserSendCount(
|
||||
},
|
||||
)
|
||||
}
|
||||
pipeline := buildPiplineForRangeUserSendCount(or, start, end, sort, pageNumber, showNumber)
|
||||
cur, err := m.MsgCollection.Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||||
if err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
defer cur.Close(ctx)
|
||||
var result []Result
|
||||
if err = cur.All(ctx, &result); err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
users = make([]*table.UserCount, len(result[0].Users))
|
||||
for i, r := range result[0].Users {
|
||||
users[i] = &table.UserCount{
|
||||
UserID: r.UserID,
|
||||
Count: r.Count,
|
||||
}
|
||||
}
|
||||
dateCount = make(map[string]int64)
|
||||
for _, r := range result[0].Dates {
|
||||
dateCount[r.Date] = r.Count
|
||||
}
|
||||
|
||||
return result[0].MsgCount, result[0].UserCount, users, dateCount, nil
|
||||
}
|
||||
|
||||
//nolint:funlen // it need to be such long
|
||||
func buildPiplineForRangeUserSendCount(or bson.A, start time.Time,
|
||||
end time.Time, sort int, pageNumber, showNumber int32,
|
||||
) bson.A {
|
||||
return bson.A{
|
||||
pipeline := bson.A{
|
||||
bson.M{
|
||||
"$match": bson.M{
|
||||
"$and": bson.A{
|
||||
@@ -841,6 +795,30 @@ func buildPiplineForRangeUserSendCount(or bson.A, start time.Time,
|
||||
},
|
||||
},
|
||||
}
|
||||
cur, err := m.MsgCollection.Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||||
if err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
defer cur.Close(ctx)
|
||||
var result []Result
|
||||
if err := cur.All(ctx, &result); err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
users = make([]*table.UserCount, len(result[0].Users))
|
||||
for i, r := range result[0].Users {
|
||||
users[i] = &table.UserCount{
|
||||
UserID: r.UserID,
|
||||
Count: r.Count,
|
||||
}
|
||||
}
|
||||
dateCount = make(map[string]int64)
|
||||
for _, r := range result[0].Dates {
|
||||
dateCount[r.Date] = r.Count
|
||||
}
|
||||
return result[0].MsgCount, result[0].UserCount, users, dateCount, nil
|
||||
}
|
||||
|
||||
func (m *MsgMongoDriver) RangeGroupSendCount(
|
||||
@@ -869,39 +847,7 @@ func (m *MsgMongoDriver) RangeGroupSendCount(
|
||||
Count int64 `bson:"count"`
|
||||
} `bson:"dates"`
|
||||
}
|
||||
pipeline := buildPiplineForRangeGroupSendCount(start, end, sort, pageNumber, showNumber)
|
||||
cur, err := m.MsgCollection.Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||||
if err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
defer cur.Close(ctx)
|
||||
var result []Result
|
||||
if err = cur.All(ctx, &result); err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
groups = make([]*table.GroupCount, len(result[0].Groups))
|
||||
for i, r := range result[0].Groups {
|
||||
groups[i] = &table.GroupCount{
|
||||
GroupID: r.GroupID,
|
||||
Count: r.Count,
|
||||
}
|
||||
}
|
||||
dateCount = make(map[string]int64)
|
||||
for _, r := range result[0].Dates {
|
||||
dateCount[r.Date] = r.Count
|
||||
}
|
||||
|
||||
return result[0].MsgCount, result[0].UserCount, groups, dateCount, nil
|
||||
}
|
||||
|
||||
//nolint:funlen //it need to has such length
|
||||
func buildPiplineForRangeGroupSendCount(start time.Time,
|
||||
end time.Time, sort int, pageNumber, showNumber int32,
|
||||
) bson.A {
|
||||
return bson.A{
|
||||
pipeline := bson.A{
|
||||
bson.M{
|
||||
"$match": bson.M{
|
||||
"$and": bson.A{
|
||||
@@ -1098,6 +1044,30 @@ func buildPiplineForRangeGroupSendCount(start time.Time,
|
||||
},
|
||||
},
|
||||
}
|
||||
cur, err := m.MsgCollection.Aggregate(ctx, pipeline, options.Aggregate().SetAllowDiskUse(true))
|
||||
if err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
defer cur.Close(ctx)
|
||||
var result []Result
|
||||
if err := cur.All(ctx, &result); err != nil {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
if len(result) == 0 {
|
||||
return 0, 0, nil, nil, errs.Wrap(err)
|
||||
}
|
||||
groups = make([]*table.GroupCount, len(result[0].Groups))
|
||||
for i, r := range result[0].Groups {
|
||||
groups[i] = &table.GroupCount{
|
||||
GroupID: r.GroupID,
|
||||
Count: r.Count,
|
||||
}
|
||||
}
|
||||
dateCount = make(map[string]int64)
|
||||
for _, r := range result[0].Dates {
|
||||
dateCount[r.Date] = r.Count
|
||||
}
|
||||
return result[0].MsgCount, result[0].UserCount, groups, dateCount, nil
|
||||
}
|
||||
|
||||
func (m *MsgMongoDriver) SearchMessage(ctx context.Context, req *msg.SearchMessageReq) (int32, []*table.MsgInfoModel, error) {
|
||||
@@ -1105,7 +1075,6 @@ func (m *MsgMongoDriver) SearchMessage(ctx context.Context, req *msg.SearchMessa
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
|
||||
return total, msgs, nil
|
||||
}
|
||||
|
||||
@@ -1150,7 +1119,7 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa
|
||||
},
|
||||
},
|
||||
)
|
||||
//nolint:govet //this is already standard
|
||||
|
||||
pipe = mongo.Pipeline{
|
||||
{
|
||||
{"$match", bson.D{
|
||||
@@ -1245,6 +1214,5 @@ func (m *MsgMongoDriver) searchMessage(ctx context.Context, req *msg.SearchMessa
|
||||
} else {
|
||||
msgs = msgs[start:]
|
||||
}
|
||||
|
||||
return n, msgs, nil
|
||||
}
|
||||
|
||||
@@ -31,14 +31,12 @@ func (m *MsgMongoDriver) ConvertMsgsDocLen(ctx context.Context, conversationIDs
|
||||
cursor, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": regex})
|
||||
if err != nil {
|
||||
log.ZError(ctx, "convertAll find msg doc failed", err, "conversationID", conversationID)
|
||||
|
||||
continue
|
||||
}
|
||||
var msgDocs []table.MsgDocModel
|
||||
err = cursor.All(ctx, &msgDocs)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "convertAll cursor all failed", err, "conversationID", conversationID)
|
||||
|
||||
continue
|
||||
}
|
||||
if len(msgDocs) < 1 {
|
||||
@@ -46,45 +44,39 @@ func (m *MsgMongoDriver) ConvertMsgsDocLen(ctx context.Context, conversationIDs
|
||||
}
|
||||
log.ZInfo(ctx, "msg doc convert", "conversationID", conversationID, "len(msgDocs)", len(msgDocs))
|
||||
if len(msgDocs[0].Msg) == int(m.model.GetSingleGocMsgNum5000()) {
|
||||
convertMsgDocs(m, ctx, msgDocs, conversationID, regex)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func convertMsgDocs(m *MsgMongoDriver, ctx context.Context, msgDocs []table.MsgDocModel, conversationID string, regex primitive.Regex) {
|
||||
var err error
|
||||
if _, err = m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": regex}); err != nil {
|
||||
log.ZError(ctx, "convertAll delete many failed", err, "conversationID", conversationID)
|
||||
|
||||
return
|
||||
}
|
||||
var newMsgDocs []interface{}
|
||||
for _, msgDoc := range msgDocs {
|
||||
if int64(len(msgDoc.Msg)) == m.model.GetSingleGocMsgNum() {
|
||||
continue
|
||||
}
|
||||
var index int64
|
||||
for index < int64(len(msgDoc.Msg)) {
|
||||
msg := msgDoc.Msg[index]
|
||||
if msg != nil && msg.Msg != nil {
|
||||
msgDocModel := table.MsgDocModel{DocID: m.model.GetDocID(conversationID, msg.Msg.Seq)}
|
||||
end := index + m.model.GetSingleGocMsgNum()
|
||||
if int(end) >= len(msgDoc.Msg) {
|
||||
msgDocModel.Msg = msgDoc.Msg[index:]
|
||||
} else {
|
||||
msgDocModel.Msg = msgDoc.Msg[index:end]
|
||||
if _, err := m.MsgCollection.DeleteMany(ctx, bson.M{"doc_id": regex}); err != nil {
|
||||
log.ZError(ctx, "convertAll delete many failed", err, "conversationID", conversationID)
|
||||
continue
|
||||
}
|
||||
var newMsgDocs []interface{}
|
||||
for _, msgDoc := range msgDocs {
|
||||
if int64(len(msgDoc.Msg)) == m.model.GetSingleGocMsgNum() {
|
||||
continue
|
||||
}
|
||||
newMsgDocs = append(newMsgDocs, msgDocModel)
|
||||
index = end
|
||||
var index int64
|
||||
for index < int64(len(msgDoc.Msg)) {
|
||||
msg := msgDoc.Msg[index]
|
||||
if msg != nil && msg.Msg != nil {
|
||||
msgDocModel := table.MsgDocModel{DocID: m.model.GetDocID(conversationID, msg.Msg.Seq)}
|
||||
end := index + m.model.GetSingleGocMsgNum()
|
||||
if int(end) >= len(msgDoc.Msg) {
|
||||
msgDocModel.Msg = msgDoc.Msg[index:]
|
||||
} else {
|
||||
msgDocModel.Msg = msgDoc.Msg[index:end]
|
||||
}
|
||||
newMsgDocs = append(newMsgDocs, msgDocModel)
|
||||
index = end
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
_, err = m.MsgCollection.InsertMany(ctx, newMsgDocs)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "convertAll insert many failed", err, "conversationID", conversationID, "len(newMsgDocs)", len(newMsgDocs))
|
||||
} else {
|
||||
break
|
||||
log.ZInfo(ctx, "msg doc convert", "conversationID", conversationID, "len(newMsgDocs)", len(newMsgDocs))
|
||||
}
|
||||
}
|
||||
}
|
||||
_, err = m.MsgCollection.InsertMany(ctx, newMsgDocs)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "convertAll insert many failed", err, "conversationID", conversationID, "len(newMsgDocs)", len(newMsgDocs))
|
||||
} else {
|
||||
log.ZInfo(ctx, "msg doc convert", "conversationID", conversationID, "len(newMsgDocs)", len(newMsgDocs))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -59,7 +59,6 @@ func (s *SuperGroupMongoDriver) CreateSuperGroup(ctx context.Context, groupID st
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -70,7 +69,6 @@ func (s *SuperGroupMongoDriver) TakeSuperGroup(
|
||||
if err := s.superGroupCollection.FindOne(ctx, bson.M{"group_id": groupID}).Decode(&group); err != nil {
|
||||
return nil, utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return group, nil
|
||||
}
|
||||
|
||||
@@ -88,7 +86,6 @@ func (s *SuperGroupMongoDriver) FindSuperGroup(
|
||||
if err := cursor.All(ctx, &groups); err != nil {
|
||||
return nil, utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return groups, nil
|
||||
}
|
||||
|
||||
@@ -116,7 +113,6 @@ func (s *SuperGroupMongoDriver) AddUserToSuperGroup(ctx context.Context, groupID
|
||||
return utils.Wrap(err, "transaction failed")
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -133,7 +129,6 @@ func (s *SuperGroupMongoDriver) RemoverUserFromSuperGroup(ctx context.Context, g
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -143,7 +138,6 @@ func (s *SuperGroupMongoDriver) GetSuperGroupByUserID(
|
||||
) (*unrelation.UserToSuperGroupModel, error) {
|
||||
var user unrelation.UserToSuperGroupModel
|
||||
err := s.userToSuperGroupCollection.FindOne(ctx, bson.M{"user_id": userID}).Decode(&user)
|
||||
|
||||
return &user, utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
@@ -155,7 +149,6 @@ func (s *SuperGroupMongoDriver) DeleteSuperGroup(ctx context.Context, groupID st
|
||||
if _, err := s.superGroupCollection.DeleteOne(ctx, bson.M{"group_id": groupID}); err != nil {
|
||||
return utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
return s.RemoveGroupFromUser(ctx, groupID, group.MemberIDs)
|
||||
}
|
||||
|
||||
@@ -165,6 +158,5 @@ func (s *SuperGroupMongoDriver) RemoveGroupFromUser(ctx context.Context, groupID
|
||||
bson.M{"user_id": bson.M{"$in": userIDs}},
|
||||
bson.M{"$pull": bson.M{"group_id_list": groupID}},
|
||||
)
|
||||
|
||||
return utils.Wrap(err, "")
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ package unrelation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
@@ -51,7 +50,6 @@ type UserMongoDriver struct {
|
||||
// AddSubscriptionList Subscriber's handling of thresholds.
|
||||
func (u *UserMongoDriver) AddSubscriptionList(ctx context.Context, userID string, userIDList []string) error {
|
||||
// Check the number of lists in the key.
|
||||
//nolint:govet //this has already been the standard format for mongo.Pipeline
|
||||
pipeline := mongo.Pipeline{
|
||||
{{"$match", bson.D{{"user_id", SubscriptionPrefix + userID}}}},
|
||||
{{"$project", bson.D{{"count", bson.D{{"$size", "$user_id_list"}}}}}},
|
||||
@@ -67,7 +65,7 @@ func (u *UserMongoDriver) AddSubscriptionList(ctx context.Context, userID string
|
||||
}
|
||||
// iterate over aggregated results
|
||||
for cursor.Next(ctx) {
|
||||
err = cursor.Decode(&cnt)
|
||||
err := cursor.Decode(&cnt)
|
||||
if err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
@@ -124,7 +122,6 @@ func (u *UserMongoDriver) AddSubscriptionList(ctx context.Context, userID string
|
||||
return utils.Wrap(err, "transaction failed")
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -142,7 +139,6 @@ func (u *UserMongoDriver) UnsubscriptionList(ctx context.Context, userID string,
|
||||
if err != nil {
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -156,7 +152,6 @@ func (u *UserMongoDriver) RemoveSubscribedListFromUser(ctx context.Context, user
|
||||
bson.M{"$pull": bson.M{"user_id_list": userID}},
|
||||
)
|
||||
}
|
||||
|
||||
return errs.Wrap(err)
|
||||
}
|
||||
|
||||
@@ -168,13 +163,12 @@ func (u *UserMongoDriver) GetAllSubscribeList(ctx context.Context, userID string
|
||||
bson.M{"user_id": SubscriptionPrefix + userID})
|
||||
err = cursor.Decode(&user)
|
||||
if err != nil {
|
||||
if errors.Is(err, mongo.ErrNoDocuments) {
|
||||
if err == mongo.ErrNoDocuments {
|
||||
return []string{}, nil
|
||||
} else {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
|
||||
return user.UserIDList, nil
|
||||
}
|
||||
|
||||
@@ -186,12 +180,11 @@ func (u *UserMongoDriver) GetSubscribedList(ctx context.Context, userID string)
|
||||
bson.M{"user_id": SubscribedPrefix + userID})
|
||||
err = cursor.Decode(&user)
|
||||
if err != nil {
|
||||
if errors.Is(err, mongo.ErrNoDocuments) {
|
||||
if err == mongo.ErrNoDocuments {
|
||||
return []string{}, nil
|
||||
} else {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
|
||||
return user.UserIDList, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user