package main

import (
	"context"
	"database/sql"
	"flag"
	"fmt"
	"log"
	"os"
	"sort"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"hyapp/pkg/appcode"
)

type countryRegion struct {
	countryID int64
	regionID  int64
}

type appDayKey struct {
	day       string
	countryID int64
	regionID  int64
}

type appDayAgg struct {
	newUsers                   int64
	activeUsers                int64
	paidUsers                  int64
	rechargeUSDMinor           int64
	newUserRechargeUSDMinor    int64
	coinSellerRechargeUSDMinor int64
	coinSellerStockCoin        int64
	mifaPayRechargeUSDMinor    int64
	googleRechargeUSDMinor     int64
	coinSellerTransferCoin     int64
	coinTotal                  int64
	consumedCoin               int64
	platformGrantCoin          int64
	manualGrantCoin            int64
	salaryUSDMinor             int64
	salaryTransferCoin         int64
	micOnlineMS                int64
	micOnlineUsers             int64
}

type userRegistrationRow struct {
	userID         int64
	registeredDay  string
	dim            countryRegion
	registeredAtMS int64
}

type payerRow struct {
	day     string
	userID  int64
	dim     countryRegion
	firstMS int64
}

type activeRow struct {
	day     string
	userID  int64
	dim     countryRegion
	firstMS int64
}

type micRow struct {
	day         string
	userID      int64
	dim         countryRegion
	micOnlineMS int64
}

type rebuildState struct {
	appDays             map[appDayKey]*appDayAgg
	registrations       []userRegistrationRow
	registeredByUserDay map[string]struct{}
	payersByKey         map[string]payerRow
	activeByKey         map[string]activeRow
	micByKey            map[string]micRow
	userIDs             map[int64]struct{}
	userDims            map[int64]countryRegion
	knownUsers          map[int64]struct{}
}

func main() {
	ctx := context.Background()
	var (
		appFlag       = flag.String("app", envDefault("APP_CODE", "lalu"), "app_code to backfill")
		tzFlag        = flag.String("timezone", envDefault("BACKFILL_TIMEZONE", "UTC"), "day boundary timezone, e.g. UTC or Asia/Shanghai")
		startFlag     = flag.String("start-day", os.Getenv("START_DAY"), "inclusive yyyy-mm-dd in timezone; defaults to last 7 days")
		endFlag       = flag.String("end-day", os.Getenv("END_DAY"), "inclusive yyyy-mm-dd in timezone; defaults to today")
		statsDSNFlag  = flag.String("statistics-dsn", firstNonEmpty(os.Getenv("STATISTICS_MYSQL_DSN"), os.Getenv("MYSQL_DSN")), "statistics mysql dsn")
		userDSNFlag   = flag.String("user-dsn", os.Getenv("USER_MYSQL_DSN"), "user-service mysql dsn")
		walletDSNFlag = flag.String("wallet-dsn", os.Getenv("WALLET_MYSQL_DSN"), "wallet-service mysql dsn")
		applyFlag     = flag.Bool("apply", envBool("APPLY"), "apply replacement; default dry-run")
	)
	flag.Parse()

	if strings.TrimSpace(*statsDSNFlag) == "" || strings.TrimSpace(*userDSNFlag) == "" || strings.TrimSpace(*walletDSNFlag) == "" {
		log.Fatal("statistics-dsn, user-dsn and wallet-dsn are required")
	}
	loc, err := time.LoadLocation(strings.TrimSpace(*tzFlag))
	if err != nil {
		log.Fatalf("load timezone: %v", err)
	}
	statTZ, err := normalizeBackfillStatTZ(loc.String())
	if err != nil {
		log.Fatal(err)
	}
	startDay, endDay, windowStartMS, windowEndMS, err := resolveWindow(*startFlag, *endFlag, loc)
	if err != nil {
		log.Fatal(err)
	}
	app := appcode.Normalize(*appFlag)

	statsDB := mustOpen(ctx, *statsDSNFlag)
	defer statsDB.Close()
	userDB := mustOpen(ctx, *userDSNFlag)
	defer userDB.Close()
	walletDB := mustOpen(ctx, *walletDSNFlag)
	defer walletDB.Close()

	state := &rebuildState{
		appDays:             map[appDayKey]*appDayAgg{},
		registeredByUserDay: map[string]struct{}{},
		payersByKey:         map[string]payerRow{},
		activeByKey:         map[string]activeRow{},
		micByKey:            map[string]micRow{},
		userIDs:             map[int64]struct{}{},
		userDims:            map[int64]countryRegion{},
		knownUsers:          map[int64]struct{}{},
	}
	// 第一阶段只收集窗口内会参与重算的 user_id，后续统一批量加载用户国家和区域，避免按充值/活跃记录逐条访问 user-service 数据库。
	if err := collectRegistrationUsers(ctx, userDB, state, app, windowStartMS, windowEndMS); err != nil {
		log.Fatalf("collect registrations: %v", err)
	}
	if err := collectActiveUserIDs(ctx, statsDB, state, app, statTZ, startDay, endDay); err != nil {
		log.Fatalf("collect active ids: %v", err)
	}
	if err := collectRechargeUserIDs(ctx, walletDB, state, app, windowStartMS, windowEndMS); err != nil {
		log.Fatalf("collect recharge ids: %v", err)
	}
	if err := collectReportMetricUserIDs(ctx, walletDB, userDB, state, app, startDay, endDay, windowEndMS); err != nil {
		log.Fatalf("collect report metric ids: %v", err)
	}
	if err := loadUserDims(ctx, userDB, state, app); err != nil {
		log.Fatalf("load user dims: %v", err)
	}
	if err := rebuildRegistrations(ctx, userDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
		log.Fatalf("rebuild registrations: %v", err)
	}
	if err := rebuildActive(ctx, statsDB, state, app, statTZ, startDay, endDay); err != nil {
		log.Fatalf("rebuild active: %v", err)
	}
	if err := rebuildWalletRecharge(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
		log.Fatalf("rebuild wallet recharge: %v", err)
	}
	if err := rebuildCoinSellerStock(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
		log.Fatalf("rebuild coin seller stock: %v", err)
	}
	if err := rebuildWalletReportMetrics(ctx, walletDB, state, app, startDay, endDay, windowStartMS, windowEndMS, loc); err != nil {
		log.Fatalf("rebuild wallet report metrics: %v", err)
	}
	if err := rebuildSalary(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
		log.Fatalf("rebuild salary: %v", err)
	}
	if err := rebuildMicStats(ctx, userDB, state, app, startDay, endDay); err != nil {
		log.Fatalf("rebuild mic stats: %v", err)
	}

	if err := stageAndMaybeApply(ctx, statsDB, state, app, statTZ, startDay, endDay, *applyFlag); err != nil {
		log.Fatalf("stage/apply: %v", err)
	}
	mode := "dry-run"
	if *applyFlag {
		mode = "applied"
	}
	fmt.Printf("%s app=%s timezone=%s window=%s..%s app_day_rows=%d registrations=%d active=%d payers=%d\n",
		mode, app, statTZ, startDay, endDay, len(state.appDays), len(state.registrations), len(state.activeByKey), len(state.payersByKey))
}

