hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复数据以及上麦问题

Browse Source
This commit is contained in:
zhx 2026-07-04 22:02:28 +08:00
parent 7a2d292c36
commit e5cd2a5ce7
15 changed files with 918 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go
View File

@ -917,7 +917,7 @@ func (s *AslanMongoDashboardSource) loadCoinFlows(ctx context.Context, startDate
} }
// applyRetention 用注册 cohort × user_daily_active_log 推 D1/D7/D30：目标观察日未到时基数保持 0， // applyRetention 用注册 cohort × user_daily_active_log 推 D1/D7/D30：目标观察日未到时基数保持 0，
// 上层比例展示为 0%，与 cdc 源口径一致。 // 上层会把 0 基数转成 nil，避免把“还不能观察”展示成真实 0% 留存。
func (s *AslanMongoDashboardSource) applyRetention(ctx context.Context, cohorts map[string]map[string][]int64, grid *legacyMongoGrid) error { func (s *AslanMongoDashboardSource) applyRetention(ctx context.Context, cohorts map[string]map[string][]int64, grid *legacyMongoGrid) error {
location, err := time.LoadLocation(s.statTimezone) location, err := time.LoadLocation(s.statTimezone)
if err != nil { if err != nil {

50
server/admin/internal/modules/dashboard/external_dashboard.go
View File

@ -489,15 +489,15 @@ func (m externalDashboardMetric) toMap() map[string]any {
"consumed_coin": nullableInt64(m.ConsumedCoin), "consumed_coin": nullableInt64(m.ConsumedCoin),
"consume_output_ratio": consumeOutputRatioValue(m.ConsumedCoin, m.OutputCoin), "consume_output_ratio": consumeOutputRatioValue(m.ConsumedCoin, m.OutputCoin),
"daily_active_user": m.ActiveUsers, "daily_active_user": m.ActiveUsers,
"d1_retention_base_users": m.D1RetentionBaseUsers, "d1_retention_base_users": retentionMetricValue(m.D1RetentionBaseUsers, m.D1RetentionBaseUsers),
"d1_retention_rate": m.D1RetentionRate, "d1_retention_rate": retentionRateValue(m.D1RetentionUsers, m.D1RetentionBaseUsers),
"d1_retention_users": m.D1RetentionUsers, "d1_retention_users": retentionMetricValue(m.D1RetentionUsers, m.D1RetentionBaseUsers),
"d7_retention_base_users": m.D7RetentionBaseUsers, "d7_retention_base_users": retentionMetricValue(m.D7RetentionBaseUsers, m.D7RetentionBaseUsers),
"d7_retention_rate": m.D7RetentionRate, "d7_retention_rate": retentionRateValue(m.D7RetentionUsers, m.D7RetentionBaseUsers),
"d7_retention_users": m.D7RetentionUsers, "d7_retention_users": retentionMetricValue(m.D7RetentionUsers, m.D7RetentionBaseUsers),
"d30_retention_base_users": m.D30RetentionBaseUsers, "d30_retention_base_users": retentionMetricValue(m.D30RetentionBaseUsers, m.D30RetentionBaseUsers),
"d30_retention_rate": m.D30RetentionRate, "d30_retention_rate": retentionRateValue(m.D30RetentionUsers, m.D30RetentionBaseUsers),
"d30_retention_users": m.D30RetentionUsers, "d30_retention_users": retentionMetricValue(m.D30RetentionUsers, m.D30RetentionBaseUsers),
"game_payout": m.GamePayout, "game_payout": m.GamePayout,
"game_players": m.GamePlayers, "game_players": m.GamePlayers,
"game_profit": m.GameProfit, "game_profit": m.GameProfit,
@ -546,15 +546,15 @@ func (m externalDashboardMetric) toMap() map[string]any {
func (m externalDashboardMetric) retentionMap() map[string]any { func (m externalDashboardMetric) retentionMap() map[string]any {
return map[string]any{ return map[string]any{
"day1_base_users": m.D1RetentionBaseUsers, "day1_base_users": retentionMetricValue(m.D1RetentionBaseUsers, m.D1RetentionBaseUsers),
"day1_rate": m.D1RetentionRate, "day1_rate": retentionRateValue(m.D1RetentionUsers, m.D1RetentionBaseUsers),
"day1_users": m.D1RetentionUsers, "day1_users": retentionMetricValue(m.D1RetentionUsers, m.D1RetentionBaseUsers),
"day7_base_users": m.D7RetentionBaseUsers, "day7_base_users": retentionMetricValue(m.D7RetentionBaseUsers, m.D7RetentionBaseUsers),
"day7_rate": m.D7RetentionRate, "day7_rate": retentionRateValue(m.D7RetentionUsers, m.D7RetentionBaseUsers),
"day7_users": m.D7RetentionUsers, "day7_users": retentionMetricValue(m.D7RetentionUsers, m.D7RetentionBaseUsers),
"day30_base_users": m.D30RetentionBaseUsers, "day30_base_users": retentionMetricValue(m.D30RetentionBaseUsers, m.D30RetentionBaseUsers),
"day30_rate": m.D30RetentionRate, "day30_rate": retentionRateValue(m.D30RetentionUsers, m.D30RetentionBaseUsers),
"day30_users": m.D30RetentionUsers, "day30_users": retentionMetricValue(m.D30RetentionUsers, m.D30RetentionBaseUsers),
"registered_users": m.NewUsers, "registered_users": m.NewUsers,
} }
} }
@ -757,6 +757,20 @@ func ratioFloat64(numerator int64, denominator int64) float64 {
return float64(numerator) / float64(denominator) return float64(numerator) / float64(denominator)
} }
func retentionMetricValue(value int64, base int64) any {
if base <= 0 {
return nil
}
return value
}
func retentionRateValue(users int64, base int64) any {
if base <= 0 {
return nil
}
return ratioFloat64(users, base)
}
func deltaRate(current int64, previous int64) float64 { func deltaRate(current int64, previous int64) float64 {
if previous == 0 { if previous == 0 {
if current == 0 { if current == 0 {

22
server/admin/internal/modules/dashboard/external_dashboard_test.go
View File

@ -120,6 +120,28 @@ func TestMySQLExternalDashboardSourceStatisticsOverview(t *testing.T) {
} }
} }
func TestExternalDashboardRetentionUsesNilWhenBaseMissing(t *testing.T) {
metric := externalDashboardMetric{
D1RetentionUsers: 0,
D1RetentionBaseUsers: 0,
D7RetentionUsers: 0,
D7RetentionBaseUsers: 5,
D30RetentionUsers: 0,
D30RetentionBaseUsers: 0,
}
row := metric.toMap()
if row["d1_retention_rate"] != nil || row["d1_retention_users"] != nil || row["d1_retention_base_users"] != nil {
t.Fatalf("d1 retention without base should be nil, got %#v", row)
}
if row["d7_retention_rate"] != float64(0) || row["d7_retention_users"] != int64(0) || row["d7_retention_base_users"] != int64(5) {
t.Fatalf("d7 observed zero retention should stay visible, got %#v", row)
}
retention := metric.retentionMap()
if retention["day30_rate"] != nil || retention["day30_users"] != nil || retention["day30_base_users"] != nil {
t.Fatalf("d30 retention without base should be nil, got %#v", retention)
}
}
func TestMySQLExternalDashboardSourceFiltersLegacyRegionCountries(t *testing.T) { func TestMySQLExternalDashboardSourceFiltersLegacyRegionCountries(t *testing.T) {
db, mock, err := sqlmock.New() db, mock, err := sqlmock.New()
if err != nil { if err != nil {

24
server/admin/internal/modules/databi/service.go
View File

@ -1345,9 +1345,33 @@ func topLevelMetrics(overview map[string]any) map[string]any {
} }
out[key] = value out[key] = value
} }
flattenRetentionMetrics(out, overview["retention"])
return out return out
} }
func flattenRetentionMetrics(out map[string]any, retention any) {
retentionRow, ok := retention.(map[string]any)
if !ok {
return
}
for _, item := range []struct {
day string
rateKey string
usersKey string
baseKey string
}{
{day: "day1", rateKey: "d1_retention_rate", usersKey: "d1_retention_users", baseKey: "d1_retention_base_users"},
{day: "day7", rateKey: "d7_retention_rate", usersKey: "d7_retention_users", baseKey: "d7_retention_base_users"},
{day: "day30", rateKey: "d30_retention_rate", usersKey: "d30_retention_users", baseKey: "d30_retention_base_users"},
} {
// statistics-service 的历史契约把整段留存放在 retention 子对象里；
// Databi v2 的 App 行统一读取扁平字段，这里只做字段搬运，不重算口径。
out[item.rateKey] = retentionRow[item.day+"_rate"]
out[item.usersKey] = retentionRow[item.day+"_users"]
out[item.baseKey] = retentionRow[item.day+"_base_users"]
}
}
func scopeInfos(scopes []model.UserMoneyScope) []ScopeInfo { func scopeInfos(scopes []model.UserMoneyScope) []ScopeInfo {
out := make([]ScopeInfo, 0, len(scopes)) out := make([]ScopeInfo, 0, len(scopes))
for _, scope := range scopes { for _, scope := range scopes {

9
server/admin/internal/modules/databi/service_test.go
View File

@ -42,12 +42,18 @@ func TestRegionDisplay(t *testing.T) {
} }
func TestTopLevelMetricsStripsStructuralKeys(t *testing.T) { func TestTopLevelMetricsStripsStructuralKeys(t *testing.T) {
day1Rate := 0.5
out := topLevelMetrics(map[string]any{ out := topLevelMetrics(map[string]any{
"recharge_usd_minor": int64(5), "recharge_usd_minor": int64(5),
"daily_series": []any{"x"}, "daily_series": []any{"x"},
"country_breakdown": []any{"y"}, "country_breakdown": []any{"y"},
"daily_country_breakdown": []any{"z"}, "daily_country_breakdown": []any{"z"},
"report_metric_sources": []any{}, "report_metric_sources": []any{},
"retention": map[string]any{
"day1_base_users": int64(6),
"day1_rate": day1Rate,
"day1_users": int64(3),
},
}) })
if _, ok := out["daily_series"]; ok { if _, ok := out["daily_series"]; ok {
t.Fatalf("expected daily_series stripped") t.Fatalf("expected daily_series stripped")
@ -55,6 +61,9 @@ func TestTopLevelMetricsStripsStructuralKeys(t *testing.T) {
if out["recharge_usd_minor"] != int64(5) { if out["recharge_usd_minor"] != int64(5) {
t.Fatalf("expected metric kept, got %v", out["recharge_usd_minor"]) t.Fatalf("expected metric kept, got %v", out["recharge_usd_minor"])
} }
if out["d1_retention_rate"] != day1Rate || out["d1_retention_users"] != int64(3) || out["d1_retention_base_users"] != int64(6) {
t.Fatalf("expected nested retention flattened, got %#v", out)
}
} }
// legacy 雪花区域 ID 超出 JS 安全整数，历史 scope 里存的是 float64 圆整值； // legacy 雪花区域 ID 超出 JS 安全整数，历史 scope 里存的是 float64 圆整值；

7
services/room-service/internal/room/command/command.go
View File

@ -248,6 +248,11 @@ type ConfirmMicPublishing struct {
RoomVersion int64 `json:"room_version"` RoomVersion int64 `json:"room_version"`
// EventTimeMS 是客户端 SDK 或 RTC webhook 的事件时间，必须晚于已接受事件。 // EventTimeMS 是客户端 SDK 或 RTC webhook 的事件时间，必须晚于已接受事件。
EventTimeMS int64 `json:"event_time_ms"` EventTimeMS int64 `json:"event_time_ms"`
// AcceptedEventTimeMS 是 room-service 接受本次客户端确认时使用的服务端时间。
//
// 客户端 event_time_ms 可能受设备时钟漂移影响，只能保留为原始观测时间；恢复回放和后续媒体事件
// 排序必须使用已经被服务端 deadline 校验过的时间，避免重启后把偏快客户端时间写回状态。
AcceptedEventTimeMS int64 `json:"accepted_event_time_ms,omitempty"`
// Source 记录确认来源，例如 client 或 rtc_webhook。 // Source 记录确认来源，例如 client 或 rtc_webhook。
Source string `json:"source,omitempty"` Source string `json:"source,omitempty"`
// TargetDisplayProfile 是发流确认目标用户展示快照，只用于 IM/UI 兜底。 // TargetDisplayProfile 是发流确认目标用户展示快照，只用于 IM/UI 兜底。
@ -686,6 +691,8 @@ func IdempotencyPayload(payload []byte) ([]byte, error) {
delete(values, "password_hash") delete(values, "password_hash")
// MicUp 的发流确认 deadline 由 room-service 生成，客户端重试同一 command_id 时不能因此冲突。 // MicUp 的发流确认 deadline 由 room-service 生成，客户端重试同一 command_id 时不能因此冲突。
delete(values, "publish_deadline_ms") delete(values, "publish_deadline_ms")
// ConfirmMicPublishing 的接受时间由 room-service 生成，不能参与客户端请求幂等语义。
delete(values, "accepted_event_time_ms")
return json.Marshal(values) return json.Marshal(values)
} }

62
services/room-service/internal/room/service/mic.go
View File

@ -379,10 +379,8 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM
// RTC webhook 可能晚于用户下麦到达；旧事件只忽略，不返回错误清掉新状态。 // RTC webhook 可能晚于用户下麦到达；旧事件只忽略，不返回错误清掉新状态。
return mutationResult{snapshot: current.ToProto()}, nil, nil return mutationResult{snapshot: current.ToProto()}, nil, nil
} }
if seat.MicSessionID != cmd.MicSessionID || receivedAtMS := micPublishConfirmReceivedAtMS(seat, cmd, s.micPublishTimeout, now)
cmd.RoomVersion < seat.MicSessionRoomVersion || if !micPublishConfirmationMatchesCurrentSession(seat, cmd, receivedAtMS) {
cmd.EventTimeMS <= seat.LastPublishEventTimeMS ||
cmd.EventTimeMS > seat.PublishDeadlineMS {
return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil
} }
if seat.PublishState == state.MicPublishPublishing { if seat.PublishState == state.MicPublishPublishing {
@ -392,11 +390,18 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM
return mutationResult{}, nil, xerr.New(xerr.Conflict, "mic is not waiting for publish confirmation") return mutationResult{}, nil, xerr.New(xerr.Conflict, "mic is not waiting for publish confirmation")
} }
acceptedCmd := cmd
acceptedCmd.AcceptedEventTimeMS = receivedAtMS
commandPayload, err := command.Serialize(acceptedCmd)
if err != nil {
return mutationResult{}, nil, err
}
index := current.SeatIndex(seat.SeatNo) index := current.SeatIndex(seat.SeatNo)
current.MicSeats[index].PublishState = state.MicPublishPublishing current.MicSeats[index].PublishState = state.MicPublishPublishing
current.MicSeats[index].LastPublishEventTimeMS = cmd.EventTimeMS current.MicSeats[index].LastPublishEventTimeMS = receivedAtMS
// 确认发流本身代表当前会话刚被服务端接受，先初始化心跳时间，后续 MicHeartbeat 只负责刷新这个独立字段。 // 确认发流本身代表当前会话刚被服务端接受，心跳起点使用同一个 accepted time，保证 command log 回放一致。
current.MicSeats[index].MicHeartbeatAtMS = now.UnixMilli() current.MicSeats[index].MicHeartbeatAtMS = receivedAtMS
if micPublishConfirmSourceIsMuted(cmd.Source) { if micPublishConfirmSourceIsMuted(cmd.Source) {
current.MicSeats[index].MicMuted = true current.MicSeats[index].MicMuted = true
} }
@ -413,7 +418,7 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM
MicSessionId: cmd.MicSessionID, MicSessionId: cmd.MicSessionID,
PublishState: state.MicPublishPublishing, PublishState: state.MicPublishPublishing,
PublishDeadlineMs: seat.PublishDeadlineMS, PublishDeadlineMs: seat.PublishDeadlineMS,
PublishEventTimeMs: cmd.EventTimeMS, PublishEventTimeMs: receivedAtMS,
MicMuted: current.MicSeats[index].MicMuted, MicMuted: current.MicSeats[index].MicMuted,
SeatStatus: seatStatus, SeatStatus: seatStatus,
TargetGiftValue: targetGiftValue, TargetGiftValue: targetGiftValue,
@ -433,6 +438,7 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM
micSessionID: cmd.MicSessionID, micSessionID: cmd.MicSessionID,
publishDeadlineMS: seat.PublishDeadlineMS, publishDeadlineMS: seat.PublishDeadlineMS,
micHeartbeatAtMS: current.MicSeats[index].MicHeartbeatAtMS, micHeartbeatAtMS: current.MicSeats[index].MicHeartbeatAtMS,
commandPayload: commandPayload,
syncEvent: &tencentim.RoomEvent{ syncEvent: &tencentim.RoomEvent{
EventID: micEvent.EventID, EventID: micEvent.EventID,
RoomID: current.RoomID, RoomID: current.RoomID,
@ -456,6 +462,46 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM
}, nil }, nil
} }
func micPublishConfirmationMatchesCurrentSession(seat state.MicSeat, cmd command.ConfirmMicPublishing, arrivedAtMS int64) bool {
if seat.MicSessionID != cmd.MicSessionID || cmd.RoomVersion < seat.MicSessionRoomVersion {
// session_id 和创建 session 的房间版本共同防止快速下麦/重上麦后旧确认误推进新会话。
return false
}
if cmd.EventTimeMS <= seat.LastPublishEventTimeMS {
// event_time_ms 只负责同一会话内的事件顺序，不能拿它判断客户端请求是否真的超时。
return false
}
if seat.PublishDeadlineMS <= 0 || arrivedAtMS >= seat.PublishDeadlineMS {
// deadline 是服务端创建的等待窗口：客户端时钟可能快于服务端，只有服务端到达时间越界才算真实超时。
return false
}
return true
}
func micPublishConfirmReceivedAtMS(seat state.MicSeat, cmd command.ConfirmMicPublishing, timeout time.Duration, now time.Time) int64 {
receivedAtMS := cmd.SentAtMS
if receivedAtMS <= 0 {
return now.UnixMilli()
}
if createdAtMS := rtcMicSessionCreatedAtMS(seat, timeout); createdAtMS > 0 && receivedAtMS < createdAtMS {
// sent_at_ms 正常来自 gateway 服务端时间；若测试夹具或异常入口给了早于会话创建的时间，
// 不能把它当成本次确认时间，否则恢复后 LastPublishEventTimeMS 会落到会话创建前。
return now.UnixMilli()
}
return receivedAtMS
}
func confirmMicPublishingAcceptedEventTimeMS(cmd *command.ConfirmMicPublishing) int64 {
if cmd == nil {
return 0
}
if cmd.AcceptedEventTimeMS > 0 {
return cmd.AcceptedEventTimeMS
}
return cmd.EventTimeMS
}
// MicHeartbeat 刷新当前 publishing 麦位会话的服务端心跳时间。 // MicHeartbeat 刷新当前 publishing 麦位会话的服务端心跳时间。
func (s *Service) MicHeartbeat(ctx context.Context, req *roomv1.MicHeartbeatRequest) (*roomv1.MicHeartbeatResponse, error) { func (s *Service) MicHeartbeat(ctx context.Context, req *roomv1.MicHeartbeatRequest) (*roomv1.MicHeartbeatResponse, error) {
ctx = contextFromMeta(ctx, req.GetMeta()) ctx = contextFromMeta(ctx, req.GetMeta())

272
services/room-service/internal/room/service/mic_test.go
View File

@ -2,10 +2,13 @@ package service_test
import ( import (
"context" "context"
"fmt"
"testing" "testing"
"time" "time"
roomv1 "hyapp.local/api/proto/room/v1" roomv1 "hyapp.local/api/proto/room/v1"
"hyapp/pkg/appcode"
"hyapp/services/room-service/internal/room/command"
"hyapp/services/room-service/internal/testutil/mysqltest" "hyapp/services/room-service/internal/testutil/mysqltest"
) )
@ -54,6 +57,161 @@ func TestMicUpReturnsExistingSeatForDuplicateUserCommand(t *testing.T) {
} }
} }
func TestConfirmMicPublishingAcceptsFastClientClockBeforeServerDeadline(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
now := &fixedRoomRocketClock{now: time.Date(2026, 6, 4, 8, 0, 0, 0, time.UTC)}
svc := newRocketTestService(t, repository, &rocketTestWallet{}, now)
roomID := "room-mic-confirm-fast-client-clock"
ownerID := int64(8551)
speakerID := int64(8552)
createRocketRoom(t, ctx, svc, roomID, ownerID, 9101)
joinRocketRoom(t, ctx, svc, roomID, speakerID)
upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-fast-clock-up"),
SeatNo: 2,
})
if err != nil {
t.Fatalf("mic up failed: %v", err)
}
arrivedAtMS := upResp.GetPublishDeadlineMs() - 1
fastClientEventMS := upResp.GetPublishDeadlineMs() + int64((5 * time.Second).Milliseconds())
now.now = time.UnixMilli(arrivedAtMS)
confirmMeta := rocketMeta(roomID, speakerID, "mic-confirm-fast-clock-confirm")
confirmResp, err := svc.ConfirmMicPublishing(ctx, &roomv1.ConfirmMicPublishingRequest{
Meta: confirmMeta,
MicSessionId: upResp.GetMicSessionId(),
RoomVersion: upResp.GetRoom().GetVersion(),
EventTimeMs: fastClientEventMS,
Source: "client",
})
if err != nil {
t.Fatalf("confirm mic publishing failed: %v", err)
}
if !confirmResp.GetResult().GetApplied() {
t.Fatalf("fast client event_time_ms must not be treated as timeout before server deadline: %+v", confirmResp.GetResult())
}
seat := seatByNo(confirmResp.GetRoom(), 2)
if seat == nil || seat.GetPublishState() != "publishing" {
t.Fatalf("seat must be confirmed as publishing: %+v", seat)
}
if seat.GetLastPublishEventTimeMs() != arrivedAtMS || seat.GetMicHeartbeatAtMs() != arrivedAtMS {
t.Fatalf("confirmation must store server accepted time, not fast client event time: seat=%+v accepted=%d client_event=%d", seat, arrivedAtMS, fastClientEventMS)
}
record, exists, err := repository.GetCommand(appcode.WithContext(ctx, appcode.Default), roomID, confirmMeta.GetCommandId())
if err != nil {
t.Fatalf("get confirm command log failed: %v", err)
}
if !exists {
t.Fatalf("confirm command log must be persisted")
}
decoded, err := command.Deserialize(command.ConfirmMicPublishing{}.Type(), record.Payload)
if err != nil {
t.Fatalf("decode confirm command log failed: %v", err)
}
confirmCmd, ok := decoded.(*command.ConfirmMicPublishing)
if !ok {
t.Fatalf("confirm command log has unexpected type: %T", decoded)
}
if confirmCmd.EventTimeMS != fastClientEventMS || confirmCmd.AcceptedEventTimeMS != arrivedAtMS {
t.Fatalf("confirm command log must keep raw client time and accepted server time: %+v raw=%d accepted=%d", confirmCmd, fastClientEventMS, arrivedAtMS)
}
}
func TestConfirmMicPublishingIgnoresStaleSessionVersionAndTimeout(t *testing.T) {
tests := []struct {
name string
adjust func(clock *fixedRoomRocketClock, upResp *roomv1.MicUpResponse)
request func(roomID string, speakerID int64, clock *fixedRoomRocketClock, upResp *roomv1.MicUpResponse) *roomv1.ConfirmMicPublishingRequest
}{
{
name: "old_session",
adjust: func(clock *fixedRoomRocketClock, _ *roomv1.MicUpResponse) {
clock.now = clock.now.Add(500 * time.Millisecond)
},
request: func(roomID string, speakerID int64, clock *fixedRoomRocketClock, upResp *roomv1.MicUpResponse) *roomv1.ConfirmMicPublishingRequest {
return &roomv1.ConfirmMicPublishingRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-old-session"),
MicSessionId: upResp.GetMicSessionId() + "-old",
RoomVersion: upResp.GetRoom().GetVersion(),
EventTimeMs: clock.Now().UnixMilli(),
Source: "client",
}
},
},
{
name: "old_room_version",
adjust: func(clock *fixedRoomRocketClock, _ *roomv1.MicUpResponse) {
clock.now = clock.now.Add(500 * time.Millisecond)
},
request: func(roomID string, speakerID int64, clock *fixedRoomRocketClock, upResp *roomv1.MicUpResponse) *roomv1.ConfirmMicPublishingRequest {
return &roomv1.ConfirmMicPublishingRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-old-room-version"),
MicSessionId: upResp.GetMicSessionId(),
RoomVersion: upResp.GetRoom().GetVersion() - 1,
EventTimeMs: clock.Now().UnixMilli(),
Source: "client",
}
},
},
{
name: "deadline_reached",
adjust: func(clock *fixedRoomRocketClock, upResp *roomv1.MicUpResponse) {
clock.now = time.UnixMilli(upResp.GetPublishDeadlineMs())
},
request: func(roomID string, speakerID int64, _ *fixedRoomRocketClock, upResp *roomv1.MicUpResponse) *roomv1.ConfirmMicPublishingRequest {
return &roomv1.ConfirmMicPublishingRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-deadline-reached"),
MicSessionId: upResp.GetMicSessionId(),
RoomVersion: upResp.GetRoom().GetVersion(),
EventTimeMs: upResp.GetPublishDeadlineMs() - 1,
Source: "client",
}
},
},
}
for index, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
now := &fixedRoomRocketClock{now: time.Date(2026, 6, 4, 8, 0, 0, 0, time.UTC)}
svc := newRocketTestService(t, repository, &rocketTestWallet{}, now)
roomID := fmt.Sprintf("room-mic-confirm-guard-%d", index)
ownerID := int64(8561 + index*10)
speakerID := int64(8562 + index*10)
createRocketRoom(t, ctx, svc, roomID, ownerID, 9101)
joinRocketRoom(t, ctx, svc, roomID, speakerID)
upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{
Meta: rocketMeta(roomID, speakerID, fmt.Sprintf("mic-confirm-guard-up-%d", index)),
SeatNo: 2,
})
if err != nil {
t.Fatalf("mic up failed: %v", err)
}
tt.adjust(now, upResp)
confirmResp, err := svc.ConfirmMicPublishing(ctx, tt.request(roomID, speakerID, now, upResp))
if err != nil {
t.Fatalf("stale confirm must be ignored without error: %v", err)
}
if confirmResp.GetResult().GetApplied() || confirmResp.GetRoom().GetVersion() != upResp.GetRoom().GetVersion() {
t.Fatalf("stale confirm must not advance room state: confirm=%+v up_version=%d", confirmResp.GetResult(), upResp.GetRoom().GetVersion())
}
seat := seatByNo(confirmResp.GetRoom(), 2)
if seat == nil || seat.GetPublishState() != "pending_publish" || seat.GetLastPublishEventTimeMs() != 0 || seat.GetMicHeartbeatAtMs() != 0 {
t.Fatalf("stale confirm must leave pending session untouched: %+v", seat)
}
})
}
}
func TestMicHeartbeatRefreshesPublishingSessionAndPresence(t *testing.T) { func TestMicHeartbeatRefreshesPublishingSessionAndPresence(t *testing.T) {
ctx := context.Background() ctx := context.Background()
repository := mysqltest.NewRepository(t) repository := mysqltest.NewRepository(t)
@ -111,6 +269,120 @@ func TestMicHeartbeatRefreshesPublishingSessionAndPresence(t *testing.T) {
} }
} }
func TestConfirmMicPublishingIgnoresBackwardEventTime(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
now := &fixedRoomRocketClock{now: time.Date(2026, 6, 4, 8, 0, 0, 0, time.UTC)}
svc := newRocketTestService(t, repository, &rocketTestWallet{}, now)
roomID := "room-mic-confirm-backward-event"
ownerID := int64(8651)
speakerID := int64(8652)
createRocketRoom(t, ctx, svc, roomID, ownerID, 9101)
joinRocketRoom(t, ctx, svc, roomID, speakerID)
upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-backward-up"),
SeatNo: 2,
})
if err != nil {
t.Fatalf("mic up failed: %v", err)
}
now.now = now.now.Add(time.Second)
acceptedEventMS := now.Now().UnixMilli()
confirmResp, err := svc.ConfirmMicPublishing(ctx, &roomv1.ConfirmMicPublishingRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-backward-first"),
MicSessionId: upResp.GetMicSessionId(),
RoomVersion: upResp.GetRoom().GetVersion(),
EventTimeMs: acceptedEventMS,
Source: "client",
})
if err != nil {
t.Fatalf("first confirm failed: %v", err)
}
if !confirmResp.GetResult().GetApplied() {
t.Fatalf("first confirm must apply: %+v", confirmResp.GetResult())
}
now.now = now.now.Add(time.Second)
backwardResp, err := svc.ConfirmMicPublishing(ctx, &roomv1.ConfirmMicPublishingRequest{
Meta: rocketMeta(roomID, speakerID, "mic-confirm-backward-retry"),
MicSessionId: upResp.GetMicSessionId(),
RoomVersion: confirmResp.GetRoom().GetVersion(),
EventTimeMs: acceptedEventMS - 1,
Source: "client",
})
if err != nil {
t.Fatalf("backward confirm must be ignored without error: %v", err)
}
if backwardResp.GetResult().GetApplied() || backwardResp.GetRoom().GetVersion() != confirmResp.GetRoom().GetVersion() {
t.Fatalf("backward event must not advance room state: backward=%+v confirmed_version=%d", backwardResp.GetResult(), confirmResp.GetRoom().GetVersion())
}
seat := seatByNo(backwardResp.GetRoom(), 2)
if seat == nil || seat.GetPublishState() != "publishing" || seat.GetLastPublishEventTimeMs() != acceptedEventMS {
t.Fatalf("backward event must not lower accepted event time: %+v want_event=%d", seat, acceptedEventMS)
}
}
func TestRTCAudioStoppedReleasesAfterFastClientConfirmWatermark(t *testing.T) {
ctx := context.Background()
repository := mysqltest.NewRepository(t)
now := &fixedRoomRocketClock{now: time.Date(2026, 6, 4, 8, 0, 0, 0, time.UTC)}
svc := newRocketTestService(t, repository, &rocketTestWallet{}, now)
roomID := "room-rtc-stop-after-fast-confirm"
ownerID := int64(8661)
speakerID := int64(8662)
createRocketRoom(t, ctx, svc, roomID, ownerID, 9101)
joinRocketRoom(t, ctx, svc, roomID, speakerID)
upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{
Meta: rocketMeta(roomID, speakerID, "rtc-stop-after-fast-confirm-up"),
SeatNo: 2,
})
if err != nil {
t.Fatalf("mic up failed: %v", err)
}
createdAtMS := upResp.GetPublishDeadlineMs() - int64((15 * time.Second).Milliseconds())
stopEventMS := createdAtMS + int64((2 * time.Second).Milliseconds())
acceptedAtMS := upResp.GetPublishDeadlineMs() - 1
now.now = time.UnixMilli(acceptedAtMS)
confirmResp, err := svc.ConfirmMicPublishing(ctx, &roomv1.ConfirmMicPublishingRequest{
Meta: rocketMeta(roomID, speakerID, "rtc-stop-after-fast-confirm-confirm"),
MicSessionId: upResp.GetMicSessionId(),
RoomVersion: upResp.GetRoom().GetVersion(),
EventTimeMs: upResp.GetPublishDeadlineMs() + 5000,
Source: "client",
})
if err != nil {
t.Fatalf("confirm mic publishing failed: %v", err)
}
confirmedSeat := seatByNo(confirmResp.GetRoom(), 2)
if confirmedSeat == nil || confirmedSeat.GetLastPublishEventTimeMs() != acceptedAtMS {
t.Fatalf("confirm must store accepted watermark before RTC stop test: %+v accepted=%d", confirmedSeat, acceptedAtMS)
}
now.now = now.now.Add(time.Second)
stopResp, err := svc.ApplyRTCEvent(ctx, &roomv1.ApplyRTCEventRequest{
Meta: rocketMeta(roomID, speakerID, "rtc-stop-after-fast-confirm-stop"),
TargetUserId: speakerID,
EventType: "audio_stopped",
EventTimeMs: stopEventMS,
Reason: "rtc_audio_stopped:0",
Source: "tencent_rtc_callback",
})
if err != nil {
t.Fatalf("rtc audio stopped failed: %v", err)
}
if !stopResp.GetResult().GetApplied() {
t.Fatalf("rtc stop must release current mic session even when provider event time is below client confirm accepted time: %+v", stopResp.GetResult())
}
seat := seatByNo(stopResp.GetRoom(), 2)
if seat == nil || seat.GetUserId() != 0 || seat.GetMicSessionId() != "" {
t.Fatalf("rtc stop must clear current mic seat: %+v", seat)
}
}
func TestMicHeartbeatRejectsPendingSession(t *testing.T) { func TestMicHeartbeatRejectsPendingSession(t *testing.T) {
ctx := context.Background() ctx := context.Background()
repository := mysqltest.NewRepository(t) repository := mysqltest.NewRepository(t)

5
services/room-service/internal/room/service/recovery.go
View File

@ -251,9 +251,10 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64)
return fmt.Errorf("confirm_mic_publishing replay target is not on seat: %d", typed.TargetUserID) return fmt.Errorf("confirm_mic_publishing replay target is not on seat: %d", typed.TargetUserID)
} }
index := current.SeatIndex(seat.SeatNo) index := current.SeatIndex(seat.SeatNo)
acceptedEventTimeMS := confirmMicPublishingAcceptedEventTimeMS(typed)
current.MicSeats[index].PublishState = state.MicPublishPublishing current.MicSeats[index].PublishState = state.MicPublishPublishing
current.MicSeats[index].LastPublishEventTimeMS = typed.EventTimeMS current.MicSeats[index].LastPublishEventTimeMS = acceptedEventTimeMS
current.MicSeats[index].MicHeartbeatAtMS = typed.SentAtMS current.MicSeats[index].MicHeartbeatAtMS = acceptedEventTimeMS
current.Version++ current.Version++
case *command.MicHeartbeat: case *command.MicHeartbeat:
seat, exists := current.SeatByUser(typed.TargetUserID) seat, exists := current.SeatByUser(typed.TargetUserID)

