feat: optimize corn tasks

config/openim-crontask.yml

@@ -1,4 +1,2 @@
-chatRecordsClearTime: "0 2 * * 3"
-msgDestructTime: "0 2 * * *"
+chatRecordsClearTime: "0 2 * * *"
 retainChatRecords: 365
-enableCronLocker: false

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible // indirect
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.64
+	github.com/openimsdk/protocol v0.0.65
 	github.com/openimsdk/tools v0.0.49-alpha.2
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0

go.sum

@@ -281,8 +281,8 @@ github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
 github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs=
 github.com/openimsdk/gomake v0.0.9 h1:ouf25ygN2PMQ68Gfgns/EQRPiLPnp+77SIr68GfE+n4=
 github.com/openimsdk/gomake v0.0.9/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.64 h1:OrjSs4CgKN9VLvJvrAsc37O7Ru0E0VllXZQSmG/ab7U=
-github.com/openimsdk/protocol v0.0.64/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+github.com/openimsdk/protocol v0.0.65 h1:SPT9qyUsFRTTKSKb/FjpS+xr6sxz/Kbnu+su1bxYagc=
+github.com/openimsdk/protocol v0.0.65/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
 github.com/openimsdk/tools v0.0.49-alpha.2 h1:8IfV6o2ySU7C54sh/MG7ctEp1h3lSNe03OCUDWSk5Ws=
 github.com/openimsdk/tools v0.0.49-alpha.2/go.mod h1:P4oGP1Pd+d4ctbLD5U/XQTgl8yu8Hd3skx640Fr69ko=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

internal/rpc/conversation/conversaion.go

@@ -536,3 +536,46 @@ func (c *conversationServer) GetConversationNotReceiveMessageUserIDs(ctx context
 	}
 	return &pbconversation.GetConversationNotReceiveMessageUserIDsResp{UserIDs: userIDs}, nil
 }
+
+func (c *conversationServer) UpdateConversation(ctx context.Context, req *pbconversation.UpdateConversationReq) (*pbconversation.UpdateConversationResp, error) {
+	m := make(map[string]any)
+	if req.RecvMsgOpt != nil {
+		m["recv_msg_opt"] = req.RecvMsgOpt.Value
+	}
+	if req.AttachedInfo != nil {
+		m["attached_info"] = req.AttachedInfo.Value
+	}
+	if req.Ex != nil {
+		m["ex"] = req.Ex.Value
+	}
+	if req.IsPinned != nil {
+		m["is_pinned"] = req.IsPinned.Value
+	}
+	if req.GroupAtType != nil {
+		m["group_at_type"] = req.GroupAtType.Value
+	}
+	if req.MsgDestructTime != nil {
+		m["msg_destruct_time"] = req.MsgDestructTime.Value
+	}
+	if req.IsMsgDestruct != nil {
+		m["is_msg_destruct"] = req.IsMsgDestruct.Value
+	}
+	if req.BurnDuration != nil {
+		m["burn_duration"] = req.BurnDuration.Value
+	}
+	if req.IsPrivateChat != nil {
+		m["is_private_chat"] = req.IsPrivateChat.Value
+	}
+	if req.MinSeq != nil {
+		m["min_seq"] = req.MinSeq.Value
+	}
+	if req.MaxSeq != nil {
+		m["max_seq"] = req.MaxSeq.Value
+	}
+	if len(m) > 0 {
+		if err := c.conversationDatabase.UpdateUsersConversationField(ctx, req.UserIDs, req.ConversationID, m); err != nil {
+			return nil, err
+		}
+	}
+	return &pbconversation.UpdateConversationResp{}, nil
+}

internal/rpc/msg/clear.go

@@ -0,0 +1,77 @@
+package msg
+
+import (
+	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+	"github.com/openimsdk/protocol/conversation"
+	"github.com/openimsdk/protocol/msg"
+	"github.com/openimsdk/protocol/wrapperspb"
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	"strings"
+	"time"
+)
+
+func (m *msgServer) ClearMsg(ctx context.Context, req *msg.ClearMsgReq) (_ *msg.ClearMsgResp, err error) {
+	if err := authverify.CheckAdmin(ctx, m.config.Share.IMAdminUserID); err != nil {
+		return nil, err
+	}
+	if req.Timestamp > time.Now().UnixMilli() {
+		return nil, errs.ErrArgs.WrapMsg("request millisecond timestamp error")
+	}
+	var (
+		docNum int
+		msgNum int
+		start  = time.Now()
+	)
+	clearMsg := func(ctx context.Context) (bool, error) {
+		conversationSeqs := make(map[string]struct{})
+		defer func() {
+			req := &conversation.UpdateConversationReq{
+				MsgDestructTime: wrapperspb.Int64(time.Now().UnixMilli()),
+			}
+			for conversationID := range conversationSeqs {
+				req.ConversationID = conversationID
+				if err := m.Conversation.UpdateConversations(ctx, req); err != nil {
+					log.ZError(ctx, "update conversation max seq failed", err, "conversationID", conversationID, "msgDestructTime", req.MsgDestructTime)
+				}
+			}
+		}()
+		msgs, err := m.MsgDatabase.GetBeforeMsg(ctx, req.Timestamp, 100)
+		if err != nil {
+			return false, err
+		}
+		if len(msgs) == 0 {
+			return false, nil
+		}
+		for _, msg := range msgs {
+			index, err := m.MsgDatabase.DeleteDocMsgBefore(ctx, req.Timestamp, msg)
+			if err != nil {
+				return false, err
+			}
+			if len(index) == 0 {
+				return false, errs.ErrInternalServer.WrapMsg("delete doc msg failed")
+			}
+			docNum++
+			msgNum += len(index)
+			conversationID := msg.DocID[:strings.LastIndex(msg.DocID, ":")]
+			if _, ok := conversationSeqs[conversationID]; !ok {
+				conversationSeqs[conversationID] = struct{}{}
+			}
+		}
+		return true, nil
+	}
+	for {
+		keep, err := clearMsg(ctx)
+		if err != nil {
+			log.ZError(ctx, "clear msg failed", err, "docNum", docNum, "msgNum", msgNum, "cost", time.Since(start))
+			return nil, err
+		}
+		if !keep {
+			log.ZInfo(ctx, "clear msg success", "docNum", docNum, "msgNum", msgNum, "cost", time.Since(start))
+			break
+		}
+		log.ZInfo(ctx, "clearing message", "docNum", docNum, "msgNum", msgNum, "cost", time.Since(start))
+	}
+	return &msg.ClearMsgResp{}, nil
+}