func normalizeBackfillStatTZ(value string) (string, error) {
	switch strings.TrimSpace(value) {
	case "UTC":
		return "UTC", nil
	case "Asia/Shanghai":
		return "Asia/Shanghai", nil
	default:
		return "", fmt.Errorf("unsupported stat_tz %q; only UTC and Asia/Shanghai are queryable", value)
	}
}

func envDefault(key string, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(key)); value != "" {
		return value
	}
	return fallback
}

func envBool(key string) bool {
	switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) {
	case "1", "true", "yes", "y":
		return true
	default:
		return false
	}
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}

func mustOpen(ctx context.Context, dsn string) *sql.DB {
	db, err := sql.Open("mysql", strings.TrimSpace(dsn))
	if err != nil {
		log.Fatal(err)
	}
	if err := db.PingContext(ctx); err != nil {
		log.Fatal(err)
	}
	return db
}

func resolveWindow(rawStart, rawEnd string, loc *time.Location) (string, string, int64, int64, error) {
	now := time.Now().In(loc)
	endDayTime := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
	startDayTime := endDayTime.AddDate(0, 0, -6)
	var err error
	if strings.TrimSpace(rawEnd) != "" {
		endDayTime, err = parseDay(rawEnd, loc)
		if err != nil {
			return "", "", 0, 0, err
		}
	}
	if strings.TrimSpace(rawStart) != "" {
		startDayTime, err = parseDay(rawStart, loc)
		if err != nil {
			return "", "", 0, 0, err
		}
	}
	if startDayTime.After(endDayTime) {
		return "", "", 0, 0, fmt.Errorf("start-day must not be after end-day")
	}
	windowEnd := endDayTime.AddDate(0, 0, 1)
	return formatDay(startDayTime), formatDay(endDayTime), startDayTime.UTC().UnixMilli(), windowEnd.UTC().UnixMilli(), nil
}

func parseDay(value string, loc *time.Location) (time.Time, error) {
	parsed, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(value), loc)
	if err != nil {
		return time.Time{}, err
	}
	return time.Date(parsed.Year(), parsed.Month(), parsed.Day(), 0, 0, 0, 0, loc), nil
}

func dayFromMS(ms int64, loc *time.Location) string {
	return formatDay(time.UnixMilli(ms).In(loc))
}

func formatDay(value time.Time) string {
	return value.Format("2006-01-02")
}

