287 lines
9.0 KiB
Go
287 lines
9.0 KiB
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。
|
||
// 注:这个方法需要 exhibition 表有 settled 列,这里先简化用 is_processed 字段保持兼容。
|
||
// 真实上线时需执行 migration 加 settled 列。
|
||
func isSettled(ctx context.Context, exhibitionID int64) bool {
|
||
var processed bool
|
||
err := database.GetDB().Table("public.exhibitions").
|
||
Select("COALESCE(is_processed, false)"). // 暂时复用作 settled
|
||
Where("id = ?", exhibitionID).
|
||
Scan(&processed).Error
|
||
if err != nil {
|
||
// 查询失败 = 视为未 settled,允许 handler 继续
|
||
return false
|
||
}
|
||
return processed
|
||
}
|
||
|
||
func markSettled(ctx context.Context, exhibitionID int64) error {
|
||
return database.GetDB().Table("public.exhibitions").
|
||
Where("id = ?", exhibitionID).
|
||
Update("is_processed", true).Error
|
||
}
|
||
|
||
// scanExpiredExhibitions 一次性迁移扫描：
|
||
//
|
||
// deleted_at IS NULL
|
||
// AND is_processed = false
|
||
// 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 is_processed = false 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))
|
||
}
|
||
}
|
||
}
|