package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"math"
	"os"
	"regexp"
	"sort"
	"strings"
	"time"

	mysql "github.com/go-sql-driver/mysql"
)

const (
	protocol           = "GB32960"
	legacySourceIP     = "legacy-mysql.lingniu-prod"
	legacySourceKey    = "GB32960:LEGACY_MILEAGE@legacy-mysql.lingniu-prod"
	legacySourceCode   = "legacy_lingniu_prod"
	legacyPlatformName = "历史生产库里程"
	legacyQuality      = "legacy_mysql_backfill_before_2026-07-21"
	maxDailyMileageKM  = 2500.0
)

var vinPattern = regexp.MustCompile(`^[A-HJ-NPR-Z0-9]{17}$`)

type sourceRecord struct {
	ID            int64
	Identifier    string
	StatDate      time.Time
	DailyKM       sql.NullFloat64
	TotalKM       sql.NullFloat64
	Now           sql.NullTime
	MappingOrigin string
	VIN           string
}

type mappingIndex struct {
	Unique    map[string]string
	Ambiguous map[string]struct{}
}

type scanReport struct {
	SourceRows                   int      `json:"source_rows"`
	SourceDistinctIdentifiers    int      `json:"source_distinct_identifiers"`
	DedupedSourceRows            int      `json:"deduped_source_rows"`
	DuplicateRows                int      `json:"duplicate_rows"`
	ConflictingDuplicateGroups   int      `json:"conflicting_duplicate_groups"`
	MappedByCurrentVehicle       int      `json:"mapped_by_current_vehicle"`
	MappedByLegacyVehicle        int      `json:"mapped_by_legacy_vehicle"`
	CurrentLegacyVINConflicts    int      `json:"current_legacy_vin_conflicts"`
	UnresolvedRows               int      `json:"unresolved_rows"`
	UnresolvedIdentifiers        int      `json:"unresolved_identifiers"`
	UnresolvedEligibleRows       int      `json:"unresolved_eligible_rows"`
	UnresolvedIdentifierValues   []string `json:"unresolved_identifier_values,omitempty"`
	NullMileageRows              int      `json:"null_mileage_rows"`
	ZeroMileageRows              int      `json:"zero_mileage_rows"`
	NegativeMileageRows          int      `json:"negative_mileage_rows"`
	OverLimitMileageRows         int      `json:"over_limit_mileage_rows"`
	EligibleRows                 int      `json:"eligible_rows"`
	VINDateCollisions            int      `json:"vin_date_collisions"`
	ConflictingVINDateCollisions int      `json:"conflicting_vin_date_collisions"`
}

type targetReport struct {
	StageRows                         int64   `json:"stage_rows"`
	MissingRows                       int64   `json:"missing_rows"`
	ZeroRows                          int64   `json:"zero_rows"`
	NonzeroProtectedRows              int64   `json:"nonzero_protected_rows"`
	ExistingPositiveCandidateRows     int64   `json:"existing_positive_candidate_rows"`
	ProtectedSameMileageRows          int64   `json:"protected_same_mileage_rows"`
	ProtectedMileageSumKM             float64 `json:"protected_mileage_sum_km"`
	ProtectedMileageFingerprint       uint64  `json:"protected_mileage_fingerprint"`
	LegacyCandidateRows               int64   `json:"legacy_candidate_rows"`
	LegacyCandidateTotalMatchedRows   int64   `json:"legacy_candidate_total_matched_rows"`
	LegacyProjectedRows               int64   `json:"legacy_projected_rows"`
	LegacyProjectedTotalMatchedRows   int64   `json:"legacy_projected_total_matched_rows"`
	LegacySelectedCandidateRows       int64   `json:"legacy_selected_candidate_rows"`
	PreexistingNonLegacyRows          int64   `json:"preexisting_non_legacy_rows"`
	PreexistingNonLegacyMileageSumKM  float64 `json:"preexisting_non_legacy_mileage_sum_km"`
	PreexistingNonLegacyFingerprint   uint64  `json:"preexisting_non_legacy_fingerprint"`
	RemainingMissingOrZeroRows        int64   `json:"remaining_missing_or_zero_rows"`
	PostProtectedRows                 int64   `json:"post_protected_rows"`
	PostProtectedMileageSumKM         float64 `json:"post_protected_mileage_sum_km"`
	PostProtectedMileageFingerprint   uint64  `json:"post_protected_mileage_fingerprint"`
	ProtectedRowsUnchanged            int64   `json:"protected_rows_unchanged"`
	ProtectedMileageFingerprintStable bool    `json:"protected_mileage_fingerprint_stable"`
}

