fix: optimize app tracking funnel aggregation

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

@@ -48,6 +48,7 @@ CREATE TABLE IF NOT EXISTS app_tracking_events (
   KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms),
   KEY idx_app_tracking_device (app_code, device_id, occurred_at_ms),
   KEY idx_app_tracking_event_window (app_code, event_name, occurred_at_ms, identity_key, device_id),
+  KEY idx_app_tracking_time_window (app_code, occurred_at_ms, event_name, identity_key, device_id),
   KEY idx_app_tracking_retention_identity (app_code, stat_day, identity_key)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='App 埋点原始明细表';
 

services/statistics-service/internal/storage/mysql/query.go

@@ -752,17 +752,13 @@ func (r *Repository) QueryAppTrackingFunnel(ctx context.Context, query AppTracki
 	}
 	out.Steps = buildAppTrackingFunnelSteps(steps)
 	out.Totals = appTrackingTotals(out.Steps)
-	d1Base, d1Users, err := r.queryAppTrackingD1Total(ctx, app, startMS, endMS, query.CountryID, query.RegionID, out.RegionIDs)
+	d1Base, d1Users, cohorts, err := r.queryAppTrackingD1Cohorts(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
 }
@@ -2097,7 +2093,7 @@ func (r *Repository) queryAppTrackingFunnelSteps(ctx context.Context, app string
 	}
 	rows, err := r.db.QueryContext(ctx, `
 		SELECT event_name, COUNT(*), COUNT(DISTINCT `+appTrackingIdentitySQL("")+`), COUNT(DISTINCT NULLIF(device_id, ''))
-		FROM app_tracking_events
+		FROM app_tracking_events FORCE INDEX (idx_app_tracking_time_window)
 		WHERE `+filter+` AND event_name IN (`+placeholders+`)
 		GROUP BY event_name`, args...)
 	if err != nil {
@@ -2172,97 +2168,119 @@ func appTrackingTotals(steps []AppTrackingFunnelStep) AppTrackingFunnelTotals {
 	}
 }
 
