From 38c674e3ebe6122443ee20e530423a16890c911e Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 30 Jun 2026 14:58:48 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E5=B7=A5=E8=B5=84=E6=80=BB?= =?UTF-8?q?=E5=92=8C=E4=BD=99=E9=A2=9D=E5=BF=AB=E7=85=A7=E5=8F=A3=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/数据大屏统计口径.md | 6 +- .../cmd/backfill-last7/main.go | 102 ++++++++---------- .../cmd/databi-real-flow-audit/main.go | 28 +++-- .../mysql/initdb/001_statistics_service.sql | 2 +- .../internal/storage/mysql/query.go | 10 +- .../internal/storage/mysql/query_test.go | 22 ++-- .../internal/storage/mysql/repository.go | 13 ++- 7 files changed, 96 insertions(+), 87 deletions(-) diff --git a/docs/数据大屏统计口径.md b/docs/数据大屏统计口径.md index 1970359d..a9ff4317 100644 --- a/docs/数据大屏统计口径.md +++ b/docs/数据大屏统计口径.md @@ -36,7 +36,7 @@ | `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_usd_minor` | 工资总和(主播/代理) | 主播和代理工资钱包 `HOST_SALARY_USD + AGENCY_SALARY_USD` 的 `available+frozen` 余额总和；历史日取当天 23:59:59.999 前最后一次余额，今日由补数任务按当前已有分录刷新；多日总览只取结束日快照，日序列展示每天快照；不包含 BD/Admin 工资和 residual。 | | `salary_transfer_coin` | 工资兑换金币 | 工资转给币商时币商 `COIN_SELLER_COIN` 入账金币数量，维度按发起转账用户。 | | `mic_online_ms` | 麦上总时长 | `user_mic_daily_stats.mic_online_ms` 按日、国家、区域聚合。 | | `mic_online_users` | 麦上用户数 | 当日 `mic_online_ms > 0` 的去重用户数。 | @@ -79,8 +79,8 @@ | 字段 | 口径 | | --- | --- | | `retention` | `registered_users` 为注册 cohort；`day1/day7/day30_users` 为注册后第 1/7/30 天有活跃的用户；rate 为对应用户数除以注册用户数。 | -| `daily_series` | 按日返回本页主要字段；`coin_total` 是每天的用户金币数快照，其它流水字段按当天累加。 | -| `country_breakdown` | 按国家返回本页主要字段；`coin_total` 是查询结束日该国家用户金币数快照，其它字段按查询周期累加。 | +| `daily_series` | 按日返回本页主要字段；`coin_total` 和 `salary_usd_minor` 是每天余额快照，其它流水字段按当天累加。 | +| `country_breakdown` | 按国家返回本页主要字段；`coin_total` 和 `salary_usd_minor` 是查询结束日该国家余额快照，其它字段按查询周期累加。 | | `daily_country_breakdown` | 按日 + 国家返回主要字段，用于国家弹窗明细。 | | `game_ranking` | 按 `platform_code/game_id` 聚合游戏流水、返奖、退款、玩家数、利润和利润率。 | | `lucky_gift_pools` | 按 `pool_id` 聚合 lucky 礼物流水、返奖、利润、返奖率和利润率。 | diff --git a/services/statistics-service/cmd/backfill-last7/main.go b/services/statistics-service/cmd/backfill-last7/main.go index 45cbff45..223231cc 100644 --- a/services/statistics-service/cmd/backfill-last7/main.go +++ b/services/statistics-service/cmd/backfill-last7/main.go @@ -170,8 +170,8 @@ func main() { 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) + if err := rebuildSalaryTotals(ctx, walletDB, state, app, startDay, endDay, windowEndMS, loc); err != nil { + log.Fatalf("rebuild salary totals: %v", err) } if err := rebuildMicStats(ctx, userDB, state, app, startDay, endDay); err != nil { log.Fatalf("rebuild mic stats: %v", err) @@ -370,25 +370,9 @@ func collectReportMetricUserIDs(ctx context.Context, walletDB *sql.DB, userDB *s query: ` SELECT DISTINCT user_id FROM wallet_accounts - WHERE app_code = ? AND asset_type = 'COIN' AND available_amount + frozen_amount <> 0`, + WHERE app_code = ? AND asset_type IN ('COIN', 'HOST_SALARY_USD', 'AGENCY_SALARY_USD') AND available_amount + frozen_amount <> 0`, args: []any{app}, }, - { - db: walletDB, - query: ` - SELECT DISTINCT user_id - FROM host_salary_settlement_records - WHERE app_code = ? AND status = 'succeeded' AND created_at_ms < ?`, - args: []any{app, windowEndMS}, - }, - { - db: walletDB, - query: ` - SELECT DISTINCT agency_owner_user_id - FROM host_salary_settlement_records - WHERE app_code = ? AND status = 'succeeded' AND agency_owner_user_id > 0 AND created_at_ms < ?`, - args: []any{app, windowEndMS}, - }, { db: userDB, query: ` @@ -675,54 +659,82 @@ func rebuildAppDayCoinMetricsFromStatistics(ctx context.Context, db *sql.DB, sta return rows.Err() } -type coinBalanceEntry struct { +type walletBalanceEntry struct { userID int64 + assetType string balance int64 createdAtMS int64 } func rebuildCoinTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location) error { + return rebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, windowEndMS, loc, []string{"COIN"}, func(agg *appDayAgg, balance int64) { + agg.coinTotal += balance + }) +} + +func rebuildSalaryTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location) error { + return rebuildWalletBalanceTotals(ctx, db, state, app, startDay, endDay, windowEndMS, loc, []string{"HOST_SALARY_USD", "AGENCY_SALARY_USD"}, func(agg *appDayAgg, balance int64) { + agg.salaryUSDMinor += balance + }) +} + +func rebuildWalletBalanceTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location, assetTypes []string, applyBalance func(*appDayAgg, int64)) error { dayEnds, err := dayEndPoints(startDay, endDay, loc) if err != nil { return err } + if len(assetTypes) == 0 { + return nil + } + placeholders := strings.TrimRight(strings.Repeat("?,", len(assetTypes)), ",") + args := []any{app} + for _, assetType := range assetTypes { + args = append(args, strings.ToUpper(strings.TrimSpace(assetType))) + } + args = append(args, windowEndMS) rows, err := db.QueryContext(ctx, ` - SELECT e.user_id, e.available_after + e.frozen_after AS balance_after, e.created_at_ms + SELECT e.user_id, e.asset_type, e.available_after + e.frozen_after AS balance_after, 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 = 'COIN' AND e.created_at_ms < ? + WHERE wt.app_code = ? AND e.asset_type IN (`+placeholders+`) AND e.created_at_ms < ? ORDER BY e.user_id ASC, e.created_at_ms ASC, e.entry_id ASC`, - app, windowEndMS) + args...) if err != nil { return err } defer rows.Close() var currentUserID int64 - entries := []coinBalanceEntry{} + entries := []walletBalanceEntry{} flush := func() { if currentUserID <= 0 || len(entries) == 0 { return } dim := state.userDims[currentUserID] index := 0 - balance := int64(0) - hasBalance := false + balancesByAsset := map[string]int64{} for _, point := range dayEnds { for index < len(entries) && entries[index].createdAtMS < point.endMS { - balance = entries[index].balance - hasBalance = true + balancesByAsset[entries[index].assetType] = entries[index].balance index++ } - if hasBalance { - state.ensureAgg(point.day, dim).coinTotal += balance + if len(balancesByAsset) == 0 { + continue } + balanceTotal := int64(0) + for _, balance := range balancesByAsset { + balanceTotal += balance + } + // 日终余额口径取该用户在当天结束前每个目标资产的最后一条账后余额；工资用户可能同时拥有主播和代理资产， + // 因此先按资产覆盖最后余额再求和，避免同一用户多资产互相覆盖。 + applyBalance(state.ensureAgg(point.day, dim), balanceTotal) } } for rows.Next() { - var entry coinBalanceEntry - if err := rows.Scan(&entry.userID, &entry.balance, &entry.createdAtMS); err != nil { + var entry walletBalanceEntry + if err := rows.Scan(&entry.userID, &entry.assetType, &entry.balance, &entry.createdAtMS); err != nil { return err } + entry.assetType = strings.ToUpper(strings.TrimSpace(entry.assetType)) if currentUserID != 0 && entry.userID != currentUserID { flush() entries = entries[:0] @@ -737,32 +749,6 @@ func rebuildCoinTotals(ctx context.Context, db *sql.DB, state *rebuildState, app return nil } -func rebuildSalary(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS int64, endMS int64, loc *time.Location) error { - rows, err := db.QueryContext(ctx, ` - SELECT user_id, agency_owner_user_id, host_salary_usd_minor_delta, agency_salary_usd_minor_delta, created_at_ms - FROM host_salary_settlement_records - WHERE app_code = ? AND status = 'succeeded' AND created_at_ms >= ? AND created_at_ms < ?`, - app, startMS, endMS) - if err != nil { - return err - } - defer rows.Close() - for rows.Next() { - var hostUserID, agencyUserID, hostSalary, agencySalary, createdAtMS int64 - if err := rows.Scan(&hostUserID, &agencyUserID, &hostSalary, &agencySalary, &createdAtMS); err != nil { - return err - } - day := dayFromMS(createdAtMS, loc) - if hostSalary > 0 { - state.ensureAgg(day, state.userDims[hostUserID]).salaryUSDMinor += hostSalary - } - if agencySalary > 0 && agencyUserID > 0 { - state.ensureAgg(day, state.userDims[agencyUserID]).salaryUSDMinor += agencySalary - } - } - return rows.Err() -} - func rebuildMicStats(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string) error { rows, err := db.QueryContext(ctx, ` SELECT DATE_FORMAT(stat_date, '%Y-%m-%d'), user_id, mic_online_ms diff --git a/services/statistics-service/cmd/databi-real-flow-audit/main.go b/services/statistics-service/cmd/databi-real-flow-audit/main.go index d0f84c9f..0acc8de8 100644 --- a/services/statistics-service/cmd/databi-real-flow-audit/main.go +++ b/services/statistics-service/cmd/databi-real-flow-audit/main.go @@ -109,9 +109,10 @@ func auditSources(ctx context.Context, dbs map[string]*sql.DB, app string, windo SELECT COUNT(*) FROM wallet_outbox WHERE app_code = ? AND event_type = 'WalletRechargeRecorded' AND created_at_ms >= ? AND created_at_ms < ? AND LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload, '$.recharge_type')), '')) IN ('coin_seller_transfer','coin_seller')`, app, window.startMS, window.endMS), - countCheck(ctx, dbs["wallet"], "coin_total", "金币数量总和", "wallet_outbox.WalletBalanceChanged(asset=COIN)", true, ` - SELECT COUNT(*) FROM wallet_outbox - WHERE app_code = ? AND event_type = 'WalletBalanceChanged' AND asset_type = 'COIN' AND created_at_ms >= ? AND created_at_ms < ?`, app, window.startMS, window.endMS), + countCheck(ctx, dbs["wallet"], "coin_total", "用户金币数", "wallet_entries(asset=COIN)", true, ` + SELECT COUNT(*) FROM wallet_entries e + JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id + WHERE wt.app_code = ? AND e.asset_type = 'COIN' AND e.created_at_ms < ?`, app, window.endMS), countCheck(ctx, dbs["wallet"], "consumed_coin", "消耗金币", "wallet_outbox.WalletBalanceChanged(COIN消费biz_type)", true, ` SELECT COUNT(*) FROM wallet_outbox WHERE app_code = ? AND event_type = 'WalletBalanceChanged' AND asset_type = 'COIN' AND available_delta + frozen_delta < 0 @@ -126,11 +127,10 @@ func auditSources(ctx context.Context, dbs map[string]*sql.DB, app string, windo WHERE app_code = ? AND event_type = 'WalletBalanceChanged' AND asset_type = 'COIN' AND available_delta + frozen_delta > 0 AND created_at_ms >= ? AND created_at_ms < ? AND LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload, '$.biz_type')), '')) = 'manual_credit'`, app, window.startMS, window.endMS), - countCheck(ctx, dbs["wallet"], "salary_usd_minor", "工资总和", "wallet_outbox.WalletBalanceChanged(biz_type=host_salary_settlement)", true, ` - SELECT COUNT(*) FROM wallet_outbox - WHERE app_code = ? AND event_type = 'WalletBalanceChanged' AND asset_type IN ('HOST_SALARY_USD','AGENCY_SALARY_USD') - AND available_delta + frozen_delta > 0 AND created_at_ms >= ? AND created_at_ms < ? - AND LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload, '$.biz_type')), '')) = 'host_salary_settlement'`, app, window.startMS, window.endMS), + countCheck(ctx, dbs["wallet"], "salary_usd_minor", "工资总和(主播/代理)", "wallet_entries(asset=HOST_SALARY_USD/AGENCY_SALARY_USD)", true, ` + SELECT COUNT(*) FROM wallet_entries e + JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id + WHERE wt.app_code = ? AND e.asset_type IN ('HOST_SALARY_USD','AGENCY_SALARY_USD') AND e.created_at_ms < ?`, app, window.endMS), countCheck(ctx, dbs["wallet"], "salary_transfer_coin", "工资兑换金币", "wallet_outbox.WalletBalanceChanged(biz_type=salary_transfer_to_coin_seller)", true, ` SELECT COUNT(*) FROM wallet_outbox WHERE app_code = ? AND event_type = 'WalletBalanceChanged' AND asset_type = 'COIN_SELLER_COIN' @@ -164,11 +164,9 @@ func auditStatistics(ctx context.Context, db *sql.DB, app string, statTZ string, {"coin_seller_transfer_coin", "币商出货金币", "SUM(coin_seller_transfer_coin)"}, {"google_recharge_usd_minor", "谷歌充值", "SUM(google_recharge_usd_minor)"}, {"mifapay_recharge_usd_minor", "三方充值", "SUM(mifapay_recharge_usd_minor)"}, - {"coin_total", "金币数量总和", "SUM(coin_total)"}, {"consumed_coin", "消耗金币", "SUM(consumed_coin)"}, {"platform_grant_coin", "平台发放金币", "SUM(platform_grant_coin)"}, {"manual_grant_coin", "人工发放金币", "SUM(manual_grant_coin)"}, - {"salary_usd_minor", "工资总和", "SUM(salary_usd_minor)"}, {"salary_transfer_coin", "工资兑换金币", "SUM(salary_transfer_coin)"}, {"mic_online_ms", "麦上总时长", "SUM(mic_online_ms)"}, {"mic_online_users", "平均麦上时间用户数", "SUM(mic_online_users)"}, @@ -183,6 +181,16 @@ func auditStatistics(ctx context.Context, db *sql.DB, app string, statTZ string, app, statTZ, window.startDay, window.endDay)) } checks = append(checks, + countCheck(ctx, db, "coin_total", "用户金币数", "stat_app_day_country.coin_total(end day)", true, ` + SELECT COALESCE(SUM(CASE WHEN stat_day = ? THEN coin_total ELSE 0 END),0) + FROM stat_app_day_country + WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`, + window.endDay, app, statTZ, window.startDay, window.endDay), + countCheck(ctx, db, "salary_usd_minor", "工资总和(主播/代理)", "stat_app_day_country.salary_usd_minor(end day)", true, ` + SELECT COALESCE(SUM(CASE WHEN stat_day = ? THEN salary_usd_minor ELSE 0 END),0) + FROM stat_app_day_country + WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`, + window.endDay, app, statTZ, window.startDay, window.endDay), countCheck(ctx, db, "super_lucky_gift_turnover", "超级幸运礼物流水", "stat_lucky_gift_pool_day_country(pool_id LIKE %super_lucky%)", true, ` SELECT COALESCE(SUM(turnover_coin),0) FROM stat_lucky_gift_pool_day_country WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ? AND LOWER(pool_id) LIKE '%super_lucky%'`, diff --git a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql index 84798bfc..c92adfe4 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -38,7 +38,7 @@ CREATE TABLE IF NOT EXISTS stat_app_day_country ( 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_usd_minor BIGINT NOT NULL DEFAULT 0 COMMENT '主播和代理工资钱包余额快照 USD 最小单位', salary_transfer_coin BIGINT NOT NULL DEFAULT 0 COMMENT '工资转币商入账金币数量', mic_online_ms BIGINT NOT NULL DEFAULT 0 COMMENT '麦上总时长毫秒', mic_online_users BIGINT NOT NULL DEFAULT 0 COMMENT '麦上用户数', diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index f9885a11..5a5571bd 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -682,7 +682,7 @@ func (r *Repository) QueryPlatformGrantCoinRecords(ctx context.Context, query Pl } func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (Overview, error) { - args := []any{endDay, app, normalizeStatTZ(statTZ), startDay, endDay} + args := []any{endDay, endDay, app, normalizeStatTZ(statTZ), startDay, endDay} filter := "" if countryID > 0 { filter += " AND country_id = ?" @@ -699,7 +699,7 @@ func (r *Repository) queryAppDayOverviewTotals(ctx context.Context, app, statTZ, COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_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(manual_grant_coin),0), COALESCE(SUM(CASE WHEN stat_day = ? THEN salary_usd_minor ELSE 0 END),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), COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payout),0), COALESCE(SUM(lucky_gift_payers),0), @@ -1002,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{endDay, app, normalizeStatTZ(statTZ), startDay, endDay} + args := []any{endDay, endDay, app, normalizeStatTZ(statTZ), startDay, endDay} filter := "" if regionID > 0 { filter = " AND region_id = ?" @@ -1016,7 +1016,7 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta COALESCE(SUM(google_recharge_usd_minor),0), COALESCE(SUM(coin_seller_transfer_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(manual_grant_coin),0), COALESCE(SUM(CASE WHEN stat_day = ? THEN salary_usd_minor ELSE 0 END),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), COALESCE(SUM(lucky_gift_turnover),0), COALESCE(SUM(lucky_gift_payout),0), COALESCE(SUM(lucky_gift_payers),0), @@ -1425,7 +1425,7 @@ func reportMetricSources() []ReportMetricSource { {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_usd_minor", Label: "工资总和(主播/代理)", Status: "available", Source: "wallet_entries / WalletBalanceChanged", Definition: "主播和代理工资钱包 HOST_SALARY_USD+AGENCY_SALARY_USD 的 available+frozen 余额总和；多日总览只取结束日余额快照，日序列展示每日日终快照，不包含 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 入账的金币数量；国家/区域按发起转账用户归属。"}, {Key: "avg_mic_online_ms", Label: "平均麦上时间", Status: "available", Source: "user_mic_daily_stats", Definition: "麦上总时长 / 当日麦上用户数，只统计 mic_online_ms > 0 的用户。"}, {Key: "mic_online_ms", Label: "麦上总时长", Status: "available", Source: "user_mic_daily_stats.mic_online_ms", Definition: "用户每日麦上时长按国家/区域聚合后的毫秒总和。"}, diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index e3565c73..4b372721 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -157,6 +157,7 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) { {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:transfer-debit", UserID: 7001, AssetType: "HOST_SALARY_USD", BizType: "salary_transfer_to_coin_seller", AvailableDelta: -23, 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}, } { if err := repository.ConsumeWalletBalanceChanged(ctx, event); err != nil { @@ -188,7 +189,7 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) { if err != nil { t.Fatalf("query overview: %v", err) } - 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 { + 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 != 100 || overview.SalaryTransferCoin != 444 { t.Fatalf("report wallet metrics mismatch: %+v", overview) } if overview.MicOnlineMS != 120_000 || overview.MicOnlineUsers != 2 || overview.AvgMicOnlineMS != 60_000 { @@ -206,7 +207,7 @@ func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) { } } -func TestQueryOverviewUsesEndDayCoinSnapshotForRange(t *testing.T) { +func TestQueryOverviewUsesEndDayBalanceSnapshotsForRange(t *testing.T) { ctx := context.Background() _, file, _, _ := runtime.Caller(0) statsSchema := mysqlschema.New(t, mysqlschema.Config{ @@ -221,10 +222,10 @@ func TestQueryOverviewUsesEndDayCoinSnapshotForRange(t *testing.T) { 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 + app_code, stat_day, country_id, region_id, coin_total, salary_usd_minor, 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 { + ('lalu', '2026-06-24', 15, 10, 100, 12, 7, 3, 1), + ('lalu', '2026-06-25', 15, 10, 250, 34, 11, 4, 1)`); err != nil { t.Fatalf("seed coin snapshots: %v", err) } @@ -233,11 +234,14 @@ func TestQueryOverviewUsesEndDayCoinSnapshotForRange(t *testing.T) { 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 overview.CoinTotal != 250 || overview.SalaryUSDMinor != 34 || overview.ConsumedCoin != 18 || overview.OutputCoin != 7 { + t.Fatalf("overview must use end-day balance snapshots 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) + if len(overview.DailySeries) != 2 || overview.DailySeries[0].CoinTotal != 100 || overview.DailySeries[0].SalaryUSDMinor != 12 || overview.DailySeries[1].CoinTotal != 250 || overview.DailySeries[1].SalaryUSDMinor != 34 { + t.Fatalf("daily series must keep each day's balance snapshots: %+v", overview.DailySeries) + } + if len(overview.CountryBreakdown) != 1 || overview.CountryBreakdown[0].CoinTotal != 250 || overview.CountryBreakdown[0].SalaryUSDMinor != 34 { + t.Fatalf("country breakdown must use end-day balance snapshots: %+v", overview.CountryBreakdown) } } diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index 6fb569db..a4866776 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -1066,7 +1066,9 @@ func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event Wall manualGrant = totalDelta } } - if totalDelta > 0 && bizType == "host_salary_settlement" && (assetType == "HOST_SALARY_USD" || assetType == "AGENCY_SALARY_USD") { + if isHostAgencySalaryAssetType(assetType) { + // 工资总和是主播/代理工资钱包余额快照，不是结算流水。这里写 available+frozen 的净变化， + // 冻结动作本身一正一负不改变总额，入账/出账/兑换才会改变当天余额快照。 salaryUSD = totalDelta } if totalDelta > 0 && bizType == "salary_transfer_to_coin_seller" && assetType == "COIN_SELLER_COIN" { @@ -1653,6 +1655,15 @@ func isOutputCoinBizType(value string) bool { } } +func isHostAgencySalaryAssetType(value string) bool { + switch strings.ToUpper(strings.TrimSpace(value)) { + case "HOST_SALARY_USD", "AGENCY_SALARY_USD": + return true + default: + return false + } +} + func isOutputCoinRewardEventType(value string) bool { switch strings.TrimSpace(value) { case "WalletWheelRewardCredited":