IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-16 23:09:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: cron task

Browse Source
This commit is contained in:
withchao
2024-04-25 16:50:24 +08:00
parent 20dfafd218
commit dc57d38856
19 changed files with 471 additions and 617 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/authverify/token.go
-1
View File
@@ -46,7 +46,6 @@ func IsAppManagerUid(ctx context.Context, imAdminUserID []string) bool {
}
func CheckAdmin(ctx context.Context, imAdminUserID []string) error {
if datautil.Contain(mcontext.GetOpUserID(ctx), imAdminUserID...) {
return nil
}
pkg/common/config/config.go
-2
View File
@@ -107,9 +107,7 @@ type API struct {
type CronTask struct {
ChatRecordsClearTime string `mapstructure:"chatRecordsClearTime"`
MsgDestructTime string `mapstructure:"msgDestructTime"`
RetainChatRecords int `mapstructure:"retainChatRecords"`
EnableCronLocker bool `yaml:"enableCronLocker"`
}
type OfflinePushConfig struct {
pkg/common/db/controller/msg.go
+75
View File
@@ -17,6 +17,8 @@ package controller
import (
"context"
"encoding/json"
"errors"
"strings"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
@@ -102,6 +104,10 @@ type CommonMsgDatabase interface {
RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*relation.UserCount, dateCount map[string]int64, err error)
RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*relation.GroupCount, dateCount map[string]int64, err error)
ConvertMsgsDocLen(ctx context.Context, conversationIDs []string)
// clear msg
GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*relation.MsgDocModel, error)
DeleteDocMsgBefore(ctx context.Context, ts int64, doc *relation.MsgDocModel) ([]int, error)
}
func NewCommonMsgDatabase(msgDocModel relation.MsgDocModelInterface, msg cache.MsgCache, seq cache.SeqCache, kafkaConf *config.Kafka) (CommonMsgDatabase, error) {
@@ -1047,3 +1053,72 @@ func (db *commonMsgDatabase) FindOneByDocIDs(ctx context.Context, conversationID
func (db *commonMsgDatabase) ConvertMsgsDocLen(ctx context.Context, conversationIDs []string) {
db.msgDocDatabase.ConvertMsgsDocLen(ctx, conversationIDs)
}
func (db *commonMsgDatabase) GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*relation.MsgDocModel, error) {
return db.msgDocDatabase.GetBeforeMsg(ctx, ts, limit)
}
func (db *commonMsgDatabase) DeleteDocMsgBefore(ctx context.Context, ts int64, doc *relation.MsgDocModel) ([]int, error) {
var notNull int
index := make([]int, 0, len(doc.Msg))
for i, message := range doc.Msg {
if message.Msg != nil {
notNull++
if message.Msg.SendTime < ts {
index = append(index, i)
}
}
}
if len(index) == 0 {
return index, nil
}
maxSeq := doc.Msg[index[len(index)-1]].Msg.Seq
conversationID := doc.DocID[:strings.LastIndex(doc.DocID, ":")]
if err := db.setMinSeq(ctx, conversationID, maxSeq+1); err != nil {
return index, err
}
if len(index) == notNull {
return index, db.msgDocDatabase.DeleteDoc(ctx, doc.DocID)
} else {
return index, db.msgDocDatabase.DeleteMsgByIndex(ctx, doc.DocID, index)
}
}
//func (db *commonMsgDatabase) ClearMsg(ctx context.Context, ts int64) (err error) {
// var (
// docNum int
// msgNum int
// start = time.Now()
// )
// for {
// msgs, err := db.msgDocDatabase.GetBeforeMsg(ctx, ts, 100)
// if err != nil {
// return err
// }
// if len(msgs) == 0 {
// return nil
// }
// for _, msg := range msgs {
// num, err := db.deleteOneMsg(ctx, ts, msg)
// if err != nil {
// return err
// }
// docNum++
// msgNum += num
// }
// }
//}
func (db *commonMsgDatabase) setMinSeq(ctx context.Context, conversationID string, seq int64) error {
dbSeq, err := db.seq.GetMinSeq(ctx, conversationID)
if err != nil {
if errors.Is(errs.Unwrap(err), redis.Nil) {
return nil
}
return err
}
if dbSeq >= seq {
return nil
}
return db.seq.SetMinSeq(ctx, conversationID, seq)
}
pkg/common/db/mgo/conversation.go
+10 -1
View File
@@ -56,7 +56,16 @@ func (c *ConversationMgo) Delete(ctx context.Context, groupIDs []string) (err er
}
func (c *ConversationMgo) UpdateByMap(ctx context.Context, userIDs []string, conversationID string, args map[string]any) (rows int64, err error) {
res, err := mongoutil.UpdateMany(ctx, c.coll, bson.M{"owner_user_id": bson.M{"$in": userIDs}, "conversation_id": conversationID}, bson.M{"$set": args})
if len(args) == 0 {
return 0, nil
}
filter := bson.M{
"conversation_id": conversationID,
}
if len(userIDs) > 0 {
filter["owner_user_id"] = bson.M{"$in": userIDs}
}
res, err := mongoutil.UpdateMany(ctx, c.coll, filter, bson.M{"$set": args})
if err != nil {
return 0, err
}
pkg/common/db/mgo/msg.go
+86
View File
@@ -896,3 +896,89 @@ func (m *MsgMgo) ConvertMsgsDocLen(ctx context.Context, conversationIDs []string
}
}
}
func (m *MsgMgo) GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*relation.MsgDocModel, error) {
return mongoutil.Aggregate[*relation.MsgDocModel](ctx, m.coll, []bson.M{
{
"$match": bson.M{
"msgs.msg.send_time": bson.M{
"$lt": ts,
},
},
},
{
"$project": bson.M{
"_id": 0,
"doc_id": 1,
"msgs.msg.send_time": 1,
"msgs.msg.seq": 1,
},
},
{
"$limit": limit,
},
})
}
func (m *MsgMgo) DeleteMsgByIndex(ctx context.Context, docID string, index []int) error {
if len(index) == 0 {
return nil
}
model := &relation.MsgInfoModel{DelList: []string{}}
set := make(map[string]any)
for i := range index {
set[fmt.Sprintf("msgs.%d", i)] = model
}
return mongoutil.UpdateOne(ctx, m.coll, bson.M{"doc_id": docID}, bson.M{"$set": set}, true)
}
//func (m *MsgMgo) ClearMsg(ctx context.Context, t time.Time) (int64, error) {
// ts := t.UnixMilli()
// var count int64
// for {
// msgs, err := m.GetBeforeMsg(ctx, ts, 100)
// if err != nil {
// return count, err
// }
// if len(msgs) == 0 {
// return count, nil
// }
// for _, msg := range msgs {
// num, err := m.deleteOneMsg(ctx, ts, msg)
// count += num
// if err != nil {
// return count, err
// }
// }
// }
//}
func (m *MsgMgo) DeleteDoc(ctx context.Context, docID string) error {
return mongoutil.DeleteOne(ctx, m.coll, bson.M{"doc_id": docID})
}
//func (m *MsgMgo) DeleteDocMsg(ctx context.Context, ts int64, doc *relation.MsgDocModel) (int64, error) {
// var notNull int
// index := make([]int, 0, len(doc.Msg))
// for i, message := range doc.Msg {
// if message.Msg != nil {
// notNull++
// if message.Msg.SendTime < ts {
// index = append(index, i)
// }
// }
// }
// if len(index) == 0 {
// return 0, errs.New("no msg to delete").WrapMsg("deleteOneMsg", "docID", doc.DocID)
// }
// if len(index) == notNull {
// if err := m.DeleteDoc(ctx, doc.DocID); err != nil {
// return 0, err
// }
// } else {
// if err := m.setNullMsg(ctx, doc.DocID, index); err != nil {
// return 0, err
// }
// }
// return int64(len(index)), nil
//}
pkg/common/db/mgo/msg_del.go
+71
View File
@@ -0,0 +1,71 @@
package mgo
import (
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
"go.mongodb.org/mongo-driver/mongo"
)
func NewMsgDelMongo(db *mongo.Database) *MsgDelMgo {
return &MsgDelMgo{coll: db.Collection(new(relation.MsgDocModel).TableName())}
}
type MsgDelMgo struct {
coll *mongo.Collection
model relation.MsgDocModel
}
//func (m *MsgDelMgo) getEmptyMsg(ctx context.Context, limit int) ([]string, error) {
// return mongoutil.Aggregate[string](ctx, m.coll, []bson.M{
// {
// "$match": bson.M{
// "msgs": bson.M{
// "$exists": true,
// },
// },
// },
// {
// "$project": bson.M{
// "_id": 0,
// "doc_id": 1,
// "all_null_msgs": bson.M{
// "$not": []bson.M{
// {
// "$anyElementTrue": bson.M{
// "$map": bson.M{
// "input": "$msgs",
// "as": "item",
// "in": "$$item.msg",
// },
// },
// },
// },
// },
// },
// },
// {
// "$project": bson.M{
// "doc_id": 1,
// },
// },
// {
// "$limit": limit,
// },
// })
//}
//
//func (m *MsgDelMgo) deleteEmptyMsgs(ctx context.Context) error {
// for {
// docIDs, err := m.getEmptyMsg(ctx, 100)
// if err != nil {
// return err
// }
// if len(docIDs) == 0 {
// return nil
// }
// for _, docID := range docIDs {
// if err := m.deleteEmptyMsg(ctx, docID); err != nil {
// return err
// }
// }
// }
//}
pkg/common/db/mgo/msg_test.go
+46
View File
@@ -0,0 +1,46 @@
package mgo
import (
"strings"
"testing"
)
func TestName(t *testing.T) {
//conf := config.Mongo{
// Address: []string{"localhost:37017"},
// Username: "openIM",
// Password: "openIM123",
// Database: "demo",
//}
//conf.URI = `mongodb://openIM:openIM123@localhost:37017/demo?maxPoolSize=100&authSource=admin`
//cli, err := mongoutil.NewMongoDB(context.Background(), conf.Build())
//if err != nil {
// panic(err)
//}
//msg, _ := NewMsgMongo(cli.GetDB())
//count, err := msg.ClearMsg(context.Background(), time.Now().Add(-time.Hour*24*61))
//if err != nil {
// t.Log("error", err)
// return
//}
//t.Log("count", count)
s := `si_5300327160_9129042887:0123`
t.Log(s[:strings.LastIndex(s, ":")])
}
func TestName2(t *testing.T) {
m := map[string]string{
"1": "1",
"2": "2",
}
t.Log(m)
clear(m)
t.Log(m)
a := []string{"1", "2"}
t.Log(a)
clear(a)
t.Log(a)
}
pkg/common/db/table/relation/msg.go
+6
View File
@@ -116,6 +116,12 @@ type MsgDocModelInterface interface {
RangeUserSendCount(ctx context.Context, start time.Time, end time.Time, group bool, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, users []*UserCount, dateCount map[string]int64, err error)
RangeGroupSendCount(ctx context.Context, start time.Time, end time.Time, ase bool, pageNumber int32, showNumber int32) (msgCount int64, userCount int64, groups []*GroupCount, dateCount map[string]int64, err error)
ConvertMsgsDocLen(ctx context.Context, conversationIDs []string)
DeleteDoc(ctx context.Context, docID string) error
DeleteMsgByIndex(ctx context.Context, docID string, index []int) error
GetBeforeMsg(ctx context.Context, ts int64, limit int) ([]*MsgDocModel, error)
//ClearMsg(ctx context.Context, t time.Time) (int64, error)
}
func (MsgDocModel) TableName() string {
pkg/rpcclient/conversation.go
+5
View File
@@ -82,6 +82,11 @@ func (c *ConversationRpcClient) SetConversations(ctx context.Context, userIDs []
return err
}
func (c *ConversationRpcClient) UpdateConversations(ctx context.Context, conversation *pbconversation.UpdateConversationReq) error {
_, err := c.Client.UpdateConversation(ctx, conversation)
return err
}
func (c *ConversationRpcClient) GetConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
resp, err := c.Client.GetConversationIDs(ctx, &pbconversation.GetConversationIDsReq{UserID: ownerUserID})
if err != nil {
pkg/rpcclient/msg.go
+5
View File
@@ -212,6 +212,11 @@ func (m *MessageRpcClient) GetConversationMaxSeq(ctx context.Context, conversati
return resp.MaxSeq, nil
}
func (m *MessageRpcClient) ClearMsg(ctx context.Context, ts int64) error {
_, err := m.Client.ClearMsg(ctx, &msg.ClearMsgReq{Timestamp: ts})
return err
}
type NotificationSender struct {
contentTypeConf map[int32]config.NotificationConfig
sessionTypeConf map[int32]int32