package bootstrapimpl

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"os"
	"regexp"
	"strconv"
	"strings"

	"github.com/nexuscrm/automation/autonumber"
)

// envTolerateLegacy lets a deploy environment opt out of PRD §8.2 step
// 3.c fail-closed on legacy rows whose format was produced by a
// pre-v2 listener and no longer matches the current format spec's
// derived regex. When "true", unparseable rows are logged at WARN and
// skipped; max is computed only from the rows that match. This is
// required for production migrations where live data pre-dates the
// format cutover — the default (unset / "false") preserves PRD
// strict behavior for CI + local dev.
const envTolerateLegacy = "AUTONUMBER_SEED_TOLERATE_LEGACY"

func tolerateLegacySeedRows() bool {
	return os.Getenv(envTolerateLegacy) == "true"
}

// legacySampleLimit caps per-row WARN output in tolerant-mode seed
// scans. Staging tables like Interface_Log__c carry ~2.87M pre-cutover
// rows; a per-row log.Printf would stall boot past the ALB health-check
// window. The first legacySampleLimit rows are logged verbatim so ops
// can still triage a representative sample; remaining skips are
// collapsed into a single summary line.
const legacySampleLimit = 5

// legacyRowSampler collects per-row WARN samples up to legacySampleLimit
// and emits one aggregate summary line on flush. Single-goroutine use
// only (matches scanLiveMaxCounter / scanLivePerPrefixCounters which
// iterate rows serially).
//
// atLeast flips the summary from "skipped N" → "skipped at least N"
// when count is a known lower bound (sample-query hit its LIMIT cap
// without a successful follow-up COUNT, or mid-scan iteration errored
// before observing the full tail). Round-1 code review Pass B HIGH-1
// + HIGH-2 fix.
type legacyRowSampler struct {
	objectName, fieldName, fieldID string
	derivedRegex                   string
	count                          int
	atLeast                        bool
}

func newLegacyRowSampler(objectName, fieldName, fieldID string, derivedRegex *regexp.Regexp) *legacyRowSampler {
	return &legacyRowSampler{
		objectName:   objectName,
		fieldName:    fieldName,
		fieldID:      fieldID,
		derivedRegex: derivedRegex.String(),
	}
}

func (l *legacyRowSampler) observe(rowID, rawValue string) {
	l.count++
	if l.count <= legacySampleLimit {
		log.Printf("⚠️  autonumber seed: field %s.%s (field_id=%s): row Id=%q value %q does not match derived regex %q — skipping (%s=true)",
			l.objectName, l.fieldName, l.fieldID, rowID, rawValue,
			l.derivedRegex, envTolerateLegacy)
	}
}

func (l *legacyRowSampler) flush() {
	if l.count > legacySampleLimit {
		qualifier := ""
		if l.atLeast {
			qualifier = "at least "
		}
		log.Printf("⚠️  autonumber seed: field %s.%s (field_id=%s): skipped %s%d legacy rows that do not match derived regex %q (%d samples logged above) (%s=true)",
			l.objectName, l.fieldName, l.fieldID, qualifier, l.count,
			l.derivedRegex, legacySampleLimit, envTolerateLegacy)
	}
}

// envMatchCountCeiling overrides the default match-count ceiling that
// triggers fallback to the iterate-all scan path. Intended for forward
// compatibility: today's post-cutover match sets are tiny (10s–100s
// rows), but a long-running field may grow into the ceiling years out.
// Raising this env lets ops defer the fallback; the default is
// defaultMatchCountCeiling.
const envMatchCountCeiling = "AUTONUMBER_SEED_MATCH_CEILING"

// defaultMatchCountCeiling is the maximum number of regex-matching
// rows the SQL fast path will iterate before falling back to the
// iterate-all path. Chosen as a round safety margin above today's
// expected post-cutover match counts (Interface_Log__c has ~42 today,
// ~4K/year growth — the ceiling holds until 2028+ at that rate).
// Fallback is correct-but-slow, not wrong; the ceiling is a perf guard,
// not a safety invariant.
const defaultMatchCountCeiling = 10000

