package services import ( "bytes" "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "mime" "path/filepath" "strconv" "strings" "time" "fwh-toolbox-server/cache" "fwh-toolbox-server/domain" "fwh-toolbox-server/repositories" minioRepo "fwh-toolbox-server/repositories/minio" mongoRepo "fwh-toolbox-server/repositories/mongo" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" "go.uber.org/zap" ) type DataDownloadPartialFailure struct { Succeeded int Failed int Cause error } func (e *DataDownloadPartialFailure) Error() string { return fmt.Sprintf("fault-record request partially completed: %d succeeded, %d failed", e.Succeeded, e.Failed) } func (e *DataDownloadPartialFailure) Unwrap() error { return e.Cause } func executeFaultRecordDataDownloadItem(ctx context.Context, item *domain.DataDownloadItem, awaitRequestID string) error { ready, lease, err := prepareDataDownloadRequest(ctx, item, awaitRequestID) if err != nil { return err } if ready != nil { _ = mongoRepo.ReleaseDataDownloadFaultCall(ctx, ready.ID, item.ID) return linkDataDownloadItem(ctx, item, ready, ready.Empty) } defer lease.abort() otherActive, err := mongoRepo.OtherDataDownloadFaultCallActive(ctx, lease.documentID, item.ID) if err != nil { return err } if otherActive { return &DataDownloadFaultCallDeferred{After: time.Minute} } deviceType, err := strconv.Atoi(item.DeviceType) if err != nil { return fmt.Errorf("invalid fault-record device type %q", item.DeviceType) } loc := dataDownloadLocation(item.Timezone) start, err := time.ParseInLocation(time.DateOnly, item.Date, loc) if err != nil { return err } end := start.AddDate(0, 0, 1).Add(-time.Second) sources, sourceErr := resolveFaultRecordSources(ctx, item, lease.documentID, deviceType, start, end) var callDeferred *DataDownloadFaultCallDeferred pendingCall := errors.As(sourceErr, &callDeferred) if errors.Is(sourceErr, context.Canceled) { _ = ReleaseFaultCallOwnership(context.Background(), item) return sourceErr } if sourceErr != nil && len(sources) == 0 { if faultCallTerminal(item, sourceErr) { _ = ReleaseFaultCallOwnership(context.Background(), item) } return sourceErr } fetchedAt := time.Now().UTC() zap.L().Debug("[DataDownload] FaultRecord 已验证录波子记录", zap.String("documentId", lease.documentID), zap.String("deviceSn", item.DeviceSN), zap.String("deviceType", item.DeviceType), zap.String("date", item.Date), zap.Int("sourceCount", len(sources)), ) for _, source := range sources { sourceID := fmt.Sprintf("%d:%d", source.record.Id, source.sub.Id) if attempt, getErr := mongoRepo.GetDataDownloadFileAttempt(ctx, item.TaskID, item.ID, sourceID); getErr == nil && attempt.Status == domain.DataDownloadItemSucceeded { if _, entryErr := mongoRepo.GetDataDownloadFaultRecordEntry(ctx, lease.documentID, sourceID); entryErr == nil { continue } } name, body, downloadErr := repositories.QueryDownloadFaultRecordData(source.record.Id, source.sub.Id) if downloadErr != nil { _ = recordFaultRecordAttempt(ctx, item, lease.documentID, sourceID, name, nil, "download", downloadErr) continue } name = filepath.Base(name) if name == "." || name == "" { name = fmt.Sprintf("fault-%d-%d.bin", source.record.Id, source.sub.Id) } zap.L().Debug("[DataDownload] FaultRecord 文件已下载到内存", zap.String("documentId", lease.documentID), zap.String("sourceId", sourceID), zap.String("fileName", name), zap.Int("bytes", len(body)), ) contentType := mime.TypeByExtension(strings.ToLower(filepath.Ext(name))) if contentType == "" { contentType = "application/octet-stream" } sum := sha256.Sum256(body) objectKey, uploadErr := minioRepo.PutBinaryDocument(ctx, lease.documentID, name, bytes.NewReader(body), int64(len(body)), contentType) if uploadErr != nil { _ = recordFaultRecordAttempt(ctx, item, lease.documentID, sourceID, name, nil, "upload", uploadErr) continue } contentLock, lockErr := cache.Lock(ctx, "data-download:content:"+lease.documentID, 10*time.Minute, 200*time.Millisecond, 30*time.Second) if lockErr != nil { _ = minioRepo.Delete(context.Background(), objectKey) _ = recordFaultRecordAttempt(ctx, item, lease.documentID, sourceID, name, nil, "publish", lockErr) continue } if validateErr := lease.validate(ctx); validateErr != nil { contentLock.SafeRelease(context.Background()) _ = minioRepo.Delete(context.Background(), objectKey) return validateErr } entry := &domain.DataDownloadFaultRecordEntry{ DocumentID: lease.documentID, SourceFileID: sourceID, FaultID: source.record.Id, SubRecordID: source.sub.Id, FileName: name, ContentType: contentType, Size: int64(len(body)), Checksum: hex.EncodeToString(sum[:]), ObjectKey: objectKey, LastSuccessfulAt: fetchedAt, UpdatedAt: fetchedAt, } oldObjectKey, publishErr := mongoRepo.UpsertDataDownloadFaultRecordEntry(ctx, entry) contentLock.SafeRelease(context.Background()) if publishErr != nil { _ = minioRepo.Delete(context.Background(), objectKey) _ = recordFaultRecordAttempt(ctx, item, lease.documentID, sourceID, name, nil, "publish", publishErr) continue } if oldObjectKey != "" && oldObjectKey != objectKey { _ = minioRepo.Delete(context.Background(), oldObjectKey) } if attemptErr := recordFaultRecordAttempt(ctx, item, lease.documentID, sourceID, name, &entry.ID, "", nil); attemptErr != nil { return attemptErr } } if sourceErr != nil && !pendingCall { if err := recordFaultRecordAttempt(ctx, item, lease.documentID, "call:*", "", nil, "call", sourceErr); err != nil { return err } } else { _ = mongoRepo.DeleteDataDownloadFileAttempt(ctx, item.TaskID, item.ID, "call:*") } attempts, err := mongoRepo.ListDataDownloadFileAttempts(ctx, item.TaskID) if err != nil { return err } succeeded, failed := 0, 0 for i := range attempts { if attempts[i].ItemID != item.ID { continue } if attempts[i].Status == domain.DataDownloadItemSucceeded { succeeded++ } else if attempts[i].Status == domain.DataDownloadFailed { failed++ } } entries, err := mongoRepo.ListDataDownloadFaultRecordEntries(ctx, lease.documentID) if err != nil { return err } manifest := strings.Builder{} for i := range entries { manifest.WriteString(entries[i].SourceFileID) manifest.WriteByte('|') manifest.WriteString(entries[i].Checksum) manifest.WriteByte('\n') } manifestSum := sha256.Sum256([]byte(manifest.String())) contentLock, err := cache.Lock(ctx, "data-download:content:"+lease.documentID, 10*time.Minute, 200*time.Millisecond, 30*time.Second) if err != nil { return err } defer contentLock.SafeRelease(context.Background()) if err := lease.validate(ctx); err != nil { return err } generation, createdAt := int64(1), fetchedAt if existing, getErr := mongoRepo.GetDataDownloadDocument(ctx, lease.documentID); getErr == nil { generation = existing.ContentGeneration + 1 createdAt = existing.CreatedAt } else if !errors.Is(getErr, mongo.ErrNoDocuments) { return getErr } completeness := domain.DataDownloadCompletenessComplete if failed > 0 || pendingCall { completeness = domain.DataDownloadCompletenessPartial } state, frozenAt := faultRecordDocumentState(time.Now(), loc, start, failed) if pendingCall { state, frozenAt = domain.DataDownloadMutable, nil } document := &domain.DataDownloadDocument{ ID: lease.documentID, RequestKey: lease.requestKey, Kind: item.Kind, SiteSN: item.SiteSN, DeviceSN: item.DeviceSN, Date: item.Date, DeviceType: item.DeviceType, Timezone: item.Timezone, State: state, Completeness: completeness, Empty: !pendingCall && failed == 0 && len(entries) == 0, FileCount: int64(len(entries)), FailedFileCount: int64(failed), ContentGeneration: generation, Checksum: hex.EncodeToString(manifestSum[:]), SourceFetchedAt: fetchedAt, FrozenAt: frozenAt, CreatedAt: createdAt, UpdatedAt: fetchedAt, } if err := mongoRepo.UpsertDataDownloadDocument(ctx, document); err != nil { return err } zap.L().Debug("[DataDownload] FaultRecord 内容已提交", zap.String("documentId", lease.documentID), zap.String("state", string(state)), zap.String("completeness", string(completeness)), zap.Int("fileCount", len(entries)), zap.Int("succeeded", succeeded), zap.Int("failed", failed), zap.Int64("generation", generation), ) if err := linkDataDownloadItem(ctx, item, document, document.Empty); err != nil { return err } if pendingCall { return callDeferred } if faultCallTerminal(item, sourceErr) { _ = ReleaseFaultCallOwnership(context.Background(), item) } if failed > 0 { return &DataDownloadPartialFailure{Succeeded: succeeded, Failed: failed, Cause: sourceErr} } if err := lease.complete(ctx); err != nil { return err } return mongoRepo.ReleaseDataDownloadFaultCall(ctx, lease.documentID, item.ID) } func faultCallTerminal(item *domain.DataDownloadItem, err error) bool { return item.CallRecordStatus != nil && *item.CallRecordStatus == 2 || errors.Is(err, ErrFaultRecordCallTimeout) || errors.Is(err, ErrFaultRecordCallFailed) || errors.Is(err, errAmbiguousFaultCall) } // ReleaseFaultCallOwnership clears an owner only when no remote call was sent, // the call is terminal, or its two-hour observation window has elapsed. A // cancelled item can otherwise leave a remote call running; its owner expires // automatically and must not be removed early by another queue execution. func ReleaseFaultCallOwnership(ctx context.Context, item *domain.DataDownloadItem) error { if item == nil || item.Kind != domain.DataDownloadFaultRecord { return nil } if item.CallRequestedAt != nil && (item.CallRecordStatus == nil || (*item.CallRecordStatus != 2 && *item.CallRecordStatus != -1)) && time.Now().Before(item.CallRequestedAt.Add(faultCallTimeout)) { return nil } requestKey, err := buildDataDownloadRequestKey(item) if err != nil { return err } return mongoRepo.ReleaseDataDownloadFaultCall(ctx, stableDataDownloadDocumentID(requestKey), item.ID) } func faultRecordDocumentState(now time.Time, loc *time.Location, dataDay time.Time, failed int) (domain.DataDownloadState, *time.Time) { if failed > 0 || now.In(loc).Before(dataDay.AddDate(0, 0, 2)) { return domain.DataDownloadMutable, nil } frozenAt := now.UTC() return domain.DataDownloadFrozen, &frozenAt } func recordFaultRecordAttempt(ctx context.Context, item *domain.DataDownloadItem, documentID, sourceID, fileName string, entryID *primitive.ObjectID, stage string, cause error) error { status, reason := domain.DataDownloadItemSucceeded, "" if cause != nil { status, reason = domain.DataDownloadFailed, cause.Error() } return mongoRepo.RecordDataDownloadFileAttempt(ctx, &domain.DataDownloadFileAttempt{ TaskID: item.TaskID, ItemID: item.ID, DocumentID: documentID, SourceFileID: sourceID, FileName: fileName, Status: status, EntryID: entryID, FailureStage: stage, FailureReason: reason, }) }