junhong_cmp_fiber/internal/application/cardobservation/series.go

package cardobservation

import (
	"context"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
)

// SeriesRequest 描述一次业务成功边界产生的观测序列请求。
type SeriesRequest struct {
	Scene         string `json:"scene"`
	ResourceType  string `json:"resource_type"`
	ResourceID    string `json:"resource_id"`
	SyncType      string `json:"sync_type"`
	ExpectedValue string `json:"expected_value,omitempty"`
	Source        string `json:"source"`
	RequestID     string `json:"request_id,omitempty"`
	CorrelationID string `json:"correlation_id,omitempty"`
}

// DeviceCardsSeriesRequest 描述需要在后台展开设备有效绑定卡的观测请求。
type DeviceCardsSeriesRequest struct {
	DeviceID uint
	Request  SeriesRequest
}

// DeviceControlSeriesRequest 描述设备控制成功后需要在后台解析的差异化观测资源。
type DeviceControlSeriesRequest struct {
	DeviceID             uint
	TargetICCID          string
	SourceCardID         uint
	TargetCardID         uint
	BoundCardIDs         []uint
	IncludeTargetTraffic bool
	Request              SeriesRequest
}

// RealnameCapabilitySeriesRequest 描述需要在后台判断运营商实名能力的观测请求。
type RealnameCapabilitySeriesRequest struct {
	CarrierID uint
	Request   SeriesRequest
}

// SeriesTaskPayload 是固定三次 Asynq 任务的结构化载荷。
type SeriesTaskPayload struct {
	SeriesID      string    `json:"series_id"`
	Attempt       int       `json:"attempt"`
	ScheduledAt   time.Time `json:"scheduled_at"`
	Scene         string    `json:"scene"`
	ResourceType  string    `json:"resource_type"`
	ResourceID    string    `json:"resource_id"`
	SyncType      string    `json:"sync_type"`
	ExpectedValue string    `json:"expected_value,omitempty"`
	Source        string    `json:"source"`
	RequestID     string    `json:"request_id,omitempty"`
	CorrelationID string    `json:"correlation_id,omitempty"`
}

// RunResult 描述一次实际 Gateway 请求及公共观测应用结果。
type RunResult struct {
	StateChanged bool
	RateLimited  bool
}

// SeriesCoordinator 管理活跃序列、尝试幂等和实际请求互斥。
type SeriesCoordinator interface {
	Reserve(ctx context.Context, request SeriesRequest, candidateSeriesID string, candidateBaseTime time.Time) (seriesID string, baseTime time.Time, originalRequest SeriesRequest, merged bool, err error)
	IsScheduled(ctx context.Context, seriesID string) (bool, error)
	ClaimSchedule(ctx context.Context, seriesID string, attempt int) (bool, error)
	ReleaseSchedule(ctx context.Context, seriesID string, attempt int)
	IsCompleted(ctx context.Context, seriesID string) (bool, error)
	ClaimAttempt(ctx context.Context, seriesID string, attempt int) (bool, error)
	AcquireRequest(ctx context.Context, payload SeriesTaskPayload, provider string) (release func(), wait time.Duration, acquired bool, err error)
	FinishAttempt(ctx context.Context, payload SeriesTaskPayload) error
	CompleteSeries(ctx context.Context, payload SeriesTaskPayload) error
	CompleteResourceSeries(ctx context.Context, resourceType, resourceID, syncType string) error
}

// SeriesScheduler 提交固定的零重试观测任务。
type SeriesScheduler interface {
	Enqueue(ctx context.Context, payload SeriesTaskPayload) error
}

// BestEffortSeriesDispatcher 为读取入口提供不返回业务错误的轻量触发端口。
type BestEffortSeriesDispatcher interface {
	Dispatch(ctx context.Context, request SeriesRequest)
	DispatchDeviceCards(ctx context.Context, request DeviceCardsSeriesRequest)
	DispatchDeviceControl(ctx context.Context, request DeviceControlSeriesRequest)
	DispatchRealnameWithCapability(ctx context.Context, request RealnameCapabilitySeriesRequest)
}

