// Package bootstrap — autonumber_seed.go implements Lane C2 of the
// Autonumber Engine v2 cutover (Wave 3). It seeds the
// `_System_AutoNumberCounter` table from:
//
//  1. live target-column data (max-parsed counter per field), and
//  2. legacy counter sources (`_System_AutoNumber.CurrentNumber` +
//     `_System_CompoundAutoNumber.CurrentNumber`), and
//  3. each field's `AutoNumberStart` floor.
//
// Scope-aware GREATEST(live_max+1, legacy+1, start) per PRD §8.2 step
// 3.e + Round-2 fixpack F8.
//
// The runner is FAIL-CLOSED on unparseable live rows (PRD §8.2
// step 3.c, C#2 + Pass B P0-3) — `log-and-skip` is explicitly rejected.
//
// IMPORTANT — fixpack F1 compliance:
// SeedAutoNumberCounters does NOT acquire a new GET_LOCK. It runs on a
// connection that the caller has already serialized under
// `nexuscrm_bootstrap_mutation` (the outer bootstrap mutex). Adding a
// nested `autonumber_cutover` lock races against the outer lock's
// timeout window — see Round-2 review Pass B #1.
//
// The runner is NOT yet wired into the bootstrap apply pipeline (that
// is Wave 4 Lane D's job). This file defines the exported function and
// a companion verify gate (see autonumber_verify.go).
package bootstrap

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

	"github.com/nexuscrm/backend/internal/features/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
}

// AutoNumberFormatSeed mirrors packs.AutoNumberFormatSeed but lives in
// the bootstrap package so SeedAutoNumberCounters does not create a
// packs → bootstrap → packs import cycle (packs already depends on
// bootstrap for ReconcileTableColumns / InitializeFieldMappings).
//
// Lane D's wire-up code is expected to construct this slice from
// packs.AutoNumberFormatSeeds via a one-liner copy; tests in this
// package build it directly.
type AutoNumberFormatSeed struct {
	ObjectName   string
	FieldApiName string
	Format       string
	Scope        string
	Start        int64
	LegacySource string
}

// autonumberSeedBatchSize bounds how many UPSERTs are flushed per
// transaction. Smaller batches keep the outer bootstrap lock
// progressing; 200 matches the per_day ceiling referenced in PRD §8.2
// step 3.f pagination note.
const autonumberSeedBatchSize = 200

// 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
}

// LegacySource constants — short enum values used by scopeSourceMap
// (per Wave 3 Lane C2 task spec). The generated seed structs in
// packs/platform_config_autonumber_formats.go use the longer table-name
// form (`_System_AutoNumber`, `_System_CompoundAutoNumber`); callers
// are free to use either convention so long as both ends agree. This
// file accepts both.
const (
	LegacySourceAutoNumber = "auto_number"
	LegacySourceCompound   = "compound"
	LegacySourceNone       = "none"

	legacyTableAutoNumber = "_System_AutoNumber"
	legacyTableCompound   = "_System_CompoundAutoNumber"
)

