package recordlocks

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

var (
	// ErrMissingParentLockScope means a caller tried to acquire parent locks
	// without installing a transaction-scoped coordinator.
	ErrMissingParentLockScope = errors.New("parent lock coordinator scope is missing")
	// ErrOutOfOrderParentLock means a nested/incremental request discovered a
	// parent key that sorts before a key already held by the transaction.
	ErrOutOfOrderParentLock = errors.New("parent lock request is out of transaction-global order")
)

// ParentKey identifies a parent row that must be locked before child-row
// mutations that depend on it. Object names are compared case-insensitively,
// while record IDs are compared exactly.
type ParentKey struct {
	ObjectName string
	RecordID   string
}

type scopeKey struct{}
type forcedPrelocksKey struct{}

// RetryablePrelockError marks errors that carry additional parent-lock keys
// the caller must acquire before retrying the whole transaction attempt.
type RetryablePrelockError interface {
	error
	ParentPrelockKeys() []ParentKey
}

// Scope tracks parent rows already held by one transaction/coordinator run.
// Held rows include successfully acquired parent locks and rows inserted by
// this transaction, which the transaction owns without a separate row-lock
// read. The scope lets nested callers re-enter with duplicate keys while
// failing closed if a newly discovered key would need to be acquired before a
// key the transaction already holds.
type Scope struct {
	mu         sync.Mutex
	txIdentity interface{}
	held       map[string]ParentKey
	order      []ParentKey
	acquired   []ParentKey
	highID     string
}

// WithScope installs a fresh transaction-scoped parent lock coordinator.
// Call this inside the actual SQL transaction attempt. It intentionally
// replaces any inherited scope so retry/detached transactions cannot reuse
// lock state from a different SQL transaction.
func WithScope(ctx context.Context) context.Context {
	return context.WithValue(ctx, scopeKey{}, &Scope{held: map[string]ParentKey{}})
}

// WithoutScope masks any inherited transaction-scoped parent lock state.
// Fresh or detached SQL transaction boundaries must call this before they
// expose a context to work that may install its own coordinator scope.
func WithoutScope(ctx context.Context) context.Context {
	return context.WithValue(ctx, scopeKey{}, (*Scope)(nil))
}

// WithExistingScope installs an existing transaction-scoped parent lock
// coordinator. Callers that cache transaction-lifetime scope state use this to
// reattach the same scope to later callbacks that received only the transaction
// context.
func WithExistingScope(ctx context.Context, scope *Scope) context.Context {
	if scope == nil {
		return WithScope(ctx)
	}
	return context.WithValue(ctx, scopeKey{}, scope)
}

// WithScopeForTransaction installs or reuses a coordinator scope for one
// transaction identity. Nested callers in the same SQL transaction reuse the
// existing scope; retry/detached transactions receive a fresh scope because
// their transaction identity changes.
func WithScopeForTransaction(ctx context.Context, txIdentity interface{}) context.Context {
	if txIdentity == nil {
		return WithScope(ctx)
	}
	if scope := ScopeFromContext(ctx); scope != nil && sameTxIdentity(scope.txIdentity, txIdentity) {
		return ctx
	}
	return context.WithValue(ctx, scopeKey{}, &Scope{
		txIdentity: txIdentity,
		held:       map[string]ParentKey{},
	})
}

// ScopeFromContext returns the active parent lock scope, if one exists.
func ScopeFromContext(ctx context.Context) *Scope {
	scope, _ := ctx.Value(scopeKey{}).(*Scope)
	return scope
}

// WithForcedParentPrelocks attaches transaction-retry parent intents to ctx.
// Retry owners use this to re-run a transaction with the newly discovered
// parent keys available to every nested planner before any child/derived lock
// acquisition starts.
func WithForcedParentPrelocks(ctx context.Context, keys []ParentKey) context.Context {
	ordered := OrderedParentKeys(keys)
	if len(ordered) == 0 {
		return ctx
	}
	return context.WithValue(ctx, forcedPrelocksKey{}, ordered)
}

// ForcedParentPrelocksFromContext returns the retry-time forced parent intents
// carried on ctx, if any.
func ForcedParentPrelocksFromContext(ctx context.Context) []ParentKey {
	keys, _ := ctx.Value(forcedPrelocksKey{}).([]ParentKey)
	return append([]ParentKey(nil), keys...)
}

func sameTxIdentity(left, right interface{}) bool {
	if left == nil || right == nil {
		return false
	}
	leftValue := reflect.ValueOf(left)
	rightValue := reflect.ValueOf(right)
	if !leftValue.IsValid() || !rightValue.IsValid() || leftValue.Type() != rightValue.Type() {
		return false
	}
	if !leftValue.Type().Comparable() {
		return false
	}
	return left == right
}

