数据大屏口径修正
This commit is contained in:
zhx
parent
e953d55bc7
commit
ff51573613
94
docs/数据大屏统计口径.md
Normal file
94
docs/数据大屏统计口径.md
Normal file
@ -0,0 +1,94 @@
|
||||
# 数据大屏统计口径
|
||||
|
||||
本文档描述 `statistics-service` 提供给后台数据大屏的字段口径。查询入口是 `server/admin` 转发到 `statistics-service` 的 `/internal/v1/statistics/overview`，返回字段同时用于 `overview`、`daily_series`、`country_breakdown` 和 `daily_country_breakdown`。
|
||||
|
||||
## 通用规则
|
||||
|
||||
- 时间范围统一是 `[start_ms, end_ms)`，包含开始毫秒，不包含结束毫秒。
|
||||
- 默认统计时区是 `UTC`；传 `stat_tz=Asia/Shanghai` 时按中国自然日查询统计读模型。
|
||||
- 国家和区域只使用事件里的国家/区域字段或统计服务保存的用户维度，不用服务器所在地、IP 或本地时区推断。
|
||||
- 查询只读 `statistics-service` 聚合表，不回查 wallet、room、game、user owner 明细表。
|
||||
- `coin_total` 为兼容前端保留 JSON 字段名，展示名改为“用户金币数”。
|
||||
|
||||
## 字段口径
|
||||
|
||||
| 字段 | 展示名 | 口径 |
|
||||
| --- | --- | --- |
|
||||
| `app_code` | 应用编码 | 当前查询的租户编码。 |
|
||||
| `stat_tz` | 统计时区 | `UTC` 或 `Asia/Shanghai`。 |
|
||||
| `start_ms` / `end_ms` | 查询时间 | 本次查询的毫秒时间范围。 |
|
||||
| `country_id` / `region_id` | 国家/区域 | 聚合维度；未筛选国家时返回国家分解。 |
|
||||
| `new_users` | 新增用户 | `UserRegistered` 注册 cohort 去重用户数。 |
|
||||
| `active_users` | 活跃用户 | `RoomUserJoined`、`GameOrderSettled` 和注册当天活跃写入 `stat_user_day_activity` 后按用户去重。 |
|
||||
| `paid_users` | 付费用户 | `WalletRechargeRecorded` 写入 `stat_recharge_day_payers` 后按用户去重。 |
|
||||
| `new_paid_users` | 新增付费用户 | 付费用户中注册 cohort 也落在查询周期内的用户数。 |
|
||||
| `recharge_usd_minor` | 总充值 | `WalletRechargeRecorded.usd_minor_amount` 累加，单位是美元最小单位。 |
|
||||
| `new_user_recharge_usd_minor` | 新增充值 | `recharge_sequence=1` 的充值金额累加。 |
|
||||
| `coin_seller_recharge_usd_minor` | 币商充值 | 币商进货 USDT 和未识别为 Google/MifaPay/USDT-TRC20/币商转账的充值美元最小单位。 |
|
||||
| `coin_seller_stock_coin` | 币商充值金币 | `WalletCoinSellerStockPurchased.coin_amount` 累加。 |
|
||||
| `mifapay_recharge_usd_minor` | 三方充值 | `recharge_type=mifapay/v5pay` 的充值美元最小单位。 |
|
||||
| `google_recharge_usd_minor` | Google 充值 | `recharge_type=google/google_play` 的充值美元最小单位。 |
|
||||
| `coin_seller_transfer_coin` | 币商出货金币 | 币商向用户转入普通 COIN 金币数量。 |
|
||||
| `coin_total` | 用户金币数 | 用户普通 COIN 钱包 `available+frozen` 余额总和；历史日取当天 23:59:59.999 前最后一次余额，今日按 `WalletBalanceChanged` 增量滚动更新；多日总览只取结束日快照，日序列展示每天快照。 |
|
||||
| `consumed_coin` | 消耗金币 | 游戏投入 + 普通礼物投入 + lucky 礼物投入 + 转盘投入 + 发红包。游戏来自 `GameOrderSettled debit/bet`，房间礼物来自 `RoomGiftSent.coin_spent`，钱包只补 `direct_gift_debit`、`wheel_draw_debit`、`red_packet_create`。 |
|
||||
| `output_coin` | 产出金币 | 游戏产出 + 收礼产出 + lucky 礼物产出 + 转盘奖励金币 + 红包领取和退回。游戏来自 `GameOrderSettled credit/payout`，收礼来自 `WalletGiftIncomeCredited`，lucky 来自 `WalletLuckyGiftRewardCredited`，转盘金币来自 `WalletWheelRewardCredited`，红包来自 `red_packet_claim/refund`。 |
|
||||
| `consume_output_ratio` | 消耗产出比 | `output_coin / consumed_coin`；消耗为 0 时返回 0。 |
|
||||
| `consume_output_delta` | 消耗产出差额 | `consumed_coin - output_coin`。 |
|
||||
| `platform_grant_coin` | 平台发放金币 | 排行榜奖励、活动奖励、签到/任务奖励、火箭奖励中的 COIN 金币发放；包括对应 `Wallet*RewardCredited` 和 `resource_grant` COIN 入账，不含人工发放、装扮价值、lucky 返奖、转盘产出。 |
|
||||
| `manual_grant_coin` | 人工发放金币 | `WalletBalanceChanged biz_type=manual_credit` 的普通 COIN 正向发放。 |
|
||||
| `salary_usd_minor` | 工资总和 | 主播和代理工资钱包结算成功金额，单位是 USD minor。 |
|
||||
| `salary_transfer_coin` | 工资兑换金币 | 工资转给币商时币商 `COIN_SELLER_COIN` 入账金币数量，维度按发起转账用户。 |
|
||||
| `mic_online_ms` | 麦上总时长 | `user_mic_daily_stats.mic_online_ms` 按日、国家、区域聚合。 |
|
||||
| `mic_online_users` | 麦上用户数 | 当日 `mic_online_ms > 0` 的去重用户数。 |
|
||||
| `avg_mic_online_ms` | 平均麦上时长 | `mic_online_ms / mic_online_users`。 |
|
||||
| `gift_coin_spent` | 礼物流水 | 非机器人真实 `RoomGiftSent.coin_spent`，历史事件无 `coin_spent` 时回退 `gift_value`。 |
|
||||
| `real_room_robot_normal_gift_coin` | 真人房机器人普通礼物 | `real_room_robot_gift=true` 且非 lucky/super lucky 的礼物金币。 |
|
||||
| `real_room_robot_lucky_gift_coin` | 真人房机器人 lucky 礼物 | 真人房机器人 lucky 礼物金币。 |
|
||||
| `real_room_robot_super_gift_coin` | 真人房机器人 super lucky 礼物 | 真人房机器人 super lucky 礼物金币。 |
|
||||
| `real_room_robot_total_gift_coin` | 真人房机器人总礼物 | 上面三类机器人礼物金币合计。 |
|
||||
| `real_room_robot_gift_room_count` | 真人房机器人投放房间数 | 有真人房机器人礼物的房间去重数。 |
|
||||
| `real_room_robot_avg_room_gift_coin` | 真人房机器人房均投放 | `real_room_robot_total_gift_coin / real_room_robot_gift_room_count`。 |
|
||||
| `lucky_gift_turnover` | lucky 礼物流水 | `RoomGiftSent.pool_id` 非空的送礼金币。 |
|
||||
| `lucky_gift_payout` | lucky 礼物产出 | `WalletLuckyGiftRewardCredited.amount`，UTC 查询会优先用 activity 的 lucky pool 日表覆盖旧统计行。 |
|
||||
| `lucky_gift_payers` | lucky 礼物付费用户 | 发送 lucky 礼物的用户按日去重。 |
|
||||
| `lucky_gift_profit` | lucky 礼物利润 | `lucky_gift_turnover - lucky_gift_payout`。 |
|
||||
| `lucky_gift_payout_rate` | lucky 礼物返奖率 | `lucky_gift_payout / lucky_gift_turnover`。 |
|
||||
| `lucky_gift_profit_rate` | lucky 礼物利润率 | `lucky_gift_profit / lucky_gift_turnover`。 |
|
||||
| `game_turnover` | 游戏流水 | `GameOrderSettled op_type=debit/bet` 的金币。 |
|
||||
| `game_payout` | 游戏返奖 | `GameOrderSettled op_type=credit/payout` 的金币。 |
|
||||
| `game_refund` | 游戏退款 | `GameOrderSettled op_type=refund/reverse` 的金币。 |
|
||||
| `game_players` | 游戏玩家数 | 有游戏投入的用户按日、游戏平台、游戏去重后汇总。 |
|
||||
| `game_profit` | 游戏利润 | `game_turnover - game_payout - game_refund`。 |
|
||||
| `game_profit_rate` | 游戏利润率 | `game_profit / game_turnover`。 |
|
||||
| `super_lucky_gift_turnover` | super lucky 流水 | `pool_id` 包含 `super_lucky` 的奖池流水。 |
|
||||
| `super_lucky_gift_payout` | super lucky 返奖 | `pool_id` 包含 `super_lucky` 的奖池返奖。 |
|
||||
| `super_lucky_gift_profit` | super lucky 利润 | `super_lucky_gift_turnover - super_lucky_gift_payout`。 |
|
||||
| `super_lucky_gift_profit_rate` | super lucky 利润率 | `super_lucky_gift_profit / super_lucky_gift_turnover`。 |
|
||||
| `arpu_usd_minor` | ARPU | `recharge_usd_minor / active_users`。 |
|
||||
| `arppu_usd_minor` | ARPPU | `recharge_usd_minor / paid_users`。 |
|
||||
| `up_value_usd_minor` | UP 值 | 当前与 ARPPU 同口径。 |
|
||||
|
||||
## 对比字段
|
||||
|
||||
以下字段是当前周期与上一等长周期的变化率：`recharge_delta_rate`、`new_user_recharge_delta_rate`、`active_users_delta_rate`、`paid_users_delta_rate`、`new_paid_users_delta_rate`、`arpu_delta_rate`、`arppu_delta_rate`、`gift_coin_spent_delta_rate`、`coin_seller_transfer_coin_delta_rate`、`lucky_gift_turnover_delta_rate`、`lucky_gift_payout_delta_rate`、`lucky_gift_profit_delta_rate`、`game_turnover_delta_rate`、`game_profit_delta_rate`。
|
||||
|
||||
变化率公式是 `(当前值 - 上期值) / abs(上期值)`；上期为 0 且当前大于 0 时返回 1。
|
||||
|
||||
## 明细对象
|
||||
|
||||
| 字段 | 口径 |
|
||||
| --- | --- |
|
||||
| `retention` | `registered_users` 为注册 cohort；`day1/day7/day30_users` 为注册后第 1/7/30 天有活跃的用户；rate 为对应用户数除以注册用户数。 |
|
||||
| `daily_series` | 按日返回本页主要字段；`coin_total` 是每天的用户金币数快照，其它流水字段按当天累加。 |
|
||||
| `country_breakdown` | 按国家返回本页主要字段；`coin_total` 是查询结束日该国家用户金币数快照，其它字段按查询周期累加。 |
|
||||
| `daily_country_breakdown` | 按日 + 国家返回主要字段，用于国家弹窗明细。 |
|
||||
| `game_ranking` | 按 `platform_code/game_id` 聚合游戏流水、返奖、退款、玩家数、利润和利润率。 |
|
||||
| `lucky_gift_pools` | 按 `pool_id` 聚合 lucky 礼物流水、返奖、利润、返奖率和利润率。 |
|
||||
| `report_metric_sources` | 返回字段来源和口径说明，供前端展示或排查。 |
|
||||
|
||||
## 当前边界
|
||||
|
||||
- 转盘礼物/道具的 RTP 价值在 activity wheel 记录里有 `rtp_value_coins`，但 wallet 资源 outbox 当前不是稳定 MQ 统计输入；统计服务现在只计转盘金币奖励，礼物价值需要资源奖励事件可稳定发布后接入。
|
||||
- 平台发放的礼物价值同理，需要 `ResourceGranted/ResourceGroupGranted` 事件携带可统计的资源金币价值和来源；当前只可靠计入 COIN 入账和已存在的 `Wallet*RewardCredited` 金币奖励。
|
||||
- 装扮价值不进入 `platform_grant_coin` 和 `output_coin`。
|
||||
- 后台人工发放只进入 `manual_grant_coin`，不进入 `platform_grant_coin`。
|
||||
@ -66,6 +66,8 @@ flowchart LR
|
||||
|
||||
## 口径定义
|
||||
|
||||
数据大屏字段级口径以 [数据大屏统计口径](./数据大屏统计口径.md) 为准；本节只保留统计服务的核心边界。
|
||||
|
||||
### 用户与活跃
|
||||
|
||||
| 指标 | 口径 |
|
||||
@ -98,8 +100,8 @@ flowchart LR
|
||||
|
||||
| 指标 | 口径 |
|
||||
| --- | --- |
|
||||
| 礼物消耗 | `RoomGiftSent.gift_value` 累加 |
|
||||
| 幸运礼物流水 | `RoomGiftSent.pool_id` 非空时的 `gift_value` 累加 |
|
||||
| 礼物消耗 | `RoomGiftSent.coin_spent` 累加，历史事件无 `coin_spent` 时回退 `gift_value` |
|
||||
| 幸运礼物流水 | `RoomGiftSent.pool_id` 非空时的 `coin_spent` 累加，历史事件无 `coin_spent` 时回退 `gift_value` |
|
||||
| 幸运礼物付费用户 | `pool_id` 非空的送礼用户按 UTC 日去重 |
|
||||
|
||||
房间礼物事件必须在 wallet 扣费成功后产生。统计服务不读取钱包流水来反推礼物消耗。
|
||||
|
||||
@ -64,7 +64,61 @@ func (h *Handler) SelfGameStatisticsOverview(c *gin.Context) {
|
||||
response.OK(c, overview)
|
||||
}
|
||||
|
||||
func (h *Handler) PlatformGrantUsers(c *gin.Context) {
|
||||
page, err := h.service.PlatformGrantUsers(c.Request.Context(), PlatformGrantStatisticsQuery{
|
||||
AppCode: c.DefaultQuery("app_code", "lalu"),
|
||||
StatTZ: c.Query("stat_tz"),
|
||||
RequestID: middleware.CurrentRequestID(c),
|
||||
StartMS: parseInt64(c.Query("start_ms")),
|
||||
EndMS: parseInt64(c.Query("end_ms")),
|
||||
StatDay: c.Query("stat_day"),
|
||||
RegionID: parseInt64(c.Query("region_id")),
|
||||
CountryID: parseInt64(c.Query("country_id")),
|
||||
Page: parseInt(c.Query("page")),
|
||||
PageSize: parseInt(firstNonEmpty(c.Query("page_size"), c.Query("pageSize"))),
|
||||
})
|
||||
if err != nil {
|
||||
response.ServerError(c, "获取平台发放金币用户明细失败")
|
||||
return
|
||||
}
|
||||
response.OK(c, page)
|
||||
}
|
||||
|
||||
func (h *Handler) PlatformGrantRecords(c *gin.Context) {
|
||||
page, err := h.service.PlatformGrantRecords(c.Request.Context(), PlatformGrantStatisticsQuery{
|
||||
AppCode: c.DefaultQuery("app_code", "lalu"),
|
||||
StatTZ: c.Query("stat_tz"),
|
||||
StartMS: parseInt64(c.Query("start_ms")),
|
||||
EndMS: parseInt64(c.Query("end_ms")),
|
||||
StatDay: c.Query("stat_day"),
|
||||
RegionID: parseInt64(c.Query("region_id")),
|
||||
CountryID: parseInt64(c.Query("country_id")),
|
||||
UserID: parseInt64(c.Query("user_id")),
|
||||
Page: parseInt(c.Query("page")),
|
||||
PageSize: parseInt(firstNonEmpty(c.Query("page_size"), c.Query("pageSize"))),
|
||||
})
|
||||
if err != nil {
|
||||
response.ServerError(c, "获取平台发放金币来源明细失败")
|
||||
return
|
||||
}
|
||||
response.OK(c, page)
|
||||
}
|
||||
|
||||
func parseInt64(value string) int64 {
|
||||
parsed, _ := strconv.ParseInt(value, 10, 64)
|
||||
return parsed
|
||||
}
|
||||
|
||||
func parseInt(value string) int {
|
||||
parsed, _ := strconv.Atoi(value)
|
||||
return parsed
|
||||
}
|
||||
|
||||
func firstNonEmpty(values ...string) string {
|
||||
for _, value := range values {
|
||||
if value != "" {
|
||||
return value
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@ -9,5 +9,7 @@ import (
|
||||
func RegisterRoutes(protected *gin.RouterGroup, h *Handler) {
|
||||
protected.GET("/dashboard/overview", middleware.RequirePermission("overview:view"), h.DashboardOverview)
|
||||
protected.GET("/statistics/overview", middleware.RequirePermission("overview:view"), h.StatisticsOverview)
|
||||
protected.GET("/statistics/platform-grants/users", middleware.RequirePermission("overview:view"), h.PlatformGrantUsers)
|
||||
protected.GET("/statistics/platform-grants/records", middleware.RequirePermission("overview:view"), h.PlatformGrantRecords)
|
||||
protected.GET("/statistics/self-games/overview", middleware.RequirePermission("overview:view"), h.SelfGameStatisticsOverview)
|
||||
}
|
||||
|
||||
@ -43,6 +43,20 @@ type SelfGameStatisticsQuery struct {
|
||||
GameID string
|
||||
}
|
||||
|
||||
type PlatformGrantStatisticsQuery struct {
|
||||
AppCode string
|
||||
StatTZ string
|
||||
RequestID string
|
||||
StartMS int64
|
||||
EndMS int64
|
||||
StatDay string
|
||||
RegionID int64
|
||||
CountryID int64
|
||||
UserID int64
|
||||
Page int
|
||||
PageSize int
|
||||
}
|
||||
|
||||
func NewService(store *repository.Store, cfg config.Config, user userclient.Client) *DashboardService {
|
||||
return &DashboardService{
|
||||
store: store,
|
||||
@ -114,6 +128,28 @@ func (s *DashboardService) SelfGameStatisticsOverview(ctx context.Context, query
|
||||
return overview, nil
|
||||
}
|
||||
|
||||
func (s *DashboardService) PlatformGrantUsers(ctx context.Context, query PlatformGrantStatisticsQuery) (map[string]any, error) {
|
||||
page, err := s.statisticsGET(ctx, "/internal/v1/statistics/platform-grants/users", platformGrantValues(query, false))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 统计服务只返回用户 ID 和金币聚合；admin 层用 user-service 批量补展示资料，避免统计库复制用户主数据。
|
||||
_ = s.enrichPlatformGrantUsers(ctx, query, page)
|
||||
return page, nil
|
||||
}
|
||||
|
||||
func (s *DashboardService) PlatformGrantRecords(ctx context.Context, query PlatformGrantStatisticsQuery) (map[string]any, error) {
|
||||
if query.UserID <= 0 {
|
||||
return nil, fmt.Errorf("user_id is required")
|
||||
}
|
||||
page, err := s.statisticsGET(ctx, "/internal/v1/statistics/platform-grants/records", platformGrantValues(query, true))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stringifyPageUserIDs(page)
|
||||
return page, nil
|
||||
}
|
||||
|
||||
func (s *DashboardService) statisticsGET(ctx context.Context, path string, values url.Values) (map[string]any, error) {
|
||||
baseURL := s.cfg.StatisticsService.BaseURL + path
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL+"?"+values.Encode(), nil)
|
||||
@ -137,6 +173,119 @@ func (s *DashboardService) statisticsGET(ctx context.Context, path string, value
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func platformGrantValues(query PlatformGrantStatisticsQuery, includeUser bool) url.Values {
|
||||
values := url.Values{}
|
||||
values.Set("app_code", query.AppCode)
|
||||
if statTZ := strings.TrimSpace(query.StatTZ); statTZ != "" {
|
||||
values.Set("stat_tz", statTZ)
|
||||
}
|
||||
if query.StartMS > 0 {
|
||||
values.Set("start_ms", strconv.FormatInt(query.StartMS, 10))
|
||||
}
|
||||
if query.EndMS > 0 {
|
||||
values.Set("end_ms", strconv.FormatInt(query.EndMS, 10))
|
||||
}
|
||||
if statDay := strings.TrimSpace(query.StatDay); statDay != "" {
|
||||
values.Set("stat_day", statDay)
|
||||
}
|
||||
if query.RegionID > 0 {
|
||||
values.Set("region_id", strconv.FormatInt(query.RegionID, 10))
|
||||
}
|
||||
if query.CountryID > 0 {
|
||||
values.Set("country_id", strconv.FormatInt(query.CountryID, 10))
|
||||
}
|
||||
if includeUser && query.UserID > 0 {
|
||||
values.Set("user_id", strconv.FormatInt(query.UserID, 10))
|
||||
}
|
||||
if query.Page > 0 {
|
||||
values.Set("page", strconv.Itoa(query.Page))
|
||||
}
|
||||
if query.PageSize > 0 {
|
||||
values.Set("page_size", strconv.Itoa(query.PageSize))
|
||||
}
|
||||
return values
|
||||
}
|
||||
|
||||
func (s *DashboardService) enrichPlatformGrantUsers(ctx context.Context, query PlatformGrantStatisticsQuery, page map[string]any) error {
|
||||
if s.user == nil || page == nil {
|
||||
return nil
|
||||
}
|
||||
rows, ok := page["items"].([]any)
|
||||
if !ok || len(rows) == 0 {
|
||||
return nil
|
||||
}
|
||||
userIDs := make([]int64, 0, len(rows))
|
||||
seen := map[int64]struct{}{}
|
||||
for _, raw := range rows {
|
||||
row, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
userID := anyInt64(row["user_id"])
|
||||
if userID <= 0 {
|
||||
continue
|
||||
}
|
||||
// user_id 可能超过 JS 安全整数，admin 统一转字符串输出，前端分页和后续来源查询都使用字符串透传。
|
||||
row["user_id"] = strconv.FormatInt(userID, 10)
|
||||
if _, exists := seen[userID]; exists {
|
||||
continue
|
||||
}
|
||||
seen[userID] = struct{}{}
|
||||
userIDs = append(userIDs, userID)
|
||||
}
|
||||
if len(userIDs) == 0 {
|
||||
return nil
|
||||
}
|
||||
users, err := s.user.BatchGetUsers(ctx, userclient.BatchGetUsersRequest{
|
||||
RequestID: query.RequestID,
|
||||
Caller: "admin-server",
|
||||
UserIDs: userIDs,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, raw := range rows {
|
||||
row, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
userID := anyInt64(row["user_id"])
|
||||
user := users[userID]
|
||||
if user == nil {
|
||||
continue
|
||||
}
|
||||
userPayload := map[string]any{
|
||||
"user_id": strconv.FormatInt(userID, 10),
|
||||
"avatar": user.Avatar,
|
||||
"username": user.Username,
|
||||
"display_user_id": user.DisplayUserID,
|
||||
}
|
||||
row["user"] = userPayload
|
||||
row["avatar"] = user.Avatar
|
||||
row["nickname"] = user.Username
|
||||
row["username"] = user.Username
|
||||
row["display_user_id"] = user.DisplayUserID
|
||||
row["short_id"] = user.DisplayUserID
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func stringifyPageUserIDs(page map[string]any) {
|
||||
rows, ok := page["items"].([]any)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
for _, raw := range rows {
|
||||
row, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if userID := anyInt64(row["user_id"]); userID > 0 {
|
||||
row["user_id"] = strconv.FormatInt(userID, 10)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *DashboardService) enrichSelfGameRiskUsers(ctx context.Context, query SelfGameStatisticsQuery, overview map[string]any) error {
|
||||
if s.user == nil || overview == nil {
|
||||
return nil
|
||||
|
||||
@ -39,6 +39,7 @@ type appDayAgg struct {
|
||||
coinSellerTransferCoin int64
|
||||
coinTotal int64
|
||||
consumedCoin int64
|
||||
outputCoin int64
|
||||
platformGrantCoin int64
|
||||
manualGrantCoin int64
|
||||
salaryUSDMinor int64
|
||||
@ -166,6 +167,9 @@ func main() {
|
||||
if err := rebuildWalletReportMetrics(ctx, walletDB, state, app, startDay, endDay, windowStartMS, windowEndMS, loc); err != nil {
|
||||
log.Fatalf("rebuild wallet report metrics: %v", err)
|
||||
}
|
||||
if err := rebuildAppDayCoinMetricsFromStatistics(ctx, statsDB, state, app, statTZ, startDay, endDay); err != nil {
|
||||
log.Fatalf("rebuild app day coin metrics from statistics: %v", err)
|
||||
}
|
||||
if err := rebuildSalary(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
|
||||
log.Fatalf("rebuild salary: %v", err)
|
||||
}
|
||||
@ -595,7 +599,8 @@ func rebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildS
|
||||
return err
|
||||
}
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT e.user_id, e.counterparty_user_id, e.asset_type, e.available_delta + e.frozen_delta AS amount_delta, wt.biz_type, e.created_at_ms
|
||||
SELECT e.user_id, e.counterparty_user_id, e.asset_type, e.available_delta + e.frozen_delta AS amount_delta,
|
||||
wt.biz_type, COALESCE(JSON_UNQUOTE(JSON_EXTRACT(wt.metadata_json, '$.command_id')), wt.command_id), e.created_at_ms
|
||||
FROM wallet_entries e
|
||||
JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id
|
||||
WHERE wt.app_code = ? AND e.asset_type IN ('COIN', 'COIN_SELLER_COIN') AND e.created_at_ms >= ? AND e.created_at_ms < ?`,
|
||||
@ -606,8 +611,8 @@ func rebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildS
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var userID, counterpartyUserID, amountDelta, createdAtMS int64
|
||||
var assetType, bizType string
|
||||
if err := rows.Scan(&userID, &counterpartyUserID, &assetType, &amountDelta, &bizType, &createdAtMS); err != nil {
|
||||
var assetType, bizType, commandID string
|
||||
if err := rows.Scan(&userID, &counterpartyUserID, &assetType, &amountDelta, &bizType, &commandID, &createdAtMS); err != nil {
|
||||
return err
|
||||
}
|
||||
day := dayFromMS(createdAtMS, loc)
|
||||
@ -630,8 +635,12 @@ func rebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildS
|
||||
switch {
|
||||
case amountDelta < 0 && isConsumedCoinBizType(normalizedBiz):
|
||||
agg.consumedCoin += -amountDelta
|
||||
case amountDelta > 0 && isOutputCoinBizType(normalizedBiz):
|
||||
agg.outputCoin += amountDelta
|
||||
case amountDelta > 0 && normalizedBiz == "manual_credit":
|
||||
agg.manualGrantCoin += amountDelta
|
||||
case amountDelta > 0 && normalizedBiz == "resource_grant" && isPlatformGrantCommandID(commandID):
|
||||
agg.platformGrantCoin += amountDelta
|
||||
case amountDelta > 0 && isPlatformGrantBizType(normalizedBiz):
|
||||
agg.platformGrantCoin += amountDelta
|
||||
}
|
||||
@ -639,6 +648,33 @@ func rebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildS
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
func rebuildAppDayCoinMetricsFromStatistics(ctx context.Context, db *sql.DB, state *rebuildState, app string, statTZ string, startDay string, endDay string) error {
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), country_id, region_id,
|
||||
COALESCE(gift_coin_spent, 0), COALESCE(game_turnover, 0), COALESCE(game_payout, 0)
|
||||
FROM stat_app_day_country
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
|
||||
app, statTZ, startDay, endDay)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var day string
|
||||
var dim countryRegion
|
||||
var giftSpent, gameTurnover, gamePayout int64
|
||||
if err := rows.Scan(&day, &dim.countryID, &dim.regionID, &giftSpent, &gameTurnover, &gamePayout); err != nil {
|
||||
return err
|
||||
}
|
||||
agg := state.ensureAgg(day, dim)
|
||||
// 本补数脚本不连接 room/game owner 原始库；礼物和游戏事实已经由统计服务实时聚合到日表，
|
||||
// 因此只复用这些 owner 事实读模型来重建总消耗/产出，避免重跑钱包补数时把礼物和游戏口径清零。
|
||||
agg.consumedCoin += giftSpent + gameTurnover
|
||||
agg.outputCoin += gamePayout
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
type coinBalanceEntry struct {
|
||||
userID int64
|
||||
balance int64
|
||||
@ -778,8 +814,16 @@ func dayEndPoints(startDay string, endDay string, loc *time.Location) ([]dayEndP
|
||||
|
||||
func isConsumedCoinBizType(value string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "gift_debit", "direct_gift_debit", "vip_purchase", "resource_shop_purchase",
|
||||
"red_packet_create", "game_debit", "wheel_draw_debit", "cp_breakup_fee":
|
||||
case "direct_gift_debit", "red_packet_create", "wheel_draw_debit":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isOutputCoinBizType(value string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "gift_debit", "direct_gift_debit", "lucky_gift_reward", "wheel_reward", "red_packet_claim", "red_packet_refund":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
@ -788,8 +832,30 @@ func isConsumedCoinBizType(value string) bool {
|
||||
|
||||
func isPlatformGrantBizType(value string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "task_reward", "lucky_gift_reward", "wheel_reward", "room_turnover_reward",
|
||||
"invite_activity_reward", "agency_opening_reward":
|
||||
case "task_reward", "room_turnover_reward", "invite_activity_reward", "agency_opening_reward":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isPlatformGrantCommandID(value string) bool {
|
||||
normalized := strings.ToLower(strings.TrimSpace(value))
|
||||
if normalized == "" {
|
||||
return false
|
||||
}
|
||||
switch {
|
||||
case strings.HasPrefix(normalized, "weekly_star:"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "cp_weekly_rank:"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "room_turnover_reward:"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "wcheckin_"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "cmd_room_rocket_grant_"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "wtask_"), strings.HasPrefix(normalized, "wfr_"), strings.HasPrefix(normalized, "wcr_"), strings.HasPrefix(normalized, "wreg_"):
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
@ -878,7 +944,7 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
|
||||
return tx.Rollback()
|
||||
}
|
||||
nowMS := time.Now().UTC().UnixMilli()
|
||||
// 本 job 只覆盖能从当前源表确定的列；礼物和游戏等不在本次源表范围内的指标保留原值，且只替换目标 stat_tz，避免重跑中国日时覆盖 UTC 读模型。
|
||||
// 本 job 只覆盖能从当前源表和既有日聚合确定的列；未在本次源表范围内的礼物/游戏明细列保留原值，且只替换目标 stat_tz，避免重跑中国日时覆盖 UTC 读模型。
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE stat_app_day_country
|
||||
SET new_users = 0, active_users = 0, paid_users = 0,
|
||||
@ -886,7 +952,7 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
|
||||
coin_seller_recharge_usd_minor = 0,
|
||||
coin_seller_stock_coin = 0,
|
||||
mifapay_recharge_usd_minor = 0, google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
|
||||
coin_total = 0, consumed_coin = 0, platform_grant_coin = 0, manual_grant_coin = 0,
|
||||
coin_total = 0, consumed_coin = 0, output_coin = 0, platform_grant_coin = 0, manual_grant_coin = 0,
|
||||
salary_usd_minor = 0, salary_transfer_coin = 0, mic_online_ms = 0, mic_online_users = 0,
|
||||
updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
|
||||
@ -929,13 +995,13 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
|
||||
app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
|
||||
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
|
||||
coin_seller_stock_coin, mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin,
|
||||
coin_total, consumed_coin, platform_grant_coin, manual_grant_coin,
|
||||
coin_total, consumed_coin, output_coin, platform_grant_coin, manual_grant_coin,
|
||||
salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users, updated_at_ms
|
||||
)
|
||||
SELECT app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
|
||||
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
|
||||
coin_seller_stock_coin, mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin,
|
||||
coin_total, consumed_coin, platform_grant_coin, manual_grant_coin,
|
||||
coin_total, consumed_coin, output_coin, platform_grant_coin, manual_grant_coin,
|
||||
salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users, ?
|
||||
FROM tmp_stat_backfill_app_day
|
||||
ON DUPLICATE KEY UPDATE
|
||||
@ -949,6 +1015,7 @@ func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, ap
|
||||
coin_seller_transfer_coin = VALUES(coin_seller_transfer_coin),
|
||||
coin_total = VALUES(coin_total),
|
||||
consumed_coin = VALUES(consumed_coin),
|
||||
output_coin = VALUES(output_coin),
|
||||
platform_grant_coin = VALUES(platform_grant_coin),
|
||||
manual_grant_coin = VALUES(manual_grant_coin),
|
||||
salary_usd_minor = VALUES(salary_usd_minor),
|
||||
@ -990,7 +1057,7 @@ func createTempTables(ctx context.Context, tx *sql.Tx) error {
|
||||
coin_seller_recharge_usd_minor BIGINT NOT NULL, coin_seller_stock_coin BIGINT NOT NULL,
|
||||
mifapay_recharge_usd_minor BIGINT NOT NULL, google_recharge_usd_minor BIGINT NOT NULL,
|
||||
coin_seller_transfer_coin BIGINT NOT NULL,
|
||||
coin_total BIGINT NOT NULL, consumed_coin BIGINT NOT NULL, platform_grant_coin BIGINT NOT NULL,
|
||||
coin_total BIGINT NOT NULL, consumed_coin BIGINT NOT NULL, output_coin BIGINT NOT NULL, platform_grant_coin BIGINT NOT NULL,
|
||||
manual_grant_coin BIGINT NOT NULL, salary_usd_minor BIGINT NOT NULL, salary_transfer_coin BIGINT NOT NULL,
|
||||
mic_online_ms BIGINT NOT NULL, mic_online_users BIGINT NOT NULL,
|
||||
PRIMARY KEY(app_code, stat_tz, stat_day, country_id, region_id)
|
||||
@ -1031,12 +1098,12 @@ func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app s
|
||||
app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
|
||||
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
|
||||
coin_seller_stock_coin, mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin,
|
||||
coin_total, consumed_coin, platform_grant_coin, manual_grant_coin, salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
coin_total, consumed_coin, output_coin, platform_grant_coin, manual_grant_coin, salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
app, statTZ, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
|
||||
agg.rechargeUSDMinor, agg.newUserRechargeUSDMinor, agg.coinSellerRechargeUSDMinor,
|
||||
agg.coinSellerStockCoin, agg.mifaPayRechargeUSDMinor, agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin,
|
||||
agg.coinTotal, agg.consumedCoin, agg.platformGrantCoin, agg.manualGrantCoin, agg.salaryUSDMinor, agg.salaryTransferCoin, agg.micOnlineMS, agg.micOnlineUsers); err != nil {
|
||||
agg.coinTotal, agg.consumedCoin, agg.outputCoin, agg.platformGrantCoin, agg.manualGrantCoin, agg.salaryUSDMinor, agg.salaryTransferCoin, agg.micOnlineMS, agg.micOnlineUsers); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@ -1077,7 +1144,7 @@ func validateStage(ctx context.Context, tx *sql.Tx) error {
|
||||
SELECT COUNT(*)
|
||||
FROM tmp_stat_backfill_app_day
|
||||
WHERE new_users < 0 OR active_users < 0 OR paid_users < 0 OR recharge_usd_minor < 0 OR coin_seller_transfer_coin < 0
|
||||
OR coin_total < 0 OR consumed_coin < 0 OR platform_grant_coin < 0 OR manual_grant_coin < 0
|
||||
OR coin_total < 0 OR consumed_coin < 0 OR output_coin < 0 OR platform_grant_coin < 0 OR manual_grant_coin < 0
|
||||
OR salary_usd_minor < 0 OR salary_transfer_coin < 0 OR mic_online_ms < 0 OR mic_online_users < 0`).Scan(&negativeCount); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -33,9 +33,10 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country (
|
||||
mifapay_recharge_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT 'MifaPay 充值美元最小单位',
|
||||
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 充值美元最小单位',
|
||||
coin_seller_transfer_coin BIGINT NOT NULL DEFAULT 0 COMMENT '币商向用户转普通金币数量',
|
||||
coin_total BIGINT NOT NULL DEFAULT 0 COMMENT '普通用户 COIN 钱包日终余额总和',
|
||||
consumed_coin BIGINT NOT NULL DEFAULT 0 COMMENT '消费类普通金币扣减',
|
||||
platform_grant_coin BIGINT NOT NULL DEFAULT 0 COMMENT '系统奖励金币发放',
|
||||
coin_total BIGINT NOT NULL DEFAULT 0 COMMENT '普通用户 COIN 钱包余额快照总和',
|
||||
consumed_coin BIGINT NOT NULL DEFAULT 0 COMMENT '游戏、礼物、转盘和红包发放消耗金币',
|
||||
output_coin BIGINT NOT NULL DEFAULT 0 COMMENT '游戏、收礼、lucky、转盘和红包领取退回产出金币',
|
||||
platform_grant_coin BIGINT NOT NULL DEFAULT 0 COMMENT '排行榜、活动、签到和火箭奖励金币发放',
|
||||
manual_grant_coin BIGINT NOT NULL DEFAULT 0 COMMENT '后台人工发放金币',
|
||||
salary_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '主播和代理工资 USD 最小单位',
|
||||
salary_transfer_coin BIGINT NOT NULL DEFAULT 0 COMMENT '工资转币商入账金币数量',
|
||||
@ -131,6 +132,23 @@ CREATE TABLE IF NOT EXISTS stat_lucky_gift_day_payers (
|
||||
KEY idx_stat_lucky_payer_region (app_code, stat_tz, stat_day, region_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物日付费去重表';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_platform_grant_coin_events (
|
||||
app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
|
||||
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
|
||||
stat_day DATE NOT NULL COMMENT '统计日，随 stat_tz 切换',
|
||||
event_id VARCHAR(160) NOT NULL COMMENT '钱包 outbox 事件 ID',
|
||||
event_type VARCHAR(64) NOT NULL COMMENT '钱包奖励事件类型',
|
||||
user_id BIGINT NOT NULL COMMENT '收币用户 ID',
|
||||
country_id BIGINT NOT NULL DEFAULT 0 COMMENT '事件发生时用户国家维度',
|
||||
region_id BIGINT NOT NULL DEFAULT 0 COMMENT '事件发生时用户区域维度',
|
||||
amount BIGINT NOT NULL DEFAULT 0 COMMENT '本次平台发放金币',
|
||||
occurred_at_ms BIGINT NOT NULL COMMENT '事件发生时间，UTC epoch ms',
|
||||
updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
|
||||
PRIMARY KEY (app_code, stat_tz, event_id),
|
||||
KEY idx_stat_platform_grant_country (app_code, stat_tz, stat_day, country_id, region_id, amount),
|
||||
KEY idx_stat_platform_grant_user (app_code, stat_tz, stat_day, country_id, user_id, occurred_at_ms)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='平台发放金币事件明细投影';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_lucky_gift_pool_day_country (
|
||||
app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
|
||||
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
|
||||
|
||||
@ -213,6 +213,12 @@ func ConsumeWalletOutboxBody(ctx context.Context, repo *mysqlstorage.Repository,
|
||||
}
|
||||
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
|
||||
@ -465,6 +471,27 @@ func platformGrantCoinEvent(body []byte) (mysqlstorage.PlatformGrantCoinEvent, b
|
||||
}, 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 {
|
||||
@ -489,6 +516,7 @@ func walletBalanceChangedEvent(body []byte) (mysqlstorage.WalletBalanceChangedEv
|
||||
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")),
|
||||
@ -500,7 +528,6 @@ func walletBalanceChangedEvent(body []byte) (mysqlstorage.WalletBalanceChangedEv
|
||||
func isPlatformGrantCoinEventType(value string) bool {
|
||||
switch strings.TrimSpace(value) {
|
||||
case "WalletTaskRewardCredited",
|
||||
"WalletLuckyGiftRewardCredited",
|
||||
"WalletWheelRewardCredited",
|
||||
"WalletRoomTurnoverRewardCredited",
|
||||
"WalletInviteActivityRewardCredited",
|
||||
|
||||
@ -252,6 +252,9 @@ func TestConsumeOutboxMessagesProjectsDataScreenMetrics(t *testing.T) {
|
||||
if overview.GameTurnover != 700 || overview.GamePayout != 200 || overview.GameProfit != 500 || overview.GamePlayers != 1 {
|
||||
t.Fatalf("game metrics mismatch: %+v", overview)
|
||||
}
|
||||
if overview.ConsumedCoin != 1_950 || overview.OutputCoin != 500 || overview.ConsumeOutputDelta != 1_450 {
|
||||
t.Fatalf("consume/output metrics mismatch: %+v", overview)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConsumeWalletOutboxBodyProjectsBalanceChangedReportMetrics(t *testing.T) {
|
||||
@ -299,7 +302,7 @@ func TestConsumeWalletOutboxBodyProjectsBalanceChangedReportMetrics(t *testing.T
|
||||
PayloadJSON: mustJSON(t, map[string]any{
|
||||
"user_id": int64(88001),
|
||||
"asset_type": "COIN",
|
||||
"biz_type": "gift_debit",
|
||||
"biz_type": "red_packet_create",
|
||||
"available_delta": int64(-88),
|
||||
"country_id": int64(15),
|
||||
"region_id": int64(10),
|
||||
@ -307,6 +310,43 @@ func TestConsumeWalletOutboxBodyProjectsBalanceChangedReportMetrics(t *testing.T
|
||||
}),
|
||||
OccurredAtMS: occurredAt,
|
||||
},
|
||||
{
|
||||
AppCode: "lalu",
|
||||
EventID: "WalletBalanceChanged:coin:gift-income-balance",
|
||||
EventType: "WalletBalanceChanged",
|
||||
TransactionID: "wallet_tx_gift_income",
|
||||
CommandID: "gift-income-balance",
|
||||
UserID: 88002,
|
||||
AssetType: "COIN",
|
||||
PayloadJSON: mustJSON(t, map[string]any{
|
||||
"user_id": int64(88002),
|
||||
"asset_type": "COIN",
|
||||
"biz_type": "gift_debit",
|
||||
"available_delta": int64(25),
|
||||
"country_id": int64(15),
|
||||
"region_id": int64(10),
|
||||
"created_at_ms": occurredAt,
|
||||
}),
|
||||
OccurredAtMS: occurredAt,
|
||||
},
|
||||
{
|
||||
AppCode: "lalu",
|
||||
EventID: "WalletGiftIncomeCredited:coin:gift-income",
|
||||
EventType: "WalletGiftIncomeCredited",
|
||||
TransactionID: "wallet_tx_gift_income",
|
||||
CommandID: "gift-income",
|
||||
UserID: 88002,
|
||||
AssetType: "COIN",
|
||||
AvailableDelta: 25,
|
||||
PayloadJSON: mustJSON(t, map[string]any{
|
||||
"user_id": int64(88002),
|
||||
"amount": int64(25),
|
||||
"country_id": int64(15),
|
||||
"region_id": int64(10),
|
||||
"created_at_ms": occurredAt,
|
||||
}),
|
||||
OccurredAtMS: occurredAt,
|
||||
},
|
||||
{
|
||||
AppCode: "lalu",
|
||||
EventID: "WalletBalanceChanged:salary:coin-seller",
|
||||
@ -346,7 +386,7 @@ func TestConsumeWalletOutboxBodyProjectsBalanceChangedReportMetrics(t *testing.T
|
||||
if err != nil {
|
||||
t.Fatalf("query overview failed: %v", err)
|
||||
}
|
||||
if overview.CoinTotal != 12 || overview.ConsumedCoin != 88 || overview.SalaryTransferCoin != 660 {
|
||||
if overview.CoinTotal != 37 || overview.ConsumedCoin != 88 || overview.OutputCoin != 25 || overview.SalaryTransferCoin != 660 {
|
||||
t.Fatalf("wallet balance report metrics mismatch: %+v", overview)
|
||||
}
|
||||
}
|
||||
|
||||
@ -32,6 +32,8 @@ func newQueryHTTPServer(addr string, repo *mysqlstorage.Repository) (*queryHTTPS
|
||||
s := &queryHTTPServer{repo: repo, listener: listener}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/internal/v1/statistics/overview", s.overview)
|
||||
mux.HandleFunc("/internal/v1/statistics/platform-grants/users", s.platformGrantUsers)
|
||||
mux.HandleFunc("/internal/v1/statistics/platform-grants/records", s.platformGrantRecords)
|
||||
mux.HandleFunc("/internal/v1/statistics/self-game/events", s.selfGameEvents)
|
||||
mux.HandleFunc("/internal/v1/statistics/self-games/overview", s.selfGameOverview)
|
||||
s.server = &http.Server{Handler: mux, ReadHeaderTimeout: 2 * time.Second}
|
||||
@ -84,6 +86,60 @@ func (s *queryHTTPServer) overview(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, overview)
|
||||
}
|
||||
|
||||
func (s *queryHTTPServer) platformGrantUsers(w http.ResponseWriter, r *http.Request) {
|
||||
query := r.URL.Query()
|
||||
pageSize := parsePositiveInt(query.Get("page_size"))
|
||||
if pageSize == 0 {
|
||||
pageSize = parsePositiveInt(query.Get("pageSize"))
|
||||
}
|
||||
page, err := s.repo.QueryPlatformGrantCoinUsers(r.Context(), mysqlstorage.PlatformGrantCoinQuery{
|
||||
AppCode: query.Get("app_code"),
|
||||
StatTZ: query.Get("stat_tz"),
|
||||
StartMS: parseInt64(query.Get("start_ms")),
|
||||
EndMS: parseInt64(query.Get("end_ms")),
|
||||
StatDay: query.Get("stat_day"),
|
||||
CountryID: parseInt64(query.Get("country_id")),
|
||||
RegionID: parseInt64(query.Get("region_id")),
|
||||
Page: parsePositiveInt(query.Get("page")),
|
||||
PageSize: pageSize,
|
||||
})
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, page)
|
||||
}
|
||||
|
||||
func (s *queryHTTPServer) platformGrantRecords(w http.ResponseWriter, r *http.Request) {
|
||||
query := r.URL.Query()
|
||||
userID := parseInt64(query.Get("user_id"))
|
||||
if userID <= 0 {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]any{"error": "user_id is required"})
|
||||
return
|
||||
}
|
||||
pageSize := parsePositiveInt(query.Get("page_size"))
|
||||
if pageSize == 0 {
|
||||
pageSize = parsePositiveInt(query.Get("pageSize"))
|
||||
}
|
||||
page, err := s.repo.QueryPlatformGrantCoinRecords(r.Context(), mysqlstorage.PlatformGrantCoinQuery{
|
||||
AppCode: query.Get("app_code"),
|
||||
StatTZ: query.Get("stat_tz"),
|
||||
StartMS: parseInt64(query.Get("start_ms")),
|
||||
EndMS: parseInt64(query.Get("end_ms")),
|
||||
StatDay: query.Get("stat_day"),
|
||||
CountryID: parseInt64(query.Get("country_id")),
|
||||
RegionID: parseInt64(query.Get("region_id")),
|
||||
UserID: userID,
|
||||
Page: parsePositiveInt(query.Get("page")),
|
||||
PageSize: pageSize,
|
||||
})
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, page)
|
||||
}
|
||||
|
||||
type selfGameH5EventRequest struct {
|
||||
AppCode string `json:"app_code"`
|
||||
EventID string `json:"event_id"`
|
||||
@ -189,6 +245,14 @@ func parseInt64(value string) int64 {
|
||||
return parsed
|
||||
}
|
||||
|
||||
func parsePositiveInt(value string) int {
|
||||
parsed, _ := strconv.Atoi(strings.TrimSpace(value))
|
||||
if parsed < 0 {
|
||||
return 0
|
||||
}
|
||||
return parsed
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, payload any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
|
||||
@ -42,6 +42,9 @@ type Overview struct {
|
||||
CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"`
|
||||
CoinTotal int64 `json:"coin_total"`
|
||||
ConsumedCoin int64 `json:"consumed_coin"`
|
||||
OutputCoin int64 `json:"output_coin"`
|
||||
ConsumeOutputRatio float64 `json:"consume_output_ratio"`
|
||||
ConsumeOutputDelta int64 `json:"consume_output_delta"`
|
||||
PlatformGrantCoin int64 `json:"platform_grant_coin"`
|
||||
ManualGrantCoin int64 `json:"manual_grant_coin"`
|
||||
SalaryUSDMinor int64 `json:"salary_usd_minor"`
|
||||
@ -115,6 +118,9 @@ type DailySeriesItem struct {
|
||||
CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"`
|
||||
CoinTotal int64 `json:"coin_total"`
|
||||
ConsumedCoin int64 `json:"consumed_coin"`
|
||||
OutputCoin int64 `json:"output_coin"`
|
||||
ConsumeOutputRatio float64 `json:"consume_output_ratio"`
|
||||
ConsumeOutputDelta int64 `json:"consume_output_delta"`
|
||||
PlatformGrantCoin int64 `json:"platform_grant_coin"`
|
||||
ManualGrantCoin int64 `json:"manual_grant_coin"`
|
||||
SalaryUSDMinor int64 `json:"salary_usd_minor"`
|
||||
@ -186,6 +192,9 @@ type CountryBreakdown struct {
|
||||
CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"`
|
||||
CoinTotal int64 `json:"coin_total"`
|
||||
ConsumedCoin int64 `json:"consumed_coin"`
|
||||
OutputCoin int64 `json:"output_coin"`
|
||||
ConsumeOutputRatio float64 `json:"consume_output_ratio"`
|
||||
ConsumeOutputDelta int64 `json:"consume_output_delta"`
|
||||
PlatformGrantCoin int64 `json:"platform_grant_coin"`
|
||||
ManualGrantCoin int64 `json:"manual_grant_coin"`
|
||||
SalaryUSDMinor int64 `json:"salary_usd_minor"`
|
||||
@ -259,6 +268,53 @@ type SelfGameOverview struct {
|
||||
Conversion SelfGameConversion `json:"conversion"`
|
||||
}
|
||||
|
||||
type PlatformGrantCoinQuery struct {
|
||||
AppCode string
|
||||
StatTZ string
|
||||
StartMS int64
|
||||
EndMS int64
|
||||
StatDay string
|
||||
CountryID int64
|
||||
RegionID int64
|
||||
UserID int64
|
||||
Page int
|
||||
PageSize int
|
||||
}
|
||||
|
||||
type PlatformGrantCoinUser struct {
|
||||
UserID int64 `json:"user_id"`
|
||||
CountryID int64 `json:"country_id"`
|
||||
RegionID int64 `json:"region_id"`
|
||||
TotalCoin int64 `json:"total_coin"`
|
||||
RecordCount int64 `json:"record_count"`
|
||||
FirstGrantedAtMS int64 `json:"first_granted_at_ms"`
|
||||
LastGrantedAtMS int64 `json:"last_granted_at_ms"`
|
||||
}
|
||||
|
||||
type PlatformGrantCoinRecord struct {
|
||||
EventID string `json:"event_id"`
|
||||
EventType string `json:"event_type"`
|
||||
UserID int64 `json:"user_id"`
|
||||
CountryID int64 `json:"country_id"`
|
||||
RegionID int64 `json:"region_id"`
|
||||
Amount int64 `json:"amount"`
|
||||
OccurredAtMS int64 `json:"occurred_at_ms"`
|
||||
}
|
||||
|
||||
type PlatformGrantCoinUserPage struct {
|
||||
Items []PlatformGrantCoinUser `json:"items"`
|
||||
Page int `json:"page"`
|
||||
PageSize int `json:"pageSize"`
|
||||
Total int64 `json:"total"`
|
||||
}
|
||||
|
||||
type PlatformGrantCoinRecordPage struct {
|
||||
Items []PlatformGrantCoinRecord `json:"items"`
|
||||
Page int `json:"page"`
|
||||
PageSize int `json:"pageSize"`
|
||||
Total int64 `json:"total"`
|
||||
}
|
||||
|
||||
type SelfGameEventStat struct {
|
||||
GameID string `json:"game_id"`
|
||||
EventName string `json:"event_name"`
|
||||
@ -432,15 +488,21 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
|
||||
if activityPools, ok, err := r.queryLuckyGiftPoolsFromActivityStats(ctx, app, startDay, endDay, query.CountryID, query.RegionID); err != nil {
|
||||
return Overview{}, err
|
||||
} else if ok {
|
||||
oldTurnover, oldPayout := overview.LuckyGiftTurnover, overview.LuckyGiftPayout
|
||||
overview.LuckyGiftPools = activityPools
|
||||
overview.LuckyGiftTurnover, overview.LuckyGiftPayout = sumLuckyGiftPools(activityPools)
|
||||
overview.ConsumedCoin = replaceEmbeddedFlow(overview.ConsumedCoin, oldTurnover, overview.LuckyGiftTurnover)
|
||||
overview.OutputCoin = replaceEmbeddedFlow(overview.OutputCoin, oldPayout, overview.LuckyGiftPayout)
|
||||
overview.SuperLuckyGiftTurnover, overview.SuperLuckyGiftPayout = sumSuperLuckyGiftPools(activityPools)
|
||||
}
|
||||
// 上一周期也使用相同的日聚合口径，保证“与昨日相比/前周期相比”不额外查询明细表。
|
||||
if activityPools, ok, err := r.queryLuckyGiftPoolsFromActivityStats(ctx, app, previousStartDay, previousEndDay, query.CountryID, query.RegionID); err != nil {
|
||||
return Overview{}, err
|
||||
} else if ok {
|
||||
oldTurnover, oldPayout := previous.LuckyGiftTurnover, previous.LuckyGiftPayout
|
||||
previous.LuckyGiftTurnover, previous.LuckyGiftPayout = sumLuckyGiftPools(activityPools)
|
||||
previous.ConsumedCoin = replaceEmbeddedFlow(previous.ConsumedCoin, oldTurnover, previous.LuckyGiftTurnover)
|
||||
previous.OutputCoin = replaceEmbeddedFlow(previous.OutputCoin, oldPayout, previous.LuckyGiftPayout)
|
||||
}
|
||||
}
|
||||
applyOverviewDerivedMetrics(&overview)
|
||||
@ -542,8 +604,85 @@ func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOv
|
||||
return overview, nil
|
||||
}
|
||||
|
||||
func (r *Repository) QueryPlatformGrantCoinUsers(ctx context.Context, query PlatformGrantCoinQuery) (PlatformGrantCoinUserPage, error) {
|
||||
app, statTZ, startDay, endDay, page, pageSize, offset := normalizePlatformGrantCoinQuery(query)
|
||||
filter, args := platformGrantCoinFilter(app, statTZ, startDay, endDay, query.CountryID, query.RegionID, 0)
|
||||
out := PlatformGrantCoinUserPage{Items: []PlatformGrantCoinUser{}, Page: page, PageSize: pageSize}
|
||||
// 明细下钻只读 statistics-service 自己维护的事件投影；钱包库是写账事实源，不能在报表查询时跨库扫账务流水。
|
||||
countRow := r.db.QueryRowContext(ctx, `
|
||||
SELECT COUNT(*)
|
||||
FROM (
|
||||
SELECT 1
|
||||
FROM stat_platform_grant_coin_events
|
||||
WHERE `+filter+`
|
||||
GROUP BY user_id, country_id, region_id
|
||||
) grouped`, args...)
|
||||
if err := countRow.Scan(&out.Total); err != nil {
|
||||
return PlatformGrantCoinUserPage{}, err
|
||||
}
|
||||
rowsArgs := append(append([]any{}, args...), pageSize, offset)
|
||||
rows, err := r.db.QueryContext(ctx, `
|
||||
SELECT user_id, country_id, region_id,
|
||||
COALESCE(SUM(amount),0),
|
||||
COUNT(*),
|
||||
COALESCE(MIN(occurred_at_ms),0),
|
||||
COALESCE(MAX(occurred_at_ms),0)
|
||||
FROM stat_platform_grant_coin_events
|
||||
WHERE `+filter+`
|
||||
GROUP BY user_id, country_id, region_id
|
||||
ORDER BY COALESCE(SUM(amount),0) DESC, COALESCE(MAX(occurred_at_ms),0) DESC, user_id DESC
|
||||
LIMIT ? OFFSET ?`, rowsArgs...)
|
||||
if err != nil {
|
||||
return PlatformGrantCoinUserPage{}, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var item PlatformGrantCoinUser
|
||||
if err := rows.Scan(&item.UserID, &item.CountryID, &item.RegionID, &item.TotalCoin, &item.RecordCount, &item.FirstGrantedAtMS, &item.LastGrantedAtMS); err != nil {
|
||||
return PlatformGrantCoinUserPage{}, err
|
||||
}
|
||||
out.Items = append(out.Items, item)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return PlatformGrantCoinUserPage{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *Repository) QueryPlatformGrantCoinRecords(ctx context.Context, query PlatformGrantCoinQuery) (PlatformGrantCoinRecordPage, error) {
|
||||
app, statTZ, startDay, endDay, page, pageSize, offset := normalizePlatformGrantCoinQuery(query)
|
||||
filter, args := platformGrantCoinFilter(app, statTZ, startDay, endDay, query.CountryID, query.RegionID, query.UserID)
|
||||
out := PlatformGrantCoinRecordPage{Items: []PlatformGrantCoinRecord{}, Page: page, PageSize: pageSize}
|
||||
// 来源页按单个用户分页读取事件事实，排序使用事件发生时间，避免同一用户大量奖励一次性塞给前端。
|
||||
if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM stat_platform_grant_coin_events WHERE `+filter, args...).Scan(&out.Total); err != nil {
|
||||
return PlatformGrantCoinRecordPage{}, err
|
||||
}
|
||||
rowsArgs := append(append([]any{}, args...), pageSize, offset)
|
||||
rows, err := r.db.QueryContext(ctx, `
|
||||
SELECT event_id, event_type, user_id, country_id, region_id, amount, occurred_at_ms
|
||||
FROM stat_platform_grant_coin_events
|
||||
WHERE `+filter+`
|
||||
ORDER BY occurred_at_ms DESC, event_id DESC
|
||||
LIMIT ? OFFSET ?`, rowsArgs...)
|
||||
if err != nil {
|
||||
return PlatformGrantCoinRecordPage{}, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var item PlatformGrantCoinRecord
|
||||
if err := rows.Scan(&item.EventID, &item.EventType, &item.UserID, &item.CountryID, &item.RegionID, &item.Amount, &item.OccurredAtMS); err != nil {
|
||||
return PlatformGrantCoinRecordPage{}, err
|
||||
}
|
||||
out.Items = append(out.Items, item)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return PlatformGrantCoinRecordPage{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (Overview, error) {
|
||||
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
|
||||
args := []any{endDay, app, normalizeStatTZ(statTZ), startDay, endDay}
|
||||
filter := ""
|
||||
if countryID > 0 {
|
||||
filter += " AND country_id = ?"
|
||||
@ -558,7 +697,8 @@ func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, statTZ,
|
||||
COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0),
|
||||
COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(coin_seller_stock_coin),0), COALESCE(SUM(mifapay_recharge_usd_minor),0),
|
||||
COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_coin),0),
|
||||
COALESCE(SUM(coin_total),0), COALESCE(SUM(consumed_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(CASE WHEN stat_day = ? THEN coin_total ELSE 0 END),0),
|
||||
COALESCE(SUM(consumed_coin),0), COALESCE(SUM(output_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(manual_grant_coin),0), COALESCE(SUM(salary_usd_minor),0), COALESCE(SUM(salary_transfer_coin),0),
|
||||
COALESCE(SUM(mic_online_ms),0), COALESCE(SUM(mic_online_users),0),
|
||||
COALESCE(SUM(gift_coin_spent),0),
|
||||
@ -568,7 +708,7 @@ func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, statTZ,
|
||||
FROM stat_app_day_country
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`+filter, args...)
|
||||
var overview Overview
|
||||
if err := row.Scan(&overview.NewUsers, &overview.ActiveUsers, &overview.PaidUsers, &overview.RechargeUSDMinor, &overview.NewUserRechargeUSDMinor, &overview.CoinSellerRechargeUSDMinor, &overview.CoinSellerStockCoin, &overview.MifaPayRechargeUSDMinor, &overview.GoogleRechargeUSDMinor, &overview.CoinSellerTransferCoin, &overview.CoinTotal, &overview.ConsumedCoin, &overview.PlatformGrantCoin, &overview.ManualGrantCoin, &overview.SalaryUSDMinor, &overview.SalaryTransferCoin, &overview.MicOnlineMS, &overview.MicOnlineUsers, &overview.GiftCoinSpent, &overview.LuckyGiftTurnover, &overview.LuckyGiftPayout, &overview.LuckyGiftPayers, &overview.GameTurnover, &overview.GamePayout, &overview.GameRefund, &overview.GamePlayers); err != nil {
|
||||
if err := row.Scan(&overview.NewUsers, &overview.ActiveUsers, &overview.PaidUsers, &overview.RechargeUSDMinor, &overview.NewUserRechargeUSDMinor, &overview.CoinSellerRechargeUSDMinor, &overview.CoinSellerStockCoin, &overview.MifaPayRechargeUSDMinor, &overview.GoogleRechargeUSDMinor, &overview.CoinSellerTransferCoin, &overview.CoinTotal, &overview.ConsumedCoin, &overview.OutputCoin, &overview.PlatformGrantCoin, &overview.ManualGrantCoin, &overview.SalaryUSDMinor, &overview.SalaryTransferCoin, &overview.MicOnlineMS, &overview.MicOnlineUsers, &overview.GiftCoinSpent, &overview.LuckyGiftTurnover, &overview.LuckyGiftPayout, &overview.LuckyGiftPayers, &overview.GameTurnover, &overview.GamePayout, &overview.GameRefund, &overview.GamePlayers); err != nil {
|
||||
return Overview{}, err
|
||||
}
|
||||
return overview, nil
|
||||
@ -643,7 +783,7 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay
|
||||
COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0),
|
||||
COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(coin_seller_stock_coin),0), COALESCE(SUM(mifapay_recharge_usd_minor),0),
|
||||
COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_coin),0),
|
||||
COALESCE(SUM(coin_total),0), COALESCE(SUM(consumed_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(coin_total),0), COALESCE(SUM(consumed_coin),0), COALESCE(SUM(output_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(manual_grant_coin),0), COALESCE(SUM(salary_usd_minor),0), COALESCE(SUM(salary_transfer_coin),0),
|
||||
COALESCE(SUM(mic_online_ms),0), COALESCE(SUM(mic_online_users),0),
|
||||
COALESCE(SUM(gift_coin_spent),0),
|
||||
@ -661,7 +801,7 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay
|
||||
for rows.Next() {
|
||||
var item DailySeriesItem
|
||||
var gamePayout, gameRefund int64
|
||||
if err := rows.Scan(&item.StatDay, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.CoinSellerStockCoin, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.CoinTotal, &item.ConsumedCoin, &item.PlatformGrantCoin, &item.ManualGrantCoin, &item.SalaryUSDMinor, &item.SalaryTransferCoin, &item.MicOnlineMS, &item.MicOnlineUsers, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.GameTurnover, &gamePayout, &gameRefund); err != nil {
|
||||
if err := rows.Scan(&item.StatDay, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.CoinSellerStockCoin, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.CoinTotal, &item.ConsumedCoin, &item.OutputCoin, &item.PlatformGrantCoin, &item.ManualGrantCoin, &item.SalaryUSDMinor, &item.SalaryTransferCoin, &item.MicOnlineMS, &item.MicOnlineUsers, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.GameTurnover, &gamePayout, &gameRefund); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item.Label = item.StatDay
|
||||
@ -744,6 +884,8 @@ func applyDailySeriesDerivedMetrics(item *DailySeriesItem) {
|
||||
}
|
||||
item.RechargeUsers = item.PaidUsers
|
||||
item.LuckyGiftProfit = item.LuckyGiftTurnover - item.LuckyGiftPayout
|
||||
item.ConsumeOutputDelta = item.ConsumedCoin - item.OutputCoin
|
||||
item.ConsumeOutputRatio = ratio(item.OutputCoin, item.ConsumedCoin)
|
||||
item.SuperLuckyGiftProfit = item.SuperLuckyGiftTurnover - item.SuperLuckyGiftPayout
|
||||
item.SuperLuckyGiftProfitRate = ratio(item.SuperLuckyGiftProfit, item.SuperLuckyGiftTurnover)
|
||||
item.ARPUUSDMinor = div(item.RechargeUSDMinor, item.ActiveUsers)
|
||||
@ -762,6 +904,8 @@ func applyOverviewDerivedMetrics(overview *Overview) {
|
||||
}
|
||||
overview.GameProfit = overview.GameTurnover - overview.GamePayout - overview.GameRefund
|
||||
overview.GameProfitRate = ratio(overview.GameProfit, overview.GameTurnover)
|
||||
overview.ConsumeOutputDelta = overview.ConsumedCoin - overview.OutputCoin
|
||||
overview.ConsumeOutputRatio = ratio(overview.OutputCoin, overview.ConsumedCoin)
|
||||
overview.ARPUUSDMinor = div(overview.RechargeUSDMinor, overview.ActiveUsers)
|
||||
overview.ARPPUUSDMinor = div(overview.RechargeUSDMinor, overview.PaidUsers)
|
||||
overview.UPValueUSDMinor = overview.ARPPUUSDMinor
|
||||
@ -858,7 +1002,7 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, statTZ, startDay
|
||||
}
|
||||
|
||||
func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, startDay, endDay string, regionID int64) ([]CountryBreakdown, error) {
|
||||
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
|
||||
args := []any{endDay, app, normalizeStatTZ(statTZ), startDay, endDay}
|
||||
filter := ""
|
||||
if regionID > 0 {
|
||||
filter = " AND region_id = ?"
|
||||
@ -870,7 +1014,8 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta
|
||||
COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0),
|
||||
COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(coin_seller_stock_coin),0), COALESCE(SUM(mifapay_recharge_usd_minor),0),
|
||||
COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_coin),0),
|
||||
COALESCE(SUM(coin_total),0), COALESCE(SUM(consumed_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(CASE WHEN stat_day = ? THEN coin_total ELSE 0 END),0),
|
||||
COALESCE(SUM(consumed_coin),0), COALESCE(SUM(output_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(manual_grant_coin),0), COALESCE(SUM(salary_usd_minor),0), COALESCE(SUM(salary_transfer_coin),0),
|
||||
COALESCE(SUM(mic_online_ms),0), COALESCE(SUM(mic_online_users),0),
|
||||
COALESCE(SUM(gift_coin_spent),0),
|
||||
@ -889,7 +1034,7 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta
|
||||
out := []CountryBreakdown{}
|
||||
for rows.Next() {
|
||||
var item CountryBreakdown
|
||||
if err := rows.Scan(&item.CountryID, &item.RegionID, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.CoinSellerStockCoin, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.CoinTotal, &item.ConsumedCoin, &item.PlatformGrantCoin, &item.ManualGrantCoin, &item.SalaryUSDMinor, &item.SalaryTransferCoin, &item.MicOnlineMS, &item.MicOnlineUsers, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.LuckyGiftPayers, &item.GameTurnover, &item.GamePayout, &item.GameRefund, &item.GamePlayers); err != nil {
|
||||
if err := rows.Scan(&item.CountryID, &item.RegionID, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.CoinSellerStockCoin, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.CoinTotal, &item.ConsumedCoin, &item.OutputCoin, &item.PlatformGrantCoin, &item.ManualGrantCoin, &item.SalaryUSDMinor, &item.SalaryTransferCoin, &item.MicOnlineMS, &item.MicOnlineUsers, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.LuckyGiftPayers, &item.GameTurnover, &item.GamePayout, &item.GameRefund, &item.GamePlayers); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
applyCountryBreakdownDerivedMetrics(&item)
|
||||
@ -925,7 +1070,7 @@ func (r *Repository) queryDailyCountryBreakdown(ctx context.Context, app, statTZ
|
||||
COALESCE(SUM(recharge_usd_minor),0), COALESCE(SUM(new_user_recharge_usd_minor),0),
|
||||
COALESCE(SUM(coin_seller_recharge_usd_minor),0), COALESCE(SUM(coin_seller_stock_coin),0), COALESCE(SUM(mifapay_recharge_usd_minor),0),
|
||||
COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_coin),0),
|
||||
COALESCE(SUM(coin_total),0), COALESCE(SUM(consumed_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(coin_total),0), COALESCE(SUM(consumed_coin),0), COALESCE(SUM(output_coin),0), COALESCE(SUM(platform_grant_coin),0),
|
||||
COALESCE(SUM(manual_grant_coin),0), COALESCE(SUM(salary_usd_minor),0), COALESCE(SUM(salary_transfer_coin),0),
|
||||
COALESCE(SUM(mic_online_ms),0), COALESCE(SUM(mic_online_users),0),
|
||||
COALESCE(SUM(gift_coin_spent),0),
|
||||
@ -944,7 +1089,7 @@ func (r *Repository) queryDailyCountryBreakdown(ctx context.Context, app, statTZ
|
||||
out := []DailyCountryStat{}
|
||||
for rows.Next() {
|
||||
var item DailyCountryStat
|
||||
if err := rows.Scan(&item.StatDay, &item.CountryID, &item.RegionID, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.CoinSellerStockCoin, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.CoinTotal, &item.ConsumedCoin, &item.PlatformGrantCoin, &item.ManualGrantCoin, &item.SalaryUSDMinor, &item.SalaryTransferCoin, &item.MicOnlineMS, &item.MicOnlineUsers, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.LuckyGiftPayers, &item.GameTurnover, &item.GamePayout, &item.GameRefund, &item.GamePlayers); err != nil {
|
||||
if err := rows.Scan(&item.StatDay, &item.CountryID, &item.RegionID, &item.NewUsers, &item.ActiveUsers, &item.PaidUsers, &item.RechargeUSDMinor, &item.NewUserRechargeUSDMinor, &item.CoinSellerRechargeUSDMinor, &item.CoinSellerStockCoin, &item.MifaPayRechargeUSDMinor, &item.GoogleRechargeUSDMinor, &item.CoinSellerTransferCoin, &item.CoinTotal, &item.ConsumedCoin, &item.OutputCoin, &item.PlatformGrantCoin, &item.ManualGrantCoin, &item.SalaryUSDMinor, &item.SalaryTransferCoin, &item.MicOnlineMS, &item.MicOnlineUsers, &item.GiftCoinSpent, &item.LuckyGiftTurnover, &item.LuckyGiftPayout, &item.LuckyGiftPayers, &item.GameTurnover, &item.GamePayout, &item.GameRefund, &item.GamePlayers); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
item.Label = item.StatDay
|
||||
@ -978,6 +1123,8 @@ func applyCountryBreakdownDerivedMetrics(item *CountryBreakdown) {
|
||||
}
|
||||
item.GameProfit = item.GameTurnover - item.GamePayout - item.GameRefund
|
||||
item.GameProfitRate = ratio(item.GameProfit, item.GameTurnover)
|
||||
item.ConsumeOutputDelta = item.ConsumedCoin - item.OutputCoin
|
||||
item.ConsumeOutputRatio = ratio(item.OutputCoin, item.ConsumedCoin)
|
||||
item.ARPUUSDMinor = div(item.RechargeUSDMinor, item.ActiveUsers)
|
||||
item.ARPPUUSDMinor = div(item.RechargeUSDMinor, item.PaidUsers)
|
||||
item.AvgMicOnlineMS = div(item.MicOnlineMS, item.MicOnlineUsers)
|
||||
@ -1119,8 +1266,11 @@ func mergeActivityLuckyGiftDaily(items []DailySeriesItem, activityDaily map[stri
|
||||
index = len(items) - 1
|
||||
byDay[day] = index
|
||||
}
|
||||
oldTurnover, oldPayout := items[index].LuckyGiftTurnover, items[index].LuckyGiftPayout
|
||||
items[index].LuckyGiftTurnover = activity.TurnoverCoin
|
||||
items[index].LuckyGiftPayout = activity.PayoutCoin
|
||||
items[index].ConsumedCoin = replaceEmbeddedFlow(items[index].ConsumedCoin, oldTurnover, activity.TurnoverCoin)
|
||||
items[index].OutputCoin = replaceEmbeddedFlow(items[index].OutputCoin, oldPayout, activity.PayoutCoin)
|
||||
applyDailySeriesDerivedMetrics(&items[index])
|
||||
}
|
||||
sort.SliceStable(items, func(left, right int) bool {
|
||||
@ -1150,6 +1300,13 @@ func sumSuperLuckyGiftPools(pools []LuckyGiftPoolStat) (int64, int64) {
|
||||
return turnover, payout
|
||||
}
|
||||
|
||||
func replaceEmbeddedFlow(total, oldValue, newValue int64) int64 {
|
||||
if total >= oldValue {
|
||||
return total - oldValue + newValue
|
||||
}
|
||||
return total + newValue
|
||||
}
|
||||
|
||||
func (r *Repository) querySuperLuckyGiftDaily(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (map[string]superLuckyAggregate, error) {
|
||||
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
|
||||
filter := ""
|
||||
@ -1262,9 +1419,11 @@ func reportMetricSources() []ReportMetricSource {
|
||||
{Key: "lucky_gift_turnover", Label: "幸运礼物流水/利润率", Status: "available", Source: "RoomGiftSent + WalletLuckyGiftRewardCredited", Definition: "带 pool_id 的送礼金币为流水，钱包返奖为返奖，利润率=(流水-返奖)/流水。"},
|
||||
{Key: "super_lucky_gift_turnover", Label: "超级幸运礼物流水/利润率", Status: "available", Source: "stat_lucky_gift_pool_day_country", Definition: "pool_id 包含 super_lucky 的奖池流水和返奖派生，利润率=(流水-返奖)/流水。"},
|
||||
{Key: "gift_coin_spent", Label: "礼物流水", Status: "available", Source: "RoomGiftSent", Definition: "非机器人真实送礼 coin_spent，历史事件无 coin_spent 时回退 gift_value。"},
|
||||
{Key: "coin_total", Label: "金币数量(总和)", Status: "available", Source: "wallet_accounts / wallet_entries", Definition: "普通用户 COIN 钱包日终 available+frozen 余额总和；当天实时行按 WalletBalanceChanged 增量滚动，回填固化日终余额。"},
|
||||
{Key: "consumed_coin", Label: "消耗金币", Status: "available", Source: "WalletBalanceChanged biz_type allowlist", Definition: "仅统计消费类普通 COIN 扣减：送礼、VIP、资源商城、红包创建、游戏下注、转盘抽奖和 CP 分手费。"},
|
||||
{Key: "platform_grant_coin", Label: "平台发放金币", Status: "available", Source: "Wallet*RewardCredited", Definition: "系统奖励类 COIN 正向发放，包括任务、幸运礼物返奖、转盘、房间流水、邀请活动和代理开通奖励。"},
|
||||
{Key: "coin_total", Label: "用户金币数", Status: "available", Source: "wallet_entries / WalletBalanceChanged", Definition: "普通用户 COIN 钱包 available+frozen 余额总和；多日总览只取结束日余额快照，日序列展示每日日终快照，当天按余额变动事件滚动更新。"},
|
||||
{Key: "consumed_coin", Label: "消耗金币", Status: "available", Source: "GameOrderSettled / RoomGiftSent / WalletBalanceChanged", Definition: "游戏投入 + 普通礼物投入 + lucky 礼物投入 + 转盘投入 + 发红包；礼物和游戏分别以 room/game owner 事实为准，钱包只补转盘、红包和私聊礼物扣费。"},
|
||||
{Key: "output_coin", Label: "产出金币", Status: "available", Source: "GameOrderSettled / WalletGiftIncomeCredited / WalletLuckyGiftRewardCredited / WalletWheelRewardCredited / WalletBalanceChanged", Definition: "游戏产出 + 收礼产出 + lucky 礼物产出 + 转盘金币奖励 + 红包领取和退回；转盘礼物价值需钱包奖励事件携带金币价值后才能统计。"},
|
||||
{Key: "consume_output_ratio", Label: "消耗产出比", Status: "available", Source: "stat_app_day_country", Definition: "比例=产出金币/消耗金币，差额=消耗金币-产出金币。"},
|
||||
{Key: "platform_grant_coin", Label: "平台发放金币", Status: "available", Source: "Wallet*RewardCredited / WalletBalanceChanged biz_type=resource_grant", Definition: "排行榜奖励、活动奖励、签到/任务奖励和火箭奖励中的金币发放；resource_grant 只统计可识别命令前缀的 COIN 入账，不包含人工发放、装扮价值、lucky 返奖和转盘产出。"},
|
||||
{Key: "manual_grant_coin", Label: "人工发放金币", Status: "available", Source: "WalletBalanceChanged biz_type=manual_credit", Definition: "后台人工调账向普通 COIN 钱包正向发放的金币。"},
|
||||
{Key: "salary_usd_minor", Label: "工资总和(主播/代理)", Status: "available", Source: "host_salary_settlement_records / WalletBalanceChanged", Definition: "成功结算的主播工资和代理工资 USD minor 合计，不包含 residual、BD/Admin 工资。"},
|
||||
{Key: "salary_transfer_coin", Label: "工资兑换金币", Status: "available", Source: "WalletBalanceChanged biz_type=salary_transfer_to_coin_seller / wallet_transactions", Definition: "用户工资钱包转给币商时，币商 COIN_SELLER_COIN 入账的金币数量；国家/区域按发起转账用户归属。"},
|
||||
@ -1592,6 +1751,51 @@ func normalizeRange(startMS, endMS int64) (int64, int64) {
|
||||
return startMS, endMS
|
||||
}
|
||||
|
||||
func normalizePlatformGrantCoinQuery(query PlatformGrantCoinQuery) (string, string, string, string, int, int, int) {
|
||||
app := appcode.Normalize(query.AppCode)
|
||||
if app == "" {
|
||||
app = "lalu"
|
||||
}
|
||||
statTZ := normalizeStatTZ(query.StatTZ)
|
||||
startDay := strings.TrimSpace(query.StatDay)
|
||||
endDay := startDay
|
||||
if startDay == "" {
|
||||
startMS, endMS := normalizeRange(query.StartMS, query.EndMS)
|
||||
startDay = statDayIn(startMS, statTZ)
|
||||
endDay = statDayIn(endMS-1, statTZ)
|
||||
}
|
||||
page := query.Page
|
||||
if page < 1 {
|
||||
page = 1
|
||||
}
|
||||
pageSize := query.PageSize
|
||||
if pageSize < 1 {
|
||||
pageSize = 20
|
||||
}
|
||||
if pageSize > 100 {
|
||||
pageSize = 100
|
||||
}
|
||||
return app, statTZ, startDay, endDay, page, pageSize, (page - 1) * pageSize
|
||||
}
|
||||
|
||||
func platformGrantCoinFilter(app string, statTZ string, startDay string, endDay string, countryID int64, regionID int64, userID int64) (string, []any) {
|
||||
filter := "app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?"
|
||||
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
|
||||
if countryID > 0 {
|
||||
filter += " AND country_id = ?"
|
||||
args = append(args, countryID)
|
||||
}
|
||||
if regionID > 0 {
|
||||
filter += " AND region_id = ?"
|
||||
args = append(args, regionID)
|
||||
}
|
||||
if userID > 0 {
|
||||
filter += " AND user_id = ?"
|
||||
args = append(args, userID)
|
||||
}
|
||||
return filter, args
|
||||
}
|
||||
|
||||
func previousStatDayRange(startDay, endDay string) (string, string) {
|
||||
start, err := time.Parse("2006-01-02", startDay)
|
||||
if err != nil {
|
||||
|
||||
@ -152,8 +152,9 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) {
|
||||
}
|
||||
for _, event := range []WalletBalanceChangedEvent{
|
||||
{AppCode: "lalu", EventID: "balance:coin:recharge", UserID: 7001, AssetType: "COIN", BizType: "google_play_recharge", AvailableDelta: 1000, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:coin:gift", UserID: 7001, AssetType: "COIN", BizType: "gift_debit", AvailableDelta: -200, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:coin:refund", UserID: 7001, AssetType: "COIN", BizType: "game_refund", AvailableDelta: 50, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:coin:red-packet-create", UserID: 7001, AssetType: "COIN", BizType: "red_packet_create", AvailableDelta: -200, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:coin:red-packet-refund", UserID: 7001, AssetType: "COIN", BizType: "red_packet_refund", AvailableDelta: 50, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:coin:checkin-resource", UserID: 7001, AssetType: "COIN", BizType: "resource_grant", CommandID: "wcheckin_claim_1", AvailableDelta: 66, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:coin:manual", UserID: 7001, AssetType: "COIN", BizType: "manual_credit", AvailableDelta: 70, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:salary:host", UserID: 7001, AssetType: "HOST_SALARY_USD", BizType: "host_salary_settlement", AvailableDelta: 123, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "balance:salary:coin-seller", UserID: 9001, DimensionUserID: 7001, AssetType: "COIN_SELLER_COIN", BizType: "salary_transfer_to_coin_seller", AvailableDelta: 444, OccurredAtMS: occurredAt},
|
||||
@ -167,6 +168,11 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) {
|
||||
}); err != nil {
|
||||
t.Fatalf("consume platform grant: %v", err)
|
||||
}
|
||||
if err := repository.ConsumePlatformGrantCoin(ctx, PlatformGrantCoinEvent{
|
||||
AppCode: "lalu", EventID: "grant:wheel:1", EventType: "WalletWheelRewardCredited", UserID: 7001, Amount: 40, OccurredAtMS: occurredAt,
|
||||
}); err != nil {
|
||||
t.Fatalf("consume wheel output: %v", err)
|
||||
}
|
||||
for _, event := range []UserMicDailyStatsUpdatedEvent{
|
||||
{AppCode: "lalu", EventID: "mic:7001:first", UserID: 7001, StatDate: "2026-06-22", MicOnlineDeltaMS: 60_000, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "mic:7001:second", UserID: 7001, StatDate: "2026-06-22", MicOnlineDeltaMS: 30_000, OccurredAtMS: occurredAt},
|
||||
@ -182,7 +188,7 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("query overview: %v", err)
|
||||
}
|
||||
if overview.CoinTotal != 920 || overview.ConsumedCoin != 200 || overview.PlatformGrantCoin != 33 || overview.ManualGrantCoin != 70 || overview.SalaryUSDMinor != 123 || overview.SalaryTransferCoin != 444 {
|
||||
if overview.CoinTotal != 986 || overview.ConsumedCoin != 200 || overview.OutputCoin != 90 || overview.ConsumeOutputDelta != 110 || overview.ConsumeOutputRatio != 0.45 || overview.PlatformGrantCoin != 99 || overview.ManualGrantCoin != 70 || overview.SalaryUSDMinor != 123 || overview.SalaryTransferCoin != 444 {
|
||||
t.Fatalf("report wallet metrics mismatch: %+v", overview)
|
||||
}
|
||||
if overview.MicOnlineMS != 120_000 || overview.MicOnlineUsers != 2 || overview.AvgMicOnlineMS != 60_000 {
|
||||
@ -192,14 +198,132 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) {
|
||||
t.Fatalf("country breakdown missing: %+v", overview.CountryBreakdown)
|
||||
}
|
||||
country := overview.CountryBreakdown[0]
|
||||
if country.CoinTotal != 920 || country.ConsumedCoin != 200 || country.SalaryTransferCoin != 444 || country.AvgMicOnlineMS != 60_000 {
|
||||
if country.CoinTotal != 986 || country.ConsumedCoin != 200 || country.OutputCoin != 90 || country.PlatformGrantCoin != 99 || country.SalaryTransferCoin != 444 || country.AvgMicOnlineMS != 60_000 {
|
||||
t.Fatalf("country report metrics mismatch: %+v", country)
|
||||
}
|
||||
if len(overview.DailySeries) != 1 || overview.DailySeries[0].SalaryTransferCoin != 444 || overview.DailySeries[0].AvgMicOnlineMS != 60_000 {
|
||||
if len(overview.DailySeries) != 1 || overview.DailySeries[0].OutputCoin != 90 || overview.DailySeries[0].SalaryTransferCoin != 444 || overview.DailySeries[0].AvgMicOnlineMS != 60_000 {
|
||||
t.Fatalf("daily report metrics mismatch: %+v", overview.DailySeries)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryOverviewUsesEndDayCoinSnapshotForRange(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
_, file, _, _ := runtime.Caller(0)
|
||||
statsSchema := mysqlschema.New(t, mysqlschema.Config{
|
||||
InitDBPath: mysqlschema.InitDBPath(t, file, "..", "..", "..", "deploy", "mysql", "initdb", "001_statistics_service.sql"),
|
||||
DatabasePrefix: "hy_stats_coin_snapshot_range_test",
|
||||
})
|
||||
repository, err := Open(ctx, statsSchema.DSN)
|
||||
if err != nil {
|
||||
t.Fatalf("open repository: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = repository.Close() })
|
||||
|
||||
if _, err := repository.db.ExecContext(ctx, `
|
||||
INSERT INTO stat_app_day_country (
|
||||
app_code, stat_day, country_id, region_id, coin_total, consumed_coin, output_coin, updated_at_ms
|
||||
) VALUES
|
||||
('lalu', '2026-06-24', 15, 10, 100, 7, 3, 1),
|
||||
('lalu', '2026-06-25', 15, 10, 250, 11, 4, 1)`); err != nil {
|
||||
t.Fatalf("seed coin snapshots: %v", err)
|
||||
}
|
||||
|
||||
start := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC).UnixMilli()
|
||||
overview, err := repository.QueryOverview(ctx, OverviewQuery{AppCode: "lalu", StartMS: start, EndMS: start + int64(48*time.Hour/time.Millisecond), CountryID: 15, RegionID: 10})
|
||||
if err != nil {
|
||||
t.Fatalf("query overview: %v", err)
|
||||
}
|
||||
if overview.CoinTotal != 250 || overview.ConsumedCoin != 18 || overview.OutputCoin != 7 {
|
||||
t.Fatalf("overview must use end-day coin snapshot and range flow sums: %+v", overview)
|
||||
}
|
||||
if len(overview.DailySeries) != 2 || overview.DailySeries[0].CoinTotal != 100 || overview.DailySeries[1].CoinTotal != 250 {
|
||||
t.Fatalf("daily series must keep each day's coin snapshot: %+v", overview.DailySeries)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryPlatformGrantCoinDrilldown(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
_, file, _, _ := runtime.Caller(0)
|
||||
statsSchema := mysqlschema.New(t, mysqlschema.Config{
|
||||
InitDBPath: mysqlschema.InitDBPath(t, file, "..", "..", "..", "deploy", "mysql", "initdb", "001_statistics_service.sql"),
|
||||
DatabasePrefix: "hy_stats_platform_grant_detail_test",
|
||||
})
|
||||
repository, err := Open(ctx, statsSchema.DSN)
|
||||
if err != nil {
|
||||
t.Fatalf("open repository: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = repository.Close() })
|
||||
|
||||
occurredAt := time.Date(2026, 6, 22, 3, 4, 5, 0, time.UTC).UnixMilli()
|
||||
for _, event := range []UserRegisteredEvent{
|
||||
{AppCode: "lalu", EventID: "user_registered:grant_detail:7001", UserID: 7001, CountryID: 15, RegionID: 10, OccurredAtMS: occurredAt},
|
||||
{AppCode: "lalu", EventID: "user_registered:grant_detail:7002", UserID: 7002, CountryID: 15, RegionID: 10, OccurredAtMS: occurredAt},
|
||||
} {
|
||||
if err := repository.ConsumeUserRegistered(ctx, event); err != nil {
|
||||
t.Fatalf("consume user dimension %d: %v", event.UserID, err)
|
||||
}
|
||||
}
|
||||
if err := repository.ConsumePlatformGrantCoin(ctx, PlatformGrantCoinEvent{
|
||||
AppCode: "lalu", EventID: "grant:task:7001", EventType: "WalletTaskRewardCredited", UserID: 7001, Amount: 33, OccurredAtMS: occurredAt,
|
||||
}); err != nil {
|
||||
t.Fatalf("consume task grant: %v", err)
|
||||
}
|
||||
if err := repository.ConsumeLuckyGiftReward(ctx, LuckyGiftRewardEvent{
|
||||
AppCode: "lalu", EventID: "grant:lucky:7001", UserID: 7001, CountryID: 15, RegionID: 10, Amount: 22, OccurredAtMS: occurredAt,
|
||||
}); err != nil {
|
||||
t.Fatalf("consume lucky grant: %v", err)
|
||||
}
|
||||
if err := repository.ConsumePlatformGrantCoin(ctx, PlatformGrantCoinEvent{
|
||||
AppCode: "lalu", EventID: "grant:task:7002", EventType: "WalletTaskRewardCredited", UserID: 7002, Amount: 12, OccurredAtMS: occurredAt,
|
||||
}); err != nil {
|
||||
t.Fatalf("consume second task grant: %v", err)
|
||||
}
|
||||
|
||||
users, err := repository.QueryPlatformGrantCoinUsers(ctx, PlatformGrantCoinQuery{
|
||||
AppCode: "lalu",
|
||||
StatDay: "2026-06-22",
|
||||
CountryID: 15,
|
||||
RegionID: 10,
|
||||
Page: 1,
|
||||
PageSize: 1,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("query grant users: %v", err)
|
||||
}
|
||||
if users.Total != 2 || users.Page != 1 || users.PageSize != 1 || len(users.Items) != 1 {
|
||||
t.Fatalf("grant user page mismatch: %+v", users)
|
||||
}
|
||||
if got := users.Items[0]; got.UserID != 7001 || got.TotalCoin != 33 || got.RecordCount != 1 || got.CountryID != 15 || got.RegionID != 10 {
|
||||
t.Fatalf("first grant user mismatch: %+v", got)
|
||||
}
|
||||
|
||||
records, err := repository.QueryPlatformGrantCoinRecords(ctx, PlatformGrantCoinQuery{
|
||||
AppCode: "lalu",
|
||||
StatDay: "2026-06-22",
|
||||
CountryID: 15,
|
||||
RegionID: 10,
|
||||
UserID: 7001,
|
||||
Page: 1,
|
||||
PageSize: 20,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("query grant records: %v", err)
|
||||
}
|
||||
if records.Total != 1 || len(records.Items) != 1 {
|
||||
t.Fatalf("grant record page mismatch: %+v", records)
|
||||
}
|
||||
amountByType := map[string]int64{}
|
||||
for _, item := range records.Items {
|
||||
amountByType[item.EventType] += item.Amount
|
||||
if item.UserID != 7001 || item.CountryID != 15 || item.RegionID != 10 {
|
||||
t.Fatalf("grant record dimension mismatch: %+v", item)
|
||||
}
|
||||
}
|
||||
if amountByType["WalletTaskRewardCredited"] != 33 || amountByType["WalletLuckyGiftRewardCredited"] != 0 {
|
||||
t.Fatalf("grant record sources mismatch: %+v", amountByType)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConsumeRechargeWritesUTCAndChinaStatDays(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
_, file, _, _ := runtime.Caller(0)
|
||||
|
||||
@ -112,7 +112,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
mifapay_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
|
||||
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, coin_seller_transfer_coin BIGINT NOT NULL DEFAULT 0,
|
||||
coin_total BIGINT NOT NULL DEFAULT 0, consumed_coin BIGINT NOT NULL DEFAULT 0,
|
||||
platform_grant_coin BIGINT NOT NULL DEFAULT 0, manual_grant_coin BIGINT NOT NULL DEFAULT 0,
|
||||
output_coin BIGINT NOT NULL DEFAULT 0, platform_grant_coin BIGINT NOT NULL DEFAULT 0, manual_grant_coin BIGINT NOT NULL DEFAULT 0,
|
||||
salary_usd_minor BIGINT NOT NULL DEFAULT 0, salary_transfer_coin BIGINT NOT NULL DEFAULT 0,
|
||||
mic_online_ms BIGINT NOT NULL DEFAULT 0, mic_online_users BIGINT NOT NULL DEFAULT 0,
|
||||
gift_coin_spent BIGINT NOT NULL DEFAULT 0,
|
||||
@ -167,6 +167,15 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
KEY idx_stat_lucky_payer_country (app_code, stat_tz, stat_day, country_id),
|
||||
KEY idx_stat_lucky_payer_region (app_code, stat_tz, stat_day, region_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_platform_grant_coin_events (
|
||||
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL,
|
||||
event_id VARCHAR(160) NOT NULL, event_type VARCHAR(64) NOT NULL,
|
||||
user_id BIGINT NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0,
|
||||
amount BIGINT NOT NULL DEFAULT 0, occurred_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, stat_tz, event_id),
|
||||
KEY idx_stat_platform_grant_country (app_code, stat_tz, stat_day, country_id, region_id, amount),
|
||||
KEY idx_stat_platform_grant_user (app_code, stat_tz, stat_day, country_id, user_id, occurred_at_ms)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_lucky_gift_pool_day_country (
|
||||
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||||
region_id BIGINT NOT NULL DEFAULT 0,
|
||||
@ -309,7 +318,8 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN coin_seller_stock_coin BIGINT NOT NULL DEFAULT 0 AFTER coin_seller_recharge_usd_minor`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN coin_total BIGINT NOT NULL DEFAULT 0 AFTER coin_seller_transfer_coin`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN consumed_coin BIGINT NOT NULL DEFAULT 0 AFTER coin_total`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN platform_grant_coin BIGINT NOT NULL DEFAULT 0 AFTER consumed_coin`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN output_coin BIGINT NOT NULL DEFAULT 0 AFTER consumed_coin`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN platform_grant_coin BIGINT NOT NULL DEFAULT 0 AFTER output_coin`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN manual_grant_coin BIGINT NOT NULL DEFAULT 0 AFTER platform_grant_coin`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN salary_usd_minor BIGINT NOT NULL DEFAULT 0 AFTER manual_grant_coin`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN salary_transfer_coin BIGINT NOT NULL DEFAULT 0 AFTER salary_usd_minor`,
|
||||
@ -348,6 +358,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
"stat_user_day_mic": {"app_code", "stat_tz", "stat_day", "user_id"},
|
||||
"stat_recharge_day_payers": {"app_code", "stat_tz", "stat_day", "user_id"},
|
||||
"stat_lucky_gift_day_payers": {"app_code", "stat_tz", "stat_day", "user_id"},
|
||||
"stat_platform_grant_coin_events": {"app_code", "stat_tz", "event_id"},
|
||||
"stat_lucky_gift_pool_day_country": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "pool_id"},
|
||||
"stat_real_room_robot_gift_day_rooms": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "room_id"},
|
||||
"stat_game_day_country": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "platform_code", "game_id"},
|
||||
@ -381,6 +392,8 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
{"stat_recharge_day_payers", "idx_stat_recharge_payer_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
|
||||
{"stat_lucky_gift_day_payers", "idx_stat_lucky_payer_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},
|
||||
{"stat_lucky_gift_day_payers", "idx_stat_lucky_payer_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
|
||||
{"stat_platform_grant_coin_events", "idx_stat_platform_grant_country", []string{"app_code", "stat_tz", "stat_day", "country_id", "region_id", "amount"}},
|
||||
{"stat_platform_grant_coin_events", "idx_stat_platform_grant_user", []string{"app_code", "stat_tz", "stat_day", "country_id", "user_id", "occurred_at_ms"}},
|
||||
{"stat_lucky_gift_pool_day_country", "idx_stat_lucky_pool_overview", []string{"app_code", "stat_tz", "stat_day", "pool_id"}},
|
||||
{"stat_lucky_gift_pool_day_country", "idx_stat_lucky_pool_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
|
||||
{"stat_real_room_robot_gift_day_rooms", "idx_stat_real_room_robot_gift_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
|
||||
@ -562,6 +575,17 @@ type PlatformGrantCoinEvent struct {
|
||||
OccurredAtMS int64
|
||||
}
|
||||
|
||||
type CoinOutputEvent struct {
|
||||
AppCode string
|
||||
EventID string
|
||||
EventType string
|
||||
UserID int64
|
||||
CountryID int64
|
||||
RegionID int64
|
||||
Amount int64
|
||||
OccurredAtMS int64
|
||||
}
|
||||
|
||||
type WalletBalanceChangedEvent struct {
|
||||
AppCode string
|
||||
EventID string
|
||||
@ -570,6 +594,7 @@ type WalletBalanceChangedEvent struct {
|
||||
DimensionUserID int64
|
||||
AssetType string
|
||||
BizType string
|
||||
CommandID string
|
||||
AvailableDelta int64
|
||||
FrozenDelta int64
|
||||
CountryID int64
|
||||
@ -856,9 +881,11 @@ func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) e
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE stat_app_day_country
|
||||
SET gift_coin_spent = gift_coin_spent + ?, lucky_gift_turnover = lucky_gift_turnover + ?,
|
||||
lucky_gift_payers = lucky_gift_payers + ?, updated_at_ms = ?
|
||||
lucky_gift_payers = lucky_gift_payers + ?,
|
||||
consumed_coin = consumed_coin + ?,
|
||||
updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||||
`, event.CoinSpent, luckyTurnover, luckyPayers, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
`, event.CoinSpent, luckyTurnover, luckyPayers, event.CoinSpent, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@ -895,8 +922,13 @@ func realRoomRobotGiftKind(event RoomGiftEvent) string {
|
||||
}
|
||||
|
||||
func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGiftRewardEvent) error {
|
||||
return r.withEvent(ctx, SourceWallet, event.EventID, "WalletLuckyGiftRewardCredited", func(tx *sql.Tx, nowMS int64) error {
|
||||
eventType := strings.TrimSpace(event.EventType)
|
||||
if eventType == "" {
|
||||
eventType = "WalletLuckyGiftRewardCredited"
|
||||
}
|
||||
return r.withEvent(ctx, SourceWallet, event.EventID, eventType, func(tx *sql.Tx, nowMS int64) error {
|
||||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||||
amount := normalizeID(event.Amount)
|
||||
poolID := strings.TrimSpace(event.PoolID)
|
||||
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
|
||||
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
|
||||
@ -911,17 +943,17 @@ func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGift
|
||||
UPDATE stat_lucky_gift_pool_day_country
|
||||
SET payout_coin = payout_coin + ?, updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
|
||||
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, poolID); err != nil {
|
||||
`, amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, poolID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE stat_app_day_country
|
||||
SET lucky_gift_payout = lucky_gift_payout + ?,
|
||||
platform_grant_coin = platform_grant_coin + ?,
|
||||
output_coin = output_coin + ?,
|
||||
updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||||
`, event.Amount, event.Amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
`, amount, amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@ -943,13 +975,59 @@ func (r *Repository) ConsumePlatformGrantCoin(ctx context.Context, event Platfor
|
||||
if amount == 0 {
|
||||
return nil
|
||||
}
|
||||
platformGrant, outputCoin := int64(0), int64(0)
|
||||
switch {
|
||||
case isOutputCoinRewardEventType(eventType):
|
||||
outputCoin = amount
|
||||
default:
|
||||
platformGrant = amount
|
||||
}
|
||||
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
|
||||
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE stat_app_day_country
|
||||
SET platform_grant_coin = platform_grant_coin + ?, updated_at_ms = ?
|
||||
SET platform_grant_coin = platform_grant_coin + ?,
|
||||
output_coin = output_coin + ?,
|
||||
updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||||
`, platformGrant, outputCoin, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
return err
|
||||
}
|
||||
if platformGrant > 0 {
|
||||
// 保存事件级来源，后续 Databi 下钻按国家 -> 用户 -> 每条奖励来源分页查询，不需要扫 owner service 账务表。
|
||||
if err := insertPlatformGrantCoinEvent(ctx, tx, event.AppCode, scope.tz, scope.day, event.EventID, eventType, event.UserID, countryID, regionID, platformGrant, event.OccurredAtMS, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Repository) ConsumeCoinOutput(ctx context.Context, event CoinOutputEvent) error {
|
||||
eventType := strings.TrimSpace(event.EventType)
|
||||
if eventType == "" {
|
||||
eventType = "WalletCoinOutputCredited"
|
||||
}
|
||||
return r.withEvent(ctx, SourceWallet, event.EventID, eventType, func(tx *sql.Tx, nowMS int64) error {
|
||||
countryID, regionID, err := r.dimensionForUser(ctx, tx, event.AppCode, event.UserID, event.CountryID, event.RegionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
amount := normalizeID(event.Amount)
|
||||
if amount == 0 {
|
||||
return nil
|
||||
}
|
||||
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
|
||||
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
// 产出金币只写统计读模型总量；来源明细仍由各 owner 的 outbox 和幂等表保留，查询层不回扫钱包流水。
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE stat_app_day_country
|
||||
SET output_coin = output_coin + ?, updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||||
`, amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
return err
|
||||
@ -970,7 +1048,7 @@ func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event Wall
|
||||
return err
|
||||
}
|
||||
totalDelta := event.AvailableDelta + event.FrozenDelta
|
||||
coinTotal, consumed, manualGrant, salaryUSD, salaryTransferCoin := int64(0), int64(0), int64(0), int64(0), int64(0)
|
||||
coinTotal, consumed, outputCoin, platformGrant, manualGrant, salaryUSD, salaryTransferCoin := int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0)
|
||||
assetType := strings.ToUpper(strings.TrimSpace(event.AssetType))
|
||||
bizType := strings.ToLower(strings.TrimSpace(event.BizType))
|
||||
if assetType == "COIN" {
|
||||
@ -978,6 +1056,12 @@ func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event Wall
|
||||
if totalDelta < 0 && isConsumedCoinBizType(bizType) {
|
||||
consumed = -totalDelta
|
||||
}
|
||||
if totalDelta > 0 && isOutputCoinBizType(bizType) {
|
||||
outputCoin = totalDelta
|
||||
}
|
||||
if totalDelta > 0 && bizType == "resource_grant" && isPlatformGrantCommandID(event.CommandID) {
|
||||
platformGrant = totalDelta
|
||||
}
|
||||
if totalDelta > 0 && bizType == "manual_credit" {
|
||||
manualGrant = totalDelta
|
||||
}
|
||||
@ -988,7 +1072,7 @@ func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event Wall
|
||||
if totalDelta > 0 && bizType == "salary_transfer_to_coin_seller" && assetType == "COIN_SELLER_COIN" {
|
||||
salaryTransferCoin = totalDelta
|
||||
}
|
||||
if coinTotal == 0 && consumed == 0 && manualGrant == 0 && salaryUSD == 0 && salaryTransferCoin == 0 {
|
||||
if coinTotal == 0 && consumed == 0 && outputCoin == 0 && platformGrant == 0 && manualGrant == 0 && salaryUSD == 0 && salaryTransferCoin == 0 {
|
||||
return nil
|
||||
}
|
||||
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
|
||||
@ -999,12 +1083,14 @@ func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event Wall
|
||||
UPDATE stat_app_day_country
|
||||
SET coin_total = coin_total + ?,
|
||||
consumed_coin = consumed_coin + ?,
|
||||
output_coin = output_coin + ?,
|
||||
platform_grant_coin = platform_grant_coin + ?,
|
||||
manual_grant_coin = manual_grant_coin + ?,
|
||||
salary_usd_minor = salary_usd_minor + ?,
|
||||
salary_transfer_coin = salary_transfer_coin + ?,
|
||||
updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||||
`, coinTotal, consumed, manualGrant, salaryUSD, salaryTransferCoin, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
`, coinTotal, consumed, outputCoin, platformGrant, manualGrant, salaryUSD, salaryTransferCoin, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@ -1120,9 +1206,12 @@ func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent)
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE stat_app_day_country
|
||||
SET game_turnover = game_turnover + ?, game_payout = game_payout + ?, game_refund = game_refund + ?,
|
||||
game_players = game_players + ?, updated_at_ms = ?
|
||||
game_players = game_players + ?,
|
||||
consumed_coin = consumed_coin + ?,
|
||||
output_coin = output_coin + ?,
|
||||
updated_at_ms = ?
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||||
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
`, turnover, payout, refund, playerDelta, turnover, payout, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@ -1548,8 +1637,48 @@ func (r *Repository) dimensionForUser(ctx context.Context, tx *sql.Tx, app strin
|
||||
|
||||
func isConsumedCoinBizType(value string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "gift_debit", "direct_gift_debit", "vip_purchase", "resource_shop_purchase",
|
||||
"red_packet_create", "game_debit", "wheel_draw_debit", "cp_breakup_fee":
|
||||
case "direct_gift_debit", "red_packet_create", "wheel_draw_debit":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isOutputCoinBizType(value string) bool {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "red_packet_claim", "red_packet_refund":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isOutputCoinRewardEventType(value string) bool {
|
||||
switch strings.TrimSpace(value) {
|
||||
case "WalletWheelRewardCredited":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isPlatformGrantCommandID(value string) bool {
|
||||
normalized := strings.ToLower(strings.TrimSpace(value))
|
||||
if normalized == "" {
|
||||
return false
|
||||
}
|
||||
switch {
|
||||
case strings.HasPrefix(normalized, "weekly_star:"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "cp_weekly_rank:"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "room_turnover_reward:"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "wcheckin_"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "cmd_room_rocket_grant_"):
|
||||
return true
|
||||
case strings.HasPrefix(normalized, "wtask_"), strings.HasPrefix(normalized, "wfr_"), strings.HasPrefix(normalized, "wcr_"), strings.HasPrefix(normalized, "wreg_"):
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
@ -1610,6 +1739,26 @@ func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, statTZ
|
||||
return err
|
||||
}
|
||||
|
||||
func insertPlatformGrantCoinEvent(ctx context.Context, tx *sql.Tx, app string, statTZ string, day string, eventID string, eventType string, userID int64, countryID int64, regionID int64, amount int64, occurredAtMS int64, nowMS int64) error {
|
||||
eventID = strings.TrimSpace(eventID)
|
||||
eventType = strings.TrimSpace(eventType)
|
||||
amount = normalizeID(amount)
|
||||
if eventID == "" || userID <= 0 || amount <= 0 {
|
||||
return nil
|
||||
}
|
||||
if eventType == "" {
|
||||
eventType = "WalletPlatformGrantCoinCredited"
|
||||
}
|
||||
// withEvent 已经保证同一个钱包事件只聚合一次；这里仍用 INSERT IGNORE，让补偿消费或历史修复遇到旧明细时不回滚当天汇总事务。
|
||||
_, err := tx.ExecContext(ctx, `
|
||||
INSERT IGNORE INTO stat_platform_grant_coin_events (
|
||||
app_code, stat_tz, stat_day, event_id, event_type, user_id, country_id, region_id,
|
||||
amount, occurred_at_ms, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`, appcode.Normalize(app), normalizeStatTZ(statTZ), day, eventID, eventType, userID, countryID, regionID, amount, occurredAtMS, nowMS)
|
||||
return err
|
||||
}
|
||||
|
||||
func ensureSelfGameH5EventDay(ctx context.Context, tx *sql.Tx, app string, statTZ string, day string, countryID int64, regionID int64, gameID string, eventName string, language string, clientVersion string, h5Version string, entrySource string, nowMS int64) error {
|
||||
_, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO stat_self_game_h5_event_day (app_code, stat_tz, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source, updated_at_ms)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user