From dc44b0bf08f931f18f7643f0580c5932a5a97d4a Mon Sep 17 00:00:00 2001 From: zhx Date: Sun, 5 Jul 2026 00:08:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=BC=8F=E6=96=97=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/modules/dashboard/service.go | 41 ++ .../admin/internal/modules/databi/handler.go | 19 + .../admin/internal/modules/databi/routes.go | 1 + .../admin/internal/modules/databi/service.go | 139 +++++ .../internal/app/query_http.go | 40 ++ .../internal/storage/mysql/query.go | 540 ++++++++++++++++++ .../internal/storage/mysql/query_test.go | 105 ++++ 7 files changed, 885 insertions(+) diff --git a/server/admin/internal/modules/dashboard/service.go b/server/admin/internal/modules/dashboard/service.go index 1b45ed71..9b035c9a 100644 --- a/server/admin/internal/modules/dashboard/service.go +++ b/server/admin/internal/modules/dashboard/service.go @@ -44,6 +44,16 @@ type SelfGameStatisticsQuery struct { GameID string } +type AppTrackingFunnelQuery struct { + AppCode string + StatTZ string + StartMS int64 + EndMS int64 + RegionID int64 + RegionIDs []int64 + CountryID int64 +} + type PlatformGrantStatisticsQuery struct { AppCode string StatTZ string @@ -167,6 +177,37 @@ func (s *DashboardService) SelfGameStatisticsOverview(ctx context.Context, query return overview, nil } +func (s *DashboardService) AppTrackingFunnel(ctx context.Context, query AppTrackingFunnelQuery) (map[string]any, error) { + values := url.Values{} + values.Set("app_code", query.AppCode) + if statTZ := strings.TrimSpace(query.StatTZ); statTZ != "" { + values.Set("stat_tz", statTZ) + } + if query.StartMS > 0 { + values.Set("start_ms", strconv.FormatInt(query.StartMS, 10)) + } + if query.EndMS > 0 { + values.Set("end_ms", strconv.FormatInt(query.EndMS, 10)) + } + if query.RegionID > 0 { + values.Set("region_id", strconv.FormatInt(query.RegionID, 10)) + } else if len(query.RegionIDs) > 0 { + parts := make([]string, 0, len(query.RegionIDs)) + for _, regionID := range query.RegionIDs { + if regionID > 0 { + parts = append(parts, strconv.FormatInt(regionID, 10)) + } + } + if len(parts) > 0 { + values.Set("region_ids", strings.Join(parts, ",")) + } + } + if query.CountryID > 0 { + values.Set("country_id", strconv.FormatInt(query.CountryID, 10)) + } + return s.statisticsGET(ctx, "/internal/v1/statistics/app/funnel", values) +} + func (s *DashboardService) PlatformGrantUsers(ctx context.Context, query PlatformGrantStatisticsQuery) (map[string]any, error) { page, err := s.statisticsGET(ctx, "/internal/v1/statistics/platform-grants/users", platformGrantValues(query, false)) if err != nil { diff --git a/server/admin/internal/modules/databi/handler.go b/server/admin/internal/modules/databi/handler.go index 449c2010..21afc31b 100644 --- a/server/admin/internal/modules/databi/handler.go +++ b/server/admin/internal/modules/databi/handler.go @@ -65,6 +65,25 @@ func (h *Handler) Overview(c *gin.Context) { response.OK(c, result) } +func (h *Handler) Funnel(c *gin.Context) { + access, ok := h.moneyAccess(c) + if !ok { + return + } + result, err := h.service.Funnel(c.Request.Context(), access, FunnelQuery{ + StatTZ: c.Query("stat_tz"), + StartMS: queryInt64(c, "start_ms"), + EndMS: queryInt64(c, "end_ms"), + AppCodes: splitCSV(c.Query("app_codes")), + RegionID: queryInt64(c, "region_id"), + }) + if err != nil { + response.ServerError(c, "获取埋点漏斗数据失败") + return + } + response.OK(c, result) +} + func (h *Handler) Kpi(c *gin.Context) { access, ok := h.moneyAccess(c) if !ok { diff --git a/server/admin/internal/modules/databi/routes.go b/server/admin/internal/modules/databi/routes.go index 90eb0c30..e6041a0c 100644 --- a/server/admin/internal/modules/databi/routes.go +++ b/server/admin/internal/modules/databi/routes.go @@ -9,6 +9,7 @@ import ( func RegisterRoutes(protected *gin.RouterGroup, h *Handler) { protected.GET("/admin/databi/social/master", middleware.RequirePermission("overview:view"), h.Master) protected.GET("/admin/databi/social/overview", middleware.RequirePermission("overview:view"), h.Overview) + protected.GET("/admin/databi/social/funnel", middleware.RequirePermission("overview:view"), h.Funnel) protected.GET("/admin/databi/social/kpi", middleware.RequirePermission("overview:view"), h.Kpi) protected.GET("/admin/databi/social/kpi-targets", middleware.RequirePermission("overview:view"), h.ListKpiTargets) protected.PUT("/admin/databi/social/kpi-targets", middleware.RequirePermission(kpiManagePermission), h.ReplaceKpiTargets) diff --git a/server/admin/internal/modules/databi/service.go b/server/admin/internal/modules/databi/service.go index 26683c64..ebdd733f 100644 --- a/server/admin/internal/modules/databi/service.go +++ b/server/admin/internal/modules/databi/service.go @@ -26,6 +26,12 @@ const ( overviewConcurrency = 4 ) +var appTrackingFunnelSupportedApps = map[string]struct{}{ + "fami": {}, + "huwaa": {}, + "lalu": {}, +} + // LegacyRegionCatalogSource 由 payment.MongoMoneyRegionSource 实现； // 社交 BI 与财务范围共用同一套 legacy 区域目录与合成 region_id 口径。 type LegacyRegionCatalogSource interface { @@ -266,6 +272,32 @@ type OverviewResult struct { Access AccessInfo `json:"access"` } +type FunnelQuery struct { + StatTZ string + StartMS int64 + EndMS int64 + AppCodes []string + RegionID int64 +} + +type AppFunnel struct { + AppCode string `json:"app_code"` + AppName string `json:"app_name"` + Kind string `json:"kind"` + Restricted bool `json:"restricted"` + AllowedRegionIDs []int64 `json:"allowed_region_ids,omitempty"` + Steps []map[string]any `json:"steps"` + Cohorts []map[string]any `json:"cohorts"` + Totals map[string]any `json:"totals"` + UpdatedAtMS int64 `json:"updated_at_ms"` + Error string `json:"error,omitempty"` +} + +type FunnelResult struct { + Apps []AppFunnel `json:"apps"` + Access AccessInfo `json:"access"` +} + // Overview 服务端并发聚合各 App 的统计总览：hyapp App 走 statistics-service， // legacy App 走 dashboard 外部源；国家行按区域目录卷积出大区行，并按用户数据范围裁剪。 func (s *Service) Overview(ctx context.Context, access repository.MoneyAccess, query OverviewQuery) (*OverviewResult, error) { @@ -298,6 +330,113 @@ func (s *Service) Overview(ctx context.Context, access repository.MoneyAccess, q }, nil } +func (s *Service) Funnel(ctx context.Context, access repository.MoneyAccess, query FunnelQuery) (*FunnelResult, error) { + apps, err := s.listAllowedApps(ctx, query.AppCodes, access) + if err != nil { + return nil, err + } + apps = filterAppTrackingFunnelApps(apps) + catalog, err := s.regionCatalog(ctx, apps) + if err != nil { + return nil, err + } + + results := make([]AppFunnel, len(apps)) + var wg sync.WaitGroup + semaphore := make(chan struct{}, overviewConcurrency) + for index, app := range apps { + wg.Add(1) + go func(index int, app AppInfo) { + defer wg.Done() + semaphore <- struct{}{} + defer func() { <-semaphore }() + results[index] = s.appFunnel(ctx, app, catalog[app.AppCode], access, query) + }(index, app) + } + wg.Wait() + + return &FunnelResult{ + Apps: results, + Access: AccessInfo{All: access.All, Scopes: scopeInfos(access.Scopes)}, + }, nil +} + +func (s *Service) appFunnel(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query FunnelQuery) AppFunnel { + out := AppFunnel{ + AppCode: app.AppCode, + AppName: app.AppName, + Kind: app.Kind, + Steps: []map[string]any{}, + Cohorts: []map[string]any{}, + Totals: map[string]any{}, + } + if app.Kind != appKindHyapp { + out.Error = "该 App 暂未接入 App 埋点漏斗" + return out + } + if !supportsAppTrackingFunnel(app.AppCode) { + out.Error = "该 App 暂未接入 App 埋点漏斗" + return out + } + allowAll, allowedIDs := allowedRegions(access, app.AppCode) + allowedIDs = normalizeRegionIDs(regions, allowedIDs) + if !allowAll && len(allowedIDs) == 0 { + out.Error = "没有该 App 的数据权限" + return out + } + if query.RegionID > 0 && !access.Allows(app.AppCode, query.RegionID) { + out.Error = "没有该区域的数据权限" + return out + } + if query.RegionID > 0 && !regionInCatalog(regions, query.RegionID) { + out.Error = "该 App 未配置此区域" + return out + } + restricted := !allowAll && query.RegionID == 0 + if restricted { + out.Restricted = true + out.AllowedRegionIDs = allowedIDs + } + regionIDs := []int64(nil) + if restricted { + regionIDs = allowedIDs + } + funnel, err := s.dashboards.AppTrackingFunnel(ctx, dashboard.AppTrackingFunnelQuery{ + AppCode: app.AppCode, + StatTZ: query.StatTZ, + StartMS: query.StartMS, + EndMS: query.EndMS, + RegionID: query.RegionID, + RegionIDs: regionIDs, + }) + if err != nil { + out.Error = fmt.Sprintf("获取埋点漏斗失败: %v", err) + return out + } + out.Steps = anyRowSlice(funnel["steps"]) + out.Cohorts = anyRowSlice(funnel["cohorts"]) + if totals, ok := funnel["totals"].(map[string]any); ok { + out.Totals = totals + } + out.UpdatedAtMS = time.Now().UnixMilli() + return out +} + +func filterAppTrackingFunnelApps(apps []AppInfo) []AppInfo { + out := make([]AppInfo, 0, len(apps)) + for _, app := range apps { + if app.Kind == appKindHyapp && supportsAppTrackingFunnel(app.AppCode) { + out = append(out, app) + } + } + return out +} + +func supportsAppTrackingFunnel(appCode string) bool { + _, ok := appTrackingFunnelSupportedApps[appctx.Normalize(appCode)] + return ok +} + func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query OverviewQuery) AppOverview { out := AppOverview{ AppCode: app.AppCode, diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go index 4d39ace5..e6adc2e6 100644 --- a/services/statistics-service/internal/app/query_http.go +++ b/services/statistics-service/internal/app/query_http.go @@ -35,6 +35,7 @@ func newQueryHTTPServer(addr string, repo *mysqlstorage.Repository) (*queryHTTPS mux.HandleFunc("/internal/v1/statistics/platform-grants/users", s.platformGrantUsers) mux.HandleFunc("/internal/v1/statistics/platform-grants/records", s.platformGrantRecords) mux.HandleFunc("/internal/v1/statistics/app/events", s.appEvents) + mux.HandleFunc("/internal/v1/statistics/app/funnel", s.appFunnel) mux.HandleFunc("/internal/v1/statistics/self-game/events", s.selfGameEvents) mux.HandleFunc("/internal/v1/statistics/self-games/overview", s.selfGameOverview) s.server = &http.Server{Handler: mux, ReadHeaderTimeout: 2 * time.Second} @@ -143,6 +144,33 @@ func (s *queryHTTPServer) platformGrantRecords(w http.ResponseWriter, r *http.Re writeJSON(w, http.StatusOK, page) } +func (s *queryHTTPServer) appFunnel(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + startMS := parseInt64(query.Get("start_ms")) + endMS := parseInt64(query.Get("end_ms")) + if startMS <= 0 { + now := time.Now().UTC() + startMS = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC).UnixMilli() + } + if endMS <= startMS { + endMS = startMS + 24*time.Hour.Milliseconds() + } + funnel, err := s.repo.QueryAppTrackingFunnel(r.Context(), mysqlstorage.AppTrackingFunnelQuery{ + AppCode: query.Get("app_code"), + StatTZ: query.Get("stat_tz"), + StartMS: startMS, + EndMS: endMS, + CountryID: parseInt64(query.Get("country_id")), + RegionID: parseInt64(query.Get("region_id")), + RegionIDs: parseInt64CSV(query.Get("region_ids")), + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, funnel) +} + type selfGameH5EventRequest struct { AppCode string `json:"app_code"` EventID string `json:"event_id"` @@ -334,6 +362,18 @@ func parsePositiveInt(value string) int { return parsed } +func parseInt64CSV(value string) []int64 { + parts := strings.Split(value, ",") + out := make([]int64, 0, len(parts)) + for _, part := range parts { + parsed := parseInt64(part) + if parsed > 0 { + out = append(out, parsed) + } + } + return out +} + func writeJSON(w http.ResponseWriter, status int, payload any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index 7ccc23e6..abc736aa 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -323,6 +323,80 @@ type PlatformGrantCoinQuery struct { PageSize int } +type AppTrackingFunnelQuery struct { + AppCode string + StatTZ string + StartMS int64 + EndMS int64 + CountryID int64 + RegionID int64 + RegionIDs []int64 +} + +type AppTrackingFunnel struct { + AppCode string `json:"app_code"` + StatTZ string `json:"stat_tz"` + StartMS int64 `json:"start_ms"` + EndMS int64 `json:"end_ms"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + RegionIDs []int64 `json:"region_ids,omitempty"` + Steps []AppTrackingFunnelStep `json:"steps"` + Cohorts []AppTrackingFunnelCohort `json:"cohorts"` + Totals AppTrackingFunnelTotals `json:"totals"` +} + +type AppTrackingFunnelStep struct { + Key string `json:"key"` + EventName string `json:"event_name"` + Label string `json:"label"` + EventCount int64 `json:"event_count"` + UserCount int64 `json:"user_count"` + DeviceCount int64 `json:"device_count"` + OverallConversionRate float64 `json:"overall_conversion_rate"` + PreviousConversionRate float64 `json:"previous_conversion_rate"` + DropoffUsers int64 `json:"dropoff_users"` + IsFailure bool `json:"is_failure,omitempty"` +} + +type AppTrackingFunnelCohort struct { + Dimension string `json:"dimension"` + DimensionLabel string `json:"dimension_label"` + Value string `json:"value"` + Label string `json:"label"` + BaseUsers int64 `json:"base_users"` + D1RetentionUsers int64 `json:"d1_retention_users"` + D1RetentionRate float64 `json:"d1_retention_rate"` + Steps []AppTrackingCohortStepCount `json:"steps"` +} + +type AppTrackingCohortStepCount struct { + EventName string `json:"event_name"` + Label string `json:"label"` + UserCount int64 `json:"user_count"` +} + +type AppTrackingFunnelTotals struct { + LoginStartUsers int64 `json:"login_start_users"` + LoginSuccessUsers int64 `json:"login_success_users"` + ProfileCompleteUsers int64 `json:"profile_complete_users"` + RoomImpressionUsers int64 `json:"room_impression_users"` + RoomClickUsers int64 `json:"room_click_users"` + RoomJoinSuccessUsers int64 `json:"room_join_success_users"` + RoomJoinFailUsers int64 `json:"room_join_fail_users"` + Stay30sUsers int64 `json:"stay_30s_users"` + Stay3mUsers int64 `json:"stay_3m_users"` + Stay10mUsers int64 `json:"stay_10m_users"` + SendMessageUsers int64 `json:"send_message_users"` + FollowHostUsers int64 `json:"follow_host_users"` + AddFriendUsers int64 `json:"add_friend_users"` + CheckInUsers int64 `json:"check_in_users"` + PushPermissionUsers int64 `json:"push_permission_users"` + D1RetentionBaseUsers int64 `json:"d1_retention_base_users"` + D1RetentionUsers int64 `json:"d1_retention_users"` + D1RetentionRate float64 `json:"d1_retention_rate"` +} + type PlatformGrantCoinUser struct { UserID int64 `json:"user_id"` CountryID int64 `json:"country_id"` @@ -654,6 +728,45 @@ func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOv return overview, nil } +func (r *Repository) QueryAppTrackingFunnel(ctx context.Context, query AppTrackingFunnelQuery) (AppTrackingFunnel, error) { + startMS, endMS := normalizeRange(query.StartMS, query.EndMS) + app := appcode.Normalize(query.AppCode) + if app == "" { + app = "lalu" + } + statTZ := normalizeStatTZ(query.StatTZ) + out := AppTrackingFunnel{ + AppCode: app, + StatTZ: statTZ, + StartMS: startMS, + EndMS: endMS, + CountryID: query.CountryID, + RegionID: query.RegionID, + RegionIDs: normalizeTrackingRegionIDs(query.RegionIDs), + Steps: []AppTrackingFunnelStep{}, + Cohorts: []AppTrackingFunnelCohort{}, + } + steps, err := r.queryAppTrackingFunnelSteps(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs) + if err != nil { + return AppTrackingFunnel{}, err + } + out.Steps = buildAppTrackingFunnelSteps(steps) + out.Totals = appTrackingTotals(out.Steps) + d1Base, d1Users, err := r.queryAppTrackingD1Total(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs) + if err != nil { + return AppTrackingFunnel{}, err + } + out.Totals.D1RetentionBaseUsers = d1Base + out.Totals.D1RetentionUsers = d1Users + out.Totals.D1RetentionRate = ratio(d1Users, d1Base) + cohorts, err := r.queryAppTrackingD1Cohorts(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs) + if err != nil { + return AppTrackingFunnel{}, err + } + out.Cohorts = cohorts + return out, nil +} + func (r *Repository) QueryPlatformGrantCoinUsers(ctx context.Context, query PlatformGrantCoinQuery) (PlatformGrantCoinUserPage, error) { app, statTZ, startDay, endDay, page, pageSize, offset := normalizePlatformGrantCoinQuery(query) filter, args := platformGrantCoinFilter(app, statTZ, startDay, endDay, query.CountryID, query.RegionID, 0, query.EventType) @@ -1928,6 +2041,433 @@ func buildSelfGameFunnel(events []SelfGameEventStat) ([]SelfGameFunnelStep, Self } } +type appTrackingStepDef struct { + Key string + EventName string + Label string + IsFailure bool +} + +var appTrackingStepDefs = []appTrackingStepDef{ + {Key: "login_start", EventName: "login_start", Label: "登录开始"}, + {Key: "login_success", EventName: "login_success", Label: "登录成功"}, + {Key: "profile_start", EventName: "profile_start", Label: "资料开始"}, + {Key: "profile_complete", EventName: "profile_complete", Label: "资料完成"}, + {Key: "home_room_impression", EventName: "home_room_impression", Label: "首页房间曝光"}, + {Key: "home_room_click", EventName: "home_room_click", Label: "首页房间点击"}, + {Key: "room_join_success", EventName: "room_join_success", Label: "进房成功"}, + {Key: "room_join_fail", EventName: "room_join_fail", Label: "进房失败", IsFailure: true}, + {Key: "stay_30s", EventName: "stay_30s", Label: "停留 30 秒"}, + {Key: "stay_3m", EventName: "stay_3m", Label: "停留 3 分钟"}, + {Key: "stay_10m", EventName: "stay_10m", Label: "停留 10 分钟"}, + {Key: "mic_up", EventName: "mic_up", Label: "上麦"}, + {Key: "send_message", EventName: "send_message", Label: "发消息"}, + {Key: "follow_host", EventName: "follow_host", Label: "关注主播"}, + {Key: "add_friend", EventName: "add_friend", Label: "加好友"}, + {Key: "check_in", EventName: "check_in", Label: "签到"}, + {Key: "push_permission", EventName: "push_permission", Label: "Push 授权"}, +} + +type appTrackingCohortDef struct { + Key string + Label string +} + +var appTrackingCohortDefs = []appTrackingCohortDef{ + {Key: "country", Label: "国家"}, + {Key: "language", Label: "语言"}, + {Key: "channel", Label: "渠道"}, + {Key: "login_method", Label: "登录方式"}, + {Key: "first_room_stay", Label: "首房停留时长"}, +} + +type appTrackingStepAggregate struct { + EventName string + EventCount int64 + UserCount int64 + DeviceCount int64 +} + +func (r *Repository) queryAppTrackingFunnelSteps(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) ([]appTrackingStepAggregate, error) { + filter, args := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) + names := appTrackingEventNames() + placeholders := sqlPlaceholders(len(names)) + for _, name := range names { + args = append(args, name) + } + rows, err := r.db.QueryContext(ctx, ` + SELECT event_name, COUNT(*), COUNT(DISTINCT `+appTrackingIdentitySQL("")+`), COUNT(DISTINCT NULLIF(device_id, '')) + FROM app_tracking_events + WHERE `+filter+` AND event_name IN (`+placeholders+`) + GROUP BY event_name`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []appTrackingStepAggregate{} + for rows.Next() { + var item appTrackingStepAggregate + if err := rows.Scan(&item.EventName, &item.EventCount, &item.UserCount, &item.DeviceCount); err != nil { + return nil, err + } + out = append(out, item) + } + return out, rows.Err() +} + +func buildAppTrackingFunnelSteps(aggregates []appTrackingStepAggregate) []AppTrackingFunnelStep { + byEvent := map[string]appTrackingStepAggregate{} + for _, item := range aggregates { + byEvent[item.EventName] = item + } + out := make([]AppTrackingFunnelStep, 0, len(appTrackingStepDefs)) + var baseUsers int64 + var previousUsers int64 + for index, def := range appTrackingStepDefs { + aggregate := byEvent[def.EventName] + if index == 0 { + baseUsers = aggregate.UserCount + } + step := AppTrackingFunnelStep{ + Key: def.Key, + EventName: def.EventName, + Label: def.Label, + EventCount: aggregate.EventCount, + UserCount: aggregate.UserCount, + DeviceCount: aggregate.DeviceCount, + OverallConversionRate: ratio(aggregate.UserCount, baseUsers), + PreviousConversionRate: ratio(aggregate.UserCount, previousUsers), + IsFailure: def.IsFailure, + } + if previousUsers > aggregate.UserCount { + step.DropoffUsers = previousUsers - aggregate.UserCount + } + out = append(out, step) + previousUsers = aggregate.UserCount + } + return out +} + +func appTrackingTotals(steps []AppTrackingFunnelStep) AppTrackingFunnelTotals { + byEvent := map[string]int64{} + for _, step := range steps { + byEvent[step.EventName] = step.UserCount + } + return AppTrackingFunnelTotals{ + LoginStartUsers: byEvent["login_start"], + LoginSuccessUsers: byEvent["login_success"], + ProfileCompleteUsers: byEvent["profile_complete"], + RoomImpressionUsers: byEvent["home_room_impression"], + RoomClickUsers: byEvent["home_room_click"], + RoomJoinSuccessUsers: byEvent["room_join_success"], + RoomJoinFailUsers: byEvent["room_join_fail"], + Stay30sUsers: byEvent["stay_30s"], + Stay3mUsers: byEvent["stay_3m"], + Stay10mUsers: byEvent["stay_10m"], + SendMessageUsers: byEvent["send_message"], + FollowHostUsers: byEvent["follow_host"], + AddFriendUsers: byEvent["add_friend"], + CheckInUsers: byEvent["check_in"], + PushPermissionUsers: byEvent["push_permission"], + } +} + +func (r *Repository) queryAppTrackingD1Total(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (int64, int64, error) { + filter, args := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) + args = append(args, app) + row := r.db.QueryRowContext(ctx, ` + WITH base AS ( + SELECT DISTINCT stat_day, `+appTrackingIdentitySQL("")+` AS identity + FROM app_tracking_events + WHERE `+filter+` AND event_name = 'login_success' + ) + SELECT COUNT(*), + COALESCE(SUM(CASE WHEN EXISTS ( + SELECT 1 FROM app_tracking_events n + WHERE n.app_code = ? AND n.stat_day = DATE_ADD(base.stat_day, INTERVAL 1 DAY) + AND `+appTrackingIdentitySQL("n")+` = base.identity + LIMIT 1 + ) THEN 1 ELSE 0 END),0) + FROM base`, args...) + var baseUsers, retainedUsers int64 + if err := row.Scan(&baseUsers, &retainedUsers); err != nil { + return 0, 0, err + } + return baseUsers, retainedUsers, nil +} + +func (r *Repository) queryAppTrackingD1Cohorts(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) ([]AppTrackingFunnelCohort, error) { + out := []AppTrackingFunnelCohort{} + for _, def := range appTrackingCohortDefs { + rows, err := r.queryAppTrackingD1CohortDimension(ctx, app, startMS, endMS, countryID, regionID, regionIDs, def) + if err != nil { + return nil, err + } + stepCounts, err := r.queryAppTrackingCohortSteps(ctx, app, startMS, endMS, countryID, regionID, regionIDs, def) + if err != nil { + return nil, err + } + for index := range rows { + rows[index].Steps = buildAppTrackingCohortSteps(stepCounts[rows[index].Value]) + } + out = append(out, rows...) + } + return out, nil +} + +func (r *Repository) queryAppTrackingD1CohortDimension(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64, def appTrackingCohortDef) ([]AppTrackingFunnelCohort, error) { + filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) + sourceSQL := appTrackingD1CohortSourceSQL(def.Key, filter) + args := appTrackingCohortSourceArgs(def.Key, filterArgs) + args = append(args, app) + rows, err := r.db.QueryContext(ctx, ` + WITH base AS (`+sourceSQL+`) + SELECT value, COUNT(*), + COALESCE(SUM(CASE WHEN EXISTS ( + SELECT 1 FROM app_tracking_events n + WHERE n.app_code = ? AND n.stat_day = DATE_ADD(base.stat_day, INTERVAL 1 DAY) + AND `+appTrackingIdentitySQL("n")+` = base.identity + LIMIT 1 + ) THEN 1 ELSE 0 END),0) + FROM base + GROUP BY value + ORDER BY COUNT(*) DESC, value ASC + LIMIT 50`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := []AppTrackingFunnelCohort{} + for rows.Next() { + var value sql.NullString + var baseUsers, retainedUsers int64 + if err := rows.Scan(&value, &baseUsers, &retainedUsers); err != nil { + return nil, err + } + cohortValue := normalizeCohortValue(value.String) + out = append(out, AppTrackingFunnelCohort{ + Dimension: def.Key, + DimensionLabel: def.Label, + Value: cohortValue, + Label: cohortValue, + BaseUsers: baseUsers, + D1RetentionUsers: retainedUsers, + D1RetentionRate: ratio(retainedUsers, baseUsers), + Steps: []AppTrackingCohortStepCount{}, + }) + } + return out, rows.Err() +} + +func (r *Repository) queryAppTrackingCohortSteps(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64, def appTrackingCohortDef) (map[string]map[string]int64, error) { + filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) + cohortSQL := appTrackingStepCohortSourceSQL(def.Key, filter) + args := appTrackingCohortSourceArgs(def.Key, filterArgs) + eventFilter, eventArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs) + args = append(args, eventArgs...) + names := appTrackingEventNames() + placeholders := sqlPlaceholders(len(names)) + for _, name := range names { + args = append(args, name) + } + rows, err := r.db.QueryContext(ctx, ` + WITH cohort AS (`+cohortSQL+`), + events AS ( + SELECT DISTINCT `+appTrackingIdentitySQL("")+` AS identity, event_name + FROM app_tracking_events + WHERE `+eventFilter+` AND event_name IN (`+placeholders+`) + ) + SELECT cohort.value, events.event_name, COUNT(DISTINCT cohort.identity) + FROM cohort + JOIN events ON events.identity = cohort.identity + GROUP BY cohort.value, events.event_name`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]map[string]int64{} + for rows.Next() { + var value sql.NullString + var eventName string + var userCount int64 + if err := rows.Scan(&value, &eventName, &userCount); err != nil { + return nil, err + } + cohortValue := normalizeCohortValue(value.String) + if out[cohortValue] == nil { + out[cohortValue] = map[string]int64{} + } + out[cohortValue][eventName] = userCount + } + return out, rows.Err() +} + +func buildAppTrackingCohortSteps(counts map[string]int64) []AppTrackingCohortStepCount { + out := make([]AppTrackingCohortStepCount, 0, len(appTrackingStepDefs)) + for _, def := range appTrackingStepDefs { + out = append(out, AppTrackingCohortStepCount{ + EventName: def.EventName, + Label: def.Label, + UserCount: counts[def.EventName], + }) + } + return out +} + +func appTrackingD1CohortSourceSQL(dimension string, filter string) string { + if dimension == "first_room_stay" { + return ` + SELECT base.stat_day, base.identity, ` + appTrackingStayBucketSQL("stay.stay_rank") + ` AS value + FROM ( + SELECT DISTINCT stat_day, ` + appTrackingIdentitySQL("") + ` AS identity + FROM app_tracking_events + WHERE ` + filter + ` AND event_name = 'login_success' + ) base + LEFT JOIN ( + SELECT stat_day, ` + appTrackingIdentitySQL("") + ` AS identity, + MAX(CASE event_name WHEN 'stay_10m' THEN 3 WHEN 'stay_3m' THEN 2 WHEN 'stay_30s' THEN 1 ELSE 0 END) AS stay_rank + FROM app_tracking_events + WHERE ` + filter + ` AND event_name IN ('stay_30s', 'stay_3m', 'stay_10m') + GROUP BY stat_day, identity + ) stay ON stay.stat_day = base.stat_day AND stay.identity = base.identity` + } + return ` + SELECT DISTINCT stat_day, ` + appTrackingIdentitySQL("") + ` AS identity, ` + appTrackingCohortValueSQL(dimension, "") + ` AS value + FROM app_tracking_events + WHERE ` + filter + ` AND event_name = 'login_success'` +} + +func appTrackingStepCohortSourceSQL(dimension string, filter string) string { + if dimension == "first_room_stay" { + return ` + SELECT DISTINCT base.identity, ` + appTrackingStayBucketSQL("stay.stay_rank") + ` AS value + FROM ( + SELECT DISTINCT stat_day, ` + appTrackingIdentitySQL("") + ` AS identity + FROM app_tracking_events + WHERE ` + filter + ` AND event_name = 'login_success' + ) base + LEFT JOIN ( + SELECT stat_day, ` + appTrackingIdentitySQL("") + ` AS identity, + MAX(CASE event_name WHEN 'stay_10m' THEN 3 WHEN 'stay_3m' THEN 2 WHEN 'stay_30s' THEN 1 ELSE 0 END) AS stay_rank + FROM app_tracking_events + WHERE ` + filter + ` AND event_name IN ('stay_30s', 'stay_3m', 'stay_10m') + GROUP BY stat_day, identity + ) stay ON stay.stat_day = base.stat_day AND stay.identity = base.identity` + } + return ` + SELECT DISTINCT ` + appTrackingIdentitySQL("") + ` AS identity, ` + appTrackingCohortValueSQL(dimension, "") + ` AS value + FROM app_tracking_events + WHERE ` + filter + ` AND event_name = 'login_success'` +} + +func appTrackingCohortSourceArgs(dimension string, filterArgs []any) []any { + args := append([]any{}, filterArgs...) + if dimension == "first_room_stay" { + args = append(args, filterArgs...) + } + return args +} + +func appTrackingStayBucketSQL(rankExpr string) string { + return "CASE COALESCE(" + rankExpr + ",0) WHEN 3 THEN '10m+' WHEN 2 THEN '3m-10m' WHEN 1 THEN '30s-3m' ELSE '<30s' END" +} + +func appTrackingCohortValueSQL(dimension string, alias string) string { + prefix := "" + if alias != "" { + prefix = alias + "." + } + jsonValue := func(path string) string { + return "NULLIF(TRIM(JSON_UNQUOTE(JSON_EXTRACT(" + prefix + "properties_json, '$." + path + "'))), '')" + } + switch dimension { + case "country": + return "COALESCE(" + jsonValue("country") + ", IF(" + prefix + "country_id > 0, CAST(" + prefix + "country_id AS CHAR), 'unknown'))" + case "language": + return "COALESCE(NULLIF(TRIM(" + prefix + "language), ''), " + jsonValue("language") + ", 'unknown')" + case "channel": + return "COALESCE(" + jsonValue("channel") + ", " + jsonValue("utm_source") + ", 'unknown')" + case "login_method": + return "COALESCE(" + jsonValue("login_method") + ", " + jsonValue("login_provider") + ", " + jsonValue("method") + ", " + jsonValue("provider") + ", 'unknown')" + default: + return "'unknown'" + } +} + +func appTrackingFilter(app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (string, []any) { + filter := "app_code = ? AND occurred_at_ms >= ? AND occurred_at_ms < ?" + args := []any{app, startMS, endMS} + if countryID > 0 { + filter += " AND country_id = ?" + args = append(args, countryID) + } + if regionID > 0 { + filter += " AND region_id = ?" + args = append(args, regionID) + return filter, args + } + regionIDs = normalizeTrackingRegionIDs(regionIDs) + if len(regionIDs) > 0 { + filter += " AND region_id IN (" + sqlPlaceholders(len(regionIDs)) + ")" + for _, id := range regionIDs { + args = append(args, id) + } + } + return filter, args +} + +func normalizeTrackingRegionIDs(regionIDs []int64) []int64 { + out := make([]int64, 0, len(regionIDs)) + seen := map[int64]struct{}{} + for _, id := range regionIDs { + if id <= 0 { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + out = append(out, id) + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out +} + +func appTrackingIdentitySQL(alias string) string { + prefix := "" + if alias != "" { + prefix = alias + "." + } + return "CASE WHEN " + prefix + "user_id > 0 THEN CONCAT('u:', " + prefix + "user_id) ELSE CONCAT('d:', " + prefix + "device_id) END" +} + +func appTrackingEventNames() []string { + out := make([]string, 0, len(appTrackingStepDefs)) + for _, def := range appTrackingStepDefs { + out = append(out, def.EventName) + } + return out +} + +func sqlPlaceholders(count int) string { + if count <= 0 { + return "NULL" + } + parts := make([]string, count) + for index := range parts { + parts[index] = "?" + } + return strings.Join(parts, ",") +} + +func normalizeCohortValue(value string) string { + value = strings.TrimSpace(value) + if value == "" || strings.EqualFold(value, "null") { + return "unknown" + } + return value +} + func normalizeRange(startMS, endMS int64) (int64, int64) { if startMS <= 0 { now := time.Now().UTC() diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index 1ab66809..41c03254 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "math" "runtime" + "strconv" "strings" "testing" "time" @@ -153,6 +154,110 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) { } } +func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(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_app_tracking_funnel_test", + }) + repository, err := Open(ctx, statsSchema.DSN) + if err != nil { + t.Fatalf("open repository: %v", err) + } + t.Cleanup(func() { _ = repository.Close() }) + + start := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + event := func(userID int64, deviceID string, eventName string, offset time.Duration, properties string) AppTrackingEvent { + return AppTrackingEvent{ + AppCode: "lalu", + EventID: "evt-" + strconv.FormatInt(userID, 10) + "-" + deviceID + "-" + eventName + "-" + strconv.FormatInt(int64(offset/time.Second), 10), + EventName: eventName, + UserID: userID, + DeviceID: deviceID, + Language: "en-US", + CountryID: 86, + RegionID: 210, + Properties: json.RawMessage(properties), + OccurredAtMS: start.Add(offset).UnixMilli(), + } + } + user1Props := `{"country":"US","channel":"ads","login_method":"google"}` + user2Props := `{"country":"BR","channel":"organic","login_method":"phone"}` + events := []AppTrackingEvent{ + event(1, "dev-1", "login_start", 0, user1Props), + event(1, "dev-1", "login_success", time.Second, user1Props), + event(1, "dev-1", "profile_start", 2*time.Second, user1Props), + event(1, "dev-1", "profile_complete", 3*time.Second, user1Props), + event(1, "dev-1", "home_room_impression", 4*time.Second, user1Props), + event(1, "dev-1", "home_room_click", 5*time.Second, user1Props), + event(1, "dev-1", "room_join_success", 6*time.Second, user1Props), + event(1, "dev-1", "stay_30s", 30*time.Second, user1Props), + event(1, "dev-1", "stay_3m", 3*time.Minute, user1Props), + event(1, "dev-1", "send_message", 4*time.Minute, user1Props), + event(1, "dev-1", "follow_host", 5*time.Minute, user1Props), + event(1, "dev-1", "push_permission", 6*time.Minute, user1Props), + event(1, "dev-1", "send_message", 24*time.Hour+time.Second, user1Props), + event(2, "dev-2", "login_start", 0, user2Props), + event(2, "dev-2", "login_success", time.Second, user2Props), + event(2, "dev-2", "home_room_impression", 2*time.Second, user2Props), + event(2, "dev-2", "home_room_click", 3*time.Second, user2Props), + event(2, "dev-2", "room_join_fail", 4*time.Second, user2Props), + } + excluded := event(3, "dev-3", "login_start", 0, user1Props) + excluded.EventID = "evt-excluded-region-login-start" + excluded.RegionID = 999 + events = append(events, excluded) + if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), events); err != nil { + t.Fatalf("consume app tracking events: %v", err) + } + + funnel, err := repository.QueryAppTrackingFunnel(ctx, AppTrackingFunnelQuery{ + AppCode: "lalu", + StartMS: start.UnixMilli(), + EndMS: start.Add(24 * time.Hour).UnixMilli(), + RegionIDs: []int64{210}, + }) + if err != nil { + t.Fatalf("query app tracking funnel: %v", err) + } + if funnel.Totals.LoginStartUsers != 2 || funnel.Totals.LoginSuccessUsers != 2 || funnel.Totals.RoomJoinSuccessUsers != 1 || funnel.Totals.RoomJoinFailUsers != 1 { + t.Fatalf("unexpected funnel totals: %+v", funnel.Totals) + } + if funnel.Totals.D1RetentionBaseUsers != 2 || funnel.Totals.D1RetentionUsers != 1 || math.Abs(funnel.Totals.D1RetentionRate-0.5) > 0.0001 { + t.Fatalf("unexpected d1 totals: %+v", funnel.Totals) + } + + stepByEvent := map[string]AppTrackingFunnelStep{} + for _, step := range funnel.Steps { + stepByEvent[step.EventName] = step + } + if stepByEvent["login_start"].UserCount != 2 || stepByEvent["profile_complete"].UserCount != 1 || stepByEvent["stay_3m"].UserCount != 1 { + t.Fatalf("unexpected step counts: %+v", stepByEvent) + } + if !stepByEvent["room_join_fail"].IsFailure { + t.Fatalf("room_join_fail must be marked as failure: %+v", stepByEvent["room_join_fail"]) + } + + cohort := func(dimension, value string) AppTrackingFunnelCohort { + for _, item := range funnel.Cohorts { + if item.Dimension == dimension && item.Value == value { + return item + } + } + t.Fatalf("missing cohort %s=%s in %+v", dimension, value, funnel.Cohorts) + return AppTrackingFunnelCohort{} + } + us := cohort("country", "US") + if us.BaseUsers != 1 || us.D1RetentionUsers != 1 || math.Abs(us.D1RetentionRate-1) > 0.0001 { + t.Fatalf("unexpected US cohort: %+v", us) + } + stay := cohort("first_room_stay", "3m-10m") + if stay.BaseUsers != 1 || stay.D1RetentionUsers != 1 { + t.Fatalf("unexpected stay cohort: %+v", stay) + } +} + func TestConsumeAppTrackingEventsRejectsInvalidProperties(t *testing.T) { ctx := context.Background() _, file, _, _ := runtime.Caller(0)