diff --git a/internal/cron_task/clear_msg.go b/internal/cron_task/clear_msg.go index 229937304..938a446ec 100644 --- a/internal/cron_task/clear_msg.go +++ b/internal/cron_task/clear_msg.go @@ -20,7 +20,6 @@ func ResetUserGroupMinSeq(operationID, groupID string, userIDList []string) erro minSeq, err := deleteMongoMsg(operationID, groupID, oldestList, &delMsgIDList) if err != nil { log.NewError(operationID, utils.GetSelfFuncName(), groupID, "deleteMongoMsg failed") - return utils.Wrap(err, "") } if minSeq == 0 { return nil @@ -55,10 +54,11 @@ func DeleteMongoMsgAndResetRedisSeq(operationID, userID string) error { } log.NewDebug(operationID, utils.GetSelfFuncName(), "delMsgIDMap: ", delMsgIDList, "minSeq", minSeq) err = db.DB.SetUserMinSeq(userID, minSeq) - return err + return utils.Wrap(err, "") } -func delMongoMsgs(operationID string, delMsgIDList *[][2]interface{}) error { +// del list +func delMongoMsgsPhysical(delMsgIDList *[][2]interface{}) error { if len(*delMsgIDList) > 0 { var IDList []string for _, v := range *delMsgIDList { @@ -72,9 +72,13 @@ func delMongoMsgs(operationID string, delMsgIDList *[][2]interface{}) error { return nil } +// index 0....19(del) 20...69 +// seq 70 +// set minSeq 21 // recursion func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[][2]interface{}) (uint32, error) { - // 从最旧的列表开始找 + // delMsgIDList [[uid:0, minSeq], [uid:1, minSeq]] + // find from oldest list msgs, err := db.DB.GetUserMsgListByIndex(ID, index) if err != nil || msgs.UID == "" { if err != nil { @@ -84,10 +88,12 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[] log.NewError(operationID, utils.GetSelfFuncName(), "GetUserMsgListByIndex failed", err.Error(), index, ID) } } - return getDelMaxSeqByIDList(*delMsgIDList), delMongoMsgs(operationID, delMsgIDList) - } - if index == 0 && msgs == nil { - return 0, nil + // 获取报错,或者获取不到了,物理删除并且返回seq + err = delMongoMsgsPhysical(delMsgIDList) + if err != nil { + return 0, err + } + return getDelMaxSeqByIDList(*delMsgIDList) + 1, nil } log.NewDebug(operationID, "ID:", ID, "index:", index, "uid:", msgs.UID, "len:", len(msgs.Msg)) if len(msgs.Msg) > db.GetSingleGocMsgNum() { @@ -95,29 +101,29 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[] } log.NewDebug(operationID, utils.GetSelfFuncName(), "get msgs: ", msgs.UID) for i, msg := range msgs.Msg { - // 找到列表中不需要删除的消息了 - if utils.GetCurrentTimestampByMill() < msg.SendTime+int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000 { - if err := delMongoMsgs(operationID, delMsgIDList); err != nil { + // 找到列表中不需要删除的消息了, 表示为递归到最后一个块 + if utils.GetCurrentTimestampByMill() < msg.SendTime+(int64(config.Config.Mongo.DBRetainChatRecords)*24*60*60*1000) { + // 删除块失败 递归结束 返回0 + if err := delMongoMsgsPhysical(delMsgIDList); err != nil { return 0, err } - minSeq := getDelMaxSeqByIDList(*delMsgIDList) + // unMarshall失败 块删除成功 返回块的最大seq+1 设置为最小seq + msgPb := &server_api_params.MsgData{} + if err = proto.Unmarshal(msg.Msg, msgPb); err != nil { + return getDelMaxSeqByIDList(*delMsgIDList) + 1, utils.Wrap(err, "") + } + // 如果不是块中第一个,就把前面比他早插入的全部设置空。 if i > 0 { - msgPb := &server_api_params.MsgData{} - err = proto.Unmarshal(msg.Msg, msgPb) + err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1) if err != nil { - log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), ID, index) - } else { - err = db.DB.ReplaceMsgToBlankByIndex(msgs.UID, i-1) - if err != nil { - log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i) - return minSeq, nil - } - minSeq = msgPb.Seq + log.NewError(operationID, utils.GetSelfFuncName(), err.Error(), msgs.UID, i) + return getDelMaxSeqByIDList(*delMsgIDList) + 1, utils.Wrap(err, "") } } - return minSeq, nil + return msgPb.Seq, nil } } + // 该列表中消息全部为老消息, 加入删除列表继续递归 if len(msgs.Msg) > 0 { msgPb := &server_api_params.MsgData{} err = proto.Unmarshal(msgs.Msg[len(msgs.Msg)-1].Msg, msgPb) @@ -127,10 +133,10 @@ func deleteMongoMsg(operationID string, ID string, index int64, delMsgIDList *[] } *delMsgIDList = append(*delMsgIDList, [2]interface{}{msgs.UID, msgPb.Seq}) } - // 没有找到 代表需要全部删除掉 继续递归查找下一个比较旧的列表 - seq, err := deleteMongoMsg(operationID, utils.GetSelfFuncName(), index+1, delMsgIDList) + // 继续递归 index+1 + seq, err := deleteMongoMsg(operationID, ID, index+1, delMsgIDList) if err != nil { - return 0, utils.Wrap(err, "deleteMongoMsg failed") + return seq, utils.Wrap(err, "deleteMongoMsg failed") } return seq, nil } @@ -166,7 +172,7 @@ func checkMaxSeqWithMongo(operationID, ID string, diffusionType int) error { return utils.Wrap(err, "") } if math.Abs(float64(msgPb.Seq-uint32(maxSeq))) > 10 { - log.NewWarn(operationID, utils.GetSelfFuncName(), maxSeq, msgPb.Seq, "redis maxSeq is different with msg.Seq") + log.NewWarn(operationID, utils.GetSelfFuncName(), maxSeq, msgPb.Seq, "redis maxSeq is different with msg.Seq > 10") } else { log.NewInfo(operationID, utils.GetSelfFuncName(), diffusionType, ID, "seq and msg OK", msgPb.Seq, uint32(maxSeq)) } diff --git a/pkg/common/db/mongoModel.go b/pkg/common/db/mongoModel.go index 389a1e0f1..6dd23deb6 100644 --- a/pkg/common/db/mongoModel.go +++ b/pkg/common/db/mongoModel.go @@ -278,7 +278,7 @@ func (d *DataBases) GetUserMsgListByIndex(ID string, index int64) (*UserChat, er return nil, utils.Wrap(err, fmt.Sprintf("cursor is %s", cursor.Current.String())) } if len(msgs) > 0 { - return &msgs[0], err + return &msgs[0], nil } else { return nil, ErrMsgListNotExist }