package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	mysqldriver "github.com/go-sql-driver/mysql"
	"hyapp/pkg/xerr"
	userdomain "hyapp/services/user-service/internal/domain/user"
)

// ResolveEnabledCountryByCode 按国家码读取 App 可选国家，注册和改国家都会经过这里。
func (r *Repository) ResolveEnabledCountryByCode(ctx context.Context, countryCode string) (userdomain.Country, bool, error) {
	return resolveCountryByCode(ctx, r.db, countryCode, true)
}

// ListRegistrationCountries 返回注册页可选国家，只包含 enabled 的国家。
func (r *Repository) ListRegistrationCountries(ctx context.Context) ([]userdomain.Country, error) {
	return listCountries(ctx, r.db, userdomain.CountryFilter{Enabled: boolPtr(true)})
}

// ResolveActiveRegionByCountry 按国家码读取当前 active 区域映射；无映射返回 ok=false。
func (r *Repository) ResolveActiveRegionByCountry(ctx context.Context, countryCode string) (userdomain.Region, bool, error) {
	country, ok, err := resolveCountryByCode(ctx, r.db, countryCode, true)
	if err != nil || !ok {
		return userdomain.Region{}, false, err
	}
	var region userdomain.Region
	err = r.db.QueryRowContext(ctx, `
		SELECT rg.region_id, rg.region_code, rg.name, rg.status, rg.sort_order,
		       COALESCE(rg.created_by_user_id, 0), COALESCE(rg.updated_by_user_id, 0), rg.created_at_ms, rg.updated_at_ms
		FROM region_countries rc
		INNER JOIN regions rg ON rg.region_id = rc.region_id
		WHERE rc.country_code = ? AND rc.status = ? AND rg.status = ?
		LIMIT 1
	`, country.CountryCode, userdomain.RegionStatusActive, userdomain.RegionStatusActive).Scan(
		&region.RegionID,
		&region.RegionCode,
		&region.Name,
		&region.Status,
		&region.SortOrder,
		&region.CreatedByUserID,
		&region.UpdatedByUserID,
		&region.CreatedAtMs,
		&region.UpdatedAtMs,
	)
	if err == sql.ErrNoRows {
		return userdomain.Region{}, false, nil
	}
	if err != nil {
		return userdomain.Region{}, false, err
	}
	countries, err := activeCountriesForRegion(ctx, r.db, region.RegionID)
	if err != nil {
		return userdomain.Region{}, false, err
	}
	region.Countries = countries

	return region, true, nil
}

// CreateCountry 创建国家主数据；enabled 是 App 注册页和改国家链路的唯一可用性开关。
func (r *Repository) CreateCountry(ctx context.Context, command userdomain.CreateCountryCommand) (userdomain.Country, error) {
	result, err := r.db.ExecContext(ctx, `
		INSERT INTO countries (
			country_name, country_code, iso_alpha3, iso_numeric, country_display_name,
			phone_country_code, flag, enabled, sort_order,
			created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
		)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
	`, command.CountryName, command.CountryCode, nullableString(command.ISOAlpha3), nullableString(command.ISONumeric), command.CountryDisplayName, nullableString(command.PhoneCountryCode), command.Flag, command.Enabled, command.SortOrder, nullableOperator(command.OperatorUserID), nullableOperator(command.OperatorUserID), command.NowMs, command.NowMs)
	if err != nil {
		return userdomain.Country{}, mapCountryDuplicateError(err)
	}
	countryID, err := result.LastInsertId()
	if err != nil {
		return userdomain.Country{}, err
	}

	return queryCountry(ctx, r.db, "WHERE country_id = ?", countryID)
}

// UpdateCountry 修改国家展示字段；country_code 不允许原地修改。
func (r *Repository) UpdateCountry(ctx context.Context, command userdomain.UpdateCountryCommand) (userdomain.Country, error) {
	result, err := r.db.ExecContext(ctx, `
		UPDATE countries
		SET country_name = ?,
			iso_alpha3 = ?,
			iso_numeric = ?,
			country_display_name = ?,
			phone_country_code = ?,
			flag = ?,
			sort_order = ?,
			updated_by_user_id = ?,
			updated_at_ms = ?
		WHERE country_id = ?
	`, command.CountryName, nullableString(command.ISOAlpha3), nullableString(command.ISONumeric), command.CountryDisplayName, nullableString(command.PhoneCountryCode), command.Flag, command.SortOrder, nullableOperator(command.OperatorUserID), command.NowMs, command.CountryID)
	if err != nil {
		return userdomain.Country{}, mapCountryDuplicateError(err)
	}
	if affected, err := result.RowsAffected(); err != nil {
		return userdomain.Country{}, err
	} else if affected == 0 {
		return userdomain.Country{}, xerr.New(xerr.CountryNotFound, "country not found")
	}

	return queryCountry(ctx, r.db, "WHERE country_id = ?", command.CountryID)
}

