125 lines
4.7 KiB
Go
125 lines
4.7 KiB
Go
package integration
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strconv"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
"hyapp/pkg/rocketmqx"
|
||
"hyapp/pkg/roommq"
|
||
)
|
||
|
||
// RocketMQOutboxPublisher publishes durable room_outbox facts to a RocketMQ topic.
|
||
type RocketMQOutboxPublisher struct {
|
||
producer *rocketmqx.Producer
|
||
topic string
|
||
tagMode roommq.EventTagMode
|
||
}
|
||
|
||
// NewRocketMQOutboxPublisher creates a room_outbox fanout publisher.
|
||
func NewRocketMQOutboxPublisher(producer *rocketmqx.Producer, topic string, tagMode roommq.EventTagMode) *RocketMQOutboxPublisher {
|
||
return &RocketMQOutboxPublisher{producer: producer, topic: topic, tagMode: tagMode}
|
||
}
|
||
|
||
// PublishOutboxEvent implements OutboxPublisher.
|
||
func (p *RocketMQOutboxPublisher) PublishOutboxEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error {
|
||
message, err := roomOutboxRocketMessage(p.topic, p.tagMode, envelope)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return p.producer.SendSync(ctx, message)
|
||
}
|
||
|
||
// roomOutboxRocketMessage keeps envelope.event_type canonical in both migration phases.
|
||
// Phase A publishes the legacy Tag, while explicit Phase B uses the exact event_type;
|
||
// the body stays byte-compatible so old replay tools and handlers require no protocol fork.
|
||
func roomOutboxRocketMessage(topic string, tagMode roommq.EventTagMode, envelope *roomeventsv1.EventEnvelope) (rocketmqx.Message, error) {
|
||
if envelope == nil {
|
||
return rocketmqx.Message{}, fmt.Errorf("room outbox envelope is required")
|
||
}
|
||
eventTag, err := roommq.EventTypeTag(envelope.GetEventType())
|
||
if err != nil {
|
||
return rocketmqx.Message{}, err
|
||
}
|
||
publishTag := roommq.TagRoomOutboxEvent
|
||
switch tagMode {
|
||
case "", roommq.EventTagModeLegacy:
|
||
// 空值覆盖旧测试构造和旧配置；只有显式 event_type 才允许进入 Phase B。
|
||
case roommq.EventTagModeEventType:
|
||
publishTag = eventTag
|
||
default:
|
||
return rocketmqx.Message{}, fmt.Errorf("unsupported room event tag mode %q", tagMode)
|
||
}
|
||
// outbox 历史数据可能含首尾空白；Tag 与消息体必须使用同一规范 event_type，
|
||
// 但不能原地修改 repository 持有的 envelope，避免 composite publisher 看到副作用。
|
||
canonicalEnvelope := proto.Clone(envelope).(*roomeventsv1.EventEnvelope)
|
||
canonicalEnvelope.EventType = eventTag
|
||
body, err := roommq.EncodeRoomOutboxMessage(canonicalEnvelope)
|
||
if err != nil {
|
||
return rocketmqx.Message{}, err
|
||
}
|
||
return rocketmqx.Message{
|
||
Topic: topic,
|
||
Tag: publishTag,
|
||
Keys: []string{canonicalEnvelope.GetEventId(), canonicalEnvelope.GetRoomId()},
|
||
Body: body,
|
||
Properties: map[string]string{
|
||
"message_type": roommq.MessageTypeRoomOutboxEvent,
|
||
"app_code": canonicalEnvelope.GetAppCode(),
|
||
"room_id": canonicalEnvelope.GetRoomId(),
|
||
"event_id": canonicalEnvelope.GetEventId(),
|
||
"event_type": eventTag,
|
||
"room_version": strconv.FormatInt(canonicalEnvelope.GetRoomVersion(), 10),
|
||
"occurred_at_ms": strconv.FormatInt(canonicalEnvelope.GetOccurredAtMs(), 10),
|
||
"source_database": "hyapp_room",
|
||
"source_table": "room_outbox",
|
||
"delivery_version": "1",
|
||
},
|
||
}, nil
|
||
}
|
||
|
||
// RocketMQRocketLaunchScheduler publishes exact-time rocket wakeup messages.
|
||
type RocketMQRocketLaunchScheduler struct {
|
||
producer *rocketmqx.Producer
|
||
topic string
|
||
}
|
||
|
||
// NewRocketMQRocketLaunchScheduler creates the delayed launch scheduler.
|
||
func NewRocketMQRocketLaunchScheduler(producer *rocketmqx.Producer, topic string) *RocketMQRocketLaunchScheduler {
|
||
return &RocketMQRocketLaunchScheduler{producer: producer, topic: topic}
|
||
}
|
||
|
||
// ScheduleRoomRocketLaunch implements RoomRocketLaunchScheduler.
|
||
func (s *RocketMQRocketLaunchScheduler) ScheduleRoomRocketLaunch(ctx context.Context, schedule RoomRocketLaunchSchedule) error {
|
||
body, err := roommq.EncodeRoomRocketLaunchDueMessage(roommq.RoomRocketLaunchDueMessage{
|
||
AppCode: schedule.AppCode,
|
||
RoomID: schedule.RoomID,
|
||
RocketID: schedule.RocketID,
|
||
Level: schedule.Level,
|
||
LaunchAtMS: schedule.LaunchAtMS,
|
||
ResetAtMS: schedule.ResetAtMS,
|
||
CommandID: schedule.CommandID,
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return s.producer.SendSync(ctx, rocketmqx.Message{
|
||
Topic: s.topic,
|
||
Tag: roommq.TagRoomRocketLaunchDue,
|
||
Keys: []string{schedule.RocketID, schedule.RoomID},
|
||
Body: body,
|
||
DeliveryAtMS: schedule.LaunchAtMS,
|
||
Properties: map[string]string{
|
||
"message_type": roommq.MessageTypeRoomRocketLaunchDue,
|
||
"app_code": schedule.AppCode,
|
||
"room_id": schedule.RoomID,
|
||
"rocket_id": schedule.RocketID,
|
||
"level": strconv.FormatInt(int64(schedule.Level), 10),
|
||
"launch_at_ms": strconv.FormatInt(schedule.LaunchAtMS, 10),
|
||
"reset_at_ms": strconv.FormatInt(schedule.ResetAtMS, 10),
|
||
},
|
||
})
|
||
}
|