func collectRegistrationUsers(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64) error {
	// 全站机器人和后台快捷账号使用 users 资料参与运营、联调或游戏展示，但不是 App 自然注册用户；补数入口必须和实时 UserRegistered 口径一致排除。
	rows, err := db.QueryContext(ctx, `
		SELECT u.user_id
		FROM users u
		WHERE u.app_code = ?
		  AND u.profile_completed = 1
		  AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) >= ?
		  AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) < ?
		  AND LOWER(COALESCE(u.source, '')) NOT IN ('game_robot', 'quick_account')`,
		app, startMS, endMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var userID int64
		if err := rows.Scan(&userID); err != nil {
			return err
		}
		state.userIDs[userID] = struct{}{}
	}
	return rows.Err()
}

func collectActiveUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app, statTZ, startDay, endDay string) error {
	rows, err := db.QueryContext(ctx, `
		SELECT DISTINCT user_id
		FROM stat_user_day_activity
		WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
		app, statTZ, startDay, endDay)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var userID int64
		if err := rows.Scan(&userID); err != nil {
			return err
		}
		state.userIDs[userID] = struct{}{}
	}
	return rows.Err()
}

func collectRechargeUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64) error {
	for _, query := range []string{
		`SELECT DISTINCT user_id FROM wallet_recharge_records WHERE app_code = ? AND created_at_ms >= ? AND created_at_ms < ?`,
		`SELECT DISTINCT seller_user_id FROM coin_seller_stock_records WHERE app_code = ? AND counts_as_seller_recharge = 1 AND created_at_ms >= ? AND created_at_ms < ?`,
	} {
		rows, err := db.QueryContext(ctx, query, app, startMS, endMS)
		if err != nil {
			return err
		}
		for rows.Next() {
			var userID int64
			if err := rows.Scan(&userID); err != nil {
				rows.Close()
				return err
			}
			state.userIDs[userID] = struct{}{}
		}
		if err := rows.Close(); err != nil {
			return err
		}
	}
	return nil
}

func collectReportMetricUserIDs(ctx context.Context, walletDB *sql.DB, userDB *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64) error {
	queries := []struct {
		db    *sql.DB
		query string
		args  []any
	}{
		{
			db: walletDB,
			query: `
				SELECT DISTINCT e.user_id
				FROM wallet_entries e
				JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id
				WHERE wt.app_code = ? AND e.asset_type IN ('COIN', 'COIN_SELLER_COIN', 'HOST_SALARY_USD', 'AGENCY_SALARY_USD') AND e.created_at_ms < ?
				UNION
				SELECT DISTINCT e.counterparty_user_id
				FROM wallet_entries e
				JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id
				WHERE wt.app_code = ? AND wt.biz_type = 'salary_transfer_to_coin_seller'
				  AND e.asset_type = 'COIN_SELLER_COIN' AND e.counterparty_user_id > 0 AND e.created_at_ms < ?`,
			args: []any{app, windowEndMS, app, windowEndMS},
		},
		{
			db: walletDB,
			query: `
				SELECT DISTINCT user_id
				FROM wallet_accounts
				WHERE app_code = ? AND asset_type = 'COIN' AND available_amount + frozen_amount <> 0`,
			args: []any{app},
		},
		{
			db: walletDB,
			query: `
				SELECT DISTINCT user_id
				FROM host_salary_settlement_records
				WHERE app_code = ? AND status = 'succeeded' AND created_at_ms < ?`,
			args: []any{app, windowEndMS},
		},
		{
			db: walletDB,
			query: `
				SELECT DISTINCT agency_owner_user_id
				FROM host_salary_settlement_records
				WHERE app_code = ? AND status = 'succeeded' AND agency_owner_user_id > 0 AND created_at_ms < ?`,
			args: []any{app, windowEndMS},
		},
		{
			db: userDB,
			query: `
				SELECT DISTINCT user_id
				FROM user_mic_daily_stats
				WHERE app_code = ? AND stat_date BETWEEN ? AND ?`,
			args: []any{app, startDay, endDay},
		},
	}
	for _, item := range queries {
		rows, err := item.db.QueryContext(ctx, item.query, item.args...)
		if err != nil {
			return err
		}
		for rows.Next() {
			var userID int64
			if err := rows.Scan(&userID); err != nil {
				rows.Close()
				return err
			}
			if userID > 0 {
				state.userIDs[userID] = struct{}{}
			}
		}
		if err := rows.Close(); err != nil {
			return err
		}
	}
	return nil
}

func loadUserDims(ctx context.Context, db *sql.DB, state *rebuildState, app string) error {
	userIDs := make([]int64, 0, len(state.userIDs))
	for userID := range state.userIDs {
		userIDs = append(userIDs, userID)
	}
	sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] })
	for start := 0; start < len(userIDs); start += 500 {
		end := start + 500
		if end > len(userIDs) {
			end = len(userIDs)
		}
		placeholders := strings.TrimRight(strings.Repeat("?,", end-start), ",")
		args := []any{app}
		for _, userID := range userIDs[start:end] {
			args = append(args, userID)
		}
		rows, err := db.QueryContext(ctx, `
			SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0)
			FROM users u
			LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country
			WHERE u.app_code = ? AND u.user_id IN (`+placeholders+`)`, args...)
		if err != nil {
			return err
		}
		for rows.Next() {
			var userID int64
			var dim countryRegion
			if err := rows.Scan(&userID, &dim.countryID, &dim.regionID); err != nil {
				rows.Close()
				return err
			}
			state.userDims[userID] = dim
			state.knownUsers[userID] = struct{}{}
		}
		if err := rows.Close(); err != nil {
			return err
		}
	}
	return nil
}

func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
	// 注册国家以资料完成后的 users.country 映射为准；region_id 只作为独立维度写入，不能再兜底成国家。
	// source=game_robot 是后台全站机器人池账号，source=quick_account 是后台快捷建号；两者都不是 App 自然注册，不能进入新增用户、注册 cohort 或留存口径。
	rows, err := db.QueryContext(ctx, `
		SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0),
		       COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) AS registered_at_ms
		FROM users u
		LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country
		WHERE u.app_code = ?
		  AND u.profile_completed = 1
		  AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) >= ?
		  AND COALESCE(NULLIF(u.profile_completed_at_ms, 0), u.created_at_ms) < ?
		  AND LOWER(COALESCE(u.source, '')) NOT IN ('game_robot', 'quick_account')`,
		app, startMS, endMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var row userRegistrationRow
		if err := rows.Scan(&row.userID, &row.dim.countryID, &row.dim.regionID, &row.registeredAtMS); err != nil {
			return err
		}
		row.registeredDay = dayFromMS(row.registeredAtMS, loc)
		state.registrations = append(state.registrations, row)
		state.registeredByUserDay[userDayKey(row.registeredDay, row.userID)] = struct{}{}
		agg := state.ensureAgg(row.registeredDay, row.dim)
		agg.newUsers++
		// 产品口径要求完成注册即算当日活跃；补数时直接把注册事实写入活跃重建集合，
		// 后续房间/游戏活跃若命中同一用户同一天，只会更新更早的 first_active_at_ms，不会重复计数。
		state.addActive(row.registeredDay, row.userID, row.dim, row.registeredAtMS)
	}
	return rows.Err()
}

func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, statTZ, startDay, endDay string) error {
	// 房间/游戏活跃已经被实时消费进对应 stat_tz 的 stat_user_day_activity；回填只读取目标时区，避免用 UTC 活跃去污染中国自然日。
	rows, err := db.QueryContext(ctx, `
		SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), user_id, MIN(first_active_at_ms)
		FROM stat_user_day_activity
		WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?
		GROUP BY stat_day, user_id`,
		app, statTZ, startDay, endDay)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var day string
		var userID, firstMS int64
		if err := rows.Scan(&day, &userID, &firstMS); err != nil {
			return err
		}
		if _, ok := state.knownUsers[userID]; !ok {
			// stat_user_day_activity 是历史活跃读模型，不是用户事实 owner；回填时必须丢弃 user-service 已不存在的孤儿活跃，避免脏事实继续汇总到 country_id=0。
			continue
		}
		state.addActive(day, userID, state.userDims[userID], firstMS)
	}
	return rows.Err()
}

func rebuildWalletRecharge(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
	// wallet_recharge_records 是到账事实表；Google 也会写入该表，payment_orders 只是 provider 审计，不能在回填里再扫一次导致双算。
	// 币商向用户转账只代表金币充值用户，不产生用户 USDT 充值；Google Play 则按记录里的 usd_minor_amount 计入用户充值。
	rows, err := db.QueryContext(ctx, `
		SELECT rr.user_id, rr.recharge_sequence, rr.coin_amount, rr.usd_minor_amount, rr.created_at_ms, wt.biz_type
		FROM wallet_recharge_records rr
		JOIN wallet_transactions wt ON wt.app_code = rr.app_code AND wt.transaction_id = rr.transaction_id
		WHERE rr.app_code = ? AND rr.created_at_ms >= ? AND rr.created_at_ms < ?`,
		app, startMS, endMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var userID, sequence, coinAmount, usdMinor, createdAtMS int64
		var bizType string
		if err := rows.Scan(&userID, &sequence, &coinAmount, &usdMinor, &createdAtMS, &bizType); err != nil {
			return err
		}
		day := dayFromMS(createdAtMS, loc)
		dim := state.userDims[userID]
		state.addPayer(day, userID, dim, createdAtMS)
		agg := state.ensureAgg(day, dim)
		switch strings.ToLower(strings.TrimSpace(bizType)) {
		case "coin_seller_transfer":
			agg.coinSellerTransferCoin += coinAmount
		case "google_play_recharge":
			agg.rechargeUSDMinor += usdMinor
			agg.googleRechargeUSDMinor += usdMinor
		case "mifapay", "v5pay":
			agg.rechargeUSDMinor += usdMinor
			agg.mifaPayRechargeUSDMinor += usdMinor
		default:
			if usdMinor > 0 {
				agg.rechargeUSDMinor += usdMinor
			}
		}
		if sequence == 1 && usdMinor > 0 {
			agg.newUserRechargeUSDMinor += usdMinor
		}
	}
	return rows.Err()
}

func rebuildCoinSellerStock(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
	// 币商进货是平台收到的 USDT 充值，按 seller 的国家/区域进入总充值，但不写入用户金币转账金额。
	rows, err := db.QueryContext(ctx, `
		SELECT seller_user_id, paid_amount_micro, coin_amount, created_at_ms
		FROM coin_seller_stock_records
		WHERE app_code = ? AND counts_as_seller_recharge = 1 AND (paid_amount_micro <> 0 OR coin_amount <> 0) AND created_at_ms >= ? AND created_at_ms < ?`,
		app, startMS, endMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var sellerUserID, amountMicro, coinAmount, createdAtMS int64
		if err := rows.Scan(&sellerUserID, &amountMicro, &coinAmount, &createdAtMS); err != nil {
			return err
		}
		usdMinor := amountMicro / 10_000
		day := dayFromMS(createdAtMS, loc)
		dim := state.userDims[sellerUserID]
		agg := state.ensureAgg(day, dim)
		agg.rechargeUSDMinor += usdMinor
		agg.coinSellerRechargeUSDMinor += usdMinor
		agg.coinSellerStockCoin += coinAmount
	}
	return rows.Err()
}

func rebuildWalletReportMetrics(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, startMS int64, endMS int64, loc *time.Location) error {
	if err := rebuildCoinTotals(ctx, db, state, app, startDay, endDay, endMS, loc); err != nil {
		return err
	}
	rows, err := db.QueryContext(ctx, `
		SELECT e.user_id, e.counterparty_user_id, e.asset_type, e.available_delta + e.frozen_delta AS amount_delta, wt.biz_type, e.created_at_ms
		FROM wallet_entries e
		JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id
		WHERE wt.app_code = ? AND e.asset_type IN ('COIN', 'COIN_SELLER_COIN') AND e.created_at_ms >= ? AND e.created_at_ms < ?`,
		app, startMS, endMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var userID, counterpartyUserID, amountDelta, createdAtMS int64
		var assetType, bizType string
		if err := rows.Scan(&userID, &counterpartyUserID, &assetType, &amountDelta, &bizType, &createdAtMS); err != nil {
			return err
		}
		day := dayFromMS(createdAtMS, loc)
		normalizedBiz := strings.ToLower(strings.TrimSpace(bizType))
		normalizedAsset := strings.ToUpper(strings.TrimSpace(assetType))
		if amountDelta > 0 && normalizedBiz == "salary_transfer_to_coin_seller" && normalizedAsset == "COIN_SELLER_COIN" {
			dimensionUserID := userID
			if counterpartyUserID > 0 {
				dimensionUserID = counterpartyUserID
			}
			agg := state.ensureAgg(day, state.userDims[dimensionUserID])
			agg.salaryTransferCoin += amountDelta
			continue
		}
		if normalizedAsset != "COIN" {
			continue
		}
		dim := state.userDims[userID]
		agg := state.ensureAgg(day, dim)
		switch {
		case amountDelta < 0 && isConsumedCoinBizType(normalizedBiz):
			agg.consumedCoin += -amountDelta
		case amountDelta > 0 && normalizedBiz == "manual_credit":
			agg.manualGrantCoin += amountDelta
		case amountDelta > 0 && isPlatformGrantBizType(normalizedBiz):
			agg.platformGrantCoin += amountDelta
		}
	}
	return rows.Err()
}

type coinBalanceEntry struct {
	userID      int64
	balance     int64
	createdAtMS int64
}

func rebuildCoinTotals(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string, windowEndMS int64, loc *time.Location) error {
	dayEnds, err := dayEndPoints(startDay, endDay, loc)
	if err != nil {
		return err
	}
	rows, err := db.QueryContext(ctx, `
		SELECT e.user_id, e.available_after + e.frozen_after AS balance_after, e.created_at_ms
		FROM wallet_entries e
		JOIN wallet_transactions wt ON wt.transaction_id = e.transaction_id
		WHERE wt.app_code = ? AND e.asset_type = 'COIN' AND e.created_at_ms < ?
		ORDER BY e.user_id ASC, e.created_at_ms ASC, e.entry_id ASC`,
		app, windowEndMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	var currentUserID int64
	entries := []coinBalanceEntry{}
	flush := func() {
		if currentUserID <= 0 || len(entries) == 0 {
			return
		}
		dim := state.userDims[currentUserID]
		index := 0
		balance := int64(0)
		hasBalance := false
		for _, point := range dayEnds {
			for index < len(entries) && entries[index].createdAtMS < point.endMS {
				balance = entries[index].balance
				hasBalance = true
				index++
			}
			if hasBalance {
				state.ensureAgg(point.day, dim).coinTotal += balance
			}
		}
	}
	for rows.Next() {
		var entry coinBalanceEntry
		if err := rows.Scan(&entry.userID, &entry.balance, &entry.createdAtMS); err != nil {
			return err
		}
		if currentUserID != 0 && entry.userID != currentUserID {
			flush()
			entries = entries[:0]
		}
		currentUserID = entry.userID
		entries = append(entries, entry)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	flush()
	return nil
}

func rebuildSalary(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS int64, endMS int64, loc *time.Location) error {
	rows, err := db.QueryContext(ctx, `
		SELECT user_id, agency_owner_user_id, host_salary_usd_minor_delta, agency_salary_usd_minor_delta, created_at_ms
		FROM host_salary_settlement_records
		WHERE app_code = ? AND status = 'succeeded' AND created_at_ms >= ? AND created_at_ms < ?`,
		app, startMS, endMS)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var hostUserID, agencyUserID, hostSalary, agencySalary, createdAtMS int64
		if err := rows.Scan(&hostUserID, &agencyUserID, &hostSalary, &agencySalary, &createdAtMS); err != nil {
			return err
		}
		day := dayFromMS(createdAtMS, loc)
		if hostSalary > 0 {
			state.ensureAgg(day, state.userDims[hostUserID]).salaryUSDMinor += hostSalary
		}
		if agencySalary > 0 && agencyUserID > 0 {
			state.ensureAgg(day, state.userDims[agencyUserID]).salaryUSDMinor += agencySalary
		}
	}
	return rows.Err()
}

func rebuildMicStats(ctx context.Context, db *sql.DB, state *rebuildState, app string, startDay string, endDay string) error {
	rows, err := db.QueryContext(ctx, `
		SELECT DATE_FORMAT(stat_date, '%Y-%m-%d'), user_id, mic_online_ms
		FROM user_mic_daily_stats
		WHERE app_code = ? AND stat_date BETWEEN ? AND ?`,
		app, startDay, endDay)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var day string
		var userID, micOnlineMS int64
		if err := rows.Scan(&day, &userID, &micOnlineMS); err != nil {
			return err
		}
		if micOnlineMS <= 0 {
			continue
		}
		dim := state.userDims[userID]
		state.addMic(day, userID, dim, micOnlineMS)
	}
	return rows.Err()
}

type dayEndPoint struct {
	day   string
	endMS int64
}

func dayEndPoints(startDay string, endDay string, loc *time.Location) ([]dayEndPoint, error) {
	start, err := parseDay(startDay, loc)
	if err != nil {
		return nil, err
	}
	end, err := parseDay(endDay, loc)
	if err != nil {
		return nil, err
	}
	points := []dayEndPoint{}
	for cursor := start; !cursor.After(end); cursor = cursor.AddDate(0, 0, 1) {
		points = append(points, dayEndPoint{
			day:   formatDay(cursor),
			endMS: cursor.AddDate(0, 0, 1).UTC().UnixMilli(),
		})
	}
	return points, nil
}

func isConsumedCoinBizType(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "gift_debit", "direct_gift_debit", "vip_purchase", "resource_shop_purchase",
		"red_packet_create", "game_debit", "wheel_draw_debit", "cp_breakup_fee":
		return true
	default:
		return false
	}
}

func isPlatformGrantBizType(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "task_reward", "lucky_gift_reward", "wheel_reward", "room_turnover_reward",
		"invite_activity_reward", "agency_opening_reward":
		return true
	default:
		return false
	}
}

func (s *rebuildState) ensureAgg(day string, dim countryRegion) *appDayAgg {
	key := appDayKey{day: day, countryID: dim.countryID, regionID: dim.regionID}
	if s.appDays[key] == nil {
		s.appDays[key] = &appDayAgg{}
	}
	return s.appDays[key]
}

func (s *rebuildState) addPayer(day string, userID int64, dim countryRegion, firstMS int64) {
	key := fmt.Sprintf("%s:%d", day, userID)
	if existing, exists := s.payersByKey[key]; exists && existing.firstMS <= firstMS {
		return
	}
	if _, exists := s.payersByKey[key]; !exists {
		s.ensureAgg(day, dim).paidUsers++
	}
	s.payersByKey[key] = payerRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
}

func (s *rebuildState) addActive(day string, userID int64, dim countryRegion, firstMS int64) {
	key := userDayKey(day, userID)
	if existing, exists := s.activeByKey[key]; exists {
		// 注册、进房和游戏都能证明同一用户同一天活跃；主键按用户日收敛，聚合只加一次，
		// 但 first_active_at_ms 要保留最早事实，便于后续排查“为什么算活跃”时能回到第一条证据。
		if firstMS > 0 && (existing.firstMS == 0 || firstMS < existing.firstMS) {
			existing.firstMS = firstMS
			s.activeByKey[key] = existing
		}
		return
	}
	s.activeByKey[key] = activeRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
	s.ensureAgg(day, dim).activeUsers++
}

func (s *rebuildState) addMic(day string, userID int64, dim countryRegion, micOnlineMS int64) {
	key := userDayKey(day, userID)
	if existing, exists := s.micByKey[key]; exists {
		agg := s.ensureAgg(day, existing.dim)
		agg.micOnlineMS += micOnlineMS - existing.micOnlineMS
		existing.micOnlineMS = micOnlineMS
		s.micByKey[key] = existing
		return
	}
	s.micByKey[key] = micRow{day: day, userID: userID, dim: dim, micOnlineMS: micOnlineMS}
	agg := s.ensureAgg(day, dim)
	agg.micOnlineMS += micOnlineMS
	agg.micOnlineUsers++
}

func (s *rebuildState) userRegisteredOn(userID int64, day string) bool {
	_, ok := s.registeredByUserDay[userDayKey(day, userID)]
	return ok
}

func userDayKey(day string, userID int64) string {
	return fmt.Sprintf("%s:%d", day, userID)
}

func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, app, statTZ, startDay, endDay string, apply bool) error {
	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if err := createTempTables(ctx, tx); err != nil {
		return err
	}
	if err := insertStageRows(ctx, tx, state, app, statTZ); err != nil {
		return err
	}
	if err := validateStage(ctx, tx); err != nil {
		return err
	}
	if !apply {
		return tx.Rollback()
	}
	nowMS := time.Now().UTC().UnixMilli()
	// 本 job 只覆盖能从当前源表确定的列；礼物和游戏等不在本次源表范围内的指标保留原值，且只替换目标 stat_tz，避免重跑中国日时覆盖 UTC 读模型。
	if _, err := tx.ExecContext(ctx, `
		UPDATE stat_app_day_country
		SET new_users = 0, active_users = 0, paid_users = 0,
		    recharge_usd_minor = 0, new_user_recharge_usd_minor = 0,
		    coin_seller_recharge_usd_minor = 0,
		    coin_seller_stock_coin = 0,
		    mifapay_recharge_usd_minor = 0, google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
		    coin_total = 0, consumed_coin = 0, platform_grant_coin = 0, manual_grant_coin = 0,
		    salary_usd_minor = 0, salary_transfer_coin = 0, mic_online_ms = 0, mic_online_users = 0,
		    updated_at_ms = ?
		WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
		nowMS, app, statTZ, startDay, endDay); err != nil {
		return err
	}
	for _, statement := range []string{
		`DELETE FROM stat_user_registration WHERE app_code = ? AND stat_tz = ? AND registered_day BETWEEN ? AND ?`,
		`DELETE FROM stat_user_day_activity WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
		`DELETE FROM stat_recharge_day_payers WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
		`DELETE FROM stat_user_day_mic WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?`,
	} {
		if _, err := tx.ExecContext(ctx, statement, app, statTZ, startDay, endDay); err != nil {
			return err
		}
	}
	for userID, dim := range state.userDims {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO stat_user_dimension (app_code, user_id, country_id, region_id, updated_at_ms)
			VALUES (?, ?, ?, ?, ?)
			ON DUPLICATE KEY UPDATE
			  country_id = VALUES(country_id),
			  region_id = VALUES(region_id),
			  updated_at_ms = VALUES(updated_at_ms)
		`, app, userID, dim.countryID, dim.regionID, nowMS); err != nil {
			return err
		}
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_app_day_country (
			app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
			recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
			coin_seller_stock_coin, mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin,
			coin_total, consumed_coin, platform_grant_coin, manual_grant_coin,
			salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users, updated_at_ms
		)
		SELECT app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
		       recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
		       coin_seller_stock_coin, mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin,
		       coin_total, consumed_coin, platform_grant_coin, manual_grant_coin,
		       salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users, ?
		FROM tmp_stat_backfill_app_day
		ON DUPLICATE KEY UPDATE
		  new_users = VALUES(new_users), active_users = VALUES(active_users), paid_users = VALUES(paid_users),
		  recharge_usd_minor = VALUES(recharge_usd_minor),
		  new_user_recharge_usd_minor = VALUES(new_user_recharge_usd_minor),
		  coin_seller_recharge_usd_minor = VALUES(coin_seller_recharge_usd_minor),
		  coin_seller_stock_coin = VALUES(coin_seller_stock_coin),
		  mifapay_recharge_usd_minor = VALUES(mifapay_recharge_usd_minor),
		  google_recharge_usd_minor = VALUES(google_recharge_usd_minor),
		  coin_seller_transfer_coin = VALUES(coin_seller_transfer_coin),
		  coin_total = VALUES(coin_total),
		  consumed_coin = VALUES(consumed_coin),
		  platform_grant_coin = VALUES(platform_grant_coin),
		  manual_grant_coin = VALUES(manual_grant_coin),
		  salary_usd_minor = VALUES(salary_usd_minor),
		  salary_transfer_coin = VALUES(salary_transfer_coin),
		  mic_online_ms = VALUES(mic_online_ms),
		  mic_online_users = VALUES(mic_online_users),
		  updated_at_ms = VALUES(updated_at_ms)`, nowMS); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_user_registration (app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms)
		SELECT app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms FROM tmp_stat_backfill_registration`); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_user_day_activity (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms)
		SELECT app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms FROM tmp_stat_backfill_active`); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_recharge_day_payers (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms)
		SELECT app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms FROM tmp_stat_backfill_payers`); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_user_day_mic (app_code, stat_tz, stat_day, country_id, region_id, user_id, mic_online_ms, updated_at_ms)
		SELECT app_code, stat_tz, stat_day, country_id, region_id, user_id, mic_online_ms, ? FROM tmp_stat_backfill_mic`, nowMS); err != nil {
		return err
	}
	return tx.Commit()
}

