// Package autonumber implements metadata-driven identifier assignment for NexusCRM.
package autonumber

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"math/rand"
	"sync"
	"time"
)

// maxReservationRetries is the optimistic-UPDATE retry budget for
// reserveOne / reserveBatch. After `maxRetries` misses the reserver
// falls back to `SELECT ... FOR UPDATE` (PRD §5.2 6.f).
//
// PRD sets maxRetries=5; aligned with §12.1 R3.
const maxReservationRetries = 5

// deriveScopeKey computes the scope_key column value for a given
// (scope, assignedAt) tuple (PRD §4.4 scope key semantics).
//
// Scope values:
//   - "global"  → ""
//   - "per_day" → "YYYY-MM-DD" (UTC)
//
// Unknown scopes fall back to "" and are logically equivalent to
// global; the seed-time validator rejects unknown scopes, so this
// only guards against runtime metadata corruption.
func deriveScopeKey(scope string, assignedAt time.Time) string {
	switch scope {
	case ScopePerDay:
		// Per PRD §4.4: scope_key is 'YYYY-MM-DD' for per_day.
		// This matches the system_tables.json description and
		// leaves room for VARCHAR(50); PRD §5.2 pseudocode showed
		// 'YYYYMMDD' but the authoritative storage shape in
		// system_tables (v2 Lane B) is hyphenated — we follow the
		// DB contract.
		return assignedAt.UTC().Format("2006-01-02")
	default:
		return ""
	}
}

// widerJitter returns a random sleep duration in [0, 2^(attempt+3) ms)
// per PRD §12.1 R3 / §5.2 step 6.e.
func widerJitter(attempt int, rng *rand.Rand) time.Duration {
	if attempt < 0 {
		attempt = 0
	}
	// 2^(attempt+3). Cap at 2^16 ms (~65s) to keep the envelope sane
	// even if somehow `attempt` leaks past maxReservationRetries.
	shift := attempt + 3
	if shift > 16 {
		shift = 16
	}
	upperMS := 1 << shift
	return time.Duration(rng.Intn(upperMS)) * time.Millisecond
}

// reserveInput bundles the parameters for reserveOne/reserveBatch so
// the two paths share a single argument shape. The engine packs this
// at the top of AssignOne / AssignBatch.
//
// NOTE (V2-H3, Round-2 review): per_day scope is NOT active at
// cutover — all 132 seeded fields ship as scope=global (PRD §8.1).
// The lazy-init path for per_day new-day rows uses in.startValue as
// the initial NextValue; there is no hook to absorb live rows from
// that day. When per_day is re-activated (v2.1), operators MUST
// pre-seed day buckets via SeedAutoNumberCounters before enabling —
// runtime lazy-init does NOT scan live data.
type reserveInput struct {
	fieldID    string
	scope      string
	startValue int64
	// scopeKeyFn, when non-nil, computes the scope_key given the
	// reservation's effective `assignedAt`. The engine sets this for
	// ScopePerPrefix fields: the closure captures `spec` +
	// `crossValues` and invokes RenderScopePrefix, so the date tokens
	// in the rendered prefix use the SAME time source the renderer
	// uses for the final identifier.
	//
	// Why a closure and not a precomputed string: date tokens in a
	// per_prefix format (e.g. `{YY}{MM}{DD}` in
	// `{Account.BillingCountryCode}-QT-{YY}{MM}{DD}-{00}`) would
	// otherwise sample Go `time.Now().UTC()` in the engine while the
	// renderer samples `UTC_TIMESTAMP(6)` inside the reservation txn
	// — a mismatch at a UTC day boundary produces Name values whose
	// date component differs from the scope_key's date component,
	// allowing two concurrent inserts to collide on the same rendered
	// Name while landing in different counter buckets. The closure
	// defers rendering until the reserver has sampled its DB time,
	// so both derivations see one consistent timestamp. (Codex
	// adversarial review 2026-04-18 finding #1.)
	//
	// When nil (all legacy call sites), scope_key falls back to
	// deriveScopeKey(scope, assignedAt) — backwards compatible.
	scopeKeyFn func(assignedAt time.Time) (string, error)
	// liveMaxFn, when non-nil, is invoked by the lazy-init UPSERT
	// path (row-missing branch for ScopePerDay / ScopePerPrefix) to
	// seed the counter row's initial NextValue based on existing
	// live-table records that already fall in this scopeKey's
	// bucket. For ScopePerPrefix this is the race-closure that
	// Codex round-3 flagged: a bucket observed in live data after
	// SeedAutoNumberCounters ran (either because a concurrent
	// writer slipped a record in mid-seed or because the bucket is
	// genuinely new) would otherwise lazy-init at startValue=1 and
	// collide with the existing record. The closure returns the
	// maximum counter value already in the bucket; reserveOne then
	// uses max(startValue, liveMax+1) as the initial NextValue.
	//
	// `assignedAt` is passed in so closures that render prefix
	// tokens with date components (the common per_prefix case)
	// derive the query's before/after anchors against the same
	// timestamp the reserver uses for scope_key and the renderer.
	//
	// When nil, initialNext falls back to in.startValue — the
	// legacy behavior, safe for genuinely-new buckets but unsafe
	// for buckets that already contain records.
	//
	// Errors from liveMaxFn are fatal to the reservation (fail-
	// closed) so duplicate names are never rendered.
	liveMaxFn func(ctx context.Context, assignedAt time.Time) (int64, error)
}

