open-im-server/internal/crontask/clear_msg.go

package cronTask

import (
	"Open_IM/pkg/common/config"
	"Open_IM/pkg/common/constant"
	"Open_IM/pkg/common/db"
	"Open_IM/pkg/common/db/cache"
	"Open_IM/pkg/common/db/controller"
	"Open_IM/pkg/common/db/mongo"
	"Open_IM/pkg/common/log"
	"Open_IM/pkg/common/tracelog"
	sdkws "Open_IM/pkg/proto/sdkws"
	"Open_IM/pkg/utils"
	"context"
	"math"
	"strconv"
	"strings"

	"github.com/go-redis/redis/v8"
	"github.com/golang/protobuf/proto"
)

type SeqCheckInterface interface {
	ClearAll() error
}

type ClearMsgCronTask struct {
	msgModel   controller.MsgInterface
	userModel  controller.UserInterface
	groupModel controller.GroupInterface
	cache      cache.Cache
}

func (c *ClearMsgCronTask) getCronTaskOperationID() string {
	return cronTaskOperationID + utils.OperationIDGenerator()
}

func (c *ClearMsgCronTask) ClearAll() {
	operationID := c.getCronTaskOperationID()
	ctx := context.Background()
	tracelog.SetOperationID(ctx, operationID)
	log.NewInfo(operationID, "========================= start del cron task =========================")
	var err error
	userIDList, err := c.userModel.GetAllUserID(ctx)
	if err == nil {
		c.StartClearMsg(operationID, userIDList)
	} else {
		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
	}
	// working group msg clear
	workingGroupIDList, err := im_mysql_model.GetGroupIDListByGroupType(constant.WorkingGroup)
	if err == nil {
		c.StartClearWorkingGroupMsg(operationID, workingGroupIDList)
	} else {
		log.NewError(operationID, utils.GetSelfFuncName(), err.Error())
	}

	log.NewInfo(operationID, "========================= start del cron finished =========================")
}

func (c *ClearMsgCronTask) StartClearMsg(operationID string, userIDList []string) {
	log.NewDebug(operationID, utils.GetSelfFuncName(), "userIDList: ", userIDList)
	for _, userID := range userIDList {
		if err := DeleteUserMsgsAndSetMinSeq(operationID, userID); err != nil {
			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), userID)
		}
		if err := checkMaxSeqWithMongo(operationID, userID, constant.WriteDiffusion); err != nil {
			log.NewError(operationID, utils.GetSelfFuncName(), userID, err)
		}
	}
}

func (c *ClearMsgCronTask) StartClearWorkingGroupMsg(operationID string, workingGroupIDList []string) {
	log.NewDebug(operationID, utils.GetSelfFuncName(), "workingGroupIDList: ", workingGroupIDList)
	for _, groupID := range workingGroupIDList {
		userIDList, err := rocksCache.GetGroupMemberIDListFromCache(groupID)
		if err != nil {
			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID)
			continue
		}
		log.NewDebug(operationID, utils.GetSelfFuncName(), "groupID:", groupID, "workingGroupIDList:", userIDList)
		if err := DeleteUserSuperGroupMsgsAndSetMinSeq(operationID, groupID, userIDList); err != nil {
			log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userIDList)
		}
		if err := checkMaxSeqWithMongo(operationID, groupID, constant.ReadDiffusion); err != nil {
			log.NewError(operationID, utils.GetSelfFuncName(), groupID, err)
		}
	}
}

func checkMaxSeqWithMongo(operationID, sourceID string, diffusionType int) error {
	var seqRedis uint64
	var err error
	if diffusionType == constant.WriteDiffusion {
		seqRedis, err = db.DB.GetUserMaxSeq(sourceID)
	} else {
		seqRedis, err = db.DB.GetGroupMaxSeq(sourceID)
	}
	if err != nil {
		if err == goRedis.Nil {
			return nil
		}
		return utils.Wrap(err, "GetUserMaxSeq failed")
	}
	msg, err := db.DB.GetNewestMsg(sourceID)
	if err != nil {
		return utils.Wrap(err, "GetNewestMsg failed")
	}
	if msg == nil {
		return nil
	}
	if math.Abs(float64(msg.Seq-uint32(seqRedis))) > 10 {
		log.NewWarn(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, sourceID, "redis maxSeq is different with msg.Seq > 10", "status: ", msg.Status, msg.SendTime)
	} else {
		log.NewInfo(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, sourceID, "seq and msg OK", "status:", msg.Status, msg.SendTime)
	}
	return nil
}