junhong_cmp_fiber/internal/service/polling/priority_expiry_service.go

package polling

import (
	"context"
	"time"

	"go.uber.org/zap"
	"gorm.io/gorm"

	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
)

// pollingPriorityExpiryBatchSize 是有效期到期单轮扫描的批量上限。
// 只限制每轮处理量，不改变语义：剩余的超期项由下一轮计划任务继续终结。
const pollingPriorityExpiryBatchSize = 200

// PriorityExpiryService 优先轮询项有效期到期处理用例。
//
// 由独立周期任务承载（查询与读侧 MUST NOT 隐式触发）：按固定有效期长度扫描有效期起算时间
// 已超期的未完成项，逐条以条件更新转已过期终态并出队；终态写入与审计同事务。
// 触发类型、尝试次数与最近失败原因全部保留，不删除任何既有事实，也不发起任何上游调用。
type PriorityExpiryService struct {
	db            *gorm.DB
	priorityStore *postgres.PollingPriorityItemStore
	auditWriter   *audit.Writer
	logger        *zap.Logger
}

// NewPriorityExpiryService 创建优先轮询项有效期到期处理用例。
func NewPriorityExpiryService(
	db *gorm.DB,
	priorityStore *postgres.PollingPriorityItemStore,
	logger *zap.Logger,
) *PriorityExpiryService {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &PriorityExpiryService{db: db, priorityStore: priorityStore, logger: logger}
}

// SetAudit 注入统一审计 Writer。
func (s *PriorityExpiryService) SetAudit(writer *audit.Writer) {
	s.auditWriter = writer
}

// ExpireOverdue 终结所有已超过固定有效期且仍未完成的优先项，返回本轮实际终结的条数。
// 逐条独立事务：单条失败不阻断其余超期项，失败原因由调用方按任务错误处理并重试。
func (s *PriorityExpiryService) ExpireOverdue(ctx context.Context, now time.Time) (int, error) {
	if s == nil || s.db == nil || s.priorityStore == nil {
		return 0, errors.New(errors.CodeServiceUnavailable, "优先轮询项有效期到期处理未配置")
	}
	deadline := now.Add(-constants.PollingPriorityValidity)
	items, err := s.priorityStore.ListOverdueActive(ctx, deadline, pollingPriorityExpiryBatchSize)
	if err != nil {
		return 0, err
	}
	if len(items) == 0 {
		return 0, nil
	}

	expiredCount := 0
	for _, item := range items {
		if item == nil {
			continue
		}
		marked := false
		err := runPollingTransaction(ctx, s.db, s.auditWriter, func(tx *gorm.DB) error {
			hit, markErr := s.priorityStore.WithTx(tx).MarkExpired(ctx, item.ID)
			if markErr != nil {
				return markErr
			}
			if !hit {
				// 并发下该项已被执行终结或人工关闭：保留既有结论，不覆盖也不重复审计。
				return nil
			}
			marked = true
			return s.writeExpiryAudit(ctx, tx, item, now)
		})
		if err != nil {
			return expiredCount, err
		}
		if marked {
			expiredCount++
		}
	}
	if expiredCount > 0 {
		s.logger.Info("优先轮询项有效期到期处理完成",
			zap.Int("expired_count", expiredCount), zap.Int("scanned_count", len(items)))
	}
	return expiredCount, nil
}

// writeExpiryAudit 以系统任务身份写入一条超期出队审计。
// 计划任务没有后台账号，因此走任务审计形态（操作者=计划任务），而不是要求账号操作者的后台入口形态。
func (s *PriorityExpiryService) writeExpiryAudit(ctx context.Context, tx *gorm.DB, item *model.PollingPriorityItem, now time.Time) error {
	if s == nil || s.auditWriter == nil {
		return errors.New(errors.CodeInvalidStatus, "统一审计 Writer 未配置")
	}
	return s.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
		ActionCode:  constants.AuditActionPollingPriorityExpired,
		Summary:     "卡轮询优先项有效期到期出队",
		TaskID:      item.ID,
		TaskNo:      pollingPriorityResourceKey(item.ID),
		DisplayName: "卡轮询优先项",
		Actor: audit.ActorInput{
			Kind: constants.AuditActorSystemTask,
			ID:   constants.TaskTypePollingPriorityExpiry,
			Name: "优先轮询项有效期到期处理任务",
		},
		Source:        constants.AuditSourceWorker,
		ScopeType:     constants.AuditScopePlatform,
		Result:        constants.AuditResultSuccess,
		CorrelationID: constants.TaskTypePollingPriorityExpiry,
		IdentitySnapshot: map[string]any{
			"id": item.ID, "card_id": item.CardID, "task_type": item.TaskType,
			"status": item.Status, "trigger_type": item.TriggerType,
			"attempt_count": item.AttemptCount,
		},
		BeforeData: map[string]any{
			"status": item.Status, "result": item.Result, "failure_reason": item.FailureReason,
			"attempt_count": item.AttemptCount, "next_run_at": item.NextRunAt,
		},
		AfterData: map[string]any{
			"status": constants.PollingPriorityStatusExpired, "dequeued_at": now,
			"priority_effective_from": item.PriorityEffectiveFrom,
			"expired_at":              now,
		},
		Metadata: map[string]any{
			"card_id": item.CardID, "task_type": item.TaskType,
			"priority_effective_from": item.PriorityEffectiveFrom,
		},
	})
}