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

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"github.com/nexuscrm/shared/pkg/models"
)

// errNotImplemented is retained for tests that asserted the Wave 1
// stub state. Wave 2 implementations do not return it anymore.
var errNotImplemented = errors.New("autonumber: not implemented")

// MetadataProvider is the engine's view of autonumber-relevant
// metadata. Keep the surface tiny so a map-backed fake is trivial to
// write.
type MetadataProvider interface {
	// GetRule returns the Rule configured for fieldID, or nil if no
	// such field / the field has no autonumber rule.
	GetRule(fieldID string) *Rule
	// ListRules returns all autonumber-configured rules. Used at
	// engine init to populate caches + pre-parse + quarantine fail
	// cases.
	ListRules() []Rule
}

// Engine renders autonumber values for records being INSERTed.
//
// See PRD §3.1 for the contract. All methods run INSIDE the caller's
// outer transaction. The engine opens its own short-lived reservation
// connection for counter increments and its own read connection for
// `fresh_read` cross-record tokens — neither of those share the
// caller txn (§5.0 dependencies).
type Engine interface {
	// AssignOne renders a single autonumber for (fieldID, sourceRecord).
	AssignOne(ctx context.Context, fieldID string, sourceRecord models.SObject) (string, error)

	// AssignBatch renders N contiguous autonumbers for one field over
	// N source records. The engine groups by scope_key + cross-record
	// resolution fingerprint so each group obtains a single reservation
	// (PRD §5.3).
	AssignBatch(ctx context.Context, fieldID string, sourceRecords []models.SObject) ([]string, error)

	// SeedFieldCounter initializes the `_System_AutoNumberCounter` row
	// for a newly-configured field. Used by the cutover seed runner
	// (PRD §8.2) and by the admin field-create rewire (§7.2 step k,
	// fixpack F6 — replaces UpsertAutoNumber).
	SeedFieldCounter(ctx context.Context, fieldID string) error

	// BootstrapField is the engine-bootstrap path invoked from the
	// rewired `field_create.go:192` (PRD §3.3 / §7.2 step k). Registers
	// a new rule in the metadata cache and ensures a counter row
	// exists for `global` scopes.
	BootstrapField(ctx context.Context, rule Rule) error

	// Quarantine returns the engine's current quarantine registry for
	// test + admin-tooling consumers. Stable reference.
	Quarantine() *QuarantineRegistry

	// ReconcileQuarantine re-runs the init-time parse sweep against
	// the current metadata snapshot and updates the quarantine
	// registry. Called by the admin refresh-metadata endpoint so a
	// live format fix takes effect without a process restart.
	ReconcileQuarantine(ctx context.Context)
}

// engine is the concrete Engine implementation.
type engine struct {
	metadata  MetadataProvider
	reserver  *reserver
	resolver  CrossRecordResolver
	quarantine *QuarantineRegistry
	logger    *slog.Logger

	// reservationDB is retained for SeedFieldCounter / BootstrapField
	// which need direct SQL access outside the reservation loop.
	reservationDB *sql.DB
}

// NewEngine constructs a production Engine implementation.
//
// Required opts:
//   - WithReservationDB — dedicated *sql.DB for counter reservation txns
//   - WithMetadataProvider — rule lookup
//
// Optional:
//   - WithCrossRecordResolver — required only when any configured
//     rule has CrossRecordRefs; if absent AND a token is encountered,
//     AssignOne returns ErrChainResolverFailure (engine wiring error,
//     mapped to 500 InternalError downstream).
//   - WithLogger — slog-style logger for quarantine events; defaults
//     to slog.Default.
//
// (V1-H1 Round-2 review: `WithFallbackDB` was documented here as an
// option but never implemented. The fallback FOR UPDATE lane shares
// the reservation DB handle today; introducing an independent handle
// is a post-cutover optimization, not a v1 requirement.)
//
// NewEngine runs field-rule pre-parse on every rule from
// metadata.ListRules() and quarantines any that fail — fixpack F9:
// engine init MUST NOT crash on parse errors.
func NewEngine(opts ...Option) (Engine, error) {
	e := &engine{
		quarantine: NewQuarantineRegistry(),
		logger:     slog.Default(),
	}
	for _, opt := range opts {
		opt(e)
	}
	if e.reservationDB == nil {
		return nil, fmt.Errorf("autonumber: NewEngine: WithReservationDB is required")
	}
	if e.metadata == nil {
		return nil, fmt.Errorf("autonumber: NewEngine: WithMetadataProvider is required")
	}

	// F11: set SESSION-level pessimistic mode on the DB for
	// TiDB-backed deployments. Non-fatal on other engines.
	tryForceSessionPessimistic(context.Background(), e.reservationDB)

	// Pre-parse + quarantine init (F9).
	rules := e.metadata.ListRules()
	e.reserver = newReserver(e.reservationDB, e.reservationDB, len(rules), newReservationMetrics())

	for _, rule := range rules {
		_, _, err := Parse(rule.Format)
		if err != nil {
			reason := QuarantineReasonFormatFunc("parse", err)
			e.quarantine.Quarantine(rule.FieldID, reason)
			emitQuarantine(rule.FieldID)
			e.logger.Warn("autonumber: quarantining field on init",
				"fieldID", rule.FieldID,
				"reason", reason,
				"format", rule.Format,
			)
		}
	}
	return e, nil
}

