FranklinWH-Toolbox/service/services/dataDownloadFaultRecordService.go

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,
	})
}