junhong_cmp_fiber/internal/infrastructure/cardobservation/series_runner.go

package cardobservation

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

	"gorm.io/gorm"

	cardapp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
	carddomain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation"
	"github.com/break/junhong_cmp_fiber/internal/gateway"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	apperrors "github.com/break/junhong_cmp_fiber/pkg/errors"
)

// SeriesAttemptLogger 将未发送的序列结果写入统一 Integration Log。
type SeriesAttemptLogger struct {
	repository *integrationlog.Repository
}

// NewSeriesAttemptLogger 创建未发送尝试日志适配器。
func NewSeriesAttemptLogger(repository *integrationlog.Repository) *SeriesAttemptLogger {
	return &SeriesAttemptLogger{repository: repository}
}

// Record 写入互斥、限频或提前完成结果。
func (l *SeriesAttemptLogger) Record(ctx context.Context, payload cardapp.SeriesTaskPayload, result, reason string) error {
	if l == nil || l.repository == nil {
		return apperrors.New(apperrors.CodeInternalError, "卡观测 Integration Log 未配置")
	}
	resourceID := payload.ResourceID
	source, scene, seriesID := payload.Source, payload.Scene, payload.SeriesID+":"+payload.SyncType
	requestID, correlationID := optionalText(payload.RequestID), optionalText(payload.CorrelationID)
	attempt, err := l.repository.Start(ctx, integrationlog.Attempt{
		IntegrationID: unsentIntegrationID(payload), Provider: constants.IntegrationProviderGateway,
		Direction: constants.IntegrationDirectionOutbound, Operation: operationForSyncType(payload.SyncType),
		ResourceType: payload.ResourceType, ResourceID: &resourceID, TriggerSource: &source,
		TriggerScene: &scene, TriggerSeries: &seriesID, ScheduledAt: &payload.ScheduledAt,
		Attempt: payload.Attempt, InitialResult: initialResult(result), RequestID: requestID, CorrelationID: correlationID,
		Metadata: map[string]any{"reason": reason, "sync_type": payload.SyncType},
	})
	if err != nil || result != constants.IntegrationResultFailed {
		return err
	}
	_, err = l.repository.Complete(ctx, attempt.IntegrationID, integrationlog.Completion{
		Result: constants.IntegrationResultFailed, ResponseSummary: map[string]any{"reason": reason},
	})
	return err
}

// RecordMerged 记录同场景重复触发被合并，不创建第二组三任务。
func (l *SeriesAttemptLogger) RecordMerged(ctx context.Context, request cardapp.SeriesRequest, seriesID string) error {
	resourceID := request.ResourceID
	source, scene, technicalSeriesID := request.Source, request.Scene, seriesID+":"+request.SyncType
	now := time.Now().UTC()
	_, err := l.repository.Start(ctx, integrationlog.Attempt{
		Provider: constants.IntegrationProviderGateway, Direction: constants.IntegrationDirectionOutbound,
		Operation: operationForSyncType(request.SyncType), ResourceType: request.ResourceType,
		ResourceID: &resourceID, TriggerSource: &source, TriggerScene: &scene, TriggerSeries: &technicalSeriesID,
		ScheduledAt: &now, Attempt: 1, InitialResult: constants.IntegrationResultMerged,
		RequestID: optionalText(request.RequestID), CorrelationID: optionalText(request.CorrelationID),
		Metadata: map[string]any{"reason": "同场景未结束序列已存在", "sync_type": request.SyncType},
	})
	return err
}

// SeriesRunner 负责卡本地预期判断、Gateway 查询和公共观测应用。
type SeriesRunner struct {
	db          *gorm.DB
	gateway     *gateway.Client
	observation *cardapp.Service
	integration *integrationlog.Repository
	carrier     *postgres.CarrierStore
	auditWriter *audit.Writer
}

// NewSeriesRunner 创建卡观测序列 Gateway 执行器。
func NewSeriesRunner(db *gorm.DB, gatewayClient *gateway.Client, observation *cardapp.Service, integration *integrationlog.Repository, auditWriter *audit.Writer) *SeriesRunner {
	return &SeriesRunner{
		db: db, gateway: gatewayClient, observation: observation, integration: integration,
		carrier: postgres.NewCarrierStore(db), auditWriter: auditWriter,
	}
}

