112 lines
3.5 KiB
Go
112 lines
3.5 KiB
Go
package broadcast
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
|
|
"hyapp/pkg/appcode"
|
|
broadcastdomain "hyapp/services/activity-service/internal/domain/broadcast"
|
|
)
|
|
|
|
// ConsumeRedPacketWalletEvent converts one MQ-delivered wallet red-packet fact into IM outbox records.
|
|
func (s *Service) ConsumeRedPacketWalletEvent(ctx context.Context, event broadcastdomain.RedPacketWalletEvent) error {
|
|
if s == nil {
|
|
return nil
|
|
}
|
|
ctx = appcode.WithContext(ctx, event.AppCode)
|
|
switch event.EventType {
|
|
case "WalletRedPacketCreated":
|
|
return s.consumeRedPacketCreated(ctx, event)
|
|
case "WalletRedPacketClaimed":
|
|
if strings.EqualFold(event.Status, "finished") {
|
|
return s.publishRedPacketRoomEvent(ctx, event, "STATUS_CHANGED", event.CreatedAtMS)
|
|
}
|
|
case "WalletRedPacketRefunded":
|
|
return s.publishRedPacketRoomEvent(ctx, event, "STATUS_CHANGED", event.CreatedAtMS)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) ConsumeRedPacketCreatedEvent(ctx context.Context, event broadcastdomain.RedPacketWalletEvent) error {
|
|
if event.EventType == "" {
|
|
event.EventType = "WalletRedPacketCreated"
|
|
}
|
|
return s.ConsumeRedPacketWalletEvent(ctx, event)
|
|
}
|
|
|
|
func (s *Service) consumeRedPacketCreated(ctx context.Context, event broadcastdomain.RedPacketWalletEvent) error {
|
|
if event.RoomID == "" {
|
|
return nil
|
|
}
|
|
if err := s.publishRedPacketRoomEvent(ctx, event, "CREATED", event.CreatedAtMS); err != nil {
|
|
return err
|
|
}
|
|
if strings.EqualFold(event.PacketType, "delayed") && event.RegionID > 0 {
|
|
if _, err := s.PublishRegionBroadcast(ctx, PublishInput{
|
|
EventID: event.EventID + ":region_created",
|
|
BroadcastType: broadcastdomain.TypeRedPacket,
|
|
RegionID: event.RegionID,
|
|
PayloadJSON: redPacketIMPayload(event, "CREATED", "REGION"),
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if strings.EqualFold(event.PacketType, "delayed") && event.OpenAtMS > 0 {
|
|
_, err := s.PublishRoomBroadcast(ctx, event.RoomID, PublishInput{
|
|
EventID: event.EventID + ":open_due",
|
|
BroadcastType: broadcastdomain.TypeRedPacket,
|
|
RegionID: event.RegionID,
|
|
PayloadJSON: redPacketIMPayload(event, "OPEN_DUE", "ROOM"),
|
|
DeliverAtMS: event.OpenAtMS,
|
|
})
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) publishRedPacketRoomEvent(ctx context.Context, event broadcastdomain.RedPacketWalletEvent, eventName string, fallbackDeliverAtMS int64) error {
|
|
if event.RoomID == "" {
|
|
return nil
|
|
}
|
|
_, err := s.PublishRoomBroadcast(ctx, event.RoomID, PublishInput{
|
|
EventID: event.EventID,
|
|
BroadcastType: broadcastdomain.TypeRedPacket,
|
|
RegionID: event.RegionID,
|
|
PayloadJSON: redPacketIMPayload(event, eventName, "ROOM"),
|
|
DeliverAtMS: fallbackDeliverAtMS,
|
|
})
|
|
return err
|
|
}
|
|
|
|
func redPacketIMPayload(event broadcastdomain.RedPacketWalletEvent, eventName string, scope string) string {
|
|
packetMode := "IMMEDIATE"
|
|
if strings.EqualFold(event.PacketType, "delayed") {
|
|
packetMode = "DELAYED"
|
|
}
|
|
status := strings.ToUpper(strings.TrimSpace(event.Status))
|
|
if status == "" {
|
|
status = "ACTIVE"
|
|
}
|
|
payload := map[string]any{
|
|
"type": "ROOM_RED_PACKET",
|
|
"data": map[string]any{
|
|
"event": eventName,
|
|
"scope": scope,
|
|
"packetNo": event.PacketID,
|
|
"roomId": event.RoomID,
|
|
"regionId": event.RegionID,
|
|
"packetMode": packetMode,
|
|
"status": status,
|
|
"claimStartTime": event.OpenAtMS,
|
|
"expireTime": event.ExpiresAtMS,
|
|
"serverTime": event.CreatedAtMS,
|
|
},
|
|
}
|
|
encoded, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return `{"type":"ROOM_RED_PACKET","data":{}}`
|
|
}
|
|
return string(encoded)
|
|
}
|