// Package outbox 提供公共 Outbox 的受控人工恢复用例。 package outbox import ( "context" stderrors "errors" "strconv" "time" "github.com/google/uuid" "gorm.io/gorm" "gorm.io/gorm/clause" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/pkg/auditfailure" "github.com/break/junhong_cmp_fiber/pkg/constants" pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors" ) // Operator 表示人工恢复操作者的授权快照。 type Operator struct { ID uint SuperAdmin bool RequestID string CorrelationID string } // RecoveryAudit 是交给统一 Audit Port 的安全恢复事实。 type RecoveryAudit struct { OperatorID uint OperationType string Description string EventIDs []string Reason string BatchID string RequestID string CorrelationID string Events []RecoveryEventAudit Result string ErrorCode string ErrorSummary string } // RecoveryEventAudit 是一次人工恢复中单个 Outbox 事件的身份和状态变化。 type RecoveryEventAudit struct { ID uint EventID string EventType string AggregateType string AggregateID string ResourceType string ResourceID string BusinessKey string BeforeStatus int AfterStatus int BeforeNextAttempt time.Time AfterNextAttempt time.Time BeforeLeaseOwner *string BeforeLeaseExpires *time.Time AfterLeaseOwner *string AfterLeaseExpires *time.Time } // AuditWriter 是 tech-global-audit 提供实现的统一审计接缝。 type AuditWriter interface { WriteRecovery(ctx context.Context, tx *gorm.DB, audit RecoveryAudit) error } // RecoveryService 执行选择性重放和过期租约释放。 type RecoveryService struct { db *gorm.DB audit AuditWriter now func() time.Time } // NewRecoveryService 创建受控恢复用例;审计接缝不可缺失。 func NewRecoveryService(db *gorm.DB, audit AuditWriter, now func() time.Time) (*RecoveryService, error) { if db == nil || audit == nil { return nil, stderrors.New("Outbox 恢复必须配置数据库和统一审计接缝") } if now == nil { now = time.Now } return &RecoveryService{db: db, audit: audit, now: now}, nil } // Replay 只重放明确选择的最终失败或租约过期事件,并保留原始内容和身份。 func (s *RecoveryService) Replay(ctx context.Context, operator Operator, ids []uint, reason string) (string, error) { if err := validateCommand(operator, ids, reason); err != nil { return "", err } batchID := uuid.NewString() now := s.now().UTC() failureAudit := RecoveryAudit{ OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReplay, Description: "人工重放 Outbox 事件失败", Reason: reason, BatchID: batchID, RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, } err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { events, err := loadSelectedForUpdate(tx, ids) if err != nil { return err } failureAudit.EventIDs, failureAudit.Events = unchangedRecoveryAudit(events) if len(events) != len(ids) { failureAudit.Events = nil return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的事件不存在或状态不允许重放") } eventIDs := make([]string, 0, len(events)) auditEvents := make([]RecoveryEventAudit, 0, len(events)) for _, event := range events { allowed := event.Status == constants.OutboxStatusFailed || (event.Status == constants.OutboxStatusDelivering && event.LeaseExpiresAt != nil && !event.LeaseExpiresAt.After(now)) if !allowed { failureAudit.Result = constants.AuditResultDenied failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeInvalidStatus) failureAudit.ErrorSummary = "选择的事件不存在或状态不允许重放" return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的事件不存在或状态不允许重放") } eventIDs = append(eventIDs, event.EventID) auditEvents = append(auditEvents, recoveryEventAudit(event, now)) } failureAudit.Result = constants.AuditResultFailed failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeDatabaseError) failureAudit.ErrorSummary = "Outbox 人工重放事务已回滚" result := tx.Model(&model.OutboxEvent{}).Where("id IN ?", ids).Updates(map[string]any{ "status": constants.OutboxStatusPending, "next_attempt_at": now, "lease_owner": nil, "lease_expires_at": nil, "updated_at": now, }) if result.Error != nil { return result.Error } return s.audit.WriteRecovery(ctx, tx, RecoveryAudit{ OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReplay, Description: "人工重放 Outbox 事件", EventIDs: eventIDs, Reason: reason, BatchID: batchID, RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, Events: auditEvents, }) }) if err != nil { s.recordFailure(ctx, failureAudit) } return batchID, err } // ReleaseExpiredLeases 只释放明确选择且已经过期的投递租约。 func (s *RecoveryService) ReleaseExpiredLeases(ctx context.Context, operator Operator, ids []uint, reason string) (string, error) { if err := validateCommand(operator, ids, reason); err != nil { return "", err } batchID := uuid.NewString() now := s.now().UTC() failureAudit := RecoveryAudit{ OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReleaseExpiredLease, Description: "人工释放 Outbox 过期租约失败", Reason: reason, BatchID: batchID, RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, } err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { events, err := loadSelectedForUpdate(tx, ids) if err != nil { return err } failureAudit.EventIDs, failureAudit.Events = unchangedRecoveryAudit(events) if len(events) != len(ids) { failureAudit.Events = nil return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的租约不存在或仍然有效") } eventIDs := make([]string, 0, len(events)) auditEvents := make([]RecoveryEventAudit, 0, len(events)) for _, event := range events { if event.Status != constants.OutboxStatusDelivering || event.LeaseExpiresAt == nil || event.LeaseExpiresAt.After(now) { failureAudit.Result = constants.AuditResultDenied failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeInvalidStatus) failureAudit.ErrorSummary = "选择的租约不存在或仍然有效" return pkgerrors.New(pkgerrors.CodeInvalidStatus, "选择的租约不存在或仍然有效") } eventIDs = append(eventIDs, event.EventID) auditEvents = append(auditEvents, recoveryEventAudit(event, now)) } failureAudit.Result = constants.AuditResultFailed failureAudit.ErrorCode = strconv.Itoa(pkgerrors.CodeDatabaseError) failureAudit.ErrorSummary = "Outbox 过期租约释放事务已回滚" result := tx.Model(&model.OutboxEvent{}).Where("id IN ? AND status = ? AND lease_expires_at <= ?", ids, constants.OutboxStatusDelivering, now). Updates(map[string]any{ "status": constants.OutboxStatusPending, "next_attempt_at": now, "lease_owner": nil, "lease_expires_at": nil, "updated_at": now, }) if result.Error != nil { return result.Error } return s.audit.WriteRecovery(ctx, tx, RecoveryAudit{ OperatorID: operator.ID, OperationType: constants.AuditOperationOutboxReleaseExpiredLease, Description: "人工释放 Outbox 过期租约", EventIDs: eventIDs, Reason: reason, BatchID: batchID, RequestID: operator.RequestID, CorrelationID: operator.CorrelationID, Events: auditEvents, }) }) if err != nil { s.recordFailure(ctx, failureAudit) } return batchID, err } func (s *RecoveryService) recordFailure(ctx context.Context, audit RecoveryAudit) { if len(audit.Events) == 0 || audit.Result == "" { return } err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { return s.audit.WriteRecovery(ctx, tx, audit) }) if err != nil { auditfailure.RecordSecondaryWriteFailure( audit.OperationType, audit.Events[0].EventID, audit.RequestID, audit.CorrelationID, audit.ErrorCode, err, ) } } func validateCommand(operator Operator, ids []uint, reason string) error { if !operator.SuperAdmin { return pkgerrors.New(pkgerrors.CodeForbidden) } if operator.ID == 0 || len(ids) == 0 || reason == "" { return pkgerrors.New(pkgerrors.CodeInvalidParam) } return nil } func loadSelectedForUpdate(tx *gorm.DB, ids []uint) ([]model.OutboxEvent, error) { var events []model.OutboxEvent err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id IN ?", ids).Order("id ASC").Find(&events).Error return events, err } func recoveryEventAudit(event model.OutboxEvent, nextAttempt time.Time) RecoveryEventAudit { return RecoveryEventAudit{ ID: event.ID, EventID: event.EventID, EventType: event.EventType, AggregateType: event.AggregateType, AggregateID: event.AggregateID, ResourceType: event.ResourceType, ResourceID: event.ResourceID, BusinessKey: event.BusinessKey, BeforeStatus: event.Status, AfterStatus: constants.OutboxStatusPending, BeforeNextAttempt: event.NextAttemptAt, AfterNextAttempt: nextAttempt, BeforeLeaseOwner: event.LeaseOwner, BeforeLeaseExpires: event.LeaseExpiresAt, } } func unchangedRecoveryAudit(events []model.OutboxEvent) ([]string, []RecoveryEventAudit) { eventIDs := make([]string, 0, len(events)) auditEvents := make([]RecoveryEventAudit, 0, len(events)) for _, event := range events { eventIDs = append(eventIDs, event.EventID) auditEvent := recoveryEventAudit(event, event.NextAttemptAt) auditEvent.AfterStatus = event.Status auditEvent.AfterLeaseOwner = event.LeaseOwner auditEvent.AfterLeaseExpires = event.LeaseExpiresAt auditEvents = append(auditEvents, auditEvent) } return eventIDs, auditEvents }