// Package bootstrap — autonumber_verify.go implements the V1-V6
// verify gate described in PRD §5.3.3. Invoked by `cmd/setup verify`
// AFTER SeedAutoNumberCounters completes. Non-zero exit blocks deploy.
//
// The six invariants (refreshed for v3.1 + Round-2 fixpack F7):
//
//	V1 — Format parseable:   every _System_Field with AutoNumberFormat
//	                         non-null must pass autonumber.Parse.
//	V2 — Canonical hash:     if AutoNumberTokenRefs + AutoNumberCanonicalHash
//	                         are both non-empty, the hash must equal
//	                         sha256(canonicalize(tokens)).
//	V3 — Counter row exists: each scope=global field has a counter row
//	                         with ScopeKey=''.
//	V4 — NextValue ≥ 1:      no counter row has NextValue < 1.
//	V5 — Live-max absorbed:  counter NextValue > max(extract-counter from
//	                         live column). Protects against shipping
//	                         with the seed behind the live data (the
//	                         write-hole regressor in PRD §5.3.3 V1).
//	V6 — RecordTypeCode:     per F7 dual-check, every RecordType row is
//	                         IS NOT NULL AND != '' if any format uses
//	                         `{RecordType.*}`.
//
// Errors are accumulated (not fail-fast) so one deploy iteration can
// see every problem at once. The final return value wraps them with
// fmt.Errorf("...: %w", errors.Join(...)).
package bootstrap

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"
	"sync"

	"github.com/nexuscrm/backend/internal/features/automation/autonumber"
)

// ErrAutoNumberVerify is the sentinel wrapping all V1-V6 violations.
// Callers classify via errors.Is(err, ErrAutoNumberVerify).
var ErrAutoNumberVerify = errors.New("autonumber verify: invariant violated")

// VerifyAutoNumberMetadataInvariants runs only the gates that are safe
// on a hot pod-readiness path: V1 (parse), V2 (canonical hash), V3
// (counter row exists for scope=global), V4 (NextValue ≥ 1), V6
// (RecordTypeCode), V8 (minimum cardinality floor).
//
// V5 and V7 are deliberately SKIPPED here — those scan the live object
// tables (Quote, Order, Contract, …) to detect seed-vs-writer races,
// which (a) makes latency proportional to live table size on every
// restart and (b) inherits a writer-quiescence precondition that a
// running cluster cannot guarantee. Running them on pod boot creates a
// `verify → apply → verify` oscillation during rolling deploys.
// Those gates remain in VerifyAutoNumberInvariants, which runs once
// from PlatformConfigPack.Apply under the controlplane apply lock with
// writers quiesced.
//
// Codex round-9 V2 HIGH: the pod-readiness gate must only catch the
// empty-metadata silent-boot class (V8), parser/hash drift (V1, V2),
// and missing counter rows that make the engine crash at first insert
// (V3, V4, V6). Anything that depends on live-data consistency belongs
// on the setup path, not the hot path.
func VerifyAutoNumberMetadataInvariants(ctx context.Context, rawDB *sql.DB) error {
	if rawDB == nil {
		return errors.New("autonumber verify: rawDB is nil")
	}

	var violations []string
	add := func(format string, args ...any) {
		violations = append(violations, fmt.Sprintf(format, args...))
	}

	fields, err := loadAutonumberFields(ctx, rawDB)
	if err != nil {
		return fmt.Errorf("autonumber verify: load fields: %w", err)
	}

	runV1(fields, add)
	runV2(fields, add)

	counters, err := loadCounterMap(ctx, rawDB)
	if err != nil {
		return fmt.Errorf("autonumber verify: load counters: %w", err)
	}
	runV3(fields, counters, add)
	runV4(counters, add)

	if err := runV6(ctx, rawDB, fields, add); err != nil {
		return fmt.Errorf("autonumber verify V6: %w", err)
	}

	if err := runV8(ctx, rawDB, fields, add); err != nil {
		return fmt.Errorf("autonumber verify V8: %w", err)
	}

	if len(violations) == 0 {
		return nil
	}
	return fmt.Errorf("%w: %d violation(s): %s",
		ErrAutoNumberVerify, len(violations), strings.Join(violations, "; "))
}

