package cronTask import ( "Open_IM/pkg/common/config" "Open_IM/pkg/common/constant" "Open_IM/pkg/common/db" "Open_IM/pkg/common/log" server_api_params "Open_IM/pkg/proto/sdk_ws" "Open_IM/pkg/utils" goRedis "github.com/go-redis/redis/v8" "github.com/golang/protobuf/proto" "math" "strconv" "strings" ) const oldestList = 0 const newestList = -1 func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) error { var delStruct delMsgRecursionStruct minSeq, err := deleteMongoMsg(operationID, groupID, oldestList, &delStruct) if err != nil { log.NewError(operationID, utils.GetSelfFuncName(), groupID, "deleteMongoMsg failed") } if minSeq == 0 { return nil } log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDList:", delStruct, "minSeq", minSeq) for _, userID := range userIDList { userMinSeq, err := db.DB.GetGroupUserMinSeq(groupID, userID) if err != nil && err != goRedis.Nil { log.NewError(operationID, utils.GetSelfFuncName(), "GetGroupUserMinSeq failed", groupID, userID, err.Error()) continue } if userMinSeq > uint64(minSeq) { err = db.DB.SetGroupUserMinSeq(groupID, userID, userMinSeq) } else { err = db.DB.SetGroupUserMinSeq(groupID, userID, uint64(minSeq)) } if err != nil { log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), groupID, userID, userMinSeq, minSeq) } } return nil } func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error { var delStruct delMsgRecursionStruct minSeq, err := deleteMongoMsg(operationID, userID, oldestList, &delStruct) if err != nil { return utils.Wrap(err, "") } if minSeq == 0 { return nil } log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDStruct: ", delStruct, "minSeq", minSeq) err = db.DB.SetUserMinSeq(userID, minSeq) return utils.Wrap(err, "") } // del list func delMongoMsgsPhysical(uidList []string) error { if len(uidList) > 0 { err := db.DB.DelMongoMsgs(uidList) if err != nil { return utils.Wrap(err, "DelMongoMsgs failed") } } return nil } type delMsgRecursionStruct struct { minSeq uint32 delUidList []string } func (d *delMsgRecursionStruct) getSetMinSeq() uint32 { return d.minSeq } // index 0....19(del) 20...69 // seq 70 // set minSeq 21 // recursion 删除list并且返回设置的最小seq func deleteMongoMsg(operationID string, ID string, index int64, delStruct *delMsgRecursionStruct) (uint32, error) { // find from oldest list msgs, err := db.DB.GetUserMsgListByIndex(ID, index) if err != nil || msgs.UID == "" { if err != nil { if err == db.ErrMsgListNotExist { log.NewInfo(operationID, utils.GetSelfFuncName(), "ID:", ID, "index:", index, err.Error()) } else { log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID) } } // 获取报错,或者获取不到了,物理删除并且返回seq err = delMongoMsgsPhysical(delStruct.delUidList) if err != nil { return 0, err } return delStruct.getSetMinSeq(), nil } log.NewDebug(operationID, "ID:", ID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg)) if len(msgs.Msg) > db.GetSingleGocMsgNum() { log.NewWarn(operationID, utils.GetSelfFuncName(), "msgs too large", len(msgs.Msg), msgs.UID) } var hasMsgDoNotNeedDel bool for i, msg := range msgs.Msg { // 找到列表中不需要删除的消息了, 表示为递归到最后一个块 if utils.GetCurrentTimestampByMill() < msg.SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) { log.NewDebug(operationID, ID, "find uid", msgs.UID) // 删除块失败 递归结束 返回0 hasMsgDoNotNeedDel = true if err := delMongoMsgsPhysical(delStruct.delUidList); err != nil { return 0, err } // unMarshall失败 块删除成功 设置为最小seq msgPb := &server_api_params.MsgData{} if err = proto.Unmarshal(msg.Msg, msgPb); err != nil { return delStruct.getSetMinSeq(), utils.Wrap(err, "") } // 如果不是块中第一个,就把前面比他早插入的全部设置空 seq字段除外。 if i > 0 { delStruct.minSeq, err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1) if err != nil { log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i) return delStruct.getSetMinSeq(), utils.Wrap(err, "") } } // 递归结束 return msgPb.Seq, nil } else { if !msgListIsFull(msgs) { } } } // 该列表中消息全部为老消息并且列表满了, 加入删除列表继续递归 lastMsgPb := &server_api_params.MsgData{} err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, lastMsgPb) if err != nil { log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), len(msgs.Msg)-1, msgs.UID) return 0, utils.Wrap(err, "proto.Unmarshal failed") } delStruct.minSeq = lastMsgPb.Seq if msgListIsFull(msgs) { log.NewDebug(operationID, "msg list is full", msgs.UID) delStruct.delUidList = append(delStruct.delUidList, msgs.UID) } else { // 列表没有满且没有不需要被删除的消息 代表他是最新的消息块 if !hasMsgDoNotNeedDel { delStruct.minSeq, err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, len(msgs.Msg)-1) if err != nil { log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, "Index:", len(msgs.Msg)-1) err = delMongoMsgsPhysical(delStruct.delUidList) if err != nil { return delStruct.getSetMinSeq(), err } return delStruct.getSetMinSeq(), nil } } } log.NewDebug(operationID, ID, "continue to", delStruct) // 继续递归 index+1 seq, err := deleteMongoMsg(operationID, ID, index+1, delStruct) return seq, utils.Wrap(err, "deleteMongoMsg failed") } func msgListIsFull(chat *db.UserChat) bool { index, _ := strconv.Atoi(strings.Split(chat.UID, ":")[1]) if index == 0 { if len(chat.Msg) >= 4999 { return true } } if len(chat.Msg) >= 5000 { return true } return false } func CheckGroupUserMinSeq(operationID, groupID, userID string) error { return nil } func CheckUserMinSeqWithMongo(operationID, userID string) error { return nil } func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error { var seqRedis uint64 var err error if diffusionType == constant.WriteDiffusion { seqRedis, err = db.DB.GetUserMaxSeq(ID) } else { seqRedis, err = db.DB.GetGroupMaxSeq(ID) } if err != nil { if err == goRedis.Nil { return nil } return utils.Wrap(err, "GetUserMaxSeq failed") } msg, err := db.DB.GetNewestMsg(ID) 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, ID, "redis maxSeq is different with msg.Seq > 10", "status: ", msg.Status, msg.SendTime) } else { log.NewInfo(operationID, utils.GetSelfFuncName(), "seqMongo, seqRedis", msg.Seq, seqRedis, ID, "seq and msg OK", "status:", msg.Status, msg.SendTime) } return nil }