IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-07 10:35:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

active user

Browse Source
This commit is contained in:
withchao
2023-07-03 20:54:03 +08:00
parent bedd554269
commit dd6b47fbbb
8 changed files with 533 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/db/controller/msg.go
+5
View File
@@ -92,6 +92,7 @@ type CommonMsgDatabase interface {
GetOneMessageAllReactionList(ctx context.Context, clientMsgID string, sessionType int32) (map[string]string, error)
DeleteOneMessageKey(ctx context.Context, clientMsgID string, sessionType int32, subKey string) error
DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensionList map[string]*sdkws.KeyValue) error
RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, err error)
}
func NewCommonMsgDatabase(msgDocModel unRelationTb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase {
@@ -925,3 +926,7 @@ func (db *commonMsgDatabase) GetExtendMsg(ctx context.Context, conversationID st
func (db *commonMsgDatabase) DeleteReactionExtendMsgSet(ctx context.Context, conversationID string, sessionType int32, clientMsgID string, msgFirstModifyTime int64, reactionExtensions map[string]*sdkws.KeyValue) error {
return db.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, conversationID, sessionType, clientMsgID, msgFirstModifyTime, db.extendMsgSetModel.Pb2Model(reactionExtensions))
}
func (db *commonMsgDatabase) RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*unRelationTb.UserCount, err error) {
return db.msgDocDatabase.RangeUserSendCount(ctx, start, end, ase, pageNumber, showNumber)
}
pkg/common/db/table/unrelation/msg.go
+2
View File
@@ -3,6 +3,7 @@ package unrelation
import (
"context"
"strconv"
"time"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
"go.mongodb.org/mongo-driver/mongo"
@@ -87,6 +88,7 @@ type MsgDocModelInterface interface {
GetMsgDocModelByIndex(ctx context.Context, conversationID string, index, sort int64) (*MsgDocModel, error)
DeleteMsgsInOneDocByIndex(ctx context.Context, docID string, indexes []int) error
MarkSingleChatMsgsAsRead(ctx context.Context, userID string, docID string, indexes []int64) error
RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*UserCount, err error)
}
func (MsgDocModel) TableName() string {
pkg/common/db/unrelation/msg.go
+13 -62
View File
@@ -310,60 +310,7 @@ func (m *MsgMongoDriver) MarkSingleChatMsgsAsRead(ctx context.Context, userID st
return err
}
func (m *MsgMongoDriver) RangeCount(ctx context.Context, start time.Time, end time.Time) (int64, error) {
type Total struct {
Total int64 `bson:"total"`
}
pipeline := bson.A{
bson.M{
"$addFields": bson.M{
"msgs": bson.M{
"$filter": bson.M{
"input": "$msgs",
"as": "item",
"cond": bson.M{
"$and": bson.A{
bson.M{
"$gte": bson.A{
"$$item.msg.send_time", start.UnixMilli(),
},
"$lt": bson.A{
"$$item.msg.send_time", end.UnixMilli(),
},
},
},
},
},
},
},
},
bson.M{
"$group": bson.M{
"_id": nil,
"total": bson.M{
"$sum": bson.M{
"$size": "$msgs",
},
},
},
},
}
cur, err := m.MsgCollection.Aggregate(ctx, pipeline)
if err != nil {
return 0, errs.Wrap(err)
}
defer cur.Close(ctx)
var total []Total
if err := cur.All(ctx, &total); err != nil {
return 0, err
}
if len(total) == 0 {
return 0, nil
}
return total[0].Total, nil
}
func (m *MsgMongoDriver) RangeUserSendCount(ctx context.Context, ase bool, start time.Time, end time.Time) (int64, []*table.UserCount, error) {
func (m *MsgMongoDriver) RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*table.UserCount, err error) {
var sort int
if ase {
sort = -1
@@ -371,8 +318,9 @@ func (m *MsgMongoDriver) RangeUserSendCount(ctx context.Context, ase bool, start
sort = 1
}
type Result struct {
Total int64 `bson:"result"`
Result []struct {
MsgCount int64 `bson:"msg_count"`
UserCount int64 `bson:"user_count"`
Result []struct {
UserID string `bson:"_id"`
Count int64 `bson:"count"`
}
@@ -438,16 +386,19 @@ func (m *MsgMongoDriver) RangeUserSendCount(ctx context.Context, ase bool, start
},
bson.M{
"addFields": bson.M{
"result": bson.M{
"user_count": bson.M{
"$size": "$result",
},
"msg_count": bson.M{
"sum": "$result.count",
},
},
},
bson.M{
"$addFields": bson.M{
"result": bson.M{
"$slice": bson.A{
"$result", 0, 10,
"$result", pageNumber - 1, showNumber,
},
},
},
@@ -455,15 +406,15 @@ func (m *MsgMongoDriver) RangeUserSendCount(ctx context.Context, ase bool, start
}
cur, err := m.MsgCollection.Aggregate(ctx, pipeline)
if err != nil {
return 0, nil, errs.Wrap(err)
return 0, 0, nil, errs.Wrap(err)
}
defer cur.Close(ctx)
var result []Result
if err := cur.All(ctx, &result); err != nil {
return 0, nil, err
return 0, 0, nil, err
}
if len(result) == 0 {
return 0, nil, nil
return 0, 0, nil, nil
}
res := make([]*table.UserCount, len(result[0].Result))
for i, r := range result[0].Result {
@@ -472,5 +423,5 @@ func (m *MsgMongoDriver) RangeUserSendCount(ctx context.Context, ase bool, start
Count: r.Count,
}
}
return result[0].Total, res, nil
return result[0].MsgCount, result[0].UserCount, res, nil
}