hyapp-server/services/wallet-service/internal/app/outbox_replay.go

package app

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/outboxreplay"
	"hyapp/pkg/rocketmqx"
	servicemq "hyapp/pkg/servicekit/mq"
	"hyapp/pkg/walletmq"
	"hyapp/services/wallet-service/internal/config"
	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
)

// ReplayWalletOutboxOnce publishes at most scope.Limit claimable facts from
// exactly one App partition, then closes both MySQL and MQ. Each iteration uses
// the production claim/publish/mark path with batch size one, so a process
// timeout cannot leave a claimed tail waiting for lease expiry.
func ReplayWalletOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope, eventTypes []string) (outboxreplay.Result, error) {
	validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500)
	if err != nil {
		return outboxreplay.Result{}, err
	}
	includeEventTypes, err := ValidateWalletReplayEventTypes(cfg, eventTypes)
	if err != nil {
		return outboxreplay.Result{}, err
	}
	if !cfg.RocketMQ.WalletOutbox.Enabled {
		return outboxreplay.Result{}, errors.New("wallet RocketMQ outbox is not enabled")
	}

	repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
	if err != nil {
		return outboxreplay.Result{}, err
	}
	defer func() { _ = repository.Close() }()
	producer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.ProducerGroup))
	if err != nil {
		return outboxreplay.Result{}, err
	}
	if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil {
		return outboxreplay.Result{}, err
	}
	defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer})

	workerOptions := cfg.OutboxWorker
	workerOptions.BatchSize = 1
	workerOptions.Concurrency = 1
	replayApp := &App{
		mysqlRepo:          repository,
		walletEventTagMode: cfg.RocketMQ.WalletEventTagMode,
		walletOutboxTopic:  cfg.RocketMQ.WalletOutbox.Topic,
		outboxWorkerCfg:    workerOptions,
	}
	workerID := fmt.Sprintf("ops-wallet-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
	return outboxreplay.Run(ctx, validatedScope, func(runCtx context.Context) (int, error) {
		return replayApp.replayOneWalletOutboxRecord(runCtx, workerID, workerOptions, producer, cfg.RocketMQ.WalletOutbox.Topic, validatedScope.AppCode, includeEventTypes)
	})
}

func (a *App) replayOneWalletOutboxRecord(ctx context.Context, workerID string, options config.OutboxWorkerConfig, producer *rocketmqx.Producer, topic string, targetAppCode string, includeEventTypes []string) (int, error) {
	if options.PublishTimeout <= 0 {
		options.PublishTimeout = 3 * time.Second
	}
	claimCtx := appcode.WithContext(ctx, targetAppCode)
	records, err := a.mysqlRepo.ClaimPendingWalletOutboxFiltered(
		claimCtx,
		workerID,
		1,
		time.Now().UTC().Add(walletOutboxSingleRecordLease(options.PublishTimeout)).UnixMilli(),
		mysqlstorage.WalletOutboxClaimFilter{IncludeEventTypes: includeEventTypes},
	)
	if err != nil || len(records) == 0 {
		return 0, err
	}
	record := records[0]
	if record.RetryCount >= options.MaxRetryCount {
		markCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
		markErr := a.mysqlRepo.MarkWalletOutboxDead(markCtx, record.EventID, deadWalletOutboxReason(record))
		cancel()
		if markErr != nil {
			return 1, markErr
		}
		return 1, fmt.Errorf("wallet outbox event %s reached max retry count and was marked dead", record.EventID)
	}

	publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
	publishErr := a.publishWalletOutboxRecord(publishCtx, producer, topic, record)
	cancel()
	markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
	defer markCancel()
	if publishErr != nil {
		nextRetryAtMS := time.Now().UTC().Add(walletOutboxBackoff(record.RetryCount+1, options)).UnixMilli()
		if markErr := a.mysqlRepo.MarkWalletOutboxRetryable(markCtx, record.EventID, publishErr.Error(), nextRetryAtMS); markErr != nil {
			return 1, errors.Join(publishErr, markErr)
		}
		// The row was safely released for a later retry, but this explicit run must
		// fail instead of reporting a merely claimed row as a delivered replay.
		return 1, publishErr
	}
	if err := a.mysqlRepo.MarkWalletOutboxDelivered(markCtx, record.EventID); err != nil {
		return 1, err
	}
	return 1, nil
}

// NormalizeWalletReplayEventTypes validates the exact MQ tags before an
// operator dry-run is promoted to an executing run.
func NormalizeWalletReplayEventTypes(values []string) ([]string, error) {
	if len(values) > 32 {
		return nil, errors.New("wallet event-type count cannot exceed 32")
	}
	normalized := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		eventType, err := walletmq.EventTypeTag(strings.TrimSpace(value))
		if err != nil {
			return nil, err
		}
		if _, exists := seen[eventType]; exists {
			continue
		}
		seen[eventType] = struct{}{}
		normalized = append(normalized, eventType)
	}
	return normalized, nil
}

// ValidateWalletReplayEventTypes applies the lane boundary in both dry-run and
// execute modes. The normal-topic repair must reject realtime event types even
// while that worker or topic is temporarily disabled; pausing a lane does not
// transfer ownership of its durable rows to another topic.
func ValidateWalletReplayEventTypes(cfg config.Config, values []string) ([]string, error) {
	normalized, err := NormalizeWalletReplayEventTypes(values)
	if err != nil {
		return nil, err
	}
	if len(normalized) == 0 {
		// Wallet fanout has both high-volume balance notices and low-volume config
		// facts. Requiring an explicit set prevents a small operational command
		// from accidentally mixing stages merely because their timestamps overlap.
		return nil, errors.New("at least one wallet event-type is required")
	}
	if err := rejectRealtimeReplayEventTypes(normalized, cfg.RealtimeOutboxWorker.EventTypes); err != nil {
		return nil, err
	}
	return normalized, nil
}

func rejectRealtimeReplayEventTypes(included []string, realtime []string) error {
	realtimeSet := make(map[string]struct{}, len(realtime))
	for _, eventType := range realtime {
		eventType = strings.TrimSpace(eventType)
		if eventType != "" {
			realtimeSet[eventType] = struct{}{}
		}
	}
	for _, eventType := range included {
		if _, exists := realtimeSet[eventType]; exists {
			return fmt.Errorf("wallet event-type %s belongs to the realtime outbox lane", eventType)
		}
	}
	return nil
}