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
	googleRechargeUSDMinor     int64
	coinSellerTransferCoin     int64
}

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

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

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

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

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)
	}
	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{},
		userIDs:             map[int64]struct{}{},
		userDims:            map[int64]countryRegion{},
	}
	// 第一阶段只收集窗口内会参与重算的 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, 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 := 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, 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 := stageAndMaybeApply(ctx, statsDB, state, app, 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, loc.String(), startDay, endDay, len(state.appDays), len(state.registrations), len(state.activeByKey), len(state.payersByKey))
}

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 {
	rows, err := db.QueryContext(ctx, `
		SELECT u.user_id
		FROM users u
		WHERE u.app_code = ? AND u.created_at_ms >= ? AND u.created_at_ms < ?`,
		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, startDay, endDay string) error {
	rows, err := db.QueryContext(ctx, `
		SELECT DISTINCT user_id
		FROM stat_user_day_activity
		WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
		app, 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 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
		}
		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 映射出的真实 country_id 为准；region_id 只作为独立维度写入，不能再兜底成国家。
	rows, err := db.QueryContext(ctx, `
		SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0), u.created_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.created_at_ms >= ? AND u.created_at_ms < ?`,
		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.createdAtMS); err != nil {
			return err
		}
		row.registeredDay = dayFromMS(row.createdAtMS, loc)
		state.registrations = append(state.registrations, row)
		state.registeredByUserDay[userDayKey(row.registeredDay, row.userID)] = struct{}{}
		agg := state.ensureAgg(row.registeredDay, row.dim)
		agg.newUsers++
	}
	return rows.Err()
}

func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
	// 活跃的原始房间事件已经被实时消费进 stat_user_day_activity；离线回填只重刷该表的国家/区域快照，避免在大窗口内扫描房间流水。
	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_day BETWEEN ? AND ?
		GROUP BY stat_day, user_id`,
		app, 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
		}
		dim := state.userDims[userID]
		key := fmt.Sprintf("%s:%d", day, userID)
		if _, exists := state.activeByKey[key]; exists {
			continue
		}
		state.activeByKey[key] = activeRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
		state.ensureAgg(day, dim).activeUsers++
	}
	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
		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, created_at_ms
		FROM coin_seller_stock_records
		WHERE app_code = ? AND counts_as_seller_recharge = 1 AND paid_amount_micro > 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, createdAtMS int64
		if err := rows.Scan(&sellerUserID, &amountMicro, &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
	}
	return rows.Err()
}

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) 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, 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); err != nil {
		return err
	}
	if err := validateStage(ctx, tx); err != nil {
		return err
	}
	if !apply {
		return tx.Rollback()
	}
	nowMS := time.Now().UTC().UnixMilli()
	// 本 job 只覆盖能从当前源表确定的近 7 天列；礼物、游戏、MifaPay 等不在本次源表范围内的指标保留原值，避免修国家维度时误删其他业务数据。
	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,
		    google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
		    updated_at_ms = ?
		WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
		nowMS, app, startDay, endDay); err != nil {
		return err
	}
	for _, statement := range []string{
		`DELETE FROM stat_user_registration WHERE app_code = ? AND registered_day BETWEEN ? AND ?`,
		`DELETE FROM stat_user_day_activity WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
		`DELETE FROM stat_recharge_day_payers WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
	} {
		if _, err := tx.ExecContext(ctx, statement, app, startDay, endDay); err != nil {
			return err
		}
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_app_day_country (
			app_code, 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,
			google_recharge_usd_minor, coin_seller_transfer_coin, updated_at_ms
		)
		SELECT app_code, 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,
		       google_recharge_usd_minor, coin_seller_transfer_coin, ?
		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),
		  google_recharge_usd_minor = VALUES(google_recharge_usd_minor),
		  coin_seller_transfer_coin = VALUES(coin_seller_transfer_coin),
		  updated_at_ms = VALUES(updated_at_ms)`, nowMS); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO stat_user_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
		SELECT app_code, 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_day, country_id, region_id, user_id, first_active_at_ms)
		SELECT app_code, 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_day, country_id, region_id, user_id, first_paid_at_ms)
		SELECT app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms FROM tmp_stat_backfill_payers`); 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_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, google_recharge_usd_minor BIGINT NOT NULL,
			coin_seller_transfer_coin BIGINT NOT NULL,
			PRIMARY KEY(app_code, stat_day, country_id, region_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_registration (
			app_code VARCHAR(32) 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, user_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_active (
			app_code VARCHAR(32) 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_day, user_id)
		)`,
		`CREATE TEMPORARY TABLE tmp_stat_backfill_payers (
			app_code VARCHAR(32) 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_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) error {
	for key, agg := range state.appDays {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_app_day (
				app_code, 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,
				google_recharge_usd_minor, coin_seller_transfer_coin
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			app, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
			agg.rechargeUSDMinor, agg.newUserRechargeUSDMinor, agg.coinSellerRechargeUSDMinor,
			agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin); err != nil {
			return err
		}
	}
	for _, row := range state.registrations {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
			VALUES (?, ?, ?, ?, ?, ?)`, app, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.createdAtMS); err != nil {
			return err
		}
	}
	for _, row := range state.activeByKey {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO tmp_stat_backfill_active (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
			VALUES (?, ?, ?, ?, ?, ?)`, app, 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_day, country_id, region_id, user_id, first_paid_at_ms)
			VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); 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`).Scan(&negativeCount); err != nil {
		return err
	}
	if negativeCount > 0 {
		return fmt.Errorf("stage contains negative counters")
	}
	return nil
}