// ListCountries 返回国家主数据，enabled 为空时返回全部。
func (r *Repository) ListCountries(ctx context.Context, filter userdomain.CountryFilter) ([]userdomain.Country, error) {
	return listCountries(ctx, r.db, filter)
}

// EnableCountry 重新开放国家给 App 注册和改国家链路。
func (r *Repository) EnableCountry(ctx context.Context, command userdomain.EnableCountryCommand) (userdomain.Country, error) {
	result, err := r.db.ExecContext(ctx, `
		UPDATE countries
		SET enabled = ?, updated_by_user_id = ?, updated_at_ms = ?
		WHERE country_id = ?
	`, true, nullableOperator(command.OperatorUserID), command.NowMs, command.CountryID)
	if err != nil {
		return userdomain.Country{}, err
	}
	if affected, err := result.RowsAffected(); err != nil {
		return userdomain.Country{}, err
	} else if affected == 0 {
		return userdomain.Country{}, xerr.New(xerr.CountryNotFound, "country not found")
	}

	return queryCountry(ctx, r.db, "WHERE country_id = ?", command.CountryID)
}

// DisableCountry 只关闭 App 选择入口，不删除国家，也不释放区域配置。
func (r *Repository) DisableCountry(ctx context.Context, command userdomain.DisableCountryCommand) (userdomain.Country, error) {
	result, err := r.db.ExecContext(ctx, `
		UPDATE countries
		SET enabled = ?, updated_by_user_id = ?, updated_at_ms = ?
		WHERE country_id = ?
	`, false, nullableOperator(command.OperatorUserID), command.NowMs, command.CountryID)
	if err != nil {
		return userdomain.Country{}, err
	}
	if affected, err := result.RowsAffected(); err != nil {
		return userdomain.Country{}, err
	} else if affected == 0 {
		return userdomain.Country{}, xerr.New(xerr.CountryNotFound, "country not found")
	}

	return queryCountry(ctx, r.db, "WHERE country_id = ?", command.CountryID)
}

// CreateRegion 创建区域并写 active 国家归属、审计和历史用户重算任务。
func (r *Repository) CreateRegion(ctx context.Context, command userdomain.CreateRegionCommand) (userdomain.Region, error) {
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return userdomain.Region{}, err
	}
	defer tx.Rollback()

	countries, err := canonicalCountryCodes(ctx, tx, command.Countries)
	if err != nil {
		return userdomain.Region{}, err
	}
	result, err := tx.ExecContext(ctx, `
		INSERT INTO regions (region_code, name, status, sort_order, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	`, command.RegionCode, command.Name, userdomain.RegionStatusActive, command.SortOrder, nullableOperator(command.OperatorUserID), nullableOperator(command.OperatorUserID), command.NowMs, command.NowMs)
	if err != nil {
		return userdomain.Region{}, mapRegionDuplicateError(err)
	}
	regionID, err := result.LastInsertId()
	if err != nil {
		return userdomain.Region{}, err
	}
	for _, country := range countries {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO region_countries (region_id, country_code, status, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms)
			VALUES (?, ?, ?, ?, ?, ?, ?)
		`, regionID, country, userdomain.RegionStatusActive, nullableOperator(command.OperatorUserID), nullableOperator(command.OperatorUserID), command.NowMs, command.NowMs); err != nil {
			return userdomain.Region{}, mapRegionCountryDuplicateError(err)
		}
		if err := insertRebuildTask(ctx, tx, regionID, country, regionID, command.NowMs); err != nil {
			return userdomain.Region{}, err
		}
	}
	region := userdomain.Region{
		RegionID:        regionID,
		RegionCode:      command.RegionCode,
		Name:            command.Name,
		Status:          userdomain.RegionStatusActive,
		Countries:       append([]string(nil), countries...),
		SortOrder:       command.SortOrder,
		CreatedByUserID: command.OperatorUserID,
		UpdatedByUserID: command.OperatorUserID,
		CreatedAtMs:     command.NowMs,
		UpdatedAtMs:     command.NowMs,
	}
	if err := insertRegionChangeLog(ctx, tx, regionID, "create_region", nil, region, command.OperatorUserID, command.RequestID, command.NowMs); err != nil {
		return userdomain.Region{}, err
	}
	if err := tx.Commit(); err != nil {
		return userdomain.Region{}, err
	}

	return region, nil
}

// UpdateRegion 修改区域展示字段，不改变国家归属。
func (r *Repository) UpdateRegion(ctx context.Context, command userdomain.UpdateRegionCommand) (userdomain.Region, error) {
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return userdomain.Region{}, err
	}
	defer tx.Rollback()

	before, err := queryRegion(ctx, tx, "WHERE region_id = ? FOR UPDATE", command.RegionID)
	if err == sql.ErrNoRows {
		return userdomain.Region{}, xerr.New(xerr.RegionNotFound, "region not found")
	}
	if err != nil {
		return userdomain.Region{}, err
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE regions
		SET region_code = ?, name = ?, sort_order = ?, updated_by_user_id = ?, updated_at_ms = ?
		WHERE region_id = ?
	`, command.RegionCode, command.Name, command.SortOrder, nullableOperator(command.OperatorUserID), command.NowMs, command.RegionID); err != nil {
		return userdomain.Region{}, mapRegionDuplicateError(err)
	}
	after := before
	after.RegionCode = command.RegionCode
	after.Name = command.Name
	after.SortOrder = command.SortOrder
	after.UpdatedByUserID = command.OperatorUserID
	after.UpdatedAtMs = command.NowMs
	if err := insertRegionChangeLog(ctx, tx, command.RegionID, "update_region", before, after, command.OperatorUserID, command.RequestID, command.NowMs); err != nil {
		return userdomain.Region{}, err
	}
	if err := tx.Commit(); err != nil {
		return userdomain.Region{}, err
	}

	return r.GetRegion(ctx, command.RegionID)
}