type output struct {
	Mode                string       `json:"mode"`
	CutoffExclusive     string       `json:"cutoff_exclusive"`
	Protocol            string       `json:"protocol"`
	SourceTable         string       `json:"source_table"`
	TotalMileageDivisor float64      `json:"total_mileage_divisor_m_to_km"`
	Scan                scanReport   `json:"scan"`
	Target              targetReport `json:"target"`
	Duration            string       `json:"duration"`
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

func run() error {
	var cutoffRaw, dsn string
	var apply bool
	var timeout time.Duration
	flag.StringVar(&cutoffRaw, "before", "2026-07-21", "exclusive upper date boundary (YYYY-MM-DD)")
	flag.StringVar(&dsn, "mysql-dsn", strings.TrimSpace(os.Getenv("MYSQL_DSN")), "target MySQL DSN")
	flag.BoolVar(&apply, "apply", false, "apply the migration; default is dry-run")
	flag.DurationVar(&timeout, "timeout", 10*time.Minute, "overall timeout")
	flag.Parse()

	started := time.Now()
	cutoff, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(cutoffRaw), time.FixedZone("Asia/Shanghai", 8*3600))
	if err != nil {
		return fmt.Errorf("invalid -before date: %w", err)
	}
	if strings.TrimSpace(dsn) == "" {
		return errors.New("MYSQL_DSN or -mysql-dsn is required")
	}
	dsn, err = normalizedDSN(dsn)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return err
	}
	defer db.Close()
	db.SetMaxOpenConns(2)
	if err := db.PingContext(ctx); err != nil {
		return err
	}

	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	currentMappings, err := loadMappingIndex(ctx, conn, `
SELECT CONVERT(plate USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM vehicle
WHERE plate IS NOT NULL AND TRIM(plate) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM vehicle_identifier
WHERE enabled = 1 AND plate IS NOT NULL AND TRIM(plate) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''`)
	if err != nil {
		return fmt.Errorf("load current vehicle mappings: %w", err)
	}
	legacyMappings, err := loadMappingIndex(ctx, conn, `
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.tab_truck
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.truck_info
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.v_vehicle_daily_stats
WHERE plate IS NOT NULL AND TRIM(plate) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.tab_truck_remote_sync_realtime_info
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.view_truck_comprehensive_info
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin_code USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.tab_hecri_hydrogen_order
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin_code IS NOT NULL AND TRIM(vin_code) <> ''
UNION ALL
SELECT CONVERT(plate USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.bi_ele_charge_record
WHERE plate IS NOT NULL AND TRIM(plate) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.v_truck_task_basic
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''
UNION ALL
SELECT CONVERT(plate_number USING utf8mb4) COLLATE utf8mb4_unicode_ci AS plate,
       CONVERT(vin USING utf8mb4) COLLATE utf8mb4_unicode_ci AS vin FROM lingniu_prod.view_vehicle_annual_inspection
WHERE plate_number IS NOT NULL AND TRIM(plate_number) <> '' AND vin IS NOT NULL AND TRIM(vin) <> ''`)
	if err != nil {
		return fmt.Errorf("load legacy vehicle mappings: %w", err)
	}
	records, report, err := loadAndResolveSource(ctx, conn, cutoff, currentMappings, legacyMappings)
	if err != nil {
		return err
	}
	if err := createAndLoadStage(ctx, conn, records); err != nil {
		return err
	}
	target, err := inspectTarget(ctx, conn)
	if err != nil {
		return err
	}

	mode := "dry_run"
	if apply {
		mode = "apply"
		if err := createProtectedSnapshot(ctx, conn); err != nil {
			return err
		}
		if err := applyMigration(ctx, conn); err != nil {
			return err
		}
		post, err := inspectTarget(ctx, conn)
		if err != nil {
			return err
		}
		post.PostProtectedRows, post.PostProtectedMileageSumKM, post.PostProtectedMileageFingerprint, post.ProtectedRowsUnchanged, err = inspectProtectedSnapshot(ctx, conn)
		if err != nil {
			return err
		}
		post.ProtectedMileageFingerprintStable =
			target.PreexistingNonLegacyRows == post.PostProtectedRows &&
				math.Abs(target.PreexistingNonLegacyMileageSumKM-post.PostProtectedMileageSumKM) < 0.0005 &&
				target.PreexistingNonLegacyFingerprint == post.PostProtectedMileageFingerprint &&
				post.ProtectedRowsUnchanged == target.PreexistingNonLegacyRows
		post.NonzeroProtectedRows = target.NonzeroProtectedRows
		post.ProtectedSameMileageRows = target.ProtectedSameMileageRows
		post.ProtectedMileageSumKM = target.ProtectedMileageSumKM
		post.ProtectedMileageFingerprint = target.ProtectedMileageFingerprint
		target = post
	}

	result := output{
		Mode: mode, CutoffExclusive: cutoff.Format("2006-01-02"), Protocol: protocol,
		SourceTable: "lingniu_prod.ln_vehicle_day_mileage", TotalMileageDivisor: 1000, Scan: report, Target: target,
		Duration: time.Since(started).Round(time.Millisecond).String(),
	}
	encoder := json.NewEncoder(os.Stdout)
	encoder.SetIndent("", "  ")
	return encoder.Encode(result)
}