// VerifyAutoNumberInvariants runs all eight gate checks, accumulating
// every violation before returning. Returns nil if all pass.
//
// Intended for use from `cmd/setup verify` and
// PlatformConfigPack.Apply post-SeedAutoNumberCounters where writers
// are quiesced under the controlplane apply lock. NOT safe on the
// pod-readiness path — use VerifyAutoNumberMetadataInvariants there.
//
// V7 covers per_prefix coverage (Codex adversarial round-3 finding):
// V3 and V5 only validate scope=global fields, so a per_prefix field
// whose seed missed a bucket (mid-seed race with a concurrent writer,
// or a bucket introduced between seed and verify) would ship
// undetected. V7 re-scans live data for every per_prefix field and
// asserts each observed prefix has a counter row with NextValue >
// live_max.
func VerifyAutoNumberInvariants(ctx context.Context, rawDB *sql.DB) error {
	if rawDB == nil {
		return errors.New("autonumber verify: rawDB is nil")
	}

	var violations []string
	add := func(format string, args ...any) {
		violations = append(violations, fmt.Sprintf(format, args...))
	}

	// Pull all autonumber-configured fields once.
	fields, err := loadAutonumberFields(ctx, rawDB)
	if err != nil {
		return fmt.Errorf("autonumber verify: load fields: %w", err)
	}

	// V1 + V2 — per-field parse + canonical hash checks.
	v1Pass := runV1(fields, add)
	runV2(fields, add)

	// V3 — counter row exists for each scope=global field.
	counters, err := loadCounterMap(ctx, rawDB)
	if err != nil {
		return fmt.Errorf("autonumber verify: load counters: %w", err)
	}
	runV3(fields, counters, add)

	// V4 — monotone NextValue.
	runV4(counters, add)

	// V5 — counters cover live_max+1 for scope=global. Only run if
	// V1 passed (otherwise we can't derive the regex reliably).
	if v1Pass {
		if err := runV5(ctx, rawDB, fields, counters, add); err != nil {
			return fmt.Errorf("autonumber verify V5: %w", err)
		}
	}

	// V6 — RecordTypeCode dual-check (F7).
	if err := runV6(ctx, rawDB, fields, add); err != nil {
		return fmt.Errorf("autonumber verify V6: %w", err)
	}

	// V7 — per_prefix coverage. Ensures every live prefix bucket has
	// an associated counter row with NextValue > live_max. Only run
	// if V1 passed (needs a parseable regex per field).
	if v1Pass {
		if err := runV7(ctx, rawDB, fields, counters, add); err != nil {
			return fmt.Errorf("autonumber verify V7: %w", err)
		}
	}

	// V8 — declared-vs-reachable gate (Codex round-8 V2 CRITICAL +
	// issue #1632). Without this check, verify silently PASSes on
	// an empty `_System_Field.auto_number_format` state because
	// V1-V7 all iterate the returned field set. Deploying a cutover
	// binary against a DB where `cmd/setup apply` never ran
	// (ordering mistake, forgotten step, crashed applier) would
	// boot a healthy-looking engine whose `LookupByObject` always
	// returns empty → `assignAutoNumbers` silent no-op → records
	// created with NULL QuoteNumber / OrderNumber / ContractNumber /
	// Name. V8 fails closed so the operator is forced to rerun the
	// applier before pods become ready.
	if err := runV8(ctx, rawDB, fields, add); err != nil {
		return fmt.Errorf("autonumber verify V8: %w", err)
	}

	if len(violations) == 0 {
		return nil
	}
	return fmt.Errorf("%w: %d violation(s): %s",
		ErrAutoNumberVerify, len(violations), strings.Join(violations, "; "))
}

// MinExpectedAutoNumberFieldsV8 is the legacy fixed-floor fallback for
// the V8 gate. Used only when no declared-seed list has been registered
// via RegisterAutoNumberDeclaredSeeds (e.g. unit tests that do not
// import controlplane/packs).
//
// Production paths always register the generated seed list, so V8 runs
// the stricter declared-vs-reachable check and this constant does not
// apply. See runV8 for the primary semantics.
const MinExpectedAutoNumberFieldsV8 = 100

// AutoNumberDeclaredSeedKey identifies one declared autonumber seed by
// its target (ObjectName, FieldApiName) pair. Registered via
// RegisterAutoNumberDeclaredSeeds so V8 can compute an accurate
// declared-vs-reachable expected count without creating a
// bootstrap → controlplane/packs import cycle.
type AutoNumberDeclaredSeedKey struct {
	ObjectName   string
	FieldApiName string
}

var (
	declaredSeedKeysMu sync.RWMutex
	declaredSeedKeys   []AutoNumberDeclaredSeedKey
)

// RegisterAutoNumberDeclaredSeeds publishes the generated seed list so
// V8 can compute its declared-vs-reachable expected count. Called by
// controlplane/packs at init-time (see packs/bootstrap_shim.go).
//
// Idempotent: repeated calls overwrite the registry. Safe for
// concurrent reads via runV8 after registration completes.
func RegisterAutoNumberDeclaredSeeds(keys []AutoNumberDeclaredSeedKey) {
	declaredSeedKeysMu.Lock()
	defer declaredSeedKeysMu.Unlock()
	// Defensive copy so callers can mutate their slice without
	// affecting the registry.
	cp := make([]AutoNumberDeclaredSeedKey, len(keys))
	copy(cp, keys)
	declaredSeedKeys = cp
}

