feat(通道流量阈值): AUG26-011 运营商通道流量阈值达量停机与周期复机

internal/application/carrierthreshold/event.go

@@ -0,0 +1,159 @@
+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)