// ListRegions 返回区域主数据列表，countries 只包含 active 映射。
func (r *Repository) ListRegions(ctx context.Context, filter userdomain.RegionFilter) ([]userdomain.Region, error) {
	query := `
		SELECT region_id, region_code, name, status, sort_order,
		       COALESCE(created_by_user_id, 0), COALESCE(updated_by_user_id, 0), created_at_ms, updated_at_ms
		FROM regions`
	args := []any{}
	if filter.Status != "" {
		query += " WHERE status = ?"
		args = append(args, filter.Status)
	}
	query += " ORDER BY sort_order ASC, region_code ASC"
	rows, err := r.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	regions := []userdomain.Region{}
	for rows.Next() {
		region, err := scanRegion(rows)
		if err != nil {
			return nil, err
		}
		countries, err := activeCountriesForRegion(ctx, r.db, region.RegionID)
		if err != nil {
			return nil, err
		}
		region.Countries = countries
		regions = append(regions, region)
	}

	return regions, rows.Err()
}

// GetRegion 返回单个区域及其 active 国家列表。
func (r *Repository) GetRegion(ctx context.Context, regionID int64) (userdomain.Region, error) {
	region, err := queryRegion(ctx, r.db, "WHERE region_id = ?", regionID)
	if err == sql.ErrNoRows {
		return userdomain.Region{}, xerr.New(xerr.RegionNotFound, "region not found")
	}
	if err != nil {
		return userdomain.Region{}, err
	}

	return region, nil
}

// ReplaceRegionCountries 原子替换区域 active 国家归属，并为受影响国家创建 rebuild task。
func (r *Repository) ReplaceRegionCountries(ctx context.Context, command userdomain.ReplaceRegionCountriesCommand) (userdomain.Region, error) {
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return userdomain.Region{}, err
	}
	defer tx.Rollback()

	before, err := queryRegion(ctx, tx, "WHERE region_id = ? FOR UPDATE", command.RegionID)
	if err == sql.ErrNoRows {
		return userdomain.Region{}, xerr.New(xerr.RegionNotFound, "region not found")
	}
	if err != nil {
		return userdomain.Region{}, err
	}
	if before.Status != userdomain.RegionStatusActive {
		return userdomain.Region{}, xerr.New(xerr.RegionDisabled, "region is disabled")
	}
	countries, err := canonicalCountryCodes(ctx, tx, command.Countries)
	if err != nil {
		return userdomain.Region{}, err
	}
	if err := ensureCountriesAssignableToRegion(ctx, tx, countries, command.RegionID); err != nil {
		return userdomain.Region{}, err
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE region_countries
		SET status = ?, updated_by_user_id = ?, updated_at_ms = ?
		WHERE region_id = ? AND status = ?
	`, userdomain.RegionStatusDisabled, nullableOperator(command.OperatorUserID), command.NowMs, command.RegionID, userdomain.RegionStatusActive); err != nil {
		return userdomain.Region{}, err
	}
	for _, country := range countries {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO region_countries (region_id, country_code, status, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms)
			VALUES (?, ?, ?, ?, ?, ?, ?)
		`, command.RegionID, country, userdomain.RegionStatusActive, nullableOperator(command.OperatorUserID), nullableOperator(command.OperatorUserID), command.NowMs, command.NowMs); err != nil {
			return userdomain.Region{}, mapRegionCountryDuplicateError(err)
		}
	}
	after := before
	after.Countries = append([]string(nil), countries...)
	after.UpdatedByUserID = command.OperatorUserID
	after.UpdatedAtMs = command.NowMs
	for _, country := range unionCountryCodes(before.Countries, countries) {
		targetRegionID := int64(0)
		if stringSliceContains(countries, country) {
			targetRegionID = command.RegionID
		}
		if err := insertRebuildTask(ctx, tx, command.RegionID, country, targetRegionID, command.NowMs); err != nil {
			return userdomain.Region{}, err
		}
	}
	if err := insertRegionChangeLog(ctx, tx, command.RegionID, "replace_region_countries", before, after, command.OperatorUserID, command.RequestID, command.NowMs); err != nil {
		return userdomain.Region{}, err
	}
	if err := tx.Commit(); err != nil {
		return userdomain.Region{}, err
	}

	return r.GetRegion(ctx, command.RegionID)
}