func matchCountCeiling() int {
	v := os.Getenv(envMatchCountCeiling)
	if v == "" {
		return defaultMatchCountCeiling
	}
	n, err := strconv.Atoi(v)
	if err != nil || n <= 0 {
		return defaultMatchCountCeiling
	}
	return n
}

// allowedRegexEscapes is the set of characters that may follow a
// backslash in a regex emitted by deriveCounterRegex. Today only `\d`
// (counter / date tokens) and the QuoteMeta set (`.\+*?()|[]{}^$` —
// the punctuation regexp.QuoteMeta escapes) are possible. If a future
// format introduces `\s`, `\w`, `\b`, etc., sqlRegexFromGoRegex fails
// loud at boot so the mismatch surfaces in unit-test / staging rather
// than as silent semantic drift between Go regex and TiDB REGEXP.
// Round-1 Pass A LOW finding.
const allowedRegexEscapes = `d.\+*?()|[]{}^$`

// sqlRegexFromGoRegex translates a Go regex produced by
// deriveCounterRegex into a MySQL/TiDB-REGEXP-compatible pattern. The
// only translation required today is `\d → [0-9]`; every other
// backslash escape in the input must come from the anchored / QuoteMeta
// set (see allowedRegexEscapes). Unknown escapes fail loud so future
// format tokens cannot silently diverge across engines (Round-1 Pass B
// HIGH-3 residual risk).
//
// NOTE: TiDB REGEXP is collation-dependent (case-INSENSITIVE by default
// on utf8mb4_unicode_ci). Callers MUST wrap the predicate with
// `COLLATE utf8mb4_bin` to match Go regexp's case-sensitive semantics.
// This function returns the pattern only; it does not inject the COLLATE
// clause (that belongs at the call site where the query is assembled).
func sqlRegexFromGoRegex(re *regexp.Regexp) (string, error) {
	s := re.String()
	for i := 0; i < len(s); i++ {
		if s[i] != '\\' {
			continue
		}
		if i+1 >= len(s) {
			return "", fmt.Errorf("sqlRegexFromGoRegex: trailing backslash in %q", s)
		}
		next := s[i+1]
		if !strings.ContainsRune(allowedRegexEscapes, rune(next)) {
			return "", fmt.Errorf(
				"sqlRegexFromGoRegex: unsupported escape \\%c in %q (expected one of %q — extend the translator before landing a new format token)",
				next, s, allowedRegexEscapes,
			)
		}
		i++ // skip the escaped char on next iteration
	}
	return strings.ReplaceAll(s, `\d`, `[0-9]`), nil
}

