package ports

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
)

// ErrHistoryTrackingConflict classifies stale CAS snapshots and interval
// ledger conflicts for TrackHistory transitions.
var ErrHistoryTrackingConflict = errors.New("history tracking transition conflict")

// ErrHistorySerializationConflict classifies an optimistic commit conflict
// after a transaction reserved a field-history sequence.
var ErrHistorySerializationConflict = errors.New("history serialization conflict")

// ErrOutOfOrderHistorySequenceLock classifies a field-history sequence row
// discovered below the high-water lock already held by the active transaction.
var ErrOutOfOrderHistorySequenceLock = errors.New("out-of-order history sequence lock")

// IsHistorySerializationConflict reports whether err is a typed field-history
// serialization conflict.
func IsHistorySerializationConflict(err error) bool {
	return errors.Is(err, ErrHistorySerializationConflict)
}

type historySequencePrelockPlanner interface {
	HistorySequencePrelockFields() []string
}

// HistorySequencePrelockFields extracts the complete ordered FieldId set that
// a fresh transaction must prelock before replaying work which discovered
// sequence rows in an unsafe order.
func HistorySequencePrelockFields(err error) ([]string, bool) {
	if err == nil {
		return nil, false
	}
	var planner historySequencePrelockPlanner
	if !errors.As(err, &planner) {
		return nil, false
	}
	fields := OrderedHistorySequenceFields(planner.HistorySequencePrelockFields())
	return fields, len(fields) > 0
}

type historyTrackingAttemptStateKey struct{}

type historyTrackingAttemptState struct {
	attempted bool
}

// WithHistoryTrackingAttemptState installs transaction-local mutable state so
// the transaction owner can classify a commit-time write conflict after the
// repository callback has returned.
func WithHistoryTrackingAttemptState(ctx context.Context) context.Context {
	return context.WithValue(ctx, historyTrackingAttemptStateKey{}, &historyTrackingAttemptState{})
}

// MarkHistoryTrackingAttempt records that this transaction performed a
// TrackHistory CAS write.
func MarkHistoryTrackingAttempt(ctx context.Context) {
	state, _ := ctx.Value(historyTrackingAttemptStateKey{}).(*historyTrackingAttemptState)
	if state != nil {
		state.attempted = true
	}
}

// IsHistoryTrackingAttempt reports whether a CAS write ran in this transaction.
func IsHistoryTrackingAttempt(ctx context.Context) bool {
	state, _ := ctx.Value(historyTrackingAttemptStateKey{}).(*historyTrackingAttemptState)
	return state != nil && state.attempted
}

type historySerializationAttemptStateKey struct{}
type historySequencePrelockRetryOwnerKey struct{}
type initialHistorySequenceLocksKey struct{}
type initialHistoryBusinessRowObjectsKey struct{}
type plannedHistorySequenceLocksKey struct{}
type historySequenceRetryPlanKey struct{}

// HistorySequenceRetryPlan is the root-transaction history lock knowledge
// accumulated across fresh transaction-manager invocations. Records may need
// to restart its outer transaction after discovering a parent row; retaining
// this plan prevents that outer restart from forgetting history fields learned
// by the inner transaction manager.
type HistorySequenceRetryPlan struct {
	mu     sync.Mutex
	fields []string
}

type plannedHistorySequenceLocks struct {
	mu       sync.Mutex
	fields   []string
	consumed bool
}

type historySerializationAttemptState struct {
	mu        sync.Mutex
	attempted bool
	held      map[string]struct{}
	high      string
}

// EnsureHistorySequenceRetryPlan preserves an enclosing root retry plan or
// installs one seeded from the caller's immutable first-attempt plan.
func EnsureHistorySequenceRetryPlan(ctx context.Context) context.Context {
	if plan, _ := ctx.Value(historySequenceRetryPlanKey{}).(*HistorySequenceRetryPlan); plan != nil {
		return ctx
	}
	return context.WithValue(ctx, historySequenceRetryPlanKey{}, &HistorySequenceRetryPlan{
		fields: InitialHistorySequenceLocks(ctx),
	})
}