// DisableRegion 停用区域并释放 active 国家归属。
func (r *Repository) DisableRegion(ctx context.Context, command userdomain.DisableRegionCommand) (userdomain.Region, error) {
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return userdomain.Region{}, err
	}
	defer tx.Rollback()

	before, err := queryRegion(ctx, tx, "WHERE region_id = ? FOR UPDATE", command.RegionID)
	if err == sql.ErrNoRows {
		return userdomain.Region{}, xerr.New(xerr.RegionNotFound, "region not found")
	}
	if err != nil {
		return userdomain.Region{}, err
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE regions
		SET status = ?, updated_by_user_id = ?, updated_at_ms = ?
		WHERE region_id = ?
	`, userdomain.RegionStatusDisabled, nullableOperator(command.OperatorUserID), command.NowMs, command.RegionID); err != nil {
		return userdomain.Region{}, err
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE region_countries
		SET status = ?, updated_by_user_id = ?, updated_at_ms = ?
		WHERE region_id = ? AND status = ?
	`, userdomain.RegionStatusDisabled, nullableOperator(command.OperatorUserID), command.NowMs, command.RegionID, userdomain.RegionStatusActive); err != nil {
		return userdomain.Region{}, err
	}
	after := before
	after.Status = userdomain.RegionStatusDisabled
	after.Countries = nil
	after.UpdatedByUserID = command.OperatorUserID
	after.UpdatedAtMs = command.NowMs
	for _, country := range before.Countries {
		if err := insertRebuildTask(ctx, tx, command.RegionID, country, 0, command.NowMs); err != nil {
			return userdomain.Region{}, err
		}
	}
	if err := insertRegionChangeLog(ctx, tx, command.RegionID, "disable_region", before, after, command.OperatorUserID, command.RequestID, command.NowMs); err != nil {
		return userdomain.Region{}, err
	}
	if err := tx.Commit(); err != nil {
		return userdomain.Region{}, err
	}

	return r.GetRegion(ctx, command.RegionID)
}

// ProcessNextRegionRebuildTask 先提交 task claim，再按 user_id cursor 分页刷新 users.region_id。
func (r *Repository) ProcessNextRegionRebuildTask(ctx context.Context, workerID string, nowMs int64, lockTTL time.Duration, batchSize int) (userdomain.RegionRebuildProcessResult, error) {
	if lockTTL <= 0 {
		lockTTL = 30 * time.Second
	}
	if batchSize <= 0 {
		batchSize = 500
	}

	// claim 必须独立提交：running/locked_until/attempt_count 要对其他 worker 和运维可见。
	task, ok, err := claimRegionRebuildTask(ctx, r.db, workerID, nowMs, lockTTL)
	if err != nil {
		return userdomain.RegionRebuildProcessResult{}, err
	}
	if !ok {
		return userdomain.RegionRebuildProcessResult{NoTask: true}, nil
	}

	return r.processClaimedRegionRebuildTask(ctx, task.TaskID, workerID, nowMs, batchSize)
}