// Option is the functional-option type used by NewEngine.
type Option func(*engine)

// WithReservationDB wires the *sql.DB used for counter-reservation
// txns. REQUIRED.
func WithReservationDB(db *sql.DB) Option {
	return func(e *engine) { e.reservationDB = db }
}

// WithMetadataProvider wires the rule lookup. REQUIRED.
func WithMetadataProvider(m MetadataProvider) Option {
	return func(e *engine) { e.metadata = m }
}

// WithCrossRecordResolver wires the cross-record lookup
// implementation. Optional — only required when a rule references
// {Object.Field} tokens.
func WithCrossRecordResolver(r CrossRecordResolver) Option {
	return func(e *engine) { e.resolver = r }
}

// WithLogger overrides the slog logger used for quarantine + error
// reporting.
func WithLogger(l *slog.Logger) Option {
	return func(e *engine) {
		if l != nil {
			e.logger = l
		}
	}
}

// AssignOne — PRD §5.1.
func (e *engine) AssignOne(ctx context.Context, fieldID string, sourceRecord models.SObject) (string, error) {
	// 1. Metadata lookup.
	rule := e.metadata.GetRule(fieldID)
	if rule == nil {
		return "", fmt.Errorf("autonumber: AssignOne: unknown fieldID %q", fieldID)
	}
	if rule.Format == "" {
		return "", fmt.Errorf("autonumber: AssignOne: field %q has no auto_number_format", fieldID)
	}

	// Quarantine gate — fixpack F9.
	if qerr, ok := buildQuarantineErrorFromRegistry(e.quarantine, fieldID); ok {
		return "", qerr
	}

	// 2. Parse (cached).
	spec, refs, err := Parse(rule.Format)
	if err != nil {
		return "", fmt.Errorf("autonumber: AssignOne: parse %q: %w", rule.Format, err)
	}

	// Merge parsed refs with any pre-stored TokenRefs from metadata.
	// Metadata-side TokenRefs carry FkChains + directField + strategy
	// configured via data-spec; parser-derived refs are the fallback.
	activeRefs := rule.TokenRefs
	if len(activeRefs) == 0 {
		activeRefs = refs
	}

	// 3. Resolve cross-record tokens.
	crossValues, err := resolveCrossRecords(ctx, e.resolver, sourceRecord, activeRefs)
	if err != nil {
		return "", fmt.Errorf("autonumber: AssignOne: %w", err)
	}

	// 4. Reserve counter.
	scope := rule.Scope
	if scope == "" {
		scope = ScopeGlobal
	}
	start := rule.Start
	if start <= 0 {
		start = 1
	}
	reserveIn := reserveInput{
		fieldID:    fieldID,
		scope:      scope,
		startValue: start,
	}
	// ScopePerPrefix: the scope key is the rendered non-counter
	// portion of the format. We pass a closure rather than a pre-
	// computed string so the reserver renders against the SAME
	// `assignedAt` used for the final identifier rendering. This
	// prevents a day-boundary race where scope_key (Go time.Now()
	// rendering) and final Name (DB UTC_TIMESTAMP sample) disagree on
	// the date component — which would let two inserts land different
	// counter buckets yet render the same Name.
	//
	// liveMaxFn closes the mid-seed / new-bucket race (Codex round-3
	// CRITICAL follow-up). When the reserver lazy-creates a counter
	// row for a per_prefix bucket, it consults liveMaxFn to absorb
	// any existing live-table records in that bucket so the initial
	// NextValue is `max(startValue, live_max+1)` rather than a
	// blind `startValue=1`. This makes post-seed drift impossible by
	// construction — first-insert into a bucket always wins against
	// prior records, regardless of whether the seed-time scan saw
	// them.
	if scope == ScopePerPrefix {
		reserveIn.scopeKeyFn = func(assignedAt time.Time) (string, error) {
			return RenderScopePrefix(spec, assignedAt, crossValues)
		}
		if rule.ObjectAPIName != "" && rule.FieldAPIName != "" && e.reservationDB != nil {
			objectAPI := rule.ObjectAPIName
			fieldAPI := rule.FieldAPIName
			reserveIn.liveMaxFn = func(ctx context.Context, assignedAt time.Time) (int64, error) {
				before, after, err := renderScopePrefixParts(spec, assignedAt, crossValues)
				if err != nil {
					return 0, fmt.Errorf("autonumber: AssignOne live_max: %w", err)
				}
				return queryLiveMaxForPrefix(ctx, e.reservationDB, objectAPI, fieldAPI, before, after)
			}
		}
	}
	assignedAt, counterValue, err := e.reserver.reserveOne(ctx, reserveIn)
	if err != nil {
		return "", fmt.Errorf("autonumber: AssignOne: %w", err)
	}

	// 5. Render.
	rendered, err := Render(spec, assignedAt, counterValue, crossValues)
	if err != nil {
		return "", fmt.Errorf("autonumber: AssignOne: render: %w", err)
	}
	return rendered, nil
}