// HistorySequenceRetryPlanFields returns the complete ordered history fields
// learned by the active root retry owner.
func HistorySequenceRetryPlanFields(ctx context.Context) []string {
	plan, _ := ctx.Value(historySequenceRetryPlanKey{}).(*HistorySequenceRetryPlan)
	if plan == nil {
		return InitialHistorySequenceLocks(ctx)
	}
	plan.mu.Lock()
	defer plan.mu.Unlock()
	return append([]string(nil), plan.fields...)
}

// AugmentHistorySequenceRetryPlan merges fields learned by a rolled-back
// attempt into the root plan. It returns the resulting ordered snapshot and
// whether the plan grew.
func AugmentHistorySequenceRetryPlan(ctx context.Context, fieldIDs []string) ([]string, bool) {
	plan, _ := ctx.Value(historySequenceRetryPlanKey{}).(*HistorySequenceRetryPlan)
	if plan == nil {
		return OrderedHistorySequenceFields(append(InitialHistorySequenceLocks(ctx), fieldIDs...)), false
	}
	plan.mu.Lock()
	defer plan.mu.Unlock()
	merged := OrderedHistorySequenceFields(append(plan.fields, fieldIDs...))
	changed := len(merged) != len(plan.fields)
	plan.fields = merged
	return append([]string(nil), plan.fields...), changed
}

// WithoutHistorySequenceRetryPlan prevents detached work from sharing mutable
// retry knowledge with the transaction it left.
func WithoutHistorySequenceRetryPlan(ctx context.Context) context.Context {
	return context.WithValue(ctx, historySequenceRetryPlanKey{}, (*HistorySequenceRetryPlan)(nil))
}

// WithHistorySerializationAttemptState installs transaction-local state used
// to classify commit-time conflicts after sequence reservation.
func WithHistorySerializationAttemptState(ctx context.Context) context.Context {
	return context.WithValue(ctx, historySerializationAttemptStateKey{}, &historySerializationAttemptState{})
}

// EnsureHistorySerializationAttemptState preserves an existing root
// transaction coordinator or installs one for a schema operation which owns
// its active transaction scope directly.
func EnsureHistorySerializationAttemptState(ctx context.Context) context.Context {
	state, _ := ctx.Value(historySerializationAttemptStateKey{}).(*historySerializationAttemptState)
	if state != nil {
		return ctx
	}
	return WithHistorySerializationAttemptState(ctx)
}

// WithHistorySequencePrelockRetryOwner marks a root transaction callback that
// can safely roll back and replay after late field discovery. Nested work
// inherits the marker, while a direct non-retrying transaction must not gain
// replay semantics it does not own.
func WithHistorySequencePrelockRetryOwner(ctx context.Context) context.Context {
	return context.WithValue(ctx, historySequencePrelockRetryOwnerKey{}, true)
}

// WithoutHistorySequencePrelockRetryOwner clears inherited replay ownership
// when work detaches from its enclosing transaction.
func WithoutHistorySequencePrelockRetryOwner(ctx context.Context) context.Context {
	return context.WithValue(ctx, historySequencePrelockRetryOwnerKey{}, false)
}

// IsHistorySequencePrelockRetryOwner reports whether the active root callback
// owns whole-transaction replay for a history-sequence prelock plan.
func IsHistorySequencePrelockRetryOwner(ctx context.Context) bool {
	owned, _ := ctx.Value(historySequencePrelockRetryOwnerKey{}).(bool)
	return owned
}

// WithInitialHistorySequenceLocks carries a complete caller-planned field set
// into the first root transaction attempt. Before the first history reservation,
// the caller must lock every existing business row whose mutation can reserve a
// declared field. The transaction owner copies this immutable seed into
// attempt-local planned locks; ReserveHistorySequences still acquires them only
// at the first actual reservation.
func WithInitialHistorySequenceLocks(ctx context.Context, fieldIDs []string) context.Context {
	fields := OrderedHistorySequenceFields(fieldIDs)
	if len(fields) == 0 {
		return ctx
	}
	return context.WithValue(ctx, initialHistorySequenceLocksKey{}, fields)
}

// InitialHistorySequenceLocks returns a defensive copy of the caller-planned
// first-attempt field set.
func InitialHistorySequenceLocks(ctx context.Context) []string {
	fields, _ := ctx.Value(initialHistorySequenceLocksKey{}).([]string)
	return append([]string(nil), fields...)
}

