junhong_cmp_fiber/internal/application/carrierthreshold/execute.go

package carrierthreshold

import (
	"context"

	"go.uber.org/zap"

	domain "github.com/break/junhong_cmp_fiber/internal/domain/carrierthreshold"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
)

// 无法从运营商确认结果时的可安全展示原因（不含内部细节与渠道报文）。
const (
	failedStopReason    = "运营商停机调用失败，等待状态查询确认"
	unknownStopReason   = "运营商停机结果未知，等待状态查询确认"
	failedResumeReason  = "运营商复机调用失败，等待状态查询确认"
	unknownResumeReason = "运营商复机结果未知，等待状态查询确认"
)

// ExecuteStop 执行一次通道阈值达量停机，保证至多一次外部调用。
//
// 流程：提交认领（stop_submitted_at IS NULL 且仍持锁）→ 卡已停机则直接确认成功，不调运营商 →
// 否则复用既有停机重试、Integration Log 与统一审计执行停机。认领失败表示该锁已提交过
// （事件重复投递或人工重放），本次只结束，绝不重复调用；结果由恢复扫描查询收敛。
// 任何分支都不删除锁行、不解锁。
func (s *Service) ExecuteStop(ctx context.Context, lockID uint) error {
	if err := s.requireExecution(); err != nil {
		return err
	}
	lock, err := s.loadLock(ctx, lockID)
	if err != nil {
		return err
	}
	if lock == nil {
		s.logger.Info("通道阈值停机事件对应锁不存在，幂等跳过", zap.Uint("lock_id", lockID))
		return nil
	}
	claimed, err := s.ClaimStopSubmission(ctx, lockID, s.now())
	if err != nil {
		return err
	}
	if !claimed {
		s.logger.Info("通道阈值停机已提交过，只等待恢复扫描确认",
			zap.Uint("lock_id", lockID), zap.String("stop_status", lock.StopStatus))
		return nil
	}
	card, err := s.loadCard(ctx, lock.CardID)
	if err != nil {
		return err
	}
	if card == nil {
		// 卡事实不存在时无法调运营商，保留 submitted 由恢复扫描按窗口标记异常转人工。
		s.logger.Warn("通道阈值停机锁对应卡不存在，等待恢复扫描处理",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID))
		return nil
	}
	if card.NetworkStatus == constants.NetworkStatusOffline {
		// 其他停因已先行停机：不重复调用运营商，直接确认本次停机目标已达成。
		if _, err := s.markTaskConfirmed(ctx, lockID, stopTask, ""); err != nil {
			return err
		}
		s.logger.Info("卡已停机，通道阈值停机直接确认成功",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID), zap.String("stop_reason", card.StopReason))
		return nil
	}
	outcome, callErr := s.commander.StopCardForThreshold(ctx, lock.CardID)
	if callErr != nil && outcome.Result == "" {
		// 基础设施故障（读卡、写审计或写卡状态失败）导致结果无法判定：保留 submitted，
		// 交由恢复扫描查询确认，本次不判定终态、不重复调用。
		s.logger.Error("通道阈值停机执行失败，等待恢复扫描确认",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID), zap.Error(callErr))
		return nil
	}
	if callErr != nil {
		s.logger.Warn("通道阈值停机运营商调用未成功，已按结果分类回填",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID),
			zap.String("result", outcome.Result), zap.Error(callErr))
	}
	_, err = s.markTaskOutcome(ctx, lockID, stopTask, []string{domain.TaskStatusSubmitted},
		taskResultOf(outcome), outcome.IntegrationID, stopFailureReason(outcome))
	if err != nil {
		return err
	}
	s.logger.Info("通道阈值停机任务已回填结果",
		zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID),
		zap.String("result", outcome.Result), zap.String("integration_id", outcome.IntegrationID))
	return nil
}