// snapshotDeclaredSeedKeys returns a defensive copy of the currently-
// registered seed keys. Returns nil when the registry is empty; callers
// MUST check len() before iterating (range over nil is safe, but nil
// must not be passed where a non-nil slice is required).
func snapshotDeclaredSeedKeys() []AutoNumberDeclaredSeedKey {
	declaredSeedKeysMu.RLock()
	defer declaredSeedKeysMu.RUnlock()
	if len(declaredSeedKeys) == 0 {
		return nil
	}
	cp := make([]AutoNumberDeclaredSeedKey, len(declaredSeedKeys))
	copy(cp, declaredSeedKeys)
	return cp
}

// runV8 enforces the declared-vs-reachable invariant: every declared
// autonumber seed whose target (Object, Field) row exists in
// `_System_Field` must have a non-empty `auto_number_format`. Mismatch
// indicates `applyAutoNumberFormats` did not finish seeding every
// reachable target — a silent-no-op hazard per the Codex round-8 V2
// CRITICAL fix, now refined (issue #1632) so partial-managed-package
// orgs do not false-positive on the previous fixed `>= 100` threshold.
//
// Computation:
//   - reachable = set of declared seeds whose (ObjectName, FieldApiName)
//     pair is present in `_System_Field` joined with `_System_Object`.
//     Applier silently skips unmatched seeds (fields the org does not
//     have), so "reachable" captures exactly the set the writer should
//     have touched.
//   - written   = subset of reachable that ALSO appears in `fields`
//     (rows with non-empty auto_number_format). Intersecting on
//     (Object, Field) rather than raw len(fields) is necessary so
//     unrelated admin-created autonumber fields do not inflate the
//     count and mask a missing declared target.
//   - Fail if written < reachable, naming the specific missing pairs
//     so operators can identify the failing seed.
//
// Fallbacks (preserving Codex round-8 V2 CRITICAL coverage when the
// declared-seed list is unavailable):
//   - No registered seeds (unit tests that do not import
//     controlplane/packs) → use fixed floor
//     MinExpectedAutoNumberFieldsV8.
//   - Env override AUTONUMBER_V8_FLOOR=<n> → use n as floor (<n> == 0
//     disables the gate; legacy dev-sandbox escape hatch).
func runV8(
	ctx context.Context,
	rawDB *sql.DB,
	fields []autonumberField,
	add func(string, ...any),
) error {
	// Env override takes precedence — legacy dev-sandbox escape hatch.
	if override := os.Getenv("AUTONUMBER_V8_FLOOR"); override != "" {
		if v, err := strconv.Atoi(override); err == nil && v >= 0 {
			if v == 0 {
				return nil
			}
			if len(fields) < v {
				add("V8: only %d `_System_Field` rows have auto_number_format set; "+
					"expected >= %d (AUTONUMBER_V8_FLOOR override). "+
					"`applyAutoNumberFormats` likely did not run on this DB — "+
					"rerun the applier before serving traffic or live inserts "+
					"will produce NULL identifiers.",
					len(fields), v)
			}
			return nil
		}
	}

	declared := snapshotDeclaredSeedKeys()
	if len(declared) == 0 {
		// Fallback: no declared-seed registry (typically unit tests).
		// Use the legacy fixed-floor check so the Codex round-8 V2
		// CRITICAL empty-cohort hazard is still caught.
		if len(fields) < MinExpectedAutoNumberFieldsV8 {
			add("V8: only %d `_System_Field` rows have auto_number_format set; "+
				"expected >= %d (fixed-floor fallback — declared seed list "+
				"not registered). `applyAutoNumberFormats` likely did not run "+
				"on this DB — rerun the applier before serving traffic or live "+
				"inserts will produce NULL identifiers.",
				len(fields), MinExpectedAutoNumberFieldsV8)
		}
		return nil
	}

	reachable, err := reachableDeclaredSeedKeys(ctx, rawDB, declared)
	if err != nil {
		return fmt.Errorf("compute reachable seeds: %w", err)
	}
	if len(reachable) == 0 {
		// No declared-seed target field exists on this DB. Legitimate
		// on a tenant that has none of the seed-targeted objects, or
		// a fresh DB where `cmd/setup apply` never ran. Fall back to
		// the fixed-floor check so the empty-DB-never-applied hazard
		// is still surfaced.
		if len(fields) < MinExpectedAutoNumberFieldsV8 {
			add("V8: only %d `_System_Field` rows have auto_number_format set "+
				"and zero declared seeds are reachable on this DB; expected "+
				">= %d (fixed-floor fallback). `applyAutoNumberFormats` "+
				"likely did not run — rerun the applier before serving "+
				"traffic or live inserts will produce NULL identifiers.",
				len(fields), MinExpectedAutoNumberFieldsV8)
		}
		return nil
	}

	// Intersect reachable declared seeds with actually-written fields.
	// Using the (Object, Field) pair (not raw len(fields)) is what
	// makes V8 resistant to custom / admin-created autonumber fields
	// inflating the count — Copilot review finding on PR #1641.
	writtenPairs := make(map[string]struct{}, len(fields))
	for _, f := range fields {
		writtenPairs[strings.ToLower(f.ObjectName)+"::"+strings.ToLower(f.FieldName)] = struct{}{}
	}
	var missing []string
	for _, key := range reachable {
		normalized := strings.ToLower(key.ObjectName) + "::" + strings.ToLower(key.FieldApiName)
		if _, ok := writtenPairs[normalized]; !ok {
			missing = append(missing, key.ObjectName+"."+key.FieldApiName)
		}
	}
	if len(missing) > 0 {
		// Cap the listed targets so the error message stays readable
		// even when the applier skipped a large cohort. Callers see
		// the first N pairs + a count; operators can rerun the
		// applier and resolve from its log.
		const maxListed = 5
		listed := missing
		suffix := ""
		if len(missing) > maxListed {
			listed = missing[:maxListed]
			suffix = fmt.Sprintf(" (and %d more)", len(missing)-maxListed)
		}
		add("V8: %d of %d reachable declared autonumber seeds are NOT "+
			"written in `_System_Field` (missing: %s%s). "+
			"`applyAutoNumberFormats` did not finish seeding every "+
			"reachable target — rerun the applier before serving traffic "+
			"or live inserts will produce NULL identifiers for these "+
			"fields.",
			len(missing), len(reachable), strings.Join(listed, ", "), suffix)
	}
	return nil
}

