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 "" }