func createProtectedSnapshot(ctx context.Context, conn *sql.Conn) error {
	_, err := conn.ExecContext(ctx, `
CREATE TEMPORARY TABLE tmp_legacy_mileage_protected AS
SELECT m.vin, m.stat_date, m.protocol, m.source_id, m.daily_mileage_km,
       m.latest_total_mileage_km, m.updated_at
FROM tmp_legacy_mileage_stage st
JOIN vehicle_daily_mileage m
  ON m.vin = st.vin AND m.stat_date = st.stat_date AND m.protocol = 'GB32960'
LEFT JOIN vehicle_data_source ds ON ds.id = m.source_id
WHERE m.daily_mileage_km > 0 AND COALESCE(ds.source_ip, '') <> ?`, legacySourceIP)
	if err != nil {
		return fmt.Errorf("snapshot protected mileage rows: %w", err)
	}
	return nil
}

func inspectProtectedSnapshot(ctx context.Context, queryer interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}) (count int64, sum float64, fingerprint uint64, unchanged int64, err error) {
	var nullableSum sql.NullFloat64
	err = queryer.QueryRowContext(ctx, `
SELECT
  COUNT(m.vin),
  SUM(CASE WHEN m.vin IS NOT NULL THEN m.daily_mileage_km ELSE 0 END),
  COALESCE(BIT_XOR(CASE WHEN m.vin IS NOT NULL THEN CRC32(CONCAT_WS('|', m.vin, m.stat_date, m.protocol, m.daily_mileage_km, COALESCE(m.latest_total_mileage_km, 'NULL'))) ELSE 0 END), 0),
  COALESCE(SUM(
    m.vin IS NOT NULL
    AND (m.source_id <=> p.source_id)
    AND (m.daily_mileage_km <=> p.daily_mileage_km)
    AND (m.latest_total_mileage_km <=> p.latest_total_mileage_km)
    AND (m.updated_at <=> p.updated_at)
  ), 0)
FROM tmp_legacy_mileage_protected p
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = p.vin AND m.stat_date = p.stat_date AND m.protocol = p.protocol`).Scan(
		&count, &nullableSum, &fingerprint, &unchanged,
	)
	if err != nil {
		return 0, 0, 0, 0, fmt.Errorf("verify protected mileage snapshot: %w", err)
	}
	if nullableSum.Valid {
		sum = nullableSum.Float64
	}
	return count, sum, fingerprint, unchanged, nil
}

func normalizedDSN(raw string) (string, error) {
	cfg, err := mysql.ParseDSN(strings.TrimSpace(raw))
	if err != nil {
		return "", fmt.Errorf("parse mysql dsn: %w", err)
	}
	cfg.ParseTime = true
	cfg.Loc = time.FixedZone("Asia/Shanghai", 8*3600)
	cfg.Params = cloneMap(cfg.Params)
	cfg.Params["charset"] = "utf8mb4"
	return cfg.FormatDSN(), nil
}

