junhong_cmp_fiber/internal/infrastructure/prioritypolling/event.go

// Package prioritypolling 实现卡轮询优先队列的可靠事件适配器。
//
// 职责分工：本包只做「业务事实 → Outbox 事件」与「Outbox 事件 → 优先项事实 + 执行提示」两侧适配，
// 不承载状态机（权威在优先项事实表）、不决定任务类型集合（权威在 pkg/constants）。
package prioritypolling

import (
	"context"
	"strconv"
	"strings"

	"github.com/bytedance/sonic"
	"go.uber.org/zap"
	"gorm.io/gorm"

	cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
	priorityapp "github.com/break/junhong_cmp_fiber/internal/application/prioritypolling"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox"
	"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"
	"github.com/break/junhong_cmp_fiber/pkg/outboxid"
)

// PriorityEventWriter 将业务成功事实写入公共 Outbox。
type PriorityEventWriter struct {
	outbox *outbox.Repository
}

// NewPriorityEventWriter 创建优先轮询事件 Writer。
func NewPriorityEventWriter(repository *outbox.Repository) *PriorityEventWriter {
	return &PriorityEventWriter{outbox: repository}
}

// AppendPriorityRequested 在原业务事务中追加稳定优先轮询请求事件。
//
// 三个不可省略的约束：
//  1. 必须在调用方的业务事务内追加（tx 非空），由 Outbox 在提交后投递，避免「提交后 goroutine」丢失可靠副作用。
//  2. 观测结果驱动的业务评估必须短路：与既有观测序列写入共用同一抑制标记，避免「观测→业务→优先轮询→观测」反向触发环。
//  3. event_id 必须使用独立命名空间前缀，否则会被 Outbox 全局幂等键静默吞掉。
func (w *PriorityEventWriter) AppendPriorityRequested(ctx context.Context, tx *gorm.DB, event priorityapp.PriorityRequestedEvent) error {
	if w == nil || w.outbox == nil {
		return errors.New(errors.CodeInternalError, "优先轮询 Outbox Writer 未配置")
	}
	if cardObservationApp.IsSeriesTriggerSuppressed(ctx) {
		return nil
	}
	if tx == nil {
		return errors.New(errors.CodeInternalError, "优先轮询事件缺少原业务事务")
	}
	if strings.TrimSpace(event.EventID) == "" || event.ResourceID == 0 || strings.TrimSpace(event.TriggerType) == "" {
		return errors.New(errors.CodeInvalidParam, "优先轮询事件载荷不完整")
	}
	if !strings.HasPrefix(event.EventID, constants.PollingPriorityEventKeyPrefix) {
		return errors.New(errors.CodeInvalidParam,
			"优先轮询事件 ID 必须使用 "+constants.PollingPriorityEventKeyPrefix+" 命名空间前缀")
	}
	// 长度契约：event_id 列上限 64，超限会让写入失败并连带回滚触发事务（如套餐轮询进失败分支）。
	// 因此在此提前按既有 Outbox 标识契约校验，把数据库层失败变成可定位的参数错误。
	if err := outboxid.Validate(event.EventID, ""); err != nil {
		return errors.Wrap(errors.CodeInvalidParam, err, "优先轮询事件 ID 超长")
	}
	if event.ResourceType != constants.AssetTypeIotCard && event.ResourceType != constants.AssetTypeDevice {
		return errors.New(errors.CodeInvalidParam, "优先轮询事件资源类型不受支持")
	}
	// 人工入队不经事件通道：manual_trigger 由 internal/service/polling/priority_manual_service.go 直写事实表，
	// 事件通道没有操作者，无法满足事实表 CHECK (trigger_type <> 'manual_trigger' OR manual_operator_id > 0)。
	// 权威约束在事实表 CHECK，这里只是把「DB 23514 → 整个投递事务回滚」提前为可定位的参数错误，
	// 避免调用方业务事务被数据库层失败牵连回滚。
	if event.TriggerType == constants.PollingPriorityTriggerManual {
		return errors.New(errors.CodeInvalidParam,
			"优先轮询事件不接受人工入队触发类型 "+constants.PollingPriorityTriggerManual+"，人工入队请调用人工入队用例")
	}

	// 冻结卡快照：独立卡为自身，绑定设备的资产为绑定状态有效的全部在用卡。
	// 快照在业务事务内写入，触发提交后绑定关系或当前卡槽变化都不追溯。
	if event.ResourceType == constants.AssetTypeDevice {
		event.ResourceIDs = nil
		if err := tx.WithContext(ctx).
			Model(&model.DeviceSimBinding{}).
			Where("device_id = ? AND bind_status = ?", event.ResourceID, constants.BindStatusBound).
			Order("slot_position ASC").
			Pluck("iot_card_id", &event.ResourceIDs).Error; err != nil {
			return errors.Wrap(errors.CodeDatabaseError, err, "冻结优先轮询设备绑定卡快照失败")
		}
	}

	if _, err := w.outbox.AppendIdempotent(ctx, tx, outbox.Envelope{
		EventID:        event.EventID,
		EventType:      constants.OutboxEventTypePollingPriorityRequested,
		PayloadVersion: constants.PollingPriorityRequestedPayloadVersionV1,
		AggregateType:  event.ResourceType,
		AggregateID:    strconv.FormatUint(uint64(event.ResourceID), 10),
		ResourceType:   event.ResourceType,
		ResourceID:     strconv.FormatUint(uint64(event.ResourceID), 10),
		BusinessKey:    event.EventID,
		RequestID:      event.RequestID,
		CorrelationID:  event.CorrelationID,
		Payload:        event,
	}); err != nil {
		return errors.Wrap(errors.CodeDatabaseError, err, "写入优先轮询 Outbox 事件失败")
	}
	return nil
}