// Provider 返回用于请求互斥的运营商接入标识。
func (r *SeriesRunner) Provider(ctx context.Context, payload cardapp.SeriesTaskPayload) (string, error) {
	if payload.ResourceType == constants.CardObservationResourceTypeDevice {
		if _, err := r.loadDevice(ctx, payload); err != nil {
			return "", err
		}
		return constants.CardObservationResourceTypeDevice, nil
	}
	card, err := r.loadCard(ctx, payload)
	if err != nil {
		return "", err
	}
	provider := strings.ToLower(strings.TrimSpace(card.CarrierType))
	if provider == "" {
		provider = "carrier-" + strconv.FormatUint(uint64(card.CarrierID), 10)
	}
	return provider, nil
}

// ExpectationMet 在访问 Gateway 前读取本地权威快照。
func (r *SeriesRunner) ExpectationMet(ctx context.Context, payload cardapp.SeriesTaskPayload) (bool, error) {
	expected := strings.ToLower(strings.TrimSpace(payload.ExpectedValue))
	if expected == "" {
		return false, nil
	}
	if payload.ResourceType == constants.CardObservationResourceTypeDevice && payload.SyncType == constants.CardObservationSyncTypeDeviceInfo {
		currentICCIDs, err := r.loadCurrentICCIDs(ctx, payload)
		if err != nil {
			return false, err
		}
		for _, currentICCID := range currentICCIDs {
			if strings.EqualFold(expected, strings.TrimSpace(currentICCID)) {
				return true, nil
			}
		}
		return false, nil
	}
	card, err := r.loadCard(ctx, payload)
	if err != nil {
		return false, err
	}
	switch payload.SyncType {
	case constants.CardObservationSyncTypeRealname:
		return (expected == "verified" || expected == "1") && card.RealNameStatus == constants.RealNameStatusVerified, nil
	case constants.CardObservationSyncTypeNetwork:
		if expected == "online" || expected == "1" {
			return card.NetworkStatus == constants.NetworkStatusOnline, nil
		}
		if expected == "offline" || expected == "stopped" || expected == "0" {
			return card.NetworkStatus == constants.NetworkStatusOffline, nil
		}
	}
	return false, nil
}

// Run 执行一次 Gateway 查询，并将结果交给公共卡观测应用服务。
func (r *SeriesRunner) Run(ctx context.Context, payload cardapp.SeriesTaskPayload) (cardapp.RunResult, error) {
	if r == nil || r.db == nil || r.gateway == nil || r.observation == nil || r.integration == nil {
		return cardapp.RunResult{}, apperrors.New(apperrors.CodeInternalError, "卡观测序列 Gateway 能力未完整配置")
	}
	if payload.ResourceType == constants.CardObservationResourceTypeDevice && payload.SyncType == constants.CardObservationSyncTypeDeviceInfo {
		return r.runDeviceInfo(ctx, payload)
	}
	card, err := r.loadCard(ctx, payload)
	if err != nil {
		return cardapp.RunResult{}, err
	}
	attempt, err := r.startAttempt(ctx, payload, card)
	if err != nil {
		return cardapp.RunResult{}, err
	}
	startedAt := time.Now()
	result, runErr := r.queryAndApply(ctx, payload, card, attempt.IntegrationID)
	completionResult := constants.IntegrationResultSuccess
	if runErr != nil {
		completionResult = constants.IntegrationResultFailed
		if isRateLimited(runErr) {
			completionResult = constants.IntegrationResultRateLimited
			result.RateLimited = true
		}
	}
	_, completeErr := r.integration.Complete(ctx, attempt.IntegrationID, integrationlog.Completion{
		Result: completionResult, DurationMS: time.Since(startedAt).Milliseconds(), StateChanged: result.StateChanged,
		ResponseSummary: map[string]any{"sync_type": payload.SyncType, "applied": runErr == nil},
	})
	if completeErr != nil {
		return result, completeErr
	}
	return result, runErr
}