func cloneMap(source map[string]string) map[string]string {
	result := make(map[string]string, len(source)+1)
	for key, value := range source {
		result[key] = value
	}
	return result
}

func loadMappingIndex(ctx context.Context, queryer interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}, query string) (mappingIndex, error) {
	rows, err := queryer.QueryContext(ctx, query)
	if err != nil {
		return mappingIndex{}, err
	}
	defer rows.Close()
	values := map[string]map[string]struct{}{}
	for rows.Next() {
		var plate, vin sql.NullString
		if err := rows.Scan(&plate, &vin); err != nil {
			return mappingIndex{}, err
		}
		plateValue := normalizeIdentifier(plate.String)
		vinValue := normalizeVIN(vin.String)
		if plateValue == "" || !vinPattern.MatchString(vinValue) {
			continue
		}
		if values[plateValue] == nil {
			values[plateValue] = map[string]struct{}{}
		}
		values[plateValue][vinValue] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return mappingIndex{}, err
	}
	result := mappingIndex{Unique: map[string]string{}, Ambiguous: map[string]struct{}{}}
	for plate, vins := range values {
		if len(vins) != 1 {
			result.Ambiguous[plate] = struct{}{}
			continue
		}
		for vin := range vins {
			result.Unique[plate] = vin
		}
	}
	return result, nil
}

