183 lines
7.0 KiB
Go
183 lines
7.0 KiB
Go
// Package roommq defines RocketMQ payloads owned by room-service.
|
||
package roommq
|
||
|
||
import (
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
)
|
||
|
||
const (
|
||
MessageTypeRoomOutboxEvent = "room_outbox_event"
|
||
MessageTypeRoomRocketLaunchDue = "room_rocket_launch_due"
|
||
|
||
// TagRoomOutboxEvent 是 typed tag 上线前所有房间事实共用的 legacy Tag。
|
||
// 迁移窗口内 consumer 必须订阅 legacy || relevant typed tags，等所有旧消息
|
||
// 和重试消息排空后才允许移除 legacy，不能直接切换造成存量事实不可见。
|
||
TagRoomOutboxEvent = "room_outbox_event"
|
||
|
||
// 以下事件类型是当前 room topic 下游实际使用的稳定事实名。producer 的
|
||
// typed 模式直接使用 envelope.event_type，consumer 复用这些常量拼 selector，
|
||
// 避免 producer/consumer 两侧手写字符串漂移后静默漏消费。
|
||
EventTypeRoomGiftSent = "RoomGiftSent"
|
||
EventTypeRoomUserJoined = "RoomUserJoined"
|
||
EventTypeRoomUserLeft = "RoomUserLeft"
|
||
EventTypeRoomUserKicked = "RoomUserKicked"
|
||
EventTypeRoomMicChanged = "RoomMicChanged"
|
||
EventTypeRoomRocketIgnited = "RoomRocketIgnited"
|
||
EventTypeRoomRocketRewardGranted = "RoomRocketRewardGranted"
|
||
EventTypeRoomPasswordChanged = "RoomPasswordChanged"
|
||
|
||
TagRoomRocketLaunchDue = "room_rocket_launch_due"
|
||
)
|
||
|
||
// EventTagMode 控制 room outbox 的 Broker Tag；它只改变 MQ 路由，不改变消息体协议。
|
||
type EventTagMode string
|
||
|
||
const (
|
||
// EventTagModeLegacy 是安全默认值，保证缺少新配置的旧环境继续发布统一 Tag。
|
||
EventTagModeLegacy EventTagMode = "legacy"
|
||
// EventTagModeEventType 只应在所有 consumer 完成兼容订阅滚动后显式启用。
|
||
EventTagModeEventType EventTagMode = "event_type"
|
||
)
|
||
|
||
// EventTypeTag 把一个受控的 room event_type 转成 RocketMQ Tag。事件类型来自
|
||
// owner outbox 而不是用户输入，但仍需拒绝表达式操作符和标点，避免坏数据扩大
|
||
// 订阅范围或发布成 consumer 无法安全表达的 Tag。
|
||
func EventTypeTag(eventType string) (string, error) {
|
||
tag := strings.TrimSpace(eventType)
|
||
if tag == "" {
|
||
return "", errors.New("room event_type tag is required")
|
||
}
|
||
if tag == TagRoomOutboxEvent {
|
||
return "", errors.New("room event_type cannot reuse the legacy room tag")
|
||
}
|
||
for _, char := range tag {
|
||
if (char >= 'a' && char <= 'z') ||
|
||
(char >= 'A' && char <= 'Z') ||
|
||
(char >= '0' && char <= '9') ||
|
||
char == '_' || char == '-' {
|
||
continue
|
||
}
|
||
return "", fmt.Errorf("room event_type %q cannot be used as a RocketMQ tag", eventType)
|
||
}
|
||
return tag, nil
|
||
}
|
||
|
||
// LegacyCompatibleTagExpression 构造无损迁移 selector：legacy 分支消费切换前
|
||
// 已存在 Broker 的消息，typed 分支只接收当前 consumer 真正处理的事件。排序和
|
||
// 去重保证同一 consumer group 的所有副本注册完全相同的订阅元数据。
|
||
func LegacyCompatibleTagExpression(eventTypes ...string) (string, error) {
|
||
typedTags := make([]string, 0, len(eventTypes))
|
||
seen := make(map[string]struct{}, len(eventTypes))
|
||
for _, eventType := range eventTypes {
|
||
tag, err := EventTypeTag(eventType)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if _, exists := seen[tag]; exists {
|
||
continue
|
||
}
|
||
seen[tag] = struct{}{}
|
||
typedTags = append(typedTags, tag)
|
||
}
|
||
if len(typedTags) == 0 {
|
||
return "", errors.New("at least one typed room event tag is required")
|
||
}
|
||
sort.Strings(typedTags)
|
||
return TagRoomOutboxEvent + " || " + strings.Join(typedTags, " || "), nil
|
||
}
|
||
|
||
// RoomOutboxMessage is the MQ representation of one durable room_outbox fact.
|
||
type RoomOutboxMessage struct {
|
||
MessageType string `json:"message_type"`
|
||
AppCode string `json:"app_code"`
|
||
EventID string `json:"event_id"`
|
||
EventType string `json:"event_type"`
|
||
RoomID string `json:"room_id"`
|
||
RoomVersion int64 `json:"room_version"`
|
||
OccurredAtMS int64 `json:"occurred_at_ms"`
|
||
Envelope []byte `json:"envelope"`
|
||
}
|
||
|
||
// RoomRocketLaunchDueMessage wakes room-service at the configured launch_at_ms.
|
||
type RoomRocketLaunchDueMessage struct {
|
||
MessageType string `json:"message_type"`
|
||
AppCode string `json:"app_code"`
|
||
RoomID string `json:"room_id"`
|
||
RocketID string `json:"rocket_id"`
|
||
Level int32 `json:"level"`
|
||
LaunchAtMS int64 `json:"launch_at_ms"`
|
||
ResetAtMS int64 `json:"reset_at_ms"`
|
||
CommandID string `json:"command_id"`
|
||
}
|
||
|
||
// EncodeRoomOutboxMessage serializes the protobuf envelope into an MQ JSON body.
|
||
func EncodeRoomOutboxMessage(envelope *roomeventsv1.EventEnvelope) ([]byte, error) {
|
||
if envelope == nil {
|
||
return nil, errors.New("room outbox envelope is required")
|
||
}
|
||
envelopeBytes, err := proto.Marshal(envelope)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return json.Marshal(RoomOutboxMessage{
|
||
MessageType: MessageTypeRoomOutboxEvent,
|
||
AppCode: envelope.GetAppCode(),
|
||
EventID: envelope.GetEventId(),
|
||
EventType: envelope.GetEventType(),
|
||
RoomID: envelope.GetRoomId(),
|
||
RoomVersion: envelope.GetRoomVersion(),
|
||
OccurredAtMS: envelope.GetOccurredAtMs(),
|
||
Envelope: envelopeBytes,
|
||
})
|
||
}
|
||
|
||
// DecodeRoomOutboxMessage validates the MQ body and restores the protobuf fact.
|
||
func DecodeRoomOutboxMessage(body []byte) (*roomeventsv1.EventEnvelope, RoomOutboxMessage, error) {
|
||
var message RoomOutboxMessage
|
||
if err := json.Unmarshal(body, &message); err != nil {
|
||
return nil, RoomOutboxMessage{}, err
|
||
}
|
||
if message.MessageType != MessageTypeRoomOutboxEvent {
|
||
return nil, RoomOutboxMessage{}, errors.New("unexpected room outbox message_type")
|
||
}
|
||
var envelope roomeventsv1.EventEnvelope
|
||
if err := proto.Unmarshal(message.Envelope, &envelope); err != nil {
|
||
return nil, RoomOutboxMessage{}, err
|
||
}
|
||
if envelope.GetEventId() == "" || envelope.GetEventType() == "" || envelope.GetRoomId() == "" {
|
||
return nil, RoomOutboxMessage{}, errors.New("room outbox envelope is incomplete")
|
||
}
|
||
return &envelope, message, nil
|
||
}
|
||
|
||
// EncodeRoomRocketLaunchDueMessage serializes a launch wakeup command.
|
||
func EncodeRoomRocketLaunchDueMessage(message RoomRocketLaunchDueMessage) ([]byte, error) {
|
||
message.MessageType = MessageTypeRoomRocketLaunchDue
|
||
if message.AppCode == "" || message.RoomID == "" || message.RocketID == "" || message.Level <= 0 || message.LaunchAtMS <= 0 {
|
||
return nil, errors.New("room rocket launch due message is incomplete")
|
||
}
|
||
return json.Marshal(message)
|
||
}
|
||
|
||
// DecodeRoomRocketLaunchDueMessage validates a launch wakeup body.
|
||
func DecodeRoomRocketLaunchDueMessage(body []byte) (RoomRocketLaunchDueMessage, error) {
|
||
var message RoomRocketLaunchDueMessage
|
||
if err := json.Unmarshal(body, &message); err != nil {
|
||
return RoomRocketLaunchDueMessage{}, err
|
||
}
|
||
if message.MessageType != MessageTypeRoomRocketLaunchDue {
|
||
return RoomRocketLaunchDueMessage{}, errors.New("unexpected room rocket message_type")
|
||
}
|
||
if message.AppCode == "" || message.RoomID == "" || message.RocketID == "" || message.Level <= 0 || message.LaunchAtMS <= 0 {
|
||
return RoomRocketLaunchDueMessage{}, errors.New("room rocket launch due message is incomplete")
|
||
}
|
||
return message, nil
|
||
}
|