// AssignBatch — PRD §5.3.
//
// Bucketing: group records by (scope_key, cross-record-fingerprint).
// Same-bucket records share ONE reservation of size = |bucket|.
// Different-bucket records get their own reservations — fixpack F13
// ensures mixed buckets never degrade to N× reserveOne.
func (e *engine) AssignBatch(ctx context.Context, fieldID string, sourceRecords []models.SObject) ([]string, error) {
	if len(sourceRecords) == 0 {
		return nil, nil
	}
	rule := e.metadata.GetRule(fieldID)
	if rule == nil {
		return nil, fmt.Errorf("autonumber: AssignBatch: unknown fieldID %q", fieldID)
	}
	if rule.Format == "" {
		return nil, fmt.Errorf("autonumber: AssignBatch: field %q has no auto_number_format", fieldID)
	}
	if qerr, ok := buildQuarantineErrorFromRegistry(e.quarantine, fieldID); ok {
		return nil, qerr
	}
	spec, refs, err := Parse(rule.Format)
	if err != nil {
		return nil, fmt.Errorf("autonumber: AssignBatch: parse: %w", err)
	}
	activeRefs := rule.TokenRefs
	if len(activeRefs) == 0 {
		activeRefs = refs
	}

	// Batch cross-record resolve.
	crossPerRecord, err := resolveCrossRecordsBatch(ctx, e.resolver, sourceRecords, activeRefs)
	if err != nil {
		return nil, fmt.Errorf("autonumber: AssignBatch: %w", err)
	}

	scope := rule.Scope
	if scope == "" {
		scope = ScopeGlobal
	}
	start := rule.Start
	if start <= 0 {
		start = 1
	}

	// Bucket by cross-record fingerprint. For scope=global across the
	// same process invocation the scope_key will be uniform ("") so
	// the outer group is just the cross fingerprint. For scope=per_day
	// the scope_key depends on assignedAt which is not yet known; we
	// cannot pre-split by scope_key (all same-day records share it).
	// The reserver loop handles the per_day assignedAt internally.
	type bucketKey struct {
		fingerprint string
	}
	type bucket struct {
		indices []int
	}
	buckets := make(map[bucketKey]*bucket)
	keyOrder := make([]bucketKey, 0)
	for i, cross := range crossPerRecord {
		fp := crossValueFingerprint(cross, activeRefs)
		bk := bucketKey{fingerprint: fp}
		b, ok := buckets[bk]
		if !ok {
			b = &bucket{}
			buckets[bk] = b
			keyOrder = append(keyOrder, bk)
		}
		b.indices = append(b.indices, i)
	}

	out := make([]string, len(sourceRecords))
	for _, bk := range keyOrder {
		b := buckets[bk]
		reserveIn := reserveInput{
			fieldID:    fieldID,
			scope:      scope,
			startValue: start,
		}
		if scope == ScopePerPrefix {
			// Records in the same bucket share crossValues by
			// construction (fingerprint-keyed), so the closure can
			// render once against the first record's cross values and
			// cover the whole bucket. The reserver invokes the closure
			// with its in-txn `assignedAt` so the scope_key's date
			// component lines up with the per-record Render below that
			// also uses that same assignedAt.
			firstIdx := b.indices[0]
			bucketCross := crossPerRecord[firstIdx]
			reserveIn.scopeKeyFn = func(assignedAt time.Time) (string, error) {
				return RenderScopePrefix(spec, assignedAt, bucketCross)
			}
			// liveMaxFn: see AssignOne for rationale (race-closure
			// at lazy-init). Uses the same bucketCross as scopeKeyFn.
			if rule.ObjectAPIName != "" && rule.FieldAPIName != "" && e.reservationDB != nil {
				objectAPI := rule.ObjectAPIName
				fieldAPI := rule.FieldAPIName
				reserveIn.liveMaxFn = func(ctx context.Context, assignedAt time.Time) (int64, error) {
					before, after, err := renderScopePrefixParts(spec, assignedAt, bucketCross)
					if err != nil {
						return 0, fmt.Errorf("autonumber: AssignBatch live_max: %w", err)
					}
					return queryLiveMaxForPrefix(ctx, e.reservationDB, objectAPI, fieldAPI, before, after)
				}
			}
		}
		assignedAt, startVal, err := e.reserver.reserveBatch(ctx, reserveIn, len(b.indices))
		if err != nil {
			return nil, fmt.Errorf("autonumber: AssignBatch: %w", err)
		}
		for i, idx := range b.indices {
			counter := startVal + int64(i)
			rendered, err := Render(spec, assignedAt, counter, crossPerRecord[idx])
			if err != nil {
				return nil, fmt.Errorf("autonumber: AssignBatch: render idx=%d: %w", idx, err)
			}
			out[idx] = rendered
		}
	}
	return out, nil
}

