hyapp-server/services/statistics-service/cmd/backfill-social-identities/main.go

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