junhong_cmp_fiber/internal/application/carrierthreshold/event.go

package carrierthreshold

import (
	"context"
	stderrors "errors"

	"github.com/bytedance/sonic"
	"github.com/jackc/pgx/v5/pgconn"
	"gorm.io/gorm"

	"github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
	"github.com/break/junhong_cmp_fiber/pkg/outboxid"
)

// EventCarrierThresholdStop 是卡流量达到通道阈值后的可靠停机事件。
const EventCarrierThresholdStop = "carrier.threshold.stop.requested"

// EventCarrierThresholdResume 是通道阈值跨期解锁且条件满足后的可靠复机事件。
const EventCarrierThresholdResume = "carrier.threshold.resume.requested"

// carrierThresholdPayloadVersion 是通道阈值事件的载荷版本。
const carrierThresholdPayloadVersion = 1

// periodLockConstraint 是周期锁部分唯一索引名，用于把 23505 精确识别为该周期已处理。
const periodLockConstraint = "uq_carrier_traffic_threshold_lock_key"

// StopPayload 是通道阈值停机事件的载荷，只携带锁与卡标识，消费者按锁 ID 认领提交权。
type StopPayload struct {
	LockID    uint `json:"lock_id"`
	CardID    uint `json:"card_id"`
	CarrierID uint `json:"carrier_id"`
}

// AppendStopRequested 在达量判定事务内幂等写入通道阈值停机事件。
// 事件 ID 由锁 ID 派生，同一周期锁重复投递不会创建第二个事件（ENG-OUTBOX-001）。
func AppendStopRequested(ctx context.Context, tx *gorm.DB, repository *outbox.Repository, lock *model.CarrierTrafficThresholdLock) error {
	if repository == nil {
		return gorm.ErrInvalidDB
	}
	if lock == nil || lock.ID == 0 {
		return gorm.ErrInvalidData
	}
	value := lockKeyValue(lock.ID)
	_, err := repository.AppendIdempotent(ctx, tx, outbox.Envelope{
		EventID:        outboxid.Stable(EventCarrierThresholdStop+":", value),
		EventType:      EventCarrierThresholdStop,
		PayloadVersion: carrierThresholdPayloadVersion,
		AggregateType:  "carrier_traffic_threshold_lock",
		AggregateID:    value,
		ResourceType:   "iot_card",
		ResourceID:     lockKeyValue(lock.CardID),
		BusinessKey:    EventCarrierThresholdStop + ":" + value,
		Payload: StopPayload{
			LockID:    lock.ID,
			CardID:    lock.CardID,
			CarrierID: lock.CarrierID,
		},
	})
	return err
}

// ResumePayload 是通道阈值复机事件的载荷，只携带锁与卡标识，消费者按锁 ID 认领提交权。
type ResumePayload struct {
	LockID    uint `json:"lock_id"`
	CardID    uint `json:"card_id"`
	CarrierID uint `json:"carrier_id"`
}

// AppendResumeRequested 在周期处理事务内幂等写入通道阈值复机事件。
// 事件 ID 由锁 ID 派生：解锁认领与复机事件同事务写入，重复投递不会创建第二个事件（ENG-OUTBOX-001）。
func AppendResumeRequested(ctx context.Context, tx *gorm.DB, repository *outbox.Repository, lock *model.CarrierTrafficThresholdLock) error {
	if repository == nil {
		return gorm.ErrInvalidDB
	}
	if lock == nil || lock.ID == 0 {
		return gorm.ErrInvalidData
	}
	value := lockKeyValue(lock.ID)
	_, err := repository.AppendIdempotent(ctx, tx, outbox.Envelope{
		EventID:        outboxid.Stable(EventCarrierThresholdResume+":", value),
		EventType:      EventCarrierThresholdResume,
		PayloadVersion: carrierThresholdPayloadVersion,
		AggregateType:  "carrier_traffic_threshold_lock",
		AggregateID:    value,
		ResourceType:   "iot_card",
		ResourceID:     lockKeyValue(lock.CardID),
		BusinessKey:    EventCarrierThresholdResume + ":" + value,
		Payload: ResumePayload{
			LockID:    lock.ID,
			CardID:    lock.CardID,
			CarrierID: lock.CarrierID,
		},
	})
	return err
}

// isPeriodLockConflict 判断错误是否为周期锁唯一键冲突，即「该周期已处理」。
// 该冲突必须与流量基线 CAS 冲突（CodeConflict）区分：前者幂等跳过，后者由调用方重放重试。
func isPeriodLockConflict(err error) bool {
	var pgErr *pgconn.PgError
	if !stderrors.As(err, &pgErr) {
		return false
	}
	return pgErr.Code == "23505" && pgErr.ConstraintName == periodLockConstraint
}

// Consumer 把通道阈值停复机事件转成一次停复机动作。
type Consumer struct {
	service *Service
}

// NewConsumer 创建通道阈值停复机事件消费者。
func NewConsumer(service *Service) *Consumer {
	return &Consumer{service: service}
}

// Consume 按事件类型幂等执行停机或复机；重复投递由锁行认领字段兜住。
func (c *Consumer) Consume(ctx context.Context, envelope outbox.DeliveryEnvelope) error {
	if c == nil || c.service == nil {
		return errors.New(errors.CodeServiceUnavailable, "通道流量阈值停复机执行能力未配置")
	}
	lockID, validationErr := decodeThresholdPayload(envelope)
	if validationErr != nil {
		return validationErr
	}
	ctx = auditcontext.With(ctx, auditcontext.Context{CorrelationID: envelope.CorrelationID, ParentEventID: envelope.EventID})
	switch envelope.EventType {
	case EventCarrierThresholdStop:
		return c.service.ExecuteStop(ctx, lockID)
	case EventCarrierThresholdResume:
		return c.service.ExecuteResume(ctx, lockID)
	default:
		return outbox.Permanent(gorm.ErrInvalidData)
	}
}

// decodeThresholdPayload 校验事件类型与载荷版本并取出锁 ID。
// 载荷不合法属永久失败：重复投递不会改变结果，必须直接终结而不是重试。
func decodeThresholdPayload(envelope outbox.DeliveryEnvelope) (uint, error) {
	if envelope.PayloadVersion != carrierThresholdPayloadVersion {
		return 0, outbox.Permanent(gorm.ErrInvalidData)
	}
	var payload struct {
		LockID uint `json:"lock_id"`
	}
	if err := sonic.Unmarshal(envelope.Payload, &payload); err != nil {
		return 0, outbox.Permanent(err)
	}
	if payload.LockID == 0 {
		return 0, outbox.Permanent(gorm.ErrInvalidData)
	}
	return payload.LockID, nil
}

// 编译期断言：通道阈值停复机消费者满足公共 Outbox 的消费边界。
var _ outbox.EventConsumer = (*Consumer)(nil)