// SeriesAttemptLogger 记录未访问 Gateway 的合并、互斥、限频和提前完成结果。
type SeriesAttemptLogger interface {
	Record(ctx context.Context, payload SeriesTaskPayload, result, reason string) error
	RecordMerged(ctx context.Context, request SeriesRequest, seriesID string) error
}

// SeriesRunner 查询本地快照并执行一次 Gateway 公共观测。
type SeriesRunner interface {
	Provider(ctx context.Context, payload SeriesTaskPayload) (string, error)
	ExpectationMet(ctx context.Context, payload SeriesTaskPayload) (bool, error)
	Run(ctx context.Context, payload SeriesTaskPayload) (RunResult, error)
}

// SeriesTrigger 创建或合并固定的立即、3 分钟、5 分钟任务序列。
type SeriesTrigger struct {
	coordinator SeriesCoordinator
	scheduler   SeriesScheduler
	logger      SeriesAttemptLogger
	now         func() time.Time
}

// NewSeriesTrigger 创建观测序列触发器。
func NewSeriesTrigger(coordinator SeriesCoordinator, scheduler SeriesScheduler, logger SeriesAttemptLogger) *SeriesTrigger {
	return &SeriesTrigger{coordinator: coordinator, scheduler: scheduler, logger: logger, now: time.Now}
}

// Trigger 创建新序列；同场景未结束序列只留合并记录，不延长原序列。
func (s *SeriesTrigger) Trigger(ctx context.Context, request SeriesRequest) (string, bool, error) {
	return s.trigger(ctx, request, uuid.NewString(), false)
}

// TriggerEvent 使用稳定业务事件键触发序列，至少一次重投不会再次创建任务。
func (s *SeriesTrigger) TriggerEvent(ctx context.Context, eventKey string, request SeriesRequest) (string, bool, error) {
	seriesID := uuid.NewSHA1(uuid.NameSpaceOID, []byte(eventKey)).String()
	return s.trigger(ctx, request, seriesID, true)
}

func (s *SeriesTrigger) trigger(ctx context.Context, request SeriesRequest, candidateSeriesID string, stable bool) (string, bool, error) {
	if s == nil || s.coordinator == nil || s.scheduler == nil || s.logger == nil {
		return "", false, errors.New(errors.CodeInternalError, "卡观测序列触发能力未完整配置")
	}
	request = normalizeSeriesTrace(request)
	if err := validateSeriesRequest(request); err != nil {
		return "", false, err
	}
	if stable {
		scheduled, err := s.coordinator.IsScheduled(ctx, candidateSeriesID)
		if err != nil {
			return "", false, err
		}
		if scheduled {
			if logErr := s.logger.RecordMerged(ctx, request, candidateSeriesID); logErr != nil {
				return candidateSeriesID, true, logErr
			}
			return candidateSeriesID, true, nil
		}
	}
	candidateBaseTime := s.now().UTC()
	seriesID, baseTime, originalRequest, merged, err := s.coordinator.Reserve(ctx, request, candidateSeriesID, candidateBaseTime)
	if err != nil {
		return "", false, err
	}
	// 重复触发仍以稳定任务 ID 补齐首次入队的局部失败；已存在任务由 Asynq 去重，不会延长原序列。
	for attempt := 1; attempt <= constants.CardObservationSeriesAttemptCount; attempt++ {
		claimed, claimErr := s.coordinator.ClaimSchedule(ctx, seriesID, attempt)
		if claimErr != nil {
			return seriesID, merged, claimErr
		}
		if !claimed {
			continue
		}
		scheduledAt := baseTime.Add(constants.CardObservationAttemptDelay(attempt))
		payload := SeriesTaskPayload{
			SeriesID: seriesID, Attempt: attempt, ScheduledAt: scheduledAt,
			Scene: originalRequest.Scene, ResourceType: originalRequest.ResourceType, ResourceID: originalRequest.ResourceID,
			SyncType: originalRequest.SyncType, ExpectedValue: originalRequest.ExpectedValue, Source: originalRequest.Source,
			RequestID: originalRequest.RequestID, CorrelationID: originalRequest.CorrelationID,
		}
		if err := s.scheduler.Enqueue(ctx, payload); err != nil {
			s.coordinator.ReleaseSchedule(ctx, seriesID, attempt)
			return seriesID, false, err
		}
	}
	if merged {
		if err := s.logger.RecordMerged(ctx, request, seriesID); err != nil {
			return "", true, err
		}
	}
	return seriesID, merged, nil
}

