diff --git a/services/gateway-service/internal/client/statistics_client.go b/services/gateway-service/internal/client/statistics_client.go index ae037763..fb925844 100644 --- a/services/gateway-service/internal/client/statistics_client.go +++ b/services/gateway-service/internal/client/statistics_client.go @@ -12,27 +12,29 @@ import ( ) type AppTrackingEvent struct { - AppCode string `json:"app_code"` - EventID string `json:"event_id"` - EventName string `json:"event_name"` - EventType string `json:"event_type"` - Screen string `json:"screen"` - TargetType string `json:"target_type"` - TargetID string `json:"target_id"` - UserID int64 `json:"user_id"` - DeviceID string `json:"device_id"` - SessionID string `json:"session_id"` - Platform string `json:"platform"` - AppVersion string `json:"app_version"` - Language string `json:"language"` - Timezone string `json:"timezone"` - CountryID int64 `json:"country_id"` - RegionID int64 `json:"region_id"` - DurationMS int64 `json:"duration_ms"` - Success bool `json:"success"` - ErrorCode string `json:"error_code"` - Properties json.RawMessage `json:"properties,omitempty"` - OccurredAtMS int64 `json:"occurred_at_ms"` + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + EventType string `json:"event_type"` + Screen string `json:"screen"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + UserID int64 `json:"user_id"` + DeviceID string `json:"device_id"` + SessionID string `json:"session_id"` + ClientSessionID string `json:"client_session_id"` + AuthSessionID string `json:"auth_session_id"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` + Language string `json:"language"` + Timezone string `json:"timezone"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + Properties json.RawMessage `json:"properties,omitempty"` + OccurredAtMS int64 `json:"occurred_at_ms"` } type AppTrackingReportResult struct { diff --git a/services/gateway-service/internal/transport/http/appapi/app_event_handler.go b/services/gateway-service/internal/transport/http/appapi/app_event_handler.go index b4e9b65a..41b4ccc1 100644 --- a/services/gateway-service/internal/transport/http/appapi/app_event_handler.go +++ b/services/gateway-service/internal/transport/http/appapi/app_event_handler.go @@ -19,23 +19,24 @@ type appEventsRequest struct { } type appEventRequest struct { - EventID string `json:"event_id"` - EventName string `json:"event_name"` - EventType string `json:"event_type"` - Screen string `json:"screen"` - TargetType string `json:"target_type"` - TargetID string `json:"target_id"` - DeviceID string `json:"device_id"` - SessionID string `json:"session_id"` - Platform string `json:"platform"` - AppVersion string `json:"app_version"` - Language string `json:"language"` - Timezone string `json:"timezone"` - DurationMS int64 `json:"duration_ms"` - Success bool `json:"success"` - ErrorCode string `json:"error_code"` - Properties json.RawMessage `json:"properties"` - OccurredAtMS int64 `json:"occurred_at_ms"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + EventType string `json:"event_type"` + Screen string `json:"screen"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + DeviceID string `json:"device_id"` + SessionID string `json:"session_id"` + ClientSessionID string `json:"client_session_id"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` + Language string `json:"language"` + Timezone string `json:"timezone"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + Properties json.RawMessage `json:"properties"` + OccurredAtMS int64 `json:"occurred_at_ms"` } func (h *Handler) reportAppEvents(writer http.ResponseWriter, request *http.Request) { @@ -147,25 +148,27 @@ func appTrackingEventFromRequest(item appEventRequest, defaults appTrackingEvent occurredAtMS = defaults.receivedAtMS } event := client.AppTrackingEvent{ - AppCode: defaults.appCode, - EventID: item.EventID, - EventName: item.EventName, - EventType: item.EventType, - Screen: item.Screen, - TargetType: item.TargetType, - TargetID: item.TargetID, - UserID: defaults.userID, - DeviceID: deviceID, - SessionID: firstTrimmed(defaults.sessionID, item.SessionID), - Platform: strings.ToLower(firstTrimmed(item.Platform, defaults.platform)), - AppVersion: firstTrimmed(item.AppVersion, defaults.appVersion), - Language: firstTrimmed(item.Language, defaults.language), - Timezone: firstTrimmed(item.Timezone, defaults.timezone), - DurationMS: item.DurationMS, - Success: item.Success, - ErrorCode: item.ErrorCode, - Properties: properties, - OccurredAtMS: occurredAtMS, + AppCode: defaults.appCode, + EventID: item.EventID, + EventName: item.EventName, + EventType: item.EventType, + Screen: item.Screen, + TargetType: item.TargetType, + TargetID: item.TargetID, + UserID: defaults.userID, + DeviceID: deviceID, + SessionID: firstTrimmed(defaults.sessionID, item.SessionID), + ClientSessionID: item.ClientSessionID, + AuthSessionID: defaults.sessionID, + Platform: strings.ToLower(firstTrimmed(item.Platform, defaults.platform)), + AppVersion: firstTrimmed(item.AppVersion, defaults.appVersion), + Language: firstTrimmed(item.Language, defaults.language), + Timezone: firstTrimmed(item.Timezone, defaults.timezone), + DurationMS: item.DurationMS, + Success: item.Success, + ErrorCode: item.ErrorCode, + Properties: properties, + OccurredAtMS: occurredAtMS, } if !normalizeAppTrackingEventText(&event) || event.EventID == "" || event.EventName == "" { return client.AppTrackingEvent{}, false @@ -201,6 +204,8 @@ func normalizeAppTrackingEventText(event *client.AppTrackingEvent) bool { {value: &event.TargetID, maxBytes: apptracking.MaxTargetIDBytes}, {value: &event.DeviceID, maxBytes: apptracking.MaxDeviceIDBytes}, {value: &event.SessionID, maxBytes: apptracking.MaxSessionIDBytes}, + {value: &event.ClientSessionID, maxBytes: apptracking.MaxSessionIDBytes}, + {value: &event.AuthSessionID, maxBytes: apptracking.MaxSessionIDBytes}, {value: &event.Platform, maxBytes: apptracking.MaxPlatformBytes}, {value: &event.AppVersion, maxBytes: apptracking.MaxAppVersionBytes}, {value: &event.Language, maxBytes: apptracking.MaxLanguageBytes}, diff --git a/services/statistics-service/Dockerfile b/services/statistics-service/Dockerfile index 5ec4ea09..63a4f396 100644 --- a/services/statistics-service/Dockerfile +++ b/services/statistics-service/Dockerfile @@ -15,13 +15,17 @@ COPY . . RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \ --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \ mkdir -p /out && \ - GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/statistics-service/cmd/server + GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/server ./services/statistics-service/cmd/server && \ + GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/backfill-social-identities ./services/statistics-service/cmd/backfill-social-identities FROM alpine:3.20 ENV TZ=UTC WORKDIR /app RUN apk add --no-cache ca-certificates wget && adduser -D -g '' appuser COPY --from=builder /out/server /app/server +# 历史身份回放只随 statistics 镜像交付，不开放 HTTP；生产由单节点 +# docker exec 显式 dry-run/apply，复用容器内只读配置与最小连接数保护。 +COPY --from=builder /out/backfill-social-identities /app/backfill-social-identities COPY services/statistics-service/configs/config.docker.yaml /app/config.yaml USER appuser EXPOSE 13110 diff --git a/services/statistics-service/cmd/backfill-social-identities/main.go b/services/statistics-service/cmd/backfill-social-identities/main.go new file mode 100644 index 00000000..2b9f83e5 --- /dev/null +++ b/services/statistics-service/cmd/backfill-social-identities/main.go @@ -0,0 +1,356 @@ +package main + +import ( + "context" + "database/sql" + "flag" + "fmt" + "log" + "os" + "sort" + "strings" + "time" + + "hyapp/pkg/appcode" + statisticsconfig "hyapp/services/statistics-service/internal/config" + + _ "github.com/go-sql-driver/mysql" +) + +const ( + legacyDayResolutionSource = "legacy_day_unique" + legacyWindowResolutionSource = "legacy_window_unique" + legacyAmbiguousResolutionSource = "legacy_ambiguous" +) + +type replayWindow struct { + start time.Time + end time.Time + loc *time.Location +} + +type deviceResolution struct { + userID int64 + ambiguous bool +} + +type dayResolution struct { + deviceID string + userID int64 + state string + source string +} + +type daySummary struct { + day string + anonymous int + resolved int + ambiguous int + unresolved int +} + +func main() { + var ( + appsValue = flag.String("apps", envDefault("APP_CODES", "lalu,huwaa,fami"), "comma-separated app_code list") + statTZ = flag.String("timezone", envDefault("STAT_TZ", "Asia/Shanghai"), "UTC or Asia/Shanghai") + startDay = flag.String("start-day", "", "inclusive day, yyyy-mm-dd") + endDay = flag.String("end-day", "", "inclusive day, yyyy-mm-dd") + apply = flag.Bool("apply", false, "write identity resolutions; default is dry-run") + configPath = flag.String("config", envDefault("STATISTICS_CONFIG", "/app/config.yaml"), "statistics-service YAML used when no DSN flag/env is set") + dsn = flag.String("statistics-dsn", firstNonEmptyEnv("STATISTICS_MYSQL_DSN", "MYSQL_DSN"), "statistics MySQL DSN") + pause = flag.Duration("pause", 250*time.Millisecond, "pause between bounded day queries") + slicePause = flag.Duration("slice-pause", 50*time.Millisecond, "pause after each one-hour raw scan") + queryLimit = flag.Duration("query-timeout", 30*time.Second, "timeout for each bounded query or day transaction") + maxOpenConn = flag.Int("max-open-conns", 1, "small connection cap for production-safe backfill") + receivedCutoff = flag.Int64("received-before-ms", time.Now().UTC().UnixMilli(), "only use raw events received at or before this fixed cutoff") + allowWindowFallback = flag.Bool("allow-window-fallback", false, "allow cross-day unique device inference; unsafe by default for legacy shared devices") + ) + flag.Parse() + + apps := normalizeApps(*appsValue) + dsnValue := strings.TrimSpace(*dsn) + if dsnValue == "" { + cfg, err := statisticsconfig.Load(strings.TrimSpace(*configPath)) + if err != nil { + log.Fatalf("load statistics config: %v", err) + } + dsnValue = strings.TrimSpace(cfg.MySQLDSN) + } + if len(apps) == 0 || dsnValue == "" { + log.Fatal("apps and statistics MySQL DSN are required") + } + window, err := resolveWindow(*startDay, *endDay, *statTZ) + if err != nil { + log.Fatal(err) + } + db, err := sql.Open("mysql", dsnValue) + if err != nil { + log.Fatal(err) + } + defer db.Close() + if *maxOpenConn < 1 { + *maxOpenConn = 1 + } + db.SetMaxOpenConns(*maxOpenConn) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(5 * time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), *queryLimit) + if err := db.PingContext(ctx); err != nil { + cancel() + log.Fatal(err) + } + cancel() + fmt.Printf("identity backfill cutoff received_before_ms=%d (%s)\n", *receivedCutoff, time.UnixMilli(*receivedCutoff).UTC().Format(time.RFC3339)) + + for _, app := range apps { + if err := backfillApp(db, app, *statTZ, window, *receivedCutoff, *allowWindowFallback, *apply, *pause, *slicePause, *queryLimit); err != nil { + log.Fatalf("app=%s: %v", app, err) + } + } +} + +func backfillApp(db *sql.DB, app, statTZ string, window replayWindow, receivedCutoff int64, allowWindowFallback, apply bool, pause, slicePause, queryLimit time.Duration) error { + windowDeviceUsers := map[string]deviceResolution{} + dayDeviceUsers := map[string]map[string]deviceResolution{} + for day := window.start; !day.After(window.end); day = day.AddDate(0, 0, 1) { + dayText := day.Format("2006-01-02") + dayDeviceUsers[dayText] = map[string]deviceResolution{} + // One-hour slices cap every index range even on the peak day. All + // authenticated event names participate: an account that only emitted + // page_open must still disqualify a shared device from "unique" mapping. + for hour := day; hour.Before(day.AddDate(0, 0, 1)); hour = hour.Add(time.Hour) { + if err := loadLoggedIdentitySlice(db, app, hour.UnixMilli(), hour.Add(time.Hour).UnixMilli(), receivedCutoff, + dayDeviceUsers[dayText], windowDeviceUsers, queryLimit); err != nil { + return fmt.Errorf("load logged identities day=%s hour=%s: %w", dayText, hour.Format("15:04"), err) + } + pauseFor(slicePause) + } + pauseFor(pause) + } + + for day := window.start; !day.After(window.end); day = day.AddDate(0, 0, 1) { + dayText := day.Format("2006-01-02") + resolutions, summary, err := planDay(db, app, statTZ, dayText, dayDeviceUsers[dayText], windowDeviceUsers, allowWindowFallback, queryLimit) + if err != nil { + return err + } + mode := "dry-run" + if apply { + mode = "applied" + if err := applyDay(db, app, statTZ, dayText, resolutions, queryLimit); err != nil { + return err + } + } + fmt.Printf("%s app=%s stat_tz=%s day=%s anonymous=%d resolved=%d ambiguous=%d unresolved=%d\n", + mode, app, statTZ, summary.day, summary.anonymous, summary.resolved, summary.ambiguous, summary.unresolved) + pauseFor(pause) + } + return nil +} + +func loadLoggedIdentitySlice(db *sql.DB, app string, startMS, endMS, receivedCutoff int64, dayUsers, windowUsers map[string]deviceResolution, queryLimit time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), queryLimit) + defer cancel() + rows, err := db.QueryContext(ctx, ` + SELECT device_id, user_id + FROM app_tracking_events FORCE INDEX (idx_app_tracking_time_window) + WHERE app_code = ? AND occurred_at_ms >= ? AND occurred_at_ms < ? AND received_at_ms <= ? + AND device_id <> '' AND user_id > 0 + `, app, startMS, endMS, receivedCutoff) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var deviceID string + var userID int64 + if err := rows.Scan(&deviceID, &userID); err != nil { + return err + } + deviceID = strings.TrimSpace(deviceID) + if deviceID == "" || userID <= 0 { + continue + } + addDeviceUser(dayUsers, deviceID, userID) + addDeviceUser(windowUsers, deviceID, userID) + } + return rows.Err() +} + +func addDeviceUser(target map[string]deviceResolution, deviceID string, userID int64) { + current, exists := target[deviceID] + if !exists { + target[deviceID] = deviceResolution{userID: userID} + return + } + if current.ambiguous || current.userID == userID { + return + } + current.ambiguous = true + current.userID = 0 + target[deviceID] = current +} + +func planDay(db *sql.DB, app, statTZ, day string, dayUsers, windowUsers map[string]deviceResolution, allowWindowFallback bool, queryLimit time.Duration) ([]dayResolution, daySummary, error) { + ctx, cancel := context.WithTimeout(context.Background(), queryLimit) + defer cancel() + rows, err := db.QueryContext(ctx, ` + SELECT device_id + FROM stat_social_user_day + WHERE app_code = ? AND stat_tz = ? AND stat_day = ? + AND user_id = 0 AND device_id <> '' + ORDER BY device_id + `, app, statTZ, day) + if err != nil { + return nil, daySummary{}, fmt.Errorf("load anonymous devices day=%s: %w", day, err) + } + defer rows.Close() + summary := daySummary{day: day} + resolutions := []dayResolution{} + for rows.Next() { + var deviceID string + if err := rows.Scan(&deviceID); err != nil { + return nil, daySummary{}, err + } + summary.anonymous++ + identity, ok := dayUsers[deviceID] + switch { + case ok && (identity.ambiguous || identity.userID <= 0): + summary.ambiguous++ + resolutions = append(resolutions, dayResolution{deviceID: deviceID, state: "ambiguous", source: legacyAmbiguousResolutionSource}) + case ok: + summary.resolved++ + resolutions = append(resolutions, dayResolution{deviceID: deviceID, userID: identity.userID, state: "resolved", source: legacyDayResolutionSource}) + default: + if !allowWindowFallback { + summary.unresolved++ + continue + } + identity, ok = windowUsers[deviceID] + if !ok { + summary.unresolved++ + continue + } + if identity.ambiguous || identity.userID <= 0 { + summary.ambiguous++ + resolutions = append(resolutions, dayResolution{deviceID: deviceID, state: "ambiguous", source: legacyAmbiguousResolutionSource}) + continue + } + // Cross-day unique fallback repairs the exact retention gap where day 1 + // was anonymous and the same device first authenticated on day 2. It is + // never used if the device showed more than one account in the window. + summary.resolved++ + resolutions = append(resolutions, dayResolution{deviceID: deviceID, userID: identity.userID, state: "resolved", source: legacyWindowResolutionSource}) + } + } + return resolutions, summary, rows.Err() +} + +func applyDay(db *sql.DB, app, statTZ, day string, resolutions []dayResolution, queryLimit time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), queryLimit) + defer cancel() + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + // Only rows produced by this legacy job are replaced. Realtime client-session + // evidence has higher authority and must survive an idempotent historical rerun. + if _, err := tx.ExecContext(ctx, ` + DELETE FROM stat_social_identity_day + WHERE app_code = ? AND stat_tz = ? AND stat_day = ? + AND resolution_source IN (?, ?, ?) + `, app, statTZ, day, legacyDayResolutionSource, legacyWindowResolutionSource, legacyAmbiguousResolutionSource); err != nil { + return err + } + stmt, err := tx.PrepareContext(ctx, ` + INSERT INTO stat_social_identity_day ( + app_code, stat_tz, stat_day, device_id, resolved_user_id, + resolution_state, resolution_source, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + resolved_user_id = IF(resolution_source = 'client_session', resolved_user_id, VALUES(resolved_user_id)), + resolution_state = IF(resolution_source = 'client_session', resolution_state, VALUES(resolution_state)), + resolution_source = IF(resolution_source = 'client_session', resolution_source, VALUES(resolution_source)), + updated_at_ms = IF(resolution_source = 'client_session', updated_at_ms, VALUES(updated_at_ms)) + `) + if err != nil { + return err + } + defer stmt.Close() + nowMS := time.Now().UTC().UnixMilli() + for _, resolution := range resolutions { + if _, err := stmt.ExecContext(ctx, app, statTZ, day, resolution.deviceID, resolution.userID, + resolution.state, resolution.source, nowMS); err != nil { + return err + } + } + return tx.Commit() +} + +func resolveWindow(startDay, endDay, statTZ string) (replayWindow, error) { + var loc *time.Location + switch strings.TrimSpace(statTZ) { + case "UTC": + loc = time.UTC + case "Asia/Shanghai": + loc = time.FixedZone("Asia/Shanghai", 8*60*60) + default: + return replayWindow{}, fmt.Errorf("unsupported timezone %q", statTZ) + } + start, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(startDay), loc) + if err != nil { + return replayWindow{}, fmt.Errorf("parse start-day: %w", err) + } + end, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(endDay), loc) + if err != nil { + return replayWindow{}, fmt.Errorf("parse end-day: %w", err) + } + if end.Before(start) { + return replayWindow{}, fmt.Errorf("end-day is before start-day") + } + if days := int(end.Sub(start).Hours()/24) + 1; days > 31 { + return replayWindow{}, fmt.Errorf("window has %d days; maximum is 31", days) + } + return replayWindow{start: start, end: end, loc: loc}, nil +} + +func normalizeApps(value string) []string { + seen := map[string]struct{}{} + apps := []string{} + for _, item := range strings.Split(value, ",") { + app := appcode.Normalize(item) + if app == "" { + continue + } + if _, ok := seen[app]; ok { + continue + } + seen[app] = struct{}{} + apps = append(apps, app) + } + sort.Strings(apps) + return apps +} + +func pauseFor(pause time.Duration) { + if pause > 0 { + time.Sleep(pause) + } +} + +func envDefault(key, fallback string) string { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + return fallback +} + +func firstNonEmptyEnv(keys ...string) string { + for _, key := range keys { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + } + return "" +} 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 0b87eccc..b9093454 100644 --- a/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql +++ b/services/statistics-service/deploy/mysql/initdb/001_statistics_service.sql @@ -28,7 +28,9 @@ CREATE TABLE IF NOT EXISTS app_tracking_events ( user_id BIGINT NOT NULL DEFAULT 0 COMMENT '服务端识别的用户 ID，匿名为 0', device_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '设备 ID', identity_key VARCHAR(160) GENERATED ALWAYS AS (CASE WHEN user_id > 0 THEN CONCAT('u:', user_id) ELSE CONCAT('d:', device_id) END) VIRTUAL COMMENT '漏斗/留存去重身份，登录用户优先 user_id，匿名用户回退 device_id', - session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '会话 ID', + session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '兼容会话 ID', + client_session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT 'App 进程会话 ID，登录前后保持不变', + auth_session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '服务端认证会话 ID', platform VARCHAR(32) NOT NULL DEFAULT '' COMMENT '平台', app_version VARCHAR(64) NOT NULL DEFAULT '' COMMENT 'App 版本', language VARCHAR(32) NOT NULL DEFAULT '' COMMENT '语言', @@ -205,6 +207,49 @@ CREATE TABLE IF NOT EXISTS stat_social_user_day ( KEY idx_stat_social_user_day_filter (app_code, stat_tz, stat_day, user_role, recharge_user) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social BI P0-P5 用户日级读模型'; +CREATE TABLE IF NOT EXISTS stat_social_identity_session_user ( + app_code VARCHAR(32) NOT NULL, + device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL, + user_id BIGINT NOT NULL, + first_seen_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, device_id, client_session_id, user_id), + KEY idx_social_identity_session_user (app_code, user_id, first_seen_at_ms) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 客户端会话登录账号观测表'; + +CREATE TABLE IF NOT EXISTS stat_social_identity_device_lock ( + app_code VARCHAR(32) NOT NULL, + device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + created_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, device_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 身份解析设备级事务锁'; + +CREATE TABLE IF NOT EXISTS stat_social_identity_session_day ( + app_code VARCHAR(32) NOT NULL, + stat_tz VARCHAR(64) NOT NULL, + stat_day DATE NOT NULL, + device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL, + first_seen_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, device_id, client_session_id, stat_tz, stat_day), + KEY idx_social_identity_session_day (app_code, stat_tz, stat_day, device_id, client_session_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 匿名会话活跃日观测表'; + +CREATE TABLE IF NOT EXISTS stat_social_identity_day ( + app_code VARCHAR(32) NOT NULL, + stat_tz VARCHAR(64) NOT NULL, + stat_day DATE NOT NULL, + device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + resolved_user_id BIGINT NOT NULL DEFAULT 0, + resolution_state VARCHAR(16) NOT NULL DEFAULT 'unresolved', + resolution_source VARCHAR(32) NOT NULL DEFAULT 'client_session', + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_tz, stat_day, device_id), + KEY idx_social_identity_day_user (app_code, stat_tz, stat_day, resolved_user_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 设备日级 canonical 身份解析表'; + CREATE TABLE IF NOT EXISTS stat_recharge_day_payers ( app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai', diff --git a/services/statistics-service/internal/app/query_http.go b/services/statistics-service/internal/app/query_http.go index b611d40f..4e011096 100644 --- a/services/statistics-service/internal/app/query_http.go +++ b/services/statistics-service/internal/app/query_http.go @@ -286,27 +286,29 @@ type socialCoinSellerRechargeOrderRequest struct { } type appTrackingEventRequest struct { - AppCode string `json:"app_code"` - EventID string `json:"event_id"` - EventName string `json:"event_name"` - EventType string `json:"event_type"` - Screen string `json:"screen"` - TargetType string `json:"target_type"` - TargetID string `json:"target_id"` - UserID int64 `json:"user_id"` - DeviceID string `json:"device_id"` - SessionID string `json:"session_id"` - Platform string `json:"platform"` - AppVersion string `json:"app_version"` - Language string `json:"language"` - Timezone string `json:"timezone"` - CountryID int64 `json:"country_id"` - RegionID int64 `json:"region_id"` - DurationMS int64 `json:"duration_ms"` - Success bool `json:"success"` - ErrorCode string `json:"error_code"` - Properties json.RawMessage `json:"properties"` - OccurredAtMS int64 `json:"occurred_at_ms"` + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventName string `json:"event_name"` + EventType string `json:"event_type"` + Screen string `json:"screen"` + TargetType string `json:"target_type"` + TargetID string `json:"target_id"` + UserID int64 `json:"user_id"` + DeviceID string `json:"device_id"` + SessionID string `json:"session_id"` + ClientSessionID string `json:"client_session_id"` + AuthSessionID string `json:"auth_session_id"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` + Language string `json:"language"` + Timezone string `json:"timezone"` + CountryID int64 `json:"country_id"` + RegionID int64 `json:"region_id"` + DurationMS int64 `json:"duration_ms"` + Success bool `json:"success"` + ErrorCode string `json:"error_code"` + Properties json.RawMessage `json:"properties"` + OccurredAtMS int64 `json:"occurred_at_ms"` } func (s *queryHTTPServer) socialCoinSellerRechargeOrder(w http.ResponseWriter, r *http.Request) { @@ -375,27 +377,29 @@ func (s *queryHTTPServer) appEvents(w http.ResponseWriter, r *http.Request) { for _, item := range body.Events { // Internal ingest 只接受 gateway 已解析过的 App、用户和维度；statistics-service 只负责幂等落表，不回查 owner service。 events = append(events, mysqlstorage.AppTrackingEvent{ - AppCode: appcode.Normalize(item.AppCode), - EventID: item.EventID, - EventName: item.EventName, - EventType: item.EventType, - Screen: item.Screen, - TargetType: item.TargetType, - TargetID: item.TargetID, - UserID: item.UserID, - DeviceID: item.DeviceID, - SessionID: item.SessionID, - Platform: item.Platform, - AppVersion: item.AppVersion, - Language: item.Language, - Timezone: item.Timezone, - CountryID: item.CountryID, - RegionID: item.RegionID, - DurationMS: item.DurationMS, - Success: item.Success, - ErrorCode: item.ErrorCode, - Properties: item.Properties, - OccurredAtMS: item.OccurredAtMS, + AppCode: appcode.Normalize(item.AppCode), + EventID: item.EventID, + EventName: item.EventName, + EventType: item.EventType, + Screen: item.Screen, + TargetType: item.TargetType, + TargetID: item.TargetID, + UserID: item.UserID, + DeviceID: item.DeviceID, + SessionID: item.SessionID, + ClientSessionID: item.ClientSessionID, + AuthSessionID: item.AuthSessionID, + Platform: item.Platform, + AppVersion: item.AppVersion, + Language: item.Language, + Timezone: item.Timezone, + CountryID: item.CountryID, + RegionID: item.RegionID, + DurationMS: item.DurationMS, + Success: item.Success, + ErrorCode: item.ErrorCode, + Properties: item.Properties, + OccurredAtMS: item.OccurredAtMS, }) } result, err := s.repo.ConsumeAppTrackingEvents(r.Context(), events) diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index b64d1de3..cc39401f 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -27,6 +27,32 @@ func TestGiftDebitConsumptionOwnersDoNotOverlap(t *testing.T) { } } +func TestCanonicalizeSocialUserDaysMergesIdentityWithoutLosingDeviceCount(t *testing.T) { + if socialRetentionLookaheadDays([]string{SocialSectionActive}) != 0 || + socialRetentionLookaheadDays([]string{SocialSectionNewUsers}) != 1 || + socialRetentionLookaheadDays([]string{SocialSectionActive, SocialSectionRetention}) != 30 { + t.Fatal("social retention lookahead must stay section-bounded") + } + rows := []socialRequirementUserDay{ + {StatDay: "2026-07-01", Subject: "d:device-A", DeviceID: "device-A", CountryID: 999, RegionID: 999, UserRole: "user", AppFirstOpen: 1, Active: 1, AppStayMS: 60_000, RoomStayMS: 20_000}, + {StatDay: "2026-07-01", Subject: "u:123", UserID: 123, DirectUser: true, DeviceID: "device-A", CountryID: 86, RegionID: 210, UserRole: "host", AppFirstOpen: 1, Active: 1, AppStayMS: 30_000, RoomOnlineMS: 45_000}, + } + resolved := map[socialIdentityDeviceDayKey]int64{ + {StatDay: "2026-07-01", DeviceID: "device-A"}: 123, + } + got := canonicalizeSocialUserDays(rows, resolved, 0, 0, []int64{210}) + if len(got) != 1 { + t.Fatalf("expected one canonical user-day, got %+v", got) + } + row := got[0] + if row.Subject != "u:123" || row.UserID != 123 || !row.DirectUser || row.CountryID != 86 || row.RegionID != 210 || row.UserRole != "host" || row.Active != 1 { + t.Fatalf("canonical identity/dimension mismatch: %+v", row) + } + if row.AppFirstOpen != 1 || row.AppStayMS != 90_000 || row.RoomStayMS != 45_000 { + t.Fatalf("canonical metric merge mismatch: %+v", row) + } +} + func TestQueryOverviewUsesActivityLuckyGiftDayStats(t *testing.T) { ctx := context.Background() _, file, _, _ := runtime.Caller(0) @@ -124,26 +150,28 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) { occurredAt := time.Date(2026, 7, 2, 23, 59, 59, 0, time.UTC).UnixMilli() result, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{ { - AppCode: "lalu", - EventID: "evt-banner-1", - EventName: "banner_view", - EventType: "view", - Screen: "home", - TargetType: "banner", - TargetID: "17", - UserID: 42, - DeviceID: "dev-1", - SessionID: "sess-1", - Platform: "ios", - AppVersion: "1.2.3", - Language: "en-US", - Timezone: "Asia/Shanghai", - CountryID: 86, - RegionID: 210, - DurationMS: 12, - Success: true, - Properties: json.RawMessage(`{"slot":"top"}`), - OccurredAtMS: occurredAt, + AppCode: "lalu", + EventID: "evt-banner-1", + EventName: "banner_view", + EventType: "view", + Screen: "home", + TargetType: "banner", + TargetID: "17", + UserID: 42, + DeviceID: "dev-1", + SessionID: "sess-1", + ClientSessionID: "client-sess-1", + AuthSessionID: "auth-sess-1", + Platform: "ios", + AppVersion: "1.2.3", + Language: "en-US", + Timezone: "Asia/Shanghai", + CountryID: 86, + RegionID: 210, + DurationMS: 12, + Success: true, + Properties: json.RawMessage(`{"slot":"top"}`), + OccurredAtMS: occurredAt, }, }) if err != nil { @@ -172,25 +200,130 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) { } var count int64 - var statDay, screen, identityKey, propertiesJSON string + var statDay, screen, identityKey, clientSessionID, authSessionID, propertiesJSON string var userID, countryID, regionID, durationMS int64 var success bool if err := repository.db.QueryRowContext(ctx, ` - SELECT COUNT(*), MAX(CAST(stat_day AS CHAR)), MAX(screen), MAX(identity_key), MAX(user_id), MAX(country_id), MAX(region_id), + SELECT COUNT(*), MAX(CAST(stat_day AS CHAR)), MAX(screen), MAX(identity_key), + MAX(client_session_id), MAX(auth_session_id), MAX(user_id), MAX(country_id), MAX(region_id), MAX(duration_ms), MAX(success), COALESCE(MAX(CAST(properties_json AS CHAR)), '{}') FROM app_tracking_events WHERE app_code = 'lalu' AND event_id = 'evt-banner-1' - `).Scan(&count, &statDay, &screen, &identityKey, &userID, &countryID, ®ionID, &durationMS, &success, &propertiesJSON); err != nil { + `).Scan(&count, &statDay, &screen, &identityKey, &clientSessionID, &authSessionID, &userID, &countryID, ®ionID, &durationMS, &success, &propertiesJSON); err != nil { t.Fatalf("query app tracking event: %v", err) } - if count != 1 || statDay != "2026-07-02" || screen != "home" || identityKey != "u:42" || userID != 42 || countryID != 86 || regionID != 210 || durationMS != 12 || !success { - t.Fatalf("stored event mismatch: count=%d statDay=%s screen=%s identityKey=%s userID=%d countryID=%d regionID=%d durationMS=%d success=%v", count, statDay, screen, identityKey, userID, countryID, regionID, durationMS, success) + if count != 1 || statDay != "2026-07-02" || screen != "home" || identityKey != "u:42" || clientSessionID != "client-sess-1" || authSessionID != "auth-sess-1" || userID != 42 || countryID != 86 || regionID != 210 || durationMS != 12 || !success { + t.Fatalf("stored event mismatch: count=%d statDay=%s screen=%s identityKey=%s clientSessionID=%s authSessionID=%s userID=%d countryID=%d regionID=%d durationMS=%d success=%v", count, statDay, screen, identityKey, clientSessionID, authSessionID, userID, countryID, regionID, durationMS, success) } if DecodeJSON(propertiesJSON)["slot"] != "top" { t.Fatalf("properties_json mismatch: %s", propertiesJSON) } } +func TestSocialIdentityCanonicalizesAnonymousAndLoggedSession(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_social_identity_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) + events := []AppTrackingEvent{ + // Both pre-login rows are stored as d:device-A. The authenticated event + // establishes the client-session bridge without rewriting those rows. + {AppCode: "lalu", EventID: "identity:first-open", EventName: "app_first_open", DeviceID: "device-A", ClientSessionID: "client-session-A", Success: true, OccurredAtMS: start.UnixMilli()}, + {AppCode: "lalu", EventID: "identity:anonymous-session", EventName: "app_session", DeviceID: "device-A", ClientSessionID: "client-session-A", DurationMS: 120_000, Success: true, OccurredAtMS: start.Add(time.Minute).UnixMilli()}, + {AppCode: "lalu", EventID: "identity:login", EventName: "home_view", UserID: 123, DeviceID: "device-A", ClientSessionID: "client-session-A", AuthSessionID: "auth-session-A", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(2 * time.Minute).UnixMilli()}, + {AppCode: "lalu", EventID: "identity:d1", EventName: "home_view", UserID: 123, DeviceID: "device-A", ClientSessionID: "client-session-A", AuthSessionID: "auth-session-A", CountryID: 86, RegionID: 999, Success: true, OccurredAtMS: start.Add(24*time.Hour + time.Minute).UnixMilli()}, + } + if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), events); err != nil { + t.Fatalf("consume identity events: %v", err) + } + + result, err := repository.QuerySocialRequirements(ctx, SocialRequirementsQuery{ + AppCode: "lalu", + StartMS: start.UnixMilli(), + EndMS: start.Add(24 * time.Hour).UnixMilli(), + RegionIDs: []int64{210}, + }) + if err != nil { + t.Fatalf("query canonical social requirements: %v", err) + } + active := socialTestSection(t, result, SocialSectionActive) + if socialMetricInt(t, active.Total, "active_users") != 1 || math.Abs(socialMetricFloat(t, active.Total, "avg_app_stay_min")-2) > 0.0001 { + t.Fatalf("anonymous and logged rows were not canonicalized: %+v", active.Total.Metrics) + } + newUsers := socialTestSection(t, result, SocialSectionNewUsers) + if socialMetricInt(t, newUsers.Total, "app_download_users") != 1 { + t.Fatalf("device download count changed after identity merge: %+v", newUsers.Total.Metrics) + } + retention := socialTestSection(t, result, SocialSectionRetention) + if socialMetricInt(t, retention.Total, "active_d1_active_base_users") != 1 || + socialMetricInt(t, retention.Total, "active_d1_active_users") != 1 || + math.Abs(socialMetricFloat(t, retention.Total, "active_d1_active_rate")-1) > 0.0001 { + t.Fatalf("canonical retention mismatch: %+v", retention.Total.Metrics) + } +} + +func TestSocialIdentityMarksConcurrentAccountSwitchAmbiguous(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_identity_switch", + }) + 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) + if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{{ + AppCode: "lalu", EventID: "switch:anonymous", EventName: "home_view", DeviceID: "shared-device", ClientSessionID: "shared-process-session", Success: true, OccurredAtMS: start.UnixMilli(), + }}); err != nil { + t.Fatalf("consume anonymous event: %v", err) + } + + // Deliberately race two accounts in one process session. The device lock + // must serialize the current reads so the last writer cannot leave a falsely + // resolved user behind. + errors := make(chan error, 2) + for userID := int64(1); userID <= 2; userID++ { + userID := userID + go func() { + _, consumeErr := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{{ + AppCode: "lalu", EventID: "switch:user:" + strconv.FormatInt(userID, 10), EventName: "home_view", UserID: userID, DeviceID: "shared-device", ClientSessionID: "shared-process-session", Success: true, OccurredAtMS: start.Add(time.Duration(userID) * time.Minute).UnixMilli(), + }}) + errors <- consumeErr + }() + } + for index := 0; index < 2; index++ { + if err := <-errors; err != nil { + t.Fatalf("consume concurrent account event: %v", err) + } + } + + var resolvedUserID int64 + var state, source string + if err := repository.db.QueryRowContext(ctx, ` + SELECT resolved_user_id, resolution_state, resolution_source + FROM stat_social_identity_day + WHERE app_code = 'lalu' AND stat_tz = 'UTC' AND stat_day = '2026-07-01' AND device_id = 'shared-device' + `).Scan(&resolvedUserID, &state, &source); err != nil { + t.Fatalf("query identity resolution: %v", err) + } + if resolvedUserID != 0 || state != "ambiguous" || source != "client_session" { + t.Fatalf("account switch must remain ambiguous: user=%d state=%s source=%s", resolvedUserID, state, source) + } +} + func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) { ctx := context.Background() _, file, _, _ := runtime.Caller(0) @@ -355,6 +488,7 @@ func TestQuerySocialRequirementsAggregatesP0ToP5(t *testing.T) { {AppCode: "lalu", EventID: "social:follow-user:1001", EventName: "follow_host", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(7 * time.Minute).UnixMilli()}, {AppCode: "lalu", EventID: "social:follow-room:1001", EventName: "follow_room", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(8 * time.Minute).UnixMilli()}, {AppCode: "lalu", EventID: "social:d1-active:1001", EventName: "home_view", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(24*time.Hour + time.Minute).UnixMilli()}, + {AppCode: "lalu", EventID: "social:d1-room:1001", EventName: "room_join_success", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(24*time.Hour + 2*time.Minute).UnixMilli()}, } { if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{event}); err != nil { t.Fatalf("consume app event %s: %v", event.EventName, err) diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index fcf945ff..1a047478 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -5,6 +5,7 @@ import ( "database/sql" "encoding/json" "errors" + "sort" "strings" "time" @@ -122,7 +123,8 @@ func (r *Repository) Migrate(ctx context.Context) error { target_type VARCHAR(64) NOT NULL DEFAULT '', target_id VARCHAR(128) NOT NULL DEFAULT '', user_id BIGINT NOT NULL DEFAULT 0, device_id VARCHAR(128) NOT NULL DEFAULT '', identity_key VARCHAR(160) GENERATED ALWAYS AS (CASE WHEN user_id > 0 THEN CONCAT('u:', user_id) ELSE CONCAT('d:', device_id) END) VIRTUAL, - session_id VARCHAR(160) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT '', + session_id VARCHAR(160) NOT NULL DEFAULT '', client_session_id VARCHAR(160) NOT NULL DEFAULT '', + auth_session_id VARCHAR(160) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT '', app_version VARCHAR(64) NOT NULL DEFAULT '', language VARCHAR(32) NOT NULL DEFAULT '', timezone VARCHAR(64) NOT NULL DEFAULT '', country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, duration_ms BIGINT NOT NULL DEFAULT 0, @@ -216,6 +218,34 @@ func (r *Repository) Migrate(ctx context.Context) error { KEY idx_stat_social_user_day_region (app_code, stat_tz, stat_day, region_id), KEY idx_stat_social_user_day_filter (app_code, stat_tz, stat_day, user_role, recharge_user) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_social_identity_session_user ( + app_code VARCHAR(32) NOT NULL, device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL, + user_id BIGINT NOT NULL, first_seen_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, device_id, client_session_id, user_id), + KEY idx_social_identity_session_user (app_code, user_id, first_seen_at_ms) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_social_identity_device_lock ( + app_code VARCHAR(32) NOT NULL, device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + created_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, device_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_social_identity_session_day ( + app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, + device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, + client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL, + first_seen_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, device_id, client_session_id, stat_tz, stat_day), + KEY idx_social_identity_session_day (app_code, stat_tz, stat_day, device_id, client_session_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, + `CREATE TABLE IF NOT EXISTS stat_social_identity_day ( + app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, + device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, resolved_user_id BIGINT NOT NULL DEFAULT 0, + resolution_state VARCHAR(16) NOT NULL DEFAULT 'unresolved', + resolution_source VARCHAR(32) NOT NULL DEFAULT 'client_session', updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, stat_tz, stat_day, device_id), + KEY idx_social_identity_day_user (app_code, stat_tz, stat_day, resolved_user_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE IF NOT EXISTS stat_recharge_day_payers ( app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, @@ -381,6 +411,10 @@ func (r *Repository) Migrate(ctx context.Context) error { } alterStatements := []string{ `ALTER TABLE app_tracking_events ADD COLUMN identity_key VARCHAR(160) GENERATED ALWAYS AS (CASE WHEN user_id > 0 THEN CONCAT('u:', user_id) ELSE CONCAT('d:', device_id) END) VIRTUAL AFTER device_id`, + // INSTANT only updates metadata for these defaulted columns. MySQL 8.4 + // rejects an explicit LOCK=NONE with INSTANT, so retain the default lock. + `ALTER TABLE app_tracking_events ADD COLUMN client_session_id VARCHAR(160) NOT NULL DEFAULT '', ALGORITHM=INSTANT`, + `ALTER TABLE app_tracking_events ADD COLUMN auth_session_id VARCHAR(160) NOT NULL DEFAULT '', ALGORITHM=INSTANT`, `ALTER TABLE stat_app_day_country ADD COLUMN paid_users BIGINT NOT NULL DEFAULT 0 AFTER active_users`, `ALTER TABLE stat_app_day_country ADD COLUMN game_refund BIGINT NOT NULL DEFAULT 0 AFTER game_payout`, `ALTER TABLE stat_app_day_country ADD COLUMN lucky_gift_payout BIGINT NOT NULL DEFAULT 0 AFTER lucky_gift_turnover`, @@ -765,27 +799,29 @@ type SelfGameH5Event struct { } type AppTrackingEvent struct { - AppCode string - EventID string - EventName string - EventType string - Screen string - TargetType string - TargetID string - UserID int64 - DeviceID string - SessionID string - Platform string - AppVersion string - Language string - Timezone string - CountryID int64 - RegionID int64 - DurationMS int64 - Success bool - ErrorCode string - Properties json.RawMessage - OccurredAtMS int64 + AppCode string + EventID string + EventName string + EventType string + Screen string + TargetType string + TargetID string + UserID int64 + DeviceID string + SessionID string + ClientSessionID string + AuthSessionID string + Platform string + AppVersion string + Language string + Timezone string + CountryID int64 + RegionID int64 + DurationMS int64 + Success bool + ErrorCode string + Properties json.RawMessage + OccurredAtMS int64 } type AppTrackingReportResult struct { @@ -1664,6 +1700,28 @@ func (r *Repository) ConsumeSelfGameH5Event(ctx context.Context, event SelfGameH } func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) { + var lastResult AppTrackingReportResult + var lastErr error + for attempt := 0; attempt < 5; attempt++ { + lastResult, lastErr = r.consumeAppTrackingEventsOnce(ctx, events) + if lastErr == nil || !isRetryableMySQLError(lastErr) { + return lastResult, lastErr + } + // Identity rows add one transactional lock per device. Concurrent batches + // can still deadlock with unrelated aggregate rows, so retry the entire + // idempotent batch just like owner-fact consumers do. + timer := time.NewTimer(time.Duration(attempt+1) * 50 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + return AppTrackingReportResult{}, ctx.Err() + case <-timer.C: + } + } + return lastResult, lastErr +} + +func (r *Repository) consumeAppTrackingEventsOnce(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) { if r == nil || r.db == nil { return AppTrackingReportResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") } @@ -1672,25 +1730,37 @@ func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppT if !apptracking.ValidBatchSize(len(events)) { return AppTrackingReportResult{}, xerr.New(xerr.InvalidArgument, "app tracking events batch is invalid") } + // Device locks live until this batch commits. A stable device/event order + // makes two overlapping batches acquire them consistently and removes the + // common A->B / B->A deadlock without changing the idempotent response. + orderedEvents := append([]AppTrackingEvent(nil), events...) + sort.SliceStable(orderedEvents, func(i, j int) bool { + leftDevice, rightDevice := strings.TrimSpace(orderedEvents[i].DeviceID), strings.TrimSpace(orderedEvents[j].DeviceID) + if leftDevice != rightDevice { + return leftDevice < rightDevice + } + return strings.TrimSpace(orderedEvents[i].EventID) < strings.TrimSpace(orderedEvents[j].EventID) + }) tx, err := r.db.BeginTx(ctx, nil) if err != nil { return AppTrackingReportResult{}, err } defer func() { _ = tx.Rollback() }() - for _, event := range events { + for _, event := range orderedEvents { normalized, propertiesJSON, err := normalizeAppTrackingEvent(ctx, event, nowMS) if err != nil { return AppTrackingReportResult{}, err } affected, err := insertUnique(ctx, tx, ` - INSERT IGNORE INTO app_tracking_events ( - app_code, event_id, event_name, event_type, screen, target_type, target_id, - user_id, device_id, session_id, platform, app_version, language, timezone, - country_id, region_id, duration_ms, success, error_code, properties_json, - occurred_at_ms, received_at_ms, stat_day - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?) - `, normalized.AppCode, normalized.EventID, normalized.EventName, normalized.EventType, normalized.Screen, + INSERT IGNORE INTO app_tracking_events ( + app_code, event_id, event_name, event_type, screen, target_type, target_id, + user_id, device_id, session_id, client_session_id, auth_session_id, platform, app_version, language, timezone, + country_id, region_id, duration_ms, success, error_code, properties_json, + occurred_at_ms, received_at_ms, stat_day + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?) + `, normalized.AppCode, normalized.EventID, normalized.EventName, normalized.EventType, normalized.Screen, normalized.TargetType, normalized.TargetID, normalized.UserID, normalized.DeviceID, normalized.SessionID, + normalized.ClientSessionID, normalized.AuthSessionID, normalized.Platform, normalized.AppVersion, normalized.Language, normalized.Timezone, normalized.CountryID, normalized.RegionID, normalized.DurationMS, normalized.Success, normalized.ErrorCode, propertiesJSON, normalized.OccurredAtMS, nowMS, statDay(normalized.OccurredAtMS)) @@ -1699,6 +1769,11 @@ func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppT } if affected > 0 { result.Stored++ + // 身份观察和原始埋点、旧宽表必须在同一事务提交；否则页面可能短暂看到 + // 新 d:/u: 行却仍使用旧解析关系，造成 DAU 与留存随请求时点抖动。 + if err := applySocialTrackingIdentity(ctx, tx, normalized, nowMS); err != nil { + return AppTrackingReportResult{}, err + } for _, scope := range statDayScopesFromContext(ctx, normalized.OccurredAtMS) { delta, ok := socialDeltaForAppTrackingEvent(normalized, scope, nowMS) if !ok { @@ -2370,6 +2445,14 @@ func normalizeAppTrackingEvent(ctx context.Context, event AppTrackingEvent, rece if err != nil { return AppTrackingEvent{}, "", err } + event.ClientSessionID, err = normalizeAppTrackingText(event.ClientSessionID, apptracking.MaxSessionIDBytes) + if err != nil { + return AppTrackingEvent{}, "", err + } + event.AuthSessionID, err = normalizeAppTrackingText(event.AuthSessionID, apptracking.MaxSessionIDBytes) + if err != nil { + return AppTrackingEvent{}, "", err + } event.Platform, err = normalizeAppTrackingText(strings.ToLower(event.Platform), apptracking.MaxPlatformBytes) if err != nil { return AppTrackingEvent{}, "", err diff --git a/services/statistics-service/internal/storage/mysql/social_identity.go b/services/statistics-service/internal/storage/mysql/social_identity.go new file mode 100644 index 00000000..9a0985f9 --- /dev/null +++ b/services/statistics-service/internal/storage/mysql/social_identity.go @@ -0,0 +1,230 @@ +package mysql + +import ( + "context" + "database/sql" + "strings" + + "hyapp/pkg/appcode" +) + +const missingClientSessionID = "__missing_client_session__" + +type socialIdentityDayKey struct { + StatTZ string + StatDay string +} + +// applySocialTrackingIdentity records only identity evidence. It deliberately +// never moves or rewrites stat_social_user_day rows: that table contains additive +// money and duration fields which cannot be safely undone when events arrive out +// of order. Query-time canonicalization consumes the small day-resolution table +// and leaves the financial read model untouched. +func applySocialTrackingIdentity(ctx context.Context, tx *sql.Tx, event AppTrackingEvent, nowMS int64) error { + app := appcode.Normalize(event.AppCode) + deviceID := strings.TrimSpace(event.DeviceID) + clientSessionID := strings.TrimSpace(event.ClientSessionID) + if app == "" || deviceID == "" { + return nil + } + + if event.UserID <= 0 { + // Only events that actually create a d: row may participate in the + // device-day resolution. page_open and failed actions otherwise introduce + // phantom unresolved sessions that permanently block a valid login merge. + contributesSocialRow := false + for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) { + if _, ok := socialDeltaForAppTrackingEvent(event, scope, nowMS); ok { + contributesSocialRow = true + break + } + } + if !contributesSocialRow { + return nil + } + } else if clientSessionID == "" { + // Old authenticated clients still produce a direct u: fact, but without + // the process session they cannot link any earlier anonymous row. Return + // before taking a device lock on this high-volume compatibility path. + return nil + } + if err := lockSocialIdentityDevice(ctx, tx, app, deviceID, nowMS); err != nil { + return err + } + + if event.UserID <= 0 { + // Old clients do not send client_session_id. A single unresolved sentinel + // per device/day blocks a different resolved session from swallowing those + // anonymous events; conservative under-count correction is safer than a + // false account merge on shared devices. + if clientSessionID == "" { + clientSessionID = missingClientSessionID + } + for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) { + affected, err := insertUnique(ctx, tx, ` + INSERT IGNORE INTO stat_social_identity_session_day ( + app_code, stat_tz, stat_day, device_id, client_session_id, first_seen_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?) + `, app, scope.tz, scope.day, deviceID, clientSessionID, event.OccurredAtMS, nowMS) + if err != nil { + return err + } + if affected > 0 { + if err := recomputeSocialIdentityDay(ctx, tx, app, scope.tz, scope.day, deviceID, nowMS); err != nil { + return err + } + } + } + return nil + } + + inserted, err := insertUnique(ctx, tx, ` + INSERT IGNORE INTO stat_social_identity_session_user ( + app_code, device_id, client_session_id, user_id, first_seen_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?) + `, app, deviceID, clientSessionID, normalizeID(event.UserID), event.OccurredAtMS, nowMS) + if err != nil || inserted == 0 { + return err + } + + // A session can cross a Beijing/UTC midnight. When its first authenticated + // event arrives, recompute every earlier anonymous day observed for that same + // device+client session so D1/D3/D7 retention sees one canonical account. + rows, err := tx.QueryContext(ctx, ` + SELECT stat_tz, DATE_FORMAT(stat_day, '%Y-%m-%d') + FROM stat_social_identity_session_day + WHERE app_code = ? AND device_id = ? AND client_session_id = ? + ORDER BY stat_tz, stat_day + FOR UPDATE + `, app, deviceID, clientSessionID) + if err != nil { + return err + } + days := make([]socialIdentityDayKey, 0, 2) + for rows.Next() { + var key socialIdentityDayKey + if err := rows.Scan(&key.StatTZ, &key.StatDay); err != nil { + _ = rows.Close() + return err + } + days = append(days, key) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return err + } + if err := rows.Close(); err != nil { + return err + } + for _, day := range days { + if err := recomputeSocialIdentityDay(ctx, tx, app, day.StatTZ, day.StatDay, deviceID, nowMS); err != nil { + return err + } + } + return nil +} + +func recomputeSocialIdentityDay(ctx context.Context, tx *sql.Tx, app, statTZ, statDay, deviceID string, nowMS int64) error { + rows, err := tx.QueryContext(ctx, ` + SELECT d.client_session_id, u.user_id + FROM stat_social_identity_session_day d + LEFT JOIN stat_social_identity_session_user u + ON u.app_code = d.app_code + AND u.device_id = d.device_id + AND u.client_session_id = d.client_session_id + WHERE d.app_code = ? AND d.stat_tz = ? AND d.stat_day = ? AND d.device_id = ? + ORDER BY d.client_session_id, u.user_id + FOR UPDATE + `, app, normalizeStatTZ(statTZ), statDay, deviceID) + if err != nil { + return err + } + sessionUsers := map[string]map[int64]struct{}{} + for rows.Next() { + var sessionID string + var userID sql.NullInt64 + if err := rows.Scan(&sessionID, &userID); err != nil { + _ = rows.Close() + return err + } + if sessionUsers[sessionID] == nil { + sessionUsers[sessionID] = map[int64]struct{}{} + } + if userID.Valid && userID.Int64 > 0 { + sessionUsers[sessionID][userID.Int64] = struct{}{} + } + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return err + } + if err := rows.Close(); err != nil { + return err + } + + state, resolvedUserID := "unresolved", int64(0) + resolutionSource := "client_session" + ambiguous := false + unresolved := false + for _, users := range sessionUsers { + if len(users) == 0 { + unresolved = true + continue + } + if len(users) > 1 { + ambiguous = true + continue + } + for userID := range users { + if resolvedUserID == 0 { + resolvedUserID = userID + } else if resolvedUserID != userID { + ambiguous = true + } + } + } + if ambiguous { + state, resolvedUserID = "ambiguous", 0 + } else if !unresolved && len(sessionUsers) > 0 && resolvedUserID > 0 { + state = "resolved" + } else { + resolvedUserID = 0 + } + if len(sessionUsers) == 1 { + if _, onlyMissingSession := sessionUsers[missingClientSessionID]; onlyMissingSession { + // A legacy client cannot establish an exact session bridge. Keep that + // evidence conservative, but let a later bounded historical replay + // replace it instead of giving it permanent client-session authority. + resolutionSource = "missing_client_session" + } + } + + _, err = tx.ExecContext(ctx, ` + INSERT INTO stat_social_identity_day ( + app_code, stat_tz, stat_day, device_id, resolved_user_id, + resolution_state, resolution_source, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + resolved_user_id = VALUES(resolved_user_id), + resolution_state = VALUES(resolution_state), + resolution_source = VALUES(resolution_source), + updated_at_ms = VALUES(updated_at_ms) + `, app, normalizeStatTZ(statTZ), statDay, deviceID, resolvedUserID, state, resolutionSource, nowMS) + return err +} + +func lockSocialIdentityDevice(ctx context.Context, tx *sql.Tx, app, deviceID string, nowMS int64) error { + if _, err := tx.ExecContext(ctx, ` + INSERT IGNORE INTO stat_social_identity_device_lock (app_code, device_id, created_at_ms) + VALUES (?, ?, ?) + `, appcode.Normalize(app), strings.TrimSpace(deviceID), nowMS); err != nil { + return err + } + var lockedDeviceID string + return tx.QueryRowContext(ctx, ` + SELECT device_id + FROM stat_social_identity_device_lock + WHERE app_code = ? AND device_id = ? + FOR UPDATE + `, appcode.Normalize(app), strings.TrimSpace(deviceID)).Scan(&lockedDeviceID) +} diff --git a/services/statistics-service/internal/storage/mysql/social_requirements_query.go b/services/statistics-service/internal/storage/mysql/social_requirements_query.go index 7ffeb34e..39e42221 100644 --- a/services/statistics-service/internal/storage/mysql/social_requirements_query.go +++ b/services/statistics-service/internal/storage/mysql/social_requirements_query.go @@ -101,6 +101,9 @@ type socialRequirementUserDay struct { StatDay string Subject string UserID int64 + DirectUser bool + DeviceID string + CountryID int64 RegionID int64 UserRole string AppFirstOpen int64 @@ -235,13 +238,19 @@ func (r *Repository) QuerySocialRequirements(ctx context.Context, query SocialRe startDay := statDayIn(startMS, statTZ) endDay := statDayIn(endMS-1, statTZ) days := socialDayRange(startDay, endDay) - // 留存指标需要读取 cohort 日之后的 D30 行；这些仍是本服务已沉淀的用户日宽表，不回查 owner service。 - rows, err := r.querySocialUserDays(ctx, app, statTZ, startDay, addStatDays(endDay, 30), query.CountryID, query.RegionID, query.RegionIDs) + sections := socialRequestedSections(query.Section) + lookaheadDays := socialRetentionLookaheadDays(sections) + // 只有 P0 次留或 P2 留存需要读取查询结束日之后的观察行；单独查看活跃、 + // 营收、送礼、游戏时不再固定多扫 30 天的 50 列用户日宽表。 + allRows, err := r.querySocialUserDays(ctx, app, statTZ, startDay, addStatDays(endDay, lookaheadDays), 0, 0, nil) if err != nil { return SocialRequirements{}, err } - byDay, bySubjectDay := indexSocialRows(rows) - sections := socialRequestedSections(query.Section) + rows := filterSocialRequirementUserDays(allRows, query.CountryID, query.RegionID, query.RegionIDs) + byDay, _ := indexSocialRows(rows) + // 区域只筛 cohort 基准日；D+N 观察日保留完整 canonical 活跃索引。 + // 用户次日跨区仍算回来，不能因为目标日维度不同被误判为流失。 + _, bySubjectDay := indexSocialRows(allRows) var gameBreakdown []SocialRequirementGame for _, section := range sections { if section == SocialSectionGame { @@ -269,7 +278,10 @@ func (r *Repository) QuerySocialRequirements(ctx context.Context, query SocialRe for _, day := range days { agg := aggregateSocialRows(byDay[day], section, query.UserRole, query.PayerType, query.NewUserType) totalAgg.add(agg) - retention := socialRetentionMetrics(day, byDay[day], bySubjectDay, section, query.UserRole, query.PayerType, query.NewUserType) + retention := map[string]retentionMetric{} + if socialSectionNeedsRetention(section) { + retention = socialRetentionMetrics(day, byDay[day], bySubjectDay, section, query.UserRole, query.PayerType, query.NewUserType) + } mergeRetentionMetrics(totalRetention, retention) sectionRows = append(sectionRows, SocialRequirementRow{ StatDay: day, @@ -326,26 +338,31 @@ func socialAverageRow(rows []SocialRequirementRow) SocialRequirementRow { return SocialRequirementRow{StatDay: "average", Label: "平均值", Metrics: metrics} } +func socialRetentionLookaheadDays(sections []string) int { + lookahead := 0 + for _, section := range sections { + switch section { + case SocialSectionRetention: + return 30 + case SocialSectionNewUsers: + lookahead = 1 + } + } + return lookahead +} + +func socialSectionNeedsRetention(section string) bool { + return section == SocialSectionRetention || section == SocialSectionNewUsers +} + func (r *Repository) querySocialUserDays(ctx context.Context, app string, statTZ string, startDay string, endDay string, countryID int64, regionID int64, regionIDs []int64) ([]socialRequirementUserDay, error) { args := []any{appcode.Normalize(app), normalizeStatTZ(statTZ), startDay, endDay} conditions := []string{"app_code = ?", "stat_tz = ?", "stat_day BETWEEN ? AND ?"} - if countryID > 0 { - conditions = append(conditions, "country_id = ?") - args = append(args, countryID) - } - if regionID > 0 { - conditions = append(conditions, "region_id = ?") - args = append(args, regionID) - } else if ids := normalizePositiveIDs(regionIDs); len(ids) > 0 { - placeholders := make([]string, 0, len(ids)) - for _, id := range ids { - placeholders = append(placeholders, "?") - args = append(args, id) - } - conditions = append(conditions, "region_id IN ("+strings.Join(placeholders, ",")+")") - } + // 匿名行通常没有国家/区域，而登录后的 u: 行才带用户维度。必须先在完整 + // app+day 小宽表范围内完成 canonical 合并，再应用维度过滤；提前在 SQL 里 + // 过滤会丢掉同一用户登录前的活跃，并让区域 DAU 再次出现 d/u 口径偏差。 rows, err := r.db.QueryContext(ctx, ` - SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), subject_key, user_id, region_id, user_role, + SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), subject_key, user_id, device_id, country_id, region_id, user_role, app_first_open, registered_success, active_user, room_join_success, room_join_fail, mic_up_success, gift_sent, normal_gift_sent, lucky_gift_sent, super_lucky_gift_sent, recharge_user, first_recharge_user, repeat_recharge_user, game_bet_user, self_game_bet_user, @@ -367,7 +384,7 @@ func (r *Repository) querySocialUserDays(ctx context.Context, app string, statTZ out := []socialRequirementUserDay{} for rows.Next() { var item socialRequirementUserDay - if err := rows.Scan(&item.StatDay, &item.Subject, &item.UserID, &item.RegionID, &item.UserRole, + if err := rows.Scan(&item.StatDay, &item.Subject, &item.UserID, &item.DeviceID, &item.CountryID, &item.RegionID, &item.UserRole, &item.AppFirstOpen, &item.Registered, &item.Active, &item.RoomJoinOK, &item.RoomJoinNG, &item.MicUp, &item.Gift, &item.NormalGift, &item.LuckyGift, &item.SuperGift, &item.Recharge, &item.FirstRecharge, &item.RepeatRecharge, &item.GameBet, &item.SelfGameBet, @@ -380,19 +397,210 @@ func (r *Repository) querySocialUserDays(ctx context.Context, app string, statTZ &item.OtherConsumeCoin, &item.OutputCoin); err != nil { return nil, err } - // 在房时长口径：客户端 room_stay 埋点（杀进程/闪退整段丢失）和服务端进退房事实 room_online_ms - // 双轨并存，取 GREATEST 避免切换期数据断层；埋点按 app 版本灰度下线后该口径自然收敛到服务端事实。 - if item.RoomStayMS < item.RoomOnlineMS { - item.RoomStayMS = item.RoomOnlineMS - } - // 在房时长下界修正：上麦必在房内，麦时长（服务端事实，权威）是房时长的天然下界。 - // 服务端在房事实上线后保留该钳制作为兜底（如回灌历史区间尚无服务端事实）。 - if item.RoomStayMS < item.MicStayMS { - item.RoomStayMS = item.MicStayMS - } + item.DirectUser = item.UserID > 0 out = append(out, item) } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + identityByDeviceDay, err := r.querySocialIdentityDays(ctx, app, statTZ, startDay, endDay) + if err != nil { + return nil, err + } + return canonicalizeSocialUserDays(out, identityByDeviceDay, countryID, regionID, regionIDs), nil +} + +type socialIdentityDeviceDayKey struct { + StatDay string + DeviceID string +} + +func (r *Repository) querySocialIdentityDays(ctx context.Context, app, statTZ, startDay, endDay string) (map[socialIdentityDeviceDayKey]int64, error) { + rows, err := r.db.QueryContext(ctx, ` + SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), device_id, resolved_user_id + FROM stat_social_identity_day + WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ? + AND resolution_state = 'resolved' AND resolved_user_id > 0 + `, appcode.Normalize(app), normalizeStatTZ(statTZ), startDay, endDay) + if err != nil { + return nil, err + } + defer rows.Close() + resolved := map[socialIdentityDeviceDayKey]int64{} + for rows.Next() { + var day, deviceID string + var userID int64 + if err := rows.Scan(&day, &deviceID, &userID); err != nil { + return nil, err + } + if deviceID = strings.TrimSpace(deviceID); deviceID != "" && userID > 0 { + resolved[socialIdentityDeviceDayKey{StatDay: day, DeviceID: deviceID}] = userID + } + } + return resolved, rows.Err() +} + +// canonicalizeSocialUserDays keeps stat_social_user_day as the immutable source +// of financial/duration facts and only rewrites identity in memory. This avoids +// destructive primary-key moves and keeps a replay idempotent even when late +// events or an account switch changes the device-day resolution. +func canonicalizeSocialUserDays(rows []socialRequirementUserDay, resolved map[socialIdentityDeviceDayKey]int64, countryID, regionID int64, regionIDs []int64) []socialRequirementUserDay { + merged := map[string]socialRequirementUserDay{} + firstOpenDevices := map[string]map[string]struct{}{} + for _, row := range rows { + if row.UserID <= 0 { + if userID := resolved[socialIdentityDeviceDayKey{StatDay: row.StatDay, DeviceID: strings.TrimSpace(row.DeviceID)}]; userID > 0 { + row.UserID = userID + row.Subject = "u:" + strconv.FormatInt(userID, 10) + } + } + key := row.StatDay + "\x00" + row.Subject + if row.AppFirstOpen > 0 { + deviceID := strings.TrimSpace(row.DeviceID) + if deviceID == "" { + // Old u: rows may not retain a device. Use the physical subject as a + // stable one-count fallback; known d/u rows still dedupe by device ID. + deviceID = "subject:" + row.Subject + } + if firstOpenDevices[key] == nil { + firstOpenDevices[key] = map[string]struct{}{} + } + firstOpenDevices[key][deviceID] = struct{}{} + } + if current, ok := merged[key]; ok { + merged[key] = mergeSocialRequirementUserDay(current, row) + } else { + merged[key] = row + } + } + out := make([]socialRequirementUserDay, 0, len(merged)) + for key, row := range merged { + row.AppFirstOpen = int64(len(firstOpenDevices[key])) + // Client room_stay and authoritative server room_online are parallel + // measurements, not additive copies. Clamp only after d:/u: fields have + // been summed so the same interval is not counted twice. + if row.RoomStayMS < row.RoomOnlineMS { + row.RoomStayMS = row.RoomOnlineMS + } + if row.RoomStayMS < row.MicStayMS { + row.RoomStayMS = row.MicStayMS + } + out = append(out, row) + } + sort.Slice(out, func(i, j int) bool { + if out[i].StatDay != out[j].StatDay { + return out[i].StatDay < out[j].StatDay + } + return out[i].Subject < out[j].Subject + }) + return filterSocialRequirementUserDays(out, countryID, regionID, regionIDs) +} + +func mergeSocialRequirementUserDay(current, next socialRequirementUserDay) socialRequirementUserDay { + if current.UserID == 0 { + current.UserID = next.UserID + } + if current.DeviceID == "" { + current.DeviceID = next.DeviceID + } + // Country, region and role are one dimension tuple. A direct same-day u: row + // is authoritative over the d: row even when the anonymous tuple is nonzero; + // copying fields independently would create a country/region pair that never + // existed. No cross-day dimension is borrowed, so history is query-window stable. + if next.DirectUser && !current.DirectUser { + current.CountryID, current.RegionID, current.UserRole = next.CountryID, next.RegionID, next.UserRole + current.DirectUser = true + } else if next.DirectUser == current.DirectUser { + if current.CountryID == 0 && current.RegionID == 0 && (next.CountryID > 0 || next.RegionID > 0) { + current.CountryID, current.RegionID = next.CountryID, next.RegionID + } + if normalizeSocialUserRole(next.UserRole) == "host" { + current.UserRole = next.UserRole + } + } + // canonicalizeSocialUserDays replaces this flag with the distinct known + // device count after all rows merge; MAX is only the empty-device fallback. + current.AppFirstOpen = maxSocialFlag(current.AppFirstOpen, next.AppFirstOpen) + current.Registered = maxSocialFlag(current.Registered, next.Registered) + current.Active = maxSocialFlag(current.Active, next.Active) + current.RoomJoinOK = maxSocialFlag(current.RoomJoinOK, next.RoomJoinOK) + current.RoomJoinNG = maxSocialFlag(current.RoomJoinNG, next.RoomJoinNG) + current.MicUp = maxSocialFlag(current.MicUp, next.MicUp) + current.Gift = maxSocialFlag(current.Gift, next.Gift) + current.NormalGift = maxSocialFlag(current.NormalGift, next.NormalGift) + current.LuckyGift = maxSocialFlag(current.LuckyGift, next.LuckyGift) + current.SuperGift = maxSocialFlag(current.SuperGift, next.SuperGift) + current.Recharge = maxSocialFlag(current.Recharge, next.Recharge) + current.FirstRecharge = maxSocialFlag(current.FirstRecharge, next.FirstRecharge) + current.RepeatRecharge = maxSocialFlag(current.RepeatRecharge, next.RepeatRecharge) + current.GameBet = maxSocialFlag(current.GameBet, next.GameBet) + current.SelfGameBet = maxSocialFlag(current.SelfGameBet, next.SelfGameBet) + current.ProbabilityBet = maxSocialFlag(current.ProbabilityBet, next.ProbabilityBet) + current.Chat = maxSocialFlag(current.Chat, next.Chat) + current.FollowUser = maxSocialFlag(current.FollowUser, next.FollowUser) + current.FollowRoom = maxSocialFlag(current.FollowRoom, next.FollowRoom) + current.GiftPanel = maxSocialFlag(current.GiftPanel, next.GiftPanel) + current.ProbabilityPanel = maxSocialFlag(current.ProbabilityPanel, next.ProbabilityPanel) + current.AppStayMS += next.AppStayMS + current.RoomStayMS += next.RoomStayMS + current.RoomOnlineMS += next.RoomOnlineMS + current.MicStayMS += next.MicStayMS + current.RechargeCount += next.RechargeCount + current.RechargeUSDMinor += next.RechargeUSDMinor + current.GoogleRechargeUSDMinor += next.GoogleRechargeUSDMinor + current.ThirdPartyRechargeUSDMinor += next.ThirdPartyRechargeUSDMinor + current.CoinSellerRechargeUSDMinor += next.CoinSellerRechargeUSDMinor + current.FirstRechargeUSDMinor += next.FirstRechargeUSDMinor + current.RepeatRechargeUSDMinor += next.RepeatRechargeUSDMinor + current.NormalGiftCoin += next.NormalGiftCoin + current.LuckyGiftCoin += next.LuckyGiftCoin + current.SuperGiftCoin += next.SuperGiftCoin + current.LuckyGiftPayoutCoin += next.LuckyGiftPayoutCoin + current.SuperGiftPayoutCoin += next.SuperGiftPayoutCoin + current.GameBetCoin += next.GameBetCoin + current.GameBetCount += next.GameBetCount + current.SelfGameBetCoin += next.SelfGameBetCoin + current.SelfGameBetCount += next.SelfGameBetCount + current.ProbabilityBetCoin += next.ProbabilityBetCoin + current.ProbabilityBetCount += next.ProbabilityBetCount + current.OtherConsumeCoin += next.OtherConsumeCoin + current.OutputCoin += next.OutputCoin + return current +} + +func maxSocialFlag(left, right int64) int64 { + if right > left { + return right + } + return left +} + +func filterSocialRequirementUserDays(rows []socialRequirementUserDay, countryID, regionID int64, regionIDs []int64) []socialRequirementUserDay { + allowedRegions := map[int64]struct{}{} + if regionID <= 0 { + for _, id := range normalizePositiveIDs(regionIDs) { + allowedRegions[id] = struct{}{} + } + } + if countryID <= 0 && regionID <= 0 && len(allowedRegions) == 0 { + return rows + } + out := make([]socialRequirementUserDay, 0, len(rows)) + for _, row := range rows { + if countryID > 0 && row.CountryID != countryID { + continue + } + if regionID > 0 && row.RegionID != regionID { + continue + } + if len(allowedRegions) > 0 { + if _, ok := allowedRegions[row.RegionID]; !ok { + continue + } + } + out = append(out, row) + } + return out } func indexSocialRows(rows []socialRequirementUserDay) (map[string][]socialRequirementUserDay, map[string]map[string]socialRequirementUserDay) { @@ -414,7 +622,9 @@ func aggregateSocialRows(rows []socialRequirementUserDay, section string, role s if !socialRowMatchesSection(row, section, role, payerType, newUserType) { continue } - agg.AppFirstOpen += flagValue(row.AppFirstOpen) + // 下载/首次打开是设备口径；身份归并只消除用户行为的 d/u 重复， + // 不能把同一账号在两台设备上的两次安装压成 1。 + agg.AppFirstOpen += normalizeID(row.AppFirstOpen) agg.Registered += flagValue(row.Registered) if flagValue(row.Registered) > 0 && normalizeSocialUserRole(row.UserRole) == "host" { agg.NewHostUsers++ @@ -723,23 +933,23 @@ func socialSectionMetrics(section string, agg socialRequirementAggregate, retent } default: return map[string]any{ - "app_download_users": agg.AppFirstOpen, - "registered_users": agg.Registered, - "new_host_users": agg.NewHostUsers, - "new_normal_users": agg.NewNormalUsers, - "registration_success_rate": ratio(agg.Registered, agg.AppFirstOpen), - "room_join_success_users": agg.RegisteredRoomJoinOK, - "room_join_fail_users": agg.RegisteredRoomJoinNG, - "room_join_success_rate": ratio(agg.RegisteredRoomJoinOK, agg.Registered), - "room_join_fail_rate": ratio(agg.RegisteredRoomJoinNG, agg.Registered), - "mic_up_users": agg.RegisteredMicUp, - "mic_up_rate": ratio(agg.RegisteredMicUp, agg.RegisteredRoomJoinOK), - "gift_users": agg.RegisteredRoomGift, - "gift_rate": ratio(agg.RegisteredRoomGift, agg.RegisteredRoomJoinOK), - "recharge_users": agg.RegisteredRecharge, - "recharge_rate": ratio(agg.RegisteredRecharge, agg.Registered), - "game_users": agg.RegisteredGame, - "game_conversion_rate": ratio(agg.RegisteredGame, agg.Registered), + "app_download_users": agg.AppFirstOpen, + "registered_users": agg.Registered, + "new_host_users": agg.NewHostUsers, + "new_normal_users": agg.NewNormalUsers, + "registration_success_rate": ratio(agg.Registered, agg.AppFirstOpen), + "room_join_success_users": agg.RegisteredRoomJoinOK, + "room_join_fail_users": agg.RegisteredRoomJoinNG, + "room_join_success_rate": ratio(agg.RegisteredRoomJoinOK, agg.Registered), + "room_join_fail_rate": ratio(agg.RegisteredRoomJoinNG, agg.Registered), + "mic_up_users": agg.RegisteredMicUp, + "mic_up_rate": ratio(agg.RegisteredMicUp, agg.RegisteredRoomJoinOK), + "gift_users": agg.RegisteredRoomGift, + "gift_rate": ratio(agg.RegisteredRoomGift, agg.RegisteredRoomJoinOK), + "recharge_users": agg.RegisteredRecharge, + "recharge_rate": ratio(agg.RegisteredRecharge, agg.Registered), + "game_users": agg.RegisteredGame, + "game_conversion_rate": ratio(agg.RegisteredGame, agg.Registered), // P0 全部限定当天注册的新用户：时长按新用户累计、除以对应行为的新用户数。 "avg_app_stay_min": minutesPerUser(agg.RegisteredAppStayMS, agg.Registered), "avg_room_stay_min": minutesPerUser(agg.RegisteredRoomStayMS, agg.RegisteredRoomJoinOK),