// initialNextFor computes the starting NextValue for a freshly-created
// counter row (lazy-init branch). When `in.liveMaxFn` is set, it is
// invoked with the reservation's assignedAt so the closure can render
// the live-scan anchors (prefix before/after the counter) against the
// same timestamp used for scope_key and the renderer. The returned
// value is `max(in.startValue, liveMax+1)`. When liveMaxFn is nil,
// falls back to `in.startValue` — the legacy behavior.
//
// Errors from liveMaxFn are propagated unchanged (fail-closed): a
// race between seed and concurrent writers must NOT silently produce
// duplicate identifiers. See reserveInput.liveMaxFn for rationale.
func initialNextFor(ctx context.Context, in reserveInput, assignedAt time.Time) (int64, error) {
	startValue := in.startValue
	if startValue < 1 {
		startValue = 1
	}
	if in.liveMaxFn == nil {
		return startValue, nil
	}
	liveMax, err := in.liveMaxFn(ctx, assignedAt)
	if err != nil {
		return 0, fmt.Errorf("autonumber reserve: lazy-init live_max: %w", err)
	}
	// Defensive: liveMax < 0 or liveMax == MaxInt64 are unreachable
	// with sane metadata + sane live data, but guard so
	// liveMax+1 cannot underflow (<0 → 0+1) or overflow
	// (MaxInt64+1 → MinInt64). Pass A round-6 MEDIUM.
	if liveMax < 0 || liveMax == math.MaxInt64 {
		return startValue, nil
	}
	if liveMax+1 > startValue {
		return liveMax + 1, nil
	}
	return startValue, nil
}

// scopeKeyFor returns the effective scope_key for a reservation,
// respecting scopeKeyFn precedence. All reservation code paths MUST
// use this helper rather than calling deriveScopeKey directly on an
// input, so ScopePerPrefix fields consistently honor the engine-
// supplied partition key computed against the same assignedAt the
// renderer will use.
//
// Precedence:
//  1. in.scopeKeyFn (if non-nil) — invoked with the caller-supplied
//     `assignedAt`. Returned string is used verbatim as scope_key.
//     Any closure error propagates up unchanged.
//  2. deriveScopeKey(in.scope, assignedAt) — the legacy time-derived
//     rule used by ScopeGlobal / ScopePerDay.
func scopeKeyFor(in reserveInput, assignedAt time.Time) (string, error) {
	if in.scopeKeyFn != nil {
		return in.scopeKeyFn(assignedAt)
	}
	return deriveScopeKey(in.scope, assignedAt), nil
}

// reserver owns the DB resources required to reserve counter values
// (§5.2). It is constructed once per Engine and shared across
// AssignOne / AssignBatch callers.
type reserver struct {
	// reservationDB is the dedicated *sql.DB handle for the short
	// optimistic-UPDATE txn. NEVER the caller's handle.
	reservationDB *sql.DB
	// fallbackDB may equal reservationDB — separated here so tests
	// can override for the `FOR UPDATE` path if desired.
	fallbackDB *sql.DB
	// sems serializes the fallback entry per (fieldID, scope_key).
	sems *semaphoreCache
	// metrics emits per-reservation counters / histograms. Optional —
	// if nil, emissions are noop.
	metrics *reservationMetrics
	// rngPool amortizes per-call rand.Rand allocation; every
	// goroutine draws a dedicated rng from the pool for jitter.
	rngPool sync.Pool
}

// newReserver constructs a reserver wired to the given DBs + LRU
// cache. `fieldCount` sizes the semaphore LRU.
func newReserver(reservationDB, fallbackDB *sql.DB, fieldCount int, metrics *reservationMetrics) *reserver {
	if fallbackDB == nil {
		fallbackDB = reservationDB
	}
	r := &reserver{
		reservationDB: reservationDB,
		fallbackDB:    fallbackDB,
		sems:          newSemaphoreCache(fieldCount, nil),
		metrics:       metrics,
	}
	r.rngPool.New = func() any {
		return rand.New(rand.NewSource(time.Now().UnixNano() ^ int64(fieldCount)))
	}
	return r
}