-func (r *Repository) queryAppTrackingD1Total(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (int64, int64, error) {
+func (r *Repository) queryAppTrackingD1Cohorts(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (int64, int64, []AppTrackingFunnelCohort, error) {
-	filter, args := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs)
+	filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs)
+	totalArgs := append([]any{}, filterArgs...)
+	dimensionSQL, dimensionArgs := appTrackingCohortUnionSQL(appTrackingD1CohortSourceSQL, filter, filterArgs, "source.stat_day, source.identity, source.value")
+	args := append(totalArgs, dimensionArgs...)
 	args = append(args, app)
-	row := r.db.QueryRowContext(ctx, `
+	// retention_base 对每个 stat_day+identity 只判断一次次日活跃，再与五个维度关联；旧实现会为每个维度重复执行同一 EXISTS。
-		WITH base AS (
+	rows, err := r.db.QueryContext(ctx, `
+		WITH total_base AS (
 			SELECT DISTINCT stat_day, `+appTrackingIdentitySQL("")+` AS identity
 			FROM app_tracking_events
 			WHERE `+filter+` AND event_name = 'login_success'
+		),
+		dimension_base AS (`+dimensionSQL+`),
+		retention_base AS (
+			SELECT total_base.stat_day, total_base.identity,
+			       CASE WHEN EXISTS (
+			         SELECT 1 FROM app_tracking_events n
+			         WHERE n.app_code = ? AND n.stat_day = DATE_ADD(total_base.stat_day, INTERVAL 1 DAY)
+			           AND `+appTrackingIdentitySQL("n")+` = total_base.identity
+			         LIMIT 1
+			       ) THEN 1 ELSE 0 END AS retained
+			FROM total_base
+		),
+		dimension_aggregated AS (
+			SELECT dimension_base.dimension, dimension_base.value, COUNT(*) AS base_users,
+			       COALESCE(SUM(retention_base.retained), 0) AS retained_users
+			FROM dimension_base
+			JOIN retention_base ON retention_base.stat_day = dimension_base.stat_day
+			                   AND retention_base.identity = dimension_base.identity
+			GROUP BY dimension_base.dimension, dimension_base.value
+		),
+		dimension_ranked AS (
+			SELECT dimension, value, base_users, retained_users,
+			       ROW_NUMBER() OVER (PARTITION BY dimension ORDER BY base_users DESC, value ASC) AS dimension_rank
+			FROM dimension_aggregated
 		)
-		SELECT COUNT(*),
+		SELECT '' AS dimension, '' AS value, COUNT(*) AS base_users,
-		       COALESCE(SUM(CASE WHEN EXISTS (
+		       COALESCE(SUM(retained), 0) AS retained_users, 0 AS dimension_rank
-		         SELECT 1 FROM app_tracking_events n
+		FROM retention_base
-		         WHERE n.app_code = ? AND n.stat_day = DATE_ADD(base.stat_day, INTERVAL 1 DAY)
+		UNION ALL
-		           AND `+appTrackingIdentitySQL("n")+` = base.identity
+		SELECT dimension, value, base_users, retained_users, dimension_rank
-		         LIMIT 1
+		FROM dimension_ranked
-		       ) THEN 1 ELSE 0 END),0)
+		WHERE dimension_rank <= 50
-		FROM base`, args...)
+		ORDER BY dimension ASC, dimension_rank ASC`, 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
+		return 0, 0, nil, err
 	}
 	defer rows.Close()
-	out := []AppTrackingFunnelCohort{}
+	var baseUsers int64
+	var retainedUsers int64
+	byDimension := map[string][]AppTrackingFunnelCohort{}
+	labels := map[string]string{}
+	for _, def := range appTrackingCohortDefs {
+		labels[def.Key] = def.Label
+	}
 	for rows.Next() {
+		var dimension string
 		var value sql.NullString
-		var baseUsers, retainedUsers int64
+		var rowBaseUsers, rowRetainedUsers, dimensionRank int64
-		if err := rows.Scan(&value, &baseUsers, &retainedUsers); err != nil {
+		if err := rows.Scan(&dimension, &value, &rowBaseUsers, &rowRetainedUsers, &dimensionRank); err != nil {
-			return nil, err
+			return 0, 0, nil, err
+		}
+		if dimension == "" {
+			baseUsers = rowBaseUsers
+			retainedUsers = rowRetainedUsers
+			continue
 		}
 		cohortValue := normalizeCohortValue(value.String)
-		out = append(out, AppTrackingFunnelCohort{
+		byDimension[dimension] = append(byDimension[dimension], AppTrackingFunnelCohort{
-			Dimension:        def.Key,
+			Dimension:        dimension,
-			DimensionLabel:   def.Label,
+			DimensionLabel:   labels[dimension],
 			Value:            cohortValue,
 			Label:            cohortValue,
-			BaseUsers:        baseUsers,
+			BaseUsers:        rowBaseUsers,
-			D1RetentionUsers: retainedUsers,
+			D1RetentionUsers: rowRetainedUsers,
-			D1RetentionRate:  ratio(retainedUsers, baseUsers),
+			D1RetentionRate:  ratio(rowRetainedUsers, rowBaseUsers),
 			Steps:            []AppTrackingCohortStepCount{},
 		})
 	}
-	return out, rows.Err()
+	if err := rows.Err(); err != nil {
+		return 0, 0, nil, err
+	}
+	stepCounts, err := r.queryAppTrackingCohortSteps(ctx, app, startMS, endMS, countryID, regionID, regionIDs)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	out := []AppTrackingFunnelCohort{}
+	for _, def := range appTrackingCohortDefs {
+		for _, cohort := range byDimension[def.Key] {
+			cohort.Steps = buildAppTrackingCohortSteps(stepCounts[def.Key][cohort.Value])
+			out = append(out, cohort)
+		}
+	}
+	return baseUsers, retainedUsers, out, nil
 }
 
-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) {
+type appTrackingCohortSourceBuilder func(dimension string, filter string) string
+
+func appTrackingCohortUnionSQL(builder appTrackingCohortSourceBuilder, filter string, filterArgs []any, columns string) (string, []any) {
+	parts := make([]string, 0, len(appTrackingCohortDefs))
+	args := []any{}
+	for _, def := range appTrackingCohortDefs {
+		parts = append(parts, "SELECT ? AS dimension, "+columns+" FROM ("+builder(def.Key, filter)+") source")
+		args = append(args, def.Key)
+		args = append(args, appTrackingCohortSourceArgs(def.Key, filterArgs)...)
+	}
+	return strings.Join(parts, "\nUNION ALL\n"), args
+}
+
+func (r *Repository) queryAppTrackingCohortSteps(ctx context.Context, app string, startMS, endMS int64, countryID int64, regionID int64, regionIDs []int64) (map[string]map[string]map[string]int64, error) {
 	filter, filterArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs)
-	cohortSQL := appTrackingStepCohortSourceSQL(def.Key, filter)
+	cohortSQL, args := appTrackingCohortUnionSQL(appTrackingStepCohortSourceSQL, filter, filterArgs, "source.identity, source.value")
-	args := appTrackingCohortSourceArgs(def.Key, filterArgs)
 	eventFilter, eventArgs := appTrackingFilter(app, startMS, endMS, countryID, regionID, regionIDs)
 	args = append(args, eventArgs...)
 	names := appTrackingEventNames()
@@ -2274,30 +2292,35 @@ func (r *Repository) queryAppTrackingCohortSteps(ctx context.Context, app string
 		WITH cohort AS (`+cohortSQL+`),
 		     events AS (
 		       SELECT DISTINCT `+appTrackingIdentitySQL("")+` AS identity, event_name
-		       FROM app_tracking_events
+		       -- 同一窗口会覆盖几乎全部漏斗事件；强制时间优先覆盖索引，避免优化器只按 app_code 扫描主键后再回表去重。
+		       FROM app_tracking_events FORCE INDEX (idx_app_tracking_time_window)
 		       WHERE `+eventFilter+` AND event_name IN (`+placeholders+`)
 		     )
-		SELECT cohort.value, events.event_name, COUNT(DISTINCT cohort.identity)
+		SELECT cohort.dimension, 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...)
+		GROUP BY cohort.dimension, cohort.value, events.event_name`, args...)
 	if err != nil {
 		return nil, err
 	}
 	defer rows.Close()
-	out := map[string]map[string]int64{}
+	out := map[string]map[string]map[string]int64{}
 	for rows.Next() {
+		var dimension string
 		var value sql.NullString
 		var eventName string
 		var userCount int64
-		if err := rows.Scan(&value, &eventName, &userCount); err != nil {
+		if err := rows.Scan(&dimension, &value, &eventName, &userCount); err != nil {
 			return nil, err
 		}
 		cohortValue := normalizeCohortValue(value.String)
-		if out[cohortValue] == nil {
+		if out[dimension] == nil {
-			out[cohortValue] = map[string]int64{}
+			out[dimension] = map[string]map[string]int64{}
 		}
-		out[cohortValue][eventName] = userCount
+		if out[dimension][cohortValue] == nil {
+			out[dimension][cohortValue] = map[string]int64{}
+		}
+		out[dimension][cohortValue][eventName] = userCount
 	}
 	return out, rows.Err()
 }

services/statistics-service/internal/storage/mysql/query_test.go

@@ -84,6 +84,31 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) {
 	}
 	t.Cleanup(func() { _ = repository.Close() })
 
+	// 漏斗读取按 App+毫秒窗口裁剪，索引顺序属于性能契约；event_name 放在时间前会让全事件漏斗退化为按 App 扫描。
+	indexRows, err := repository.db.QueryContext(ctx, `
+		SELECT COLUMN_NAME
+		FROM information_schema.STATISTICS
+		WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'app_tracking_events' AND INDEX_NAME = 'idx_app_tracking_time_window'
+		ORDER BY SEQ_IN_INDEX`)
+	if err != nil {
+		t.Fatalf("query app tracking time window index: %v", err)
+	}
+	var indexColumns []string
+	for indexRows.Next() {
+		var column string
+		if err := indexRows.Scan(&column); err != nil {
+			_ = indexRows.Close()
+			t.Fatalf("scan app tracking time window index: %v", err)
+		}
+		indexColumns = append(indexColumns, column)
+	}
+	if err := indexRows.Close(); err != nil {
+		t.Fatalf("close app tracking time window index rows: %v", err)
+	}
+	if got, want := strings.Join(indexColumns, ","), "app_code,occurred_at_ms,event_name,identity_key,device_id"; got != want {
+		t.Fatalf("app tracking time window index mismatch: got=%s want=%s", got, want)
+	}
+
 	occurredAt := time.Date(2026, 7, 2, 23, 59, 59, 0, time.UTC).UnixMilli()
 	result, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{
 		{
@@ -183,10 +208,17 @@ func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) {
 		}
 	}
 	user1Props := `{"country":"US","channel":"ads","login_method":"google"}`
+	user1AffiliateProps := `{"country":"US","channel":"affiliate","login_method":"google"}`
 	user2Props := `{"country":"BR","channel":"organic","login_method":"phone"}`
+	affiliateLogin := event(1, "dev-1", "login_success", 1500*time.Millisecond, user1AffiliateProps)
+	affiliateLogin.EventID += "-affiliate"
+	retainedOutsideFilter := event(1, "dev-1", "send_message", 24*time.Hour+time.Second, user1Props)
+	// D1 表示次日任意活跃，不能继承首日查询的区域过滤，否则用户次日跨区会被错误判定流失。
+	retainedOutsideFilter.RegionID = 999
 	events := []AppTrackingEvent{
 		event(1, "dev-1", "login_start", 0, user1Props),
 		event(1, "dev-1", "login_success", time.Second, user1Props),
+		affiliateLogin,
 		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),
@@ -197,7 +229,7 @@ func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) {
 		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),
+		retainedOutsideFilter,
 		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),
@@ -252,6 +284,25 @@ func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) {
 	if us.BaseUsers != 1 || us.D1RetentionUsers != 1 || math.Abs(us.D1RetentionRate-1) > 0.0001 {
 		t.Fatalf("unexpected US cohort: %+v", us)
 	}
+	cohortSteps := func(item AppTrackingFunnelCohort) map[string]int64 {
+		steps := map[string]int64{}
+		for _, step := range item.Steps {
+			steps[step.EventName] = step.UserCount
+		}
+		return steps
+	}
+	usSteps := cohortSteps(us)
+	if usSteps["login_success"] != 1 || usSteps["room_join_success"] != 1 || usSteps["send_message"] != 1 {
+		t.Fatalf("unexpected US cohort steps: %+v", usSteps)
+	}
+	brSteps := cohortSteps(cohort("country", "BR"))
+	if brSteps["login_success"] != 1 || brSteps["room_join_success"] != 0 || brSteps["room_join_fail"] != 1 {
+		t.Fatalf("unexpected BR cohort steps: %+v", brSteps)
+	}
+	// 同一 identity+day 的相同国家只计一次；同日出现两个渠道值时，各渠道仍按既有口径分别计入。
+	if cohort("country", "US").BaseUsers != 1 || cohort("channel", "ads").BaseUsers != 1 || cohort("channel", "affiliate").BaseUsers != 1 {
+		t.Fatalf("multi-value cohort deduplication mismatch: %+v", funnel.Cohorts)
+	}
 	stay := cohort("first_room_stay", "3m-10m")
 	if stay.BaseUsers != 1 || stay.D1RetentionUsers != 1 {
 		t.Fatalf("unexpected stay cohort: %+v", stay)

services/statistics-service/internal/storage/mysql/repository.go

@@ -139,6 +139,7 @@ func (r *Repository) Migrate(ctx context.Context) error {
 		  KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms),
 		  KEY idx_app_tracking_device (app_code, device_id, occurred_at_ms),
 		  KEY idx_app_tracking_event_window (app_code, event_name, occurred_at_ms, identity_key, device_id),
+		  KEY idx_app_tracking_time_window (app_code, occurred_at_ms, event_name, identity_key, device_id),
 		  KEY idx_app_tracking_retention_identity (app_code, stat_day, identity_key)
 		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
 		`CREATE TABLE IF NOT EXISTS stat_app_day_country (
@@ -457,6 +458,8 @@ func (r *Repository) Migrate(ctx context.Context) error {
 		{"app_tracking_events", "idx_app_tracking_user", []string{"app_code", "user_id", "occurred_at_ms"}},
 		{"app_tracking_events", "idx_app_tracking_device", []string{"app_code", "device_id", "occurred_at_ms"}},
 		{"app_tracking_events", "idx_app_tracking_event_window", []string{"app_code", "event_name", "occurred_at_ms", "identity_key", "device_id"}},
+		// 漏斗步骤会一次读取全部事件名；时间列必须紧跟 app_code，避免 event_name IN 覆盖大多数事件时退化为按 App 扫描主键。
+		{"app_tracking_events", "idx_app_tracking_time_window", []string{"app_code", "occurred_at_ms", "event_name", "identity_key", "device_id"}},
 		{"app_tracking_events", "idx_app_tracking_retention_identity", []string{"app_code", "stat_day", "identity_key"}},
 		{"stat_app_day_country", "idx_stat_app_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
 		{"stat_user_day_activity", "idx_stat_user_day_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},