415 lines
13 KiB
Go
415 lines
13 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"log/slog"
|
||
"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"
|
||
"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)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
consumers, err := newConsumers(cfg, repo)
|
||
if err != nil {
|
||
_ = repo.Close()
|
||
return nil, err
|
||
}
|
||
health := grpchealth.NewServingChecker("statistics-service", grpchealth.Dependency{Name: "mysql", Check: repo.Ping})
|
||
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
|
||
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 {
|
||
event, ok, err := userRegisteredEvent(message.Body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserRegistered(appcode.WithContext(ctx, event.AppCode), event)
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := add(cfg.RocketMQ.WalletOutbox, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
event, ok, err := rechargeEvent(message.Body)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if ok {
|
||
return repo.ConsumeRecharge(appcode.WithContext(ctx, event.AppCode), event)
|
||
}
|
||
reward, ok, err := luckyGiftRewardEvent(message.Body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeLuckyGiftReward(appcode.WithContext(ctx, reward.AppCode), reward)
|
||
}); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return nil, err
|
||
}
|
||
if err := add(cfg.RocketMQ.RoomOutbox, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
if gift, ok, err := roomGiftEvent(message.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.VisibleRegionID,
|
||
OccurredAtMS: gift.OccurredAtMS,
|
||
})
|
||
}
|
||
active, ok, err := roomJoinActiveEvent(message.Body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeUserActive(appcode.WithContext(ctx, active.AppCode), active)
|
||
}); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return nil, err
|
||
}
|
||
if err := add(cfg.RocketMQ.GameOutbox, gamemq.TagGameOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
event, ok, err := gameOrderEvent(message.Body)
|
||
if err != nil || !ok {
|
||
return err
|
||
}
|
||
return repo.ConsumeGameOrder(appcode.WithContext(ctx, event.AppCode), event)
|
||
}); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return nil, err
|
||
}
|
||
return consumers, nil
|
||
}
|
||
|
||
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,
|
||
RegionID: mysqlstorage.Int64(payload, "region_id"),
|
||
OccurredAtMS: 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)
|
||
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,
|
||
RechargeSequence: mysqlstorage.Int64(payload, "recharge_sequence"),
|
||
TargetRegionID: regionID,
|
||
RechargeType: rechargeType,
|
||
OccurredAtMS: message.OccurredAtMS,
|
||
}, true, nil
|
||
}
|
||
|
||
func rechargeUSDMinorFromPayload(payload map[string]any, rechargeType string) int64 {
|
||
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 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, "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 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
|
||
}
|
||
return mysqlstorage.RoomGiftEvent{
|
||
AppCode: envelope.GetAppCode(),
|
||
EventID: envelope.GetEventId(),
|
||
SenderUserID: gift.GetSenderUserId(),
|
||
VisibleRegionID: gift.GetVisibleRegionId(),
|
||
GiftValue: gift.GetGiftValue(),
|
||
PoolID: gift.GetPoolId(),
|
||
OccurredAtMS: envelope.GetOccurredAtMs(),
|
||
}, true, nil
|
||
}
|
||
|
||
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
|
||
}
|
||
return mysqlstorage.UserActiveEvent{
|
||
AppCode: envelope.GetAppCode(),
|
||
EventID: envelope.GetEventId(),
|
||
EventType: envelope.GetEventType(),
|
||
UserID: joined.GetUserId(),
|
||
CountryID: 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, "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 startConsumers(consumers []*rocketmqx.Consumer) error {
|
||
for _, consumer := range consumers {
|
||
if err := consumer.Start(); err != nil {
|
||
shutdownConsumers(consumers)
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func shutdownConsumers(consumers []*rocketmqx.Consumer) {
|
||
for _, consumer := range consumers {
|
||
if err := consumer.Shutdown(); err != nil {
|
||
logx.Warn(context.Background(), "rocketmq_consumer_shutdown_failed", slog.String("error", err.Error()))
|
||
}
|
||
}
|
||
}
|
||
|
||
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
|
||
}
|