// reachableDeclaredSeedKeys returns the subset of declared seeds whose
// target (ObjectName, FieldApiName) pair exists as a row in
// `_System_Field` (joined with `_System_Object`). This is the exact
// set `applyAutoNumberFormats` is expected to have written, because
// its UPDATE silently no-ops on pairs that do not exist (the applier
// is intentionally tolerant of partial-managed-package orgs).
//
// Implementation note: pulls every existing (Object, Field) pair once
// and intersects in Go. Cheaper than a large parameterized IN clause
// (`_System_Field` has ~O(10^3) rows) and avoids per-driver placeholder
// limits.
func reachableDeclaredSeedKeys(
	ctx context.Context,
	rawDB *sql.DB,
	declared []AutoNumberDeclaredSeedKey,
) ([]AutoNumberDeclaredSeedKey, error) {
	if rawDB == nil {
		return nil, errors.New("rawDB is nil")
	}
	if len(declared) == 0 {
		return nil, nil
	}
	rows, err := rawDB.QueryContext(ctx,
		"SELECT o.ApiName, f.ApiName "+
			"FROM `_System_Field` f "+
			"JOIN `_System_Object` o ON o.Id = f.ObjectId")
	if err != nil {
		return nil, fmt.Errorf("query _System_Field pairs: %w", err)
	}
	defer rows.Close()

	// Key normalization: the applier's WHERE clauses compare ApiName
	// under the column's collation (TiDB default is typically CI on
	// utf8mb4 tenants). If prod ever ends up with a row whose casing
	// differs from the seed (e.g. legacy import wrote `account_id__c`
	// while the seed declares `Account_ID__c`), a case-sensitive Go
	// map lookup would miss and under-report `reachable`, causing V8
	// to fail-open. Lowercase both sides so the intersection matches
	// the applier's case-insensitive behavior.
	existing := make(map[string]struct{})
	for rows.Next() {
		var obj, field sql.NullString
		if err := rows.Scan(&obj, &field); err != nil {
			return nil, fmt.Errorf("scan pair: %w", err)
		}
		if !obj.Valid || !field.Valid {
			continue
		}
		existing[strings.ToLower(obj.String)+"::"+strings.ToLower(field.String)] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate pairs: %w", err)
	}

	var out []AutoNumberDeclaredSeedKey
	for _, s := range declared {
		key := strings.ToLower(s.ObjectName) + "::" + strings.ToLower(s.FieldApiName)
		if _, ok := existing[key]; ok {
			out = append(out, s)
		}
	}
	return out, nil
}

// autonumberField is the DB view of one _System_Field row with
// autonumber configured. Only fields relevant to V1-V6 are loaded.
type autonumberField struct {
	ID             string
	ObjectName     string
	FieldName      string
	Format         string
	Scope          string
	Start          int64
	TokenRefs      string // raw JSON; may be "" / NULL
	CanonicalHash  string // may be "" / NULL
}