func (r *Repository) processClaimedRegionRebuildTask(ctx context.Context, taskID int64, workerID string, nowMs int64, batchSize int) (userdomain.RegionRebuildProcessResult, error) {
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return userdomain.RegionRebuildProcessResult{}, err
	}
	defer tx.Rollback()

	task, ok, err := lockClaimedRegionRebuildTask(ctx, tx, taskID, workerID)
	if err != nil {
		return userdomain.RegionRebuildProcessResult{}, err
	}
	if !ok {
		// task 在 claim 提交后被接管或人工改动时，本轮不应继续写 users。
		return userdomain.RegionRebuildProcessResult{TaskID: taskID}, xerr.New(xerr.Conflict, "region rebuild task lock is lost")
	}
	result := userdomain.RegionRebuildProcessResult{
		TaskID:         task.TaskID,
		CountryCode:    task.CountryCode,
		TargetRegionID: task.TargetRegionID,
		Status:         userdomain.RegionRebuildTaskStatusRunning,
		CursorUserID:   task.CursorUserID,
	}

	stale, err := newerRegionRebuildTaskExists(ctx, tx, task.CountryCode, task.MappingRevision, task.TaskID)
	if err != nil {
		return result, err
	}
	if stale {
		// 旧任务不能覆盖新映射；直接标记 skipped，等待较新任务刷新历史用户。
		if err := finishRegionRebuildTask(ctx, tx, task.TaskID, workerID, userdomain.RegionRebuildTaskStatusSkipped, task.CursorUserID, "stale mapping revision", nowMs); err != nil {
			return result, err
		}
		if err := tx.Commit(); err != nil {
			return result, err
		}
		result.Status = userdomain.RegionRebuildTaskStatusSkipped
		return result, nil
	}

	userIDs, err := selectRegionRebuildUserIDs(ctx, tx, task.CountryCode, task.CursorUserID, batchSize)
	if err != nil {
		return result, err
	}
	if len(userIDs) == 0 {
		if err := finishRegionRebuildTask(ctx, tx, task.TaskID, workerID, userdomain.RegionRebuildTaskStatusCompleted, task.CursorUserID, "", nowMs); err != nil {
			return result, err
		}
		if err := tx.Commit(); err != nil {
			return result, err
		}
		result.Status = userdomain.RegionRebuildTaskStatusCompleted
		return result, nil
	}

	if err := updateUsersRegionByIDs(ctx, tx, userIDs, task.TargetRegionID, nowMs); err != nil {
		return result, err
	}
	newCursor := userIDs[len(userIDs)-1]
	status := userdomain.RegionRebuildTaskStatusPending
	if len(userIDs) < batchSize {
		status = userdomain.RegionRebuildTaskStatusCompleted
	}
	if err := finishRegionRebuildTask(ctx, tx, task.TaskID, workerID, status, newCursor, "", nowMs); err != nil {
		return result, err
	}
	if err := tx.Commit(); err != nil {
		return result, err
	}

	result.Status = status
	result.ProcessedUsers = len(userIDs)
	result.CursorUserID = newCursor
	return result, nil
}

func queryCountry(ctx context.Context, q queryer, clause string, args ...any) (userdomain.Country, error) {
	row := q.QueryRowContext(ctx, `
		SELECT country_id, country_name, country_code,
		       COALESCE(iso_alpha3, ''), COALESCE(iso_numeric, ''),
		       country_display_name, COALESCE(phone_country_code, ''), COALESCE(flag, ''), enabled,
		       sort_order,
		       COALESCE(created_by_user_id, 0), COALESCE(updated_by_user_id, 0), created_at_ms, updated_at_ms
		FROM countries
		`+clause, args...)
	return scanCountry(row)
}