// beginPessimisticTx acquires a dedicated *sql.Conn, forces
// SESSION-level TiDB pessimistic mode on THAT conn, and returns a txn
// rooted on the same conn plus a cleanup closure the caller MUST
// invoke after commit/rollback. The cleanup resets the session var
// back to its server-default BEFORE returning the conn to the pool —
// otherwise the conn re-enters the shared pool still pinned to
// pessimistic mode and silently flips future pooled queries over
// time (V2-R3-C1).
//
// V2-C2 Round-2 review: the previous implementation called
// `SET SESSION tidb_txn_mode = 'pessimistic'` inside the txn via
// tx.ExecContext — under TiDB that statement is silently ignored for
// the active txn (pessimistic mode must be set BEFORE BEGIN), and
// calling it on `db.ExecContext` hits a random pooled conn that has
// no relationship to the conn used for the next BeginTx.
//
// V2-R3-H2: the SET SESSION error is no longer swallowed. If
// pessimistic mode can't be set, we fail the reservation rather than
// silently running the FOR UPDATE lane in optimistic mode (which
// defeats the whole fallback serialization).
//
// V2-R3-C1: cleanup performs `SET SESSION tidb_txn_mode = DEFAULT`
// before `conn.Close()`. `DEFAULT` restores the server-configured
// default — safer than hard-coding 'optimistic' in case a future
// TiDB config sets something else. If the reset itself fails we
// still close the conn (calling Close marks the conn broken and
// the pool discards it, which is the correct fallback).
//
// By routing through `db.Conn(ctx)` we guarantee that the SESSION
// statement AND the BeginTx see the SAME underlying connection.
//
// NOTE (pool sizing): each active reservation pins a pool conn for
// the full SET + BEGIN + UPDATE + COMMIT duration. Under high
// concurrency the reservation pool can starve unrelated app queries
// if it shares the app's main *sql.DB. Size
// `ReservationDB.SetMaxOpenConns(n)` with n ≥ peak autonumber QPS ×
// avg reservation latency; or dedicate a separate *sql.DB for the
// reservation path.
func beginPessimisticTx(ctx context.Context, db *sql.DB) (*sql.Tx, func() error, error) {
	conn, err := db.Conn(ctx)
	if err != nil {
		return nil, nil, fmt.Errorf("acquire conn: %w", err)
	}

	cleanup := func() error {
		// Reset BEFORE returning the conn to the pool so we don't
		// pollute other callers sharing this *sql.DB. Use a background
		// context so the reset still runs even if the caller's ctx
		// is already cancelled. `DEFAULT` restores the
		// server-configured default, which is 'pessimistic' under
		// TiDB 4.0+ defaults but may be 'optimistic' on some
		// clusters — either way, `DEFAULT` is the correct "undo".
		resetErr := resetTxnModeDefault(context.Background(), conn)
		closeErr := conn.Close()
		// Prefer surfacing the reset error so callers can alarm on
		// pool pollution; close errors on a broken conn are
		// expected.
		if resetErr != nil {
			return resetErr
		}
		return closeErr
	}

	if _, err := conn.ExecContext(ctx, "SET SESSION tidb_txn_mode = 'pessimistic'"); err != nil {
		// Fatal: without pessimistic mode the fallback FOR UPDATE
		// lane degrades to an optimistic read with no blocking lock
		// (V2-R3-H2). Run cleanup (which will try to DEFAULT the
		// already-errored conn, then Close) so the conn does not
		// stay pinned in whatever half-state the failed SET left.
		_ = cleanup()
		return nil, nil, fmt.Errorf("set tidb_txn_mode=pessimistic: %w", err)
	}

	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		_ = cleanup()
		return nil, nil, fmt.Errorf("begin tx: %w", err)
	}
	return tx, cleanup, nil
}

// resetTxnModeDefault restores the session `tidb_txn_mode` to the
// server default. Split out so the cleanup path stays readable.
func resetTxnModeDefault(ctx context.Context, conn *sql.Conn) error {
	if _, err := conn.ExecContext(ctx, "SET SESSION tidb_txn_mode = DEFAULT"); err != nil {
		return fmt.Errorf("reset tidb_txn_mode to DEFAULT: %w", err)
	}
	return nil
}

// reserveOne — short optimistic-UPDATE txn with livelock-free first-use
// (PRD §5.2 lines 497-533 VERBATIM). Returns (assignedAt, counter).
//
// P0-R2-B DO's & DON'Ts:
//   - Scope=global missing row → ErrCounterMissing (NOT reseed).
//   - Scope=per_day missing row → own-txn UPSERT with
//     `ON DUPLICATE KEY UPDATE next_value = next_value`, then retry
//     outer loop. Do NOT `Rollback()` after INSERT (would roll back
//     the insert and livelock).
//   - `assignedAt` sampled ONCE inside `BeginTx` BEFORE sem.Acquire on
//     the fallback path (P2-R3 + fixpack F11).
//   - V2-C2: txn obtained via beginPessimisticTx so TiDB sees
//     pessimistic mode BEFORE BEGIN (required; in-txn SET is ignored).
func (r *reserver) reserveOne(ctx context.Context, in reserveInput) (time.Time, int64, error) {
	var zeroTime time.Time
	startT := time.Now()

	rng := r.rngPool.Get().(*rand.Rand)
	defer r.rngPool.Put(rng)

	for attempt := 0; attempt < maxReservationRetries; attempt++ {
		r.metrics.observeRetryAttempt(in.fieldID, attempt)

		assignedAt, counterValue, done, retryable, err := r.reserveOneAttempt(ctx, in, attempt, rng)
		if done {
			r.metrics.observeReservationDuration(in.fieldID, time.Since(startT))
			return assignedAt, counterValue, err
		}
		if err != nil {
			// Non-retryable failure.
			r.metrics.observeReservationDuration(in.fieldID, time.Since(startT))
			return zeroTime, 0, err
		}
		if !retryable {
			// Shouldn't happen — guard anyway.
			break
		}
		// otherwise: CAS miss or per-day lazy-row retry → loop.
	}

	// Starvation fallback.
	at, val, err := r.reserveOneFallbackFORUPDATE(ctx, in)
	r.metrics.observeReservationDuration(in.fieldID, time.Since(startT))
	return at, val, err
}