// PriorityRequestedConsumer 消费业务成功事实，按卡 × 纳入任务类型建立或合并优先项并下发执行提示。
type PriorityRequestedConsumer struct {
	db        *gorm.DB
	store     *postgres.PollingPriorityItemStore
	publisher priorityapp.PromptPublisher
	audit     *audit.Writer
	logger    *zap.Logger
}

// NewPriorityRequestedConsumer 创建优先轮询请求消费者。
func NewPriorityRequestedConsumer(
	db *gorm.DB,
	store *postgres.PollingPriorityItemStore,
	publisher priorityapp.PromptPublisher,
	auditWriter *audit.Writer,
	logger *zap.Logger,
) *PriorityRequestedConsumer {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &PriorityRequestedConsumer{db: db, store: store, publisher: publisher, audit: auditWriter, logger: logger}
}

// Consume 将一个业务事件展开为逐卡逐任务类型的优先项；重复投递只合并，不新建行也不重复调用。
func (c *PriorityRequestedConsumer) Consume(ctx context.Context, envelope outbox.DeliveryEnvelope) error {
	if c == nil || c.db == nil || c.store == nil || c.publisher == nil {
		return errors.New(errors.CodeInternalError, "优先轮询消费者未配置")
	}
	if envelope.EventType != constants.OutboxEventTypePollingPriorityRequested ||
		envelope.PayloadVersion != constants.PollingPriorityRequestedPayloadVersionV1 {
		return errors.New(errors.CodeInvalidParam, "优先轮询事件类型或版本不受支持")
	}
	var event priorityapp.PriorityRequestedEvent
	if err := sonic.Unmarshal(envelope.Payload, &event); err != nil {
		return errors.Wrap(errors.CodeInvalidParam, err, "优先轮询事件载荷无法解析")
	}
	if event.EventID != envelope.EventID || event.ResourceID == 0 || strings.TrimSpace(event.TriggerType) == "" {
		return errors.New(errors.CodeInvalidParam, "优先轮询事件载荷不完整")
	}
	if event.ResourceType != constants.AssetTypeIotCard && event.ResourceType != constants.AssetTypeDevice {
		return errors.New(errors.CodeInvalidParam, "优先轮询事件资源类型不受支持")
	}
	cardIDs := []uint{event.ResourceID}
	if event.ResourceType == constants.AssetTypeDevice {
		cardIDs = event.ResourceIDs
	}
	// 设备在触发事务内没有绑定卡：没有执行对象，事件成功结束且不重试。
	// 记一条 Info 以便外部区分「触发已发生但无卡」与「触发未发生」。
	if len(cardIDs) == 0 {
		c.logger.Info("优先轮询触发未解析到绑定卡，事件按无执行对象结束",
			zap.String("event_id", event.EventID),
			zap.String("trigger_type", event.TriggerType),
			zap.String("resource_type", event.ResourceType),
			zap.Uint("resource_id", event.ResourceID))
		return nil
	}

	taskTypes := constants.PollingPriorityTaskTypes()
	createdCount, mergedCount := 0, 0
	err := c.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		shopSnapshots, lookupErr := loadCardShopSnapshots(ctx, tx, cardIDs)
		if lookupErr != nil {
			return lookupErr
		}
		store := c.store.WithTx(tx)
		assetSnapshots, assetErr := store.AssetSnapshots(ctx, cardIDs)
		if assetErr != nil {
			return assetErr
		}
		createdCount, mergedCount = 0, 0
		for _, cardID := range cardIDs {
			for _, taskType := range taskTypes {
				asset := assetSnapshots[cardID]
				item := &model.PollingPriorityItem{
					CardID:               cardID,
					TaskType:             taskType,
					TriggerType:          event.TriggerType,
					SourceOrderID:        optionalUintPointer(event.SourceOrderID),
					SourcePackageUsageID: optionalUintPointer(event.SourcePackageUsageID),
					ShopIDSnapshot:       shopSnapshots[cardID],
					AssetType:            asset.AssetType,
					AssetID:              optionalUintPointer(asset.AssetID),
					DeviceNoSnapshot:     asset.DeviceNo,
					AgentShopIDSnapshot:  asset.AgentShopID,
					LastTriggeredAt:      event.OccurredAt,
				}
				created, insertErr := store.InsertOrMerge(ctx, item)
				if insertErr != nil {
					return insertErr
				}
				if created {
					createdCount++
					continue
				}
				mergedCount++
			}
		}
		// 入队事实与审计同事务：任一逐卡逐类型写入失败都整体回滚，由 Outbox 重投，不存在部分静默成功。
		return c.writeEnqueueAudit(ctx, tx, envelope.EventID, event, len(cardIDs), taskTypes, createdCount, mergedCount)
	})
	if err != nil {
		return err
	}

	// 提交后再下发执行提示。提示通道不是权威，但下发失败仍返回错误交 Outbox 重投：
	// 合并语义使重投幂等，且无人接收的提示只会让加急退化为延迟一个普通轮询周期，
	// 因此不能用「吞掉错误」的方式掩盖通道故障。
	for _, cardID := range cardIDs {
		for _, taskType := range taskTypes {
			if publishErr := c.publisher.EnqueuePriority(ctx, cardID, taskType); publishErr != nil {
				return errors.Wrap(errors.CodeInternalError, publishErr, "下发优先轮询执行提示失败")
			}
		}
	}
	return nil
}

