1017 lines
34 KiB
Go
1017 lines
34 KiB
Go
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(
|
||
®ion.RegionID,
|
||
®ion.RegionCode,
|
||
®ion.Name,
|
||
®ion.Status,
|
||
®ion.SortOrder,
|
||
®ion.CreatedByUserID,
|
||
®ion.UpdatedByUserID,
|
||
®ion.CreatedAtMs,
|
||
®ion.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(
|
||
®ion.RegionID,
|
||
®ion.RegionCode,
|
||
®ion.Name,
|
||
®ion.Status,
|
||
®ion.SortOrder,
|
||
®ion.CreatedByUserID,
|
||
®ion.UpdatedByUserID,
|
||
®ion.CreatedAtMs,
|
||
®ion.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(®ionID)
|
||
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,
|
||
®ionID,
|
||
&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)
|
||
}
|
||
}
|