// reserveOneAttempt runs ONE optimistic-UPDATE attempt for reserveOne.
// Return contract:
//   - done=true, err=nil   → reservation succeeded; (assignedAt, counter) valid
//   - done=true, err!=nil  → terminal failure (including ErrCounterMissing)
//   - done=false, err==nil → caller should continue the retry loop
//     (CAS miss, per-day lazy-row retry, etc.)
//   - retryable is true whenever done=false (redundant flag kept for
//     call-site clarity)
//
// Encapsulates the conn+tx lifecycle so the single-attempt path can
// use deferred cleanup without leaking a conn between retries
// (V2-C2 Round-2 review).
func (r *reserver) reserveOneAttempt(
	ctx context.Context,
	in reserveInput,
	attempt int,
	rng *rand.Rand,
) (assignedAt time.Time, counter int64, done bool, retryable bool, err error) {
	var zeroTime time.Time

	tx, cleanup, err := beginPessimisticTx(ctx, r.reservationDB)
	if err != nil {
		return zeroTime, 0, true, false, fmt.Errorf("autonumber reserve: %w", err)
	}
	// Ensure the conn returns to the pool (with session reset) after
	// commit OR rollback. `tx.Rollback()` after commit is a no-op.
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	// 6.a. DB wall-clock inside reservation txn (C#1).
	assignedAt, err = sampleAssignedAt(ctx, tx)
	if err != nil {
		return zeroTime, 0, true, false, err
	}

	// 6.b. Derive scope_key. Closure-based derivation (per_prefix)
	// uses THIS assignedAt so the scope_key's date component matches
	// the renderer's date component — preventing cross-day collisions
	// (Codex adversarial finding #1).
	scopeKey, err := scopeKeyFor(in, assignedAt)
	if err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserve: scope_key: %w", err)
	}

	// 6.c. Read current next_value (plain read).
	var currentNext int64
	err = tx.QueryRowContext(ctx,
		`SELECT `+"`NextValue`"+` FROM `+"`_System_AutoNumberCounter`"+`
           WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=?`,
		in.fieldID, scopeKey,
	).Scan(&currentNext)

	if errors.Is(err, sql.ErrNoRows) {
		// Policy gate — scope=global row-missing is CORRUPTION.
		// Return early; deferred rollback+close releases the conn.
		if in.scope != ScopePerDay && in.scope != ScopePerPrefix {
			return zeroTime, 0, true, false,
				fmt.Errorf("%w: fieldID=%s scopeKey=%q",
					ErrCounterMissing, in.fieldID, scopeKey)
		}

		// Own-txn UPSERT (Approach Y) for per_day new-day or
		// per_prefix fresh bucket. initialNextFor consults
		// in.liveMaxFn (race-closure for per_prefix — Codex
		// round-3 follow-up); for per_day or when liveMaxFn is
		// unwired, falls back to in.startValue.
		//
		// upsertPerDayRow uses its own conn+txn; we let the deferred
		// rollback+close clean up THIS attempt's conn before
		// returning "retry outer loop".
		initialNext, err := initialNextFor(ctx, in, assignedAt)
		if err != nil {
			return zeroTime, 0, true, false, err
		}
		if err := r.upsertPerDayRow(ctx, in.fieldID, scopeKey, initialNext, assignedAt); err != nil {
			return zeroTime, 0, true, false, err
		}
		return zeroTime, 0, false, true, nil // retry outer loop
	} else if err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserve: read next_value: %w", err)
	}

	// 6.e. Optimistic UPDATE (CAS on NextValue).
	res, err := tx.ExecContext(ctx,
		`UPDATE `+"`_System_AutoNumberCounter`"+`
            SET `+"`NextValue`"+` = `+"`NextValue`"+` + 1,
                `+"`LastModifiedDate`"+` = ?
          WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=? AND `+"`NextValue`"+`=?`,
		assignedAt, in.fieldID, scopeKey, currentNext,
	)
	if err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserve: CAS update: %w", err)
	}
	affected, _ := res.RowsAffected()
	if affected == 0 {
		// Jitter only when we will retry.
		if attempt < maxReservationRetries-1 {
			sleep := widerJitter(attempt, rng)
			select {
			case <-ctx.Done():
				return zeroTime, 0, true, false, ctx.Err()
			case <-time.After(sleep):
			}
		}
		return zeroTime, 0, false, true, nil // retry outer loop
	}
	if err := tx.Commit(); err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserve: commit: %w", err)
	}
	committed = true
	return assignedAt, currentNext, true, false, nil
}

