hyapp-server/services/room-service/internal/room/service/outbox_worker.go

package service

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	"hyapp/pkg/appcode"
)

const (
	// outboxRetryFixedInterval 是 V1 唯一支持的补偿策略，失败记录下一轮继续扫描。
	outboxRetryFixedInterval = "fixed_interval"
	// outboxWorkerName 固定写入日志，便于后续按 worker 聚合检索。
	outboxWorkerName = "room_outbox"
)

// OutboxWorkerOptions 控制 room_outbox pending 事件的持续补偿循环。
type OutboxWorkerOptions struct {
	// PollInterval 是两轮 pending 扫描之间的间隔；非正数会被归一到安全默认值。
	PollInterval time.Duration
	// BatchSize 是单轮最多处理条数；非正数会被归一，避免全表扫描。
	BatchSize int
	// PublishTimeout 限制单条外部投递耗时，shutdown 时当前事件最多阻塞该时长。
	PublishTimeout time.Duration
	// RetryStrategy 当前只支持 fixed_interval。
	RetryStrategy string
}

func defaultOutboxWorkerOptions() OutboxWorkerOptions {
	// 默认值对应文档里的 V1 固定间隔补偿策略，优先避免 busy loop 和无限外部调用。
	return OutboxWorkerOptions{
		PollInterval:   time.Second,
		BatchSize:      100,
		PublishTimeout: 3 * time.Second,
		RetryStrategy:  outboxRetryFixedInterval,
	}
}

func normalizeOutboxWorkerOptions(options OutboxWorkerOptions) OutboxWorkerOptions {
	// worker 层再次归一化，防止测试或手动构造绕过 config.Normalize。
	defaults := defaultOutboxWorkerOptions()
	if options.PollInterval <= 0 {
		// 非正间隔不能直接传给 ticker，否则会 panic 或形成忙循环。
		options.PollInterval = defaults.PollInterval
	}
	if options.BatchSize <= 0 {
		// 非正批量按安全默认值处理，避免 repository 扫描不受控。
		options.BatchSize = defaults.BatchSize
	}
	if options.PublishTimeout <= 0 {
		// 单条外部投递必须有 deadline，shutdown 才能等待有界。
		options.PublishTimeout = defaults.PublishTimeout
	}
	options.RetryStrategy = strings.TrimSpace(options.RetryStrategy)
	if options.RetryStrategy == "" {
		// 空策略按 fixed_interval 处理，保持和配置默认值一致。
		options.RetryStrategy = defaults.RetryStrategy
	}

	return options
}

// RunOutboxWorker 持续扫描 pending outbox；它不执行房间命令，只投递已持久化事件。
func (s *Service) RunOutboxWorker(ctx context.Context, options OutboxWorkerOptions) {
	options = normalizeOutboxWorkerOptions(options)
	if options.RetryStrategy != outboxRetryFixedInterval {
		// 配置层会拒绝未知策略；这里兜底避免测试或手写构造静默改变语义。
		log.Printf("worker=%s node_id=%s status=stopped error=%q", outboxWorkerName, s.nodeID, "unsupported retry_strategy")
		return
	}

	if err := s.ProcessPendingOutbox(ctx, options); err != nil && ctx.Err() == nil {
		log.Printf("worker=%s node_id=%s status=batch_failed error=%q", outboxWorkerName, s.nodeID, trimOutboxError(err.Error()))
	}

	ticker := time.NewTicker(options.PollInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := s.ProcessPendingOutbox(ctx, options); err != nil && ctx.Err() == nil {
				log.Printf("worker=%s node_id=%s status=batch_failed error=%q", outboxWorkerName, s.nodeID, trimOutboxError(err.Error()))
			}
		}
	}
}

// ProcessPendingOutbox 把待投递事件补偿性地推送给异步消费者。
func (s *Service) ProcessPendingOutbox(ctx context.Context, options OutboxWorkerOptions) error {
	options = normalizeOutboxWorkerOptions(options)
	if options.RetryStrategy != outboxRetryFixedInterval {
		return fmt.Errorf("unsupported outbox retry_strategy %q", options.RetryStrategy)
	}

	// 补偿 worker 先抢占 pending/retryable 事件，再投递，避免多机同时处理同一批。
	records, err := s.repository.ClaimPendingOutbox(ctx, s.nodeID, options.BatchSize, time.Now().Add(options.PublishTimeout))
	if err != nil {
		return err
	}

	for _, record := range records {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		publishCtx, cancel := context.WithTimeout(context.Background(), options.PublishTimeout)
		err := s.outboxPublisher.PublishOutboxEvent(publishCtx, record.Envelope)
		cancel()
		if err != nil {
			// 投递失败转为 retryable，并记录错误和重试次数，等待下一轮 worker 再抢占。
			markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
			markErr := s.repository.MarkOutboxFailed(markCtx, record.EventID, trimOutboxError(err.Error()))
			markCancel()
			if markErr != nil {
				return markErr
			}

			log.Printf("worker=%s node_id=%s event_id=%s event_type=%s room_id=%s retry_count=%d status=retryable error=%q", outboxWorkerName, s.nodeID, record.EventID, record.EventType, record.RoomID, record.RetryCount+1, trimOutboxError(err.Error()))
			continue
		}

		// 只有外部发布成功后才能标记 delivered。
		markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout)
		markErr := s.repository.MarkOutboxDelivered(markCtx, record.EventID)
		markCancel()
		if markErr != nil {
			return markErr
		}

		log.Printf("worker=%s node_id=%s event_id=%s event_type=%s room_id=%s retry_count=%d status=delivered", outboxWorkerName, s.nodeID, record.EventID, record.EventType, record.RoomID, record.RetryCount)
	}

	return nil
}

func trimOutboxError(value string) string {
	// last_error 会写入 MySQL 和日志，先去掉首尾空白避免不可见差异影响排障。
	value = strings.TrimSpace(value)
	if len(value) <= 512 {
		return value
	}

	// V1 只保留错误前缀，避免外部响应过大撑爆日志或 TEXT 字段查询成本。
	return value[:512]
}