// loadAutonumberFields fetches every _System_Field where
// AutoNumberFormat IS NOT NULL, joined with its object so we can
// name the live table in V5.
func loadAutonumberFields(ctx context.Context, rawDB *sql.DB) ([]autonumberField, error) {
	const q = "SELECT f.Id, o.ApiName, f.ApiName, " +
		"f.auto_number_format, f.auto_number_scope, f.auto_number_start, " +
		"COALESCE(f.auto_number_token_refs, ''), COALESCE(f.auto_number_canonical_hash, '') " +
		"FROM `_System_Field` f " +
		"JOIN `_System_Object` o ON o.Id = f.ObjectId " +
		"WHERE f.auto_number_format IS NOT NULL AND f.auto_number_format != ''"

	rows, err := rawDB.QueryContext(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []autonumberField
	for rows.Next() {
		var f autonumberField
		var scope sql.NullString
		var start sql.NullInt64
		if err := rows.Scan(
			&f.ID, &f.ObjectName, &f.FieldName,
			&f.Format, &scope, &start,
			&f.TokenRefs, &f.CanonicalHash,
		); err != nil {
			return nil, err
		}
		if scope.Valid {
			f.Scope = scope.String
		} else {
			f.Scope = autonumber.ScopeGlobal
		}
		if start.Valid {
			f.Start = start.Int64
		} else {
			f.Start = 1
		}
		out = append(out, f)
	}
	return out, rows.Err()
}

// counterRow is the DB view of one _System_AutoNumberCounter row used
// by V3, V4, V5.
type counterRow struct {
	FieldID   string
	ScopeKey  string
	NextValue int64
}

// loadCounterMap fetches all counter rows indexed by (FieldID, ScopeKey).
func loadCounterMap(ctx context.Context, rawDB *sql.DB) (map[string]counterRow, error) {
	// _System_AutoNumberCounter may not exist on a pre-cutover DB —
	// treat absent-table as "zero counters" so V3 surfaces the
	// missing rows rather than V-loading erroring out ambiguously.
	ok, err := tableExistsRaw(ctx, rawDB, "_System_AutoNumberCounter")
	if err != nil {
		return nil, err
	}
	if !ok {
		return map[string]counterRow{}, nil
	}
	rows, err := rawDB.QueryContext(ctx,
		"SELECT `FieldId`, `ScopeKey`, `NextValue` FROM `_System_AutoNumberCounter`",
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := make(map[string]counterRow)
	for rows.Next() {
		var r counterRow
		if err := rows.Scan(&r.FieldID, &r.ScopeKey, &r.NextValue); err != nil {
			return nil, err
		}
		out[r.FieldID+"::"+r.ScopeKey] = r
	}
	return out, rows.Err()
}

// runV1 parses every format and reports unparseable ones. Returns true
// if every format parsed successfully (V5 gates on this).
func runV1(fields []autonumberField, add func(string, ...any)) bool {
	allOK := true
	for _, f := range fields {
		if _, _, err := autonumber.Parse(f.Format); err != nil {
			allOK = false
			add("V1: field %s.%s (id=%s): format %q does not parse: %v",
				f.ObjectName, f.FieldName, f.ID, f.Format, err)
		}
	}
	return allOK
}

// runV2 recomputes sha256(canonicalize(tokens)) from TokenRefs and
// compares to the stored AutoNumberCanonicalHash. Only runs when BOTH
// are non-empty (V2 is a drift detector, not a presence requirement —
// Lane C2 is allowed to leave these NULL until Lane D writes them).
func runV2(fields []autonumberField, add func(string, ...any)) {
	for _, f := range fields {
		if f.TokenRefs == "" || f.CanonicalHash == "" {
			continue
		}
		// Parse TokenRefs to normalize the JSON shape; then
		// canonicalize the (format, scope, tokenRefs, start) tuple.
		var tokenRefs []autonumber.CrossRecordRef
		if err := json.Unmarshal([]byte(f.TokenRefs), &tokenRefs); err != nil {
			add("V2: field %s.%s (id=%s): AutoNumberTokenRefs is not valid JSON: %v",
				f.ObjectName, f.FieldName, f.ID, err)
			continue
		}
		canonical := canonicalizeForHashV2(f.Format, f.Scope, f.Start, tokenRefs)
		sum := sha256.Sum256([]byte(canonical))
		got := hex.EncodeToString(sum[:])
		if !strings.EqualFold(got, f.CanonicalHash) {
			add("V2: field %s.%s (id=%s): canonical hash drift — stored=%q recomputed=%q",
				f.ObjectName, f.FieldName, f.ID, f.CanonicalHash, got)
		}
	}
}

// canonicalizeForHashV2 builds a deterministic string for hash
// comparison. NOT the engine's canonicalize.go (which owns the
// authoritative format). This is a minimal V2 drift detector — if the
// stored hash was produced by a different canonicalize path (e.g. an
// admin edit bypassing data-spec-apply), this surfaces the drift. The
// engine's canonicalize test suite is responsible for correctness of
// the runtime path.
func canonicalizeForHashV2(
	format, scope string, start int64,
	tokenRefs []autonumber.CrossRecordRef,
) string {
	// Compact JSON encoding of tokenRefs — stable key order via
	// json.Marshal default (struct field order is stable).
	// canonicalizer-exempt: V2 drift detector by design uses an independent
	// serialization so it can catch drift in autonumber.CanonicalizeTokenRefs.
	// See the function comment above.
	b, _ := json.Marshal(tokenRefs) // canonicalizer-exempt
	return fmt.Sprintf("format=%s;scope=%s;start=%d;refs=%s",
		format, scope, start, string(b))
}

// runV3 checks that every scope=global autonumber field has a counter
// row with ScopeKey=''. scope=per_day and scope=per_prefix are exempt
// (lazy init — counter rows are created on first-insert because their
// ScopeKey is data-dependent; see autonumber_seed.go).
func runV3(
	fields []autonumberField,
	counters map[string]counterRow,
	add func(string, ...any),
) {
	for _, f := range fields {
		if f.Scope != autonumber.ScopeGlobal {
			continue
		}
		key := f.ID + "::"
		if _, ok := counters[key]; !ok {
			add("V3: field %s.%s (id=%s): scope=global but no counter row with ScopeKey='' in _System_AutoNumberCounter",
				f.ObjectName, f.FieldName, f.ID)
		}
	}
}

// runV4 checks that no counter row has NextValue < 1.
func runV4(counters map[string]counterRow, add func(string, ...any)) {
	for key, r := range counters {
		if r.NextValue < 1 {
			add("V4: counter row (field_id=%s scope=%q, key=%s) has NextValue=%d < 1",
				r.FieldID, r.ScopeKey, key, r.NextValue)
		}
	}
}

// runV5 verifies that the seeded counter's NextValue strictly exceeds
// the live-column max (seed already absorbed live data + 1). If the
// live column is empty, this check is a no-op. Only runs for
// scope=global fields.
func runV5(
	ctx context.Context,
	rawDB *sql.DB,
	fields []autonumberField,
	counters map[string]counterRow,
	add func(string, ...any),
) error {
	for _, f := range fields {
		if f.Scope != autonumber.ScopeGlobal {
			continue
		}
		counter, ok := counters[f.ID+"::"]
		if !ok {
			// V3 already reports this; skip V5 check.
			continue
		}
		spec, _, err := autonumber.Parse(f.Format)
		if err != nil {
			// V1 already reports; skip.
			continue
		}
		derived, err := deriveCounterRegex(spec)
		if err != nil {
			add("V5: field %s.%s (id=%s): cannot derive regex: %v",
				f.ObjectName, f.FieldName, f.ID, err)
			continue
		}
		colExists, err := physicalColumnExists(ctx, rawDB, f.ObjectName, f.FieldName)
		if err != nil {
			return err
		}
		if !colExists {
			continue
		}
		liveMax, err := queryLiveMaxCounter(ctx, rawDB, f.ObjectName, f.FieldName, derived)
		if err != nil {
			// V5 is tolerant of unparseable live rows because
			// SeedAutoNumberCounters is fail-closed — if we got
			// here, the seed succeeded, meaning every live row
			// was parseable. An unparseable row now would be
			// new drift since the seed ran; surface as violation
			// rather than erroring the whole verify pass.
			add("V5: field %s.%s (id=%s): live scan failed: %v",
				f.ObjectName, f.FieldName, f.ID, err)
			continue
		}
		if counter.NextValue <= liveMax {
			add("V5: field %s.%s (id=%s): NextValue=%d ≤ live_max=%d (seed did not absorb live data)",
				f.ObjectName, f.FieldName, f.ID, counter.NextValue, liveMax)
		}
	}
	return nil
}

// runV7 verifies scope=per_prefix coverage: every prefix bucket
// observed in live data must have a corresponding counter row with
// NextValue > live_max(prefix).
//
// This closes the Codex adversarial round-3 finding that
// scanLivePerPrefixCounters + mid-seed concurrent writes could leave
// a bucket with stale NextValue (or no row at all), producing
// duplicate rendered Names post-cutover. V7 catches drift that V3/V5
// miss because they only run for scope=global fields.
//
// Failure to parse a row is reported as a violation (tolerant) rather
// than bubbling up — same policy as V5: SeedAutoNumberCounters is
// fail-closed, so if verify reaches V7 the seed already accepted
// every parseable row. Subsequent unparseable rows represent drift
// that operators should surface without aborting the whole pass.
//
// Operator note: for V7 to pass reliably, writers MUST be quiesced
// between SeedAutoNumberCounters and VerifyAutoNumberInvariants.
// Concurrent inserts during that window can still race — V7 only
// detects the race, it does not prevent it. See
// SeedAutoNumberCounters' docstring for the cutover prerequisite.
func runV7(
	ctx context.Context,
	rawDB *sql.DB,
	fields []autonumberField,
	counters map[string]counterRow,
	add func(string, ...any),
) error {
	for _, f := range fields {
		if f.Scope != autonumber.ScopePerPrefix {
			continue
		}
		spec, _, err := autonumber.Parse(f.Format)
		if err != nil {
			// V1 already reports; skip.
			continue
		}
		derived, err := deriveCounterRegex(spec)
		if err != nil {
			add("V7: field %s.%s (id=%s): cannot derive regex: %v",
				f.ObjectName, f.FieldName, f.ID, err)
			continue
		}
		// queryLivePerPrefixCounters is the TOLERANT variant of the
		// seed-time per-prefix scanner — matches V5's policy so a
		// single malformed legacy row does not short-circuit drift
		// detection for every other bucket on this field. Malformed
		// rows are surfaced as separate violations and the remaining
		// good rows are still evaluated.
		perPrefixMax, err := queryLivePerPrefixCounters(
			ctx, rawDB,
			f.ObjectName, f.FieldName,
			derived, f.ID, add,
		)
		if err != nil {
			add("V7: field %s.%s (id=%s): live scan failed: %v",
				f.ObjectName, f.FieldName, f.ID, err)
			continue
		}
		for prefix, liveMax := range perPrefixMax {
			key := f.ID + "::" + prefix
			counter, ok := counters[key]
			if !ok {
				add("V7: field %s.%s (id=%s): prefix bucket %q observed in live data (live_max=%d) but no counter row (seed missed it — likely mid-seed race)",
					f.ObjectName, f.FieldName, f.ID, prefix, liveMax)
				continue
			}
			if counter.NextValue <= liveMax {
				add("V7: field %s.%s (id=%s): prefix bucket %q NextValue=%d ≤ live_max=%d (counter stale — likely mid-seed race or drift)",
					f.ObjectName, f.FieldName, f.ID, prefix, counter.NextValue, liveMax)
			}
		}
	}
	return nil
}

// queryLivePerPrefixCounters is the tolerant variant of
// scanLivePerPrefixCounters (autonumber_seed.go). It mirrors V5's
// queryLiveMaxCounter policy: any malformed / non-matching / counter-
// overflow row is reported as a separate violation via `add` and then
// skipped, so the remainder of the buckets still receive drift
// checks.
//
// Returns an empty map (and no error) when the physical table/column
// does not exist — same semantics as scanLivePerPrefixCounters.
func queryLivePerPrefixCounters(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	derived *regexp.Regexp,
	fieldID string,
	add func(string, ...any),
) (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, err
	}
	if !ok {
		return nil, nil
	}
	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, err
	}
	defer rows.Close()
	out := make(map[string]int64)
	for rows.Next() {
		var rowID, v sql.NullString
		if err := rows.Scan(&rowID, &v); err != nil {
			return nil, err
		}
		if !v.Valid || v.String == "" {
			continue
		}
		loc := derived.FindStringSubmatchIndex(v.String)
		if loc == nil || len(loc) < 4 {
			add("V7: field %s.%s (id=%s): row Id=%q value %q does not match derived regex (drift since seed)",
				objectName, fieldName, fieldID, rowID.String, v.String)
			continue
		}
		counterStr := v.String[loc[2]:loc[3]]
		counter, perr := strconv.ParseInt(counterStr, 10, 64)
		if perr != nil {
			add("V7: field %s.%s (id=%s): row Id=%q counter %q overflows int64",
				objectName, fieldName, fieldID, rowID.String, counterStr)
			continue
		}
		prefix := v.String[loc[0]:loc[2]] + v.String[loc[3]:loc[1]]
		if prev, ok := out[prefix]; !ok || counter > prev {
			out[prefix] = counter
		}
	}
	if err := rows.Err(); err != nil {
		return out, fmt.Errorf("autonumber verify V7 %s.%s: row iteration: %w",
			objectName, fieldName, err)
	}
	return out, nil
}

