531 lines
21 KiB
Go
531 lines
21 KiB
Go
// Package broadcast 实现 activity-service 拥有的腾讯云 IM 播报能力。
|
||
// 该包只处理群生命周期、播报入队、outbox 投递和 room 事件消费策略，不拥有房间状态和资金事实。
|
||
package broadcast
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/imgroup"
|
||
"hyapp/pkg/tencentim"
|
||
"hyapp/pkg/xerr"
|
||
broadcastdomain "hyapp/services/activity-service/internal/domain/broadcast"
|
||
)
|
||
|
||
const (
|
||
defaultBroadcastBatchSize = 100
|
||
defaultBroadcastLockTTL = 30 * time.Second
|
||
defaultBroadcastMaxRetry = 8
|
||
)
|
||
|
||
// Repository 是播报消息的持久化 outbox 边界。
|
||
// service 只通过这个接口写入/抢占/标记记录，确保腾讯 REST 失败后仍可重试和人工补偿。
|
||
type Repository interface {
|
||
SaveBroadcastOutbox(ctx context.Context, record broadcastdomain.OutboxRecord) (bool, broadcastdomain.OutboxRecord, error)
|
||
ClaimPendingBroadcastOutbox(ctx context.Context, workerID string, nowMS int64, lockUntilMS int64, limit int, maxRetry int) ([]broadcastdomain.OutboxRecord, error)
|
||
MarkBroadcastOutboxDelivered(ctx context.Context, eventID string, nowMS int64) error
|
||
MarkBroadcastOutboxFailed(ctx context.Context, eventID string, workerID string, nowMS int64, nextRetryAtMS int64, maxRetry int, lastError string) error
|
||
}
|
||
|
||
// Publisher 隐藏腾讯云 IM REST 细节，只暴露 activity-service 真正拥有的建群和发群自定义消息能力。
|
||
type Publisher interface {
|
||
EnsureGroup(ctx context.Context, spec tencentim.GroupSpec) error
|
||
PublishGroupCustomMessage(ctx context.Context, message tencentim.CustomGroupMessage) error
|
||
DeleteGroupMember(ctx context.Context, groupID string, userID int64) error
|
||
}
|
||
|
||
// RegionSource 从 user-service 读取 active 区域。
|
||
// 区域归属是用户主数据，activity-service 不复制区域配置，只在建群 reconciler 中读取。
|
||
type RegionSource interface {
|
||
ListActiveRegionIDs(ctx context.Context) ([]int64, error)
|
||
}
|
||
|
||
// Config 保存播报策略和 worker 参数。
|
||
// SuperGiftMinValue 是产品策略阈值；Worker* 是 outbox 投递吞吐和故障恢复参数。
|
||
type Config struct {
|
||
NodeID string
|
||
SuperGiftMinValue int64
|
||
WorkerBatchSize int
|
||
WorkerLockTTL time.Duration
|
||
WorkerMaxRetry int
|
||
WorkerPollInterval time.Duration
|
||
EnsureGroupsOnStartup bool
|
||
}
|
||
|
||
// PublishInput 是服务端入队播报请求。
|
||
// 调用方必须提供稳定 EventID；重复请求会命中 outbox 幂等，而不是重复发 IM。
|
||
type PublishInput struct {
|
||
EventID string
|
||
BroadcastType string
|
||
RegionID int64
|
||
PayloadJSON string
|
||
DeliverAtMS int64
|
||
}
|
||
|
||
// Service 拥有播报群建群补偿、播报入队和 outbox 投递决策。
|
||
// 它不直接向客户端返回 UI 文案，只生产带 action 的结构化 JSON，让客户端自行决定展示形态。
|
||
type Service struct {
|
||
cfg Config
|
||
repository Repository
|
||
publisher Publisher
|
||
regionSource RegionSource
|
||
now func() time.Time
|
||
}
|
||
|
||
// New 创建播报服务。
|
||
// publisher 可以为 nil，此时仍允许消费事件并写 outbox，但真正投递会 fail-closed；本地环境默认不误发腾讯云消息。
|
||
func New(cfg Config, repository Repository, publisher Publisher, regionSource RegionSource) *Service {
|
||
cfg = normalizeConfig(cfg)
|
||
return &Service{cfg: cfg, repository: repository, publisher: publisher, regionSource: regionSource, now: time.Now}
|
||
}
|
||
|
||
// SetClock 注入确定性时间，测试中用它验证 payload 和 outbox 的 Unix ms 字段。
|
||
func (s *Service) SetClock(now func() time.Time) {
|
||
if now != nil {
|
||
s.now = now
|
||
}
|
||
}
|
||
|
||
// EnsureBroadcastGroups 幂等创建全局播报群和所有 active 区域播报群。
|
||
// 这个动作放在启动/定时 reconciler，不放在 /im/usersig 路径，避免用户登录被外部腾讯 REST 拖慢。
|
||
func (s *Service) EnsureBroadcastGroups(ctx context.Context) (int, error) {
|
||
if s == nil || s.publisher == nil {
|
||
return 0, xerr.New(xerr.Unavailable, "broadcast publisher is not configured")
|
||
}
|
||
app := appcode.FromContext(ctx)
|
||
count := 0
|
||
globalID, err := imgroup.GlobalBroadcastGroupID(app)
|
||
if err != nil {
|
||
return 0, xerr.New(xerr.InvalidArgument, err.Error())
|
||
}
|
||
if err := s.publisher.EnsureGroup(ctx, broadcastGroupSpec(globalID)); err != nil {
|
||
return count, err
|
||
}
|
||
count++
|
||
|
||
if s.regionSource == nil {
|
||
// 没有区域源时至少保证全局群存在；区域群等待下次有 user-service 依赖的 reconciler 补齐。
|
||
return count, nil
|
||
}
|
||
regionIDs, err := s.regionSource.ListActiveRegionIDs(ctx)
|
||
if err != nil {
|
||
return count, err
|
||
}
|
||
for _, regionID := range regionIDs {
|
||
if regionID <= 0 {
|
||
// GLOBAL/未知区域不对应 IM 区域群，避免客户端加入一个语义不清的“0 区域群”。
|
||
continue
|
||
}
|
||
groupID, err := imgroup.RegionBroadcastGroupID(app, regionID)
|
||
if err != nil {
|
||
return count, xerr.New(xerr.InvalidArgument, err.Error())
|
||
}
|
||
if err := s.publisher.EnsureGroup(ctx, broadcastGroupSpec(groupID)); err != nil {
|
||
return count, err
|
||
}
|
||
count++
|
||
}
|
||
return count, nil
|
||
}
|
||
|
||
// PublishRegionBroadcast 把区域播报写入持久化 outbox。
|
||
// 这里不直接调用腾讯 REST，保证调用方提交成功后即使进程崩溃也能由 worker 继续投递。
|
||
func (s *Service) PublishRegionBroadcast(ctx context.Context, input PublishInput) (broadcastdomain.PublishResult, error) {
|
||
if input.RegionID <= 0 {
|
||
return broadcastdomain.PublishResult{}, xerr.New(xerr.InvalidArgument, "region_id is required")
|
||
}
|
||
groupID, err := imgroup.RegionBroadcastGroupID(appcode.FromContext(ctx), input.RegionID)
|
||
if err != nil {
|
||
return broadcastdomain.PublishResult{}, xerr.New(xerr.InvalidArgument, err.Error())
|
||
}
|
||
return s.enqueue(ctx, broadcastdomain.ScopeRegion, groupID, input)
|
||
}
|
||
|
||
func (s *Service) PublishRoomBroadcast(ctx context.Context, roomID string, input PublishInput) (broadcastdomain.PublishResult, error) {
|
||
roomID = strings.TrimSpace(roomID)
|
||
if roomID == "" {
|
||
return broadcastdomain.PublishResult{}, xerr.New(xerr.InvalidArgument, "room_id is required")
|
||
}
|
||
return s.enqueue(ctx, broadcastdomain.ScopeRoom, roomID, input)
|
||
}
|
||
|
||
// PublishGlobalBroadcast 把全局播报写入持久化 outbox。
|
||
// 全局播报和区域播报共用同一张表，scope/group_id 决定最终投递目标。
|
||
func (s *Service) PublishGlobalBroadcast(ctx context.Context, input PublishInput) (broadcastdomain.PublishResult, error) {
|
||
groupID, err := imgroup.GlobalBroadcastGroupID(appcode.FromContext(ctx))
|
||
if err != nil {
|
||
return broadcastdomain.PublishResult{}, xerr.New(xerr.InvalidArgument, err.Error())
|
||
}
|
||
return s.enqueue(ctx, broadcastdomain.ScopeGlobal, groupID, input)
|
||
}
|
||
|
||
// RemoveRegionBroadcastMember 把一个用户从旧区域播报群移除。
|
||
// 这是成员关系更新，不是区域群生命周期操作；区域群仍由 reconciler 长期维护，其他用户不受影响。
|
||
func (s *Service) RemoveRegionBroadcastMember(ctx context.Context, userID int64, regionID int64) (string, bool, error) {
|
||
if userID <= 0 {
|
||
return "", false, xerr.New(xerr.InvalidArgument, "user_id is required")
|
||
}
|
||
if regionID <= 0 {
|
||
// GLOBAL/无区域没有对应区域播报群，调用方跨区域比较时可以安全把 0 传进来。
|
||
return "", false, nil
|
||
}
|
||
if s == nil || s.publisher == nil {
|
||
return "", false, xerr.New(xerr.Unavailable, "broadcast publisher is not configured")
|
||
}
|
||
groupID, err := imgroup.RegionBroadcastGroupID(appcode.FromContext(ctx), regionID)
|
||
if err != nil {
|
||
return "", false, xerr.New(xerr.InvalidArgument, err.Error())
|
||
}
|
||
if err := s.publisher.DeleteGroupMember(ctx, groupID, userID); err != nil {
|
||
return groupID, false, err
|
||
}
|
||
return groupID, true, nil
|
||
}
|
||
|
||
func (s *Service) enqueue(ctx context.Context, scope string, groupID string, input PublishInput) (broadcastdomain.PublishResult, error) {
|
||
if s == nil || s.repository == nil {
|
||
return broadcastdomain.PublishResult{}, xerr.New(xerr.Unavailable, "broadcast repository is not configured")
|
||
}
|
||
input.EventID = strings.TrimSpace(input.EventID)
|
||
input.BroadcastType = strings.TrimSpace(input.BroadcastType)
|
||
if input.EventID == "" || input.BroadcastType == "" {
|
||
return broadcastdomain.PublishResult{}, xerr.New(xerr.InvalidArgument, "event_id and broadcast_type are required")
|
||
}
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
payloadJSON, err := normalizedPayloadJSON(input.PayloadJSON, map[string]any{
|
||
"event_id": input.EventID,
|
||
"broadcast_type": input.BroadcastType,
|
||
"scope": scope,
|
||
"app_code": appcode.FromContext(ctx),
|
||
"region_id": input.RegionID,
|
||
"sent_at_ms": nowMS,
|
||
})
|
||
if err != nil {
|
||
return broadcastdomain.PublishResult{}, err
|
||
}
|
||
// outbox 记录保存的是完整 JSON，而不是只保存引用 ID；客户端收到 IM 后无需再同步请求才能完成入口跳转。
|
||
record := broadcastdomain.OutboxRecord{
|
||
AppCode: appcode.FromContext(ctx),
|
||
EventID: input.EventID,
|
||
Scope: scope,
|
||
GroupID: groupID,
|
||
BroadcastType: input.BroadcastType,
|
||
PayloadJSON: payloadJSON,
|
||
Status: broadcastdomain.StatusPending,
|
||
NextRetryAtMS: nowMS,
|
||
CreatedAtMS: nowMS,
|
||
UpdatedAtMS: nowMS,
|
||
}
|
||
if input.DeliverAtMS > nowMS {
|
||
record.NextRetryAtMS = input.DeliverAtMS
|
||
}
|
||
created, saved, err := s.repository.SaveBroadcastOutbox(ctx, record)
|
||
if err != nil {
|
||
return broadcastdomain.PublishResult{}, err
|
||
}
|
||
return broadcastdomain.PublishResult{EventID: saved.EventID, GroupID: saved.GroupID, Status: saved.Status, Created: created}, nil
|
||
}
|
||
|
||
// ProcessPendingBroadcasts 抢占一批待投递 outbox，发送到腾讯 IM，并更新成功/重试状态。
|
||
// claim 和 mark 分开做：抢占阶段保护多 worker 并发，发送阶段允许单条失败不影响同批其他消息。
|
||
func (s *Service) ProcessPendingBroadcasts(ctx context.Context, options WorkerOptions) (broadcastdomain.ProcessResult, error) {
|
||
if s == nil || s.repository == nil {
|
||
return broadcastdomain.ProcessResult{}, xerr.New(xerr.Unavailable, "broadcast repository is not configured")
|
||
}
|
||
if s.publisher == nil {
|
||
return broadcastdomain.ProcessResult{}, xerr.New(xerr.Unavailable, "broadcast publisher is not configured")
|
||
}
|
||
options = normalizeWorkerOptions(options, s.cfg)
|
||
nowMS := s.now().UTC().UnixMilli()
|
||
records, err := s.repository.ClaimPendingBroadcastOutbox(ctx, options.WorkerID, nowMS, nowMS+options.LockTTL.Milliseconds(), options.BatchSize, options.MaxRetry)
|
||
if err != nil {
|
||
return broadcastdomain.ProcessResult{}, err
|
||
}
|
||
result := broadcastdomain.ProcessResult{ClaimedCount: len(records), HasMore: len(records) >= options.BatchSize}
|
||
for _, record := range records {
|
||
// 腾讯 REST 只接受已序列化 JSON；event_id/desc/ext 让客户端和日志都能快速识别消息类型。
|
||
err := s.publisher.PublishGroupCustomMessage(ctx, tencentim.CustomGroupMessage{
|
||
GroupID: record.GroupID,
|
||
EventID: record.EventID,
|
||
Desc: record.BroadcastType,
|
||
Ext: "im_broadcast",
|
||
PayloadJSON: json.RawMessage(record.PayloadJSON),
|
||
})
|
||
if err != nil {
|
||
result.FailureCount++
|
||
// Mark 失败只记录为 best-effort：当前记录锁会过期，下一轮 worker 仍可重新接管。
|
||
_ = s.repository.MarkBroadcastOutboxFailed(ctx, record.EventID, options.WorkerID, s.now().UTC().UnixMilli(), s.now().UTC().Add(options.PollInterval).UnixMilli(), options.MaxRetry, trimError(err.Error()))
|
||
continue
|
||
}
|
||
if err := s.repository.MarkBroadcastOutboxDelivered(ctx, record.EventID, s.now().UTC().UnixMilli()); err != nil {
|
||
return result, err
|
||
}
|
||
result.SuccessCount++
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// HandleRoomEvent 把已提交的 room outbox 事实转换为服务端播报。
|
||
// 只有 RoomGiftSent 且满足区域和礼物价值阈值时才生成区域播报，避免把展示策略侵入 Room Cell 主链路。
|
||
func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) (broadcastdomain.ConsumeRoomEventResult, error) {
|
||
if envelope == nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, xerr.New(xerr.InvalidArgument, "room event envelope is required")
|
||
}
|
||
eventCtx := appcode.WithContext(ctx, envelope.GetAppCode())
|
||
result := broadcastdomain.ConsumeRoomEventResult{EventID: envelope.GetEventId(), Status: broadcastdomain.StatusSkipped}
|
||
if envelope.GetEventType() == "RoomTreasureCountdownStarted" {
|
||
return s.handleRoomTreasureCountdown(eventCtx, envelope)
|
||
}
|
||
if envelope.GetEventType() != "RoomGiftSent" {
|
||
// 当前只消费礼物和房间宝箱事实；未来红包或运营事件应显式增加事件类型分支和测试。
|
||
return result, nil
|
||
}
|
||
var gift roomeventsv1.RoomGiftSent
|
||
if err := proto.Unmarshal(envelope.GetBody(), &gift); err != nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, err
|
||
}
|
||
if gift.GetVisibleRegionId() <= 0 || gift.GetGiftValue() < s.cfg.SuperGiftMinValue {
|
||
// visible_region_id 来自房间/主播可见区域，不能用客户端当前 IP 或本地时区推断。
|
||
return result, nil
|
||
}
|
||
broadcastEventID := superGiftBroadcastEventID(envelope, &gift)
|
||
payloadJSON, err := superGiftPayload(envelope, &gift, broadcastEventID, s.now().UTC().UnixMilli())
|
||
if err != nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, err
|
||
}
|
||
published, err := s.PublishRegionBroadcast(eventCtx, PublishInput{
|
||
EventID: broadcastEventID,
|
||
BroadcastType: broadcastdomain.TypeSuperGift,
|
||
RegionID: gift.GetVisibleRegionId(),
|
||
PayloadJSON: payloadJSON,
|
||
})
|
||
if err != nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, err
|
||
}
|
||
// 返回上游 event_id 和下游 broadcast_event_id，方便排查“房间事件已消费但播报是否生成”。
|
||
return broadcastdomain.ConsumeRoomEventResult{
|
||
EventID: envelope.GetEventId(),
|
||
Status: published.Status,
|
||
BroadcastEventID: published.EventID,
|
||
BroadcastCreated: published.Created,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) handleRoomTreasureCountdown(ctx context.Context, envelope *roomeventsv1.EventEnvelope) (broadcastdomain.ConsumeRoomEventResult, error) {
|
||
result := broadcastdomain.ConsumeRoomEventResult{EventID: envelope.GetEventId(), Status: broadcastdomain.StatusSkipped}
|
||
var treasure roomeventsv1.RoomTreasureCountdownStarted
|
||
if err := proto.Unmarshal(envelope.GetBody(), &treasure); err != nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, err
|
||
}
|
||
switch treasure.GetBroadcastScope() {
|
||
case broadcastdomain.ScopeRegion:
|
||
if treasure.GetVisibleRegionId() <= 0 {
|
||
return result, nil
|
||
}
|
||
case broadcastdomain.ScopeGlobal:
|
||
default:
|
||
return result, nil
|
||
}
|
||
|
||
broadcastEventID := roomTreasureBroadcastEventID(envelope, &treasure)
|
||
payloadJSON, err := roomTreasurePayload(envelope, &treasure, broadcastEventID, s.now().UTC().UnixMilli())
|
||
if err != nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, err
|
||
}
|
||
var published broadcastdomain.PublishResult
|
||
if treasure.GetBroadcastScope() == broadcastdomain.ScopeGlobal {
|
||
published, err = s.PublishGlobalBroadcast(ctx, PublishInput{
|
||
EventID: broadcastEventID,
|
||
BroadcastType: broadcastdomain.TypeRoomTreasure,
|
||
PayloadJSON: payloadJSON,
|
||
})
|
||
} else {
|
||
published, err = s.PublishRegionBroadcast(ctx, PublishInput{
|
||
EventID: broadcastEventID,
|
||
BroadcastType: broadcastdomain.TypeRoomTreasure,
|
||
RegionID: treasure.GetVisibleRegionId(),
|
||
PayloadJSON: payloadJSON,
|
||
})
|
||
}
|
||
if err != nil {
|
||
return broadcastdomain.ConsumeRoomEventResult{}, err
|
||
}
|
||
return broadcastdomain.ConsumeRoomEventResult{
|
||
EventID: envelope.GetEventId(),
|
||
Status: published.Status,
|
||
BroadcastEventID: published.EventID,
|
||
BroadcastCreated: published.Created,
|
||
}, nil
|
||
}
|
||
|
||
// WorkerOptions 描述一次 worker 批处理请求；cron 和常驻 worker 都走同一套参数归一化。
|
||
type WorkerOptions struct {
|
||
WorkerID string
|
||
BatchSize int
|
||
LockTTL time.Duration
|
||
MaxRetry int
|
||
PollInterval time.Duration
|
||
}
|
||
|
||
func normalizeConfig(cfg Config) Config {
|
||
cfg.NodeID = strings.TrimSpace(cfg.NodeID)
|
||
if cfg.NodeID == "" {
|
||
cfg.NodeID = "activity-broadcast"
|
||
}
|
||
if cfg.SuperGiftMinValue <= 0 {
|
||
cfg.SuperGiftMinValue = 100000
|
||
}
|
||
if cfg.WorkerBatchSize <= 0 {
|
||
cfg.WorkerBatchSize = defaultBroadcastBatchSize
|
||
}
|
||
if cfg.WorkerLockTTL <= 0 {
|
||
cfg.WorkerLockTTL = defaultBroadcastLockTTL
|
||
}
|
||
if cfg.WorkerMaxRetry <= 0 {
|
||
cfg.WorkerMaxRetry = defaultBroadcastMaxRetry
|
||
}
|
||
if cfg.WorkerPollInterval <= 0 {
|
||
cfg.WorkerPollInterval = time.Second
|
||
}
|
||
return cfg
|
||
}
|
||
|
||
func normalizeWorkerOptions(options WorkerOptions, cfg Config) WorkerOptions {
|
||
if strings.TrimSpace(options.WorkerID) == "" {
|
||
options.WorkerID = cfg.NodeID + "-broadcast"
|
||
}
|
||
if options.BatchSize <= 0 {
|
||
options.BatchSize = cfg.WorkerBatchSize
|
||
}
|
||
if options.LockTTL <= 0 {
|
||
options.LockTTL = cfg.WorkerLockTTL
|
||
}
|
||
if options.MaxRetry <= 0 {
|
||
options.MaxRetry = cfg.WorkerMaxRetry
|
||
}
|
||
if options.PollInterval <= 0 {
|
||
options.PollInterval = cfg.WorkerPollInterval
|
||
}
|
||
return options
|
||
}
|
||
|
||
func broadcastGroupSpec(groupID string) tencentim.GroupSpec {
|
||
return tencentim.GroupSpec{
|
||
GroupID: groupID,
|
||
Name: groupID,
|
||
Type: tencentim.DefaultGroupType,
|
||
// 入群表面允许 FreeAccess，真正准入由腾讯回调回查 user-service；这样客户端 SDK 体验简单，权限仍在服务端。
|
||
ApplyJoinOption: "FreeAccess",
|
||
}
|
||
}
|
||
|
||
func normalizedPayloadJSON(input string, forced map[string]any) (string, error) {
|
||
input = strings.TrimSpace(input)
|
||
if input == "" {
|
||
input = "{}"
|
||
}
|
||
var payload map[string]any
|
||
if err := json.Unmarshal([]byte(input), &payload); err != nil {
|
||
return "", xerr.New(xerr.InvalidArgument, "payload_json is invalid")
|
||
}
|
||
if payload == nil {
|
||
payload = map[string]any{}
|
||
}
|
||
for key, value := range forced {
|
||
if key == "region_id" && value.(int64) <= 0 {
|
||
// 全局播报不写 region_id，避免客户端误把 0 当作一个真实区域。
|
||
continue
|
||
}
|
||
// forced 字段覆盖调用方 payload，保证 event_id/scope/app_code/sent_at_ms 不被外部请求伪造。
|
||
payload[key] = value
|
||
}
|
||
encoded, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return string(encoded), nil
|
||
}
|
||
|
||
func superGiftBroadcastEventID(envelope *roomeventsv1.EventEnvelope, gift *roomeventsv1.RoomGiftSent) string {
|
||
if strings.TrimSpace(gift.GetCommandId()) != "" {
|
||
// command_id 是 SendGift 幂等键，用它生成播报 event_id 可避免同一房间事件重放时重复播报。
|
||
return fmt.Sprintf("gift_broadcast:%s:%s", envelope.GetRoomId(), gift.GetCommandId())
|
||
}
|
||
// 兼容没有 command_id 的旧/测试事件；当前开发阶段仍以 envelope event_id 保持幂等。
|
||
return "gift_broadcast:" + envelope.GetEventId()
|
||
}
|
||
|
||
func superGiftPayload(envelope *roomeventsv1.EventEnvelope, gift *roomeventsv1.RoomGiftSent, eventID string, sentAtMS int64) (string, error) {
|
||
payload := map[string]any{
|
||
"event_id": eventID,
|
||
"broadcast_type": broadcastdomain.TypeSuperGift,
|
||
"scope": broadcastdomain.ScopeRegion,
|
||
"app_code": envelope.GetAppCode(),
|
||
"region_id": gift.GetVisibleRegionId(),
|
||
"room_id": envelope.GetRoomId(),
|
||
"sender_user_id": gift.GetSenderUserId(),
|
||
"target_user_id": gift.GetTargetUserId(),
|
||
"gift_id": gift.GetGiftId(),
|
||
"gift_count": gift.GetGiftCount(),
|
||
"gift_value": gift.GetGiftValue(),
|
||
"sent_at_ms": sentAtMS,
|
||
"room_version": envelope.GetRoomVersion(),
|
||
"source_event_id": envelope.GetEventId(),
|
||
"action": map[string]any{
|
||
"type": "enter_room",
|
||
"room_id": envelope.GetRoomId(),
|
||
},
|
||
}
|
||
encoded, err := json.Marshal(payload)
|
||
return string(encoded), err
|
||
}
|
||
|
||
func roomTreasureBroadcastEventID(envelope *roomeventsv1.EventEnvelope, treasure *roomeventsv1.RoomTreasureCountdownStarted) string {
|
||
if strings.TrimSpace(treasure.GetBoxId()) != "" {
|
||
return fmt.Sprintf("room_treasure_broadcast:%s:%s", envelope.GetRoomId(), treasure.GetBoxId())
|
||
}
|
||
return "room_treasure_broadcast:" + envelope.GetEventId()
|
||
}
|
||
|
||
func roomTreasurePayload(envelope *roomeventsv1.EventEnvelope, treasure *roomeventsv1.RoomTreasureCountdownStarted, eventID string, sentAtMS int64) (string, error) {
|
||
payload := map[string]any{
|
||
"event_id": eventID,
|
||
"broadcast_type": broadcastdomain.TypeRoomTreasure,
|
||
"scope": treasure.GetBroadcastScope(),
|
||
"app_code": envelope.GetAppCode(),
|
||
"region_id": treasure.GetVisibleRegionId(),
|
||
"room_id": envelope.GetRoomId(),
|
||
"box_id": treasure.GetBoxId(),
|
||
"level": treasure.GetLevel(),
|
||
"open_at_ms": treasure.GetOpenAtMs(),
|
||
"top1_user_id": treasure.GetTop1UserId(),
|
||
"igniter_user_id": treasure.GetIgniterUserId(),
|
||
"sent_at_ms": sentAtMS,
|
||
"room_version": envelope.GetRoomVersion(),
|
||
"source_event_id": envelope.GetEventId(),
|
||
"current_progress": treasure.GetCurrentProgress(),
|
||
"energy_threshold": treasure.GetEnergyThreshold(),
|
||
"countdown_started_at_ms": treasure.GetCountdownStartedAtMs(),
|
||
"action": map[string]any{
|
||
"type": "enter_room",
|
||
"room_id": envelope.GetRoomId(),
|
||
},
|
||
}
|
||
encoded, err := json.Marshal(payload)
|
||
return string(encoded), err
|
||
}
|
||
|
||
func trimError(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if len(value) > 512 {
|
||
return value[:512]
|
||
}
|
||
return value
|
||
}
|