8
services/room-service/internal/room/service/rtc_event.go
View File

@ -236,6 +236,10 @@ func rtcEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, tim
if !rtcStopEventMatchesCurrentMicSession(seat, eventTimeMS, timeout) { if !rtcStopEventMatchesCurrentMicSession(seat, eventTimeMS, timeout) {
return false return false
} }
if eventTimeMS <= seat.LastPublishEventTimeMS {
// audio_started 必须晚于已接受发布水位，防止腾讯重试或乱序开始事件反复推进状态。
return false
}
if seat.PublishDeadlineMS > 0 && eventTimeMS > seat.PublishDeadlineMS { if seat.PublishDeadlineMS > 0 && eventTimeMS > seat.PublishDeadlineMS {
// audio_started 过了确认 deadline 不再接受，超时 worker 会按 mic_session_id 释放麦位。 // audio_started 过了确认 deadline 不再接受，超时 worker 会按 mic_session_id 释放麦位。
return false return false
@ -246,8 +250,8 @@ func rtcEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, tim
// rtcStopEventMatchesCurrentMicSession 校验停止/退房事件是否能作用于当前 mic_session。 // rtcStopEventMatchesCurrentMicSession 校验停止/退房事件是否能作用于当前 mic_session。
func rtcStopEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, timeout time.Duration) bool { func rtcStopEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, timeout time.Duration) bool {
if seat.MicSessionID == "" || eventTimeMS <= seat.LastPublishEventTimeMS { if seat.MicSessionID == "" {
// 没有会话或事件时间不递增时，事件不能改变当前麦位。 // 没有会话时，停止/退房事件只能是旧回调。
return false return false
} }
if createdAtMS := rtcMicSessionCreatedAtMS(seat, timeout); createdAtMS > 0 && eventTimeMS < createdAtMS { if createdAtMS := rtcMicSessionCreatedAtMS(seat, timeout); createdAtMS > 0 && eventTimeMS < createdAtMS {

4
services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql
View File

@ -109,7 +109,9 @@ CREATE TABLE IF NOT EXISTS stat_user_registration (
registered_at_ms BIGINT NOT NULL, registered_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_tz, user_id), PRIMARY KEY (app_code, stat_tz, user_id),
KEY idx_stat_user_registration_day (app_code, stat_tz, registered_day, country_id), KEY idx_stat_user_registration_day (app_code, stat_tz, registered_day, country_id),
KEY idx_stat_user_registration_region (app_code, stat_tz, registered_day, region_id) KEY idx_stat_user_registration_region (app_code, stat_tz, registered_day, region_id),
KEY idx_stat_user_registration_country_day (app_code, stat_tz, country_id, registered_day, user_id),
KEY idx_stat_user_registration_region_day (app_code, stat_tz, region_id, registered_day, country_id, user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户注册日留存 cohort 表'; ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='用户注册日留存 cohort 表';
CREATE TABLE IF NOT EXISTS stat_user_dimension ( CREATE TABLE IF NOT EXISTS stat_user_dimension (

47
services/statistics-service/deploy/mysql/migrations/001_retention_registration_filter_indexes.sql Normal file
View File

@ -0,0 +1,47 @@
SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci;
-- Databi 留存查询会按注册日 cohort 读取 stat_user_registration，并可叠加国家或区域筛选。
-- 这两个索引只用于老库受控补齐：请在低峰执行，并在执行后对留存 SQL 做 EXPLAIN。
-- 不放进服务启动自动迁移，避免大表普通 ALTER 在生产启动时造成元数据锁或 IO 峰值。
SET @add_retention_country_day := (
SELECT IF(
COUNT(*) = 0,
'ALTER TABLE stat_user_registration ADD INDEX idx_stat_user_registration_country_day (app_code, stat_tz, country_id, registered_day, user_id), ALGORITHM=INPLACE, LOCK=NONE',
'SELECT 1'
)
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = DATABASE()
AND TABLE_NAME = 'stat_user_registration'
AND INDEX_NAME = 'idx_stat_user_registration_country_day'
);
PREPARE stmt FROM @add_retention_country_day;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SET @add_retention_region_day := (
SELECT IF(
COUNT(*) = 0,
'ALTER TABLE stat_user_registration ADD INDEX idx_stat_user_registration_region_day (app_code, stat_tz, region_id, registered_day, country_id, user_id), ALGORITHM=INPLACE, LOCK=NONE',
'SELECT 1'
)
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = DATABASE()
AND TABLE_NAME = 'stat_user_registration'
AND INDEX_NAME = 'idx_stat_user_registration_region_day'
);
PREPARE stmt FROM @add_retention_region_day;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;
SELECT
COUNT(*) AS retention_registration_filter_index_count
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = DATABASE()
AND TABLE_NAME = 'stat_user_registration'
AND INDEX_NAME IN (
'idx_stat_user_registration_country_day',
'idx_stat_user_registration_region_day'
);

414
services/statistics-service/internal/storage/mysql/query.go
View File

@ -111,59 +111,71 @@ type Overview struct {
} }
type DailySeriesItem struct { type DailySeriesItem struct {
StatDay string `json:"stat_day"` StatDay string `json:"stat_day"`
Label string `json:"label"` Label string `json:"label"`
NewUsers int64 `json:"new_users"` NewUsers int64 `json:"new_users"`
RegisteredUsers int64 `json:"registered_users"` RegisteredUsers int64 `json:"registered_users"`
Registrations int64 `json:"registrations"` Registrations int64 `json:"registrations"`
ActiveUsers int64 `json:"active_users"` ActiveUsers int64 `json:"active_users"`
PaidUsers int64 `json:"paid_users"` PaidUsers int64 `json:"paid_users"`
NewPaidUsers int64 `json:"new_paid_users"` NewPaidUsers int64 `json:"new_paid_users"`
RechargeUsers int64 `json:"recharge_users"` RechargeUsers int64 `json:"recharge_users"`
RechargeUSDMinor int64 `json:"recharge_usd_minor"` RechargeUSDMinor int64 `json:"recharge_usd_minor"`
NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"`
CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"`
CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"` CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"`
MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"`
GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"`
CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"`
CoinTotal int64 `json:"coin_total"` CoinTotal int64 `json:"coin_total"`
ConsumedCoin int64 `json:"consumed_coin"` ConsumedCoin int64 `json:"consumed_coin"`
OutputCoin int64 `json:"output_coin"` OutputCoin int64 `json:"output_coin"`
ConsumeOutputRatio float64 `json:"consume_output_ratio"` ConsumeOutputRatio float64 `json:"consume_output_ratio"`
ConsumeOutputDelta int64 `json:"consume_output_delta"` ConsumeOutputDelta int64 `json:"consume_output_delta"`
PlatformGrantCoin int64 `json:"platform_grant_coin"` PlatformGrantCoin int64 `json:"platform_grant_coin"`
ManualGrantCoin int64 `json:"manual_grant_coin"` ManualGrantCoin int64 `json:"manual_grant_coin"`
SalaryUSDMinor int64 `json:"salary_usd_minor"` SalaryUSDMinor int64 `json:"salary_usd_minor"`
SalaryTransferCoin int64 `json:"salary_transfer_coin"` SalaryTransferCoin int64 `json:"salary_transfer_coin"`
MicOnlineMS int64 `json:"mic_online_ms"` MicOnlineMS int64 `json:"mic_online_ms"`
MicOnlineUsers int64 `json:"mic_online_users"` MicOnlineUsers int64 `json:"mic_online_users"`
AvgMicOnlineMS int64 `json:"avg_mic_online_ms"` AvgMicOnlineMS int64 `json:"avg_mic_online_ms"`
GiftCoinSpent int64 `json:"gift_coin_spent"` GiftCoinSpent int64 `json:"gift_coin_spent"`
LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` LuckyGiftTurnover int64 `json:"lucky_gift_turnover"`
LuckyGiftPayout int64 `json:"lucky_gift_payout"` LuckyGiftPayout int64 `json:"lucky_gift_payout"`
LuckyGiftProfit int64 `json:"lucky_gift_profit"` LuckyGiftProfit int64 `json:"lucky_gift_profit"`
GameTurnover int64 `json:"game_turnover"` GameTurnover int64 `json:"game_turnover"`
GameProfit int64 `json:"game_profit"` GameProfit int64 `json:"game_profit"`
SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"` SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"`
SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"` SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"`
SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"` SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"`
SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"` SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"`
ARPUUSDMinor int64 `json:"arpu_usd_minor"` ARPUUSDMinor int64 `json:"arpu_usd_minor"`
ARPPUUSDMinor int64 `json:"arppu_usd_minor"` ARPPUUSDMinor int64 `json:"arppu_usd_minor"`
PayerRate float64 `json:"payer_rate"` PayerRate float64 `json:"payer_rate"`
PaidConversionRate float64 `json:"paid_conversion_rate"` PaidConversionRate float64 `json:"paid_conversion_rate"`
RechargeConversionRate float64 `json:"recharge_conversion_rate"` RechargeConversionRate float64 `json:"recharge_conversion_rate"`
D1RetentionUsers *int64 `json:"d1_retention_users"`
D1RetentionBaseUsers *int64 `json:"d1_retention_base_users"`
D1RetentionRate *float64 `json:"d1_retention_rate"`
D7RetentionUsers *int64 `json:"d7_retention_users"`
D7RetentionBaseUsers *int64 `json:"d7_retention_base_users"`
D7RetentionRate *float64 `json:"d7_retention_rate"`
D30RetentionUsers *int64 `json:"d30_retention_users"`
D30RetentionBaseUsers *int64 `json:"d30_retention_base_users"`
D30RetentionRate *float64 `json:"d30_retention_rate"`
} }
type Retention struct { type Retention struct {
RegisteredUsers int64 `json:"registered_users"` RegisteredUsers int64 `json:"registered_users"`
Day1Users int64 `json:"day1_users"` Day1Users *int64 `json:"day1_users"`
Day7Users int64 `json:"day7_users"` Day1BaseUsers *int64 `json:"day1_base_users"`
Day30Users int64 `json:"day30_users"` Day1Rate *float64 `json:"day1_rate"`
Day1Rate float64 `json:"day1_rate"` Day7Users *int64 `json:"day7_users"`
Day7Rate float64 `json:"day7_rate"` Day7BaseUsers *int64 `json:"day7_base_users"`
Day30Rate float64 `json:"day30_rate"` Day7Rate *float64 `json:"day7_rate"`
Day30Users *int64 `json:"day30_users"`
Day30BaseUsers *int64 `json:"day30_base_users"`
Day30Rate *float64 `json:"day30_rate"`
} }
type GameRank struct { type GameRank struct {
@ -192,55 +204,64 @@ type superLuckyAggregate struct {
} }
type CountryBreakdown struct { type CountryBreakdown struct {
CountryID int64 `json:"country_id"` CountryID int64 `json:"country_id"`
RegionID int64 `json:"region_id"` RegionID int64 `json:"region_id"`
NewUsers int64 `json:"new_users"` NewUsers int64 `json:"new_users"`
RegisteredUsers int64 `json:"registered_users"` RegisteredUsers int64 `json:"registered_users"`
Registrations int64 `json:"registrations"` Registrations int64 `json:"registrations"`
ActiveUsers int64 `json:"active_users"` ActiveUsers int64 `json:"active_users"`
PaidUsers int64 `json:"paid_users"` PaidUsers int64 `json:"paid_users"`
RechargeUsers int64 `json:"recharge_users"` RechargeUsers int64 `json:"recharge_users"`
RechargeUSDMinor int64 `json:"recharge_usd_minor"` RechargeUSDMinor int64 `json:"recharge_usd_minor"`
NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"`
CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"`
CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"` CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"`
MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"`
GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"`
CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"`
CoinTotal int64 `json:"coin_total"` CoinTotal int64 `json:"coin_total"`
ConsumedCoin int64 `json:"consumed_coin"` ConsumedCoin int64 `json:"consumed_coin"`
OutputCoin int64 `json:"output_coin"` OutputCoin int64 `json:"output_coin"`
ConsumeOutputRatio float64 `json:"consume_output_ratio"` ConsumeOutputRatio float64 `json:"consume_output_ratio"`
ConsumeOutputDelta int64 `json:"consume_output_delta"` ConsumeOutputDelta int64 `json:"consume_output_delta"`
PlatformGrantCoin int64 `json:"platform_grant_coin"` PlatformGrantCoin int64 `json:"platform_grant_coin"`
ManualGrantCoin int64 `json:"manual_grant_coin"` ManualGrantCoin int64 `json:"manual_grant_coin"`
SalaryUSDMinor int64 `json:"salary_usd_minor"` SalaryUSDMinor int64 `json:"salary_usd_minor"`
SalaryTransferCoin int64 `json:"salary_transfer_coin"` SalaryTransferCoin int64 `json:"salary_transfer_coin"`
MicOnlineMS int64 `json:"mic_online_ms"` MicOnlineMS int64 `json:"mic_online_ms"`
MicOnlineUsers int64 `json:"mic_online_users"` MicOnlineUsers int64 `json:"mic_online_users"`
AvgMicOnlineMS int64 `json:"avg_mic_online_ms"` AvgMicOnlineMS int64 `json:"avg_mic_online_ms"`
GiftCoinSpent int64 `json:"gift_coin_spent"` GiftCoinSpent int64 `json:"gift_coin_spent"`
LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` LuckyGiftTurnover int64 `json:"lucky_gift_turnover"`
LuckyGiftPayout int64 `json:"lucky_gift_payout"` LuckyGiftPayout int64 `json:"lucky_gift_payout"`
LuckyGiftPayers int64 `json:"lucky_gift_payers"` LuckyGiftPayers int64 `json:"lucky_gift_payers"`
LuckyGiftProfit int64 `json:"lucky_gift_profit"` LuckyGiftProfit int64 `json:"lucky_gift_profit"`
LuckyGiftPayoutRate float64 `json:"lucky_gift_payout_rate"` LuckyGiftPayoutRate float64 `json:"lucky_gift_payout_rate"`
LuckyGiftProfitRate float64 `json:"lucky_gift_profit_rate"` LuckyGiftProfitRate float64 `json:"lucky_gift_profit_rate"`
GameTurnover int64 `json:"game_turnover"` GameTurnover int64 `json:"game_turnover"`
GamePayout int64 `json:"game_payout"` GamePayout int64 `json:"game_payout"`
GameRefund int64 `json:"game_refund"` GameRefund int64 `json:"game_refund"`
GamePlayers int64 `json:"game_players"` GamePlayers int64 `json:"game_players"`
GameProfit int64 `json:"game_profit"` GameProfit int64 `json:"game_profit"`
GameProfitRate float64 `json:"game_profit_rate"` GameProfitRate float64 `json:"game_profit_rate"`
SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"` SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"`
SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"` SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"`
SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"` SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"`
SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"` SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"`
ARPUUSDMinor int64 `json:"arpu_usd_minor"` ARPUUSDMinor int64 `json:"arpu_usd_minor"`
ARPPUUSDMinor int64 `json:"arppu_usd_minor"` ARPPUUSDMinor int64 `json:"arppu_usd_minor"`
PayerRate float64 `json:"payer_rate"` PayerRate float64 `json:"payer_rate"`
PaidConversionRate float64 `json:"paid_conversion_rate"` PaidConversionRate float64 `json:"paid_conversion_rate"`
RechargeConversionRate float64 `json:"recharge_conversion_rate"` RechargeConversionRate float64 `json:"recharge_conversion_rate"`
D1RetentionUsers *int64 `json:"d1_retention_users"`
D1RetentionBaseUsers *int64 `json:"d1_retention_base_users"`
D1RetentionRate *float64 `json:"d1_retention_rate"`
D7RetentionUsers *int64 `json:"d7_retention_users"`
D7RetentionBaseUsers *int64 `json:"d7_retention_base_users"`
D7RetentionRate *float64 `json:"d7_retention_rate"`
D30RetentionUsers *int64 `json:"d30_retention_users"`
D30RetentionBaseUsers *int64 `json:"d30_retention_base_users"`
D30RetentionRate *float64 `json:"d30_retention_rate"`
} }
type DailyCountryStat struct { type DailyCountryStat struct {
@ -529,23 +550,23 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
applyOverviewDerivedMetrics(&overview) applyOverviewDerivedMetrics(&overview)
applyOverviewDerivedMetrics(&previous) applyOverviewDerivedMetrics(&previous)
overview.applyDeltaRates(previous) overview.applyDeltaRates(previous)
retention, err := r.queryRetention(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID) retentionCube, err := r.queryRetentionCube(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID)
if err != nil { if err != nil {
return Overview{}, err return Overview{}, err
} }
overview.Retention = retention overview.Retention = retentionCube.Total.toRetention()
ranking, err := r.queryGameRanking(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID) ranking, err := r.queryGameRanking(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID)
if err != nil { if err != nil {
return Overview{}, err return Overview{}, err
} }
overview.GameRanking = ranking overview.GameRanking = ranking
if query.CountryID <= 0 { if query.CountryID <= 0 {
breakdown, err := r.queryCountryBreakdown(ctx, app, statTZ, startDay, endDay, query.RegionID) breakdown, err := r.queryCountryBreakdown(ctx, app, statTZ, startDay, endDay, query.RegionID, retentionCube.ByCountry)
if err != nil { if err != nil {
return Overview{}, err return Overview{}, err
} }
overview.CountryBreakdown = breakdown overview.CountryBreakdown = breakdown
dailyCountryBreakdown, err := r.queryDailyCountryBreakdown(ctx, app, statTZ, startDay, endDay, query.RegionID) dailyCountryBreakdown, err := r.queryDailyCountryBreakdown(ctx, app, statTZ, startDay, endDay, query.RegionID, retentionCube.ByDayCountry)
if err != nil { if err != nil {
return Overview{}, err return Overview{}, err
} }
@ -556,8 +577,16 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov
seriesStartDay = statDayIn(query.SeriesStartMS, statTZ) seriesStartDay = statDayIn(query.SeriesStartMS, statTZ)
seriesEndDay = statDayIn(query.SeriesEndMS-1, statTZ) seriesEndDay = statDayIn(query.SeriesEndMS-1, statTZ)
} }
seriesRetentionByDay := retentionCube.ByDay
if seriesStartDay != startDay || seriesEndDay != endDay {
seriesRetentionCube, err := r.queryRetentionCube(ctx, app, statTZ, seriesStartDay, seriesEndDay, query.CountryID, query.RegionID)
if err != nil {
return Overview{}, err
}
seriesRetentionByDay = seriesRetentionCube.ByDay
}
// daily_series 服务卡片内 sparkline 和弹窗折线图；一条 GROUP BY stat_day 查询返回全部指标，避免前端逐日请求。 // daily_series 服务卡片内 sparkline 和弹窗折线图；一条 GROUP BY stat_day 查询返回全部指标，避免前端逐日请求。
dailySeries, err := r.queryDailySeries(ctx, app, statTZ, seriesStartDay, seriesEndDay, query.CountryID, query.RegionID) dailySeries, err := r.queryDailySeries(ctx, app, statTZ, seriesStartDay, seriesEndDay, query.CountryID, query.RegionID, seriesRetentionByDay)
if err != nil { if err != nil {
return Overview{}, err return Overview{}, err
} }
@ -787,7 +816,7 @@ func (r *Repository) queryRealRoomRobotGiftOverview(ctx context.Context, app, st
return overview, nil return overview, nil
} }
func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) ([]DailySeriesItem, error) { func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64, retentionByDay map[string]retentionCount) ([]DailySeriesItem, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay} args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := "" filter := ""
if countryID > 0 { if countryID > 0 {
@ -840,6 +869,13 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay
for index := range out { for index := range out {
out[index].NewPaidUsers = newPaidUsersByDay[out[index].StatDay] out[index].NewPaidUsers = newPaidUsersByDay[out[index].StatDay]
} }
for index := range out {
if retention, ok := retentionByDay[out[index].StatDay]; ok {
out[index].D1RetentionUsers, out[index].D1RetentionBaseUsers, out[index].D1RetentionRate = retentionPointers(retention.D1Users, retention.D1Base)
out[index].D7RetentionUsers, out[index].D7RetentionBaseUsers, out[index].D7RetentionRate = retentionPointers(retention.D7Users, retention.D7Base)
out[index].D30RetentionUsers, out[index].D30RetentionBaseUsers, out[index].D30RetentionRate = retentionPointers(retention.D30Users, retention.D30Base)
}
}
superByDay, err := r.querySuperLuckyGiftDaily(ctx, app, statTZ, startDay, endDay, countryID, regionID) superByDay, err := r.querySuperLuckyGiftDaily(ctx, app, statTZ, startDay, endDay, countryID, regionID)
if err != nil { if err != nil {
return nil, err return nil, err
@ -895,6 +931,118 @@ func (r *Repository) queryDailyNewPaidUsers(ctx context.Context, app, statTZ, st
return out, rows.Err() return out, rows.Err()
} }
type retentionCount struct {
RegisteredUsers int64
D1Users int64
D1Base int64
D7Users int64
D7Base int64
D30Users int64
D30Base int64
}
func (count *retentionCount) add(next retentionCount) {
count.RegisteredUsers += next.RegisteredUsers
count.D1Users += next.D1Users
count.D1Base += next.D1Base
count.D7Users += next.D7Users
count.D7Base += next.D7Base
count.D30Users += next.D30Users
count.D30Base += next.D30Base
}
func (count retentionCount) toRetention() Retention {
out := Retention{RegisteredUsers: count.RegisteredUsers}
out.Day1Users, out.Day1BaseUsers, out.Day1Rate = retentionPointers(count.D1Users, count.D1Base)
out.Day7Users, out.Day7BaseUsers, out.Day7Rate = retentionPointers(count.D7Users, count.D7Base)
out.Day30Users, out.Day30BaseUsers, out.Day30Rate = retentionPointers(count.D30Users, count.D30Base)
return out
}
type retentionCube struct {
Total retentionCount
ByDay map[string]retentionCount
ByCountry map[int64]retentionCount
ByDayCountry map[string]retentionCount
}
func newRetentionCube() retentionCube {
return retentionCube{
ByDay: map[string]retentionCount{},
ByCountry: map[int64]retentionCount{},
ByDayCountry: map[string]retentionCount{},
}
}
func (cube *retentionCube) add(day string, countryID int64, count retentionCount) {
cube.Total.add(count)
dayCount := cube.ByDay[day]
dayCount.add(count)
cube.ByDay[day] = dayCount
countryCount := cube.ByCountry[countryID]
countryCount.add(count)
cube.ByCountry[countryID] = countryCount
dayCountryKey := dailyCountryKey(day, countryID)
dayCountryCount := cube.ByDayCountry[dayCountryKey]
dayCountryCount.add(count)
cube.ByDayCountry[dayCountryKey] = dayCountryCount
}
func (r *Repository) queryRetentionCube(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (retentionCube, error) {
tz := normalizeStatTZ(statTZ)
today := statDayIn(time.Now().UnixMilli(), tz)
args := []any{
today, today,
today, today,
today, today,
app, tz, startDay, endDay,
}
filter := ""
if countryID > 0 {
filter += " AND r.country_id = ?"
args = append(args, countryID)
}
if regionID > 0 {
filter += " AND r.region_id = ?"
args = append(args, regionID)
}
rows, err := r.db.QueryContext(ctx, `
SELECT DATE_FORMAT(r.registered_day, '%Y-%m-%d'), r.country_id, COUNT(*),
COALESCE(SUM(CASE WHEN d1.user_id IS NOT NULL THEN 1 ELSE 0 END),0),
COALESCE(SUM(CASE WHEN DATE_ADD(r.registered_day, INTERVAL 1 DAY) <= ? THEN 1 ELSE 0 END),0),
COALESCE(SUM(CASE WHEN d7.user_id IS NOT NULL THEN 1 ELSE 0 END),0),
COALESCE(SUM(CASE WHEN DATE_ADD(r.registered_day, INTERVAL 7 DAY) <= ? THEN 1 ELSE 0 END),0),
COALESCE(SUM(CASE WHEN d30.user_id IS NOT NULL THEN 1 ELSE 0 END),0),
COALESCE(SUM(CASE WHEN DATE_ADD(r.registered_day, INTERVAL 30 DAY) <= ? THEN 1 ELSE 0 END),0)
FROM stat_user_registration r
LEFT JOIN stat_user_day_activity d1 ON DATE_ADD(r.registered_day, INTERVAL 1 DAY) <= ? AND d1.app_code = r.app_code AND d1.stat_tz = r.stat_tz AND d1.stat_day = DATE_ADD(r.registered_day, INTERVAL 1 DAY) AND d1.user_id = r.user_id
LEFT JOIN stat_user_day_activity d7 ON DATE_ADD(r.registered_day, INTERVAL 7 DAY) <= ? AND d7.app_code = r.app_code AND d7.stat_tz = r.stat_tz AND d7.stat_day = DATE_ADD(r.registered_day, INTERVAL 7 DAY) AND d7.user_id = r.user_id
LEFT JOIN stat_user_day_activity d30 ON DATE_ADD(r.registered_day, INTERVAL 30 DAY) <= ? AND d30.app_code = r.app_code AND d30.stat_tz = r.stat_tz AND d30.stat_day = DATE_ADD(r.registered_day, INTERVAL 30 DAY) AND d30.user_id = r.user_id
WHERE r.app_code = ? AND r.stat_tz = ? AND r.registered_day BETWEEN ? AND ?`+filter+`
GROUP BY r.registered_day, r.country_id`, args...)
if err != nil {
return retentionCube{}, err
}
defer rows.Close()
out := newRetentionCube()
for rows.Next() {
var day string
var countryID int64
var count retentionCount
if err := rows.Scan(&day, &countryID, &count.RegisteredUsers, &count.D1Users, &count.D1Base, &count.D7Users, &count.D7Base, &count.D30Users, &count.D30Base); err != nil {
return retentionCube{}, err
}
out.add(day, countryID, count)
}
if err := rows.Err(); err != nil {
return retentionCube{}, err
}
return out, nil
}
func applyDailySeriesDerivedMetrics(item *DailySeriesItem) { func applyDailySeriesDerivedMetrics(item *DailySeriesItem) {
if item == nil { if item == nil {
return return
@ -975,34 +1123,6 @@ func (overview *Overview) applyDeltaRates(previous Overview) {
overview.GameProfitDeltaRate = deltaRate(overview.GameProfit, previous.GameProfit) overview.GameProfitDeltaRate = deltaRate(overview.GameProfit, previous.GameProfit)
} }
func (r *Repository) queryRetention(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) (Retention, error) {
filter := ""
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
if countryID > 0 {
filter = " AND r.country_id = ?"
args = append(args, countryID)
}
if regionID > 0 {
filter += " AND r.region_id = ?"
args = append(args, regionID)
}
row := r.db.QueryRowContext(ctx, `
SELECT COUNT(*),
COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.stat_tz=r.stat_tz AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 1 DAY))),0),
COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.stat_tz=r.stat_tz AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 7 DAY))),0),
COALESCE(SUM(EXISTS(SELECT 1 FROM stat_user_day_activity a WHERE a.app_code=r.app_code AND a.stat_tz=r.stat_tz AND a.user_id=r.user_id AND a.stat_day=DATE_ADD(r.registered_day, INTERVAL 30 DAY))),0)
FROM stat_user_registration r
WHERE r.app_code = ? AND r.stat_tz = ? AND r.registered_day BETWEEN ? AND ?`+filter, args...)
var out Retention
if err := row.Scan(&out.RegisteredUsers, &out.Day1Users, &out.Day7Users, &out.Day30Users); err != nil && err != sql.ErrNoRows {
return Retention{}, err
}
out.Day1Rate = ratio(out.Day1Users, out.RegisteredUsers)
out.Day7Rate = ratio(out.Day7Users, out.RegisteredUsers)
out.Day30Rate = ratio(out.Day30Users, out.RegisteredUsers)
return out, nil
}
func (r *Repository) queryGameRanking(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) ([]GameRank, error) { func (r *Repository) queryGameRanking(ctx context.Context, app, statTZ, startDay, endDay string, countryID int64, regionID int64) ([]GameRank, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay} args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := "" filter := ""
@ -1039,7 +1159,7 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, statTZ, startDay
return out, rows.Err() return out, rows.Err()
} }
func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, startDay, endDay string, regionID int64) ([]CountryBreakdown, error) { func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, startDay, endDay string, regionID int64, retentionByCountry map[int64]retentionCount) ([]CountryBreakdown, error) {
args := []any{endDay, endDay, app, normalizeStatTZ(statTZ), startDay, endDay} args := []any{endDay, endDay, app, normalizeStatTZ(statTZ), startDay, endDay}
filter := "" filter := ""
if regionID > 0 { if regionID > 0 {
@ -1081,6 +1201,13 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta
if err := rows.Err(); err != nil { if err := rows.Err(); err != nil {
return nil, err return nil, err
} }
for index := range out {
if retention, ok := retentionByCountry[out[index].CountryID]; ok {
out[index].D1RetentionUsers, out[index].D1RetentionBaseUsers, out[index].D1RetentionRate = retentionPointers(retention.D1Users, retention.D1Base)
out[index].D7RetentionUsers, out[index].D7RetentionBaseUsers, out[index].D7RetentionRate = retentionPointers(retention.D7Users, retention.D7Base)
out[index].D30RetentionUsers, out[index].D30RetentionBaseUsers, out[index].D30RetentionRate = retentionPointers(retention.D30Users, retention.D30Base)
}
}
superByCountry, err := r.querySuperLuckyGiftCountry(ctx, app, statTZ, startDay, endDay, regionID) superByCountry, err := r.querySuperLuckyGiftCountry(ctx, app, statTZ, startDay, endDay, regionID)
if err != nil { if err != nil {
return nil, err return nil, err
@ -1095,7 +1222,7 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta
return out, nil return out, nil
} }
func (r *Repository) queryDailyCountryBreakdown(ctx context.Context, app, statTZ, startDay, endDay string, regionID int64) ([]DailyCountryStat, error) { func (r *Repository) queryDailyCountryBreakdown(ctx context.Context, app, statTZ, startDay, endDay string, regionID int64, retentionByDayCountry map[string]retentionCount) ([]DailyCountryStat, error) {
args := []any{app, normalizeStatTZ(statTZ), startDay, endDay} args := []any{app, normalizeStatTZ(statTZ), startDay, endDay}
filter := "" filter := ""
if regionID > 0 { if regionID > 0 {
@ -1137,6 +1264,13 @@ func (r *Repository) queryDailyCountryBreakdown(ctx context.Context, app, statTZ
if err := rows.Err(); err != nil { if err := rows.Err(); err != nil {
return nil, err return nil, err
} }
for index := range out {
if retention, ok := retentionByDayCountry[dailyCountryKey(out[index].StatDay, out[index].CountryID)]; ok {
out[index].D1RetentionUsers, out[index].D1RetentionBaseUsers, out[index].D1RetentionRate = retentionPointers(retention.D1Users, retention.D1Base)
out[index].D7RetentionUsers, out[index].D7RetentionBaseUsers, out[index].D7RetentionRate = retentionPointers(retention.D7Users, retention.D7Base)
out[index].D30RetentionUsers, out[index].D30RetentionBaseUsers, out[index].D30RetentionRate = retentionPointers(retention.D30Users, retention.D30Base)
}
}
superByDayCountry, err := r.querySuperLuckyGiftDailyCountry(ctx, app, statTZ, startDay, endDay, regionID) superByDayCountry, err := r.querySuperLuckyGiftDailyCountry(ctx, app, statTZ, startDay, endDay, regionID)
if err != nil { if err != nil {
return nil, err return nil, err
@ -1912,6 +2046,16 @@ func ratio(numerator, denominator int64) float64 {
return float64(numerator) / float64(denominator) return float64(numerator) / float64(denominator)
} }
func retentionPointers(users, base int64) (*int64, *int64, *float64) {
if base <= 0 {
return nil, nil, nil
}
usersValue := users
baseValue := base
rateValue := ratio(users, base)
return &usersValue, &baseValue, &rateValue
}
func selfGameMetricDefinitions() []MetricDefinition { func selfGameMetricDefinitions() []MetricDefinition {
return []MetricDefinition{ return []MetricDefinition{
{Metric: "game_pv", Definition: "H5 page_open event_count；同一用户多次打开会重复计数。"}, {Metric: "game_pv", Definition: "H5 page_open event_count；同一用户多次打开会重复计数。"},

156
services/statistics-service/internal/storage/mysql/query_test.go
View File

@ -3,11 +3,13 @@ package mysql
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"math"
"runtime" "runtime"
"strings" "strings"
"testing" "testing"
"time" "time"
"github.com/DATA-DOG/go-sqlmock"
"hyapp/internal/testutil/mysqlschema" "hyapp/internal/testutil/mysqlschema"
"hyapp/pkg/appcode" "hyapp/pkg/appcode"
) )
@ -615,6 +617,139 @@ func TestQueryOverviewReturnsPreviousPeriodDeltaRates(t *testing.T) {
} }
} }
func TestQueryOverviewReturnsRetentionForAllBreakdowns(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_query_retention_test",
})
repository, err := Open(ctx, statsSchema.DSN)
if err != nil {
t.Fatalf("open repository: %v", err)
}
t.Cleanup(func() { _ = repository.Close() })
today, err := time.Parse("2006-01-02", statDayIn(time.Now().UnixMilli(), StatTZUTC))
if err != nil {
t.Fatalf("parse today: %v", err)
}
registeredAt := today.AddDate(0, 0, -1)
registeredDay := registeredAt.Format("2006-01-02")
retainedDay := today.Format("2006-01-02")
if _, err := repository.db.ExecContext(ctx, `
INSERT INTO stat_app_day_country (
app_code, stat_day, country_id, region_id, new_users, active_users, updated_at_ms
) VALUES ('lalu', ?, 86, 210, 2, 2, 1)`, registeredDay); err != nil {
t.Fatalf("seed day country stats: %v", err)
}
if _, err := repository.db.ExecContext(ctx, `
INSERT INTO stat_user_registration (
app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms
) VALUES
('lalu', 'UTC', 9001, ?, 86, 210, 1),
('lalu', 'UTC', 9002, ?, 86, 210, 2)`, registeredDay, registeredDay); err != nil {
t.Fatalf("seed registrations: %v", err)
}
if _, err := repository.db.ExecContext(ctx, `
INSERT INTO stat_user_day_activity (
app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms
) VALUES ('lalu', 'UTC', ?, 86, 210, 9001, 3)`, retainedDay); err != nil {
t.Fatalf("seed retained activity: %v", err)
}
overview, err := repository.QueryOverview(ctx, OverviewQuery{
AppCode: "lalu",
StartMS: registeredAt.UnixMilli(),
EndMS: registeredAt.AddDate(0, 0, 1).UnixMilli(),
RegionID: 210,
})
if err != nil {
t.Fatalf("query overview: %v", err)
}
if overview.Retention.RegisteredUsers != 2 {
t.Fatalf("retention registered users mismatch: %+v", overview.Retention)
}
requireInt64Ptr(t, "overview d1 users", overview.Retention.Day1Users, 1)
requireInt64Ptr(t, "overview d1 base", overview.Retention.Day1BaseUsers, 2)
requireFloat64Ptr(t, "overview d1 rate", overview.Retention.Day1Rate, 0.5)
requireNilRetention(t, "overview d7", overview.Retention.Day7Users, overview.Retention.Day7BaseUsers, overview.Retention.Day7Rate)
requireNilRetention(t, "overview d30", overview.Retention.Day30Users, overview.Retention.Day30BaseUsers, overview.Retention.Day30Rate)
if len(overview.DailySeries) != 1 {
t.Fatalf("daily series missing retention day: %+v", overview.DailySeries)
}
requireInt64Ptr(t, "daily d1 users", overview.DailySeries[0].D1RetentionUsers, 1)
requireInt64Ptr(t, "daily d1 base", overview.DailySeries[0].D1RetentionBaseUsers, 2)
requireFloat64Ptr(t, "daily d1 rate", overview.DailySeries[0].D1RetentionRate, 0.5)
requireNilRetention(t, "daily d7", overview.DailySeries[0].D7RetentionUsers, overview.DailySeries[0].D7RetentionBaseUsers, overview.DailySeries[0].D7RetentionRate)
if len(overview.CountryBreakdown) != 1 {
t.Fatalf("country breakdown missing retention country: %+v", overview.CountryBreakdown)
}
requireInt64Ptr(t, "country d1 users", overview.CountryBreakdown[0].D1RetentionUsers, 1)
requireInt64Ptr(t, "country d1 base", overview.CountryBreakdown[0].D1RetentionBaseUsers, 2)
requireFloat64Ptr(t, "country d1 rate", overview.CountryBreakdown[0].D1RetentionRate, 0.5)
if len(overview.DailyCountryBreakdown) != 1 {
t.Fatalf("daily country breakdown missing retention country: %+v", overview.DailyCountryBreakdown)
}
requireInt64Ptr(t, "daily country d1 users", overview.DailyCountryBreakdown[0].D1RetentionUsers, 1)
requireInt64Ptr(t, "daily country d1 base", overview.DailyCountryBreakdown[0].D1RetentionBaseUsers, 2)
requireFloat64Ptr(t, "daily country d1 rate", overview.DailyCountryBreakdown[0].D1RetentionRate, 0.5)
}
func TestQueryRetentionCubeRollsUpDailyCountryRows(t *testing.T) {
ctx := context.Background()
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
repository := &Repository{db: db}
today := statDayIn(time.Now().UnixMilli(), StatTZUTC)
rows := sqlmock.NewRows([]string{
"registered_day", "country_id", "registered_users",
"d1_users", "d1_base", "d7_users", "d7_base", "d30_users", "d30_base",
}).
AddRow("2026-06-01", int64(86), int64(2), int64(1), int64(2), int64(0), int64(0), int64(0), int64(0)).
AddRow("2026-06-01", int64(66), int64(3), int64(0), int64(3), int64(1), int64(3), int64(0), int64(0)).
AddRow("2026-06-02", int64(86), int64(4), int64(2), int64(4), int64(0), int64(0), int64(0), int64(0))
mock.ExpectQuery("SELECT DATE_FORMAT").
WithArgs(today, today, today, today, today, today, "lalu", StatTZUTC, "2026-06-01", "2026-06-02", int64(210)).
WillReturnRows(rows)
cube, err := repository.queryRetentionCube(ctx, "lalu", StatTZUTC, "2026-06-01", "2026-06-02", 0, 210)
if err != nil {
t.Fatalf("query retention cube: %v", err)
}
if cube.Total.RegisteredUsers != 9 || cube.Total.D1Users != 3 || cube.Total.D1Base != 9 || cube.Total.D7Users != 1 || cube.Total.D7Base != 3 {
t.Fatalf("total retention rollup mismatch: %+v", cube.Total)
}
retention := cube.Total.toRetention()
requireInt64Ptr(t, "total d1 users", retention.Day1Users, 3)
requireFloat64Ptr(t, "total d1 rate", retention.Day1Rate, 1.0/3.0)
requireNilRetention(t, "total d30", retention.Day30Users, retention.Day30BaseUsers, retention.Day30Rate)
day := cube.ByDay["2026-06-01"]
if day.RegisteredUsers != 5 || day.D1Users != 1 || day.D1Base != 5 || day.D7Users != 1 || day.D7Base != 3 {
t.Fatalf("day retention rollup mismatch: %+v", day)
}
country := cube.ByCountry[86]
if country.RegisteredUsers != 6 || country.D1Users != 3 || country.D1Base != 6 {
t.Fatalf("country retention rollup mismatch: %+v", country)
}
dayCountry := cube.ByDayCountry[dailyCountryKey("2026-06-01", 66)]
if dayCountry.RegisteredUsers != 3 || dayCountry.D1Users != 0 || dayCountry.D1Base != 3 || dayCountry.D7Users != 1 || dayCountry.D7Base != 3 {
t.Fatalf("daily country retention rollup mismatch: %+v", dayCountry)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestQueryOverviewReturnsRealRoomRobotGiftStats(t *testing.T) { func TestQueryOverviewReturnsRealRoomRobotGiftStats(t *testing.T) {
ctx := context.Background() ctx := context.Background()
_, file, _, _ := runtime.Caller(0) _, file, _, _ := runtime.Caller(0)
@ -860,3 +995,24 @@ func TestQuerySelfGameOverviewUsesLatestPoolBalanceSnapshot(t *testing.T) {
t.Fatalf("pool balance should be latest snapshot, not max balance: %+v", overview.PoolStats) t.Fatalf("pool balance should be latest snapshot, not max balance: %+v", overview.PoolStats)
} }
} }
func requireInt64Ptr(t *testing.T, name string, got *int64, want int64) {
t.Helper()
if got == nil || *got != want {
t.Fatalf("%s mismatch: got=%v want=%d", name, got, want)
}
}
func requireFloat64Ptr(t *testing.T, name string, got *float64, want float64) {
t.Helper()
if got == nil || math.Abs(*got-want) > 0.000001 {
t.Fatalf("%s mismatch: got=%v want=%f", name, got, want)
}
}
func requireNilRetention(t *testing.T, name string, users *int64, base *int64, rate *float64) {
t.Helper()
if users != nil || base != nil || rate != nil {
t.Fatalf("%s should be nil before cohort matures: users=%v base=%v rate=%v", name, users, base, rate)
}
}

4
services/statistics-service/internal/storage/mysql/repository.go
View File

@ -168,7 +168,9 @@ func (r *Repository) Migrate(ctx context.Context) error {
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', user_id BIGINT NOT NULL, registered_day DATE NOT NULL, app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_tz, user_id), KEY idx_stat_user_registration_day (app_code, stat_tz, registered_day, country_id), PRIMARY KEY (app_code, stat_tz, user_id), KEY idx_stat_user_registration_day (app_code, stat_tz, registered_day, country_id),
KEY idx_stat_user_registration_region (app_code, stat_tz, registered_day, region_id) KEY idx_stat_user_registration_region (app_code, stat_tz, registered_day, region_id),
KEY idx_stat_user_registration_country_day (app_code, stat_tz, country_id, registered_day, user_id),
KEY idx_stat_user_registration_region_day (app_code, stat_tz, region_id, registered_day, country_id, user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_user_dimension ( `CREATE TABLE IF NOT EXISTS stat_user_dimension (
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,