// Package tasks 集中管理所有 TaskType 常量 + Payload 结构。 // // 设计原则: // // - 所有 TaskType 字符串集中在此,避免业务代码散落字面量 // - 每个 TaskType 配套一个 Payload struct,业务侧用 marshal/unmarshal 工具类型安全地序列化 // - 后续切 broker (RabbitMQ 等) 时只改 adapter 层,本包不动 // // 业务侧用法: // // import "github.com/topfans/backend/pkg/mq/tasks" // import "github.com/topfans/backend/pkg/mq/adapter" // // // 生产 // tasks.MarshalToPayload(tasks.RevenueExhibitionPayload{ExhibitionID: 123, ...}) // adapter.Get().TaskProducer().Enqueue(ctx, adapter.Task{Type: tasks.TypeRevenueExhibition, Payload: p, MaxRetry: 3}) // // // 消费 // var p tasks.RevenueExhibitionPayload // if err := tasks.UnmarshalPayload(t.Payload, &p); err != nil { ... } // // ... 业务逻辑 package tasks import ( "encoding/json" "errors" ) // TaskType 字符串常量统一在此,handler 注册用同一个字符串。 const ( // ===== Phase 1: 收益类(这次实施范围)===== // TypeRevenueExhibition 展览展示收益结算 // 生产者: galleryService(gallery:exhibition-settled handler) // 消费者: taskService // 重试: 3 次,死信队列 TypeRevenueExhibition = "revenue:exhibition" // TypeRevenueLikeBet 展览点赞押注收益结算 // 生产者: galleryService(gallery:exhibition-settled handler) // 消费者: taskService // 重试: 3 次,死信队列 TypeRevenueLikeBet = "revenue:like-bet" // TypeGalleryExhibitionSettled 通用展览结算入口 // 生产者: galleryService 业务侧 (RemoveFromSlot / RemoveExhibitionByAsset / cleanup_worker) // 消费者: galleryService (本服务消费) // 重试: 3 次 TypeGalleryExhibitionSettled = "gallery:exhibition-settled" // TypeGalleryExhibitionExpire 展览自然到期精确触发 // 生产者: galleryService 业务侧(上架时 EnqueueAt) // 消费者: galleryService (本服务消费,ProcessAt 触发后转 settled) // 重试: 3 次 TypeGalleryExhibitionExpire = "gallery:exhibition-expire" // TypeUserAccumulateHours 用户累计上架时长累加 // 生产者: galleryService(gallery:exhibition-settled handler) // 消费者: userService // 重试: 3 次,handler 内按 exhibition_id 幂等 TypeUserAccumulateHours = "user:accumulate-hours" // TypeTaskEvent 通用业务事件入口(daily-task 引擎消费) // 生产者: assetService(铸造成功处 emit daily_mint) + 前端 reportEvent RPC(daily_login/daily_browse_asset/daily_place_asset) // 消费者: taskService // 重试: 3 次,handler 内按 user_id+star_id+task_key 幂等(pending→completed 一次性) TypeTaskEvent = "task:event" // ===== Phase 2+: 预留(本期不实现,仅占位方便后续查阅)===== TypeAssetAccumulateHours = "asset:accumulate-hours" TypeNotificationCreate = "notification:create" TypeNotificationPush = "notification:push" TypeModerationAutoHide = "moderation:auto-hide" TypeAichatChat = "aichat:chat" TypeStarbookCollectionNew = "starbook:collection-create" TypeGalleryCleanupDisplayStatus = "gallery:cleanup-display-status" TypeGalleryExpiredFallback = "gallery:expired-exhibition-fallback" TypeTaskDailyReset = "task:daily-reset" TypeAssetSeasonReset = "asset:season-reset" TypeStatisticMaterialize = "statistic:materialize" TypeStatisticWeeklyIncome = "statistic:weekly-income" TypeStatisticLevelUp = "statistic:level-up" TypeStatisticPartitionCreate = "statistic:partition-create" TypeStatisticPartitionDrop = "statistic:partition-drop" // StreamTopic 集中管理 StreamUser = "stream:user" StreamAsset = "stream:asset" StreamSocial = "stream:social" StreamExhibition = "stream:exhibition" StreamActivity = "stream:activity" StreamModeration = "stream:moderation" ) // ===== Phase 1 Payload 结构 ===== // RevenueExhibitionPayload exhibition 收益结算参数。 // 来源: gallery:exhibition-settled handler 内组装。 type RevenueExhibitionPayload struct { ExhibitionID int64 `json:"exhibition_id"` AssetID int64 `json:"asset_id"` SlotID int64 `json:"slot_id"` OccupierUID int64 `json:"occupier_uid"` OccupierStarID int64 `json:"occupier_star_id"` SlotOwnerUID int64 `json:"slot_owner_uid"` StartTime int64 `json:"start_time"` // 毫秒 ExpireAt int64 `json:"expire_at"` // 毫秒 LikeCount int32 `json:"like_count"` Source string `json:"source"` // "natural" | "manual" | "kick" } // RevenueLikeBetPayload exhibition 点赞押注收益结算参数。 type RevenueLikeBetPayload struct { ExhibitionID int64 `json:"exhibition_id"` AssetID int64 `json:"asset_id"` StartTime int64 `json:"start_time"` ExpireAt int64 `json:"expire_at"` Source string `json:"source"` } // GalleryExhibitionSettledPayload exhibition 通用结算入口参数。 // 所有路径(自然到期/手动下架/踢走)统一入队这个 payload。 type GalleryExhibitionSettledPayload struct { ExhibitionID int64 `json:"exhibition_id"` AssetID int64 `json:"asset_id"` SlotID int64 `json:"slot_id"` OccupierUID int64 `json:"occupier_uid"` OccupierStarID int64 `json:"occupier_star_id"` SlotOwnerUID int64 `json:"slot_owner_uid"` StartTime int64 `json:"start_time"` ExpireAt int64 `json:"expire_at"` SettledAt int64 `json:"settled_at"` // 实际下架时间 LikeCount int32 `json:"like_count"` ActualHours int32 `json:"actual_hours"` Source string `json:"source"` // "natural" | "manual" | "kick" } // GalleryExhibitionExpirePayload exhibition 自然到期参数。 // 上架时 EnqueueAt(expireAt),到点才执行。 type GalleryExhibitionExpirePayload struct { ExhibitionID int64 `json:"exhibition_id"` AssetID int64 `json:"asset_id"` SlotID int64 `json:"slot_id"` StartTime int64 `json:"start_time"` ExpireAt int64 `json:"expire_at"` // 绝对到期时刻 } // UserAccumulateHoursPayload 累加用户上架时长参数。 // 来源: gallery:exhibition-settled handler 派发。 type UserAccumulateHoursPayload struct { UserID int64 `json:"user_id"` StarID int64 `json:"star_id"` Hours int32 `json:"hours"` SourceID string `json:"source_id"` // 关联业务 ID(通常为 exhibition_id 字符串) Source string `json:"source"` // "natural" | "manual" | "kick" } // ===== Marshal / Unmarshal 工具 ===== // MarshalToPayload 把结构体序列化为 map[string]any(供 adapter.Task.Payload 使用)。 func MarshalToPayload(v any) (map[string]any, error) { b, err := json.Marshal(v) if err != nil { return nil, err } m := make(map[string]any) if err := json.Unmarshal(b, &m); err != nil { return nil, err } return m, nil } // UnmarshalPayload 把 map 反序列化为指定 Payload 结构体。 // payload 通常来自 adapter.Task.Payload(handler 接收端)。 func UnmarshalPayload(payload map[string]any, out any) error { if payload == nil { return errors.New("mq: payload is nil") } b, err := json.Marshal(payload) if err != nil { return err } return json.Unmarshal(b, out) } // MustMarshalToPayload panic 版本的 MarshalToPayload,使用前保证不会失败场景用。 func MustMarshalToPayload(v any) map[string]any { m, err := MarshalToPayload(v) if err != nil { panic(err) } return m }