// validSQLIdentifier is the whitelist regex for object / field names
// that we splice into SQL via fmt.Sprintf (scanLiveMaxCounter,
// queryLiveMaxCounter, V5 live scan). Round-2 review V1-C1: backticks
// alone are NOT sufficient hardening — a name containing a backtick
// would escape the quoting and enable injection. The set below matches
// the NexusCRM physical schema convention (letter-led, ASCII
// alphanumeric + underscore, ≤63 chars, matching MySQL/TiDB identifier
// limits).
var validSQLIdentifier = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_]{0,62}$`)

// assertValidSQLIdentifier returns an error if name is not a safe SQL
// identifier per validSQLIdentifier. Call before fmt.Sprintf'ing name
// into a query string.
func assertValidSQLIdentifier(kind, name string) error {
	if !validSQLIdentifier.MatchString(name) {
		return fmt.Errorf("autonumber seed: invalid %s %q (must match %s)",
			kind, name, validSQLIdentifier.String())
	}
	return nil
}

// deriveCounterRegex compiles a regex that matches the *rendered* form
// of a value generated under spec. Capture group 1 is the counter.
// Non-counter tokens use broad wildcards so we do not need to resolve
// cross-record values during seed time (PRD §8.2 step 3.a).
//
// Tokens map to:
//
//	LITERAL       → QuoteMeta(literal bytes)
//	COUNTER       → (\d{width})        CAPTURED (exact width)
//	DATE(YYYY)    → \d{4}              non-captured
//	DATE(YY/MM/DD)→ \d{2}              non-captured
//	FIELD / RT    → .*                 non-captured (lazy-ish; .* is safe
//	                                     because the regex is anchored)
//
// The result is anchored with `^...$`.
func deriveCounterRegex(spec autonumber.FormatSpec) (*regexp.Regexp, error) {
	var b strings.Builder
	b.WriteByte('^')
	counterSeen := false
	for _, tok := range spec.Tokens {
		switch tok.Kind {
		case autonumber.TokenKindLiteral:
			b.WriteString(regexp.QuoteMeta(tok.Lit))
		case autonumber.TokenKindCounter:
			if counterSeen {
				return nil, errors.New("derive regex: multiple counter tokens (parser should have rejected)")
			}
			counterSeen = true
			// Minimum-width match. The renderer uses `%0*d` so a
			// counter of width W outputs `max(W, len(counter))`
			// digits: values within width are zero-padded; values
			// that exceed the declared width grow naturally. A
			// `{0}` field (width=1) legitimately renders 1, 2, …,
			// 1234 — the regex must accept all of them. Using
			// `\d{W,}` keeps the short-form minimum-width bound
			// while allowing legitimate growth.
			fmt.Fprintf(&b, `(\d{%d,})`, tok.Width)
		case autonumber.TokenKindDate:
			switch tok.DateName {
			case "YYYY":
				b.WriteString(`\d{4}`)
			case "YY", "MM", "DD":
				b.WriteString(`\d{2}`)
			default:
				return nil, fmt.Errorf("derive regex: unknown date token %q", tok.DateName)
			}
		case autonumber.TokenKindField, autonumber.TokenKindRecordType:
			// Cross-record or RecordType taxonomy token — both match any
			// non-empty stretch of chars; the anchored `^...$` bounds it.
			// Non-captured. Parser distinguishes the two kinds at
			// validation time; derive regex treats them uniformly
			// (live-data width is open-ended).
			b.WriteString(`.*`)
		default:
			// Future-proof: an added TokenKind triggers a runtime hit
			// that surfaces in seed tests before we ever consult DB data.
			return nil, fmt.Errorf("derive regex: unknown token kind %v", tok.Kind)
		}
	}
	b.WriteByte('$')
	if !counterSeen {
		return nil, errors.New("derive regex: no counter token (parser should have rejected)")
	}
	return regexp.Compile(b.String())
}

// scanLiveMaxCounter returns the maximum counter value observed in the
// live target column for a `global` scope field. FAIL-CLOSED on legacy
// rows in strict mode; tolerant mode (envTolerateLegacy=true) samples +
// summarizes legacy rows instead. Returns 0 if the target table/column
// does not exist or the column is empty.
//
// Fast path: push the derived regex into SQL via
// `<field> COLLATE utf8mb4_bin REGEXP ?`. DB returns only the
// small set of post-cutover matching rows (10s–100s on a post-cutover
// table) rather than every live row (millions on tables like
// Interface_Log__c). Go-side iteration builds the max; legacy-tail
// handling uses one more targeted query (merged count+samples for
// tolerant mode; LIMIT 1 abort query for strict mode).
//
// The fast path falls back to scanLiveMaxCounterIterateAll when:
//   - matching-row count exceeds matchCountCeiling (pathological growth
//     guard — Round-2 HIGH-N2), or
//   - a row TiDB-REGEXP-matched but Go regexp rejects (engine parity
//     divergence, e.g. a format shape we did not anticipate). Falling
//     back preserves the pre-v2 safety net — Round-2 HIGH-3 residual
//     risk.
func scanLiveMaxCounter(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	derivedRegex *regexp.Regexp,
	fieldID string,
) (int64, error) {
	// V1-C1 SQL injection guard: reject non-whitelisted identifiers
	// BEFORE fmt.Sprintf'ing them into a query. Backtick-quoting
	// alone is not sufficient (a backtick inside the name would
	// escape).
	if err := assertValidSQLIdentifier("objectName", objectName); err != nil {
		return 0, err
	}
	if err := assertValidSQLIdentifier("fieldName", fieldName); err != nil {
		return 0, err
	}

	// Probe column existence — fresh DB may not have the physical
	// table yet (applyAutoNumberFormats seeds metadata first; live
	// tables are created by other phases). Match the skip pattern
	// used by ensureStandardAutoNumbers (schema.go:1855+).
	ok, err := physicalColumnExists(ctx, rawDB, objectName, fieldName)
	if err != nil {
		return 0, fmt.Errorf("autonumber seed %s.%s: probe column: %w",
			objectName, fieldName, err)
	}
	if !ok {
		return 0, nil
	}

	// Round-2 CRITICAL-1: COLLATE utf8mb4_bin forces case-sensitive
	// REGEXP to match Go regexp's semantics.
	// Round-2 CRITICAL-2: regex passed as a bound `?` param — never
	// interpolated into the query string.
	sqlRegex, err := sqlRegexFromGoRegex(derivedRegex)
	if err != nil {
		return 0, fmt.Errorf("autonumber seed %s.%s: %w", objectName, fieldName, err)
	}

	matchQuery := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` "+
			"WHERE `%s` IS NOT NULL AND `%s` != '' "+
			"AND `%s` COLLATE utf8mb4_bin REGEXP ?",
		fieldName, objectName, fieldName, fieldName, fieldName,
	)
	rows, err := rawDB.QueryContext(ctx, matchQuery, sqlRegex)
	if err != nil {
		return 0, fmt.Errorf("autonumber seed %s.%s: match scan: %w",
			objectName, fieldName, err)
	}
	defer rows.Close()

	ceiling := matchCountCeiling()
	var matchCount int
	var maxCounter int64
	var fallbackReason string
	for rows.Next() {
		matchCount++
		if matchCount > ceiling {
			fallbackReason = fmt.Sprintf(
				"%d matching rows exceeds ceiling %d (%s; raise via %s or optimize this field's format)",
				matchCount, ceiling, envMatchCountCeiling, envMatchCountCeiling,
			)
			break
		}
		var rowID, rawValue sql.NullString
		if err := rows.Scan(&rowID, &rawValue); err != nil {
			return 0, fmt.Errorf("autonumber seed %s.%s: scan row: %w",
				objectName, fieldName, err)
		}
		if !rawValue.Valid || rawValue.String == "" {
			continue
		}
		// Re-validate Go-side — TiDB REGEXP and Go regexp share most
		// semantics under COLLATE utf8mb4_bin but could theoretically
		// diverge on exotic backtracking (Round-2 HIGH-3). Any row
		// TiDB matched but Go rejects triggers fallback rather than a
		// bogus max value.
		m := derivedRegex.FindStringSubmatch(rawValue.String)
		if m == nil || len(m) < 2 {
			fallbackReason = fmt.Sprintf(
				"row Id=%q value %q passed TiDB REGEXP but fails Go regex %q (engine parity divergence)",
				rowID.String, rawValue.String, derivedRegex.String(),
			)
			break
		}
		counter, err := strconv.ParseInt(m[1], 10, 64)
		if err != nil {
			// A \d+ match that ParseInt rejects means the captured
			// number overflows int64. Structured abort — admin must
			// widen or rotate.
			return 0, fmt.Errorf(
				"autonumber seed %s.%s: row Id=%q counter %q overflows int64: %w",
				objectName, fieldName, rowID.String, m[1], err,
			)
		}
		if counter > maxCounter {
			maxCounter = counter
		}
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("autonumber seed %s.%s: row iteration: %w",
			objectName, fieldName, err)
	}
	rows.Close()

	if fallbackReason != "" {
		log.Printf("⚠️  autonumber seed: field %s.%s (field_id=%s): %s — falling back to iterate-all scan",
			objectName, fieldName, fieldID, fallbackReason)
		return scanLiveMaxCounterIterateAll(ctx, rawDB, objectName, fieldName, derivedRegex, fieldID)
	}

	// Legacy-tail handling. Strict mode aborts on the first legacy
	// row; tolerant mode logs a sample + count.
	if tolerateLegacySeedRows() {
		if err := logLegacyTolerant(ctx, rawDB, objectName, fieldName, fieldID, derivedRegex, sqlRegex); err != nil {
			return 0, err
		}
	} else {
		if err := abortOnFirstLegacy(ctx, rawDB, objectName, fieldName, fieldID, derivedRegex, sqlRegex); err != nil {
			return 0, err
		}
	}
	return maxCounter, nil
}

