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

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/nexuscrm/automation/autonumber"
	bootstate "github.com/nexuscrm/db/pkg/bootstate"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/autonumberseed"
)

// 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).
//
// Relocated verbatim to nexuscrm-contract/pkg/autonumberseed (Wave-0 SLICE B);
// this alias keeps existing importers compiling unchanged.
type AutoNumberFormatSeed = autonumberseed.AutoNumberFormatSeed

// autonumberSeedBatchSize bounds how many rows are written per UPSERT
// statement. 200 matches the per_day ceiling referenced in PRD §8.2
// step 3.f pagination note.
const autonumberSeedBatchSize = 200

// LegacySource constants — short enum values used by scopeSourceMap
// (per Wave 3 Lane C2 task spec). Pack seed structs may 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.
//
// Relocated verbatim to nexuscrm-contract/pkg/autonumberseed (Wave-0 SLICE B);
// these aliases keep existing importers (and the DB code below) compiling
// unchanged.
const (
	LegacySourceAutoNumber = autonumberseed.LegacySourceAutoNumber
	LegacySourceCompound   = autonumberseed.LegacySourceCompound
	LegacySourceNone       = autonumberseed.LegacySourceNone

	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
}

// 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). Wave 5 relocated the generic
// query to the neutral sibling github.com/nexuscrm/db/pkg/bootstate so the
// relocated controlplane content-compat backfill reaches it without importing
// internal/bootstrap; this unexported wrapper delegates so stay-behind bootstrap
// callers are unchanged. SQL and return shape are identical.
func tableExistsRaw(ctx context.Context, rawDB *sql.DB, table string) (bool, error) {
	return bootstate.TableExistsRaw(ctx, rawDB, table)
}

// 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 stmtPrefix = "INSERT INTO `_System_AutoNumberCounter` " +
		"(`Id`, `FieldId`, `ScopeKey`, `NextValue`, `CreatedDate`, `LastModifiedDate`) " +
		"VALUES "
	const valueTuple = "(?, ?, ?, ?, ?, ?)"
	const stmtSuffix = " ON DUPLICATE KEY UPDATE " +
		"`NextValue` = GREATEST(`NextValue`, VALUES(`NextValue`)), " +
		"`LastModifiedDate` = VALUES(`LastModifiedDate`)"
	for start := 0; start < len(rows); start += autonumberSeedBatchSize {
		end := start + autonumberSeedBatchSize
		if end > len(rows) {
			end = len(rows)
		}
		chunk := rows[start:end]
		stmt := stmtPrefix + strings.TrimSuffix(strings.Repeat(valueTuple+", ", len(chunk)), ", ") + stmtSuffix
		args := make([]any, 0, len(chunk)*6)
		for _, r := range chunk {
			args = append(args, r.ID, r.FieldID, r.ScopeKey, r.NextValue, now, now)
		}
		if _, err := tx.ExecContext(ctx, stmt, args...); err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("autonumber seed flush: upsert rows %d-%d: %w", start, end-1, 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 declared seeds. It converts
// the long table-name form into the short enum.
//
// Relocated verbatim to nexuscrm-contract/pkg/autonumberseed (Wave-0 SLICE B);
// this alias keeps existing importers compiling unchanged.
var BuildScopeSourceMapFromSeeds = autonumberseed.BuildScopeSourceMapFromSeeds