813 lines
31 KiB
Go
813 lines
31 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"google.golang.org/protobuf/proto"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/gamemq"
|
||
"hyapp/pkg/grpchealth"
|
||
"hyapp/pkg/healthhttp"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/rocketmqx"
|
||
"hyapp/pkg/roommq"
|
||
servicehealth "hyapp/pkg/servicekit/health"
|
||
servicemq "hyapp/pkg/servicekit/mq"
|
||
"hyapp/pkg/usermq"
|
||
"hyapp/pkg/walletmq"
|
||
"hyapp/services/statistics-service/internal/config"
|
||
mysqlstorage "hyapp/services/statistics-service/internal/storage/mysql"
|
||
)
|
||
|
||
// App wires statistics MQ consumers to the aggregate repository.
|
||
type App struct {
|
||
cfg config.Config
|
||
repo *mysqlstorage.Repository
|
||
health *grpchealth.ServingChecker
|
||
healthHTTP *healthhttp.Server
|
||
queryHTTP *queryHTTPServer
|
||
consumers []*rocketmqx.Consumer
|
||
closeOnce sync.Once
|
||
}
|
||
|
||
func New(cfg config.Config) (*App, error) {
|
||
startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
repo, err := mysqlstorage.Open(startupCtx, cfg.MySQLDSN, cfg.ActivityMySQLDSN)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
consumers, err := newConsumers(cfg, repo)
|
||
if err != nil {
|
||
_ = repo.Close()
|
||
return nil, err
|
||
}
|
||
health, healthHTTP, err := servicehealth.NewHTTP(servicehealth.Config{
|
||
ServiceName: "statistics-service",
|
||
HTTPAddr: cfg.HealthHTTPAddr,
|
||
NodeID: cfg.NodeID,
|
||
Dependencies: []grpchealth.Dependency{{Name: "mysql", Check: repo.Ping}},
|
||
})
|
||
if err != nil {
|
||
shutdownConsumers(consumers)
|
||
_ = repo.Close()
|
||
return nil, err
|
||
}
|
||
queryHTTP, err := newQueryHTTPServer(cfg.QueryHTTPAddr, repo)
|
||
if err != nil {
|
||
shutdownConsumers(consumers)
|
||
_ = healthHTTP.Close(context.Background())
|
||
_ = repo.Close()
|
||
return nil, err
|
||
}
|
||
return &App{cfg: cfg, repo: repo, health: health, healthHTTP: healthHTTP, queryHTTP: queryHTTP, consumers: consumers}, nil
|
||
}
|
||
|
||
func (a *App) Run() error {
|
||
a.runHealthHTTP()
|
||
a.runQueryHTTP()
|
||
if err := startConsumers(a.consumers); err != nil {
|
||
a.health.MarkStopped()
|
||
return err
|
||
}
|
||
a.health.MarkServing()
|
||
select {}
|
||
}
|
||
|
||
func (a *App) runQueryHTTP() {
|
||
if a.queryHTTP == nil {
|
||
return
|
||
}
|
||
go func() {
|
||
if err := a.queryHTTP.Run(); err != nil {
|
||
logx.Error(context.Background(), "query_http_run_failed", err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
func (a *App) Close() {
|
||
a.closeOnce.Do(func() {
|
||
a.health.MarkDraining()
|
||
shutdownConsumers(a.consumers)
|
||
a.closeHealthHTTP()
|
||
if a.repo != nil {
|
||
_ = a.repo.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)
|
||
if a.queryHTTP != nil {
|
||
_ = a.queryHTTP.Close(ctx)
|
||
}
|
||
}
|
||
|
||
func newConsumers(cfg config.Config, repo *mysqlstorage.Repository) ([]*rocketmqx.Consumer, error) {
|
||
consumers := make([]*rocketmqx.Consumer, 0, 4)
|
||
add := func(mq config.ConsumerMQConfig, tag string, handler rocketmqx.Handler) error {
|
||
if !mq.Enabled {
|
||
return nil
|
||
}
|
||
consumer, err := rocketmqx.NewConsumer(consumerConfig(cfg.RocketMQ, mq))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := consumer.Subscribe(mq.Topic, tag, handler); err != nil {
|
||
_ = consumer.Shutdown()
|
||
return err
|
||
}
|
||
consumers = append(consumers, consumer)
|
||
return nil
|
||
}
|
||
if err := add(cfg.RocketMQ.UserOutbox, usermq.TagUserOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
return ConsumeUserOutboxBody(ctx, repo, message.Body)
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := add(cfg.RocketMQ.WalletOutbox, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
return ConsumeWalletOutboxBody(ctx, repo, message.Body)
|
||
}); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return nil, err
|
||
}
|
||
if err := add(cfg.RocketMQ.RoomOutbox, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
return ConsumeRoomOutboxBody(ctx, repo, message.Body)
|
||
}); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return nil, err
|
||
}
|
||
if err := add(cfg.RocketMQ.GameOutbox, gamemq.TagGameOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
return ConsumeGameOutboxBody(ctx, repo, message.Body)
|
||
}); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return nil, err
|
||
}
|
||
return consumers, nil
|
||
}
|
||
|
||
// ConsumeUserOutboxBody 是 user outbox 的唯一统计投影入口；实时 MQ 和离线回灌共用它，避免同一 payload 出现两套解析口径。
|
||
func ConsumeUserOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
|
||
if event, ok, err := userRegisteredEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserRegistered(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
if event, ok, err := userRegionChangedEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserRegionChanged(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
event, ok, err := userMicDailyStatsUpdatedEvent(body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserMicDailyStatsUpdated(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
|
||
// ConsumeWalletOutboxBody 按钱包事实类型分派充值、币商补货、奖励和余额变动；未知事件只提交消费位点，不写统计读模型。
|
||
func ConsumeWalletOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
|
||
if event, ok, err := rechargeEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeRecharge(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
if stock, ok, err := coinSellerStockEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeCoinSellerStock(appcode.WithContext(ctx, stock.AppCode), stock)
|
||
}
|
||
if reward, ok, err := luckyGiftRewardEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeLuckyGiftReward(appcode.WithContext(ctx, reward.AppCode), reward)
|
||
}
|
||
if grant, ok, err := platformGrantCoinEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumePlatformGrantCoin(appcode.WithContext(ctx, grant.AppCode), grant)
|
||
}
|
||
if output, ok, err := giftIncomeOutputEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeCoinOutput(appcode.WithContext(ctx, output.AppCode), output)
|
||
}
|
||
event, ok, err := walletBalanceChangedEvent(body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeWalletBalanceChanged(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
|
||
// ConsumeRoomOutboxBody 只投影房间统计需要的礼物和真人进房事件；礼物事实额外产生 sender 当日活跃，保持实时和回灌一致。
|
||
func ConsumeRoomOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
|
||
if gift, ok, err := roomGiftEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
if err := repo.ConsumeRoomGift(appcode.WithContext(ctx, gift.AppCode), gift); err != nil {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserActive(appcode.WithContext(ctx, gift.AppCode), mysqlstorage.UserActiveEvent{
|
||
AppCode: gift.AppCode,
|
||
EventID: gift.EventID + ":active",
|
||
EventType: "RoomGiftSentActive",
|
||
UserID: gift.SenderUserID,
|
||
CountryID: gift.CountryID,
|
||
RegionID: gift.VisibleRegionID,
|
||
OccurredAtMS: gift.OccurredAtMS,
|
||
})
|
||
}
|
||
active, ok, err := roomJoinActiveEvent(body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserActive(appcode.WithContext(ctx, active.AppCode), active)
|
||
}
|
||
|
||
// ConsumeGameOutboxBody 同时承接厂商游戏订单和自研游戏局级事实；回灌复用它可保证利润、奖池和参与者分布口径一致。
|
||
func ConsumeGameOutboxBody(ctx context.Context, repo *mysqlstorage.Repository, body []byte) error {
|
||
if event, ok, err := gameOrderEvent(body); err != nil || ok {
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeGameOrder(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
match, ok, err := selfGameMatchEvent(body)
|
||
if err != nil || !ok {
|
||
pool, poolOK, poolErr := selfGamePoolAdjustmentEvent(body)
|
||
if poolErr != nil || !poolOK {
|
||
return poolErr
|
||
}
|
||
return repo.ConsumeSelfGamePoolAdjustment(appcode.WithContext(ctx, pool.AppCode), pool)
|
||
}
|
||
return repo.ConsumeSelfGameMatch(appcode.WithContext(ctx, match.AppCode), match)
|
||
}
|
||
|
||
func userRegisteredEvent(body []byte) (mysqlstorage.UserRegisteredEvent, bool, error) {
|
||
message, err := usermq.DecodeUserOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.UserRegisteredEvent{}, false, err
|
||
}
|
||
if message.EventType != "UserRegistered" {
|
||
return mysqlstorage.UserRegisteredEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.UserRegisteredEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
UserID: message.AggregateID,
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")),
|
||
RegionID: mysqlstorage.Int64(payload, "region_id"),
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func userRegionChangedEvent(body []byte) (mysqlstorage.UserRegionChangedEvent, bool, error) {
|
||
message, err := usermq.DecodeUserOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.UserRegionChangedEvent{}, false, err
|
||
}
|
||
if message.EventType != "UserRegionChanged" {
|
||
return mysqlstorage.UserRegionChangedEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.UserRegionChangedEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.AggregateID),
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "new_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "new_region_id")),
|
||
OccurredAtMS: firstNonZeroInt64(mysqlstorage.Int64(payload, "changed_at_ms"), message.OccurredAtMS),
|
||
}, true, nil
|
||
}
|
||
|
||
func userMicDailyStatsUpdatedEvent(body []byte) (mysqlstorage.UserMicDailyStatsUpdatedEvent, bool, error) {
|
||
message, err := usermq.DecodeUserOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.UserMicDailyStatsUpdatedEvent{}, false, err
|
||
}
|
||
if message.EventType != "UserMicDailyStatsUpdated" {
|
||
return mysqlstorage.UserMicDailyStatsUpdatedEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.UserMicDailyStatsUpdatedEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.AggregateID),
|
||
StatDate: mysqlstorage.String(payload, "stat_date"),
|
||
MicOnlineDeltaMS: mysqlstorage.Int64(payload, "mic_online_delta_ms"),
|
||
OccurredAtMS: firstNonZeroInt64(mysqlstorage.Int64(payload, "updated_at_ms"), message.OccurredAtMS),
|
||
}, true, nil
|
||
}
|
||
|
||
func rechargeEvent(body []byte) (mysqlstorage.RechargeEvent, bool, error) {
|
||
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.RechargeEvent{}, false, err
|
||
}
|
||
if message.EventType != "WalletRechargeRecorded" {
|
||
return mysqlstorage.RechargeEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
rechargeType := firstNonEmpty(mysqlstorage.String(payload, "recharge_type"), mysqlstorage.String(payload, "channel"), mysqlstorage.String(payload, "provider"), "coin_seller_transfer")
|
||
usdMinor := rechargeUSDMinorFromPayload(payload, rechargeType)
|
||
coinAmount := int64(0)
|
||
if isCoinSellerRechargeType(rechargeType) {
|
||
usdMinor = 0
|
||
coinAmount = firstNonZeroInt64(mysqlstorage.Int64(payload, "amount"), mysqlstorage.Int64(payload, "coin_amount"), mysqlstorage.Int64(payload, "available_delta"))
|
||
}
|
||
regionID := firstNonZeroInt64(mysqlstorage.Int64(payload, "target_region_id"), mysqlstorage.Int64(payload, "region_id"))
|
||
return mysqlstorage.RechargeEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
EventType: message.EventType,
|
||
UserID: message.UserID,
|
||
USDMinor: usdMinor,
|
||
CoinAmount: coinAmount,
|
||
RechargeSequence: mysqlstorage.Int64(payload, "recharge_sequence"),
|
||
TargetCountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "target_country_id"), mysqlstorage.Int64(payload, "country_id")),
|
||
TargetRegionID: regionID,
|
||
RechargeType: rechargeType,
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func coinSellerStockEvent(body []byte) (mysqlstorage.CoinSellerStockEvent, bool, error) {
|
||
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.CoinSellerStockEvent{}, false, err
|
||
}
|
||
if message.EventType != "WalletCoinSellerStockPurchased" {
|
||
return mysqlstorage.CoinSellerStockEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
if !boolValue(payload, "counts_as_seller_recharge") {
|
||
return mysqlstorage.CoinSellerStockEvent{}, false, nil
|
||
}
|
||
usdMinor := amountMicroToUSDMinor(mysqlstorage.Int64(payload, "paid_amount_micro"))
|
||
if usdMinor == 0 {
|
||
return mysqlstorage.CoinSellerStockEvent{}, false, nil
|
||
}
|
||
return mysqlstorage.CoinSellerStockEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
SellerUserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "seller_user_id"), message.UserID),
|
||
USDMinor: usdMinor,
|
||
CoinAmount: firstNonZeroInt64(mysqlstorage.Int64(payload, "coin_amount"), mysqlstorage.Int64(payload, "available_delta")),
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "seller_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "seller_region_id")),
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func rechargeUSDMinorFromPayload(payload map[string]any, rechargeType string) int64 {
|
||
// 币商向用户转账是普通金币到账事实，只用于充值用户和金币金额统计；即使历史 payload 携带 recharge_usd_minor，也不能进入用户 USDT/USD 充值。
|
||
if isCoinSellerRechargeType(rechargeType) {
|
||
return 0
|
||
}
|
||
if value := mysqlstorage.Int64(payload, "recharge_usd_minor"); value > 0 {
|
||
return value
|
||
}
|
||
// 旧 Google outbox 只有 amount_micro；这里按微单位折算成美分，避免 0.99 被统计成 9900。
|
||
if isGoogleRechargeType(rechargeType) {
|
||
return amountMicroToUSDMinor(mysqlstorage.Int64(payload, "amount_micro"))
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func isGoogleRechargeType(value string) bool {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case "google", "google_play":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func isCoinSellerRechargeType(value string) bool {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case "coin_seller_transfer", "coin_seller":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func amountMicroToUSDMinor(amountMicro int64) int64 {
|
||
if amountMicro == 0 {
|
||
return 0
|
||
}
|
||
return amountMicro / 10_000
|
||
}
|
||
|
||
func luckyGiftRewardEvent(body []byte) (mysqlstorage.LuckyGiftRewardEvent, bool, error) {
|
||
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.LuckyGiftRewardEvent{}, false, err
|
||
}
|
||
if message.EventType != "WalletLuckyGiftRewardCredited" {
|
||
return mysqlstorage.LuckyGiftRewardEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.LuckyGiftRewardEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
EventType: message.EventType,
|
||
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.UserID),
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id")),
|
||
PoolID: mysqlstorage.String(payload, "pool_id"),
|
||
Amount: mysqlstorage.Int64(payload, "amount"),
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func platformGrantCoinEvent(body []byte) (mysqlstorage.PlatformGrantCoinEvent, bool, error) {
|
||
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.PlatformGrantCoinEvent{}, false, err
|
||
}
|
||
if !isPlatformGrantCoinEventType(message.EventType) || message.EventType == "WalletLuckyGiftRewardCredited" {
|
||
return mysqlstorage.PlatformGrantCoinEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.PlatformGrantCoinEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
EventType: message.EventType,
|
||
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.UserID),
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id")),
|
||
Amount: firstNonZeroInt64(mysqlstorage.Int64(payload, "amount"), message.AvailableDelta+message.FrozenDelta),
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func giftIncomeOutputEvent(body []byte) (mysqlstorage.CoinOutputEvent, bool, error) {
|
||
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.CoinOutputEvent{}, false, err
|
||
}
|
||
if message.EventType != "WalletGiftIncomeCredited" {
|
||
return mysqlstorage.CoinOutputEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.CoinOutputEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
EventType: message.EventType,
|
||
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), mysqlstorage.Int64(payload, "target_user_id"), message.UserID),
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id")),
|
||
Amount: firstNonZeroInt64(mysqlstorage.Int64(payload, "amount"), message.AvailableDelta+message.FrozenDelta),
|
||
OccurredAtMS: firstNonZeroInt64(mysqlstorage.Int64(payload, "created_at_ms"), message.OccurredAtMS),
|
||
}, true, nil
|
||
}
|
||
|
||
func walletBalanceChangedEvent(body []byte) (mysqlstorage.WalletBalanceChangedEvent, bool, error) {
|
||
message, err := walletmq.DecodeWalletOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.WalletBalanceChangedEvent{}, false, err
|
||
}
|
||
if message.EventType != "WalletBalanceChanged" {
|
||
return mysqlstorage.WalletBalanceChangedEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
metadata := objectValue(payload, "metadata")
|
||
assetType := firstNonEmpty(mysqlstorage.String(payload, "asset_type"), message.AssetType)
|
||
bizType := firstNonEmpty(mysqlstorage.String(payload, "biz_type"), mysqlstorage.String(payload, "business_type"))
|
||
dimensionUserID := int64(0)
|
||
if strings.EqualFold(strings.TrimSpace(bizType), "salary_transfer_to_coin_seller") && strings.EqualFold(strings.TrimSpace(assetType), "COIN_SELLER_COIN") {
|
||
dimensionUserID = firstNonZeroInt64(mysqlstorage.Int64(payload, "dimension_user_id"), mysqlstorage.Int64(metadata, "source_user_id"), mysqlstorage.Int64(metadata, "sourceUserID"))
|
||
}
|
||
return mysqlstorage.WalletBalanceChangedEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
EventType: message.EventType,
|
||
UserID: firstNonZeroInt64(mysqlstorage.Int64(payload, "user_id"), message.UserID),
|
||
DimensionUserID: dimensionUserID,
|
||
AssetType: assetType,
|
||
BizType: bizType,
|
||
CommandID: firstNonEmpty(mysqlstorage.String(payload, "command_id"), mysqlstorage.String(metadata, "command_id"), message.CommandID),
|
||
AvailableDelta: firstNonZeroInt64(mysqlstorage.Int64(payload, "available_delta"), message.AvailableDelta),
|
||
FrozenDelta: firstNonZeroInt64(mysqlstorage.Int64(payload, "frozen_delta"), message.FrozenDelta),
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id"), mysqlstorage.Int64(payload, "target_region_id"), mysqlstorage.Int64(metadata, "region_id"), mysqlstorage.Int64(metadata, "regionId")),
|
||
OccurredAtMS: firstNonZeroInt64(mysqlstorage.Int64(payload, "created_at_ms"), message.OccurredAtMS),
|
||
}, true, nil
|
||
}
|
||
|
||
func isPlatformGrantCoinEventType(value string) bool {
|
||
switch strings.TrimSpace(value) {
|
||
case "WalletTaskRewardCredited",
|
||
"WalletWheelRewardCredited",
|
||
"WalletRoomTurnoverRewardCredited",
|
||
"WalletInviteActivityRewardCredited",
|
||
"WalletAgencyOpeningRewardCredited":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func roomGiftEvent(body []byte) (mysqlstorage.RoomGiftEvent, bool, error) {
|
||
envelope, _, err := roommq.DecodeRoomOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.RoomGiftEvent{}, false, err
|
||
}
|
||
if envelope.GetEventType() != "RoomGiftSent" {
|
||
return mysqlstorage.RoomGiftEvent{}, false, nil
|
||
}
|
||
var gift roomeventsv1.RoomGiftSent
|
||
if err := proto.Unmarshal(envelope.GetBody(), &gift); err != nil {
|
||
return mysqlstorage.RoomGiftEvent{}, false, err
|
||
}
|
||
if gift.GetIsRobotGift() && !gift.GetRealRoomRobotGift() {
|
||
return mysqlstorage.RoomGiftEvent{}, false, nil
|
||
}
|
||
return mysqlstorage.RoomGiftEvent{
|
||
AppCode: envelope.GetAppCode(),
|
||
EventID: envelope.GetEventId(),
|
||
RoomID: envelope.GetRoomId(),
|
||
SenderUserID: gift.GetSenderUserId(),
|
||
CountryID: gift.GetCountryId(),
|
||
VisibleRegionID: firstNonZeroInt64(gift.GetRegionId(), gift.GetVisibleRegionId()),
|
||
GiftValue: gift.GetGiftValue(),
|
||
CoinSpent: roomGiftCoinSpent(&gift),
|
||
PoolID: gift.GetPoolId(),
|
||
GiftTypeCode: gift.GetGiftTypeCode(),
|
||
RealRoomRobot: gift.GetRealRoomRobotGift(),
|
||
OccurredAtMS: envelope.GetOccurredAtMs(),
|
||
}, true, nil
|
||
}
|
||
|
||
func roomGiftCoinSpent(gift *roomeventsv1.RoomGiftSent) int64 {
|
||
if gift == nil {
|
||
return 0
|
||
}
|
||
// Databi 的礼物消费和幸运礼物流水按金币展示；新版 RoomGiftSent 会直接携带 coin_spent，历史事件没有该字段时才回退旧的 gift_value 热度口径。
|
||
if gift.GetCoinSpent() > 0 {
|
||
return gift.GetCoinSpent()
|
||
}
|
||
return gift.GetGiftValue()
|
||
}
|
||
|
||
func roomJoinActiveEvent(body []byte) (mysqlstorage.UserActiveEvent, bool, error) {
|
||
envelope, _, err := roommq.DecodeRoomOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.UserActiveEvent{}, false, err
|
||
}
|
||
if envelope.GetEventType() != "RoomUserJoined" {
|
||
return mysqlstorage.UserActiveEvent{}, false, nil
|
||
}
|
||
var joined roomeventsv1.RoomUserJoined
|
||
if err := proto.Unmarshal(envelope.GetBody(), &joined); err != nil {
|
||
return mysqlstorage.UserActiveEvent{}, false, err
|
||
}
|
||
if joined.GetIsRobot() {
|
||
// 房间机器人进房只服务房间热闹度和 IM 展示，不代表真人用户活跃；这里返回 ok=false，让 room outbox 消费继续提交位点但不写统计聚合。
|
||
return mysqlstorage.UserActiveEvent{}, false, nil
|
||
}
|
||
return mysqlstorage.UserActiveEvent{
|
||
AppCode: envelope.GetAppCode(),
|
||
EventID: envelope.GetEventId(),
|
||
EventType: envelope.GetEventType(),
|
||
UserID: joined.GetUserId(),
|
||
CountryID: joined.GetCountryId(),
|
||
RegionID: firstNonZeroInt64(joined.GetRegionId(), joined.GetVisibleRegionId()),
|
||
OccurredAtMS: envelope.GetOccurredAtMs(),
|
||
}, true, nil
|
||
}
|
||
|
||
func gameOrderEvent(body []byte) (mysqlstorage.GameOrderEvent, bool, error) {
|
||
message, err := gamemq.DecodeGameOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.GameOrderEvent{}, false, err
|
||
}
|
||
if message.EventType != "GameOrderSettled" {
|
||
return mysqlstorage.GameOrderEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.GameOrderEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
UserID: message.UserID,
|
||
CountryID: firstNonZeroInt64(mysqlstorage.Int64(payload, "country_id"), mysqlstorage.Int64(payload, "target_country_id")),
|
||
RegionID: firstNonZeroInt64(mysqlstorage.Int64(payload, "visible_region_id"), mysqlstorage.Int64(payload, "region_id")),
|
||
PlatformCode: message.PlatformCode,
|
||
GameID: message.GameID,
|
||
OpType: message.OpType,
|
||
CoinAmount: message.CoinAmount,
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func selfGameMatchEvent(body []byte) (mysqlstorage.SelfGameMatchEvent, bool, error) {
|
||
message, err := gamemq.DecodeGameOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.SelfGameMatchEvent{}, false, err
|
||
}
|
||
switch message.EventType {
|
||
case gamemq.EventTypeSelfGameMatchCreated,
|
||
gamemq.EventTypeSelfGameMatchReady,
|
||
gamemq.EventTypeSelfGameMatchSettled,
|
||
gamemq.EventTypeSelfGameMatchCanceled,
|
||
gamemq.EventTypeSelfGameMatchFailed:
|
||
default:
|
||
return mysqlstorage.SelfGameMatchEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
event := mysqlstorage.SelfGameMatchEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
EventType: message.EventType,
|
||
MatchID: firstNonEmpty(mysqlstorage.String(payload, "match_id"), message.OrderID),
|
||
GameID: firstNonEmpty(mysqlstorage.String(payload, "game_id"), message.GameID),
|
||
PlatformCode: firstNonEmpty(mysqlstorage.String(payload, "platform_code"), message.PlatformCode),
|
||
RoomID: mysqlstorage.String(payload, "room_id"),
|
||
CountryID: mysqlstorage.Int64(payload, "country_id"),
|
||
RegionID: mysqlstorage.Int64(payload, "region_id"),
|
||
StakeCoin: mysqlstorage.Int64(payload, "stake_coin"),
|
||
Status: mysqlstorage.String(payload, "status"),
|
||
Result: mysqlstorage.String(payload, "result"),
|
||
MatchMode: mysqlstorage.String(payload, "match_mode"),
|
||
ForcedResult: mysqlstorage.String(payload, "forced_result"),
|
||
UserParticipants: mysqlstorage.Int64(payload, "user_participants"),
|
||
RobotParticipants: mysqlstorage.Int64(payload, "robot_participants"),
|
||
WinnerCount: mysqlstorage.Int64(payload, "winner_count"),
|
||
LoserCount: mysqlstorage.Int64(payload, "loser_count"),
|
||
DrawCount: mysqlstorage.Int64(payload, "draw_count"),
|
||
UserStakeCoin: mysqlstorage.Int64(payload, "user_stake_coin"),
|
||
RobotStakeCoin: mysqlstorage.Int64(payload, "robot_stake_coin"),
|
||
PayoutCoin: mysqlstorage.Int64(payload, "payout_coin"),
|
||
RefundCoin: mysqlstorage.Int64(payload, "refund_coin"),
|
||
PlatformProfitCoin: mysqlstorage.Int64(payload, "platform_profit_coin"),
|
||
PoolDeltaCoin: mysqlstorage.Int64(payload, "pool_delta_coin"),
|
||
WaitMS: mysqlstorage.Int64(payload, "wait_ms"),
|
||
DurationMS: mysqlstorage.Int64(payload, "duration_ms"),
|
||
Participants: selfGameParticipants(payload["participants"]),
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}
|
||
return event, true, nil
|
||
}
|
||
|
||
func selfGamePoolAdjustmentEvent(body []byte) (mysqlstorage.SelfGamePoolAdjustmentEvent, bool, error) {
|
||
message, err := gamemq.DecodeGameOutboxMessage(body)
|
||
if err != nil {
|
||
return mysqlstorage.SelfGamePoolAdjustmentEvent{}, false, err
|
||
}
|
||
if message.EventType != gamemq.EventTypeSelfGamePoolAdjusted {
|
||
return mysqlstorage.SelfGamePoolAdjustmentEvent{}, false, nil
|
||
}
|
||
payload := mysqlstorage.DecodeJSON(message.PayloadJSON)
|
||
return mysqlstorage.SelfGamePoolAdjustmentEvent{
|
||
AppCode: message.AppCode,
|
||
EventID: message.EventID,
|
||
GameID: firstNonEmpty(mysqlstorage.String(payload, "game_id"), message.GameID),
|
||
Direction: mysqlstorage.String(payload, "direction"),
|
||
Reason: mysqlstorage.String(payload, "reason"),
|
||
AmountCoin: mysqlstorage.Int64(payload, "amount_coin"),
|
||
BalanceAfter: mysqlstorage.Int64(payload, "balance_after"),
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func selfGameParticipants(raw any) []mysqlstorage.SelfGameMatchParticipantEvent {
|
||
values, ok := raw.([]any)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
participants := make([]mysqlstorage.SelfGameMatchParticipantEvent, 0, len(values))
|
||
for _, value := range values {
|
||
payload, ok := value.(map[string]any)
|
||
if !ok {
|
||
continue
|
||
}
|
||
participants = append(participants, mysqlstorage.SelfGameMatchParticipantEvent{
|
||
UserID: mysqlstorage.Int64(payload, "user_id"),
|
||
ParticipantType: mysqlstorage.String(payload, "participant_type"),
|
||
StakeCoin: mysqlstorage.Int64(payload, "stake_coin"),
|
||
DicePoints: int64Slice(payload["dice_points"]),
|
||
RPSGesture: mysqlstorage.String(payload, "rps_gesture"),
|
||
Result: mysqlstorage.String(payload, "result"),
|
||
PayoutCoin: mysqlstorage.Int64(payload, "payout_coin"),
|
||
NetWinCoin: mysqlstorage.Int64(payload, "net_win_coin"),
|
||
})
|
||
}
|
||
return participants
|
||
}
|
||
|
||
func int64Slice(raw any) []int64 {
|
||
values, ok := raw.([]any)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
out := make([]int64, 0, len(values))
|
||
for _, value := range values {
|
||
switch typed := value.(type) {
|
||
case float64:
|
||
out = append(out, int64(typed))
|
||
case int64:
|
||
out = append(out, typed)
|
||
case int:
|
||
out = append(out, int64(typed))
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func startConsumers(consumers []*rocketmqx.Consumer) error {
|
||
return servicemq.StartConsumers(consumers)
|
||
}
|
||
|
||
func shutdownConsumers(consumers []*rocketmqx.Consumer) {
|
||
servicemq.ShutdownConsumers(consumers)
|
||
}
|
||
|
||
func consumerConfig(cfg config.RocketMQConfig, mq config.ConsumerMQConfig) 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: mq.ConsumerGroup,
|
||
MaxReconsumeTimes: mq.ConsumerMaxReconsumeTimes,
|
||
ConsumeRetryDelay: time.Second,
|
||
ConsumePullBatch: 32,
|
||
ConsumeGoroutines: 1,
|
||
}
|
||
}
|
||
|
||
func firstNonEmpty(values ...string) string {
|
||
for _, value := range values {
|
||
if value = strings.TrimSpace(value); value != "" {
|
||
return value
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func firstNonZeroInt64(values ...int64) int64 {
|
||
for _, value := range values {
|
||
if value != 0 {
|
||
return value
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func objectValue(payload map[string]any, key string) map[string]any {
|
||
if value, ok := payload[key].(map[string]any); ok {
|
||
return value
|
||
}
|
||
return map[string]any{}
|
||
}
|
||
|
||
func boolValue(payload map[string]any, key string) bool {
|
||
switch value := payload[key].(type) {
|
||
case bool:
|
||
return value
|
||
case string:
|
||
return strings.EqualFold(strings.TrimSpace(value), "true") || strings.TrimSpace(value) == "1"
|
||
case float64:
|
||
return value != 0
|
||
case int64:
|
||
return value != 0
|
||
default:
|
||
return false
|
||
}
|
||
}
|