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
	}
	processed := 0
	for processed < options.BatchSize {
		// Sync publish is intentionally claimed one record at a time. Claiming an entire
		// batch with one short lock lets later rows expire while earlier SendSync calls
		// are still running, so another replica can publish the same tail concurrently.
		records, err := a.mysqlRepo.ClaimPendingWalletOutboxFiltered(ctx, workerID, 1, time.Now().UTC().Add(walletOutboxSingleRecordLease(options.PublishTimeout)).UnixMilli(), mysqlstorage.WalletOutboxClaimFilter{
			IncludeEventTypes: includeEventTypes,
			ExcludeEventTypes: excludeEventTypes,
		})
		if err != nil {
			return processed, err
		}
		if len(records) == 0 {
			return processed, nil
		}
		record := records[0]
		processed++
		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 processed, 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 processed, 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 processed, markErr
		}
	}
	return processed, nil
}

func walletOutboxSingleRecordLease(publishTimeout time.Duration) time.Duration {
	if publishTimeout <= 0 {
		publishTimeout = 3 * time.Second
	}
	// 一条事实最多经历一次 publish deadline 和一次状态落库 deadline；再保留
	// 一个 deadline 作为调度余量。进程崩溃后的接管时间仍与单条消息耗时有界，
	// 不会随着 batch_size 放大成数十分钟。
	return 3 * publishTimeout
}

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()
}