func createTempTables(ctx context.Context, tx *sql.Tx) error {
	statements := []string{
		`CREATE TEMPORARY TABLE tmp_stat_backfill_app_day (
			app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL, region_id BIGINT NOT NULL,
			new_users BIGINT NOT NULL, active_users BIGINT NOT NULL, paid_users BIGINT NOT NULL,
			recharge_usd_minor BIGINT NOT NULL, new_user_recharge_usd_minor BIGINT NOT NULL,
			coin_seller_recharge_usd_minor BIGINT NOT NULL, coin_seller_stock_coin BIGINT NOT NULL,
			mifapay_recharge_usd_minor BIGINT NOT NULL, google_recharge_usd_minor BIGINT NOT NULL,
			coin_seller_transfer_coin BIGINT NOT NULL,
			coin_total BIGINT NOT NULL, consumed_coin BIGINT NOT NULL, platform_grant_coin BIGINT NOT NULL,
			manual_grant_coin BIGINT NOT NULL, salary_usd_minor BIGINT NOT NULL, salary_transfer_coin BIGINT NOT NULL,
			mic_online_ms BIGINT NOT NULL, mic_online_users BIGINT NOT NULL,
			PRIMARY KEY(app_code, stat_tz, stat_day, country_id, region_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_registration (
			app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
			country_id BIGINT NOT NULL, region_id BIGINT NOT NULL, registered_at_ms BIGINT NOT NULL,
			PRIMARY KEY(app_code, stat_tz, user_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_active (
			app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
			region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL,
			PRIMARY KEY(app_code, stat_tz, stat_day, user_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_payers (
			app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
			region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
			PRIMARY KEY(app_code, stat_tz, stat_day, user_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_mic (
			app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
			region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, mic_online_ms BIGINT NOT NULL,
			PRIMARY KEY(app_code, stat_tz, stat_day, user_id)
		)`,
	}
	for _, statement := range statements {
		if _, err := tx.ExecContext(ctx, statement); err != nil {
			return err
		}
	}
	return nil
}