// crossValueFingerprint produces a deterministic string fingerprint of
// a record's resolved cross-record values so AssignBatch bucketing is
// scope-safe. Records with identical token values share a bucket.
func crossValueFingerprint(cross map[string]any, refs []CrossRecordRef) string {
	if len(refs) == 0 || len(cross) == 0 {
		return ""
	}
	var buf []byte
	// Canonicalize over refs in declared order (they are already
	// stable per metadata).
	for _, ref := range refs {
		buf = append(buf, ref.TokenLiteral...)
		buf = append(buf, '\x00')
		if v, ok := cross[ref.TokenLiteral]; ok {
			switch t := v.(type) {
			case string:
				buf = append(buf, t...)
			case []byte:
				buf = append(buf, t...)
			default:
				buf = append(buf, fmt.Sprintf("%v", t)...)
			}
		}
		buf = append(buf, '\x01')
	}
	return string(buf)
}

// SeedFieldCounter UPSERTs the initial counter row for (fieldID, '').
// Global-scope only; per_day creates rows lazily on first-insert.
// Replaces the deleted repo.UpsertAutoNumber (fixpack F6).
func (e *engine) SeedFieldCounter(ctx context.Context, fieldID string) error {
	rule := e.metadata.GetRule(fieldID)
	if rule == nil {
		return fmt.Errorf("autonumber: SeedFieldCounter: unknown fieldID %q", fieldID)
	}
	scope := rule.Scope
	if scope == "" {
		scope = ScopeGlobal
	}
	if scope != ScopeGlobal {
		// Per-day is lazy; nothing to seed at field create time.
		return nil
	}
	start := rule.Start
	if start <= 0 {
		start = 1
	}
	rowID := generateCounterRowID(fieldID, "")
	_, err := e.reservationDB.ExecContext(ctx,
		`INSERT INTO `+"`_System_AutoNumberCounter`"+`
           (`+"`Id`"+`,`+"`FieldId`"+`,`+"`ScopeKey`"+`,`+"`NextValue`"+`,`+"`CreatedDate`"+`,`+"`LastModifiedDate`"+`)
         VALUES (?, ?, '', ?, UTC_TIMESTAMP(6), UTC_TIMESTAMP(6))
         ON DUPLICATE KEY UPDATE `+"`NextValue`"+` = GREATEST(`+"`NextValue`"+`, VALUES(`+"`NextValue`"+`))`,
		rowID, fieldID, start,
	)
	if err != nil {
		return fmt.Errorf("autonumber: SeedFieldCounter %q: %w", fieldID, err)
	}
	return nil
}