// WithInitialHistoryBusinessRowObjects identifies object types for which every
// existing row mutated by the initial sequence plan must already be held by the
// transaction's parent-lock coordinator.
func WithInitialHistoryBusinessRowObjects(ctx context.Context, objectAPIName []string) context.Context {
	unique := make(map[string]struct{}, len(objectAPIName))
	for _, name := range objectAPIName {
		if name = strings.ToLower(strings.TrimSpace(name)); name != "" {
			unique[name] = struct{}{}
		}
	}
	objects := make([]string, 0, len(unique))
	for name := range unique {
		objects = append(objects, name)
	}
	sort.Strings(objects)
	if len(objects) == 0 {
		return ctx
	}
	return context.WithValue(ctx, initialHistoryBusinessRowObjectsKey{}, objects)
}

// RequiresInitialHistoryBusinessRowPrelock reports whether objectAPIName is
// covered by the active first-attempt history plan.
func RequiresInitialHistoryBusinessRowPrelock(ctx context.Context, objectAPIName string) bool {
	objects, _ := ctx.Value(initialHistoryBusinessRowObjectsKey{}).([]string)
	objectAPIName = strings.ToLower(strings.TrimSpace(objectAPIName))
	index := sort.SearchStrings(objects, objectAPIName)
	return objectAPIName != "" && index < len(objects) && objects[index] == objectAPIName
}

// WithoutInitialHistoryBusinessRowObjects prevents detached work from
// inheriting the enclosing Flow's business-row coverage contract.
func WithoutInitialHistoryBusinessRowObjects(ctx context.Context) context.Context {
	return context.WithValue(ctx, initialHistoryBusinessRowObjectsKey{}, []string(nil))
}

// WithoutInitialHistorySequenceLocks prevents a detached transaction from
// inheriting an enclosing operation's first-attempt plan.
func WithoutInitialHistorySequenceLocks(ctx context.Context) context.Context {
	return context.WithValue(ctx, initialHistorySequenceLocksKey{}, []string(nil))
}

// WithPlannedHistorySequenceLocks carries the complete field-history sequence
// set discovered by a failed attempt into one fresh retry. The retry must not
// acquire these rows at transaction entry: record writes acquire their business
// rows first, and a history-first retry can deadlock with that normal path.
// ReserveHistorySequences consumes the plan at the first actual reservation,
// preserving business-row -> history-sequence order while still prelocking the
// complete field set in deterministic order.
func WithPlannedHistorySequenceLocks(ctx context.Context, fieldIDs []string) context.Context {
	fields := OrderedHistorySequenceFields(fieldIDs)
	return context.WithValue(ctx, plannedHistorySequenceLocksKey{}, &plannedHistorySequenceLocks{fields: fields})
}

