FranklinWH-Toolbox/service/services/dataDownloadFaultRecordCall.go

package services

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/repositories"
	mongoRepo "fwh-toolbox-server/repositories/mongo"
	"go.uber.org/zap"
)

const (
	faultCallPageSize     = 50
	faultCallPollInterval = time.Minute
	faultCallTimeout      = 2 * time.Hour
)

type DataDownloadFaultCallDeferred struct{ After time.Duration }

func (e *DataDownloadFaultCallDeferred) Error() string {
	return "fault recording call is still running"
}

var errAmbiguousFaultCall = errors.New("ambiguous fault recording call")
var ErrFaultRecordCallTimeout = errors.New("fault recording call timed out")
var ErrFaultRecordCallFailed = errors.New("fault recording call failed")

func faultCallTimeoutError(item *domain.DataDownloadItem, lastStatus string) error {
	return fmt.Errorf("%w after 2h (recordId=%d, lastStatus=%s)", ErrFaultRecordCallTimeout, item.CallRecordID, lastStatus)
}

type faultRecordSource struct {
	record domain.FaultRecordInfo
	sub    domain.FaultRecordSubInfo
}

// resolveFaultRecordSources checks the callable catalog first. Only a subrecord
// whose failure time and fault code match a catalog entry proves that an
// existing remote record can be reused.
func resolveFaultRecordSources(ctx context.Context, item *domain.DataDownloadItem, documentID string, deviceType int, start, end time.Time) ([]faultRecordSource, error) {
	callable, err := repositories.QueryFaultRecordCallableList(deviceType, item.DeviceSN)
	if err != nil {
		return nil, fmt.Errorf("query callable fault records: %w", err)
	}
	selected := make([]domain.FaultRecordCallableInfo, 0)
	for _, candidate := range callable {
		failureTime, parseErr := time.ParseInLocation(time.DateTime, candidate.FailureTime, start.Location())
		if parseErr != nil {
			return nil, fmt.Errorf("invalid catalog failureTime %q: %w", candidate.FailureTime, parseErr)
		}
		if !failureTime.Before(start) && !failureTime.After(end) {
			selected = append(selected, candidate)
		}
	}
	if len(selected) == 0 {
		if item.CallRequestedAt != nil {
			return nil, fmt.Errorf("callable fault catalog became empty while call %d was active", item.CallRecordID)
		}
		return nil, nil
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	records, err := repositories.QueryFaultRecordList(item.DeviceSN, start.Format(time.DateTime), end.Format(time.DateTime))
	if err != nil {
		return nil, fmt.Errorf("query existing fault records: %w", err)
	}
	matched := make([]bool, len(selected))
	sources := make([]faultRecordSource, 0)
	for _, record := range records {
		if record.DeviceType != deviceType {
			continue
		}
		sn, identityErr := dataDownloadFaultRecordDeviceSN(record)
		if identityErr != nil {
			return sources, identityErr
		}
		if sn != item.DeviceSN {
			continue
		}
		if err := ctx.Err(); err != nil {
			return sources, err
		}
		subs, subErr := repositories.QueryFaultRecordSubList(record.Id)
		if subErr != nil {
			return sources, fmt.Errorf("list existing fault record %d: %w", record.Id, subErr)
		}
		if len(subs) != record.FaultRecordsCount {
			zap.L().Warn("[DataDownload] 已召唤录波子记录数量不一致，忽略该记录", zap.Int64("faultId", record.Id), zap.Int("expected", record.FaultRecordsCount), zap.Int("actual", len(subs)))
		}
		sources = append(sources, matchExistingFaultRecord(record, subs, selected, matched)...)
	}
	missing := make([]domain.FaultRecordCallableInfo, 0)
	indices := make([]int, 0)
	for i, candidate := range selected {
		if !matched[i] {
			missing = append(missing, candidate)
			indices = append(indices, candidate.Idx)
		}
	}
	if len(indices) == 0 {
		if item.CallRequestedAt != nil {
			if item.CallRecordStatus == nil || *item.CallRecordStatus != 2 {
				if _, err := summonFaultRecords(ctx, item, documentID, deviceType, nil); err != nil {
					return sources, err
				}
			}
			found := false
			for _, source := range sources {
				if source.record.Id == item.CallRecordID {
					found = true
					break
				}
			}
			if !found {
				return sources, fmt.Errorf("summoned fault record %d is not in the verified subrecords", item.CallRecordID)
			}
		}
		return sources, nil
	}
	called, err := summonFaultRecords(ctx, item, documentID, deviceType, indices)
	if err != nil {
		return sources, err
	}
	if err := ctx.Err(); err != nil {
		return sources, err
	}
	newRecords, err := repositories.QueryFaultRecordList(item.DeviceSN, start.Format(time.DateTime), end.Format(time.DateTime))
	if err != nil {
		return sources, fmt.Errorf("query summoned fault record %d: %w", called.Id, err)
	}
	var created *domain.FaultRecordInfo
	for i := range newRecords {
		if newRecords[i].Id == called.Id {
			created = &newRecords[i]
			break
		}
	}
	if created == nil {
		return sources, fmt.Errorf("summoned fault record %d is missing from faultRecordingLst", called.Id)
	}
	expected := item.CallRequestedCount
	if expected == 0 {
		expected = len(missing)
	}
	if created.FaultRecordsCount != expected {
		return sources, fmt.Errorf("summoned fault record %d count %d, expected %d", created.Id, created.FaultRecordsCount, expected)
	}
	subs, err := repositories.QueryFaultRecordSubList(created.Id)
	if err != nil {
		return sources, fmt.Errorf("list summoned fault record %d: %w", created.Id, err)
	}
	if len(subs) != expected {
		return sources, fmt.Errorf("summoned fault record %d has %d subrecords, expected %d", created.Id, len(subs), expected)
	}
	used := make([]bool, len(selected))
	for _, sub := range subs {
		match := -1
		for i, candidate := range selected {
			if !used[i] && faultSubMatchesCallable(sub, candidate) {
				match = i
				break
			}
		}
		if match < 0 {
			return sources, fmt.Errorf("summoned fault record %d contains unexpected subrecord %d", created.Id, sub.Id)
		}
		used[match] = true
		present := false
		for _, source := range sources {
			if source.record.Id == created.Id && source.sub.Id == sub.Id {
				present = true
				break
			}
		}
		if !present {
			sources = append(sources, faultRecordSource{record: *created, sub: sub})
		}
	}
	return sources, nil
}

func matchExistingFaultRecord(record domain.FaultRecordInfo, subs []domain.FaultRecordSubInfo, selected []domain.FaultRecordCallableInfo, matched []bool) []faultRecordSource {
	if len(subs) != record.FaultRecordsCount {
		return nil
	}
	var sources []faultRecordSource
	for _, sub := range subs {
		for i, candidate := range selected {
			if matched[i] || !faultSubMatchesCallable(sub, candidate) {
				continue
			}
			matched[i] = true
			sources = append(sources, faultRecordSource{record: record, sub: sub})
			break
		}
	}
	return sources
}

func faultSubMatchesCallable(sub domain.FaultRecordSubInfo, candidate domain.FaultRecordCallableInfo) bool {
	return strings.TrimSpace(sub.FailureTime) == strings.TrimSpace(candidate.FailureTime) &&
		strings.TrimSpace(sub.EventTriggerSource) == strings.TrimSpace(candidate.TriggerSourceCode)
}

func summonFaultRecords(ctx context.Context, item *domain.DataDownloadItem, documentID string, deviceType int, indices []int) (*domain.FaultRecordInfo, error) {
	requestedAt := item.CallRequestedAt
	if requestedAt != nil && !time.Now().Before(requestedAt.Add(faultCallTimeout)) {
		return nil, faultCallTimeoutError(item, callStatusText(item))
	}
	if requestedAt == nil {
		now := time.Now().UTC()
		requestedAt = &now
	}
	owned, err := mongoRepo.TryOwnDataDownloadFaultCall(ctx, documentID, item.ID, *requestedAt)
	if err != nil {
		return nil, err
	}
	if !owned {
		return nil, &DataDownloadFaultCallDeferred{After: faultCallPollInterval}
	}
	if item.CallRequestedAt == nil {
		if err := mongoRepo.UpdateDataDownloadFaultCall(ctx, item.ID, requestedAt, len(indices), 0, nil); err != nil {
			return nil, err
		}
		item.CallRequestedAt = requestedAt
		item.CallRequestedCount = len(indices)
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		if err := repositories.QueryCellFaultRecord(deviceType, item.DeviceSN, indices); err != nil {
			return nil, fmt.Errorf("call fault recording: %w", err)
		}
	}
	deadline := requestedAt.Add(faultCallTimeout)
	pollCtx, cancel := context.WithDeadline(ctx, deadline)
	defer cancel()
	if err := pollCtx.Err(); err != nil {
		return nil, faultCallTimeoutError(item, callStatusText(item))
	}
	task, err := mongoRepo.GetDataDownloadTask(pollCtx, item.TaskID)
	if err != nil {
		return nil, err
	}
	if task.Status == domain.DataDownloadCancelling || task.Status == domain.DataDownloadCancelled {
		return nil, context.Canceled
	}
	var record *domain.FaultRecordInfo
	if item.CallRecordID != 0 {
		record, err = findFaultCallByID(pollCtx, item.CallRecordID)
	} else {
		record, err = findFaultCallByRequest(pollCtx, item.DeviceSN, deviceType, item.CallRequestedCount, *requestedAt)
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return nil, faultCallTimeoutError(item, callStatusText(item))
	}
	if errors.Is(err, errAmbiguousFaultCall) {
		return nil, err
	}
	if err != nil {
		zap.L().Warn("[DataDownload] 录波召唤状态查询失败，下一轮重试",
			zap.String("itemId", item.ID.Hex()), zap.Int64("recordId", item.CallRecordID), zap.Error(err))
	}
	if record != nil {
		item.CallRecordID = record.Id
		item.CallRecordStatus = record.Status
		if err := mongoRepo.UpdateDataDownloadFaultCall(ctx, item.ID, nil, 0, record.Id, record.Status); err != nil {
			return nil, err
		}
		if record.Status != nil {
			switch *record.Status {
			case 2:
				return record, nil
			case -1:
				return nil, fmt.Errorf("%w: recordId=%d reason=%s", ErrFaultRecordCallFailed, record.Id, faultReason(record))
			}
		}
	}
	remaining := time.Until(deadline)
	if remaining <= 0 {
		return nil, faultCallTimeoutError(item, callStatusText(item))
	}
	if remaining > faultCallPollInterval {
		remaining = faultCallPollInterval
	}
	return nil, &DataDownloadFaultCallDeferred{After: remaining}
}

func callStatusText(item *domain.DataDownloadItem) string {
	if item.CallRecordStatus == nil {
		return "unknown"
	}
	return strconv.Itoa(*item.CallRecordStatus)
}

func faultReason(record *domain.FaultRecordInfo) string {
	if record.FaultReason == nil {
		return "upstream reported failure"
	}
	return *record.FaultReason
}

func findFaultCallByRequest(ctx context.Context, sn string, deviceType, count int, requestedAt time.Time) (*domain.FaultRecordInfo, error) {
	var found *domain.FaultRecordInfo
	lowerTime := requestedAt.Add(-time.Minute)
	for page := 1; page <= 200; page++ {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		records, err := repositories.QueryFaultRecordCallRecordPageContext(ctx, page, faultCallPageSize)
		if err != nil {
			return nil, err
		}
		for i := range records {
			record := &records[i]
			if record.DeviceType != deviceType || record.FaultRecordsCount != count || record.TriggerType != 2 {
				continue
			}
			deviceSN, snErr := dataDownloadFaultRecordDeviceSN(*record)
			if snErr != nil || deviceSN != sn {
				continue
			}
			callTime, parseErr := time.ParseInLocation(time.DateTime, record.CellTime, time.UTC)
			if parseErr != nil || callTime.Before(requestedAt.Add(-time.Minute)) || callTime.After(requestedAt.Add(time.Minute)) {
				continue
			}
			if found != nil && found.Id != record.Id {
				return nil, fmt.Errorf("%w for %s at %s", errAmbiguousFaultCall, sn, requestedAt.Format(time.RFC3339))
			}
			copy := *record
			found = &copy
		}
		// IDs are returned newest first. Once the oldest row is older than the
		// match window, later pages cannot contain this request.
		if len(records) > 0 {
			oldest, parseErr := time.ParseInLocation(time.DateTime, records[len(records)-1].CellTime, time.UTC)
			if parseErr == nil && oldest.Before(lowerTime) {
				return found, nil
			}
		}
		if len(records) < faultCallPageSize {
			return found, nil
		}
	}
	return nil, fmt.Errorf("fault recording call search exceeded 200 pages")
}

func findFaultCallByID(ctx context.Context, id int64) (*domain.FaultRecordInfo, error) {
	probe := func(page int) (*domain.FaultRecordInfo, bool, error) {
		if err := ctx.Err(); err != nil {
			return nil, false, err
		}
		records, err := repositories.QueryFaultRecordCallRecordPageContext(ctx, page, faultCallPageSize)
		if err != nil {
			return nil, false, err
		}
		for i := range records {
			if records[i].Id == id {
				return &records[i], false, nil
			}
		}
		return nil, len(records) > 0 && records[len(records)-1].Id > id, nil
	}
	// Exponential probing bounds the target page; binary search then locates
	// it without scanning every newer page on each 60-second poll.
	lower, upper := 0, 1
	for upper <= 1024 {
		record, newer, err := probe(upper)
		if err != nil {
			return nil, err
		}
		if record != nil {
			return record, nil
		}
		if !newer {
			break
		}
		lower, upper = upper, upper*2
	}
	if upper > 1024 {
		return nil, fmt.Errorf("fault recording call %d is older than 1024 pages", id)
	}
	for lower+1 < upper {
		middle := lower + (upper-lower)/2
		record, newer, err := probe(middle)
		if err != nil {
			return nil, err
		}
		if record != nil {
			return record, nil
		}
		if newer {
			lower = middle
		} else {
			upper = middle
		}
	}
	return nil, fmt.Errorf("fault recording call %d is absent from lstFaultWaveCall", id)
}