push

2026-05-19 16:29:01 +08:00 · 2023-02-23 18:20:45 +08:00
parent 4fc00109ef
commit 9c76722c90
8 changed files with 211 additions and 120 deletions
@@ -0,0 +1,170 @@
+package task
+
+import (
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/db/controller"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/tracelog"
+	"Open_IM/pkg/utils"
+	"context"
+	"fmt"
+	"github.com/go-redis/redis/v8"
+	"math"
+)
+
+type msgTool struct {
+	msgInterface   controller.MsgDatabase
+	userInterface  controller.UserDatabase
+	groupInterface controller.GroupDatabase
+}
+
+func (c *msgTool) getCronTaskOperationID() string {
+	return cronTaskOperationID + utils.OperationIDGenerator()
+}
+
+func (c *msgTool) ClearAll() {
+	operationID := c.getCronTaskOperationID()
+	ctx := context.Background()
+	tracelog.SetOperationID(ctx, operationID)
+	log.NewInfo(operationID, "============================ start del cron task ============================")
+	var err error
+	userIDList, err := c.userInterface.GetAllUserID(ctx)
+	if err == nil {
+		c.ClearUsersMsg(ctx, userIDList)
+	} else {
+		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
+	}
+	// working group msg clear
+	workingGroupIDList, err := c.groupInterface.GetGroupIDsByGroupType(ctx, constant.WorkingGroup)
+	if err == nil {
+		c.ClearSuperGroupMsg(ctx, workingGroupIDList)
+	} else {
+		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
+	}
+	log.NewInfo(operationID, "============================ start del cron finished ============================")
+}
+
+func (c *msgTool) ClearUsersMsg(ctx context.Context, userIDList []string) {
+	for _, userID := range userIDList {
+		if err := c.msgInterface.DeleteUserMsgsAndSetMinSeq(ctx, userID, int64(config.Config.Mongo.DBRetainChatRecords*24*60*60)); err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), userID)
+		}
+		_, maxSeqMongo, minSeqCache, maxSeqCache, err := c.msgInterface.GetUserMinMaxSeqInMongoAndCache(ctx, userID)
+		if err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), "GetUserMinMaxSeqInMongoAndCache failed", userID)
+			continue
+		}
+		c.FixUserSeq(ctx, userID, minSeqCache, maxSeqCache)
+		c.CheckMaxSeqWithMongo(ctx, userID, maxSeqCache, maxSeqMongo, constant.WriteDiffusion)
+	}
+}
+
+func (c *msgTool) ClearSuperGroupMsg(ctx context.Context, workingGroupIDList []string) {
+	for _, groupID := range workingGroupIDList {
+		userIDs, err := c.groupInterface.FindGroupMemberUserID(ctx, groupID)
+		if err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), "FindGroupMemberUserID", err.Error(), groupID)
+			continue
+		}
+		if err := c.msgInterface.DeleteUserSuperGroupMsgsAndSetMinSeq(ctx, groupID, userIDs, int64(config.Config.Mongo.DBRetainChatRecords*24*60*60)); err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), "DeleteUserSuperGroupMsgsAndSetMinSeq failed", groupID, userIDs, config.Config.Mongo.DBRetainChatRecords)
+		}
+		_, maxSeqMongo, maxSeqCache, err := c.msgInterface.GetSuperGroupMinMaxSeqInMongoAndCache(ctx, groupID)
+		if err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), utils.GetSelfFuncName(), err.Error(), "GetUserMinMaxSeqInMongoAndCache failed", groupID)
+			continue
+		}
+		for _, userID := range userIDs {
+			minSeqCache, err := c.msgInterface.GetGroupUserMinSeq(ctx, groupID, userID)
+			if err != nil {
+				log.NewError(tracelog.GetOperationID(ctx), "GetGroupUserMinSeq failed", groupID, userID)
+				continue
+			}
+			c.FixGroupUserSeq(ctx, userID, groupID, minSeqCache, maxSeqCache)
+
+		}
+		c.CheckMaxSeqWithMongo(ctx, groupID, maxSeqCache, maxSeqMongo, constant.WriteDiffusion)
+	}
+}
+
+func (c *msgTool) FixUserSeq(ctx context.Context, userID string, minSeqCache, maxSeqCache int64) {
+	if minSeqCache > maxSeqCache {
+		if err := c.msgInterface.SetUserMinSeq(ctx, userID, maxSeqCache); err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), "SetUserMinSeq failed", userID, minSeqCache, maxSeqCache)
+		} else {
+			log.NewWarn(tracelog.GetOperationID(ctx), "SetUserMinSeq success", userID, minSeqCache, maxSeqCache)
+		}
+	}
+}
+
+func (c *msgTool) FixGroupUserSeq(ctx context.Context, userID string, groupID string, minSeqCache, maxSeqCache int64) {
+	if minSeqCache > maxSeqCache {
+		if err := c.msgInterface.SetGroupUserMinSeq(ctx, groupID, userID, maxSeqCache); err != nil {
+			log.NewError(tracelog.GetOperationID(ctx), "SetGroupUserMinSeq failed", userID, minSeqCache, maxSeqCache)
+		} else {
+			log.NewWarn(tracelog.GetOperationID(ctx), "SetGroupUserMinSeq success", userID, minSeqCache, maxSeqCache)
+		}
+	}
+}
+
+func (c *msgTool) CheckMaxSeqWithMongo(ctx context.Context, sourceID string, maxSeqCache, maxSeqMongo int64, diffusionType int) {
+	if math.Abs(float64(maxSeqMongo-maxSeqCache)) > 10 {
+		log.NewWarn(tracelog.GetOperationID(ctx), "cache max seq and mongo max seq is diff > 10", sourceID, maxSeqCache, maxSeqMongo, diffusionType)
+	}
+}
+
+func (c *msgTool) FixAllSeq(ctx context.Context) {
+	userIDs, err := c.userInterface.GetAllUserID(ctx)
+	if err != nil {
+		panic(err.Error())
+	}
+	for _, userID := range userIDs {
+		userCurrentMinSeq, err := c.msgInterface.GetUserMinSeq(ctx, userID)
+		if err != nil && err != redis.Nil {
+			continue
+		}
+		userCurrentMaxSeq, err := c.msgInterface.GetUserMaxSeq(ctx, userID)
+		if err != nil && err != redis.Nil {
+			continue
+		}
+		if userCurrentMinSeq > userCurrentMaxSeq {
+			if err = c.msgInterface.SetUserMinSeq(ctx, userID, userCurrentMaxSeq); err != nil {
+				fmt.Println("SetUserMinSeq failed", userID, userCurrentMaxSeq)
+			}
+			fmt.Println("fix", userID, userCurrentMaxSeq)
+		}
+	}
+	fmt.Println("fix users seq success")
+
+	groupIDs, err := c.groupInterface.GetGroupIDsByGroupType(ctx, constant.WorkingGroup)
+	if err != nil {
+		panic(err.Error())
+	}
+	for _, groupID := range groupIDs {
+		maxSeq, err := c.msgInterface.GetGroupMaxSeq(ctx, groupID)
+		if err != nil {
+			fmt.Println("GetGroupMaxSeq failed", groupID)
+			continue
+		}
+		userIDs, err := c.groupInterface.FindGroupMemberUserID(ctx, groupID)
+		if err != nil {
+			fmt.Println("get groupID", groupID, "failed, try again later")
+			continue
+		}
+		for _, userID := range userIDs {
+			userMinSeq, err := c.msgInterface.GetGroupUserMinSeq(ctx, groupID, userID)
+			if err != nil && err != redis.Nil {
+				fmt.Println("GetGroupUserMinSeq failed", groupID, userID)
+				continue
+			}
+			if userMinSeq > maxSeq {
+				if err = c.msgInterface.SetGroupUserMinSeq(ctx, groupID, userID, maxSeq); err != nil {
+					fmt.Println("SetGroupUserMinSeq failed", err.Error(), groupID, userID, maxSeq)
+				}
+				fmt.Println("fix", groupID, userID, maxSeq, userMinSeq)
+			}
+		}
+	}
+	fmt.Println("fix all seq finished")
+}
@@ -0,0 +1,103 @@
+package task
+
+import (
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/proto/sdkws"
+	"context"
+	"fmt"
+	"strconv"
+
+	"github.com/go-redis/redis/v8"
+	"github.com/golang/protobuf/proto"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+
+	"testing"
+	"time"
+)
+
+var (
+	redisClient *redis.Client
+	mongoClient *mongo.Collection
+)
+
+func GenUserChat(startSeq, stopSeq, delSeq, index uint32, userID string) *mongo2.UserChat {
+	chat := &mongo2.UserChat{UID: userID + strconv.Itoa(int(index))}
+	for i := startSeq; i <= stopSeq; i++ {
+		msg := sdkws.MsgData{
+			SendID:           "sendID1",
+			RecvID:           "recvID1",
+			GroupID:          "",
+			ClientMsgID:      "xxx",
+			ServerMsgID:      "xxx",
+			SenderPlatformID: 1,
+			SenderNickname:   "testNickName",
+			SenderFaceURL:    "testFaceURL",
+			SessionType:      1,
+			MsgFrom:          100,
+			ContentType:      101,
+			Content:          []byte("testFaceURL"),
+			Seq:              uint32(i),
+			SendTime:         time.Now().Unix(),
+			CreateTime:       time.Now().Unix(),
+			Status:           1,
+		}
+		bytes, _ := proto.Marshal(&msg)
+		sendTime := 0
+		chat.Msg = append(chat.Msg, mongo2.MsgInfo{SendTime: int64(sendTime), Msg: bytes})
+	}
+	return chat
+}
+
+func SetUserMaxSeq(userID string, seq int) error {
+	return redisClient.Set(context.Background(), "REDIS_USER_INCR_SEQ"+userID, seq, 0).Err()
+}
+
+func CreateChat(userChat *mongo2.UserChat) error {
+	_, err := mongoClient.InsertOne(context.Background(), userChat)
+	return err
+}
+
+func TestDeleteUserMsgsAndSetMinSeq(t *testing.T) {
+	operationID := getCronTaskOperationID()
+	redisClient = redis.NewClient(&redis.Options{
+		Addr:     "127.0.0.1:16379",
+		Password: "openIM123", // no password set
+		DB:       13,          // use default DB
+	})
+	mongoUri := fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d&authSource=admin",
+		"root", "openIM123", "127.0.0.1:37017",
+		"openIM", 100)
+	client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(mongoUri))
+	mongoClient = client.Database("openIM").Collection("msg")
+	testUID1 := "test_del_id1"
+	//testUID2 := "test_del_id2"
+	//testUID3 := "test_del_id3"
+	//testUID4 := "test_del_id4"
+	//testUID5 := "test_del_id5"
+	//testUID6 := "test_del_id6"
+	err = SetUserMaxSeq(testUID1, 600)
+	userChat := GenUserChat(1, 500, 200, 0, testUID1)
+	err = CreateChat(userChat)
+
+	if err := DeleteUserMsgsAndSetMinSeq(operationID, testUID1); err != nil {
+		t.Error("checkMaxSeqWithMongo failed", testUID1)
+	}
+	if err := checkMaxSeqWithMongo(operationID, testUID1, constant.WriteDiffusion); err != nil {
+		t.Error("checkMaxSeqWithMongo failed", testUID1)
+	}
+	if err != nil {
+		t.Error("err is not nil", testUID1, err.Error())
+	}
+	// testWorkingGroupIDList := []string{"test_del_id1", "test_del_id2", "test_del_id3", "test_del_id4", "test_del_id5"}
+	// for _, groupID := range testWorkingGroupIDList {
+	// 	operationID = groupID + "-" + operationID
+	// 	log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "userIDList:", testUserIDList)
+	// 	if err := DeleteUserSuperGroupMsgsAndSetMinSeq(operationID, groupID, testUserIDList); err != nil {
+	// 		t.Error("checkMaxSeqWithMongo failed", groupID)
+	// 	}
+	// 	if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
+	// 		t.Error("checkMaxSeqWithMongo failed", groupID)
+	// 	}
+	// }
+}
@@ -0,0 +1,56 @@
+package task
+
+import (
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/tracelog"
+	"Open_IM/pkg/utils"
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/robfig/cron/v3"
+)
+
+const cronTaskOperationID = "cronTaskOperationID-"
+const moduleName = "cron"
+
+func StartCronTask() error {
+	log.NewPrivateLog(moduleName)
+	log.NewInfo(utils.OperationIDGenerator(), "start cron task", "cron config", config.Config.Mongo.ChatRecordsClearTime)
+	fmt.Println("cron task start, config", config.Config.Mongo.ChatRecordsClearTime)
+	clearCronTask := msgTool{}
+	ctx := context.Background()
+	operationID := clearCronTask.getCronTaskOperationID()
+	tracelog.SetOperationID(ctx, operationID)
+	c := cron.New()
+	_, err := c.AddFunc(config.Config.Mongo.ChatRecordsClearTime, clearCronTask.ClearAll)
+	if err != nil {
+		fmt.Println("start cron failed", err.Error(), config.Config.Mongo.ChatRecordsClearTime)
+		return err
+	}
+	c.Start()
+	fmt.Println("start cron task success")
+	for {
+		time.Sleep(10 * time.Second)
+	}
+}
+
+func FixSeq(userID, workingGroupID string, fixAllSeq bool) {
+	log.NewPrivateLog(moduleName)
+	log.NewInfo(utils.OperationIDGenerator(), "start cron task", "cron config", config.Config.Mongo.ChatRecordsClearTime)
+	clearCronTask := msgTool{}
+	ctx := context.Background()
+	operationID := clearCronTask.getCronTaskOperationID()
+	tracelog.SetOperationID(ctx, operationID)
+	if userID != "" {
+		clearCronTask.ClearUsersMsg(ctx, []string{userID})
+	}
+	if workingGroupID != "" {
+		clearCronTask.ClearSuperGroupMsg(ctx, []string{workingGroupID})
+	}
+	if fixAllSeq {
+		clearCronTask.FixAllSeq(ctx)
+	}
+	fmt.Println("fix seq finished")
+}