func loadAndResolveSource(ctx context.Context, queryer interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}, cutoff time.Time, current, legacy mappingIndex) ([]sourceRecord, scanReport, error) {
	rows, err := queryer.QueryContext(ctx, `
SELECT id, plateNumber, dates, dayMileage, totalMileage, nowDatetime
FROM lingniu_prod.ln_vehicle_day_mileage
WHERE dates < ?
ORDER BY plateNumber, dates, (nowDatetime IS NULL), nowDatetime DESC, id DESC`, cutoff)
	if err != nil {
		return nil, scanReport{}, fmt.Errorf("query source mileage: %w", err)
	}
	defer rows.Close()

	var report scanReport
	distinctIdentifiers := map[string]struct{}{}
	unresolvedIdentifiers := map[string]struct{}{}
	byIdentifierDate := map[string]sourceRecord{}
	duplicateValues := map[string]float64{}
	conflictingDuplicateKeys := map[string]struct{}{}
	for rows.Next() {
		var item sourceRecord
		var rawIdentifier sql.NullString
		if err := rows.Scan(&item.ID, &rawIdentifier, &item.StatDate, &item.DailyKM, &item.TotalKM, &item.Now); err != nil {
			return nil, report, err
		}
		report.SourceRows++
		item.Identifier = normalizeIdentifier(rawIdentifier.String)
		distinctIdentifiers[item.Identifier] = struct{}{}
		key := item.Identifier + "\x00" + item.StatDate.Format("2006-01-02")
		if existing, ok := byIdentifierDate[key]; ok {
			report.DuplicateRows++
			if item.DailyKM.Valid && existing.DailyKM.Valid && math.Abs(item.DailyKM.Float64-existing.DailyKM.Float64) > 0.0005 {
				conflictingDuplicateKeys[key] = struct{}{}
			}
			continue
		}
		byIdentifierDate[key] = item
		if item.DailyKM.Valid {
			duplicateValues[key] = item.DailyKM.Float64
		}
	}
	if err := rows.Err(); err != nil {
		return nil, report, err
	}
	_ = duplicateValues
	report.SourceDistinctIdentifiers = len(distinctIdentifiers)
	report.DedupedSourceRows = len(byIdentifierDate)
	report.ConflictingDuplicateGroups = len(conflictingDuplicateKeys)

	keys := make([]string, 0, len(byIdentifierDate))
	for key := range byIdentifierDate {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	stageByVINDate := map[string]sourceRecord{}
	var stageOrder []string
	for _, key := range keys {
		item := byIdentifierDate[key]
		vin, origin, conflict := resolveVIN(item.Identifier, current, legacy)
		if conflict {
			report.CurrentLegacyVINConflicts++
		}
		if vin == "" {
			report.UnresolvedRows++
			unresolvedIdentifiers[item.Identifier] = struct{}{}
			if item.DailyKM.Valid && item.DailyKM.Float64 > 0 && item.DailyKM.Float64 <= maxDailyMileageKM {
				report.UnresolvedEligibleRows++
			}
			continue
		}
		item.VIN = vin
		item.MappingOrigin = origin
		if origin == "current_vehicle" {
			report.MappedByCurrentVehicle++
		} else {
			report.MappedByLegacyVehicle++
		}
		if !item.DailyKM.Valid {
			report.NullMileageRows++
			continue
		}
		switch {
		case item.DailyKM.Float64 == 0:
			report.ZeroMileageRows++
			continue
		case item.DailyKM.Float64 < 0:
			report.NegativeMileageRows++
			continue
		case item.DailyKM.Float64 > maxDailyMileageKM:
			report.OverLimitMileageRows++
			continue
		}
		report.EligibleRows++
		stageKey := item.VIN + "\x00" + item.StatDate.Format("2006-01-02")
		if existing, ok := stageByVINDate[stageKey]; ok {
			report.VINDateCollisions++
			if math.Abs(existing.DailyKM.Float64-item.DailyKM.Float64) > 0.0005 {
				report.ConflictingVINDateCollisions++
			}
			if item.ID <= existing.ID {
				continue
			}
		} else {
			stageOrder = append(stageOrder, stageKey)
		}
		stageByVINDate[stageKey] = item
	}
	report.UnresolvedIdentifiers = len(unresolvedIdentifiers)
	for identifier := range unresolvedIdentifiers {
		report.UnresolvedIdentifierValues = append(report.UnresolvedIdentifierValues, identifier)
	}
	sort.Strings(report.UnresolvedIdentifierValues)
	sort.Strings(stageOrder)
	stage := make([]sourceRecord, 0, len(stageOrder))
	for _, key := range stageOrder {
		stage = append(stage, stageByVINDate[key])
	}
	return stage, report, nil
}

func resolveVIN(identifier string, current, legacy mappingIndex) (vin, origin string, conflict bool) {
	if vinPattern.MatchString(identifier) {
		return identifier, "source_vin", false
	}
	currentVIN, currentOK := current.Unique[identifier]
	legacyVIN, legacyOK := legacy.Unique[identifier]
	if currentOK && legacyOK && currentVIN != legacyVIN {
		return "", "", true
	}
	if currentOK {
		return currentVIN, "current_vehicle", false
	}
	if legacyOK {
		return legacyVIN, "legacy_vehicle", false
	}
	return "", "", false
}

func normalizeIdentifier(value string) string {
	return strings.ToUpper(strings.Join(strings.Fields(strings.TrimSpace(value)), ""))
}

func normalizeVIN(value string) string {
	return strings.ToUpper(strings.TrimSpace(value))
}

func createAndLoadStage(ctx context.Context, conn *sql.Conn, records []sourceRecord) error {
	if _, err := conn.ExecContext(ctx, `
CREATE TEMPORARY TABLE tmp_legacy_mileage_stage (
  vin VARCHAR(32) NOT NULL,
  stat_date DATE NOT NULL,
  daily_mileage_km DECIMAL(18,3) NOT NULL,
  total_mileage_km DECIMAL(18,3) NULL,
  source_record_id BIGINT NOT NULL,
  mapping_origin VARCHAR(32) NOT NULL,
  PRIMARY KEY (vin, stat_date)
) ENGINE=InnoDB`); err != nil {
		return fmt.Errorf("create temporary stage: %w", err)
	}
	const batchSize = 500
	for start := 0; start < len(records); start += batchSize {
		end := min(start+batchSize, len(records))
		var query strings.Builder
		query.WriteString("INSERT INTO tmp_legacy_mileage_stage (vin, stat_date, daily_mileage_km, total_mileage_km, source_record_id, mapping_origin) VALUES ")
		args := make([]any, 0, (end-start)*6)
		for index, item := range records[start:end] {
			if index > 0 {
				query.WriteByte(',')
			}
			query.WriteString("(?,?,?,?,?,?)")
			args = append(args, item.VIN, item.StatDate, item.DailyKM.Float64, nullableTotalMileageKM(item.TotalKM), item.ID, item.MappingOrigin)
		}
		if _, err := conn.ExecContext(ctx, query.String(), args...); err != nil {
			return fmt.Errorf("load temporary stage batch %d: %w", start/batchSize+1, err)
		}
	}
	return nil
}

func nullableTotalMileageKM(value sql.NullFloat64) any {
	if !value.Valid {
		return nil
	}
	return value.Float64 / 1000
}

func inspectTarget(ctx context.Context, queryer interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}) (targetReport, error) {
	var report targetReport
	var protectedSum sql.NullFloat64
	err := queryer.QueryRowContext(ctx, `
SELECT
  COUNT(*),
  COALESCE(SUM(m.vin IS NULL), 0),
  COALESCE(SUM(m.vin IS NOT NULL AND m.daily_mileage_km = 0), 0),
  COALESCE(SUM(m.daily_mileage_km > 0), 0),
  COALESCE(SUM(m.daily_mileage_km > 0 AND ABS(m.daily_mileage_km - st.daily_mileage_km) <= 0.011), 0),
  SUM(CASE WHEN m.daily_mileage_km > 0 THEN m.daily_mileage_km ELSE 0 END),
  COALESCE(BIT_XOR(CASE WHEN m.daily_mileage_km > 0 THEN CRC32(CONCAT_WS('|', m.vin, m.stat_date, m.protocol, m.daily_mileage_km, COALESCE(m.latest_total_mileage_km, 'NULL'))) ELSE 0 END), 0),
  COALESCE(SUM(candidate.vin IS NOT NULL AND (m.vin IS NULL OR m.daily_mileage_km = 0)), 0)
FROM tmp_legacy_mileage_stage st
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = st.vin AND m.stat_date = st.stat_date AND m.protocol = 'GB32960'
LEFT JOIN (
  SELECT vin, stat_date
  FROM vehicle_daily_mileage_source
  WHERE protocol = 'GB32960' AND quality_status = 'OK' AND daily_mileage_km > 0
  GROUP BY vin, stat_date
) candidate ON candidate.vin = st.vin AND candidate.stat_date = st.stat_date`).Scan(
		&report.StageRows, &report.MissingRows, &report.ZeroRows, &report.NonzeroProtectedRows,
		&report.ProtectedSameMileageRows, &protectedSum, &report.ProtectedMileageFingerprint,
		&report.ExistingPositiveCandidateRows,
	)
	if err != nil {
		return report, fmt.Errorf("inspect migration target: %w", err)
	}
	if protectedSum.Valid {
		report.ProtectedMileageSumKM = protectedSum.Float64
	}
	if err := queryer.QueryRowContext(ctx, `
SELECT COUNT(*), COALESCE(SUM(s.latest_total_mileage_km <=> st.total_mileage_km), 0)
FROM vehicle_daily_mileage_source s
JOIN tmp_legacy_mileage_stage st ON st.vin = s.vin AND st.stat_date = s.stat_date
WHERE s.protocol = 'GB32960' AND s.source_key = ?`, legacySourceKey).Scan(
		&report.LegacyCandidateRows, &report.LegacyCandidateTotalMatchedRows,
	); err != nil {
		return report, err
	}
	if err := queryer.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM vehicle_daily_mileage_source s
JOIN tmp_legacy_mileage_stage st ON st.vin = s.vin AND st.stat_date = s.stat_date
WHERE s.protocol = 'GB32960' AND s.source_key = ? AND s.is_selected = 1`, legacySourceKey).Scan(&report.LegacySelectedCandidateRows); err != nil {
		return report, err
	}
	if err := queryer.QueryRowContext(ctx, `
SELECT COUNT(*), COALESCE(SUM(m.latest_total_mileage_km <=> st.total_mileage_km), 0)
FROM vehicle_daily_mileage m
JOIN tmp_legacy_mileage_stage st ON st.vin = m.vin AND st.stat_date = m.stat_date
JOIN vehicle_data_source ds ON ds.id = m.source_id
WHERE m.protocol = 'GB32960' AND ds.protocol = 'GB32960' AND ds.source_ip = ?
  AND ABS(m.daily_mileage_km - st.daily_mileage_km) <= 0.0005`, legacySourceIP).Scan(
		&report.LegacyProjectedRows, &report.LegacyProjectedTotalMatchedRows,
	); err != nil {
		return report, err
	}
	if err := queryer.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM tmp_legacy_mileage_stage st
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = st.vin AND m.stat_date = st.stat_date AND m.protocol = 'GB32960'
WHERE m.vin IS NULL OR m.daily_mileage_km = 0`).Scan(&report.RemainingMissingOrZeroRows); err != nil {
		return report, err
	}
	var nonLegacySum sql.NullFloat64
	if err := queryer.QueryRowContext(ctx, `
SELECT
  COUNT(*),
  SUM(m.daily_mileage_km),
  COALESCE(BIT_XOR(CRC32(CONCAT_WS('|', m.vin, m.stat_date, m.protocol, m.daily_mileage_km, COALESCE(m.latest_total_mileage_km, 'NULL')))), 0)
FROM vehicle_daily_mileage m
JOIN tmp_legacy_mileage_stage st ON st.vin = m.vin AND st.stat_date = m.stat_date
LEFT JOIN vehicle_data_source ds ON ds.id = m.source_id
WHERE m.protocol = 'GB32960' AND m.daily_mileage_km > 0
  AND COALESCE(ds.source_ip, '') <> ?`, legacySourceIP).Scan(
		&report.PreexistingNonLegacyRows, &nonLegacySum, &report.PreexistingNonLegacyFingerprint,
	); err != nil {
		return report, err
	}
	if nonLegacySum.Valid {
		report.PreexistingNonLegacyMileageSumKM = nonLegacySum.Float64
	}
	return report, nil
}