func (r *SeriesRunner) runDeviceInfo(ctx context.Context, payload cardapp.SeriesTaskPayload) (cardapp.RunResult, error) {
	device, err := r.loadDevice(ctx, payload)
	if err != nil {
		return cardapp.RunResult{}, err
	}
	attempt, err := r.startDeviceAttempt(ctx, payload, device)
	if err != nil {
		return cardapp.RunResult{}, err
	}
	startedAt := time.Now()
	response, runErr := r.gateway.SyncDeviceInfo(ctx, &gateway.SyncDeviceInfoReq{CardNo: deviceGatewayIdentifier(device)})
	result := cardapp.RunResult{}
	if runErr == nil {
		result.StateChanged, runErr = r.applyDeviceInfo(ctx, device, response, attempt.IntegrationID)
	}
	completionResult := constants.IntegrationResultSuccess
	if runErr != nil {
		completionResult = constants.IntegrationResultFailed
		if isRateLimited(runErr) {
			completionResult = constants.IntegrationResultRateLimited
			result.RateLimited = true
		}
	}
	_, completeErr := r.integration.Complete(ctx, attempt.IntegrationID, integrationlog.Completion{
		Result: completionResult, DurationMS: time.Since(startedAt).Milliseconds(), StateChanged: result.StateChanged,
		ResponseSummary: map[string]any{"sync_type": payload.SyncType, "applied": runErr == nil},
	})
	if completeErr != nil {
		return result, completeErr
	}
	return result, runErr
}

func (r *SeriesRunner) startDeviceAttempt(ctx context.Context, payload cardapp.SeriesTaskPayload, device *model.Device) (*model.IntegrationLog, error) {
	resourceID, source, scene, seriesID := payload.ResourceID, payload.Source, payload.Scene, payload.SeriesID+":"+payload.SyncType
	resourceKey := "device:" + strconv.FormatUint(uint64(device.ID), 10)
	return r.integration.Start(ctx, integrationlog.Attempt{
		IntegrationID: gatewayIntegrationID(payload), Provider: constants.IntegrationProviderGateway,
		Direction: constants.IntegrationDirectionOutbound, Operation: operationForSyncType(payload.SyncType),
		ResourceType: payload.ResourceType, ResourceID: &resourceID, ResourceKey: &resourceKey,
		TriggerSource: &source, TriggerScene: &scene, TriggerSeries: &seriesID,
		ScheduledAt: &payload.ScheduledAt, Attempt: payload.Attempt,
		RequestSummary: map[string]any{"device_id": device.ID, "sync_type": payload.SyncType},
		RequestID:      optionalText(payload.RequestID), CorrelationID: optionalText(payload.CorrelationID),
	})
}

func (r *SeriesRunner) applyDeviceInfo(ctx context.Context, device *model.Device, response *gateway.SyncDeviceInfoResp, integrationID string) (bool, error) {
	if response == nil {
		return false, apperrors.New(apperrors.CodeGatewayError, "Gateway 设备信息响应为空")
	}
	now := time.Now()
	updates := map[string]any{
		"online_status":        int(response.OnlineStatus),
		"software_version":     string(response.SoftwareVersion),
		"switch_mode":          string(response.SwitchMode),
		"last_gateway_sync_at": now,
	}
	if lastOnlineTime := parseGatewayTime(response.LastOnlineTime); lastOnlineTime != nil {
		updates["last_online_time"] = lastOnlineTime
	}
	deviceChanged := device.OnlineStatus != int(response.OnlineStatus) ||
		device.SoftwareVersion != string(response.SoftwareVersion) ||
		device.SwitchMode != string(response.SwitchMode)
	changed := deviceChanged
	err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		state, err := loadDeviceObservationState(ctx, tx, device.ID, int(response.CurrentSlotNo))
		if err != nil {
			return err
		}
		beforeSlot := bindingSlot(state.current)
		afterSlot := bindingSlot(state.target)
		changed = deviceChanged || beforeSlot != int(response.CurrentSlotNo)
		auditChanged := deviceChanged || beforeSlot != afterSlot
		if err := tx.Model(&model.Device{}).Where("id = ?", device.ID).Updates(updates).Error; err != nil {
			return err
		}
		if err := tx.Model(&model.DeviceSimBinding{}).
			Where("device_id = ? AND bind_status = ?", device.ID, constants.BindStatusBound).
			Update("is_current", false).Error; err != nil {
			return err
		}
		if int(response.CurrentSlotNo) > 0 {
			if err := tx.Model(&model.DeviceSimBinding{}).
				Where("device_id = ? AND slot_position = ? AND bind_status = ?", device.ID, int(response.CurrentSlotNo), constants.BindStatusBound).
				Update("is_current", true).Error; err != nil {
				return err
			}
		}
		if !auditChanged {
			return nil
		}
		return r.appendDeviceObservationAudit(ctx, tx, device, state, beforeSlot, afterSlot, integrationID, map[string]any{
			"online_status": device.OnlineStatus, "software_version": device.SoftwareVersion,
			"switch_mode": device.SwitchMode, "current_slot": beforeSlot,
		}, map[string]any{
			"online_status": int(response.OnlineStatus), "software_version": string(response.SoftwareVersion),
			"switch_mode": string(response.SwitchMode), "current_slot": afterSlot,
		})
	})
	if err != nil {
		wrapped := apperrors.Wrap(apperrors.CodeDatabaseError, err, "回写设备 Gateway 信息失败")
		r.recordDeviceObservationFailure(ctx, device, integrationID, wrapped)
		return false, wrapped
	}
	return changed, nil
}

