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

package mq

import (
	"context"
	"fmt"
	"time"

	"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"
	"go.uber.org/zap"
)

// RegisterHandlers 注册 galleryService 涉及的所有 MQ handler。
// 在 main.go 启动时(数据库初始化后)调一次即可。
func RegisterHandlers() error {
	tc := adapter.Get().TaskConsumer()
	if err := tc.RegisterTask(tasks.TypeGalleryExhibitionExpire, HandleExhibitionExpire, adapter.TaskRegisterOptions{
		MaxRetry: 3,
		Queue:    "gallery",
	}); err != nil {
		return fmt.Errorf("register gallery:exhibition-expire: %w", err)
	}
	if err := tc.RegisterTask(tasks.TypeGalleryExhibitionSettled, HandleExhibitionSettled, adapter.TaskRegisterOptions{
		MaxRetry: 3,
		Queue:    "gallery",
	}); err != nil {
		return fmt.Errorf("register gallery:exhibition-settled: %w", err)
	}
	logger.Logger.Info("gallery mq handlers registered",
		zap.String("types", tasks.TypeGalleryExhibitionExpire+","+tasks.TypeGalleryExhibitionSettled))
	return nil
}

// StartConsumers 启动 worker (阻塞直到 ctx 取消)。
//
// 业务侧应在 main.go 用 goroutine 调用:
//
//	go mq.StartConsumers(ctx)
func StartConsumers(ctx context.Context) error {
	// 一次性迁移扫描：处理 MQ 上线前创建的旧展览（它们没有延迟任务）
	go func() {
		time.Sleep(5 * time.Second)
		scanExpiredExhibitions(context.Background())
	}()
	return adapter.Get().TaskConsumer().Run(ctx)
}

// HandleExhibitionExpire 自然到期精确触发 → EnqueueSettled。
//
// 入参是 GalleryExhibitionExpirePayload(只有 exhibition_id + 时间)。
// 这里只是把"自然到期"信号转成统一的"settlement"事件;
// 真正算收益/累加时长的是 HandleExhibitionSettled。
func HandleExhibitionExpire(ctx context.Context, t *adapter.Task) error {
	var p tasks.GalleryExhibitionExpirePayload
	if err := tasks.UnmarshalPayload(t.Payload, &p); err != nil {
		logger.Logger.Error("handle exhibition expire: unmarshal failed", zap.Error(err))
		return fmt.Errorf("unmarshal: %w", err)
	}

	// 计算实际下架时间 = now 或 expireAt,取较小(以防时序问题)
	settledAt := time.Now().UnixMilli()
	if settledAt > p.ExpireAt {
		settledAt = p.ExpireAt
	}
	actualHours := int32((settledAt - p.StartTime) / 3600000)
	if actualHours < 0 {
		actualHours = 0
	}

	// 从 DB 补全 occupier_uid / star_id / slot_owner（自然到期路径在 PlaceAsset 时只传了 exhibition_id + 时间）
	exh := fetchExhibitionForSettle(ctx, p.ExhibitionID)
	slotOwnerUID := exh.SlotOwnerUID
	occupierUID := exh.OccupierUID
	occupierStarID := exh.OccupierStarID
	if occupierUID == 0 {
		// 展览可能已被手动删除或其他边界,跳过
		logger.Logger.Warn("exhibition occupier missing, skip settle",
			zap.Int64("exhibition_id", p.ExhibitionID))
		return nil
	}

	return EnqueueExhibitSettled(
		ctx,
		p.ExhibitionID, p.AssetID, p.SlotID,
		occupierUID, occupierStarID, slotOwnerUID,
		p.StartTime, p.ExpireAt, settledAt,
		0, // likeCount 由 settled handler 内部通过 RPC 查询
		actualHours, "natural",
	)
}

// HandleExhibitionSettled 统一结算入口(三场景:natural / manual / kick)。
//
// 流程:
//  1. 幂等检查 — exhibition 已经 settled 过则直接 return
//  2. 标记 settled=true
//  3. 查点赞数 + occupier_uid/star_id(原本 cleanup_worker 通过 RPC 查)
//  4. 派发 4 个子任务:revenue:exhibition / revenue:like-bet / user:accumulate-hours / asset:accumulate-hours
func HandleExhibitionSettled(ctx context.Context, t *adapter.Task) error {
	var p tasks.GalleryExhibitionSettledPayload
	if err := tasks.UnmarshalPayload(t.Payload, &p); err != nil {
		logger.Logger.Error("handle exhibition settled: unmarshal failed", zap.Error(err))
		return fmt.Errorf("unmarshal: %w", err)
	}

	alreadySettled := isSettled(ctx, p.ExhibitionID)

	// 1. 幂等检查: 首次结算才标记 + 派发全量子任务
	if !alreadySettled {
		// 标记 settled
		if err := markSettled(ctx, p.ExhibitionID); err != nil {
			logger.Logger.Warn("mark settled failed, but continue",
				zap.Int64("exhibition_id", p.ExhibitionID),
				zap.Error(err))
		}
	}

	// 2. 派发子任务
	helperEnqueue := func(name string, err error) {
		if err != nil {
			logger.Logger.Warn("enqueue sub-task failed",
				zap.String("sub_task", name),
				zap.Int64("exhibition_id", p.ExhibitionID),
				zap.Error(err))
		}
	}

	// user:accumulate-hours — 仅首次结算时派发(已累加过则跳过)
	if !alreadySettled && p.OccupierUID > 0 && p.OccupierStarID > 0 && p.ActualHours > 0 {
		helperEnqueue("user:accumulate-hours",
			EnqueueUserAccumulateHours(
				ctx,
				p.OccupierUID, p.OccupierStarID,
				p.ActualHours,
				exhibitionSourceID(p.ExhibitionID),
				p.Source,
			))
	}

	if p.OccupierUID > 0 && p.OccupierStarID > 0 {
		// revenue:exhibition — 仅首次结算时派发
		if !alreadySettled {
			helperEnqueue("revenue:exhibition",
				EnqueueRevenueExhibition(
					ctx,
					p.ExhibitionID, p.AssetID, p.SlotID,
					p.OccupierUID, p.OccupierStarID, p.SlotOwnerUID,
					p.StartTime, p.ExpireAt,
					p.LikeCount,
					p.Source,
				))
		}
		// revenue:like-bet — 始终派发！即使已结算也补发（幂等，修复旧代码遗漏）
		helperEnqueue("revenue:like-bet",
			EnqueueRevenueLikeBet(
				ctx,
				p.ExhibitionID, p.AssetID,
				p.StartTime, p.ExpireAt,
				p.Source,
			))
	}

	logger.Logger.Info("exhibition settled dispatch done",
		zap.Int64("exhibition_id", p.ExhibitionID),
		zap.String("source", p.Source),
		zap.Int32("actual_hours", p.ActualHours),
		zap.Int32("like_count", p.LikeCount))

	return nil
}

