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

package app

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/logx"
	"hyapp/pkg/rocketmqx"
	"hyapp/pkg/walletmq"
	"hyapp/services/wallet-service/internal/config"
	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
)

func (a *App) startWalletOutboxWorkers(ctx context.Context, workerName string, options config.OutboxWorkerConfig, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) {
	if options.Concurrency <= 0 {
		options.Concurrency = 1
	}
	for workerIndex := 1; workerIndex <= options.Concurrency; workerIndex++ {
		workerID := fmt.Sprintf("%s-%s-%02d", workerName, a.nodeID, workerIndex)
		a.workers.Add(1)
		go func() {
			defer a.workers.Done()
			a.runWalletOutboxWorker(ctx, workerID, options, producer, topic, includeEventTypes, excludeEventTypes)
		}()
	}
}

func (a *App) runWalletOutboxWorker(ctx context.Context, workerID string, options config.OutboxWorkerConfig, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) {
	ticker := time.NewTicker(options.PollInterval)
	defer ticker.Stop()
	for {
		processed, err := a.processWalletOutboxBatch(ctx, workerID, options, producer, topic, includeEventTypes, excludeEventTypes)
		if err != nil && !errors.Is(err, context.Canceled) {
			logx.Error(ctx, "wallet_outbox_publish_failed", err, slog.String("worker_id", workerID))
		}
		if processed >= options.BatchSize {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (a *App) processWalletOutboxBatch(ctx context.Context, workerID string, options config.OutboxWorkerConfig, producer *rocketmqx.Producer, topic string, includeEventTypes []string, excludeEventTypes []string) (int, error) {
	if a.mysqlRepo == nil || producer == nil {
		return 0, nil
	}
	records, err := a.mysqlRepo.ClaimPendingWalletOutboxFiltered(ctx, workerID, options.BatchSize, time.Now().UTC().Add(options.PublishTimeout).UnixMilli(), mysqlstorage.WalletOutboxClaimFilter{
		IncludeEventTypes: includeEventTypes,
		ExcludeEventTypes: excludeEventTypes,
	})
	if err != nil {
		return 0, err
	}
	for _, record := range records {
		if record.RetryCount >= options.MaxRetryCount {
			markCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
			err := a.mysqlRepo.MarkWalletOutboxDead(markCtx, record.EventID, deadWalletOutboxReason(record))
			cancel()
			if err != nil {
				return 0, err
			}
			continue
		}
		publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
		err := a.publishWalletOutboxRecord(publishCtx, producer, topic, record)
		cancel()
		if err != nil {
			nextRetryAtMS := time.Now().UTC().Add(walletOutboxBackoff(record.RetryCount+1, options)).UnixMilli()
			markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
			markErr := a.mysqlRepo.MarkWalletOutboxRetryable(markCtx, record.EventID, err.Error(), nextRetryAtMS)
			markCancel()
			if markErr != nil {
				return 0, markErr
			}
			continue
		}
		markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
		markErr := a.mysqlRepo.MarkWalletOutboxDelivered(markCtx, record.EventID)
		markCancel()
		if markErr != nil {
			return 0, markErr
		}
	}
	return len(records), nil
}

func (a *App) publishWalletOutboxRecord(ctx context.Context, producer *rocketmqx.Producer, topic string, record mysqlstorage.WalletOutboxRecord) error {
	message, err := walletOutboxRocketMessage(topic, a.walletEventTagMode, record)
	if err != nil {
		return err
	}
	return producer.SendSync(ctx, message)
}

// walletOutboxRocketMessage keeps the envelope event_type canonical in both phases.
// Phase A deliberately publishes the legacy Tag; Phase B uses that same event_type as
// the Broker routing Tag, avoiding any derived category that could drift from handlers.
func walletOutboxRocketMessage(topic string, tagMode config.WalletEventTagMode, record mysqlstorage.WalletOutboxRecord) (rocketmqx.Message, error) {
	eventTag, err := walletmq.EventTypeTag(record.EventType)
	if err != nil {
		return rocketmqx.Message{}, err
	}
	publishTag := walletmq.TagWalletOutboxEvent
	switch tagMode {
	case "", config.WalletEventTagModeLegacy:
		// 空值属于旧配置或测试构造，必须保持 legacy；只有显式 event_type 才能进入 Phase B。
	case config.WalletEventTagModeEventType:
		publishTag = eventTag
	default:
		return rocketmqx.Message{}, fmt.Errorf("unsupported wallet event tag mode %q", tagMode)
	}
	body, err := walletmq.EncodeWalletOutboxMessage(walletmq.WalletOutboxMessage{
		AppCode:        record.AppCode,
		EventID:        record.EventID,
		EventType:      eventTag,
		TransactionID:  record.TransactionID,
		CommandID:      record.CommandID,
		UserID:         record.UserID,
		AssetType:      record.AssetType,
		AvailableDelta: record.AvailableDelta,
		FrozenDelta:    record.FrozenDelta,
		PayloadJSON:    record.PayloadJSON,
		OccurredAtMS:   record.CreatedAtMS,
	})
	if err != nil {
		return rocketmqx.Message{}, err
	}
	return rocketmqx.Message{
		Topic: topic,
		Tag:   publishTag,
		Keys:  []string{record.EventID, record.TransactionID, record.CommandID},
		Body:  body,
	}, nil
}

func deadWalletOutboxReason(record mysqlstorage.WalletOutboxRecord) string {
	last := strings.TrimSpace(record.LastError)
	if last == "" {
		return fmt.Sprintf("wallet outbox exceeded retry limit: retry_count=%d", record.RetryCount)
	}
	return last
}

func walletOutboxBackoff(retryCount int, options config.OutboxWorkerConfig) time.Duration {
	if retryCount <= 0 {
		return options.InitialBackoff
	}
	backoff := options.InitialBackoff
	for i := 1; i < retryCount; i++ {
		backoff *= 2
		if backoff >= options.MaxBackoff {
			return options.MaxBackoff
		}
	}
	return backoff
}

func (a *App) closeBackgroundWorkers() {
	if a.stopWorker != nil {
		a.stopWorker()
	}
	a.workers.Wait()
}