// SeriesAttemptService 执行单次事件观测，不改变后续阶梯任务。
type SeriesAttemptService struct {
	coordinator SeriesCoordinator
	runner      SeriesRunner
	logger      SeriesAttemptLogger
}

// NewSeriesAttemptService 创建序列尝试服务。
func NewSeriesAttemptService(coordinator SeriesCoordinator, runner SeriesRunner, logger SeriesAttemptLogger) *SeriesAttemptService {
	return &SeriesAttemptService{coordinator: coordinator, runner: runner, logger: logger}
}

// Execute 执行一次幂等尝试；失败只结束当前任务。
func (s *SeriesAttemptService) Execute(ctx context.Context, payload SeriesTaskPayload) error {
	if s == nil || s.coordinator == nil || s.runner == nil || s.logger == nil {
		return errors.New(errors.CodeInternalError, "卡观测序列执行能力未完整配置")
	}
	if err := validateSeriesPayload(payload); err != nil {
		return err
	}
	claimed, err := s.coordinator.ClaimAttempt(ctx, payload.SeriesID, payload.Attempt)
	if err != nil || !claimed {
		return err
	}
	defer func() { _ = s.coordinator.FinishAttempt(context.Background(), payload) }()

	completed, err := s.coordinator.IsCompleted(ctx, payload.SeriesID)
	if err != nil {
		return s.recordPreGatewayFailure(ctx, payload, "读取序列完成状态失败", err)
	}
	if completed {
		return s.completeRemainingAttempts(ctx, payload, "序列已提前完成")
	}
	met, err := s.runner.ExpectationMet(ctx, payload)
	if err != nil {
		return s.recordPreGatewayFailure(ctx, payload, "读取本地预期快照失败", err)
	}
	if met {
		return s.completeRemainingAttempts(ctx, payload, "本地快照已达到预期")
	}
	provider, err := s.runner.Provider(ctx, payload)
	if err != nil {
		return s.recordPreGatewayFailure(ctx, payload, "解析运营商接入失败", err)
	}
	release, wait, acquired, err := s.coordinator.AcquireRequest(ctx, payload, provider)
	if err != nil {
		return s.recordPreGatewayFailure(ctx, payload, "获取实际请求互斥失败", err)
	}
	if !acquired {
		return s.logger.Record(ctx, payload, constants.IntegrationResultIgnored, "实际 Gateway 请求正在执行")
	}
	defer release()
	if wait > 0 {
		timer := time.NewTimer(wait)
		defer timer.Stop()
		select {
		case <-ctx.Done():
			return s.logger.Record(ctx, payload, constants.IntegrationResultRateLimited, "最小请求间隔等待被取消")
		case <-timer.C:
		}
	}
	result, err := s.runner.Run(ctx, payload)
	if result.RateLimited {
		return nil
	}
	if err != nil {
		return err
	}
	if payload.Attempt == constants.CardObservationSeriesAttemptCount {
		return s.coordinator.CompleteSeries(ctx, payload)
	}
	return nil
}

