junhong_cmp_fiber/internal/application/outbox/recovery.go

// 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
}