// BootstrapField runs parse validation + SeedFieldCounter for a newly
// created (admin) or newly configured (seed runner) rule.
//
// Per fixpack F9 the parse failure does NOT crash — it quarantines
// the field and returns the parse error so the caller can surface it
// in UI/seed logs, but the engine stays up.
func (e *engine) BootstrapField(ctx context.Context, rule Rule) error {
	if rule.FieldID == "" {
		return fmt.Errorf("autonumber: BootstrapField: FieldID is empty")
	}
	if rule.Format == "" {
		return fmt.Errorf("autonumber: BootstrapField: Format is empty")
	}
	if _, _, err := Parse(rule.Format); err != nil {
		reason := QuarantineReasonFormatFunc("parse", err)
		e.quarantine.Quarantine(rule.FieldID, reason)
		emitQuarantine(rule.FieldID)
		e.logger.Warn("autonumber: BootstrapField quarantine",
			"fieldID", rule.FieldID,
			"reason", reason,
		)
		return fmt.Errorf("autonumber: BootstrapField: %w", err)
	}
	// Clear any stale quarantine entry (F9 subtlety — a prior init
	// may have flagged this fieldID before the format was fixed).
	e.quarantine.Clear(rule.FieldID)
	emitUnquarantine(rule.FieldID)

	return e.SeedFieldCounter(ctx, rule.FieldID)
}

// Quarantine returns the engine's registry for test + admin-tooling
// consumers.
func (e *engine) Quarantine() *QuarantineRegistry { return e.quarantine }

// ReconcileQuarantine re-runs the init-time parse sweep (fixpack F9)
// against the CURRENT metadata snapshot and updates the quarantine
// registry accordingly:
//   - fields whose format newly fails Parse → quarantine
//   - fields whose format now parses cleanly → un-quarantine
//
// Used by the admin refresh endpoint to keep quarantine state aligned
// with metadata after a live reload (Codex round-8 V2 HIGH): previously
// a field fixed via `cmd/setup apply` + `/admin/autonumber/refresh-
// metadata` would reload rules but leave the startup quarantine
// sticky until process restart; and a newly-malformed rule loaded
// mid-process would crash on first insert instead of quarantining.
func (e *engine) ReconcileQuarantine(_ context.Context) {
	if e.metadata == nil || e.quarantine == nil {
		return
	}
	rules := e.metadata.ListRules()
	seen := make(map[string]struct{}, len(rules))
	for _, rule := range rules {
		seen[rule.FieldID] = struct{}{}
		_, _, err := Parse(rule.Format)
		alreadyQ, _ := e.quarantine.IsQuarantined(rule.FieldID)
		if err != nil {
			if !alreadyQ {
				reason := QuarantineReasonFormatFunc("parse", err)
				e.quarantine.Quarantine(rule.FieldID, reason)
				emitQuarantine(rule.FieldID)
				e.logger.Warn("autonumber: quarantining field on refresh",
					"fieldID", rule.FieldID,
					"reason", reason,
					"format", rule.Format,
				)
			}
			continue
		}
		if alreadyQ {
			e.quarantine.Clear(rule.FieldID)
			emitUnquarantine(rule.FieldID)
			e.logger.Info("autonumber: clearing quarantine on refresh",
				"fieldID", rule.FieldID,
				"format", rule.Format,
			)
		}
	}
	// Fields removed from metadata between the last run and this one
	// should not stay in quarantine indefinitely. If a field is no
	// longer in the snapshot, drop its quarantine entry so future
	// re-configs start from a clean state.
	for fieldID := range e.quarantine.List() {
		if _, ok := seen[fieldID]; !ok {
			e.quarantine.Clear(fieldID)
			emitUnquarantine(fieldID)
		}
	}
}