// reserveOneFallbackFORUPDATE is the short FOR UPDATE lane gated by a
// per-counter size-1 semaphore. Runs on `fallbackDB` NOT on the
// caller's txn (MUST NOT reintroduce Round-1 P1-1 outer-txn held
// FOR UPDATE).
//
// V2-C2: uses beginPessimisticTx — TiDB requires SET SESSION
// tidb_txn_mode='pessimistic' BEFORE BEGIN. The previous in-txn SET
// was silently ignored under TiDB, leaving the FOR UPDATE in
// optimistic mode (no blocking lock), which defeated the fallback's
// serialization goal.
//
// Round-5 Codex V2-Codex-H5 — sem-first reorder:
//
// The semaphore is now acquired BEFORE `beginPessimisticTx` so a
// blocked goroutine does NOT hold a DB connection from the pool while
// waiting. The previous order (conn → sem) meant that when N
// goroutines piled up on the same counter under fallback, all N
// reserved a pool connection each; a bounded pool could run dry
// before any fallback completed.
//
// scope_key derivation:
//   - scope=global → scope_key is "" (deriveScopeKey is value-only).
//     Pre-tx derivation is exact; the sem key matches what the UPDATE
//     will use.
//   - scope=per_day → we pre-derive using `time.Now().UTC()` as an
//     approximation of the DB's UTC_TIMESTAMP(6). In the steady state
//     these differ by < a few ms and produce the same YYYY-MM-DD
//     bucket. Within seconds of a UTC day boundary the sem key MAY
//     differ from the actual per-row scope_key sampled inside the
//     txn; that temporarily weakens the fallback's thundering-herd
//     throttle but correctness is unchanged because the FOR UPDATE
//     on the actual counter row serializes at the row level.
//   - scope=per_prefix → scope_key comes from in.scopeKeyOverride
//     (populated by the engine via RenderScopePrefix before calling
//     the reserver). The override is time-independent, so the sem
//     key and the in-txn scope_key are identical by construction —
//     no day-boundary window exists for per_prefix. The sem key is
//     just `scopeKeyFor(in, time.Now().UTC())` = the override.
func (r *reserver) reserveOneFallbackFORUPDATE(ctx context.Context, in reserveInput) (time.Time, int64, error) {
	var zeroTime time.Time

	// 1. Pre-derive scope_key and acquire the per-counter semaphore
	// BEFORE touching the pool. A blocked Acquire does NOT hold a DB
	// conn (V2-Codex-H5 — pool-starvation fix).
	//
	// The pre-txn scope_key is an APPROXIMATION: for per_day and
	// per_prefix formats with date tokens it uses `time.Now().UTC()`,
	// which may differ by a few ms (or days in a misconfigured DSN)
	// from the DB-sampled `assignedAt` used inside the txn. Any drift
	// only weakens the thundering-herd throttle for this fallback —
	// correctness is preserved because the in-txn scopeKey (line
	// below) is what the counter row is keyed on.
	preScopeKey, err := scopeKeyFor(in, time.Now().UTC())
	if err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber reserve: pre-scope_key: %w", err)
	}
	sem := r.sems.Get(in.fieldID, preScopeKey)
	if err := sem.Acquire(ctx, 1); err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber fallback: sem acquire: %w", err)
	}
	defer sem.Release(1)

	// F15: fallback-in-flight gauge increments on entry, decrements on
	// return. Label includes fieldID only — scope_key is derived from
	// assignedAt which isn't yet known; scope_key added at decrement.
	r.metrics.incFallbackInFlight(in.fieldID, "")
	defer r.metrics.decFallbackInFlight(in.fieldID, "")

	// 2. Now take a pool conn and start the pessimistic txn.
	tx, cleanup, err := beginPessimisticTx(ctx, r.fallbackDB)
	if err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber fallback: %w", err)
	}
	rollback := true
	defer func() {
		if rollback {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	// 3. Single clock sample inside the txn — authoritative scope_key
	// used by the row-targeting SQL below. If the DB-side sample
	// crossed a UTC day boundary since we pre-derived, the actual row
	// lookup uses the NEW scope_key; the pre-derived sem key covers
	// the old bucket. The FOR UPDATE still serializes the target row.
	assignedAt, err := sampleAssignedAt(ctx, tx)
	if err != nil {
		return zeroTime, 0, err
	}
	scopeKey, err := scopeKeyFor(in, assignedAt)
	if err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber reserve: scope_key: %w", err)
	}

	// 3. FOR UPDATE on exactly the row.
	var currentNext int64
	err = tx.QueryRowContext(ctx,
		`SELECT `+"`NextValue`"+` FROM `+"`_System_AutoNumberCounter`"+`
           WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=? FOR UPDATE`,
		in.fieldID, scopeKey,
	).Scan(&currentNext)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			if in.scope != ScopePerDay && in.scope != ScopePerPrefix {
				return zeroTime, 0, fmt.Errorf("%w: fieldID=%s scopeKey=%q",
					ErrCounterMissing, in.fieldID, scopeKey)
			}
			// per_day / per_prefix fallback row-missing — allocate
			// it under the already-held sem+txn. Use INSERT ... ON
			// DUPLICATE KEY UPDATE so concurrent winners reconcile.
			// initialNextFor consults in.liveMaxFn (per_prefix
			// race-closure).
			initialNext, err := initialNextFor(ctx, in, assignedAt)
			if err != nil {
				return zeroTime, 0, err
			}
			if _, err := tx.ExecContext(ctx,
				`INSERT INTO `+"`_System_AutoNumberCounter`"+`
                   (`+"`Id`"+`,`+"`FieldId`"+`,`+"`ScopeKey`"+`,`+"`NextValue`"+`,`+"`CreatedDate`"+`,`+"`LastModifiedDate`"+`)
                 VALUES (?, ?, ?, ?, ?, ?)
                 ON DUPLICATE KEY UPDATE `+"`NextValue`"+` = `+"`NextValue`"+``,
				generateCounterRowID(in.fieldID, scopeKey),
				in.fieldID, scopeKey, initialNext, assignedAt, assignedAt,
			); err != nil {
				return zeroTime, 0, fmt.Errorf("autonumber fallback: lazy-init upsert: %w", err)
			}
			// Re-read within the same txn to pick up whichever value
			// now lives in the row.
			if err := tx.QueryRowContext(ctx,
				`SELECT `+"`NextValue`"+` FROM `+"`_System_AutoNumberCounter`"+`
                   WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=? FOR UPDATE`,
				in.fieldID, scopeKey,
			).Scan(&currentNext); err != nil {
				return zeroTime, 0, fmt.Errorf("autonumber fallback: re-read: %w", err)
			}
		} else {
			return zeroTime, 0, fmt.Errorf("autonumber fallback: FOR UPDATE: %w", err)
		}
	}

	if _, err := tx.ExecContext(ctx,
		`UPDATE `+"`_System_AutoNumberCounter`"+`
            SET `+"`NextValue`"+` = `+"`NextValue`"+` + 1,
                `+"`LastModifiedDate`"+` = ?
          WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=?`,
		assignedAt, in.fieldID, scopeKey,
	); err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber fallback: post-lock update: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber fallback: commit: %w", err)
	}
	rollback = false
	return assignedAt, currentNext, nil
}