// scanLiveMaxCounterIterateAll is the pre-v2 iterate-every-row
// implementation, retained as a fallback when the fast path's guards
// (HIGH-N2 ceiling, HIGH-3 engine parity) trigger. Semantics identical
// to the original scanLiveMaxCounter body so the fallback path is
// correct-but-slow, never wrong.
func scanLiveMaxCounterIterateAll(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	derivedRegex *regexp.Regexp,
	fieldID string,
) (int64, error) {
	query := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` WHERE `%s` IS NOT NULL AND `%s` != ''",
		fieldName, objectName, fieldName, fieldName,
	)
	rows, err := rawDB.QueryContext(ctx, query)
	if err != nil {
		return 0, fmt.Errorf("autonumber seed %s.%s: scan live data: %w",
			objectName, fieldName, err)
	}
	defer rows.Close()

	var maxCounter int64
	tolerateLegacy := tolerateLegacySeedRows()
	sampler := newLegacyRowSampler(objectName, fieldName, fieldID, derivedRegex)
	defer sampler.flush()
	for rows.Next() {
		var rowID, rawValue sql.NullString
		if err := rows.Scan(&rowID, &rawValue); err != nil {
			return 0, fmt.Errorf("autonumber seed %s.%s: scan row: %w",
				objectName, fieldName, err)
		}
		if !rawValue.Valid || rawValue.String == "" {
			continue
		}
		m := derivedRegex.FindStringSubmatch(rawValue.String)
		if m == nil {
			if tolerateLegacy {
				sampler.observe(rowID.String, rawValue.String)
				continue
			}
			return 0, fmt.Errorf(
				"autonumber seed: field %s.%s (field_id=%s): row Id=%q "+
					"has value %q which does not match derived regex %q. "+
					"Fix the row (e.g. SET %s = NULL) and re-run",
				objectName, fieldName, fieldID, rowID.String, rawValue.String,
				derivedRegex.String(), fieldName,
			)
		}
		if len(m) < 2 {
			return 0, fmt.Errorf(
				"autonumber seed %s.%s: derived regex %q matched but produced no counter capture on value %q",
				objectName, fieldName, derivedRegex.String(), rawValue.String,
			)
		}
		counter, err := strconv.ParseInt(m[1], 10, 64)
		if err != nil {
			return 0, fmt.Errorf(
				"autonumber seed %s.%s: row Id=%q counter %q overflows int64: %w",
				objectName, fieldName, rowID.String, m[1], err,
			)
		}
		if counter > maxCounter {
			maxCounter = counter
		}
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("autonumber seed %s.%s: row iteration: %w",
			objectName, fieldName, err)
	}
	return maxCounter, nil
}

// logLegacyTolerant emits the hotfix sampler contract (≤legacySampleLimit
// samples + one summary line) using a single SQL query that returns
// samples + enough rows to detect the >ceiling-legacy case.
// ORDER BY Id makes samples deterministic across boots (Round-2
// MEDIUM-N1).
func logLegacyTolerant(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName, fieldID string,
	derivedRegex *regexp.Regexp,
	sqlRegex string,
) error {
	// LIMIT N+1: if we get N+1 rows, we know count > N without
	// counting the full tail. The summary cites `≥N+1` in that case.
	sampleQuery := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` "+
			"WHERE `%s` IS NOT NULL AND `%s` != '' "+
			"AND `%s` COLLATE utf8mb4_bin NOT REGEXP ? "+
			"ORDER BY `Id` LIMIT ?",
		fieldName, objectName, fieldName, fieldName, fieldName,
	)
	// Fetch up to legacySampleLimit+1 legacy rows. Count query fills
	// in the exact total when the sample LIMIT is breached.
	rows, err := rawDB.QueryContext(ctx, sampleQuery, sqlRegex, legacySampleLimit+1)
	if err != nil {
		return fmt.Errorf("autonumber seed %s.%s: legacy sample: %w",
			objectName, fieldName, err)
	}
	defer rows.Close()

	sampler := newLegacyRowSampler(objectName, fieldName, fieldID, derivedRegex)
	defer sampler.flush()
	var sampledCount int
	for rows.Next() {
		sampledCount++
		var rowID, rawValue sql.NullString
		if err := rows.Scan(&rowID, &rawValue); err != nil {
			// Iteration aborted — sampler.count reflects what we
			// observed but may under-report the true tail. Mark
			// as lower bound so the deferred summary is honest.
			sampler.atLeast = true
			return fmt.Errorf("autonumber seed %s.%s: scan legacy row: %w",
				objectName, fieldName, err)
		}
		sampler.observe(rowID.String, rawValue.String)
	}
	if err := rows.Err(); err != nil {
		sampler.atLeast = true
		return fmt.Errorf("autonumber seed %s.%s: legacy iteration: %w",
			objectName, fieldName, err)
	}
	rows.Close()

	// If the sample query filled its LIMIT (returned
	// legacySampleLimit+1), there ARE more legacy rows than we
	// fetched — mark as lower bound so flush() says "skipped at
	// least N legacy rows". We deliberately do NOT fire a follow-up
	// COUNT(*) NOT REGEXP query: on tables with a large legacy tail
	// (e.g. staging's Interface_Log__c with 4.47M rows), that COUNT
	// cannot early-terminate and costs as much as the pre-v2
	// iterate-all scan — cancelling the whole fast-path gain. Ops
	// get a lower bound instead of an exact number, which is
	// acceptable: samples are for triage, not accounting. Exact
	// legacy counts are still available via `cmd/setup verify
	// --deep` if ever needed.
	if sampledCount > legacySampleLimit {
		sampler.atLeast = true
	}
	return nil
}

