diff --git a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go index 64b5fc39..097736f4 100644 --- a/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go +++ b/server/admin/internal/modules/dashboard/aslan_mongo_dashboard.go @@ -917,7 +917,7 @@ func (s *AslanMongoDashboardSource) loadCoinFlows(ctx context.Context, startDate } // 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 { location, err := time.LoadLocation(s.statTimezone) if err != nil { diff --git a/server/admin/internal/modules/dashboard/external_dashboard.go b/server/admin/internal/modules/dashboard/external_dashboard.go index 3a61be7d..11146239 100644 --- a/server/admin/internal/modules/dashboard/external_dashboard.go +++ b/server/admin/internal/modules/dashboard/external_dashboard.go @@ -489,15 +489,15 @@ func (m externalDashboardMetric) toMap() map[string]any { "consumed_coin": nullableInt64(m.ConsumedCoin), "consume_output_ratio": consumeOutputRatioValue(m.ConsumedCoin, m.OutputCoin), "daily_active_user": m.ActiveUsers, - "d1_retention_base_users": m.D1RetentionBaseUsers, - "d1_retention_rate": m.D1RetentionRate, - "d1_retention_users": m.D1RetentionUsers, - "d7_retention_base_users": m.D7RetentionBaseUsers, - "d7_retention_rate": m.D7RetentionRate, - "d7_retention_users": m.D7RetentionUsers, - "d30_retention_base_users": m.D30RetentionBaseUsers, - "d30_retention_rate": m.D30RetentionRate, - "d30_retention_users": m.D30RetentionUsers, + "d1_retention_base_users": retentionMetricValue(m.D1RetentionBaseUsers, m.D1RetentionBaseUsers), + "d1_retention_rate": retentionRateValue(m.D1RetentionUsers, m.D1RetentionBaseUsers), + "d1_retention_users": retentionMetricValue(m.D1RetentionUsers, m.D1RetentionBaseUsers), + "d7_retention_base_users": retentionMetricValue(m.D7RetentionBaseUsers, m.D7RetentionBaseUsers), + "d7_retention_rate": retentionRateValue(m.D7RetentionUsers, m.D7RetentionBaseUsers), + "d7_retention_users": retentionMetricValue(m.D7RetentionUsers, m.D7RetentionBaseUsers), + "d30_retention_base_users": retentionMetricValue(m.D30RetentionBaseUsers, m.D30RetentionBaseUsers), + "d30_retention_rate": retentionRateValue(m.D30RetentionUsers, m.D30RetentionBaseUsers), + "d30_retention_users": retentionMetricValue(m.D30RetentionUsers, m.D30RetentionBaseUsers), "game_payout": m.GamePayout, "game_players": m.GamePlayers, "game_profit": m.GameProfit, @@ -546,15 +546,15 @@ func (m externalDashboardMetric) toMap() map[string]any { func (m externalDashboardMetric) retentionMap() map[string]any { return map[string]any{ - "day1_base_users": m.D1RetentionBaseUsers, - "day1_rate": m.D1RetentionRate, - "day1_users": m.D1RetentionUsers, - "day7_base_users": m.D7RetentionBaseUsers, - "day7_rate": m.D7RetentionRate, - "day7_users": m.D7RetentionUsers, - "day30_base_users": m.D30RetentionBaseUsers, - "day30_rate": m.D30RetentionRate, - "day30_users": m.D30RetentionUsers, + "day1_base_users": retentionMetricValue(m.D1RetentionBaseUsers, m.D1RetentionBaseUsers), + "day1_rate": retentionRateValue(m.D1RetentionUsers, m.D1RetentionBaseUsers), + "day1_users": retentionMetricValue(m.D1RetentionUsers, m.D1RetentionBaseUsers), + "day7_base_users": retentionMetricValue(m.D7RetentionBaseUsers, m.D7RetentionBaseUsers), + "day7_rate": retentionRateValue(m.D7RetentionUsers, m.D7RetentionBaseUsers), + "day7_users": retentionMetricValue(m.D7RetentionUsers, m.D7RetentionBaseUsers), + "day30_base_users": retentionMetricValue(m.D30RetentionBaseUsers, m.D30RetentionBaseUsers), + "day30_rate": retentionRateValue(m.D30RetentionUsers, m.D30RetentionBaseUsers), + "day30_users": retentionMetricValue(m.D30RetentionUsers, m.D30RetentionBaseUsers), "registered_users": m.NewUsers, } } @@ -757,6 +757,20 @@ func ratioFloat64(numerator int64, denominator int64) float64 { 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 { if previous == 0 { if current == 0 { diff --git a/server/admin/internal/modules/dashboard/external_dashboard_test.go b/server/admin/internal/modules/dashboard/external_dashboard_test.go index c376f1e0..e26da98d 100644 --- a/server/admin/internal/modules/dashboard/external_dashboard_test.go +++ b/server/admin/internal/modules/dashboard/external_dashboard_test.go @@ -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) { db, mock, err := sqlmock.New() if err != nil { diff --git a/server/admin/internal/modules/databi/service.go b/server/admin/internal/modules/databi/service.go index f3f612fb..26683c64 100644 --- a/server/admin/internal/modules/databi/service.go +++ b/server/admin/internal/modules/databi/service.go @@ -1345,9 +1345,33 @@ func topLevelMetrics(overview map[string]any) map[string]any { } out[key] = value } + flattenRetentionMetrics(out, overview["retention"]) 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 { out := make([]ScopeInfo, 0, len(scopes)) for _, scope := range scopes { diff --git a/server/admin/internal/modules/databi/service_test.go b/server/admin/internal/modules/databi/service_test.go index 7d9c6689..35c67255 100644 --- a/server/admin/internal/modules/databi/service_test.go +++ b/server/admin/internal/modules/databi/service_test.go @@ -42,12 +42,18 @@ func TestRegionDisplay(t *testing.T) { } func TestTopLevelMetricsStripsStructuralKeys(t *testing.T) { + day1Rate := 0.5 out := topLevelMetrics(map[string]any{ "recharge_usd_minor": int64(5), "daily_series": []any{"x"}, "country_breakdown": []any{"y"}, "daily_country_breakdown": []any{"z"}, "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 { t.Fatalf("expected daily_series stripped") @@ -55,6 +61,9 @@ func TestTopLevelMetricsStripsStructuralKeys(t *testing.T) { if out["recharge_usd_minor"] != int64(5) { 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 圆整值； diff --git a/services/room-service/internal/room/command/command.go b/services/room-service/internal/room/command/command.go index 6b45bb81..f20edc3f 100644 --- a/services/room-service/internal/room/command/command.go +++ b/services/room-service/internal/room/command/command.go @@ -248,6 +248,11 @@ type ConfirmMicPublishing struct { RoomVersion int64 `json:"room_version"` // EventTimeMS 是客户端 SDK 或 RTC webhook 的事件时间，必须晚于已接受事件。 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 string `json:"source,omitempty"` // TargetDisplayProfile 是发流确认目标用户展示快照，只用于 IM/UI 兜底。 @@ -686,6 +691,8 @@ func IdempotencyPayload(payload []byte) ([]byte, error) { delete(values, "password_hash") // MicUp 的发流确认 deadline 由 room-service 生成，客户端重试同一 command_id 时不能因此冲突。 delete(values, "publish_deadline_ms") + // ConfirmMicPublishing 的接受时间由 room-service 生成，不能参与客户端请求幂等语义。 + delete(values, "accepted_event_time_ms") return json.Marshal(values) } diff --git a/services/room-service/internal/room/service/mic.go b/services/room-service/internal/room/service/mic.go index a6280a09..a287ce0f 100644 --- a/services/room-service/internal/room/service/mic.go +++ b/services/room-service/internal/room/service/mic.go @@ -379,10 +379,8 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM // RTC webhook 可能晚于用户下麦到达；旧事件只忽略，不返回错误清掉新状态。 return mutationResult{snapshot: current.ToProto()}, nil, nil } - if seat.MicSessionID != cmd.MicSessionID || - cmd.RoomVersion < seat.MicSessionRoomVersion || - cmd.EventTimeMS <= seat.LastPublishEventTimeMS || - cmd.EventTimeMS > seat.PublishDeadlineMS { + receivedAtMS := micPublishConfirmReceivedAtMS(seat, cmd, s.micPublishTimeout, now) + if !micPublishConfirmationMatchesCurrentSession(seat, cmd, receivedAtMS) { return mutationResult{snapshot: current.ToProto(), seatNo: seat.SeatNo}, nil, nil } 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") } + acceptedCmd := cmd + acceptedCmd.AcceptedEventTimeMS = receivedAtMS + commandPayload, err := command.Serialize(acceptedCmd) + if err != nil { + return mutationResult{}, nil, err + } + index := current.SeatIndex(seat.SeatNo) current.MicSeats[index].PublishState = state.MicPublishPublishing - current.MicSeats[index].LastPublishEventTimeMS = cmd.EventTimeMS - // 确认发流本身代表当前会话刚被服务端接受，先初始化心跳时间，后续 MicHeartbeat 只负责刷新这个独立字段。 - current.MicSeats[index].MicHeartbeatAtMS = now.UnixMilli() + current.MicSeats[index].LastPublishEventTimeMS = receivedAtMS + // 确认发流本身代表当前会话刚被服务端接受，心跳起点使用同一个 accepted time，保证 command log 回放一致。 + current.MicSeats[index].MicHeartbeatAtMS = receivedAtMS if micPublishConfirmSourceIsMuted(cmd.Source) { current.MicSeats[index].MicMuted = true } @@ -413,7 +418,7 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM MicSessionId: cmd.MicSessionID, PublishState: state.MicPublishPublishing, PublishDeadlineMs: seat.PublishDeadlineMS, - PublishEventTimeMs: cmd.EventTimeMS, + PublishEventTimeMs: receivedAtMS, MicMuted: current.MicSeats[index].MicMuted, SeatStatus: seatStatus, TargetGiftValue: targetGiftValue, @@ -433,6 +438,7 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM micSessionID: cmd.MicSessionID, publishDeadlineMS: seat.PublishDeadlineMS, micHeartbeatAtMS: current.MicSeats[index].MicHeartbeatAtMS, + commandPayload: commandPayload, syncEvent: &tencentim.RoomEvent{ EventID: micEvent.EventID, RoomID: current.RoomID, @@ -456,6 +462,46 @@ func (s *Service) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmM }, 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 麦位会话的服务端心跳时间。 func (s *Service) MicHeartbeat(ctx context.Context, req *roomv1.MicHeartbeatRequest) (*roomv1.MicHeartbeatResponse, error) { ctx = contextFromMeta(ctx, req.GetMeta()) diff --git a/services/room-service/internal/room/service/mic_test.go b/services/room-service/internal/room/service/mic_test.go index 4cfb40e9..dd5c1fbf 100644 --- a/services/room-service/internal/room/service/mic_test.go +++ b/services/room-service/internal/room/service/mic_test.go @@ -2,10 +2,13 @@ package service_test import ( "context" + "fmt" "testing" "time" roomv1 "hyapp.local/api/proto/room/v1" + "hyapp/pkg/appcode" + "hyapp/services/room-service/internal/room/command" "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) { ctx := context.Background() 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) { ctx := context.Background() repository := mysqltest.NewRepository(t) diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index b6434349..1892b928 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -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) } index := current.SeatIndex(seat.SeatNo) + acceptedEventTimeMS := confirmMicPublishingAcceptedEventTimeMS(typed) current.MicSeats[index].PublishState = state.MicPublishPublishing - current.MicSeats[index].LastPublishEventTimeMS = typed.EventTimeMS - current.MicSeats[index].MicHeartbeatAtMS = typed.SentAtMS + current.MicSeats[index].LastPublishEventTimeMS = acceptedEventTimeMS + current.MicSeats[index].MicHeartbeatAtMS = acceptedEventTimeMS current.Version++ case *command.MicHeartbeat: seat, exists := current.SeatByUser(typed.TargetUserID) diff --git a/services/room-service/internal/room/service/rtc_event.go b/services/room-service/internal/room/service/rtc_event.go index 00da11b8..19c621fc 100644 --- a/services/room-service/internal/room/service/rtc_event.go +++ b/services/room-service/internal/room/service/rtc_event.go @@ -236,6 +236,10 @@ func rtcEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, tim if !rtcStopEventMatchesCurrentMicSession(seat, eventTimeMS, timeout) { return false } + if eventTimeMS <= seat.LastPublishEventTimeMS { + // audio_started 必须晚于已接受发布水位，防止腾讯重试或乱序开始事件反复推进状态。 + return false + } if seat.PublishDeadlineMS > 0 && eventTimeMS > seat.PublishDeadlineMS { // audio_started 过了确认 deadline 不再接受，超时 worker 会按 mic_session_id 释放麦位。 return false @@ -246,8 +250,8 @@ func rtcEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, tim // rtcStopEventMatchesCurrentMicSession 校验停止/退房事件是否能作用于当前 mic_session。 func rtcStopEventMatchesCurrentMicSession(seat state.MicSeat, eventTimeMS int64, timeout time.Duration) bool { - if seat.MicSessionID == "" || eventTimeMS <= seat.LastPublishEventTimeMS { - // 没有会话或事件时间不递增时，事件不能改变当前麦位。 + if seat.MicSessionID == "" { + // 没有会话时，停止/退房事件只能是旧回调。 return false } if createdAtMS := rtcMicSessionCreatedAtMS(seat, timeout); createdAtMS > 0 && eventTimeMS < createdAtMS { diff --git a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql index d45961de..ac2d4ae8 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -109,7 +109,9 @@ CREATE TABLE IF NOT EXISTS stat_user_registration ( 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), - 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 表'; CREATE TABLE IF NOT EXISTS stat_user_dimension ( diff --git a/services/statistics-service/deploy/mysql/migrations/001_retention_registration_filter_indexes.sql b/services/statistics-service/deploy/mysql/migrations/001_retention_registration_filter_indexes.sql new file mode 100644 index 00000000..6df546c7 --- /dev/null +++ b/services/statistics-service/deploy/mysql/migrations/001_retention_registration_filter_indexes.sql @@ -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' + ); diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index a44b10b5..7ccc23e6 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -111,59 +111,71 @@ type Overview struct { } type DailySeriesItem struct { - StatDay string `json:"stat_day"` - Label string `json:"label"` - NewUsers int64 `json:"new_users"` - RegisteredUsers int64 `json:"registered_users"` - Registrations int64 `json:"registrations"` - ActiveUsers int64 `json:"active_users"` - PaidUsers int64 `json:"paid_users"` - NewPaidUsers int64 `json:"new_paid_users"` - RechargeUsers int64 `json:"recharge_users"` - RechargeUSDMinor int64 `json:"recharge_usd_minor"` - NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` - CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` - CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"` - MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` - GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` - CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` - CoinTotal int64 `json:"coin_total"` - ConsumedCoin int64 `json:"consumed_coin"` - OutputCoin int64 `json:"output_coin"` - ConsumeOutputRatio float64 `json:"consume_output_ratio"` - ConsumeOutputDelta int64 `json:"consume_output_delta"` - PlatformGrantCoin int64 `json:"platform_grant_coin"` - ManualGrantCoin int64 `json:"manual_grant_coin"` - SalaryUSDMinor int64 `json:"salary_usd_minor"` - SalaryTransferCoin int64 `json:"salary_transfer_coin"` - MicOnlineMS int64 `json:"mic_online_ms"` - MicOnlineUsers int64 `json:"mic_online_users"` - AvgMicOnlineMS int64 `json:"avg_mic_online_ms"` - GiftCoinSpent int64 `json:"gift_coin_spent"` - LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` - LuckyGiftPayout int64 `json:"lucky_gift_payout"` - LuckyGiftProfit int64 `json:"lucky_gift_profit"` - GameTurnover int64 `json:"game_turnover"` - GameProfit int64 `json:"game_profit"` - SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"` - SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"` - SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"` - SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"` - ARPUUSDMinor int64 `json:"arpu_usd_minor"` - ARPPUUSDMinor int64 `json:"arppu_usd_minor"` - PayerRate float64 `json:"payer_rate"` - PaidConversionRate float64 `json:"paid_conversion_rate"` - RechargeConversionRate float64 `json:"recharge_conversion_rate"` + StatDay string `json:"stat_day"` + Label string `json:"label"` + NewUsers int64 `json:"new_users"` + RegisteredUsers int64 `json:"registered_users"` + Registrations int64 `json:"registrations"` + ActiveUsers int64 `json:"active_users"` + PaidUsers int64 `json:"paid_users"` + NewPaidUsers int64 `json:"new_paid_users"` + RechargeUsers int64 `json:"recharge_users"` + RechargeUSDMinor int64 `json:"recharge_usd_minor"` + NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` + CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` + CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"` + MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` + GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` + CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` + CoinTotal int64 `json:"coin_total"` + ConsumedCoin int64 `json:"consumed_coin"` + OutputCoin int64 `json:"output_coin"` + ConsumeOutputRatio float64 `json:"consume_output_ratio"` + ConsumeOutputDelta int64 `json:"consume_output_delta"` + PlatformGrantCoin int64 `json:"platform_grant_coin"` + ManualGrantCoin int64 `json:"manual_grant_coin"` + SalaryUSDMinor int64 `json:"salary_usd_minor"` + SalaryTransferCoin int64 `json:"salary_transfer_coin"` + MicOnlineMS int64 `json:"mic_online_ms"` + MicOnlineUsers int64 `json:"mic_online_users"` + AvgMicOnlineMS int64 `json:"avg_mic_online_ms"` + GiftCoinSpent int64 `json:"gift_coin_spent"` + LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` + LuckyGiftPayout int64 `json:"lucky_gift_payout"` + LuckyGiftProfit int64 `json:"lucky_gift_profit"` + GameTurnover int64 `json:"game_turnover"` + GameProfit int64 `json:"game_profit"` + SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"` + SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"` + SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"` + SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"` + ARPUUSDMinor int64 `json:"arpu_usd_minor"` + ARPPUUSDMinor int64 `json:"arppu_usd_minor"` + PayerRate float64 `json:"payer_rate"` + PaidConversionRate float64 `json:"paid_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 { - RegisteredUsers int64 `json:"registered_users"` - Day1Users int64 `json:"day1_users"` - Day7Users int64 `json:"day7_users"` - Day30Users int64 `json:"day30_users"` - Day1Rate float64 `json:"day1_rate"` - Day7Rate float64 `json:"day7_rate"` - Day30Rate float64 `json:"day30_rate"` + RegisteredUsers int64 `json:"registered_users"` + Day1Users *int64 `json:"day1_users"` + Day1BaseUsers *int64 `json:"day1_base_users"` + Day1Rate *float64 `json:"day1_rate"` + Day7Users *int64 `json:"day7_users"` + Day7BaseUsers *int64 `json:"day7_base_users"` + Day7Rate *float64 `json:"day7_rate"` + Day30Users *int64 `json:"day30_users"` + Day30BaseUsers *int64 `json:"day30_base_users"` + Day30Rate *float64 `json:"day30_rate"` } type GameRank struct { @@ -192,55 +204,64 @@ type superLuckyAggregate struct { } type CountryBreakdown struct { - CountryID int64 `json:"country_id"` - RegionID int64 `json:"region_id"` - NewUsers int64 `json:"new_users"` - RegisteredUsers int64 `json:"registered_users"` - Registrations int64 `json:"registrations"` - ActiveUsers int64 `json:"active_users"` - PaidUsers int64 `json:"paid_users"` - RechargeUsers int64 `json:"recharge_users"` - RechargeUSDMinor int64 `json:"recharge_usd_minor"` - NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` - CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` - CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"` - MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` - GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` - CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` - CoinTotal int64 `json:"coin_total"` - ConsumedCoin int64 `json:"consumed_coin"` - OutputCoin int64 `json:"output_coin"` - ConsumeOutputRatio float64 `json:"consume_output_ratio"` - ConsumeOutputDelta int64 `json:"consume_output_delta"` - PlatformGrantCoin int64 `json:"platform_grant_coin"` - ManualGrantCoin int64 `json:"manual_grant_coin"` - SalaryUSDMinor int64 `json:"salary_usd_minor"` - SalaryTransferCoin int64 `json:"salary_transfer_coin"` - MicOnlineMS int64 `json:"mic_online_ms"` - MicOnlineUsers int64 `json:"mic_online_users"` - AvgMicOnlineMS int64 `json:"avg_mic_online_ms"` - GiftCoinSpent int64 `json:"gift_coin_spent"` - LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` - LuckyGiftPayout int64 `json:"lucky_gift_payout"` - LuckyGiftPayers int64 `json:"lucky_gift_payers"` - LuckyGiftProfit int64 `json:"lucky_gift_profit"` - LuckyGiftPayoutRate float64 `json:"lucky_gift_payout_rate"` - LuckyGiftProfitRate float64 `json:"lucky_gift_profit_rate"` - GameTurnover int64 `json:"game_turnover"` - GamePayout int64 `json:"game_payout"` - GameRefund int64 `json:"game_refund"` - GamePlayers int64 `json:"game_players"` - GameProfit int64 `json:"game_profit"` - GameProfitRate float64 `json:"game_profit_rate"` - SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"` - SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"` - SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"` - SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"` - ARPUUSDMinor int64 `json:"arpu_usd_minor"` - ARPPUUSDMinor int64 `json:"arppu_usd_minor"` - PayerRate float64 `json:"payer_rate"` - PaidConversionRate float64 `json:"paid_conversion_rate"` - RechargeConversionRate float64 `json:"recharge_conversion_rate"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + NewUsers int64 `json:"new_users"` + RegisteredUsers int64 `json:"registered_users"` + Registrations int64 `json:"registrations"` + ActiveUsers int64 `json:"active_users"` + PaidUsers int64 `json:"paid_users"` + RechargeUsers int64 `json:"recharge_users"` + RechargeUSDMinor int64 `json:"recharge_usd_minor"` + NewUserRechargeUSDMinor int64 `json:"new_user_recharge_usd_minor"` + CoinSellerRechargeUSDMinor int64 `json:"coin_seller_recharge_usd_minor"` + CoinSellerStockCoin int64 `json:"coin_seller_stock_coin"` + MifaPayRechargeUSDMinor int64 `json:"mifapay_recharge_usd_minor"` + GoogleRechargeUSDMinor int64 `json:"google_recharge_usd_minor"` + CoinSellerTransferCoin int64 `json:"coin_seller_transfer_coin"` + CoinTotal int64 `json:"coin_total"` + ConsumedCoin int64 `json:"consumed_coin"` + OutputCoin int64 `json:"output_coin"` + ConsumeOutputRatio float64 `json:"consume_output_ratio"` + ConsumeOutputDelta int64 `json:"consume_output_delta"` + PlatformGrantCoin int64 `json:"platform_grant_coin"` + ManualGrantCoin int64 `json:"manual_grant_coin"` + SalaryUSDMinor int64 `json:"salary_usd_minor"` + SalaryTransferCoin int64 `json:"salary_transfer_coin"` + MicOnlineMS int64 `json:"mic_online_ms"` + MicOnlineUsers int64 `json:"mic_online_users"` + AvgMicOnlineMS int64 `json:"avg_mic_online_ms"` + GiftCoinSpent int64 `json:"gift_coin_spent"` + LuckyGiftTurnover int64 `json:"lucky_gift_turnover"` + LuckyGiftPayout int64 `json:"lucky_gift_payout"` + LuckyGiftPayers int64 `json:"lucky_gift_payers"` + LuckyGiftProfit int64 `json:"lucky_gift_profit"` + LuckyGiftPayoutRate float64 `json:"lucky_gift_payout_rate"` + LuckyGiftProfitRate float64 `json:"lucky_gift_profit_rate"` + GameTurnover int64 `json:"game_turnover"` + GamePayout int64 `json:"game_payout"` + GameRefund int64 `json:"game_refund"` + GamePlayers int64 `json:"game_players"` + GameProfit int64 `json:"game_profit"` + GameProfitRate float64 `json:"game_profit_rate"` + SuperLuckyGiftTurnover int64 `json:"super_lucky_gift_turnover"` + SuperLuckyGiftPayout int64 `json:"super_lucky_gift_payout"` + SuperLuckyGiftProfit int64 `json:"super_lucky_gift_profit"` + SuperLuckyGiftProfitRate float64 `json:"super_lucky_gift_profit_rate"` + ARPUUSDMinor int64 `json:"arpu_usd_minor"` + ARPPUUSDMinor int64 `json:"arppu_usd_minor"` + PayerRate float64 `json:"payer_rate"` + PaidConversionRate float64 `json:"paid_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 { @@ -529,23 +550,23 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov applyOverviewDerivedMetrics(&overview) applyOverviewDerivedMetrics(&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 { return Overview{}, err } - overview.Retention = retention + overview.Retention = retentionCube.Total.toRetention() ranking, err := r.queryGameRanking(ctx, app, statTZ, startDay, endDay, query.CountryID, query.RegionID) if err != nil { return Overview{}, err } overview.GameRanking = ranking 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 { return Overview{}, err } 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 { return Overview{}, err } @@ -556,8 +577,16 @@ func (r *Repository) QueryOverview(ctx context.Context, query OverviewQuery) (Ov seriesStartDay = statDayIn(query.SeriesStartMS, 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 查询返回全部指标，避免前端逐日请求。 - 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 { return Overview{}, err } @@ -787,7 +816,7 @@ func (r *Repository) queryRealRoomRobotGiftOverview(ctx context.Context, app, st 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} filter := "" if countryID > 0 { @@ -840,6 +869,13 @@ func (r *Repository) queryDailySeries(ctx context.Context, app, statTZ, startDay for index := range out { 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) if err != nil { return nil, err @@ -895,6 +931,118 @@ func (r *Repository) queryDailyNewPaidUsers(ctx context.Context, app, statTZ, st 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) { if item == nil { return @@ -975,34 +1123,6 @@ func (overview *Overview) applyDeltaRates(previous Overview) { 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) { args := []any{app, normalizeStatTZ(statTZ), startDay, endDay} filter := "" @@ -1039,7 +1159,7 @@ func (r *Repository) queryGameRanking(ctx context.Context, app, statTZ, startDay 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} filter := "" if regionID > 0 { @@ -1081,6 +1201,13 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta if err := rows.Err(); err != nil { 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) if err != nil { return nil, err @@ -1095,7 +1222,7 @@ func (r *Repository) queryCountryBreakdown(ctx context.Context, app, statTZ, sta 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} filter := "" if regionID > 0 { @@ -1137,6 +1264,13 @@ func (r *Repository) queryDailyCountryBreakdown(ctx context.Context, app, statTZ if err := rows.Err(); err != nil { 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) if err != nil { return nil, err @@ -1912,6 +2046,16 @@ func ratio(numerator, denominator int64) float64 { 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 { return []MetricDefinition{ {Metric: "game_pv", Definition: "H5 page_open event_count；同一用户多次打开会重复计数。"}, diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index d44d306b..1ab66809 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -3,11 +3,13 @@ package mysql import ( "context" "encoding/json" + "math" "runtime" "strings" "testing" "time" + "github.com/DATA-DOG/go-sqlmock" "hyapp/internal/testutil/mysqlschema" "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) { ctx := context.Background() _, 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) } } + +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) + } +} diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index ce5474a5..0cfc1062 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -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, 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), - 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`, `CREATE TABLE IF NOT EXISTS stat_user_dimension ( app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,