// exhibitionSourceID 把 exhibition_id 拼成 sourceID 字符串。
func exhibitionSourceID(id int64) string {
	return fmt.Sprintf("exhibition_%d", id)
}

// exhForSettle 从 DB 查展览关键字段（轻量）。
type exhForSettle struct {
	OccupierUID    int64
	OccupierStarID int64
	SlotOwnerUID   int64
}

// fetchExhibitionForSettle 从 exhibitions 表补全 occupier/slot_owner 信息。
func fetchExhibitionForSettle(ctx context.Context, exhibitionID int64) exhForSettle {
	var r exhForSettle
	err := database.GetDB().Table("public.exhibitions").
		Select("COALESCE(occupier_uid,0) as occupier_uid, COALESCE(occupier_star_id,0) as occupier_star_id, COALESCE(host_profile_id,0) as slot_owner_uid").
		Where("id = ?", exhibitionID).
		Scan(&r).Error
	if err != nil {
		logger.Logger.Warn("fetch exhibition for settle failed",
			zap.Int64("exhibition_id", exhibitionID),
			zap.Error(err))
	}
	return r
}

// isSettled 查 exhibition 是否已 settled。
// settled_at 非空且大于 0 视为已结算。
func isSettled(ctx context.Context, exhibitionID int64) bool {
	var settledAt *int64
	err := database.GetDB().Table("public.exhibitions").
		Select("settled_at").
		Where("id = ?", exhibitionID).
		Scan(&settledAt).Error
	if err != nil {
		// 查询失败 = 视为未 settled,允许 handler 继续
		return false
	}
	return settledAt != nil && *settledAt > 0
}

func markSettled(ctx context.Context, exhibitionID int64) error {
	return database.GetDB().Table("public.exhibitions").
		Where("id = ?", exhibitionID).
		Update("settled_at", time.Now().UnixMilli()).Error
}

// scanExpiredExhibitions 一次性迁移扫描：
//
//	deleted_at IS NULL
//	AND settled_at IS NULL
//	AND expire_at < now
//
// 对每条记录补发 EnqueueExhibitSettled(recovery) 结算任务。
func scanExpiredExhibitions(ctx context.Context) {
	type expiredRow struct {
		ID              int64
		AssetID         int64
		SlotID          int64
		OccupierUID     int64
		OccupierStarID  int64
		SlotOwnerUID    int64
		StartTime       int64
		ExpireAt        int64
	}

	var rows []expiredRow
	nowMs := time.Now().UnixMilli()
	if err := database.GetDB().Table("public.exhibitions").
		Select("id, asset_id, slot_id, COALESCE(occupier_uid,0) AS occupier_uid, COALESCE(occupier_star_id,0) AS occupier_star_id, COALESCE(host_profile_id,0) AS slot_owner_uid, start_time, expire_at").
		Where("deleted_at IS NULL AND settled_at IS NULL AND expire_at < ?", nowMs).
		Limit(500).
		Scan(&rows).Error; err != nil {
		logger.Logger.Error("expiry scanner: query failed", zap.Error(err))
		return
	}

	if len(rows) == 0 {
		return
	}

	logger.Logger.Info("expiry scanner: found expired exhibitions",
		zap.Int("count", len(rows)))

	for _, r := range rows {
		if r.OccupierUID == 0 {
			logger.Logger.Warn("expiry scanner: skip exhibition without occupier",
				zap.Int64("exhibition_id", r.ID))
			continue
		}
		settledAt := nowMs
		if settledAt > r.ExpireAt {
			settledAt = r.ExpireAt
		}
		actualHours := int32((settledAt - r.StartTime) / 3600000)
		if actualHours < 0 {
			actualHours = 0
		}
		if err := EnqueueExhibitSettled(
			ctx,
			r.ID, r.AssetID, r.SlotID,
			r.OccupierUID, r.OccupierStarID, r.SlotOwnerUID,
			r.StartTime, r.ExpireAt, settledAt,
			0, actualHours, "recovery",
		); err != nil {
			logger.Logger.Warn("expiry scanner: enqueue failed",
				zap.Int64("exhibition_id", r.ID),
				zap.Error(err))
		}
	}
}