type deviceObservationState struct {
	bindings []model.DeviceSimBinding
	cards    map[uint]*model.IotCard
	current  *model.DeviceSimBinding
	target   *model.DeviceSimBinding
}

func loadDeviceObservationState(ctx context.Context, tx *gorm.DB, deviceID uint, targetSlot int) (*deviceObservationState, error) {
	state := &deviceObservationState{cards: make(map[uint]*model.IotCard)}
	if err := tx.WithContext(ctx).Where("device_id = ? AND bind_status = ?", deviceID, constants.BindStatusBound).
		Order("slot_position ASC").Find(&state.bindings).Error; err != nil {
		return nil, apperrors.Wrap(apperrors.CodeDatabaseError, err, "读取设备有效卡槽失败")
	}
	cardIDs := make([]uint, 0, len(state.bindings))
	for index := range state.bindings {
		binding := &state.bindings[index]
		cardIDs = append(cardIDs, binding.IotCardID)
		if binding.IsCurrent && state.current == nil {
			state.current = binding
		}
		if targetSlot > 0 && binding.SlotPosition == targetSlot {
			state.target = binding
		}
	}
	if len(cardIDs) == 0 {
		return state, nil
	}
	var cards []*model.IotCard
	if err := tx.WithContext(ctx).Where("id IN ?", cardIDs).Find(&cards).Error; err != nil {
		return nil, apperrors.Wrap(apperrors.CodeDatabaseError, err, "读取设备卡槽关联 IoT 卡失败")
	}
	for _, card := range cards {
		state.cards[card.ID] = card
	}
	return state, nil
}

func bindingSlot(binding *model.DeviceSimBinding) int {
	if binding == nil {
		return 0
	}
	return binding.SlotPosition
}

func (r *SeriesRunner) appendDeviceObservationAudit(
	ctx context.Context,
	tx *gorm.DB,
	device *model.Device,
	state *deviceObservationState,
	beforeSlot, afterSlot int,
	integrationID string,
	beforeData, afterData map[string]any,
) error {
	if r.auditWriter == nil {
		return apperrors.New(apperrors.CodeInternalError, "设备观测统一审计能力未配置")
	}
	deviceID := strconv.FormatUint(uint64(device.ID), 10)
	resources := []audit.ResourceInput{{
		Type: constants.AuditResourceDevice, ID: &deviceID,
		Key: audit.DeviceResourceKey(device), DisplayName: device.VirtualNo,
		Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleDeviceTarget,
		IdentitySnapshot: audit.DeviceIdentitySnapshot(device), BeforeData: beforeData, AfterData: afterData,
		SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: "Worker 同步设备观测事实",
	}}
	if beforeSlot != afterSlot {
		resources = appendDeviceObservationBindingResources(resources, device, state.current, state.cards[stateCardID(state.current)], false,
			constants.AuditResourceRoleDeviceOldCurrentCard, constants.AuditResourceRoleDeviceOldCurrentBinding)
		resources = appendDeviceObservationBindingResources(resources, device, state.target, state.cards[stateCardID(state.target)], true,
			constants.AuditResourceRoleDeviceNewCurrentCard, constants.AuditResourceRoleDeviceNewCurrentBinding)
	}
	resources = append(resources, deviceObservationIntegrationResource(ctx, device, integrationID))
	return r.auditWriter.Append(ctx, tx, audit.AppendInput{
		ActionCode: constants.AuditActionDeviceWorkerObservationSynced,
		Summary:    "Worker 同步设备观测事实", ScopeType: constants.AuditScopePlatform,
		Result: constants.AuditResultSuccess, Resources: resources,
	})
}