func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app string, statTZ string) error {
	for key, agg := range state.appDays {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_app_day (
				app_code, stat_tz, stat_day, country_id, region_id, new_users, active_users, paid_users,
				recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
				coin_seller_stock_coin, mifapay_recharge_usd_minor, google_recharge_usd_minor, coin_seller_transfer_coin,
				coin_total, consumed_coin, platform_grant_coin, manual_grant_coin, salary_usd_minor, salary_transfer_coin, mic_online_ms, mic_online_users
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			app, statTZ, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
			agg.rechargeUSDMinor, agg.newUserRechargeUSDMinor, agg.coinSellerRechargeUSDMinor,
			agg.coinSellerStockCoin, agg.mifaPayRechargeUSDMinor, agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin,
			agg.coinTotal, agg.consumedCoin, agg.platformGrantCoin, agg.manualGrantCoin, agg.salaryUSDMinor, agg.salaryTransferCoin, agg.micOnlineMS, agg.micOnlineUsers); err != nil {
			return err
		}
	}
	for _, row := range state.registrations {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_registration (app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms)
			VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.registeredAtMS); err != nil {
			return err
		}
	}
	for _, row := range state.activeByKey {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_active (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms)
			VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
			return err
		}
	}
	for _, row := range state.payersByKey {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_payers (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms)
			VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
			return err
		}
	}
	for _, row := range state.micByKey {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_mic (app_code, stat_tz, stat_day, country_id, region_id, user_id, mic_online_ms)
			VALUES (?, ?, ?, ?, ?, ?, ?)`, app, statTZ, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.micOnlineMS); err != nil {
			return err
		}
	}
	return nil
}

func validateStage(ctx context.Context, tx *sql.Tx) error {
	var negativeCount int64
	if err := tx.QueryRowContext(ctx, `
		SELECT COUNT(*)
		FROM tmp_stat_backfill_app_day
		WHERE new_users < 0 OR active_users < 0 OR paid_users < 0 OR recharge_usd_minor < 0 OR coin_seller_transfer_coin < 0
		   OR coin_total < 0 OR consumed_coin < 0 OR platform_grant_coin < 0 OR manual_grant_coin < 0
		   OR salary_usd_minor < 0 OR salary_transfer_coin < 0 OR mic_online_ms < 0 OR mic_online_users < 0`).Scan(&negativeCount); err != nil {
		return err
	}
	if negativeCount > 0 {
		return fmt.Errorf("stage contains negative counters")
	}
	return nil
}