56 lines
1.8 KiB
Go
56 lines
1.8 KiB
Go
package integration
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"strings"
|
||
|
||
luckygifteventsv1 "hyapp.local/api/proto/events/luckygift/v1"
|
||
"hyapp/pkg/luckygiftmq"
|
||
"hyapp/pkg/rocketmqx"
|
||
)
|
||
|
||
// RocketMQEventPublisher 把已经完成钱包入账的幸运礼物事实发布到 owner topic。
|
||
// 它不读取 outbox 表；claim、重试和 delivered 状态仍由 lucky-gift-service 的 settlement worker 独占。
|
||
type RocketMQEventPublisher struct {
|
||
producer *rocketmqx.Producer
|
||
topic string
|
||
}
|
||
|
||
func NewRocketMQEventPublisher(producer *rocketmqx.Producer, topic string) *RocketMQEventPublisher {
|
||
return &RocketMQEventPublisher{producer: producer, topic: strings.TrimSpace(topic)}
|
||
}
|
||
|
||
// PublishLuckyGiftEvent 只有在 Broker ACK 后返回成功；worker 随后才能把 owner outbox 标为 delivered。
|
||
func (p *RocketMQEventPublisher) PublishLuckyGiftEvent(ctx context.Context, envelope *luckygifteventsv1.EventEnvelope) error {
|
||
if p == nil || p.producer == nil {
|
||
return errors.New("lucky gift RocketMQ producer is not configured")
|
||
}
|
||
message, err := luckyGiftRocketMessage(p.topic, envelope)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return p.producer.SendSync(ctx, message)
|
||
}
|
||
|
||
func luckyGiftRocketMessage(topic string, envelope *luckygifteventsv1.EventEnvelope) (rocketmqx.Message, error) {
|
||
topic = strings.TrimSpace(topic)
|
||
if topic == "" {
|
||
return rocketmqx.Message{}, errors.New("lucky gift outbox topic is required")
|
||
}
|
||
body, err := luckygiftmq.EncodeLuckyGiftOutboxMessage(envelope)
|
||
if err != nil {
|
||
return rocketmqx.Message{}, err
|
||
}
|
||
tag, err := luckygiftmq.EventTypeTag(envelope.GetEventType())
|
||
if err != nil {
|
||
return rocketmqx.Message{}, err
|
||
}
|
||
return rocketmqx.Message{
|
||
Topic: topic,
|
||
Tag: tag,
|
||
Keys: []string{envelope.GetEventId(), envelope.GetDrawId()},
|
||
Body: body,
|
||
}, nil
|
||
}
|