// AugmentPlannedHistorySequenceLocks adds metadata-derived fields to the
// fresh attempt's history plan before its first sequence reservation. Records
// uses this while it is already discovering and locking the complete rollup
// parent closure, so nested derived writes do not discover lower-sorting
// history rows only after the transaction has begun mutating data.
//
// The function returns false when no retry owner installed attempt-local plan
// state or when the first reservation already consumed that plan. In that
// case the existing out-of-order detector remains the safe replay fallback.
func AugmentPlannedHistorySequenceLocks(ctx context.Context, fieldIDs []string) bool {
	added := OrderedHistorySequenceFields(fieldIDs)
	state, _ := ctx.Value(plannedHistorySequenceLocksKey{}).(*plannedHistorySequenceLocks)
	if state == nil {
		return false
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	if state.consumed {
		return false
	}
	if len(added) == 0 {
		return true
	}
	state.fields = OrderedHistorySequenceFields(append(state.fields, added...))
	return true
}

// PlannedHistorySequenceLocks returns the retry plan until it has been
// consumed by a successful no-increment prelock.
func PlannedHistorySequenceLocks(ctx context.Context) []string {
	state, _ := ctx.Value(plannedHistorySequenceLocksKey{}).(*plannedHistorySequenceLocks)
	if state == nil {
		return nil
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	if state.consumed {
		return nil
	}
	return append([]string(nil), state.fields...)
}

// MarkPlannedHistorySequenceLocksConsumed records that the retry's complete
// plan was locked successfully. A failed prelock intentionally remains
// unconsumed so the enclosing transaction retry gets a fresh plan state.
func MarkPlannedHistorySequenceLocksConsumed(ctx context.Context) {
	state, _ := ctx.Value(plannedHistorySequenceLocksKey{}).(*plannedHistorySequenceLocks)
	if state == nil {
		return
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	state.consumed = true
}

// WithoutPlannedHistorySequenceLocks prevents a detached transaction from
// inheriting retry-only lock planning from its parent transaction.
func WithoutPlannedHistorySequenceLocks(ctx context.Context) context.Context {
	return context.WithValue(ctx, plannedHistorySequenceLocksKey{}, (*plannedHistorySequenceLocks)(nil))
}

// MarkHistorySerializationAttempt records that this transaction reserved at
// least one field-history sequence.
func MarkHistorySerializationAttempt(ctx context.Context) {
	state, _ := ctx.Value(historySerializationAttemptStateKey{}).(*historySerializationAttemptState)
	if state != nil {
		state.mu.Lock()
		defer state.mu.Unlock()
		state.attempted = true
	}
}

// IsHistorySerializationAttempt reports whether sequence reservation ran in
// this transaction.
func IsHistorySerializationAttempt(ctx context.Context) bool {
	state, _ := ctx.Value(historySerializationAttemptStateKey{}).(*historySerializationAttemptState)
	if state == nil {
		return false
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	return state.attempted
}

// OrderedHistorySequenceFields normalizes, deduplicates, and sorts stable
// FieldIds for deterministic row-lock planning.
func OrderedHistorySequenceFields(fieldIDs []string) []string {
	unique := make(map[string]struct{}, len(fieldIDs))
	for _, fieldID := range fieldIDs {
		fieldID = strings.TrimSpace(fieldID)
		if fieldID != "" {
			unique[fieldID] = struct{}{}
		}
	}
	ordered := make([]string, 0, len(unique))
	for fieldID := range unique {
		ordered = append(ordered, fieldID)
	}
	sort.Strings(ordered)
	return ordered
}

// PlanHistorySequenceLocks rejects a newly discovered field that sorts below
// the transaction's held high-water mark when the root owns whole-transaction
// replay. A non-retrying owner retains the legacy per-call order because it
// cannot safely consume a retry plan.
func PlanHistorySequenceLocks(ctx context.Context, fieldIDs []string) ([]string, error) {
	ordered := OrderedHistorySequenceFields(fieldIDs)
	state, _ := ctx.Value(historySerializationAttemptStateKey{}).(*historySerializationAttemptState)
	if state == nil {
		return nil, fmt.Errorf("history sequence lock planning requires transaction-local state")
	}
	state.mu.Lock()
	defer state.mu.Unlock()

	toAcquire := make([]string, 0, len(ordered))
	for _, fieldID := range ordered {
		if _, ok := state.held[fieldID]; ok {
			continue
		}
		if state.high != "" && fieldID < state.high && IsHistorySequencePrelockRetryOwner(ctx) {
			return nil, fmt.Errorf("%w: field %s sorts below held high-water %s", ErrOutOfOrderHistorySequenceLock, fieldID, state.high)
		}
		toAcquire = append(toAcquire, fieldID)
	}
	return toAcquire, nil
}

// MarkHistorySequenceLockAcquired records a successfully acquired sequence row
// lock in the active transaction scope.
func MarkHistorySequenceLockAcquired(ctx context.Context, fieldID string) {
	fieldID = strings.TrimSpace(fieldID)
	state, _ := ctx.Value(historySerializationAttemptStateKey{}).(*historySerializationAttemptState)
	if state == nil || fieldID == "" {
		return
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	if state.held == nil {
		state.held = make(map[string]struct{})
	}
	state.held[fieldID] = struct{}{}
	if fieldID > state.high {
		state.high = fieldID
	}
}

// HeldHistorySequenceLockFields returns the ordered FieldIds whose sequence
// rows are already locked by the active transaction.
func HeldHistorySequenceLockFields(ctx context.Context) []string {
	state, _ := ctx.Value(historySerializationAttemptStateKey{}).(*historySerializationAttemptState)
	if state == nil {
		return nil
	}
	state.mu.Lock()
	defer state.mu.Unlock()
	held := make([]string, 0, len(state.held))
	for fieldID := range state.held {
		held = append(held, fieldID)
	}
	sort.Strings(held)
	return held
}