// SeedAutoNumberCounters executes Lane C2's migration routine.
//
// For each seed in seedMap:
//  1. Resolve the `_System_Field.Id` (skip silently if the field
//     doesn't exist — matches applyAutoNumberFormats behavior).
//  2. Parse the format via the engine's canonical parser (same code
//     path used at runtime and in `cmd/setup verify`).
//  3. Derive a counter regex from the format.
//  4. Scan the live object table. Any row whose value is non-NULL and
//     non-empty yet does not match the derived regex aborts the whole
//     seed (fail-closed).
//  5. Compute the scope-aware GREATEST:
//     live_max_counter + 1
//     legacy_counter   + 1   (from scopeSourceMap's declared source)
//     start
//     and UPSERT one row into `_System_AutoNumberCounter` with
//     `ScopeKey=''` (scope=global) or per-prefix (scope=per_prefix)
//     and `NextValue=GREATEST(...)`.
//
// Idempotency: each UPSERT uses
// `ON DUPLICATE KEY UPDATE NextValue = GREATEST(NextValue, VALUES(NextValue))`
// so re-running against the same data produces identical row counts
// and non-decreasing NextValue. PRD §7.1 / §8.4.
//
// CUTOVER PREREQUISITE — quiesce writers:
// scope=per_prefix fields derive their counter rows by scanning live
// data and grouping by prefix. The scan is NOT serialized against
// concurrent INSERTs: if a writer inserts a new record into a prefix
// bucket AFTER the scan has read that bucket, the UPSERT writes
// NextValue = stale_max + 1, which collides with the just-inserted
// row the first time post-cutover the engine reserves from that
// bucket. Same failure mode if a brand-new prefix bucket appears
// between scan and counter-row insert: seed emits no row, runtime
// lazy-init creates a fresh counter at NextValue=1, colliding with
// the just-inserted row(s).
//
// Operators MUST stop all writers before running
// SeedAutoNumberCounters (standard cutover procedure: stop backend →
// run setup → start backend). Drift AFTER the seed window is caught
// by runV7 in VerifyAutoNumberInvariants, but prevention is the
// operator's responsibility.
//
// scopeSourceMap keys are `"<ObjectName>.<FieldApiName>"`; values are
// one of the LegacySource* constants OR the literal table-name form
// produced by the generator (`_System_AutoNumber`,
// `_System_CompoundAutoNumber`). Missing keys default to
// LegacySourceNone (equivalent to `legacy=0`, i.e. seed uses
// GREATEST(live_max+1, start)).
func SeedAutoNumberCounters(
	ctx context.Context,
	rawDB *sql.DB,
	seedMap []AutoNumberFormatSeed,
	scopeSourceMap map[string]string,
) error {
	if rawDB == nil {
		return errors.New("autonumber seed: rawDB is nil")
	}

	// P1-R3-RT-ordering assertion (PRD §8.2 preconditions + §8.5 +
	// Wave 3 Lane C2 task spec step 3): if ANY seeded format uses a
	// {RecordType.*} token, _System_RecordType.RecordTypeCode must
	// already be seeded (Lane C1). Checked here so a wrong
	// bootstrap-ordering diff fails fast with a structured message
	// instead of producing silently-broken counters.
	if err := assertRecordTypeCodeSeeded(ctx, rawDB, seedMap); err != nil {
		return err
	}

	// Accumulate UPSERT rows in memory, flush in batches of
	// autonumberSeedBatchSize to keep the outer bootstrap lock
	// progressing. F1: no new GET_LOCK — we reuse the outer
	// `nexuscrm_bootstrap_mutation` lock already held by our caller.
	var pending []counterUpsertRow

	for _, seed := range seedMap {
		rows, err := computeSeedRowsForField(ctx, rawDB, seed, scopeSourceMap)
		if err != nil {
			return err
		}
		pending = append(pending, rows...)
		if len(pending) >= autonumberSeedBatchSize {
			if err := flushCounterUpserts(ctx, rawDB, pending); err != nil {
				return err
			}
			pending = pending[:0]
		}
	}
	if len(pending) > 0 {
		if err := flushCounterUpserts(ctx, rawDB, pending); err != nil {
			return err
		}
	}

	return nil
}

// counterUpsertRow is the flat representation of one
// `_System_AutoNumberCounter` row to insert.
type counterUpsertRow struct {
	ID        string
	FieldID   string
	ScopeKey  string
	NextValue int64
}

// computeSeedRowsForField handles one seed entry end-to-end: parse,
// derive regex, scan live data, read legacy counter, compute
// GREATEST, and return the UPSERT rows. Returns an empty slice if the
// field is missing from the DB (matches applyAutoNumberFormats).
//
// Scope handling:
//   - `global`     → one counter row per field at ScopeKey=''.
//     NextValue = GREATEST(live_max+1, legacy+1, seed.Start).
//   - `per_prefix` → N counter rows per field, one per distinct prefix
//     observed in live data. NextValue_i = live_max(prefix_i)+1.
//     This absorbs pre-cutover records produced by legacy Apex
//     triggers so the first post-cutover insert into an existing
//     prefix bucket does NOT collide with an existing rendered Name
//     (Codex adversarial round-2 finding).
//   - `per_day`    → lazy-init only (PRD §8.2 step 3.f); pre-seeding
//     would require enumerating future UTC-date buckets and is not
//     meaningful.
func computeSeedRowsForField(
	ctx context.Context,
	rawDB *sql.DB,
	seed AutoNumberFormatSeed,
	scopeSourceMap map[string]string,
) ([]counterUpsertRow, error) {
	switch seed.Scope {
	case autonumber.ScopeGlobal:
		return computeSeedRowsForGlobalField(ctx, rawDB, seed, scopeSourceMap)
	case autonumber.ScopePerPrefix:
		return computeSeedRowsForPerPrefixField(ctx, rawDB, seed)
	case autonumber.ScopePerDay:
		// per_day is lazy-init only; no pre-seeding (PRD §8.2 step 3.f).
		return nil, nil
	default:
		return nil, fmt.Errorf("autonumber seed %s.%s: unknown scope %q",
			seed.ObjectName, seed.FieldApiName, seed.Scope)
	}
}