// reserveBatch reserves `n` contiguous counter values in ONE short
// reservation txn (PRD §5.3). Returns the starting value + the shared
// assignedAt. For scope=per_day the row may need lazy creation, same
// rules as reserveOne 6.d.
func (r *reserver) reserveBatch(ctx context.Context, in reserveInput, n int) (time.Time, int64, error) {
	var zeroTime time.Time
	if n <= 0 {
		return zeroTime, 0, fmt.Errorf("autonumber reserveBatch: n must be > 0, got %d", n)
	}
	if n == 1 {
		return r.reserveOne(ctx, in)
	}

	startT := time.Now()
	rng := r.rngPool.Get().(*rand.Rand)
	defer r.rngPool.Put(rng)

	for attempt := 0; attempt < maxReservationRetries; attempt++ {
		r.metrics.observeRetryAttempt(in.fieldID, attempt)

		assignedAt, startVal, done, retryable, err := r.reserveBatchAttempt(ctx, in, n, attempt, rng)
		if done {
			r.metrics.observeReservationDuration(in.fieldID, time.Since(startT))
			return assignedAt, startVal, err
		}
		if err != nil {
			r.metrics.observeReservationDuration(in.fieldID, time.Since(startT))
			return zeroTime, 0, err
		}
		if !retryable {
			break
		}
	}
	// Fallback path: degrade to a semaphore-serialized FOR UPDATE
	// batch reservation. Semantically identical to reserveOne fallback
	// but reserves `n` in one update.
	at, start, err := r.reserveBatchFallback(ctx, in, n)
	r.metrics.observeReservationDuration(in.fieldID, time.Since(startT))
	return at, start, err
}

// reserveBatchAttempt — single-attempt counterpart of reserveOneAttempt
// for batch reservations. See reserveOneAttempt for the done/retryable
// contract.
func (r *reserver) reserveBatchAttempt(
	ctx context.Context,
	in reserveInput,
	n, attempt int,
	rng *rand.Rand,
) (assignedAt time.Time, startVal int64, done bool, retryable bool, err error) {
	var zeroTime time.Time

	tx, cleanup, err := beginPessimisticTx(ctx, r.reservationDB)
	if err != nil {
		return zeroTime, 0, true, false, fmt.Errorf("autonumber reserveBatch: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	assignedAt, err = sampleAssignedAt(ctx, tx)
	if err != nil {
		return zeroTime, 0, true, false, err
	}
	scopeKey, err := scopeKeyFor(in, assignedAt)
	if err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserve-batch: scope_key: %w", err)
	}

	var currentNext int64
	err = tx.QueryRowContext(ctx,
		`SELECT `+"`NextValue`"+` FROM `+"`_System_AutoNumberCounter`"+`
           WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=?`,
		in.fieldID, scopeKey,
	).Scan(&currentNext)

	if errors.Is(err, sql.ErrNoRows) {
		if in.scope != ScopePerDay && in.scope != ScopePerPrefix {
			return zeroTime, 0, true, false,
				fmt.Errorf("%w: fieldID=%s scopeKey=%q",
					ErrCounterMissing, in.fieldID, scopeKey)
		}
		// initialNextFor consults in.liveMaxFn (per_prefix race-
		// closure). Legacy behavior preserved when liveMaxFn is nil.
		initialNext, err := initialNextFor(ctx, in, assignedAt)
		if err != nil {
			return zeroTime, 0, true, false, err
		}
		if err := r.upsertPerDayRow(ctx, in.fieldID, scopeKey, initialNext, assignedAt); err != nil {
			return zeroTime, 0, true, false, err
		}
		return zeroTime, 0, false, true, nil
	} else if err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserveBatch: read: %w", err)
	}

	// Block-reserve n counters.
	res, err := tx.ExecContext(ctx,
		`UPDATE `+"`_System_AutoNumberCounter`"+`
            SET `+"`NextValue`"+` = `+"`NextValue`"+` + ?,
                `+"`LastModifiedDate`"+` = ?
          WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=? AND `+"`NextValue`"+`=?`,
		int64(n), assignedAt, in.fieldID, scopeKey, currentNext,
	)
	if err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserveBatch: CAS update: %w", err)
	}
	affected, _ := res.RowsAffected()
	if affected == 0 {
		if attempt < maxReservationRetries-1 {
			sleep := widerJitter(attempt, rng)
			select {
			case <-ctx.Done():
				return zeroTime, 0, true, false, ctx.Err()
			case <-time.After(sleep):
			}
		}
		return zeroTime, 0, false, true, nil
	}
	if err := tx.Commit(); err != nil {
		return zeroTime, 0, true, false,
			fmt.Errorf("autonumber reserveBatch: commit: %w", err)
	}
	committed = true
	return assignedAt, currentNext, true, false, nil
}