// OrderedParentKeys trims, drops empty keys, deduplicates, and returns the
// stable acquisition order for parent-row locks.
func OrderedParentKeys(keys []ParentKey) []ParentKey {
	unique := make(map[string]ParentKey, len(keys))
	for _, key := range keys {
		normalized, ok := NormalizeParentKey(key)
		if !ok {
			continue
		}
		unique[KeyID(normalized)] = normalized
	}

	ordered := make([]ParentKey, 0, len(unique))
	for _, key := range unique {
		ordered = append(ordered, key)
	}
	sort.Slice(ordered, func(i, j int) bool {
		return CompareParentKeys(ordered[i], ordered[j]) < 0
	})
	return ordered
}

// NormalizeParentKey trims both key parts and reports whether the key is usable.
func NormalizeParentKey(key ParentKey) (ParentKey, bool) {
	key.ObjectName = strings.TrimSpace(key.ObjectName)
	key.RecordID = strings.TrimSpace(key.RecordID)
	return key, key.ObjectName != "" && key.RecordID != ""
}

// KeyID returns the canonical dedupe key for a parent lock key.
func KeyID(key ParentKey) string {
	key, ok := NormalizeParentKey(key)
	if !ok {
		return "\x00"
	}
	return strings.ToLower(key.ObjectName) + "\x00" + key.RecordID
}

// CompareParentKeys returns the global parent-lock ordering relation.
func CompareParentKeys(left, right ParentKey) int {
	left, _ = NormalizeParentKey(left)
	right, _ = NormalizeParentKey(right)
	leftObject := strings.ToLower(left.ObjectName)
	rightObject := strings.ToLower(right.ObjectName)
	if leftObject < rightObject {
		return -1
	}
	if leftObject > rightObject {
		return 1
	}
	if left.RecordID < right.RecordID {
		return -1
	}
	if left.RecordID > right.RecordID {
		return 1
	}
	return 0
}

// KeysToAcquire returns the new keys a caller may lock now. Duplicate keys
// already held by the transaction are skipped. Newly discovered keys must sort
// after every non-duplicate key already held by this scope.
func (s *Scope) KeysToAcquire(keys []ParentKey) ([]ParentKey, error) {
	if s == nil {
		return nil, ErrMissingParentLockScope
	}
	ordered := OrderedParentKeys(keys)
	s.mu.Lock()
	defer s.mu.Unlock()

	var out []ParentKey
	for _, key := range ordered {
		keyID := KeyID(key)
		if _, ok := s.held[keyID]; ok {
			continue
		}
		if s.highID != "" && keyID < s.highID {
			return nil, fmt.Errorf("%w: %s/%s sorts before already-held key %s", ErrOutOfOrderParentLock, key.ObjectName, key.RecordID, s.highID)
		}
		out = append(out, key)
	}
	return out, nil
}

// MarkAcquired records a successfully acquired parent lock.
func (s *Scope) MarkAcquired(key ParentKey) {
	s.markHeld(key, true)
}

// MarkTransactionOwned records a row inserted by this transaction. Later
// parent-lock requests for the same row can be treated as duplicates because
// the transaction already owns the row.
func (s *Scope) MarkTransactionOwned(key ParentKey) {
	s.markHeld(key, false)
}

func (s *Scope) markHeld(key ParentKey, acquired bool) {
	if s == nil {
		return
	}
	key, ok := NormalizeParentKey(key)
	if !ok {
		return
	}
	keyID := KeyID(key)
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, exists := s.held[keyID]; exists {
		return
	}
	s.held[keyID] = key
	s.order = append(s.order, key)
	if acquired {
		s.acquired = append(s.acquired, key)
	}
	if keyID > s.highID {
		s.highID = keyID
	}
}

// AcquiredKeys returns only rows that were physically locked from durable
// storage. Transaction-owned inserts participate in this attempt's ordering,
// but disappear on rollback and therefore must never become retry prelocks.
func (s *Scope) AcquiredKeys() []ParentKey {
	if s == nil {
		return nil
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]ParentKey(nil), s.acquired...)
}

// HasAcquired reports whether this scope already holds the given parent row.
func (s *Scope) HasAcquired(key ParentKey) bool {
	if s == nil {
		return false
	}
	key, ok := NormalizeParentKey(key)
	if !ok {
		return false
	}
	keyID := KeyID(key)
	s.mu.Lock()
	defer s.mu.Unlock()
	_, exists := s.held[keyID]
	return exists
}

// HeldKeys returns the parent rows held in transaction order.
func (s *Scope) HeldKeys() []ParentKey {
	if s == nil {
		return nil
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]ParentKey(nil), s.order...)
}