// writeEnqueueAudit 写入一条入队事实审计（自动触发与人工入队共用同一动作）。
func (c *PriorityRequestedConsumer) writeEnqueueAudit(
	ctx context.Context,
	tx *gorm.DB,
	eventID string,
	event priorityapp.PriorityRequestedEvent,
	cardCount int,
	taskTypes []string,
	createdCount, mergedCount int,
) error {
	afterData := map[string]any{
		"trigger_type":            event.TriggerType,
		"source_order_id":         event.SourceOrderID,
		"source_package_usage_id": event.SourcePackageUsageID,
		"resource_type":           event.ResourceType,
		"resource_id":             event.ResourceID,
		"card_count":              cardCount,
		"task_types":              taskTypes,
		"created_count":           createdCount,
		"merged_count":            mergedCount,
		"source":                  constants.AuditSourceWorker,
		"occurred_at":             event.OccurredAt,
	}
	return c.audit.WriteTask(ctx, tx, audit.TaskInput{
		// 与 Outbox 事件同 ID：至少一次重投时审计按 event_id 幂等去重，不产生重复审计行。
		EventID:    eventID,
		ActionCode: constants.AuditActionPollingPriorityEnqueued,
		Summary:    "卡轮询优先项入队",
		TaskID:     0, TaskNo: eventID, DisplayName: "卡轮询优先项入队",
		Actor: audit.ActorInput{
			Kind: constants.AuditActorSystemTask,
			ID:   constants.OutboxEventTypePollingPriorityRequested,
			Name: "优先轮询请求消费任务",
		},
		Source:        constants.AuditSourceWorker,
		ScopeType:     constants.AuditScopePlatform,
		Result:        constants.AuditResultSuccess,
		CorrelationID: event.CorrelationID,
		AfterData:     afterData,
		Metadata: map[string]any{
			"trigger_type":            event.TriggerType,
			"source_order_id":         event.SourceOrderID,
			"source_package_usage_id": event.SourcePackageUsageID,
			"created_count":           createdCount,
			"merged_count":            mergedCount,
		},
	})
}

// loadCardShopSnapshots 读取卡的店铺快照：平台卡返回 nil，读侧据此把平台卡与空范围一并挡在代理之外。
// 快照取消费时点的卡归属，卡改归属不追溯已入队的加急事实；已软删的卡不产生快照，
// 其优先项会在执行时以「卡不存在」进入失败终态。
func loadCardShopSnapshots(ctx context.Context, tx *gorm.DB, cardIDs []uint) (map[uint]*uint, error) {
	type cardShopRow struct {
		ID     uint  `gorm:"column:id"`
		ShopID *uint `gorm:"column:shop_id"`
	}
	var rows []cardShopRow
	if err := tx.WithContext(ctx).
		Model(&model.IotCard{}).
		Select("id, shop_id").
		Where("id IN ? AND deleted_at IS NULL", cardIDs).
		Scan(&rows).Error; err != nil {
		return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询优先轮询卡店铺快照失败")
	}
	snapshots := make(map[uint]*uint, len(rows))
	for _, row := range rows {
		snapshots[row.ID] = row.ShopID
	}
	return snapshots, nil
}

// optionalUintPointer 把 0 视为「无来源」，避免用哨兵 0 占用可空来源列。
func optionalUintPointer(value uint) *uint {
	if value == 0 {
		return nil
	}
	return &value
}