// abortOnFirstLegacy implements the strict-mode FAIL-CLOSED check via a
// targeted LIMIT 1 query. ORDER BY Id makes the reported row
// deterministic across boots (Round-2 MEDIUM-2).
//
// The `Id` PK column is a NexusCRM-wide convention (see
// system_tables.json — every physical SObject table has an `Id`
// VARCHAR(255) PRIMARY KEY). scanLiveMaxCounter has assumed this
// since its original introduction (see the shared comment near the
// SELECT Id, `<field>` shape). If a future object deviates, both the
// fast-path matching query and this abort query would break at
// runtime with "unknown column Id" rather than silently returning
// wrong data.
func abortOnFirstLegacy(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName, fieldID string,
	derivedRegex *regexp.Regexp,
	sqlRegex string,
) error {
	query := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` "+
			"WHERE `%s` IS NOT NULL AND `%s` != '' "+
			"AND `%s` COLLATE utf8mb4_bin NOT REGEXP ? "+
			"ORDER BY `Id` LIMIT 1",
		fieldName, objectName, fieldName, fieldName, fieldName,
	)
	var rowID, rawValue sql.NullString
	err := rawDB.QueryRowContext(ctx, query, sqlRegex).Scan(&rowID, &rawValue)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("autonumber seed %s.%s: strict legacy check: %w",
			objectName, fieldName, err)
	}
	return fmt.Errorf(
		"autonumber seed: field %s.%s (field_id=%s): row Id=%q "+
			"has value %q which does not match derived regex %q. "+
			"Fix the row (e.g. SET %s = NULL) and re-run",
		objectName, fieldName, fieldID, rowID.String, rawValue.String,
		derivedRegex.String(), fieldName,
	)
}

// scanLivePerPrefixCounters scans the live column and groups observed
// counter values by the rendered non-counter prefix (the value that
// autonumber.RenderScopePrefix would produce for a record). Returns a
// map keyed by prefix → max counter seen in that bucket. Skips
// unparseable rows ONLY when they are NULL/empty; any non-empty row
// that fails to match `derivedRegex` aborts fail-closed, matching the
// per-object live-scan behavior of scanLiveMaxCounter.
//
// The prefix is reconstructed from the regex's capture-group index:
// `match[0:counterStart] + match[counterEnd:matchEnd]` — i.e. the
// full matched substring with the counter carved out. For all
// cutover per_prefix formats the counter is the last token, so the
// prefix is simply everything before it; the carve-out keeps the
// function correct for hypothetical future formats where the counter
// is not last.
func scanLivePerPrefixCounters(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	derivedRegex *regexp.Regexp,
	fieldID string,
) (map[string]int64, error) {
	if err := assertValidSQLIdentifier("objectName", objectName); err != nil {
		return nil, err
	}
	if err := assertValidSQLIdentifier("fieldName", fieldName); err != nil {
		return nil, err
	}

	ok, err := physicalColumnExists(ctx, rawDB, objectName, fieldName)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: probe column: %w",
			objectName, fieldName, err)
	}
	if !ok {
		return nil, nil
	}

	// Same fast-path pattern as scanLiveMaxCounter (see that function
	// for Round-2 CRITICAL-1/-2/HIGH-N2 rationale). Shared helper
	// logic is not extracted because the iteration body diverges:
	// per-prefix builds a prefix→max map from the regex capture
	// indices, while global scope tracks a single max.
	sqlRegex, err := sqlRegexFromGoRegex(derivedRegex)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: %w", objectName, fieldName, err)
	}

	matchQuery := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` "+
			"WHERE `%s` IS NOT NULL AND `%s` != '' "+
			"AND `%s` COLLATE utf8mb4_bin REGEXP ?",
		fieldName, objectName, fieldName, fieldName, fieldName,
	)
	rows, err := rawDB.QueryContext(ctx, matchQuery, sqlRegex)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: match scan: %w",
			objectName, fieldName, err)
	}
	defer rows.Close()

	ceiling := matchCountCeiling()
	out := make(map[string]int64)
	var matchCount int
	var fallbackReason string
	for rows.Next() {
		matchCount++
		if matchCount > ceiling {
			fallbackReason = fmt.Sprintf(
				"%d matching rows exceeds ceiling %d (%s; raise via %s or optimize this field's format)",
				matchCount, ceiling, envMatchCountCeiling, envMatchCountCeiling,
			)
			break
		}
		var rowID, rawValue sql.NullString
		if err := rows.Scan(&rowID, &rawValue); err != nil {
			return nil, fmt.Errorf("autonumber seed %s.%s: scan row: %w",
				objectName, fieldName, err)
		}
		if !rawValue.Valid || rawValue.String == "" {
			continue
		}
		loc := derivedRegex.FindStringSubmatchIndex(rawValue.String)
		if loc == nil || len(loc) < 4 {
			fallbackReason = fmt.Sprintf(
				"row Id=%q value %q passed TiDB REGEXP but fails Go regex %q (engine parity divergence)",
				rowID.String, rawValue.String, derivedRegex.String(),
			)
			break
		}
		counterStr := rawValue.String[loc[2]:loc[3]]
		counter, err := strconv.ParseInt(counterStr, 10, 64)
		if err != nil {
			return nil, fmt.Errorf(
				"autonumber seed %s.%s: row Id=%q counter %q overflows int64: %w",
				objectName, fieldName, rowID.String, counterStr, err,
			)
		}
		// Carve the counter substring out of the match to build the
		// scope_key prefix. For counter-last formats (our 4 cutover
		// seeds) this is just the substring before loc[2]; the
		// concatenation keeps the result correct for counter-middle
		// formats too.
		prefix := rawValue.String[loc[0]:loc[2]] + rawValue.String[loc[3]:loc[1]]
		if prev, ok := out[prefix]; !ok || counter > prev {
			out[prefix] = counter
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: row iteration: %w",
			objectName, fieldName, err)
	}
	rows.Close()

	if fallbackReason != "" {
		log.Printf("⚠️  autonumber seed: field %s.%s (field_id=%s): %s — falling back to iterate-all scan",
			objectName, fieldName, fieldID, fallbackReason)
		return scanLivePerPrefixCountersIterateAll(ctx, rawDB, objectName, fieldName, derivedRegex, fieldID)
	}

	// Legacy-tail handling mirrors scanLiveMaxCounter.
	if tolerateLegacySeedRows() {
		if err := logLegacyTolerant(ctx, rawDB, objectName, fieldName, fieldID, derivedRegex, sqlRegex); err != nil {
			return nil, err
		}
	} else {
		if err := abortOnFirstLegacy(ctx, rawDB, objectName, fieldName, fieldID, derivedRegex, sqlRegex); err != nil {
			return nil, err
		}
	}
	return out, nil
}