func listCountries(ctx context.Context, q queryer, filter userdomain.CountryFilter) ([]userdomain.Country, error) {
	query := `
		SELECT country_id, country_name, country_code,
		       COALESCE(iso_alpha3, ''), COALESCE(iso_numeric, ''),
		       country_display_name, COALESCE(phone_country_code, ''), COALESCE(flag, ''), enabled,
		       sort_order,
		       COALESCE(created_by_user_id, 0), COALESCE(updated_by_user_id, 0), created_at_ms, updated_at_ms
		FROM countries`
	args := []any{}
	conditions := []string{}
	if filter.Enabled != nil {
		conditions = append(conditions, "enabled = ?")
		args = append(args, *filter.Enabled)
	}
	if len(conditions) > 0 {
		query += " WHERE " + strings.Join(conditions, " AND ")
	}
	query += " ORDER BY sort_order ASC, country_code ASC"
	rows, err := queryRows(ctx, q, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	countries := []userdomain.Country{}
	for rows.Next() {
		country, err := scanCountry(rows)
		if err != nil {
			return nil, err
		}
		countries = append(countries, country)
	}

	return countries, rows.Err()
}

func scanCountry(scanner interface{ Scan(dest ...any) error }) (userdomain.Country, error) {
	var country userdomain.Country
	err := scanner.Scan(
		&country.CountryID,
		&country.CountryName,
		&country.CountryCode,
		&country.ISOAlpha3,
		&country.ISONumeric,
		&country.CountryDisplayName,
		&country.PhoneCountryCode,
		&country.Flag,
		&country.Enabled,
		&country.SortOrder,
		&country.CreatedByUserID,
		&country.UpdatedByUserID,
		&country.CreatedAtMs,
		&country.UpdatedAtMs,
	)
	return country, err
}

func boolPtr(value bool) *bool {
	return &value
}

func queryRegion(ctx context.Context, q queryer, clause string, args ...any) (userdomain.Region, error) {
	row := q.QueryRowContext(ctx, `
		SELECT region_id, region_code, name, status, sort_order,
		       COALESCE(created_by_user_id, 0), COALESCE(updated_by_user_id, 0), created_at_ms, updated_at_ms
		FROM regions
		`+clause, args...)
	region, err := scanRegion(row)
	if err != nil {
		return userdomain.Region{}, err
	}
	countries, err := activeCountriesForRegion(ctx, q, region.RegionID)
	if err != nil {
		return userdomain.Region{}, err
	}
	region.Countries = countries

	return region, nil
}

func scanRegion(scanner interface{ Scan(dest ...any) error }) (userdomain.Region, error) {
	var region userdomain.Region
	err := scanner.Scan(
		&region.RegionID,
		&region.RegionCode,
		&region.Name,
		&region.Status,
		&region.SortOrder,
		&region.CreatedByUserID,
		&region.UpdatedByUserID,
		&region.CreatedAtMs,
		&region.UpdatedAtMs,
	)
	return region, err
}

func activeCountriesForRegion(ctx context.Context, q queryer, regionID int64) ([]string, error) {
	rows, err := queryRows(ctx, q, `
		SELECT country_code
		FROM region_countries
		WHERE region_id = ? AND status = ?
		ORDER BY country_code ASC
	`, regionID, userdomain.RegionStatusActive)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	countries := []string{}
	for rows.Next() {
		var country string
		if err := rows.Scan(&country); err != nil {
			return nil, err
		}
		countries = append(countries, country)
	}

	return countries, rows.Err()
}

func canonicalCountryCodes(ctx context.Context, q queryer, countries []string) ([]string, error) {
	seen := make(map[string]struct{}, len(countries))
	result := make([]string, 0, len(countries))
	for _, input := range countries {
		country, ok, err := resolveCountryByCode(ctx, q, input, false)
		if err != nil {
			return nil, err
		}
		if !ok {
			return nil, xerr.New(xerr.InvalidArgument, "country is not supported")
		}
		if _, exists := seen[country.CountryCode]; exists {
			// 区域国家列表不能重复，避免重复写入 active 国家归属。
			return nil, xerr.New(xerr.InvalidArgument, "countries must not contain duplicate canonical codes")
		}
		seen[country.CountryCode] = struct{}{}
		result = append(result, country.CountryCode)
	}

	return result, nil
}

func resolveCountryByCode(ctx context.Context, q queryer, countryCode string, requireEnabled bool) (userdomain.Country, bool, error) {
	countryCode = userdomain.NormalizeCountryCode(countryCode)
	clause := "WHERE country_code = ?"
	args := []any{countryCode}
	if requireEnabled {
		clause += " AND enabled = ?"
		args = append(args, true)
	}
	country, err := queryCountry(ctx, q, clause, args...)
	if err == sql.ErrNoRows {
		return userdomain.Country{}, false, nil
	}
	if err != nil {
		return userdomain.Country{}, false, err
	}

	return country, true, nil
}

func ensureCountriesAssignableToRegion(ctx context.Context, q queryer, countries []string, regionID int64) error {
	for _, country := range countries {
		mappedRegionID, mapped, err := activeRegionIDByCountry(ctx, q, country)
		if err != nil {
			return err
		}
		if mapped && mappedRegionID != regionID {
			return xerr.New(xerr.RegionCountryConflict, "country already belongs to another active region")
		}
	}

	return nil
}

func activeRegionIDByCountry(ctx context.Context, q queryer, countryCode string) (int64, bool, error) {
	var regionID int64
	err := q.QueryRowContext(ctx, `
		SELECT region_id
		FROM region_countries
		WHERE country_code = ? AND status = ?
		LIMIT 1
	`, countryCode, userdomain.RegionStatusActive).Scan(&regionID)
	if err == sql.ErrNoRows {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, err
	}

	return regionID, true, nil
}

func insertRebuildTask(ctx context.Context, tx *sql.Tx, regionID int64, countryCode string, targetRegionID int64, revision int64) error {
	_, err := tx.ExecContext(ctx, `
		INSERT INTO user_region_rebuild_tasks (region_id, country_code, target_region_id, mapping_revision, status, cursor_user_id, created_at_ms, updated_at_ms)
		VALUES (?, ?, ?, ?, ?, 0, ?, ?)
	`, nullableRegionID(regionID), countryCode, nullableRegionID(targetRegionID), revision, userdomain.RegionRebuildTaskStatusPending, revision, revision)
	return err
}

func claimRegionRebuildTask(ctx context.Context, db *sql.DB, workerID string, nowMs int64, lockTTL time.Duration) (userdomain.UserRegionRebuildTask, bool, error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	}
	defer tx.Rollback()

	task, ok, err := selectClaimableRegionRebuildTask(ctx, tx, nowMs)
	if err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	}
	if !ok {
		return userdomain.UserRegionRebuildTask{}, false, nil
	}

	lockedUntilMs := nowMs + lockTTL.Milliseconds()
	result, err := tx.ExecContext(ctx, `
		UPDATE user_region_rebuild_tasks
		SET status = ?, locked_by = ?, locked_until_ms = ?, attempt_count = attempt_count + 1, error_message = NULL, updated_at_ms = ?
		WHERE task_id = ?
	`, userdomain.RegionRebuildTaskStatusRunning, workerID, lockedUntilMs, nowMs, task.TaskID)
	if err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	}
	if affected, err := result.RowsAffected(); err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	} else if affected != 1 {
		return userdomain.UserRegionRebuildTask{}, false, xerr.New(xerr.Conflict, "region rebuild task claim conflict")
	}
	if err := tx.Commit(); err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	}
	task.Status = userdomain.RegionRebuildTaskStatusRunning
	task.LockedBy = workerID
	task.LockedUntilMs = lockedUntilMs
	task.AttemptCount++
	task.UpdatedAtMs = nowMs

	return task, true, nil
}