// computeSeedRowsForGlobalField is the original single-counter-row
// global path; extracted from computeSeedRowsForField so the scope
// dispatcher above reads cleanly.
func computeSeedRowsForGlobalField(
	ctx context.Context,
	rawDB *sql.DB,
	seed AutoNumberFormatSeed,
	scopeSourceMap map[string]string,
) ([]counterUpsertRow, error) {

	// 1. Resolve field ID.
	fieldID, ok, err := resolveFieldID(ctx, rawDB, seed.ObjectName, seed.FieldApiName)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: resolve field: %w",
			seed.ObjectName, seed.FieldApiName, err)
	}
	if !ok {
		// Field missing — applyAutoNumberFormats silently skips
		// too. Seed follows the same convention so fresh DBs
		// without the full object graph don't fail.
		return nil, nil
	}

	// 2. Parse the format via the engine's canonicalizer.
	spec, _, err := autonumber.Parse(seed.Format)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: parse format %q: %w",
			seed.ObjectName, seed.FieldApiName, seed.Format, err)
	}

	// 3. Derive counter regex from the format.
	derivedRegex, err := deriveCounterRegex(spec)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: derive regex from %q: %w",
			seed.ObjectName, seed.FieldApiName, seed.Format, err)
	}

	// 4. Scan live data; compute live_max_counter. Abort on any
	// unparseable non-NULL / non-empty row (FAIL-CLOSED — PRD §8.2
	// step 3.c + C#2).
	liveMax, err := scanLiveMaxCounter(
		ctx, rawDB,
		seed.ObjectName, seed.FieldApiName,
		derivedRegex, fieldID,
	)
	if err != nil {
		return nil, err
	}

	// 5. Read legacy counter from the declared scope-source.
	legacyCurrent, err := readLegacyCurrent(
		ctx, rawDB,
		seed.ObjectName, seed.FieldApiName,
		scopeSourceMap,
	)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: read legacy counter: %w",
			seed.ObjectName, seed.FieldApiName, err)
	}

	// 6. Scope-aware GREATEST (F8).
	start := seed.Start
	if start < 1 {
		start = 1
	}
	next := start
	if liveMax+1 > next {
		next = liveMax + 1
	}
	if legacyCurrent+1 > next {
		next = legacyCurrent + 1
	}

	return []counterUpsertRow{{
		ID:        counterRowID(fieldID, ""),
		FieldID:   fieldID,
		ScopeKey:  "",
		NextValue: next,
	}}, nil
}

// computeSeedRowsForPerPrefixField emits one counter row per distinct
// prefix bucket observed in live data for a per_prefix-scoped field.
// This absorbs legacy Apex-generated records (e.g. "JP-QT-260418-07")
// so the first post-cutover insert into the same bucket continues the
// sequence rather than restarting at 01.
//
// Returns an empty slice when:
//   - the field is missing from _System_Field (fresh DB),
//   - the physical column does not exist yet,
//   - the column has no rows (no prefix buckets to seed).
//
// Lazy-init still handles the case where a NEW prefix shows up post-
// cutover: the engine's reserveOneAttempt sees ErrNoRows, UPSERTs a
// fresh counter at NextValue=1, then retries. That's correct for a
// genuinely-new bucket because there is no prior record to collide
// with.
//
// This function does NOT read legacy counter tables
// (_System_AutoNumber / _System_CompoundAutoNumber): those tables
// store per-object totals, not per-prefix bucket breakdowns, so
// their values are not informative for per_prefix. Live data is the
// authoritative source.
func computeSeedRowsForPerPrefixField(
	ctx context.Context,
	rawDB *sql.DB,
	seed AutoNumberFormatSeed,
) ([]counterUpsertRow, error) {
	fieldID, ok, err := resolveFieldID(ctx, rawDB, seed.ObjectName, seed.FieldApiName)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: resolve field: %w",
			seed.ObjectName, seed.FieldApiName, err)
	}
	if !ok {
		return nil, nil
	}

	spec, _, err := autonumber.Parse(seed.Format)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: parse format %q: %w",
			seed.ObjectName, seed.FieldApiName, seed.Format, err)
	}

	derivedRegex, err := deriveCounterRegex(spec)
	if err != nil {
		return nil, fmt.Errorf("autonumber seed %s.%s: derive regex from %q: %w",
			seed.ObjectName, seed.FieldApiName, seed.Format, err)
	}

	perPrefixMax, err := scanLivePerPrefixCounters(
		ctx, rawDB,
		seed.ObjectName, seed.FieldApiName,
		derivedRegex, fieldID,
	)
	if err != nil {
		return nil, err
	}
	if len(perPrefixMax) == 0 {
		return nil, nil
	}

	start := seed.Start
	if start < 1 {
		start = 1
	}
	out := make([]counterUpsertRow, 0, len(perPrefixMax))
	for prefix, maxCounter := range perPrefixMax {
		next := maxCounter + 1
		if start > next {
			next = start
		}
		out = append(out, counterUpsertRow{
			ID:        counterRowID(fieldID, prefix),
			FieldID:   fieldID,
			ScopeKey:  prefix,
			NextValue: next,
		})
	}
	return out, nil
}