// reserveBatchFallback is the batch counterpart of
// reserveOneFallbackFORUPDATE. Serialization identical.
// V2-C2: uses beginPessimisticTx — see reserveOneFallbackFORUPDATE.
//
// Round-5 Codex V2-Codex-H5 sem-first reorder: the semaphore is
// acquired BEFORE beginPessimisticTx so a blocked goroutine does NOT
// hold a pool conn while waiting. See reserveOneFallbackFORUPDATE
// doc-comment for the scope_key pre-derivation trade-off.
func (r *reserver) reserveBatchFallback(ctx context.Context, in reserveInput, n int) (time.Time, int64, error) {
	var zeroTime time.Time

	// 1. Pre-derive scope_key and acquire the per-counter semaphore
	// BEFORE touching the pool. Pre-txn key is approximate; see
	// reserveOneFallbackFORUPDATE for the full semantics.
	preScopeKey, err := scopeKeyFor(in, time.Now().UTC())
	if err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber reserve-batch: pre-scope_key: %w", err)
	}
	sem := r.sems.Get(in.fieldID, preScopeKey)
	if err := sem.Acquire(ctx, 1); err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber batch fallback: sem acquire: %w", err)
	}
	defer sem.Release(1)

	r.metrics.incFallbackInFlight(in.fieldID, "")
	defer r.metrics.decFallbackInFlight(in.fieldID, "")

	// 2. Pool conn + pessimistic txn.
	tx, cleanup, err := beginPessimisticTx(ctx, r.fallbackDB)
	if err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber batch fallback: %w", err)
	}
	rollback := true
	defer func() {
		if rollback {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	// 3. Authoritative clock sample inside the txn.
	assignedAt, err := sampleAssignedAt(ctx, tx)
	if err != nil {
		return zeroTime, 0, err
	}
	scopeKey, err := scopeKeyFor(in, assignedAt)
	if err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber reserve-batch: scope_key: %w", err)
	}

	var currentNext int64
	err = tx.QueryRowContext(ctx,
		`SELECT `+"`NextValue`"+` FROM `+"`_System_AutoNumberCounter`"+`
           WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=? FOR UPDATE`,
		in.fieldID, scopeKey,
	).Scan(&currentNext)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			if in.scope != ScopePerDay && in.scope != ScopePerPrefix {
				return zeroTime, 0, fmt.Errorf("%w: fieldID=%s scopeKey=%q",
					ErrCounterMissing, in.fieldID, scopeKey)
			}
			// initialNextFor consults in.liveMaxFn (per_prefix race-
			// closure). Legacy behavior preserved when liveMaxFn is nil.
			initialNext, err := initialNextFor(ctx, in, assignedAt)
			if err != nil {
				return zeroTime, 0, err
			}
			if _, err := tx.ExecContext(ctx,
				`INSERT INTO `+"`_System_AutoNumberCounter`"+`
                   (`+"`Id`"+`,`+"`FieldId`"+`,`+"`ScopeKey`"+`,`+"`NextValue`"+`,`+"`CreatedDate`"+`,`+"`LastModifiedDate`"+`)
                 VALUES (?, ?, ?, ?, ?, ?)
                 ON DUPLICATE KEY UPDATE `+"`NextValue`"+` = `+"`NextValue`"+``,
				generateCounterRowID(in.fieldID, scopeKey),
				in.fieldID, scopeKey, initialNext, assignedAt, assignedAt,
			); err != nil {
				return zeroTime, 0, fmt.Errorf("autonumber batch fallback: per-day upsert: %w", err)
			}
			if err := tx.QueryRowContext(ctx,
				`SELECT `+"`NextValue`"+` FROM `+"`_System_AutoNumberCounter`"+`
                   WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=? FOR UPDATE`,
				in.fieldID, scopeKey,
			).Scan(&currentNext); err != nil {
				return zeroTime, 0, fmt.Errorf("autonumber batch fallback: re-read: %w", err)
			}
		} else {
			return zeroTime, 0, fmt.Errorf("autonumber batch fallback: FOR UPDATE: %w", err)
		}
	}

	if _, err := tx.ExecContext(ctx,
		`UPDATE `+"`_System_AutoNumberCounter`"+`
            SET `+"`NextValue`"+` = `+"`NextValue`"+` + ?,
                `+"`LastModifiedDate`"+` = ?
          WHERE `+"`FieldId`"+`=? AND `+"`ScopeKey`"+`=?`,
		int64(n), assignedAt, in.fieldID, scopeKey,
	); err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber batch fallback: post-lock update: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return zeroTime, 0, fmt.Errorf("autonumber batch fallback: commit: %w", err)
	}
	rollback = false
	return assignedAt, currentNext, nil
}