func selectClaimableRegionRebuildTask(ctx context.Context, tx *sql.Tx, nowMs int64) (userdomain.UserRegionRebuildTask, bool, error) {
	row := tx.QueryRowContext(ctx, `
		SELECT task_id, region_id, country_code, target_region_id, mapping_revision, status, cursor_user_id,
		       locked_by, locked_until_ms, attempt_count, error_message, created_at_ms, updated_at_ms
		FROM user_region_rebuild_tasks
		WHERE status = ?
		   OR (status = ? AND COALESCE(locked_until_ms, 0) <= ?)
		ORDER BY updated_at_ms ASC, task_id ASC
		LIMIT 1
		FOR UPDATE SKIP LOCKED
	`, userdomain.RegionRebuildTaskStatusPending, userdomain.RegionRebuildTaskStatusRunning, nowMs)
	task, err := scanRegionRebuildTask(row)
	if err == sql.ErrNoRows {
		return userdomain.UserRegionRebuildTask{}, false, nil
	}
	if err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	}

	return task, true, nil
}

func lockClaimedRegionRebuildTask(ctx context.Context, tx *sql.Tx, taskID int64, workerID string) (userdomain.UserRegionRebuildTask, bool, error) {
	row := tx.QueryRowContext(ctx, `
		SELECT task_id, region_id, country_code, target_region_id, mapping_revision, status, cursor_user_id,
		       locked_by, locked_until_ms, attempt_count, error_message, created_at_ms, updated_at_ms
		FROM user_region_rebuild_tasks
		WHERE task_id = ? AND status = ? AND locked_by = ?
		FOR UPDATE
	`, taskID, userdomain.RegionRebuildTaskStatusRunning, workerID)
	task, err := scanRegionRebuildTask(row)
	if err == sql.ErrNoRows {
		return userdomain.UserRegionRebuildTask{}, false, nil
	}
	if err != nil {
		return userdomain.UserRegionRebuildTask{}, false, err
	}

	return task, true, nil
}

func scanRegionRebuildTask(scanner interface{ Scan(dest ...any) error }) (userdomain.UserRegionRebuildTask, error) {
	var task userdomain.UserRegionRebuildTask
	var regionID sql.NullInt64
	var targetRegionID sql.NullInt64
	var lockedBy sql.NullString
	var lockedUntil sql.NullInt64
	var errorMessage sql.NullString
	err := scanner.Scan(
		&task.TaskID,
		&regionID,
		&task.CountryCode,
		&targetRegionID,
		&task.MappingRevision,
		&task.Status,
		&task.CursorUserID,
		&lockedBy,
		&lockedUntil,
		&task.AttemptCount,
		&errorMessage,
		&task.CreatedAtMs,
		&task.UpdatedAtMs,
	)
	if err != nil {
		return userdomain.UserRegionRebuildTask{}, err
	}
	if regionID.Valid {
		task.RegionID = regionID.Int64
	}
	if targetRegionID.Valid {
		task.TargetRegionID = targetRegionID.Int64
	}
	if lockedBy.Valid {
		task.LockedBy = lockedBy.String
	}
	if lockedUntil.Valid {
		task.LockedUntilMs = lockedUntil.Int64
	}
	if errorMessage.Valid {
		task.ErrorMessage = errorMessage.String
	}

	return task, nil
}

func newerRegionRebuildTaskExists(ctx context.Context, q queryer, countryCode string, revision int64, taskID int64) (bool, error) {
	var one int
	err := q.QueryRowContext(ctx, `
		SELECT 1
		FROM user_region_rebuild_tasks
		WHERE country_code = ?
		  AND (mapping_revision > ? OR (mapping_revision = ? AND task_id > ?))
		LIMIT 1
	`, countryCode, revision, revision, taskID).Scan(&one)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	return true, nil
}

