diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 9f01ae04..1986ce20 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -7075,6 +7075,14 @@ func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { }, }, }, + hostSalaryProgressResp: &walletv1.GetHostSalaryProgressResponse{Progress: &walletv1.HostSalaryProgress{ + HostUserId: 42, + CycleKey: expectedCycle, + RegionId: 31, + AgencyOwnerUserId: 42, + TotalDiamonds: 30000, + GiftDiamondTotal: 30000, + }}, } handler := NewHandlerWithClients(&fakeRoomClient{}, nil, nil, &fakeUserProfileClient{}) handler.SetUserHostClient(hostClient) @@ -7096,8 +7104,8 @@ func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { if walletClient.lastHostSalaryPolicy == nil || walletClient.lastHostSalaryPolicy.GetRegionId() != 31 || walletClient.lastHostSalaryPolicy.GetCycleKey() != expectedCycle || walletClient.lastHostSalaryPolicy.GetSettlementTriggerMode() != "" || walletClient.lastHostSalaryPolicy.GetAppCode() == "" || walletClient.lastHostSalaryPolicy.GetRequestId() == "" { t.Fatalf("agency policy request mismatch: %+v", walletClient.lastHostSalaryPolicy) } - if walletClient.lastHostSalaryProgress != nil { - t.Fatalf("agency policy should not request host progress: %+v", walletClient.lastHostSalaryProgress) + if walletClient.lastHostSalaryProgress == nil || walletClient.lastHostSalaryProgress.GetHostUserId() != 42 || walletClient.lastHostSalaryProgress.GetCycleKey() != expectedCycle || walletClient.lastHostSalaryProgress.GetAppCode() == "" || walletClient.lastHostSalaryProgress.GetRequestId() == "" { + t.Fatalf("agency owner salary progress request mismatch: %+v", walletClient.lastHostSalaryProgress) } var response httpkit.ResponseEnvelope if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { @@ -7107,9 +7115,13 @@ func TestAgencyCenterPlatformPolicyUsesAgencyRegion(t *testing.T) { policy := data["policy"].(map[string]any) levels := policy["levels"].([]any) secondLevel := levels[1].(map[string]any) + progress := data["progress"].(map[string]any) if data["found"] != true || data["agency_region_id"] != float64(31) || policy["policy_id"] != "9101" || policy["cycle_key"] != expectedCycle || policy["policy_version"] != float64(9101) || len(levels) != 2 || secondLevel["agency_salary_usd"] != 7.2 { t.Fatalf("agency policy response mismatch: %+v", data) } + if progress["host_user_id"] != float64(42) || progress["cycle_key"] != expectedCycle || progress["total_diamonds"] != float64(30000) { + t.Fatalf("agency owner salary progress response mismatch: %+v", progress) + } } func TestBDCenterOverviewUsesBDSalaryAndDirectAgencies(t *testing.T) { diff --git a/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go b/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go index 46397f5b..6395ec99 100644 --- a/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go +++ b/services/gateway-service/internal/transport/http/userapi/agency_center_handler.go @@ -28,12 +28,6 @@ type agencyCenterSalaryData struct { DisplayAmount float64 `json:"display_amount"` } -type agencyCenterDiamondProgressData struct { - CycleKey string `json:"cycle_key"` - TotalDiamonds int64 `json:"total_diamonds"` - ActiveHostCount int32 `json:"active_host_count"` -} - type agencyCenterUserData struct { UserID string `json:"user_id"` DisplayUserID string `json:"display_user_id"` @@ -126,8 +120,7 @@ func (h *Handler) getAgencyCenterPlatformPolicy(writer http.ResponseWriter, requ return } - now := time.Now().UTC() - cycleKey := now.Format("2006-01") + cycleKey := time.Now().UTC().Format("2006-01") // Agency Center 不是主播身份，不能复用 host-center 的 active host 校验；这里固定用当前 agency 的区域读取完整工资政策。 resp, err := h.walletClient.GetActiveHostSalaryPolicy(request.Context(), &walletv1.GetActiveHostSalaryPolicyRequest{ RequestId: httpkit.RequestIDFromContext(request.Context()), @@ -141,35 +134,24 @@ func (h *Handler) getAgencyCenterPlatformPolicy(writer http.ResponseWriter, requ return } - // Agency 卡片展示的是本工资周期内归属该 owner 的主播钻石合计。聚合依据入账时固化的 - // agency_owner_user_id 快照，既不会把 owner 的用户 COIN 钱包混进来，也不会因主播后来换公会改写历史归属。 - statsResp, err := h.walletClient.GetTeamHostSalaryStats(request.Context(), &walletv1.GetTeamHostSalaryStatsRequest{ - RequestId: httpkit.RequestIDFromContext(request.Context()), - AppCode: appcode.FromContext(request.Context()), - AgencyOwnerUserIds: []int64{agency.GetOwnerUserId()}, - CycleKeys: []string{cycleKey}, - NowMs: now.UnixMilli(), + // Agency 和 Host Center 的钻石卡片都代表当前登录用户本人的周期主播账户。 + // 这里直接按 owner user_id 读取同一份账务事实，避免 Agency 把下属主播合计误显示成 owner 本人的钻石。 + progressResp, err := h.walletClient.GetHostSalaryProgress(request.Context(), &walletv1.GetHostSalaryProgressRequest{ + RequestId: httpkit.RequestIDFromContext(request.Context()), + AppCode: appcode.FromContext(request.Context()), + HostUserId: agency.GetOwnerUserId(), + CycleKey: cycleKey, }) if err != nil { httpkit.WriteRPCError(writer, request, err) return } - progress := agencyCenterDiamondProgressData{CycleKey: cycleKey} - for _, stat := range statsResp.GetStats() { - if stat.GetCycleKey() != cycleKey { - continue - } - progress.TotalDiamonds = stat.GetTotalDiamonds() - progress.ActiveHostCount = stat.GetActiveHostCount() - break - } - - // H5 需要完整政策明细和 Agency 周期合计；Agency 没有单主播等级，因此不返回 host-center 的 level_progress。 + // H5 需要完整政策明细和 owner 本人的周期进度；Agency 没有单主播等级卡，因此不返回 level_progress。 httpkit.WriteOK(writer, request, map[string]any{ "found": resp.GetFound(), "agency_region_id": agency.GetRegionId(), "policy": hostCenterPolicyFromProto(resp.GetPolicy()), - "progress": progress, + "progress": hostCenterSalaryProgressFromProto(progressResp.GetProgress()), }) } diff --git a/services/user-service/internal/storage/mysql/host/stats.go b/services/user-service/internal/storage/mysql/host/stats.go index 92759286..b41786eb 100644 --- a/services/user-service/internal/storage/mysql/host/stats.go +++ b/services/user-service/internal/storage/mysql/host/stats.go @@ -31,12 +31,44 @@ func (r *Repository) GetHostEngagementStats(ctx context.Context, query hostdomai return hostdomain.EngagementStats{}, err } - if err := r.db.QueryRowContext(ctx, ` - SELECT COUNT(DISTINCT sender_user_id) - FROM user_private_message_events - WHERE app_code = ? AND target_user_id = ? AND occurred_at_ms >= ? AND occurred_at_ms < ?`, - appCode, query.HostUserID, query.StartAtMS, query.EndAtMS, - ).Scan(&stats.PrivateMessageSenders); err != nil { + // 旧版 C2C 回调缺 app_code 时会把 Fami 事实写到默认租户。读取仅扩展到这两个索引范围， + // 并用 users 全局主键校验发送者真实属于当前 App，既恢复历史数据又不放大跨租户计数。 + privateMessageSQL := ` + SELECT COUNT(DISTINCT matched.sender_user_id) + FROM ( + SELECT sender_user_id + FROM user_private_message_events FORCE INDEX (idx_private_message_target_time) + WHERE app_code = ? AND target_user_id = ? + AND occurred_at_ms >= ? AND occurred_at_ms < ? + ) matched + INNER JOIN users sender + ON sender.user_id = matched.sender_user_id AND sender.app_code = ?` + privateMessageArgs := []any{appCode, query.HostUserID, query.StartAtMS, query.EndAtMS, appCode} + if appCode != appcode.Default { + // 分成两个精确 app_code 区间，避免 MySQL 对 IN 条件选择覆盖索引全扫； + // UNION ALL 后在最终 COUNT(DISTINCT) 收敛同一发送者，历史/新记录重叠时也不会重复计数。 + privateMessageSQL = ` + SELECT COUNT(DISTINCT matched.sender_user_id) + FROM ( + SELECT sender_user_id + FROM user_private_message_events FORCE INDEX (idx_private_message_target_time) + WHERE app_code = ? AND target_user_id = ? + AND occurred_at_ms >= ? AND occurred_at_ms < ? + UNION ALL + SELECT sender_user_id + FROM user_private_message_events FORCE INDEX (idx_private_message_target_time) + WHERE app_code = ? AND target_user_id = ? + AND occurred_at_ms >= ? AND occurred_at_ms < ? + ) matched + INNER JOIN users sender + ON sender.user_id = matched.sender_user_id AND sender.app_code = ?` + privateMessageArgs = []any{ + appCode, query.HostUserID, query.StartAtMS, query.EndAtMS, + appcode.Default, query.HostUserID, query.StartAtMS, query.EndAtMS, + appCode, + } + } + if err := r.db.QueryRowContext(ctx, privateMessageSQL, privateMessageArgs...).Scan(&stats.PrivateMessageSenders); err != nil { return hostdomain.EngagementStats{}, err } @@ -109,13 +141,19 @@ func (r *Repository) hostOnlineDuration(ctx context.Context, query hostdomain.En return total, nil } -// RecordPrivateMessageEvent 用 app_code + event_id 幂等，不保存消息正文或扩展负载。 +// RecordPrivateMessageEvent 由 users 主数据反查租户，不信任公网回调缺省的 app_code。 +// users.user_id 是全局主键；两次主键查找同时确认发送者和接收者属于同一 App，避免 Fami 私信误落到默认租户。 func (r *Repository) RecordPrivateMessageEvent(ctx context.Context, event hostdomain.PrivateMessageEvent) (bool, error) { result, err := r.db.ExecContext(ctx, ` INSERT IGNORE INTO user_private_message_events ( app_code, event_id, sender_user_id, target_user_id, occurred_at_ms, created_at_ms - ) VALUES (?, ?, ?, ?, ?, ?)`, - appcode.FromContext(ctx), event.EventID, event.SenderUserID, event.TargetUserID, event.OccurredAtMS, time.Now().UTC().UnixMilli(), + ) + SELECT target.app_code, ?, sender.user_id, target.user_id, ?, ? + FROM users target + INNER JOIN users sender + ON sender.user_id = ? AND sender.app_code = target.app_code + WHERE target.user_id = ?`, + event.EventID, event.OccurredAtMS, time.Now().UTC().UnixMilli(), event.SenderUserID, event.TargetUserID, ) if err != nil { return false, err diff --git a/services/user-service/internal/storage/mysql/host/stats_test.go b/services/user-service/internal/storage/mysql/host/stats_test.go index bffa69e1..c98eb8b0 100644 --- a/services/user-service/internal/storage/mysql/host/stats_test.go +++ b/services/user-service/internal/storage/mysql/host/stats_test.go @@ -20,8 +20,26 @@ func TestGetHostEngagementStatsAggregatesOwnedFacts(t *testing.T) { mock.ExpectQuery("FROM user_mic_daily_stats"). WithArgs(validMicDayThresholdMS, "fami", int64(42), "2026-07-06", "2026-07-12"). WillReturnRows(sqlmock.NewRows([]string{"mic_ms", "valid_days"}).AddRow(int64(5400000), int64(1))) - mock.ExpectQuery("FROM user_private_message_events"). - WithArgs("fami", int64(42), int64(1783296000000), int64(1783900800000)). + mock.ExpectQuery(regexp.QuoteMeta(` + SELECT COUNT(DISTINCT matched.sender_user_id) + FROM ( + SELECT sender_user_id + FROM user_private_message_events FORCE INDEX (idx_private_message_target_time) + WHERE app_code = ? AND target_user_id = ? + AND occurred_at_ms >= ? AND occurred_at_ms < ? + UNION ALL + SELECT sender_user_id + FROM user_private_message_events FORCE INDEX (idx_private_message_target_time) + WHERE app_code = ? AND target_user_id = ? + AND occurred_at_ms >= ? AND occurred_at_ms < ? + ) matched + INNER JOIN users sender + ON sender.user_id = matched.sender_user_id AND sender.app_code = ?`)). + WithArgs( + "fami", int64(42), int64(1783296000000), int64(1783900800000), + "lalu", int64(42), int64(1783296000000), int64(1783900800000), + "fami", + ). WillReturnRows(sqlmock.NewRows([]string{"senders"}).AddRow(int64(6))) mock.ExpectQuery("FROM user_follow_events"). WithArgs("fami", int64(42), int64(1783296000000), int64(1783900800000)). @@ -53,3 +71,40 @@ func TestGetHostEngagementStatsAggregatesOwnedFacts(t *testing.T) { t.Fatalf("unmet SQL expectations: %v", err) } } + +func TestRecordPrivateMessageEventResolvesAppFromUsers(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("create sqlmock: %v", err) + } + defer db.Close() + + mock.ExpectExec(regexp.QuoteMeta(` + INSERT IGNORE INTO user_private_message_events ( + app_code, event_id, sender_user_id, target_user_id, occurred_at_ms, created_at_ms + ) + SELECT target.app_code, ?, sender.user_id, target.user_id, ?, ? + FROM users target + INNER JOIN users sender + ON sender.user_id = ? AND sender.app_code = target.app_code + WHERE target.user_id = ?`)). + WithArgs("im-message-1", int64(1783296000000), sqlmock.AnyArg(), int64(41), int64(42)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + repo := New(db) + created, err := repo.RecordPrivateMessageEvent(appcode.WithContext(context.Background(), "lalu"), hostdomain.PrivateMessageEvent{ + EventID: "im-message-1", + SenderUserID: 41, + TargetUserID: 42, + OccurredAtMS: 1783296000000, + }) + if err != nil { + t.Fatalf("RecordPrivateMessageEvent failed: %v", err) + } + if !created { + t.Fatal("event must be created for users in the same authoritative app") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet SQL expectations: %v", err) + } +} diff --git a/services/wallet-service/internal/domain/ledger/host_revenue_stats.go b/services/wallet-service/internal/domain/ledger/host_revenue_stats.go index 17885d06..cec9dfe2 100644 --- a/services/wallet-service/internal/domain/ledger/host_revenue_stats.go +++ b/services/wallet-service/internal/domain/ledger/host_revenue_stats.go @@ -1,6 +1,6 @@ package ledger -// HostRevenueStats 是主播中心按日期展示的 POINT 收益、已兑换数量和去重送礼人数。 +// HostRevenueStats 是主播中心按日期展示的周期钻石收益、POINT 已兑换数量和去重送礼人数。 type HostRevenueStats struct { DiamondEarnings int64 DiamondExchanged int64 diff --git a/services/wallet-service/internal/storage/mysql/agency_host_gift_stats.go b/services/wallet-service/internal/storage/mysql/agency_host_gift_stats.go index 82c6f49d..3e0a0525 100644 --- a/services/wallet-service/internal/storage/mysql/agency_host_gift_stats.go +++ b/services/wallet-service/internal/storage/mysql/agency_host_gift_stats.go @@ -10,8 +10,10 @@ import ( "hyapp/services/wallet-service/internal/domain/ledger" ) -// GetAgencyHostGiftStats 直接读取送礼交易里的 host_point_added 快照，只统计礼物收益； -// 任务奖励、提现释放和 Agency 分成虽然也进入 POINT 账户，但都不能混入 BD 的公会收礼口径。 +// GetAgencyHostGiftStats 只从送礼交易的不可变快照统计礼物收益；新政策写 host_point_added。 +// Fami 未发布收益政策的历史/过渡区域仍写 host_period_diamond_added，且它与新政策都是先按 +// 礼物类型倍率得到的主播实际收益，因此 Fami 可在政策切换日安全求和；其他 App 仍保持 POINT 口径。 +// 任务奖励、提现释放和 Agency 分成不在礼物交易快照内，不会混入公会收礼口径。 func (r *Repository) GetAgencyHostGiftStats(ctx context.Context, query ledger.AgencyHostGiftStatsQuery) (ledger.AgencyHostGiftStats, error) { if r == nil || r.db == nil { return ledger.AgencyHostGiftStats{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") @@ -19,6 +21,11 @@ func (r *Repository) GetAgencyHostGiftStats(ctx context.Context, query ledger.Ag if len(query.HostUserIDs) == 0 { return ledger.AgencyHostGiftStats{}, nil } + legacyIncomeSQL := "0" + if appcode.FromContext(ctx) == "fami" { + // 该片段是固定 SQL 常量，不包含用户输入；非 Fami 不解析旧字段，避免改变其他 App 的收益单位。 + legacyIncomeSQL = "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.host_period_diamond_added')), '0') AS SIGNED)" + } placeholders := make([]string, len(query.HostUserIDs)) args := make([]any, 0, len(query.HostUserIDs)+5) args = append(args, appcode.FromContext(ctx), bizTypeGiftDebit, bizTypeDirectGiftDebit, query.StartAtMS, query.EndAtMS) @@ -28,15 +35,19 @@ func (r *Repository) GetAgencyHostGiftStats(ctx context.Context, query ledger.Ag } querySQL := fmt.Sprintf(` SELECT - COALESCE(SUM(CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.host_point_added')), '0') AS SIGNED)), 0), + COALESCE(SUM( + CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.host_point_added')), '0') AS SIGNED) + + %s + ), 0), COUNT(DISTINCT CASE - WHEN CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.host_point_added')), '0') AS SIGNED) > 0 + WHEN CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.host_point_added')), '0') AS SIGNED) + + %s > 0 THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.target_user_id')) AS SIGNED) END) FROM wallet_transactions WHERE app_code = ? AND biz_type IN (?, ?) AND status = 'succeeded' AND created_at_ms >= ? AND created_at_ms < ? - AND CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.target_user_id')) AS SIGNED) IN (%s)`, strings.Join(placeholders, ",")) + AND CAST(JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.target_user_id')) AS SIGNED) IN (%s)`, legacyIncomeSQL, legacyIncomeSQL, strings.Join(placeholders, ",")) var stats ledger.AgencyHostGiftStats err := r.db.QueryRowContext(ctx, querySQL, args...).Scan(&stats.GiftIncome, &stats.GiftedHostCount) return stats, err diff --git a/services/wallet-service/internal/storage/mysql/host_revenue_stats.go b/services/wallet-service/internal/storage/mysql/host_revenue_stats.go index e4a0823d..21b05da9 100644 --- a/services/wallet-service/internal/storage/mysql/host_revenue_stats.go +++ b/services/wallet-service/internal/storage/mysql/host_revenue_stats.go @@ -2,30 +2,51 @@ package mysql import ( "context" + "time" "hyapp/pkg/appcode" "hyapp/pkg/xerr" "hyapp/services/wallet-service/internal/domain/ledger" ) -// GetHostRevenueStats 聚合 POINT 分录；兑换按 available 侧冻结扣减计数，审核结算不会再次累计。 +// GetHostRevenueStats 的“钻石收益/送礼人数”只读主播周期钻石流水； +// POINT 是可兑换资产，任务奖励、Agency 分成和提现退回都不能被误算成收礼钻石。 func (r *Repository) GetHostRevenueStats(ctx context.Context, query ledger.HostRevenueStatsQuery) (ledger.HostRevenueStats, error) { if r == nil || r.db == nil { return ledger.HostRevenueStats{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") } var stats ledger.HostRevenueStats - err := r.db.QueryRowContext(ctx, ` + startCycle := time.UnixMilli(query.StartAtMS).UTC().Format("2006-01") + endCycle := time.UnixMilli(query.EndAtMS - 1).UTC().Format("2006-01") + // (app_code,user_id,cycle_key,created_at_ms) 索引先限定单主播和最多 13 个周期， + // 再在已收敛的送礼流水上按时间求和/去重，不扫全表也不解析大 JSON 字段。 + if err := r.db.QueryRowContext(ctx, ` SELECT - COALESCE(SUM(CASE WHEN e.available_delta > 0 AND t.biz_type <> ? THEN e.available_delta ELSE 0 END), 0), - COALESCE(SUM(CASE WHEN e.available_delta < 0 THEN -e.available_delta ELSE 0 END), 0), - COUNT(DISTINCT CASE WHEN e.available_delta > 0 AND e.counterparty_user_id > 0 THEN e.counterparty_user_id END) - FROM wallet_entries e + COALESCE(SUM(diamond_delta), 0), + COUNT(DISTINCT CASE WHEN sender_user_id > 0 THEN sender_user_id END) + FROM host_period_diamond_entries FORCE INDEX (idx_host_period_diamond_entries_user_cycle) + WHERE app_code = ? AND user_id = ? AND cycle_key BETWEEN ? AND ? + AND created_at_ms >= ? AND created_at_ms < ?`, + appcode.FromContext(ctx), query.HostUserID, startCycle, endCycle, query.StartAtMS, query.EndAtMS, + ).Scan(&stats.DiamondEarnings, &stats.GiftSenders); err != nil { + return ledger.HostRevenueStats{}, err + } + + // “已兑换”保留现有 POINT 资产口径，但只计用户主动兑金币、转币商和平台提现首次冻结； + // 审核扣 frozen 、驳回释放和其他 POINT 变动不重复计入。 + if err := r.db.QueryRowContext(ctx, ` + SELECT COALESCE(SUM(-e.available_delta), 0) + FROM wallet_entries e FORCE INDEX (idx_wallet_entries_asset_user_time) INNER JOIN wallet_transactions t ON t.app_code = e.app_code AND t.transaction_id = e.transaction_id WHERE e.app_code = ? AND e.user_id = ? AND e.asset_type = ? - AND e.created_at_ms >= ? AND e.created_at_ms < ?`, - bizTypeSalaryWithdrawalRelease, + AND e.created_at_ms >= ? AND e.created_at_ms < ? + AND e.available_delta < 0 + AND t.biz_type IN (?, ?, ?)`, appcode.FromContext(ctx), query.HostUserID, ledger.AssetPoint, query.StartAtMS, query.EndAtMS, - ).Scan(&stats.DiamondEarnings, &stats.DiamondExchanged, &stats.GiftSenders) - return stats, err + bizTypePointExchangeToCoin, bizTypePointTransferToCoinSeller, bizTypeSalaryWithdrawalFreeze, + ).Scan(&stats.DiamondExchanged); err != nil { + return ledger.HostRevenueStats{}, err + } + return stats, nil } diff --git a/services/wallet-service/internal/storage/mysql/host_revenue_stats_test.go b/services/wallet-service/internal/storage/mysql/host_revenue_stats_test.go index 426dacfc..9b3f5cb0 100644 --- a/services/wallet-service/internal/storage/mysql/host_revenue_stats_test.go +++ b/services/wallet-service/internal/storage/mysql/host_revenue_stats_test.go @@ -10,7 +10,7 @@ import ( "hyapp/services/wallet-service/internal/domain/ledger" ) -func TestGetHostRevenueStatsAggregatesPointLedger(t *testing.T) { +func TestGetHostRevenueStatsAggregatesHostDiamondsAndPointExchanges(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("create sqlmock: %v", err) @@ -19,16 +19,25 @@ func TestGetHostRevenueStatsAggregatesPointLedger(t *testing.T) { mock.ExpectQuery(regexp.QuoteMeta(` SELECT - COALESCE(SUM(CASE WHEN e.available_delta > 0 AND t.biz_type <> ? THEN e.available_delta ELSE 0 END), 0), - COALESCE(SUM(CASE WHEN e.available_delta < 0 THEN -e.available_delta ELSE 0 END), 0), - COUNT(DISTINCT CASE WHEN e.available_delta > 0 AND e.counterparty_user_id > 0 THEN e.counterparty_user_id END) - FROM wallet_entries e + COALESCE(SUM(diamond_delta), 0), + COUNT(DISTINCT CASE WHEN sender_user_id > 0 THEN sender_user_id END) + FROM host_period_diamond_entries FORCE INDEX (idx_host_period_diamond_entries_user_cycle) + WHERE app_code = ? AND user_id = ? AND cycle_key BETWEEN ? AND ? + AND created_at_ms >= ? AND created_at_ms < ?`)). + WithArgs("fami", int64(42), "1970-01", "1970-01", int64(1000), int64(2000)). + WillReturnRows(sqlmock.NewRows([]string{"earnings", "senders"}).AddRow(int64(8800), int64(7))) + mock.ExpectQuery(regexp.QuoteMeta(` + SELECT COALESCE(SUM(-e.available_delta), 0) + FROM wallet_entries e FORCE INDEX (idx_wallet_entries_asset_user_time) INNER JOIN wallet_transactions t ON t.app_code = e.app_code AND t.transaction_id = e.transaction_id WHERE e.app_code = ? AND e.user_id = ? AND e.asset_type = ? - AND e.created_at_ms >= ? AND e.created_at_ms < ?`)). - WithArgs(bizTypeSalaryWithdrawalRelease, "fami", int64(42), ledger.AssetPoint, int64(1000), int64(2000)). - WillReturnRows(sqlmock.NewRows([]string{"earnings", "exchanged", "senders"}).AddRow(int64(8800), int64(1200), int64(7))) + AND e.created_at_ms >= ? AND e.created_at_ms < ? + AND e.available_delta < 0 + AND t.biz_type IN (?, ?, ?)`)). + WithArgs("fami", int64(42), ledger.AssetPoint, int64(1000), int64(2000), + bizTypePointExchangeToCoin, bizTypePointTransferToCoinSeller, bizTypeSalaryWithdrawalFreeze). + WillReturnRows(sqlmock.NewRows([]string{"exchanged"}).AddRow(int64(1200))) repo := &Repository{db: db} stats, err := repo.GetHostRevenueStats(appcode.WithContext(context.Background(), "fami"), ledger.HostRevenueStatsQuery{ diff --git a/services/wallet-service/internal/transport/grpc/host_revenue_stats.go b/services/wallet-service/internal/transport/grpc/host_revenue_stats.go index dcdd34e5..8adab713 100644 --- a/services/wallet-service/internal/transport/grpc/host_revenue_stats.go +++ b/services/wallet-service/internal/transport/grpc/host_revenue_stats.go @@ -9,7 +9,7 @@ import ( "hyapp/services/wallet-service/internal/domain/ledger" ) -// GetHostRevenueStats 返回钱包账本内的 POINT 收益、兑换和去重送礼用户。 +// GetHostRevenueStats 返回主播周期钻石收益、POINT 兑换和去重送礼用户。 func (s *Server) GetHostRevenueStats(ctx context.Context, req *walletv1.GetHostRevenueStatsRequest) (*walletv1.GetHostRevenueStatsResponse, error) { ctx = appcode.WithContext(ctx, req.GetAppCode()) stats, err := s.svc.GetHostRevenueStats(ctx, req.GetAppCode(), ledger.HostRevenueStatsQuery{