386 lines
12 KiB
Go
386 lines
12 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)
|
|
usdMinor := firstNonZeroInt64(mysqlstorage.Int64(payload, "recharge_usd_minor"), mysqlstorage.Int64(payload, "amount_micro"))
|
|
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: firstNonEmpty(mysqlstorage.String(payload, "recharge_type"), mysqlstorage.String(payload, "channel"), mysqlstorage.String(payload, "provider"), "coin_seller_transfer"),
|
|
OccurredAtMS: message.OccurredAtMS,
|
|
}, true, nil
|
|
}
|
|
|
|
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")),
|
|
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
|
|
}
|