func appendDeviceObservationBindingResources(
	resources []audit.ResourceInput,
	device *model.Device,
	binding *model.DeviceSimBinding,
	card *model.IotCard,
	afterCurrent bool,
	cardRole, bindingRole string,
) []audit.ResourceInput {
	if binding == nil {
		return resources
	}
	if card != nil {
		cardID := strconv.FormatUint(uint64(card.ID), 10)
		resources = append(resources, audit.ResourceInput{
			Type: constants.AuditResourceIotCard, ID: &cardID,
			Key: audit.IotCardResourceKey(card), DisplayName: card.ICCID,
			Relation: constants.AuditResourceRelationAffected, Role: cardRole,
			IdentitySnapshot: audit.IotCardIdentitySnapshot(card),
			BeforeData:       map[string]any{"is_current": binding.IsCurrent}, AfterData: map[string]any{"is_current": afterCurrent},
			SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: "设备当前卡槽已同步",
		})
	}
	bindingID := strconv.FormatUint(uint64(binding.ID), 10)
	identity := map[string]any{
		"id": binding.ID, "device_id": binding.DeviceID, "device_virtual_no": device.VirtualNo,
		"slot_position": binding.SlotPosition, "iot_card_id": binding.IotCardID, "is_current": afterCurrent,
	}
	if card != nil {
		identity["iccid"] = card.ICCID
		identity["virtual_no"] = card.VirtualNo
	}
	return append(resources, audit.ResourceInput{
		Type: constants.AuditResourceDeviceSIMBinding, ID: &bindingID,
		Key: bindingID, DisplayName: device.VirtualNo,
		Relation: constants.AuditResourceRelationAffected, Role: bindingRole,
		IdentitySnapshot: identity,
		BeforeData:       map[string]any{"is_current": binding.IsCurrent}, AfterData: map[string]any{"is_current": afterCurrent},
		SubjectVisibility: constants.AuditSubjectInternalOnly,
	})
}

func stateCardID(binding *model.DeviceSimBinding) uint {
	if binding == nil {
		return 0
	}
	return binding.IotCardID
}

func deviceObservationIntegrationResource(ctx context.Context, device *model.Device, integrationID string) audit.ResourceInput {
	deviceID := strconv.FormatUint(uint64(device.ID), 10)
	return audit.ResourceInput{
		Type: constants.AuditResourceIntegrationLog, Key: integrationID, DisplayName: integrationID,
		Relation: constants.AuditResourceRelationReference, Role: constants.AuditResourceRoleWorkerIntegration,
		IdentitySnapshot: map[string]any{
			"integration_id": integrationID, "provider": constants.IntegrationProviderGateway,
			"direction": constants.IntegrationDirectionOutbound, "operation": constants.IntegrationOperationGatewayDeviceInfo,
			"resource_type": constants.CardObservationResourceTypeDevice, "resource_id": deviceID,
			"resource_key": "device:" + deviceID, "correlation_id": auditcontext.From(ctx).CorrelationID,
		},
		SubjectVisibility: constants.AuditSubjectInternalOnly,
	}
}

func (r *SeriesRunner) recordDeviceObservationFailure(ctx context.Context, device *model.Device, integrationID string, businessErr error) {
	deviceID := strconv.FormatUint(uint64(device.ID), 10)
	r.auditWriter.RecordFailure(ctx, r.db, audit.AppendInput{
		ActionCode: constants.AuditActionDeviceWorkerObservationSynced,
		Summary:    "Worker 同步设备观测事实失败", ScopeType: constants.AuditScopePlatform,
		Resources: []audit.ResourceInput{{
			Type: constants.AuditResourceDevice, ID: &deviceID,
			Key: audit.DeviceResourceKey(device), DisplayName: device.VirtualNo,
			Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleDeviceTarget,
			IdentitySnapshot: audit.DeviceIdentitySnapshot(device),
			BeforeData: map[string]any{
				"online_status": device.OnlineStatus, "software_version": device.SoftwareVersion,
				"switch_mode": device.SwitchMode,
			},
			SubjectVisibility: constants.AuditSubjectResult,
		}, deviceObservationIntegrationResource(ctx, device, integrationID)},
	}, businessErr)
}

