diff --git a/server/admin/internal/modules/dashboard/handler.go b/server/admin/internal/modules/dashboard/handler.go index fb2feb07..f2cff57f 100644 --- a/server/admin/internal/modules/dashboard/handler.go +++ b/server/admin/internal/modules/dashboard/handler.go @@ -74,6 +74,7 @@ func (h *Handler) PlatformGrantUsers(c *gin.Context) { StatDay: c.Query("stat_day"), RegionID: parseInt64(c.Query("region_id")), CountryID: parseInt64(c.Query("country_id")), + Source: firstNonEmpty(c.Query("source"), c.Query("event_type")), Page: parseInt(c.Query("page")), PageSize: parseInt(firstNonEmpty(c.Query("page_size"), c.Query("pageSize"))), }) @@ -94,6 +95,7 @@ func (h *Handler) PlatformGrantRecords(c *gin.Context) { RegionID: parseInt64(c.Query("region_id")), CountryID: parseInt64(c.Query("country_id")), UserID: parseInt64(c.Query("user_id")), + Source: firstNonEmpty(c.Query("source"), c.Query("event_type")), Page: parseInt(c.Query("page")), PageSize: parseInt(firstNonEmpty(c.Query("page_size"), c.Query("pageSize"))), }) diff --git a/server/admin/internal/modules/dashboard/service.go b/server/admin/internal/modules/dashboard/service.go index c897315d..4c87c562 100644 --- a/server/admin/internal/modules/dashboard/service.go +++ b/server/admin/internal/modules/dashboard/service.go @@ -53,6 +53,7 @@ type PlatformGrantStatisticsQuery struct { RegionID int64 CountryID int64 UserID int64 + Source string Page int PageSize int } @@ -197,6 +198,9 @@ func platformGrantValues(query PlatformGrantStatisticsQuery, includeUser bool) u if includeUser && query.UserID > 0 { values.Set("user_id", strconv.FormatInt(query.UserID, 10)) } + if source := strings.TrimSpace(query.Source); source != "" && source != "all" { + values.Set("source", source) + } if query.Page > 0 { values.Set("page", strconv.Itoa(query.Page)) } 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 0acc8de8..31c983d5 100644 --- a/services/statistics-service/cmd/databi-real-flow-audit/main.go +++ b/services/statistics-service/cmd/databi-real-flow-audit/main.go @@ -118,10 +118,21 @@ 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')), '')) IN ('gift_debit','direct_gift_debit','vip_purchase','resource_shop_purchase','red_packet_create','game_debit','wheel_draw_debit','cp_breakup_fee')`, app, window.startMS, window.endMS), - countCheck(ctx, dbs["wallet"], "platform_grant_coin", "平台发放金币", "wallet_outbox.Wallet*RewardCredited", true, ` + countCheck(ctx, dbs["wallet"], "platform_grant_coin", "平台发放金币", "wallet_outbox 平台奖励/资源发放", true, ` SELECT COUNT(*) FROM wallet_outbox - WHERE app_code = ? AND event_type IN ('WalletTaskRewardCredited','WalletLuckyGiftRewardCredited','WalletWheelRewardCredited','WalletRoomTurnoverRewardCredited','WalletInviteActivityRewardCredited','WalletAgencyOpeningRewardCredited') - AND created_at_ms >= ? AND created_at_ms < ?`, app, window.startMS, window.endMS), + WHERE app_code = ? AND created_at_ms >= ? AND created_at_ms < ? + AND ( + event_type IN ('WalletTaskRewardCredited','WalletRoomTurnoverRewardCredited','WalletInviteActivityRewardCredited','WalletAgencyOpeningRewardCredited') + OR ( + event_type = 'WalletBalanceChanged' AND asset_type = 'COIN' AND available_delta + frozen_delta > 0 + AND LOWER(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(payload, '$.biz_type')), '')) = 'resource_grant' + AND ( + LOWER(command_id) LIKE 'weekly_star:%' OR LOWER(command_id) LIKE 'cp_weekly_rank:%' OR LOWER(command_id) LIKE 'room_turnover_reward:%' + OR LOWER(command_id) LIKE 'wcheckin_%' OR LOWER(command_id) LIKE 'cmd_room_rocket_grant_%' + OR LOWER(command_id) LIKE 'wtask_%' OR LOWER(command_id) LIKE 'wfr_%' OR LOWER(command_id) LIKE 'wcr_%' OR LOWER(command_id) LIKE 'wreg_%' + ) + ) + )`, app, window.startMS, window.endMS), countCheck(ctx, dbs["wallet"], "manual_grant_coin", "人工发放金币", "wallet_outbox.WalletBalanceChanged(biz_type=manual_credit)", true, ` SELECT COUNT(*) FROM wallet_outbox WHERE app_code = ? AND event_type = 'WalletBalanceChanged' AND asset_type = 'COIN' AND available_delta + frozen_delta > 0 diff --git a/services/statistics-service/cmd/replay-stat-tz/main.go b/services/statistics-service/cmd/replay-stat-tz/main.go index 66aee9c1..398e2027 100644 --- a/services/statistics-service/cmd/replay-stat-tz/main.go +++ b/services/statistics-service/cmd/replay-stat-tz/main.go @@ -235,6 +235,7 @@ func resetReplayTarget(ctx context.Context, db *sql.DB, app string, statTZ strin {"stat_recharge_day_payers", "stat_day"}, {"stat_lucky_gift_day_payers", "stat_day"}, {"stat_lucky_gift_pool_day_country", "stat_day"}, + {"stat_platform_grant_coin_events", "stat_day"}, {"stat_real_room_robot_gift_day_rooms", "stat_day"}, {"stat_game_day_country", "stat_day"}, {"stat_game_day_players", "stat_day"}, 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 c92adfe4..cdc2f9b9 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -146,7 +146,8 @@ CREATE TABLE IF NOT EXISTS stat_platform_grant_coin_events ( 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) + KEY idx_stat_platform_grant_user (app_code, stat_tz, stat_day, country_id, user_id, occurred_at_ms), + KEY idx_stat_platform_grant_source (app_code, stat_tz, stat_day, event_type, 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 ( diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go index 5ff82032..0b9f4088 100644 --- a/services/statistics-service/internal/app/query_http.go +++ b/services/statistics-service/internal/app/query_http.go @@ -100,6 +100,7 @@ func (s *queryHTTPServer) platformGrantUsers(w http.ResponseWriter, r *http.Requ StatDay: query.Get("stat_day"), CountryID: parseInt64(query.Get("country_id")), RegionID: parseInt64(query.Get("region_id")), + EventType: firstNonEmpty(query.Get("source"), query.Get("event_type")), Page: parsePositiveInt(query.Get("page")), PageSize: pageSize, }) @@ -130,6 +131,7 @@ func (s *queryHTTPServer) platformGrantRecords(w http.ResponseWriter, r *http.Re CountryID: parseInt64(query.Get("country_id")), RegionID: parseInt64(query.Get("region_id")), UserID: userID, + EventType: firstNonEmpty(query.Get("source"), query.Get("event_type")), Page: parsePositiveInt(query.Get("page")), PageSize: pageSize, }) diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index 5a5571bd..327f6547 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -277,6 +277,7 @@ type PlatformGrantCoinQuery struct { CountryID int64 RegionID int64 UserID int64 + EventType string Page int PageSize int } @@ -606,7 +607,7 @@ func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOv 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) + filter, args := platformGrantCoinFilter(app, statTZ, startDay, endDay, query.CountryID, query.RegionID, 0, query.EventType) out := PlatformGrantCoinUserPage{Items: []PlatformGrantCoinUser{}, Page: page, PageSize: pageSize} // 明细下钻只读 statistics-service 自己维护的事件投影；钱包库是写账事实源，不能在报表查询时跨库扫账务流水。 countRow := r.db.QueryRowContext(ctx, ` @@ -651,7 +652,7 @@ func (r *Repository) QueryPlatformGrantCoinUsers(ctx context.Context, query Plat 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) + filter, args := platformGrantCoinFilter(app, statTZ, startDay, endDay, query.CountryID, query.RegionID, query.UserID, query.EventType) 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 { @@ -1778,7 +1779,7 @@ func normalizePlatformGrantCoinQuery(query PlatformGrantCoinQuery) (string, stri 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) { +func platformGrantCoinFilter(app string, statTZ string, startDay string, endDay string, countryID int64, regionID int64, userID int64, eventType string) (string, []any) { filter := "app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?" args := []any{app, normalizeStatTZ(statTZ), startDay, endDay} if countryID > 0 { @@ -1793,6 +1794,10 @@ func platformGrantCoinFilter(app string, statTZ string, startDay string, endDay filter += " AND user_id = ?" args = append(args, userID) } + if normalizedEventType := strings.TrimSpace(eventType); normalizedEventType != "" && normalizedEventType != "all" { + filter += " AND event_type = ?" + args = append(args, normalizedEventType) + } return filter, args } diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index 4b372721..7805e12b 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -272,6 +272,11 @@ func TestQueryPlatformGrantCoinDrilldown(t *testing.T) { }); err != nil { t.Fatalf("consume task grant: %v", err) } + if err := repository.ConsumeWalletBalanceChanged(ctx, WalletBalanceChangedEvent{ + AppCode: "lalu", EventID: "grant:resource:7001", EventType: "WalletBalanceChanged", UserID: 7001, AssetType: "COIN", BizType: "resource_grant", CommandID: "weekly_star:2026-06-22", AvailableDelta: 7, OccurredAtMS: occurredAt, + }); err != nil { + t.Fatalf("consume resource 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 { @@ -297,7 +302,7 @@ func TestQueryPlatformGrantCoinDrilldown(t *testing.T) { 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 { + if got := users.Items[0]; got.UserID != 7001 || got.TotalCoin != 40 || got.RecordCount != 2 || got.CountryID != 15 || got.RegionID != 10 { t.Fatalf("first grant user mismatch: %+v", got) } @@ -313,7 +318,7 @@ func TestQueryPlatformGrantCoinDrilldown(t *testing.T) { if err != nil { t.Fatalf("query grant records: %v", err) } - if records.Total != 1 || len(records.Items) != 1 { + if records.Total != 2 || len(records.Items) != 2 { t.Fatalf("grant record page mismatch: %+v", records) } amountByType := map[string]int64{} @@ -323,9 +328,25 @@ func TestQueryPlatformGrantCoinDrilldown(t *testing.T) { t.Fatalf("grant record dimension mismatch: %+v", item) } } - if amountByType["WalletTaskRewardCredited"] != 33 || amountByType["WalletLuckyGiftRewardCredited"] != 0 { + if amountByType["WalletTaskRewardCredited"] != 33 || amountByType["WalletBalanceChanged"] != 7 || amountByType["WalletLuckyGiftRewardCredited"] != 0 { t.Fatalf("grant record sources mismatch: %+v", amountByType) } + taskRecords, err := repository.QueryPlatformGrantCoinRecords(ctx, PlatformGrantCoinQuery{ + AppCode: "lalu", + StatDay: "2026-06-22", + CountryID: 15, + RegionID: 10, + UserID: 7001, + EventType: "WalletTaskRewardCredited", + Page: 1, + PageSize: 20, + }) + if err != nil { + t.Fatalf("query source filtered grant records: %v", err) + } + if taskRecords.Total != 1 || len(taskRecords.Items) != 1 || taskRecords.Items[0].EventType != "WalletTaskRewardCredited" || taskRecords.Items[0].Amount != 33 { + t.Fatalf("source filtered grant records mismatch: %+v", taskRecords) + } } func TestConsumeRechargeWritesUTCAndChinaStatDays(t *testing.T) { diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index a4866776..459053e5 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -174,7 +174,8 @@ func (r *Repository) Migrate(ctx context.Context) error { 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) + KEY idx_stat_platform_grant_user (app_code, stat_tz, stat_day, country_id, user_id, occurred_at_ms), + KEY idx_stat_platform_grant_source (app_code, stat_tz, stat_day, event_type, 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, @@ -394,6 +395,7 @@ func (r *Repository) Migrate(ctx context.Context) error { {"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_platform_grant_coin_events", "idx_stat_platform_grant_source", []string{"app_code", "stat_tz", "stat_day", "event_type", "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"}}, @@ -1095,6 +1097,12 @@ func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event Wall `, coinTotal, consumed, outputCoin, platformGrant, manualGrant, salaryUSD, salaryTransferCoin, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil { return err } + if platformGrant > 0 { + // resource_grant 类入账也是平台发放金币，写入同一事件投影，确保总计、用户列表和来源明细的统计口径一致。 + if err := insertPlatformGrantCoinEvent(ctx, tx, event.AppCode, scope.tz, scope.day, event.EventID, event.EventType, event.UserID, countryID, regionID, platformGrant, event.OccurredAtMS, nowMS); err != nil { + return err + } + } } return nil })