// queryLiveMaxCounter is a quieter variant of scanLiveMaxCounter —
// unparseable rows are skipped (not fail-closed) so V5 surfaces the
// real mismatch rather than bubbling the parse error upward.
func queryLiveMaxCounter(
	ctx context.Context,
	rawDB *sql.DB,
	objectName, fieldName string,
	derived *regexp.Regexp,
) (int64, error) {
	// V1-C1 SQL injection guard — shared whitelist (validSQLIdentifier)
	// with scanLiveMaxCounter in autonumber_seed.go. V5 is read-only
	// but still splices names into SQL via fmt.Sprintf; reject
	// non-whitelisted identifiers early.
	if err := assertValidSQLIdentifier("objectName", objectName); err != nil {
		return 0, err
	}
	if err := assertValidSQLIdentifier("fieldName", fieldName); err != nil {
		return 0, err
	}
	query := fmt.Sprintf(
		"SELECT `%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, err
	}
	defer rows.Close()
	var maxV int64
	for rows.Next() {
		var v sql.NullString
		if err := rows.Scan(&v); err != nil {
			return 0, err
		}
		if !v.Valid || v.String == "" {
			continue
		}
		m := derived.FindStringSubmatch(v.String)
		if m == nil || len(m) < 2 {
			continue
		}
		n, err := strconv.ParseInt(m[1], 10, 64)
		if err != nil {
			continue
		}
		if n > maxV {
			maxV = n
		}
	}
	return maxV, rows.Err()
}

