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)
	}
	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 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,
		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",
		"WalletLuckyGiftRewardCredited",
		"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
	}
}