// upsertPerDayRow creates (or no-ops) the counter row for a lazy-init
// scope — either scope=per_day (first use of a new UTC day bucket) or
// scope=per_prefix (first use of a data-derived rendered-prefix bucket;
// see RenderScopePrefix). The function is named for historical reasons
// (v3 started with only per_day lazy init); `scopeKey` is passed in
// opaquely and the helper is scope-agnostic. Executed in an INDEPENDENT
// txn (Approach Y / PRD §5.2 step 6.d) so the outer-loop retry picks up
// the row via the optimistic path.
func (r *reserver) upsertPerDayRow(
	ctx context.Context,
	fieldID, scopeKey string,
	initialNext int64,
	assignedAt time.Time,
) error {
	ownTx, err := r.reservationDB.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("autonumber upsert per-day: begin tx: %w", err)
	}
	rowID := generateCounterRowID(fieldID, scopeKey)
	if _, err := ownTx.ExecContext(ctx,
		`INSERT INTO `+"`_System_AutoNumberCounter`"+`
           (`+"`Id`"+`,`+"`FieldId`"+`,`+"`ScopeKey`"+`,`+"`NextValue`"+`,`+"`CreatedDate`"+`,`+"`LastModifiedDate`"+`)
         VALUES (?, ?, ?, ?, ?, ?)
         ON DUPLICATE KEY UPDATE `+"`NextValue`"+` = `+"`NextValue`"+``,
		rowID, fieldID, scopeKey, initialNext, assignedAt, assignedAt,
	); err != nil {
		_ = ownTx.Rollback()
		return fmt.Errorf("autonumber upsert per-day: insert: %w", err)
	}
	if err := ownTx.Commit(); err != nil {
		return fmt.Errorf("autonumber upsert per-day: commit: %w", err)
	}
	return nil
}

// sampleAssignedAt reads UTC_TIMESTAMP(6) inside an active txn.
// Fail-closed: any DB error propagates up so the reservation aborts
// rather than silently using app-host time (which can drift from DB
// time across UTC day boundaries, producing a wrong-bucket insert).
//
// Codex round-8 V1 HIGH: the previous unconditional Go-time fallback
// masked real TiDB failures — a transient `SELECT UTC_TIMESTAMP(6)`
// error near a UTC day boundary would silently substitute app-host
// time, producing a rendered date/prefix that diverged from the DB
// time the rest of the reservation assumed. Under the fail-closed
// policy the reservation aborts and the INSERT is rejected; the
// operator investigates the DB hiccup instead of shipping a
// wrong-bucket record.
func sampleAssignedAt(ctx context.Context, tx *sql.Tx) (time.Time, error) {
	var t time.Time
	err := tx.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&t)
	if err != nil {
		return time.Time{}, fmt.Errorf("autonumber reserve: sample UTC_TIMESTAMP(6): %w", err)
	}
	return t.UTC(), nil
}

// generateCounterRowID produces a deterministic `Id` PK for the
// per-(field_id, scope_key) counter row. The schema has a synthetic
// Id PK + a UNIQUE(FieldId, ScopeKey); this function yields an Id
// string that encodes the unique pair.
//
// Format: "counter::<fieldID>::<scopeKey>" — stable across inserts,
// so ON DUPLICATE KEY UPDATE sees the same row even if the FK UNIQUE
// index is missed by the planner.
func generateCounterRowID(fieldID, scopeKey string) string {
	return "counter::" + fieldID + "::" + scopeKey
}

// tryForceSessionPessimistic is retained as a NO-OP stub for existing
// callers (engine.NewEngine). V2-C2 Round-2 review: setting
// SET SESSION on the *sql.DB hits a random pooled conn that has no
// guaranteed relationship to the conn used for the next BeginTx —
// pessimistic mode is now forced per-conn via beginPessimisticTx
// immediately before BeginTx. This stub is kept so we don't break
// engine.go wiring during this fix; a follow-up cleanup can remove
// both the call and the stub.
func tryForceSessionPessimistic(_ context.Context, _ *sql.DB) {
	// intentionally empty — see beginPessimisticTx.
}