func selectRegionRebuildUserIDs(ctx context.Context, q queryer, countryCode string, cursorUserID int64, batchSize int) ([]int64, error) {
	rows, err := queryRows(ctx, q, `
		SELECT user_id
		FROM users
		WHERE country = ? AND user_id > ?
		ORDER BY user_id ASC
		LIMIT ?
	`, countryCode, cursorUserID, batchSize)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	userIDs := make([]int64, 0, batchSize)
	for rows.Next() {
		var userID int64
		if err := rows.Scan(&userID); err != nil {
			return nil, err
		}
		userIDs = append(userIDs, userID)
	}

	return userIDs, rows.Err()
}

func updateUsersRegionByIDs(ctx context.Context, tx *sql.Tx, userIDs []int64, targetRegionID int64, nowMs int64) error {
	if len(userIDs) == 0 {
		return nil
	}
	placeholders := make([]string, 0, len(userIDs))
	args := make([]any, 0, len(userIDs)+2)
	args = append(args, nullableRegionID(targetRegionID), nowMs)
	for _, userID := range userIDs {
		placeholders = append(placeholders, "?")
		args = append(args, userID)
	}
	_, err := tx.ExecContext(ctx, `
		UPDATE users
		SET region_id = ?, updated_at_ms = ?
		WHERE user_id IN (`+strings.Join(placeholders, ",")+`)
	`, args...)
	return err
}

func finishRegionRebuildTask(ctx context.Context, tx *sql.Tx, taskID int64, workerID string, status string, cursorUserID int64, errorMessage string, nowMs int64) error {
	result, err := tx.ExecContext(ctx, `
		UPDATE user_region_rebuild_tasks
		SET status = ?, cursor_user_id = ?, locked_by = NULL, locked_until_ms = NULL, error_message = ?, updated_at_ms = ?
		WHERE task_id = ? AND status = ? AND locked_by = ?
	`, status, cursorUserID, nullableString(errorMessage), nowMs, taskID, userdomain.RegionRebuildTaskStatusRunning, workerID)
	if err != nil {
		return err
	}
	if affected, err := result.RowsAffected(); err != nil {
		return err
	} else if affected != 1 {
		// finish 和 users 更新在同一个事务里；丢失 claim 时让事务回滚，避免旧 worker 覆盖新进度。
		return xerr.New(xerr.Conflict, "region rebuild task lock is lost")
	}

	return nil
}

func insertRegionChangeLog(ctx context.Context, tx *sql.Tx, regionID int64, operation string, before any, after any, operatorUserID int64, requestID string, nowMs int64) error {
	beforeJSON, err := marshalNullableJSON(before)
	if err != nil {
		return err
	}
	afterJSON, err := marshalNullableJSON(after)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `
		INSERT INTO region_change_logs (region_id, operation, before_json, after_json, operator_user_id, request_id, created_at_ms)
		VALUES (?, ?, ?, ?, ?, ?, ?)
	`, nullableRegionID(regionID), operation, beforeJSON, afterJSON, nullableOperator(operatorUserID), requestID, nowMs)
	return err
}

func marshalNullableJSON(value any) (any, error) {
	if value == nil {
		return nil, nil
	}
	payload, err := json.Marshal(value)
	if err != nil {
		return nil, err
	}

	return string(payload), nil
}

func mapCountryDuplicateError(err error) error {
	var mysqlErr *mysqldriver.MySQLError
	if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
		return xerr.New(xerr.Conflict, "country_code already exists")
	}
	return err
}

func mapRegionDuplicateError(err error) error {
	var mysqlErr *mysqldriver.MySQLError
	if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
		return xerr.New(xerr.Conflict, "region_code already exists")
	}
	return err
}

func mapRegionCountryDuplicateError(err error) error {
	var mysqlErr *mysqldriver.MySQLError
	if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
		return xerr.New(xerr.RegionCountryConflict, "country already belongs to another active region")
	}
	return err
}

func unionCountryCodes(left []string, right []string) []string {
	seen := make(map[string]struct{}, len(left)+len(right))
	result := make([]string, 0, len(left)+len(right))
	for _, country := range append(append([]string(nil), left...), right...) {
		if _, exists := seen[country]; exists {
			continue
		}
		seen[country] = struct{}{}
		result = append(result, country)
	}
	sort.Strings(result)

	return result
}

func stringSliceContains(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}

	return false
}

func queryRows(ctx context.Context, q queryer, query string, args ...any) (*sql.Rows, error) {
	switch typed := q.(type) {
	case *sql.DB:
		return typed.QueryContext(ctx, query, args...)
	case *sql.Tx:
		return typed.QueryContext(ctx, query, args...)
	default:
		return nil, fmt.Errorf("unsupported queryer %T", q)
	}
}