// resolveFieldID looks up `_System_Field.Id` by (ObjectApiName,
// FieldApiName). Second return is false when the field does not exist
// (no error — just absent).
func resolveFieldID(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
) (string, bool, error) {
	var id string
	err := rawDB.QueryRowContext(ctx,
		"SELECT f.Id FROM `_System_Field` f "+
			"JOIN `_System_Object` o ON o.Id = f.ObjectId "+
			"WHERE o.ApiName = ? AND f.ApiName = ?",
		objectName, fieldName,
	).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}
	return id, true, 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
}

// readLegacyCurrent reads the legacy counter for the given
// (objectName, fieldName) from the source declared by scopeSourceMap.
// Missing row / missing table / none-source → returns 0 without
// error.
func readLegacyCurrent(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	scopeSourceMap map[string]string,
) (int64, error) {
	key := objectName + "." + fieldName
	source := scopeSourceMap[key]
	switch source {
	case "", LegacySourceNone:
		return 0, nil
	case LegacySourceAutoNumber, legacyTableAutoNumber:
		return readLegacyAutoNumberCurrent(ctx, rawDB, objectName, fieldName)
	case LegacySourceCompound, legacyTableCompound:
		return readLegacyCompoundMax(ctx, rawDB, objectName)
	default:
		return 0, fmt.Errorf("autonumber seed %s.%s: unknown legacy source %q",
			objectName, fieldName, source)
	}
}

// readLegacyAutoNumberCurrent reads `_System_AutoNumber.CurrentNumber`
// by (ObjectApiName, FieldApiName). Returns 0 if the legacy table is
// missing or the row is absent.
func readLegacyAutoNumberCurrent(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
) (int64, error) {
	ok, err := tableExistsRaw(ctx, rawDB, legacyTableAutoNumber)
	if err != nil {
		return 0, err
	}
	if !ok {
		return 0, nil
	}
	var current int64
	err = rawDB.QueryRowContext(ctx,
		"SELECT `CurrentNumber` FROM `_System_AutoNumber` "+
			"WHERE `ObjectApiName` = ? AND `FieldApiName` = ?",
		objectName, fieldName,
	).Scan(&current)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	return current, nil
}

// readLegacyCompoundMax reads MAX(`_System_CompoundAutoNumber.CurrentNumber`)
// across all prefix rows for the given object. Compound counters
// bucket by prefix (e.g. "CN-CSA-260418" → one row per daily prefix);
// the seed takes the max across prefixes for the single global
// counter row. Matches PRD §8.2 step 3.d.
func readLegacyCompoundMax(
	ctx context.Context,
	rawDB *sql.DB,
	objectName string,
) (int64, error) {
	ok, err := tableExistsRaw(ctx, rawDB, legacyTableCompound)
	if err != nil {
		return 0, err
	}
	if !ok {
		return 0, nil
	}
	var maxVal sql.NullInt64
	err = rawDB.QueryRowContext(ctx,
		"SELECT MAX(`CurrentNumber`) FROM `_System_CompoundAutoNumber` "+
			"WHERE `ObjectApiName` = ?",
		objectName,
	).Scan(&maxVal)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	if !maxVal.Valid {
		return 0, nil
	}
	return maxVal.Int64, nil
}

// tableExistsRaw is a local helper (separate from the test-only
// tableExists in schema_autonumber_v2_test.go).
func tableExistsRaw(ctx context.Context, rawDB *sql.DB, table string) (bool, error) {
	var n int
	err := rawDB.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES "+
			"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?",
		table,
	).Scan(&n)
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

