591 lines
21 KiB
Go
591 lines
21 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
|
|
activityv1 "hyapp.local/api/proto/activity/v1"
|
|
walletv1 "hyapp.local/api/proto/wallet/v1"
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/grpchealth"
|
|
"hyapp/pkg/grpcshutdown"
|
|
"hyapp/pkg/healthhttp"
|
|
"hyapp/pkg/logx"
|
|
"hyapp/pkg/rocketmqx"
|
|
"hyapp/pkg/roommq"
|
|
"hyapp/pkg/tencentim"
|
|
"hyapp/pkg/walletmq"
|
|
"hyapp/services/activity-service/internal/client"
|
|
"hyapp/services/activity-service/internal/config"
|
|
broadcastdomain "hyapp/services/activity-service/internal/domain/broadcast"
|
|
firstrechargedomain "hyapp/services/activity-service/internal/domain/firstrecharge"
|
|
achievementservice "hyapp/services/activity-service/internal/service/achievement"
|
|
activityservice "hyapp/services/activity-service/internal/service/activity"
|
|
broadcastservice "hyapp/services/activity-service/internal/service/broadcast"
|
|
firstrechargeservice "hyapp/services/activity-service/internal/service/firstrecharge"
|
|
growthservice "hyapp/services/activity-service/internal/service/growth"
|
|
luckygiftservice "hyapp/services/activity-service/internal/service/luckygift"
|
|
messageservice "hyapp/services/activity-service/internal/service/message"
|
|
registrationrewardservice "hyapp/services/activity-service/internal/service/registrationreward"
|
|
sevendaycheckinservice "hyapp/services/activity-service/internal/service/sevendaycheckin"
|
|
taskservice "hyapp/services/activity-service/internal/service/task"
|
|
mysqlstorage "hyapp/services/activity-service/internal/storage/mysql"
|
|
grpcserver "hyapp/services/activity-service/internal/transport/grpc"
|
|
)
|
|
|
|
// App 装配 activity-service gRPC 入口和底座依赖。
|
|
type App struct {
|
|
server *grpc.Server
|
|
listener net.Listener
|
|
health *grpchealth.ServingChecker
|
|
healthHTTP *healthhttp.Server
|
|
mysqlRepo *mysqlstorage.Repository
|
|
broadcast *broadcastservice.Service
|
|
luckyGift *luckygiftservice.Service
|
|
firstRechargeReward *firstrechargeservice.Service
|
|
broadcastWorkerEnabled bool
|
|
luckyGiftWorkerEnabled bool
|
|
workerNodeID string
|
|
luckyGiftWorkerOptions luckygiftservice.WorkerOptions
|
|
mqConsumers []*rocketmqx.Consumer
|
|
userConn *grpc.ClientConn
|
|
walletConn *grpc.ClientConn
|
|
workerCtx context.Context
|
|
workerStop context.CancelFunc
|
|
workerWG sync.WaitGroup
|
|
closeOnce sync.Once
|
|
}
|
|
|
|
// New 初始化 activity-service 应用。
|
|
func New(cfg config.Config) (*App, error) {
|
|
startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
// activity-service 的活动状态、消费幂等和 outbox 都必须使用 MySQL。
|
|
repository, err := mysqlstorage.Open(startupCtx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if cfg.MySQLAutoMigrate {
|
|
if err := repository.Migrate(startupCtx); err != nil {
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
listener, err := net.Listen("tcp", cfg.GRPCAddr)
|
|
if err != nil {
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
userConn, err := grpc.Dial(cfg.UserServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
walletConn, err := grpc.Dial(cfg.WalletServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
|
|
server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("activity-service")))
|
|
svc := activityservice.New(activityservice.Config{NodeID: cfg.NodeID}, repository)
|
|
messageSvc := messageservice.New(messageservice.Config{NodeID: cfg.NodeID}, repository, messageservice.WithTargetSource(client.NewGRPCUserTargetSource(userConn)))
|
|
taskSvc := taskservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
registrationRewardSvc := registrationrewardservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
sevenDayCheckInSvc := sevendaycheckinservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
growthSvc := growthservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
achievementSvc := achievementservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
var tencentClient *tencentim.RESTClient
|
|
var broadcastPublisher broadcastservice.Publisher
|
|
if cfg.TencentIM.Enabled {
|
|
tencentClient, err = tencentim.NewRESTClient(cfg.TencentIM.RESTConfig())
|
|
if err != nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
broadcastPublisher = tencentClient
|
|
}
|
|
if cfg.Broadcast.Enabled && broadcastPublisher == nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, errors.New("broadcast worker requires tencent_im.enabled")
|
|
}
|
|
broadcastSvc := broadcastservice.New(broadcastservice.Config{
|
|
NodeID: cfg.NodeID,
|
|
SuperGiftMinValue: cfg.Broadcast.SuperGiftMinValue,
|
|
WorkerBatchSize: cfg.Broadcast.WorkerBatchSize,
|
|
WorkerLockTTL: cfg.Broadcast.WorkerLockTTL,
|
|
WorkerMaxRetry: cfg.Broadcast.WorkerMaxRetry,
|
|
WorkerPollInterval: cfg.Broadcast.WorkerPollInterval,
|
|
EnsureGroupsOnStartup: cfg.Broadcast.EnsureGroupsOnStartup,
|
|
GroupIDPrefix: cfg.TencentIM.GroupIDPrefix,
|
|
}, repository, broadcastPublisher, client.NewGRPCRegionSource(userConn))
|
|
broadcastSvc.SetSenderProfileSource(client.NewGRPCUserProfileSource(userConn))
|
|
luckyGiftSvc := luckygiftservice.New(repository,
|
|
luckygiftservice.WithWallet(walletv1.NewWalletServiceClient(walletConn)),
|
|
luckygiftservice.WithRoomPublisher(tencentClient),
|
|
luckygiftservice.WithRegionBroadcaster(broadcastSvc),
|
|
)
|
|
activityv1.RegisterActivityServiceServer(server, grpcserver.NewServer(svc))
|
|
activityv1.RegisterMessageInboxServiceServer(server, grpcserver.NewMessageServer(messageSvc))
|
|
activityv1.RegisterActivityCronServiceServer(server, grpcserver.NewCronServer(messageSvc, growthSvc, achievementSvc))
|
|
activityv1.RegisterTaskServiceServer(server, grpcserver.NewTaskServer(taskSvc))
|
|
activityv1.RegisterAdminTaskServiceServer(server, grpcserver.NewAdminTaskServer(taskSvc))
|
|
activityv1.RegisterGrowthLevelServiceServer(server, grpcserver.NewGrowthLevelServer(growthSvc))
|
|
activityv1.RegisterAdminGrowthLevelServiceServer(server, grpcserver.NewAdminGrowthLevelServer(growthSvc))
|
|
activityv1.RegisterAchievementServiceServer(server, grpcserver.NewAchievementServer(achievementSvc))
|
|
activityv1.RegisterAdminAchievementServiceServer(server, grpcserver.NewAdminAchievementServer(achievementSvc))
|
|
activityv1.RegisterLuckyGiftServiceServer(server, grpcserver.NewLuckyGiftServer(luckyGiftSvc))
|
|
activityv1.RegisterAdminLuckyGiftServiceServer(server, grpcserver.NewAdminLuckyGiftServer(luckyGiftSvc))
|
|
activityv1.RegisterRegistrationRewardServiceServer(server, grpcserver.NewRegistrationRewardServer(registrationRewardSvc))
|
|
activityv1.RegisterAdminRegistrationRewardServiceServer(server, grpcserver.NewAdminRegistrationRewardServer(registrationRewardSvc))
|
|
activityv1.RegisterSevenDayCheckInServiceServer(server, grpcserver.NewSevenDayCheckInServer(sevenDayCheckInSvc))
|
|
activityv1.RegisterAdminSevenDayCheckInServiceServer(server, grpcserver.NewAdminSevenDayCheckInServer(sevenDayCheckInSvc))
|
|
broadcastServer := grpcserver.NewBroadcastServer(broadcastSvc, growthSvc)
|
|
activityv1.RegisterBroadcastServiceServer(server, broadcastServer)
|
|
activityv1.RegisterRoomEventConsumerServiceServer(server, broadcastServer)
|
|
firstRechargeRewardSvc := firstrechargeservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
activityv1.RegisterFirstRechargeRewardServiceServer(server, grpcserver.NewFirstRechargeRewardServer(firstRechargeRewardSvc))
|
|
activityv1.RegisterAdminFirstRechargeRewardServiceServer(server, grpcserver.NewAdminFirstRechargeRewardServer(firstRechargeRewardSvc))
|
|
mqConsumers := make([]*rocketmqx.Consumer, 0, 3)
|
|
if cfg.RocketMQ.RoomOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(roomOutboxConsumerConfig(cfg.RocketMQ))
|
|
if err != nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
envelope, _, err := roommq.DecodeRoomOutboxMessage(message.Body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
eventCtx := appcode.WithContext(ctx, envelope.GetAppCode())
|
|
if _, err := broadcastSvc.HandleRoomEvent(eventCtx, envelope); err != nil {
|
|
return err
|
|
}
|
|
_, err = growthSvc.HandleRoomEvent(eventCtx, envelope)
|
|
return err
|
|
}); err != nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
if cfg.FirstRechargeRewardWorker.Enabled && cfg.RocketMQ.WalletOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(walletOutboxConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.FirstRechargeConsumerGroup))
|
|
if err != nil {
|
|
shutdownConsumers(mqConsumers)
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
event, ok, err := firstRechargeEventFromWalletMessage(message.Body)
|
|
if err != nil || !ok {
|
|
return err
|
|
}
|
|
return firstRechargeRewardSvc.ConsumeWalletRechargeEvent(appcode.WithContext(ctx, event.AppCode), event)
|
|
}); err != nil {
|
|
_ = consumer.Shutdown()
|
|
shutdownConsumers(mqConsumers)
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
if cfg.RedPacketBroadcastWorker.Enabled && cfg.RocketMQ.WalletOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(walletOutboxConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.RedPacketBroadcastConsumerGroup))
|
|
if err != nil {
|
|
shutdownConsumers(mqConsumers)
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
event, ok, err := redPacketEventFromWalletMessage(message.Body)
|
|
if err != nil || !ok {
|
|
return err
|
|
}
|
|
return broadcastSvc.ConsumeRedPacketWalletEvent(appcode.WithContext(ctx, event.AppCode), event)
|
|
}); err != nil {
|
|
_ = consumer.Shutdown()
|
|
shutdownConsumers(mqConsumers)
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
healthDependencies := []grpchealth.Dependency{{
|
|
Name: "mysql",
|
|
Check: repository.Ping,
|
|
}}
|
|
health := grpchealth.NewServingChecker("activity-service", healthDependencies...)
|
|
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
|
|
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
|
|
if err != nil {
|
|
shutdownConsumers(mqConsumers)
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
workerCtx, workerStop := context.WithCancel(context.Background())
|
|
|
|
return &App{
|
|
server: server,
|
|
listener: listener,
|
|
health: health,
|
|
healthHTTP: healthHTTP,
|
|
mysqlRepo: repository,
|
|
broadcast: broadcastSvc,
|
|
luckyGift: luckyGiftSvc,
|
|
firstRechargeReward: firstRechargeRewardSvc,
|
|
broadcastWorkerEnabled: cfg.Broadcast.Enabled,
|
|
luckyGiftWorkerEnabled: cfg.LuckyGiftWorker.Enabled,
|
|
workerNodeID: cfg.NodeID,
|
|
luckyGiftWorkerOptions: luckyGiftWorkerOptions(cfg.NodeID, cfg.LuckyGiftWorker),
|
|
mqConsumers: mqConsumers,
|
|
userConn: userConn,
|
|
walletConn: walletConn,
|
|
workerCtx: workerCtx,
|
|
workerStop: workerStop,
|
|
}, nil
|
|
}
|
|
|
|
// Run 启动 gRPC 服务。
|
|
func (a *App) Run() error {
|
|
a.runHealthHTTP()
|
|
a.health.MarkServing()
|
|
if err := a.startMQ(); err != nil {
|
|
a.health.MarkStopped()
|
|
return err
|
|
}
|
|
if a.broadcastWorkerEnabled && a.broadcast != nil && a.workerCtx != nil {
|
|
a.workerWG.Go(func() {
|
|
a.broadcast.RunWorker(a.workerCtx, broadcastservice.WorkerOptions{})
|
|
})
|
|
}
|
|
if a.luckyGiftWorkerEnabled && a.luckyGift != nil && a.workerCtx != nil {
|
|
a.workerWG.Go(func() {
|
|
a.luckyGift.RunWorker(a.workerCtx, a.luckyGiftWorkerOptions)
|
|
})
|
|
}
|
|
err := a.server.Serve(a.listener)
|
|
if a.workerStop != nil {
|
|
a.workerStop()
|
|
}
|
|
a.workerWG.Wait()
|
|
a.shutdownMQ()
|
|
a.health.MarkStopped()
|
|
if errors.Is(err, grpc.ErrServerStopped) {
|
|
return nil
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Close 关闭 gRPC 服务。
|
|
func (a *App) Close() {
|
|
a.closeOnce.Do(func() {
|
|
a.health.MarkDraining()
|
|
if a.workerStop != nil {
|
|
a.workerStop()
|
|
}
|
|
a.workerWG.Wait()
|
|
a.shutdownMQ()
|
|
grpcshutdown.GracefulStop(a.server, 15*time.Second)
|
|
a.closeHealthHTTP()
|
|
if a.userConn != nil {
|
|
_ = a.userConn.Close()
|
|
}
|
|
if a.walletConn != nil {
|
|
_ = a.walletConn.Close()
|
|
}
|
|
if a.mysqlRepo != nil {
|
|
// MySQL 连接池最后关闭，保证 drain 中的查询或消费提交能完成。
|
|
_ = a.mysqlRepo.Close()
|
|
}
|
|
})
|
|
}
|
|
|
|
func (a *App) runHealthHTTP() {
|
|
if a.healthHTTP == nil {
|
|
return
|
|
}
|
|
go func() {
|
|
if err := a.healthHTTP.Run(); err != nil {
|
|
logx.Error(context.Background(), "health_http_run_failed", err)
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (a *App) closeHealthHTTP() {
|
|
if a.healthHTTP == nil {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
_ = a.healthHTTP.Close(ctx)
|
|
}
|
|
|
|
func (a *App) startMQ() error {
|
|
for _, consumer := range a.mqConsumers {
|
|
if err := consumer.Start(); err != nil {
|
|
a.shutdownMQ()
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *App) shutdownMQ() {
|
|
for _, consumer := range a.mqConsumers {
|
|
if err := consumer.Shutdown(); err != nil {
|
|
logx.Warn(context.Background(), "rocketmq_consumer_shutdown_failed", slog.String("error", err.Error()))
|
|
}
|
|
}
|
|
}
|
|
|
|
func roomOutboxConsumerConfig(cfg config.RocketMQConfig) rocketmqx.ConsumerConfig {
|
|
return rocketMQConsumerConfig(cfg, cfg.RoomOutbox.ConsumerGroup, cfg.RoomOutbox.ConsumerMaxReconsumeTimes)
|
|
}
|
|
|
|
func walletOutboxConsumerConfig(cfg config.RocketMQConfig, group string) rocketmqx.ConsumerConfig {
|
|
return rocketMQConsumerConfig(cfg, group, cfg.WalletOutbox.ConsumerMaxReconsumeTimes)
|
|
}
|
|
|
|
func rocketMQConsumerConfig(cfg config.RocketMQConfig, group string, maxReconsume int32) rocketmqx.ConsumerConfig {
|
|
return rocketmqx.ConsumerConfig{
|
|
EndpointConfig: rocketmqx.EndpointConfig{
|
|
NameServers: cfg.NameServers,
|
|
NameServerDomain: cfg.NameServerDomain,
|
|
AccessKey: cfg.AccessKey,
|
|
SecretKey: cfg.SecretKey,
|
|
SecurityToken: cfg.SecurityToken,
|
|
Namespace: cfg.Namespace,
|
|
VIPChannel: cfg.VIPChannel,
|
|
},
|
|
GroupName: group,
|
|
MaxReconsumeTimes: maxReconsume,
|
|
ConsumeRetryDelay: time.Second,
|
|
ConsumePullBatch: 32,
|
|
ConsumePullTimeout: 15 * time.Minute,
|
|
}
|
|
}
|
|
|
|
func shutdownConsumers(consumers []*rocketmqx.Consumer) {
|
|
for _, consumer := range consumers {
|
|
_ = consumer.Shutdown()
|
|
}
|
|
}
|
|
|
|
func firstRechargeEventFromWalletMessage(body []byte) (firstrechargedomain.RechargeEvent, bool, error) {
|
|
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
|
if err != nil {
|
|
return firstrechargedomain.RechargeEvent{}, false, err
|
|
}
|
|
if message.EventType != firstrechargedomain.RechargeEventWalletRecorded {
|
|
return firstrechargedomain.RechargeEvent{}, false, nil
|
|
}
|
|
sequence, rechargeType := rechargeFactFromWalletPayload(message.PayloadJSON)
|
|
return firstrechargedomain.RechargeEvent{
|
|
AppCode: appcode.Normalize(message.AppCode),
|
|
EventID: message.EventID,
|
|
EventType: message.EventType,
|
|
TransactionID: message.TransactionID,
|
|
CommandID: message.CommandID,
|
|
UserID: message.UserID,
|
|
RechargeCoinAmount: message.AvailableDelta,
|
|
RechargeSequence: sequence,
|
|
RechargeType: rechargeType,
|
|
PayloadJSON: message.PayloadJSON,
|
|
OccurredAtMS: message.OccurredAtMS,
|
|
}, true, nil
|
|
}
|
|
|
|
func redPacketEventFromWalletMessage(body []byte) (broadcastdomain.RedPacketWalletEvent, bool, error) {
|
|
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
|
if err != nil {
|
|
return broadcastdomain.RedPacketWalletEvent{}, false, err
|
|
}
|
|
if message.EventType != "WalletRedPacketCreated" && message.EventType != "WalletRedPacketClaimed" && message.EventType != "WalletRedPacketRefunded" {
|
|
return broadcastdomain.RedPacketWalletEvent{}, false, nil
|
|
}
|
|
fields := redPacketFieldsFromWalletPayload(message.PayloadJSON, message.EventType)
|
|
if fields.PacketID == "" {
|
|
return broadcastdomain.RedPacketWalletEvent{}, false, nil
|
|
}
|
|
return broadcastdomain.RedPacketWalletEvent{
|
|
AppCode: appcode.Normalize(message.AppCode),
|
|
EventID: message.EventID,
|
|
EventType: message.EventType,
|
|
TransactionID: message.TransactionID,
|
|
CommandID: message.CommandID,
|
|
PayloadJSON: message.PayloadJSON,
|
|
PacketID: fields.PacketID,
|
|
PacketType: fields.PacketType,
|
|
RoomID: fields.RoomID,
|
|
RegionID: fields.RegionID,
|
|
RegionCode: fields.RegionCode,
|
|
SenderUserID: fields.SenderUserID,
|
|
TotalAmount: fields.TotalAmount,
|
|
TotalCount: fields.TotalCount,
|
|
RemainingAmount: fields.RemainingAmount,
|
|
RemainingCount: fields.RemainingCount,
|
|
RefundedAmount: fields.RefundedAmount,
|
|
Status: fields.Status,
|
|
OpenAtMS: fields.OpenAtMS,
|
|
ExpiresAtMS: fields.ExpiresAtMS,
|
|
CreatedAtMS: message.OccurredAtMS,
|
|
}, true, nil
|
|
}
|
|
|
|
func rechargeFactFromWalletPayload(payload string) (int64, string) {
|
|
var decoded map[string]any
|
|
if err := json.Unmarshal([]byte(payload), &decoded); err != nil {
|
|
return 0, firstrechargedomain.RechargeTypeCoinSeller
|
|
}
|
|
var sequence int64
|
|
switch value := decoded["recharge_sequence"].(type) {
|
|
case float64:
|
|
sequence = int64(value)
|
|
case int64:
|
|
sequence = value
|
|
case json.Number:
|
|
sequence, _ = value.Int64()
|
|
}
|
|
rechargeType := firstrechargedomain.RechargeTypeCoinSeller
|
|
if value, ok := decoded["recharge_type"].(string); ok && value != "" {
|
|
rechargeType = value
|
|
}
|
|
return sequence, rechargeType
|
|
}
|
|
|
|
type redPacketPayloadFields struct {
|
|
PacketID string
|
|
PacketType string
|
|
RoomID string
|
|
RegionID int64
|
|
RegionCode string
|
|
SenderUserID int64
|
|
TotalAmount int64
|
|
TotalCount int32
|
|
RemainingAmount int64
|
|
RemainingCount int32
|
|
RefundedAmount int64
|
|
Status string
|
|
OpenAtMS int64
|
|
ExpiresAtMS int64
|
|
}
|
|
|
|
func redPacketFieldsFromWalletPayload(payload string, eventType string) redPacketPayloadFields {
|
|
var decoded map[string]any
|
|
if err := json.Unmarshal([]byte(payload), &decoded); err != nil {
|
|
return redPacketPayloadFields{}
|
|
}
|
|
packetID := stringFromDecoded(decoded, "packet_id")
|
|
if packetID == "" {
|
|
packetID = stringFromDecoded(decoded, "packetId")
|
|
}
|
|
packetType := stringFromDecoded(decoded, "packet_type")
|
|
roomID := stringFromDecoded(decoded, "room_id")
|
|
regionID := int64FromDecoded(decoded, "region_id")
|
|
status := stringFromDecoded(decoded, "status")
|
|
openAtMS := int64FromDecoded(decoded, "open_at_ms")
|
|
expiresAtMS := int64FromDecoded(decoded, "expires_at_ms")
|
|
if eventType == "WalletRedPacketClaimed" || eventType == "WalletRedPacketRefunded" {
|
|
status = stringFromDecoded(decoded, "packet_status")
|
|
}
|
|
totalCount := int32FromDecoded(decoded, "packet_count")
|
|
remainingCount := int32FromDecoded(decoded, "remaining_count")
|
|
if remainingCount == 0 && eventType == "WalletRedPacketCreated" {
|
|
remainingCount = totalCount
|
|
}
|
|
return redPacketPayloadFields{
|
|
PacketID: packetID,
|
|
PacketType: packetType,
|
|
RoomID: roomID,
|
|
RegionID: regionID,
|
|
RegionCode: stringFromDecoded(decoded, "region_code"),
|
|
SenderUserID: int64FromDecoded(decoded, "sender_user_id"),
|
|
TotalAmount: int64FromDecoded(decoded, "total_amount"),
|
|
TotalCount: totalCount,
|
|
RemainingAmount: int64FromDecoded(decoded, "remaining_amount"),
|
|
RemainingCount: remainingCount,
|
|
RefundedAmount: int64FromDecoded(decoded, "refund_amount"),
|
|
Status: status,
|
|
OpenAtMS: openAtMS,
|
|
ExpiresAtMS: expiresAtMS,
|
|
}
|
|
}
|
|
|
|
func stringFromDecoded(decoded map[string]any, key string) string {
|
|
value, _ := decoded[key].(string)
|
|
return value
|
|
}
|
|
|
|
func int64FromDecoded(decoded map[string]any, key string) int64 {
|
|
switch value := decoded[key].(type) {
|
|
case float64:
|
|
return int64(value)
|
|
case int64:
|
|
return value
|
|
case json.Number:
|
|
parsed, _ := value.Int64()
|
|
return parsed
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func int32FromDecoded(decoded map[string]any, key string) int32 {
|
|
return int32(int64FromDecoded(decoded, key))
|
|
}
|
|
|
|
func luckyGiftWorkerOptions(nodeID string, cfg config.LuckyGiftWorkerConfig) luckygiftservice.WorkerOptions {
|
|
return luckygiftservice.WorkerOptions{
|
|
WorkerID: nodeID + "-lucky-gift",
|
|
PollInterval: cfg.WorkerPollInterval,
|
|
BatchSize: cfg.WorkerBatchSize,
|
|
Concurrency: cfg.WorkerConcurrency,
|
|
LockTTL: cfg.WorkerLockTTL,
|
|
MaxRetry: cfg.WorkerMaxRetry,
|
|
PublishTimeout: cfg.PublishTimeout,
|
|
}
|
|
}
|