// runV6 enforces the F7 dual-check: for each autonumber format that
// references `{RecordType.*}`, every _System_RecordType row belonging
// to the *referencing object* must have record_type_code IS NOT NULL
// AND != ''. NULL / empty blocks deploy.
//
// Scope by object: only RecordTypes whose parent object has a format
// using `{RecordType.*}` need a code. An Order RecordType (whose
// Order.Name format uses `{Account.BillingCountryCode}` but not
// `{RecordType.*}`) does NOT need a code; flagging it would block
// deploy on false grounds.
func runV6(
	ctx context.Context,
	rawDB *sql.DB,
	fields []autonumberField,
	add func(string, ...any),
) error {
	objectsNeedingRT := map[string]struct{}{}
	var referencingFields []string
	for _, f := range fields {
		if strings.Contains(f.Format, "{RecordType.") {
			objectsNeedingRT[f.ObjectName] = struct{}{}
			referencingFields = append(referencingFields,
				fmt.Sprintf("%s.%s", f.ObjectName, f.FieldName))
		}
	}
	if len(objectsNeedingRT) == 0 {
		return nil
	}

	// Check the table first — its absence is itself a V6 violation.
	ok, err := tableExistsRaw(ctx, rawDB, "_System_RecordType")
	if err != nil {
		return err
	}
	if !ok {
		add("V6: formats %v reference {RecordType.*} but _System_RecordType table is absent",
			referencingFields)
		return nil
	}

	// Resolve the set of ObjectIds for objects that need record types.
	objectNames := make([]string, 0, len(objectsNeedingRT))
	for name := range objectsNeedingRT {
		objectNames = append(objectNames, name)
	}
	placeholders := strings.Repeat("?,", len(objectNames))
	placeholders = placeholders[:len(placeholders)-1]
	args := make([]any, len(objectNames))
	for i, n := range objectNames {
		args[i] = n
	}
	objIDQuery := fmt.Sprintf(
		"SELECT Id FROM `_System_Object` WHERE ApiName IN (%s)",
		placeholders,
	)
	idRows, err := rawDB.QueryContext(ctx, objIDQuery, args...)
	if err != nil {
		return err
	}
	var objectIds []string
	for idRows.Next() {
		var id string
		if err := idRows.Scan(&id); err != nil {
			idRows.Close()
			return err
		}
		objectIds = append(objectIds, id)
	}
	idRows.Close()
	if err := idRows.Err(); err != nil {
		return err
	}
	if len(objectIds) == 0 {
		// No matching parent object rows — nothing to check. This is a
		// legitimate pre-apply state and not a violation.
		return nil
	}

	// F7 dual-check: find RecordTypes on those specific objects whose
	// code is NULL OR empty. Report each offender so admins can fix in
	// bulk. Uses Name because _System_RecordType lacks a DeveloperName
	// column; SF-parity of DeveloperName is a follow-up.
	rtPlaceholders := strings.Repeat("?,", len(objectIds))
	rtPlaceholders = rtPlaceholders[:len(rtPlaceholders)-1]
	rtArgs := make([]any, len(objectIds))
	for i, id := range objectIds {
		rtArgs[i] = id
	}
	rows, err := rawDB.QueryContext(ctx,
		fmt.Sprintf(
			"SELECT Id, Name FROM `_System_RecordType` "+
				"WHERE (`record_type_code` IS NULL OR `record_type_code` = '') "+
				"AND ObjectId IN (%s)",
			rtPlaceholders,
		),
		rtArgs...,
	)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var id, name sql.NullString
		if err := rows.Scan(&id, &name); err != nil {
			return err
		}
		add("V6: RecordType id=%q Name=%q has record_type_code NULL or empty "+
			"(referencing formats: %v)",
			id.String, name.String, referencingFields)
	}
	return rows.Err()
}