func (r *SeriesRunner) loadDevice(ctx context.Context, payload cardapp.SeriesTaskPayload) (*model.Device, error) {
	if payload.ResourceType != constants.CardObservationResourceTypeDevice {
		return nil, apperrors.New(apperrors.CodeInvalidParam, "设备信息观测资源类型无效")
	}
	deviceID, err := strconv.ParseUint(payload.ResourceID, 10, 64)
	if err != nil || deviceID == 0 {
		return nil, apperrors.New(apperrors.CodeInvalidParam, "设备观测资源 ID 无效")
	}
	var device model.Device
	if err := r.db.WithContext(ctx).Where("id = ?", uint(deviceID)).First(&device).Error; err != nil {
		if err == gorm.ErrRecordNotFound {
			return nil, apperrors.New(apperrors.CodeNotFound, "设备不存在")
		}
		return nil, apperrors.Wrap(apperrors.CodeDatabaseError, err, "查询设备观测本地快照失败")
	}
	if deviceGatewayIdentifier(&device) == "" {
		return nil, apperrors.New(apperrors.CodeInvalidParam, "设备缺少 Gateway 标识")
	}
	return &device, nil
}

func (r *SeriesRunner) loadCurrentICCIDs(ctx context.Context, payload cardapp.SeriesTaskPayload) ([]string, error) {
	device, err := r.loadDevice(ctx, payload)
	if err != nil {
		return nil, err
	}
	var current struct {
		ICCID   string
		ICCID19 string
		ICCID20 *string
	}
	err = r.db.WithContext(ctx).Table("tb_device_sim_binding AS binding").
		Select("card.iccid, card.iccid_19, card.iccid_20").
		Joins("JOIN tb_iot_card AS card ON card.id = binding.iot_card_id AND card.deleted_at IS NULL").
		Where("binding.device_id = ? AND binding.bind_status = ? AND binding.is_current = ? AND binding.deleted_at IS NULL", device.ID, constants.BindStatusBound, true).
		Take(&current).Error
	if err == gorm.ErrRecordNotFound {
		return nil, nil
	}
	if err != nil {
		return nil, apperrors.Wrap(apperrors.CodeDatabaseError, err, "查询设备当前卡失败")
	}
	iccids := []string{current.ICCID, current.ICCID19}
	if current.ICCID20 != nil {
		iccids = append(iccids, *current.ICCID20)
	}
	return iccids, nil
}

func deviceGatewayIdentifier(device *model.Device) string {
	if strings.TrimSpace(device.IMEI) != "" {
		return strings.TrimSpace(device.IMEI)
	}
	return strings.TrimSpace(device.SN)
}

func parseGatewayTime(raw gateway.FlexString) *time.Time {
	value := strings.TrimSpace(string(raw))
	if value == "" {
		return nil
	}
	if parsed, err := time.Parse(time.RFC3339, value); err == nil {
		return &parsed
	}
	timestamp, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		return nil
	}
	if timestamp > 1_000_000_000_000 {
		timestamp /= 1000
	}
	parsed := time.Unix(timestamp, 0)
	return &parsed
}

func (r *SeriesRunner) startAttempt(ctx context.Context, payload cardapp.SeriesTaskPayload, card *model.IotCard) (*model.IntegrationLog, error) {
	resourceID, source, scene, seriesID := payload.ResourceID, payload.Source, payload.Scene, payload.SeriesID+":"+payload.SyncType
	resourceKey := "card:" + formatCardID(card.ID)
	return r.integration.Start(ctx, integrationlog.Attempt{
		IntegrationID: gatewayIntegrationID(payload), Provider: constants.IntegrationProviderGateway,
		Direction: constants.IntegrationDirectionOutbound, Operation: operationForSyncType(payload.SyncType),
		ResourceType: payload.ResourceType, ResourceID: &resourceID, ResourceKey: &resourceKey,
		TriggerSource: &source, TriggerScene: &scene, TriggerSeries: &seriesID,
		ScheduledAt: &payload.ScheduledAt, Attempt: payload.Attempt,
		RequestSummary: map[string]any{"card_id": card.ID, "sync_type": payload.SyncType},
		RequestID:      optionalText(payload.RequestID), CorrelationID: optionalText(payload.CorrelationID),
	})
}

