package mongo import ( "context" "errors" "strings" "time" "fwh-toolbox-server/database" "fwh-toolbox-server/domain" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" ) const CollectionName_FaultRecordOperation = "FaultRecordOperation" var ErrFaultRecordOpNotFound = errors.New("fault record operation not found") func faultRecordOps() *mongo.Collection { return database.GetCollection(CollectionName_FaultRecordOperation) } func InitFaultRecordIndexes(ctx context.Context) error { _, err := faultRecordOps().Indexes().CreateMany(ctx, []mongo.IndexModel{ {Keys: bson.D{{Key: "env", Value: 1}, {Key: "gatewaySn", Value: 1}, {Key: "createdAt", Value: -1}}}, {Keys: bson.D{{Key: "status", Value: 1}}}, {Keys: bson.D{{Key: "createdAt", Value: -1}}}, }) return err } func InsertFaultRecordOp(ctx context.Context, op *domain.FaultRecordOperation) error { now := time.Now() if op.ID.IsZero() { op.ID = primitive.NewObjectID() } op.CreatedAt, op.UpdatedAt = now, now if op.StageTimes == nil { op.StageTimes = map[string]time.Time{} } op.StageTimes[string(op.Status)] = now if op.Messages == nil { op.Messages = []domain.FaultRecordOpMessage{} } _, err := faultRecordOps().InsertOne(ctx, op) return err } func GetFaultRecordOp(ctx context.Context, id primitive.ObjectID) (*domain.FaultRecordOperation, error) { var op domain.FaultRecordOperation err := faultRecordOps().FindOne(ctx, bson.M{"_id": id}).Decode(&op) if errors.Is(err, mongo.ErrNoDocuments) { return nil, ErrFaultRecordOpNotFound } if err != nil { return nil, err } return &op, nil } // UpdateFaultRecordOpStatus 更新状态并记录阶段时间;extra 为同时写入的其他字段。 // 只更新尚未终止的操作,返回是否更新成功(已终止时返回 false)。 func UpdateFaultRecordOpStatus(ctx context.Context, id primitive.ObjectID, status domain.FaultRecordOpStatus, extra bson.M) (bool, error) { now := time.Now() set := bson.M{"status": status, "updatedAt": now, "stageTimes." + string(status): now} for k, v := range extra { set[k] = v } res, err := faultRecordOps().UpdateOne(ctx, bson.M{"_id": id, "status": bson.M{"$nin": domain.FaultRecordTerminalStatuses()}}, bson.M{"$set": set}) if err != nil { return false, err } return res.ModifiedCount > 0, nil } // SetFaultRecordOpFields 更新字段,不改变状态 func SetFaultRecordOpFields(ctx context.Context, id primitive.ObjectID, fields bson.M) error { fields["updatedAt"] = time.Now() _, err := faultRecordOps().UpdateOne(ctx, bson.M{"_id": id}, bson.M{"$set": fields}) return err } func AppendFaultRecordOpMessage(ctx context.Context, id primitive.ObjectID, msg domain.FaultRecordOpMessage) error { _, err := faultRecordOps().UpdateOne(ctx, bson.M{"_id": id}, bson.M{"$push": bson.M{"messages": msg}, "$set": bson.M{"updatedAt": time.Now()}}) return err } // RequestFaultRecordOpCancel 标记取消请求,仅对未终止的操作生效 func RequestFaultRecordOpCancel(ctx context.Context, id primitive.ObjectID) (bool, error) { res, err := faultRecordOps().UpdateOne(ctx, bson.M{"_id": id, "status": bson.M{"$nin": domain.FaultRecordTerminalStatuses()}}, bson.M{"$set": bson.M{"cancelRequested": true, "updatedAt": time.Now()}}) if err != nil { return false, err } return res.MatchedCount > 0, nil } // ReopenFaultRecordOpForDownloadRetry 将“上传已完成但下载失败/超时”的操作重新置为 uploaded,供重试下载。 // 条件在同一次更新中校验,避免并发重复触发;返回是否成功重开。 func ReopenFaultRecordOpForDownloadRetry(ctx context.Context, id primitive.ObjectID) (bool, error) { now := time.Now() res, err := faultRecordOps().UpdateOne(ctx, bson.M{ "_id": id, "kind": domain.FaultRecordOpRecordFetch, "status": bson.M{"$in": []domain.FaultRecordOpStatus{domain.FaultRecordStatusDownloadFailed, domain.FaultRecordStatusTimeout}}, "stageTimes.uploaded": bson.M{"$exists": true}, "fileName": bson.M{"$nin": []any{"", nil}}, }, bson.M{ "$set": bson.M{"status": domain.FaultRecordStatusUploaded, "downloadRetryAt": now, "cancelRequested": false, "updatedAt": now}, "$unset": bson.M{"error": "", "timeoutStage": ""}, "$inc": bson.M{"downloadRetryCount": 1}, }) if err != nil { return false, err } return res.ModifiedCount > 0, nil } // FaultRecordOpFilter 操作历史查询条件 type FaultRecordOpFilter struct { Env string GatewaySn string Kind string // 多个用逗号分隔 } // ListFaultRecordOps 按创建时间倒序分页,不返回报文原文 func ListFaultRecordOps(ctx context.Context, f FaultRecordOpFilter, page, size int64) ([]domain.FaultRecordOperation, int64, error) { filter := bson.M{} if f.Env != "" { filter["env"] = f.Env } if f.GatewaySn != "" { filter["gatewaySn"] = f.GatewaySn } if f.Kind != "" { if kinds := strings.Split(f.Kind, ","); len(kinds) > 1 { filter["kind"] = bson.M{"$in": kinds} } else { filter["kind"] = f.Kind } } total, err := faultRecordOps().CountDocuments(ctx, filter) if err != nil { return nil, 0, err } opts := options.Find(). SetSort(bson.D{{Key: "createdAt", Value: -1}}). SetSkip((page - 1) * size). SetLimit(size). SetProjection(bson.M{"messages": 0}) cur, err := faultRecordOps().Find(ctx, filter, opts) if err != nil { return nil, 0, err } var list []domain.FaultRecordOperation if err := cur.All(ctx, &list); err != nil { return nil, 0, err } return list, total, nil } // ListActiveFaultRecordOps 全部未终止的操作,用于启动时恢复 func ListActiveFaultRecordOps(ctx context.Context) ([]domain.FaultRecordOperation, error) { cur, err := faultRecordOps().Find(ctx, bson.M{"status": bson.M{"$nin": domain.FaultRecordTerminalStatuses()}}, options.Find().SetProjection(bson.M{"messages": 0})) if err != nil { return nil, err } var list []domain.FaultRecordOperation err = cur.All(ctx, &list) return list, err }