feat: optimize corn tasks (#2237)

* fix: GroupApplicationAcceptedNotification

* fix: GroupApplicationAcceptedNotification

* fix: NotificationUserInfoUpdate

* cicd: robot automated Change

* fix: component

* fix: getConversationInfo

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

* feat: cron task

---------

Co-authored-by: withchao <withchao@users.noreply.github.com>

pkg/authverify/token.go

@@ -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/cmd/cron_task.go

@@ -34,11 +34,8 @@ func NewCronTaskCmd() *CronTaskCmd {
 	ret := &CronTaskCmd{cronTaskConfig: &cronTaskConfig}
 	ret.configMap = map[string]any{
 		OpenIMCronTaskCfgFileName: &cronTaskConfig.CronTask,
-		RedisConfigFileName:       &cronTaskConfig.RedisConfig,
-		MongodbConfigFileName:     &cronTaskConfig.MongodbConfig,
 		ZookeeperConfigFileName:   &cronTaskConfig.ZookeeperConfig,
 		ShareFileName:             &cronTaskConfig.Share,
-		KafkaConfigFileName:       &cronTaskConfig.KafkaConfig,
 	}
 	ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap))
 	ret.ctx = context.WithValue(context.Background(), "version", config.Version)

pkg/common/cmd/msg_utils.go

@@ -15,16 +15,11 @@
 package cmd
 
 import (
-	"context"
-
-	"github.com/openimsdk/open-im-server/v3/internal/tools"
-	"github.com/openimsdk/tools/system/program"
 	"github.com/spf13/cobra"
 )
 
 type MsgUtilsCmd struct {
 	cobra.Command
-	MsgTool *tools.MsgTool
 }
 
 func (m *MsgUtilsCmd) AddUserIDFlag() {
@@ -146,27 +141,7 @@ func NewSeqCmd() *SeqCmd {
 
 func (s *SeqCmd) GetSeqCmd() *cobra.Command {
 	s.Command.Run = func(cmdLines *cobra.Command, args []string) {
-		_, err := tools.InitMsgTool(context.Background(), nil)
-		if err != nil {
-			program.ExitWithError(err)
-		}
-		userID := s.getUserIDFlag(cmdLines)
-		superGroupID := s.getSuperGroupIDFlag(cmdLines)
-		// beginSeq := s.getBeginSeqFlag(cmdLines)
-		// limit := s.getLimitFlag(cmdLines)
-		if userID != "" {
-			// seq, err := msgTool.s(context.Background(), userID)
-			if err != nil {
-				panic(err)
-			}
-			// println(seq)
-		} else if superGroupID != "" {
-			// seq, err := msgTool.GetSuperGroupSeq(context.Background(), superGroupID)
-			if err != nil {
-				panic(err)
-			}
-			// println(seq)
-		}
+
 	}
 	return &s.Command
 }

pkg/common/config/config.go

@@ -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

@@ -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

@@ -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

@@ -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/table/relation/msg.go

@@ -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

@@ -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

@@ -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