func (r *SeriesRunner) queryAndApply(ctx context.Context, payload cardapp.SeriesTaskPayload, card *model.IotCard, observationID string) (cardapp.RunResult, error) {
	metadata := carddomain.ObservationMetadata{
		ObservationID: observationID, Source: constants.CardObservationSourceBusinessEvent,
		Scene: payload.Scene, ObservedAt: time.Now().UTC(), RequestID: payload.RequestID,
		CorrelationID: payload.CorrelationID,
	}
	switch payload.SyncType {
	case constants.CardObservationSyncTypeRealname:
		response, err := r.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
		if err != nil {
			return cardapp.RunResult{}, err
		}
		decision, err := r.observation.ApplyCardObservation(ctx, carddomain.RealnameObservation{CardID: card.ID, Verified: response.RealStatus, Metadata: metadata})
		return cardapp.RunResult{StateChanged: decision.StatusChanged}, err
	case constants.CardObservationSyncTypeTraffic:
		response, err := r.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID})
		if err != nil {
			return cardapp.RunResult{}, err
		}
		decision, err := r.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{
			CardID: card.ID, GatewayReadingMB: float64(response.Used), ResetDay: r.carrier.GetDataResetDay(ctx, card.CarrierID), Metadata: metadata,
		})
		return cardapp.RunResult{StateChanged: decision.IncrementMB > 0 || decision.CrossMonth}, err
	case constants.CardObservationSyncTypeNetwork:
		response, err := r.gateway.QueryCardStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
		if err != nil {
			return cardapp.RunResult{}, err
		}
		decision, err := r.observation.ApplyNetworkObservation(ctx, carddomain.NetworkObservation{
			CardID: card.ID, GatewayStatus: response.CardStatus, GatewayExtend: response.Extend,
			GatewayIMEI: response.IMEI, Metadata: metadata,
		})
		return cardapp.RunResult{StateChanged: decision.StatusChanged}, err
	default:
		return cardapp.RunResult{}, apperrors.New(apperrors.CodeInvalidParam, "设备信息观测执行器尚未注册")
	}
}

func (r *SeriesRunner) loadCard(ctx context.Context, payload cardapp.SeriesTaskPayload) (*model.IotCard, error) {
	if payload.ResourceType != constants.CardObservationResourceTypeCard {
		return nil, apperrors.New(apperrors.CodeInvalidParam, "当前观测执行器仅支持 IoT 卡资源")
	}
	cardID, err := strconv.ParseUint(payload.ResourceID, 10, 64)
	if err != nil || cardID == 0 {
		return nil, apperrors.New(apperrors.CodeInvalidParam, "卡观测资源 ID 无效")
	}
	var card model.IotCard
	if err := r.db.WithContext(ctx).Where("id = ?", uint(cardID)).First(&card).Error; err != nil {
		if err == gorm.ErrRecordNotFound {
			return nil, apperrors.New(apperrors.CodeNotFound, "IoT卡不存在")
		}
		return nil, apperrors.Wrap(apperrors.CodeDatabaseError, err, "查询卡观测本地快照失败")
	}
	return &card, nil
}

func isRateLimited(err error) bool {
	var appErr *apperrors.AppError
	if stderrors.As(err, &appErr) && appErr.Code == apperrors.CodeTooManyRequests {
		return true
	}
	return strings.Contains(err.Error(), "429") || strings.Contains(strings.ToLower(err.Error()), "rate limit")
}

func operationForSyncType(syncType string) string {
	switch syncType {
	case constants.CardObservationSyncTypeRealname:
		return constants.IntegrationOperationGatewayRealname
	case constants.CardObservationSyncTypeTraffic:
		return constants.IntegrationOperationGatewayTraffic
	case constants.CardObservationSyncTypeNetwork:
		return constants.IntegrationOperationGatewayNetwork
	default:
		return constants.IntegrationOperationGatewayDeviceInfo
	}
}

func gatewayIntegrationID(payload cardapp.SeriesTaskPayload) string {
	return "card-series:" + payload.SeriesID + ":" + strconv.Itoa(payload.Attempt)
}

func unsentIntegrationID(payload cardapp.SeriesTaskPayload) string {
	return gatewayIntegrationID(payload) + ":unsent"
}

func optionalText(value string) *string {
	value = strings.TrimSpace(value)
	if value == "" {
		return nil
	}
	return &value
}

func initialResult(result string) string {
	if result == constants.IntegrationResultFailed {
		return constants.IntegrationResultPending
	}
	return result
}

var _ cardapp.SeriesAttemptLogger = (*SeriesAttemptLogger)(nil)
var _ cardapp.SeriesRunner = (*SeriesRunner)(nil)