// flushCounterUpserts executes a batch of INSERT ... ON DUPLICATE KEY
// UPDATE statements against `_System_AutoNumberCounter`. Each batch
// runs in its own transaction so partial progress is committed between
// batches (re-runs use the GREATEST idempotency guarantee to converge).
// PRD §8.2 step 3.f + §8.4.
func flushCounterUpserts(
	ctx context.Context,
	rawDB *sql.DB,
	rows []counterUpsertRow,
) error {
	if len(rows) == 0 {
		return nil
	}
	tx, err := rawDB.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("autonumber seed flush: begin tx: %w", err)
	}
	now := time.Now().UTC()
	const stmt = "INSERT INTO `_System_AutoNumberCounter` " +
		"(`Id`, `FieldId`, `ScopeKey`, `NextValue`, `CreatedDate`, `LastModifiedDate`) " +
		"VALUES (?, ?, ?, ?, ?, ?) " +
		"ON DUPLICATE KEY UPDATE " +
		"`NextValue` = GREATEST(`NextValue`, VALUES(`NextValue`)), " +
		"`LastModifiedDate` = VALUES(`LastModifiedDate`)"
	for _, r := range rows {
		if _, err := tx.ExecContext(ctx, stmt,
			r.ID, r.FieldID, r.ScopeKey, r.NextValue, now, now,
		); err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("autonumber seed flush: upsert field_id=%s scope=%q: %w",
				r.FieldID, r.ScopeKey, err)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("autonumber seed flush: commit: %w", err)
	}
	return nil
}

// counterRowID mirrors generateCounterRowID in the autonumber package
// (reserve.go:604). Reproduced here to avoid exporting the engine
// symbol or creating a circular import between bootstrap and
// features/automation/autonumber.
//
// Format: "counter::<fieldID>::<scopeKey>".
func counterRowID(fieldID, scopeKey string) string {
	return "counter::" + fieldID + "::" + scopeKey
}

// assertRecordTypeCodeSeeded verifies the RT-ordering invariant
// referenced in PRD §8.2 preconditions + Wave 3 task spec:
// if any seeded format contains a {RecordType.*} token, at least one
// `_System_RecordType.RecordTypeCode` must be populated. Detects the
// wrong-order bootstrap apply that would otherwise produce
// silently-broken counters (rendered values would have empty
// `{RecordType.RecordTypeCode}` slots).
func assertRecordTypeCodeSeeded(
	ctx context.Context,
	rawDB *sql.DB,
	seedMap []AutoNumberFormatSeed,
) error {
	needsRT := false
	for _, seed := range seedMap {
		if strings.Contains(seed.Format, "{RecordType.") {
			needsRT = true
			break
		}
	}
	if !needsRT {
		return nil
	}
	// _System_RecordType may itself not exist yet on a very fresh
	// DB; treat that as the RT-ordering violation.
	ok, err := tableExistsRaw(ctx, rawDB, "_System_RecordType")
	if err != nil {
		return fmt.Errorf("autonumber seed: probe _System_RecordType: %w", err)
	}
	if !ok {
		return errors.New("autonumber seed: RecordTypeCode not seeded before autonumber counter seed " +
			"(_System_RecordType table is absent; bootstrap ordering violation)")
	}
	var n int
	err = rawDB.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM `_System_RecordType` "+
			"WHERE `record_type_code` IS NOT NULL AND `record_type_code` != ''",
	).Scan(&n)
	if err != nil {
		return fmt.Errorf("autonumber seed: count RecordTypeCode: %w", err)
	}
	if n == 0 {
		return errors.New("autonumber seed: RecordTypeCode not seeded before autonumber counter seed " +
			"(_System_RecordType has zero rows with RecordTypeCode IS NOT NULL AND != '')")
	}
	return nil
}

// BuildScopeSourceMapFromSeeds is a convenience translator for
// callers that want to derive the scopeSourceMap directly from the
// generated `AutoNumberFormatSeeds` — it converts the long table-name
// form into the short enum. Tests use this to avoid duplicating the
// 132-row map.
func BuildScopeSourceMapFromSeeds(seeds []AutoNumberFormatSeed) map[string]string {
	out := make(map[string]string, len(seeds))
	for _, s := range seeds {
		key := s.ObjectName + "." + s.FieldApiName
		switch s.LegacySource {
		case legacyTableAutoNumber:
			out[key] = LegacySourceAutoNumber
		case legacyTableCompound:
			out[key] = LegacySourceCompound
		case "", LegacySourceNone:
			out[key] = LegacySourceNone
		default:
			// Unknown — preserve raw value; SeedAutoNumberCounters
			// will surface it as an error at runtime.
			out[key] = s.LegacySource
		}
	}
	return out
}