topfans/backend/services/userService/mq/consumer.go

// Package mq 是 userService 的消息队列适配入口。
//
// 业务侧用法 (main.go 启动时):
//
//	if err := mq.RegisterHandlers(); err != nil { ... }
//	go mq.StartConsumers(ctx)
//
// 本期范围:只接入 user:accumulate-hours handler(接收 galleryService 派发的累计时长任务,
// 补回手动下架丢失的累计时长,见 backend/docs/superpowers/specs/2026-07-01-message-queue-design.md)。
package mq

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/topfans/backend/pkg/database"
	"github.com/topfans/backend/pkg/logger"
	"github.com/topfans/backend/pkg/mq/adapter"
	"github.com/topfans/backend/pkg/mq/tasks"
	"github.com/topfans/backend/services/userService/repository"
	"go.uber.org/zap"
)

// FanProfileRepository 是 AddExhibitionHours 必需的依赖,通过参数传入避免循环 import。
//
// 业务侧通常传 service 包装,这里只接受最小接口(handler 自己只调这一个方法)。
type FanProfileRepository interface {
	// AddExhibitionHours 增加用户累计上架时长并触发升级(已有实现)。
	// 幂等:handler 内部按 source_id 做幂等键(目前依赖业务侧保证唯一性)。
	AddExhibitionHours(userID, starID int64, hours int64, sourceID string) (newLevel, levelDelta int32, crystalReward int64, err error)
}

// RegisterHandlers 注册 userService 涉及的所有 MQ handler。
func RegisterHandlers(fanProfileRepo repository.FanProfileRepository) error {
	tc := adapter.Get().TaskConsumer()
	if err := tc.RegisterTask(tasks.TypeUserAccumulateHours, newHandleAccumulateHours(fanProfileRepo), adapter.TaskRegisterOptions{
		MaxRetry: 3,
		Queue:    "default",
	}); err != nil {
		return fmt.Errorf("register user:accumulate-hours: %w", err)
	}
	logger.Logger.Info("user mq handlers registered",
		zap.String("types", tasks.TypeUserAccumulateHours))
	return nil
}

// StartConsumers 启动 worker (阻塞直到 ctx 取消)。
//
// 业务侧应在 main.go 用 goroutine 调用:
//
//	go mq.StartConsumers(ctx)
func StartConsumers(ctx context.Context) error {
	return adapter.Get().TaskConsumer().Run(ctx)
}

// newHandleAccumulateHours 返回 adapter.TaskHandler,内部完成幂等后调 fanProfileRepo.AddExhibitionHours。
func newHandleAccumulateHours(repo FanProfileRepository) adapter.TaskHandler {
	return func(ctx context.Context, t *adapter.Task) error {
		var p tasks.UserAccumulateHoursPayload
		if err := tasks.UnmarshalPayload(t.Payload, &p); err != nil {
			logger.Logger.Error("handle user:accumulate-hours: unmarshal failed", zap.Error(err))
			return fmt.Errorf("unmarshal: %w", err)
		}
		if p.UserID <= 0 || p.StarID <= 0 || p.Hours <= 0 {
			logger.Logger.Warn("invalid accumulate hours payload",
				zap.Int64("user_id", p.UserID),
				zap.Int64("star_id", p.StarID),
				zap.Int32("hours", p.Hours))
			// 无效 payload 不重试,直接 drop
			return nil
		}

		// 幂等:galleryService 派发时 source_id = "exhibition_<id>",
		// 同一个 exhibition_id 多次派发只会累加一次。
		if isAlreadyProcessed(ctx, p.SourceID) {
			logger.Logger.Info("exhibition hours already accumulated, skip",
				zap.Int64("user_id", p.UserID),
				zap.String("source_id", p.SourceID))
			return nil
		}

		if _, _, _, err := repo.AddExhibitionHours(
			p.UserID, p.StarID,
			int64(p.Hours),
			p.SourceID,
		); err != nil {
			logger.Logger.Error("AddExhibitionHours failed",
				zap.Int64("user_id", p.UserID),
				zap.String("source_id", p.SourceID),
				zap.Error(err))
			return err
		}

		logger.Logger.Info("user exhibition hours accumulated",
			zap.Int64("user_id", p.UserID),
			zap.Int64("star_id", p.StarID),
			zap.Int32("hours", p.Hours),
			zap.String("source_id", p.SourceID),
			zap.String("source_type", p.Source))
		return nil
	}
}

// isAlreadyProcessed 查 records 表(level_up_logs / crystal_transactions 等)看 source_id 是否已出现。
// 由于 AddExhibitionHours 内部会用 source_id 做幂等(version_id 或 idempotency key),
// 这里额外包一层轻量级只读检查,以减少数据库调用。
//
// 实现:用 source_id 在 crystal_transaction_records 里查;如果没有该表就退化到 log 警告。
func isAlreadyProcessed(ctx context.Context, sourceID string) bool {
	if sourceID == "" {
		return false
	}

	// 优先查 crystal_transaction_records(source_id 上有索引)
	var count int64
	err := database.GetDB().Table("public.crystal_transaction_records").
		Where("source_id = ? AND change_type = ?", sourceID, "exhibition_revenue").
		Count(&count).Error
	if err == nil {
		return count > 0
	}

	// 退化:用 user_exhibition_hours 配合 source 文字搜索(简化措施,生产应换 schema)
	// 暂时返回 false 让 handler 继续,依赖 AddExhibitionHours 内部幂等。
	logger.Logger.Debug("isAlreadyProcessed fallback (table missing?)",
		zap.String("source_id", sourceID),
		zap.Error(err))
	return false
}

// parseUserIDFromSourceID 辅助函数 — 从 source_id 字符串 ("exhibition_<id>") 解析。
// 本期范围内未使用,保留供后续扩展。
func parseUserIDFromSourceID(sourceID string) int64 {
	parts := strings.SplitN(sourceID, "_", 2)
	if len(parts) < 2 {
		return 0
	}
	id, _ := strconv.ParseInt(parts[1], 10, 64)
	return id
}