// ExecuteResume 执行一次通道阈值新周期复机，保证至多一次外部调用。
//
// 流程：提交认领（resume_submitted_at IS NULL）→ 卡已在线则直接确认成功，不调运营商 →
// 否则复用既有复机重试、Integration Log 与统一审计执行复机（成功时同一事务写回卡状态，
// 且只在停因为通道阈值时清除停因）。认领失败表示已提交过，绝不重复调用。
func (s *Service) ExecuteResume(ctx context.Context, lockID uint) error {
	if err := s.requireExecution(); err != nil {
		return err
	}
	lock, err := s.loadLock(ctx, lockID)
	if err != nil {
		return err
	}
	if lock == nil {
		s.logger.Info("通道阈值复机事件对应锁不存在，幂等跳过", zap.Uint("lock_id", lockID))
		return nil
	}
	claimed, err := s.ClaimResumeSubmission(ctx, lockID, s.now())
	if err != nil {
		return err
	}
	if !claimed {
		s.logger.Info("通道阈值复机已提交过，只等待恢复扫描确认",
			zap.Uint("lock_id", lockID), zap.String("resume_status", lock.ResumeStatus))
		return nil
	}
	card, err := s.loadCard(ctx, lock.CardID)
	if err != nil {
		return err
	}
	if card == nil {
		s.logger.Warn("通道阈值复机锁对应卡不存在，等待恢复扫描处理",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID))
		return nil
	}
	if card.NetworkStatus == constants.NetworkStatusOnline {
		if _, err := s.markTaskConfirmed(ctx, lockID, resumeTask, ""); err != nil {
			return err
		}
		s.logger.Info("卡已在线，通道阈值复机直接确认成功",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID))
		return nil
	}
	outcome, callErr := s.commander.ResumeCardForThreshold(ctx, lock.CardID)
	if callErr != nil && outcome.Result == "" {
		s.logger.Error("通道阈值复机执行失败，等待恢复扫描确认",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID), zap.Error(callErr))
		return nil
	}
	if callErr != nil {
		s.logger.Warn("通道阈值复机运营商调用未成功，已按结果分类回填",
			zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID),
			zap.String("result", outcome.Result), zap.Error(callErr))
	}
	_, err = s.markTaskOutcome(ctx, lockID, resumeTask, []string{domain.TaskStatusSubmitted},
		taskResultOf(outcome), outcome.IntegrationID, resumeFailureReason(outcome))
	if err != nil {
		return err
	}
	s.logger.Info("通道阈值复机任务已回填结果",
		zap.Uint("lock_id", lockID), zap.Uint("card_id", lock.CardID),
		zap.String("result", outcome.Result), zap.String("integration_id", outcome.IntegrationID))
	return nil
}

// requireExecution 校验消费者与周期处理所需的端口已配置。
func (s *Service) requireExecution() error {
	if s == nil || s.db == nil || s.repository == nil {
		return errors.New(errors.CodeServiceUnavailable, "通道流量阈值执行能力未完整配置")
	}
	if s.commander == nil {
		return errors.New(errors.CodeServiceUnavailable, "通道流量阈值停复机执行端口未配置")
	}
	return nil
}

// taskResultOf 把运营商调用结果映射为子任务终态。
func taskResultOf(outcome domain.CommandOutcome) string {
	switch outcome.Result {
	case constants.AuditResultSuccess:
		return domain.TaskStatusConfirmed
	case constants.AuditResultUnknown:
		return domain.TaskStatusUnknown
	default:
		return domain.TaskStatusFailed
	}
}

// stopFailureReason 生成停机子任务的可安全失败原因，成功时为空。
func stopFailureReason(outcome domain.CommandOutcome) string {
	if outcome.Confirmed() {
		return ""
	}
	if outcome.Unresolved() {
		return unknownStopReason
	}
	return failedStopReason
}

// resumeFailureReason 生成复机子任务的可安全失败原因，成功时为空。
func resumeFailureReason(outcome domain.CommandOutcome) string {
	if outcome.Confirmed() {
		return ""
	}
	if outcome.Unresolved() {
		return unknownResumeReason
	}
	return failedResumeReason
}