// scanLivePerPrefixCountersIterateAll is the pre-v2 iterate-every-row
// fallback for the per_prefix scan, retained behind the fast-path
// guards (matchCountCeiling + engine parity). Same semantics as the
// original scanLivePerPrefixCounters body.
func scanLivePerPrefixCountersIterateAll(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	derivedRegex *regexp.Regexp,
	fieldID string,
) (map[string]int64, error) {
	query := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` WHERE `%s` IS NOT NULL AND `%s` != ''",
		fieldName, objectName, fieldName, fieldName,
	)
	rows, err := rawDB.QueryContext(ctx, query)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: scan live data: %w",
			objectName, fieldName, err)
	}
	defer rows.Close()

	out := make(map[string]int64)
	tolerateLegacy := tolerateLegacySeedRows()
	sampler := newLegacyRowSampler(objectName, fieldName, fieldID, derivedRegex)
	defer sampler.flush()
	for rows.Next() {
		var rowID, rawValue sql.NullString
		if err := rows.Scan(&rowID, &rawValue); err != nil {
			return nil, fmt.Errorf("autonumber seed %s.%s: scan row: %w",
				objectName, fieldName, err)
		}
		if !rawValue.Valid || rawValue.String == "" {
			continue
		}
		loc := derivedRegex.FindStringSubmatchIndex(rawValue.String)
		if loc == nil {
			if tolerateLegacy {
				sampler.observe(rowID.String, rawValue.String)
				continue
			}
			return nil, fmt.Errorf(
				"autonumber seed: field %s.%s (field_id=%s): row Id=%q "+
					"has value %q which does not match derived regex %q. "+
					"Fix the row (e.g. SET %s = NULL) and re-run",
				objectName, fieldName, fieldID, rowID.String, rawValue.String,
				derivedRegex.String(), fieldName,
			)
		}
		if len(loc) < 4 {
			return nil, fmt.Errorf(
				"autonumber seed %s.%s: derived regex %q matched but produced no counter capture on value %q",
				objectName, fieldName, derivedRegex.String(), rawValue.String,
			)
		}
		counterStr := rawValue.String[loc[2]:loc[3]]
		counter, err := strconv.ParseInt(counterStr, 10, 64)
		if err != nil {
			return nil, fmt.Errorf(
				"autonumber seed %s.%s: row Id=%q counter %q overflows int64: %w",
				objectName, fieldName, rowID.String, counterStr, err,
			)
		}
		// Carve the counter substring out of the match to build the
		// scope_key prefix. For counter-last formats (our 4 cutover
		// seeds) this is just the substring before loc[2]; the
		// concatenation keeps the result correct for counter-middle
		// formats too.
		prefix := rawValue.String[loc[0]:loc[2]] + rawValue.String[loc[3]:loc[1]]
		if prev, ok := out[prefix]; !ok || counter > prev {
			out[prefix] = counter
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: row iteration: %w",
			objectName, fieldName, err)
	}
	return out, nil
}

// physicalColumnExists reports whether INFORMATION_SCHEMA records a
// column named <column> in table <table> of the current database.
func physicalColumnExists(
	ctx context.Context,
	rawDB *sql.DB,
	table, column string,
) (bool, error) {
	var n int
	err := rawDB.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS "+
			"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?",
		table, column,
	).Scan(&n)
	if err != nil {
		return false, err
	}
	return n > 0, nil
}