break
b3499adfca
Constraint: 切换 main 前必须保存当前七月分支全部项目进展，套餐生效提案仅属于 Iteration/7-11。 Rejected: 将七月套餐修复直接移植到 main | 两个分支的可靠投递架构不同。 Confidence: medium Scope-risk: broad Directive: 不得将本提交整体 cherry-pick 到 main；main 套餐热修必须基于其纯 Asynq 代码独立实施。 Tested: git diff --check；openspec validate fix-package-activation-starvation --strict。 Not-tested: 按用户要求未运行自动化测试；go build ./... 因当前审计改造中的 Enterprise 模型字面量和 role.recordFailure 参数类型错误未通过。
106 lines
3.8 KiB
Go
106 lines
3.8 KiB
Go
// Package outbox 实现公共 Outbox 的持久化与投递基础设施。
|
|
package outbox
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/bytedance/sonic"
|
|
"github.com/google/uuid"
|
|
"gorm.io/datatypes"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
|
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
|
"github.com/break/junhong_cmp_fiber/pkg/asynctask"
|
|
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
)
|
|
|
|
// Envelope 是业务 UseCase 在事务内追加的公共事件信封。
|
|
type Envelope struct {
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
PayloadVersion int `json:"payload_version"`
|
|
AggregateType string `json:"aggregate_type"`
|
|
AggregateID string `json:"aggregate_id"`
|
|
ResourceType string `json:"resource_type"`
|
|
ResourceID string `json:"resource_id"`
|
|
BusinessKey string `json:"business_key,omitempty"`
|
|
RequestID string `json:"request_id,omitempty"`
|
|
CorrelationID string `json:"correlation_id,omitempty"`
|
|
ParentEventID string `json:"parent_event_id,omitempty"`
|
|
Payload any `json:"payload"`
|
|
}
|
|
|
|
// Repository 通过调用方显式传入的 GORM 事务句柄追加事件。
|
|
type Repository struct{}
|
|
|
|
// NewRepository 创建公共 Outbox Repository。
|
|
func NewRepository() *Repository {
|
|
return &Repository{}
|
|
}
|
|
|
|
// Append 在业务事务中持久化稳定事件身份和必要快照。
|
|
func (r *Repository) Append(ctx context.Context, tx *gorm.DB, envelope Envelope) (*model.OutboxEvent, error) {
|
|
return r.append(ctx, tx, envelope, false)
|
|
}
|
|
|
|
// AppendIdempotent 在稳定 EventID 重投时保持原事件，不重复创建事实。
|
|
func (r *Repository) AppendIdempotent(ctx context.Context, tx *gorm.DB, envelope Envelope) (*model.OutboxEvent, error) {
|
|
return r.append(ctx, tx, envelope, true)
|
|
}
|
|
|
|
func (r *Repository) append(ctx context.Context, tx *gorm.DB, envelope Envelope, idempotent bool) (*model.OutboxEvent, error) {
|
|
if tx == nil {
|
|
return nil, stderrors.New("Outbox 追加必须传入 GORM 事务句柄")
|
|
}
|
|
if strings.TrimSpace(envelope.EventType) == "" || strings.TrimSpace(envelope.AggregateType) == "" ||
|
|
strings.TrimSpace(envelope.AggregateID) == "" || strings.TrimSpace(envelope.ResourceType) == "" ||
|
|
strings.TrimSpace(envelope.ResourceID) == "" {
|
|
return nil, stderrors.New("Outbox 事件类型和资源定位不能为空")
|
|
}
|
|
if err := asynctask.ValidatePayload(envelope.Payload); err != nil {
|
|
return nil, err
|
|
}
|
|
payload, err := sonic.Marshal(envelope.Payload)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if envelope.EventID == "" {
|
|
envelope.EventID = uuid.NewString()
|
|
}
|
|
if envelope.PayloadVersion <= 0 {
|
|
envelope.PayloadVersion = 1
|
|
}
|
|
linkage := auditcontext.From(ctx)
|
|
if envelope.RequestID == "" {
|
|
envelope.RequestID = linkage.RequestID
|
|
}
|
|
if envelope.CorrelationID == "" {
|
|
envelope.CorrelationID = linkage.CorrelationID
|
|
}
|
|
if envelope.ParentEventID == "" {
|
|
envelope.ParentEventID = linkage.ParentEventID
|
|
}
|
|
now := time.Now().UTC()
|
|
event := &model.OutboxEvent{
|
|
EventID: envelope.EventID, EventType: envelope.EventType, PayloadVersion: envelope.PayloadVersion,
|
|
AggregateType: envelope.AggregateType, AggregateID: envelope.AggregateID,
|
|
ResourceType: envelope.ResourceType, ResourceID: envelope.ResourceID, BusinessKey: envelope.BusinessKey,
|
|
RequestID: envelope.RequestID, CorrelationID: envelope.CorrelationID, ParentEventID: envelope.ParentEventID,
|
|
Payload: datatypes.JSON(payload),
|
|
Status: constants.OutboxStatusPending, MaxRetries: constants.OutboxDefaultMaxRetries, NextAttemptAt: now,
|
|
}
|
|
create := tx.WithContext(ctx)
|
|
if idempotent {
|
|
create = create.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "event_id"}}, DoNothing: true})
|
|
}
|
|
if err := create.Create(event).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
return event, nil
|
|
}
|