func applyMigration(ctx context.Context, conn *sql.Conn) error {
	tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, `
INSERT INTO vehicle_data_source
  (protocol, source_ip, latest_source_endpoint, platform_name, source_code, source_kind,
   trust_priority, enabled, first_seen_at, latest_seen_at, remark)
VALUES ('GB32960', ?, 'lingniu_prod.ln_vehicle_day_mileage', ?, ?, 'UNKNOWN', 1000, 1,
        '2024-10-17 00:00:00', '2026-07-20 23:59:59',
        '仅用于迁移 2026-07-21 之前缺失或为 0 的 GB32960 日里程')
ON DUPLICATE KEY UPDATE
  latest_source_endpoint = VALUES(latest_source_endpoint),
  platform_name = VALUES(platform_name),
  source_code = VALUES(source_code),
  source_kind = VALUES(source_kind),
  trust_priority = VALUES(trust_priority),
  enabled = 1,
  remark = VALUES(remark)`, legacySourceIP, legacyPlatformName, legacySourceCode); err != nil {
		return fmt.Errorf("upsert legacy data source: %w", err)
	}
	var sourceID int64
	if err := tx.QueryRowContext(ctx, `SELECT id FROM vehicle_data_source WHERE protocol = 'GB32960' AND source_ip = ?`, legacySourceIP).Scan(&sourceID); err != nil {
		return fmt.Errorf("lookup legacy data source: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
INSERT INTO vehicle_daily_mileage_source
  (vin, stat_date, protocol, source_key, source_ip, source_endpoint, platform_name,
   first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, sample_count,
   first_event_time, latest_event_time, quality_status, quality_reason, is_selected)
SELECT
  st.vin, st.stat_date, 'GB32960', ?, ?, 'lingniu_prod.ln_vehicle_day_mileage', ?,
  CASE WHEN st.total_mileage_km IS NOT NULL AND st.total_mileage_km >= st.daily_mileage_km
       THEN st.total_mileage_km - st.daily_mileage_km ELSE NULL END,
  st.total_mileage_km, st.daily_mileage_km, 1,
  TIMESTAMP(st.stat_date, '00:00:00'), TIMESTAMP(st.stat_date, '23:59:59'),
  'OK', ?, 0
FROM tmp_legacy_mileage_stage st
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = st.vin AND m.stat_date = st.stat_date AND m.protocol = 'GB32960'
WHERE m.vin IS NULL OR m.daily_mileage_km = 0 OR m.source_id = ?
ON DUPLICATE KEY UPDATE
	updated_at = IF(
		NOT (vehicle_daily_mileage_source.first_total_mileage_km <=> VALUES(first_total_mileage_km))
		OR NOT (vehicle_daily_mileage_source.latest_total_mileage_km <=> VALUES(latest_total_mileage_km))
		OR NOT (vehicle_daily_mileage_source.daily_mileage_km <=> VALUES(daily_mileage_km))
		OR NOT (vehicle_daily_mileage_source.quality_status <=> VALUES(quality_status))
		OR NOT (vehicle_daily_mileage_source.quality_reason <=> VALUES(quality_reason)),
		CURRENT_TIMESTAMP,
		vehicle_daily_mileage_source.updated_at
	),
	source_ip = VALUES(source_ip),
  source_endpoint = VALUES(source_endpoint),
  platform_name = VALUES(platform_name),
  first_total_mileage_km = VALUES(first_total_mileage_km),
  latest_total_mileage_km = VALUES(latest_total_mileage_km),
  daily_mileage_km = VALUES(daily_mileage_km),
  sample_count = VALUES(sample_count),
  first_event_time = VALUES(first_event_time),
  latest_event_time = VALUES(latest_event_time),
	quality_status = VALUES(quality_status),
	quality_reason = VALUES(quality_reason)`, legacySourceKey, legacySourceIP, legacyPlatformName, legacyQuality, sourceID); err != nil {
		return fmt.Errorf("upsert legacy mileage candidates: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
INSERT INTO vehicle_daily_mileage
  (vin, stat_date, protocol, source_id, daily_mileage_km, latest_total_mileage_km)
SELECT st.vin, st.stat_date, 'GB32960', ?, st.daily_mileage_km, st.total_mileage_km
FROM tmp_legacy_mileage_stage st
LEFT JOIN vehicle_daily_mileage m
  ON m.vin = st.vin AND m.stat_date = st.stat_date AND m.protocol = 'GB32960'
WHERE m.vin IS NULL OR m.daily_mileage_km = 0 OR m.source_id = ?
ON DUPLICATE KEY UPDATE
	updated_at = IF(
		vehicle_daily_mileage.daily_mileage_km = 0
		OR (
			(vehicle_daily_mileage.source_id <=> VALUES(source_id))
			AND (
				NOT (vehicle_daily_mileage.daily_mileage_km <=> VALUES(daily_mileage_km))
				OR NOT (vehicle_daily_mileage.latest_total_mileage_km <=> VALUES(latest_total_mileage_km))
			)
		),
		CURRENT_TIMESTAMP,
		vehicle_daily_mileage.updated_at
	),
	source_id = IF(vehicle_daily_mileage.daily_mileage_km = 0 OR (vehicle_daily_mileage.source_id <=> VALUES(source_id)), VALUES(source_id), vehicle_daily_mileage.source_id),
	latest_total_mileage_km = IF(vehicle_daily_mileage.daily_mileage_km = 0 OR (vehicle_daily_mileage.source_id <=> VALUES(source_id)), VALUES(latest_total_mileage_km), vehicle_daily_mileage.latest_total_mileage_km),
	daily_mileage_km = IF(vehicle_daily_mileage.daily_mileage_km = 0 OR (vehicle_daily_mileage.source_id <=> VALUES(source_id)), VALUES(daily_mileage_km), vehicle_daily_mileage.daily_mileage_km)`, sourceID, sourceID); err != nil {
		return fmt.Errorf("upsert final daily mileage: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
UPDATE vehicle_daily_mileage_source s
JOIN tmp_legacy_mileage_stage st
  ON st.vin = s.vin AND st.stat_date = s.stat_date
JOIN vehicle_daily_mileage m
  ON m.vin = st.vin AND m.stat_date = st.stat_date AND m.protocol = 'GB32960'
SET s.is_selected = CASE WHEN s.source_key = ? THEN 1 ELSE 0 END
WHERE s.protocol = 'GB32960'
  AND m.source_id = ?
  AND ABS(m.daily_mileage_km - st.daily_mileage_km) <= 0.0005`, legacySourceKey, sourceID); err != nil {
		return fmt.Errorf("mark legacy mileage selections: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit migration: %w", err)
	}
	return nil
}