func (s *SeriesAttemptService) recordPreGatewayFailure(ctx context.Context, payload SeriesTaskPayload, reason string, original error) error {
	if logErr := s.logger.Record(ctx, payload, constants.IntegrationResultFailed, reason); logErr != nil {
		return errors.Wrap(errors.CodeInternalError, original, reason+"，且 Integration Log 写入失败")
	}
	return original
}

func (s *SeriesAttemptService) completeRemainingAttempts(ctx context.Context, payload SeriesTaskPayload, reason string) error {
	baseTime := payload.ScheduledAt.Add(-constants.CardObservationAttemptDelay(payload.Attempt))
	for attempt := payload.Attempt; attempt <= constants.CardObservationSeriesAttemptCount; attempt++ {
		remaining := payload
		remaining.Attempt = attempt
		remaining.ScheduledAt = baseTime.Add(constants.CardObservationAttemptDelay(attempt))
		if err := s.logger.Record(ctx, remaining, constants.IntegrationResultCompleted, reason); err != nil {
			return err
		}
	}
	return s.coordinator.CompleteSeries(ctx, payload)
}

// CompleteResourceSeries 供可信回调在公共观测成功后提前完成同资源序列。
func (s *SeriesAttemptService) CompleteResourceSeries(ctx context.Context, resourceType, resourceID, syncType string) error {
	if s == nil || s.coordinator == nil {
		return errors.New(errors.CodeInternalError, "卡观测序列协调器未配置")
	}
	return s.coordinator.CompleteResourceSeries(ctx, resourceType, resourceID, syncType)
}

func validateSeriesRequest(request SeriesRequest) error {
	if strings.TrimSpace(request.Scene) == "" || strings.TrimSpace(request.ResourceType) == "" ||
		strings.TrimSpace(request.ResourceID) == "" || !validSyncType(request.SyncType) || !validObservationSource(request.Source) ||
		strings.TrimSpace(request.RequestID) == "" || strings.TrimSpace(request.CorrelationID) == "" {
		return errors.New(errors.CodeInvalidParam, "卡观测序列参数不完整")
	}
	return nil
}

func validateSeriesPayload(payload SeriesTaskPayload) error {
	if payload.SeriesID == "" || payload.Attempt < 1 || payload.Attempt > constants.CardObservationSeriesAttemptCount {
		return errors.New(errors.CodeInvalidParam, "卡观测序列任务载荷无效")
	}
	return validateSeriesRequest(SeriesRequest{
		Scene: payload.Scene, ResourceType: payload.ResourceType, ResourceID: payload.ResourceID,
		SyncType: payload.SyncType, Source: payload.Source, RequestID: payload.RequestID, CorrelationID: payload.CorrelationID,
	})
}

func normalizeSeriesTrace(request SeriesRequest) SeriesRequest {
	request.RequestID = strings.TrimSpace(request.RequestID)
	request.CorrelationID = strings.TrimSpace(request.CorrelationID)
	if request.RequestID == "" && request.CorrelationID == "" {
		traceID := uuid.NewString()
		request.RequestID = traceID
		request.CorrelationID = traceID
	} else if request.RequestID == "" {
		request.RequestID = request.CorrelationID
	} else if request.CorrelationID == "" {
		request.CorrelationID = request.RequestID
	}
	return request
}

func validObservationSource(source string) bool {
	switch source {
	case constants.CardObservationSourcePolling, constants.CardObservationSourceManualSync,
		constants.CardObservationSourceManualOverride, constants.CardObservationSourceCarrierCallback,
		constants.CardObservationSourceBusinessEvent:
		return true
	default:
		return false
	}
}

func validSyncType(syncType string) bool {
	switch syncType {
	case constants.CardObservationSyncTypeRealname, constants.CardObservationSyncTypeTraffic,
		constants.CardObservationSyncTypeNetwork, constants.CardObservationSyncTypeDeviceInfo:
		return true
	default:
		return false
	}
}