package middleware

import (
	"context"
	"errors"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/singleflight"

	"github.com/pingcap-inc/nexuscrm-api-host/hostruntime"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
)

const DefaultPermissionStateCacheTTL = 5 * time.Minute
const permissionStateResolveAttempts = 2

type permissionStateCacheEntry struct {
	snapshot  permissionstatev1.EffectiveSnapshot
	expiresAt time.Time
}

type permissionStateLineage struct {
	active     string
	retired    map[string]struct{}
	generation uint64
}

type permissionStateGeneration struct {
	lineage            uint64
	globalInvalidation uint64
	userInvalidation   uint64
}

func (generation permissionStateGeneration) loadKey() string {
	return strconv.FormatUint(generation.lineage, 10) + "|" +
		strconv.FormatUint(generation.globalInvalidation, 10) + "|" +
		strconv.FormatUint(generation.userInvalidation, 10)
}

// PermissionStateCache stores immutable aggregate snapshots qualified by their
// contract Version. It never publishes a lower revision, and a lineage switch
// fences older in-flight loads before they can publish.
type PermissionStateCache struct {
	mu                 sync.Mutex
	entries            map[string]permissionStateCacheEntry
	highWatermarks     map[string]permissionstatev1.EffectiveSnapshot
	lineage            permissionStateLineage
	globalInvalidation uint64
	userInvalidations  map[string]uint64
	ttl                time.Duration
	now                func() time.Time
	loads              singleflight.Group
}

// NewPermissionStateCache creates a version-qualified cache. A positive TTL is
// required; TTL expiry removes availability only and never permits downgrade.
func NewPermissionStateCache(ttl time.Duration) (*PermissionStateCache, error) {
	if ttl <= 0 {
		return nil, errors.New("permission state cache: positive TTL is required")
	}
	return &PermissionStateCache{
		entries:           make(map[string]permissionStateCacheEntry),
		highWatermarks:    make(map[string]permissionstatev1.EffectiveSnapshot),
		userInvalidations: make(map[string]uint64),
		ttl:               ttl,
		now:               time.Now,
	}, nil
}

// Resolve returns a snapshot in the admitted floor's lineage at or above its
// revision. A higher-floor follower gets one bounded second load rather than
// accepting the lower-floor leader's singleflight result.
func (cache *PermissionStateCache) Resolve(
	ctx context.Context,
	provider hostruntime.EffectivePermissionSnapshotProvider,
	subject permissionstatev1.Subject,
	floor permissionstatev1.Version,
) (permissionstatev1.EffectiveSnapshot, error) {
	if !cache.configured() || provider == nil {
		return permissionstatev1.EffectiveSnapshot{}, permissionStateFailure(
			permissionstatev1.FailureAuthorityUnavailable,
		)
	}
	if ctx == nil {
		return permissionstatev1.EffectiveSnapshot{}, permissionStateFailure(
			permissionstatev1.FailureInvalidValue,
		)
	}
	if err := subject.Validate(); err != nil {
		return permissionstatev1.EffectiveSnapshot{}, err
	}
	if err := floor.Validate(); err != nil {
		return permissionstatev1.EffectiveSnapshot{}, err
	}

	key := permissionStateSubjectKey(subject)
	generation, cached, err := cache.prepare(key, subject.UserID(), floor)
	if err != nil {
		return permissionstatev1.EffectiveSnapshot{}, err
	}
	if cached != nil {
		return cached.snapshot, nil
	}

	for attempt := 0; attempt < permissionStateResolveAttempts; attempt++ {
		loadKey := key + "|" + generation.loadKey()
		if attempt > 0 {
			loadKey += "|" + strconv.FormatUint(floor.Revision(), 10)
		}
		value, loadErr, _ := cache.loads.Do(loadKey, func() (interface{}, error) {
			snapshot, loadErr := provider.LoadCurrentEffectivePermissionSnapshot(ctx, subject)
			if loadErr != nil {
				return nil, loadErr
			}
			if err := snapshot.Validate(); err != nil {
				return nil, err
			}
			satisfies, err := snapshot.Version().Satisfies(floor)
			if err != nil {
				return nil, err
			}
			if !satisfies {
				return nil, permissionStateFailure(permissionstatev1.FailureStaleSnapshot)
			}
			if publishErr := cache.publish(
				key,
				subject.UserID(),
				generation,
				floor,
				snapshot,
			); publishErr != nil {
				var failure *permissionstatev1.Failure
				if errors.As(publishErr, &failure) &&
					failure.Category == permissionstatev1.FailureStaleSnapshot {
					_, current, currentErr := cache.prepare(
						key,
						subject.UserID(),
						floor,
					)
					if currentErr != nil {
						return nil, currentErr
					}
					if current != nil {
						comparison, compareErr := current.snapshot.Version().Compare(
							snapshot.Version(),
						)
						if compareErr != nil {
							return nil, compareErr
						}
						if comparison > 0 {
							return current.snapshot, nil
						}
					}
				}
				return nil, publishErr
			}
			return snapshot, nil
		})
		if loadErr != nil {
			return permissionstatev1.EffectiveSnapshot{}, loadErr
		}
		snapshot, ok := value.(permissionstatev1.EffectiveSnapshot)
		if !ok {
			return permissionstatev1.EffectiveSnapshot{}, permissionStateFailure(
				permissionstatev1.FailureAuthorityUnavailable,
			)
		}
		satisfies, err := snapshot.Version().Satisfies(floor)
		if err != nil {
			return permissionstatev1.EffectiveSnapshot{}, err
		}
		if satisfies {
			return snapshot, nil
		}
	}
	return permissionstatev1.EffectiveSnapshot{}, permissionStateFailure(
		permissionstatev1.FailureStaleSnapshot,
	)
}

// Warm loads one authoritative aggregate and publishes it through the same
// lineage and monotonicity checks used by Resolve. It has no caller floor
// because the provider's current validated snapshot is the warming authority.
// A later admitted request still proves its own floor independently.
func (cache *PermissionStateCache) Warm(
	ctx context.Context,
	provider hostruntime.EffectivePermissionSnapshotProvider,
	subject permissionstatev1.Subject,
) error {
	if !cache.configured() || provider == nil {
		return permissionStateFailure(
			permissionstatev1.FailureAuthorityUnavailable,
		)
	}
	if ctx == nil {
		return permissionStateFailure(permissionstatev1.FailureInvalidValue)
	}
	if err := subject.Validate(); err != nil {
		return err
	}
	invalidation := cache.invalidationGeneration(subject.UserID())
	snapshot, err := provider.LoadCurrentEffectivePermissionSnapshot(ctx, subject)
	if err != nil {
		return err
	}
	if err := snapshot.Validate(); err != nil {
		return err
	}
	floor := snapshot.Version()
	key := permissionStateSubjectKey(subject)
	generation, _, err := cache.prepare(key, subject.UserID(), floor)
	if err != nil {
		return err
	}
	if generation.globalInvalidation != invalidation.globalInvalidation ||
		generation.userInvalidation != invalidation.userInvalidation {
		return permissionStateFailure(permissionstatev1.FailureStaleSnapshot)
	}
	return cache.publish(key, subject.UserID(), generation, floor, snapshot)
}

// Invalidate removes cache-served aggregates without changing lineage truth or
// the high-watermarks needed to reject an older or conflicting reload.
// Invalidation is a latency optimization only: every request still proves its
// admitted floor and monotonic publication rejects an older reload.
func (cache *PermissionStateCache) Invalidate() {
	if cache == nil {
		return
	}
	cache.mu.Lock()
	cache.globalInvalidation++
	cache.entries = make(map[string]permissionStateCacheEntry)
	cache.mu.Unlock()
}

// InvalidateUser removes every cache-served Profile subject for one user while
// retaining the subject high-watermarks used for monotonic publication.
func (cache *PermissionStateCache) InvalidateUser(userID string) {
	userID = strings.TrimSpace(userID)
	if cache == nil || userID == "" {
		return
	}
	prefix := userID + "\x00"
	cache.mu.Lock()
	if cache.userInvalidations == nil {
		cache.userInvalidations = make(map[string]uint64)
	}
	cache.userInvalidations[userID]++
	for key := range cache.entries {
		if strings.HasPrefix(key, prefix) {
			delete(cache.entries, key)
		}
	}
	cache.mu.Unlock()
}

func (cache *PermissionStateCache) configured() bool {
	return cache != nil &&
		cache.ttl > 0 &&
		cache.now != nil
}

func (cache *PermissionStateCache) prepare(
	key string,
	userID string,
	floor permissionstatev1.Version,
) (permissionStateGeneration, *permissionStateCacheEntry, error) {
	lineageKey := string(floor.Lineage().Bytes())
	cache.mu.Lock()
	defer cache.mu.Unlock()

	state := cache.lineage
	if state.retired == nil {
		state.retired = make(map[string]struct{})
	}
	switch {
	case state.active == "":
		state.active = lineageKey
		state.generation = 1
	case state.active == lineageKey:
	case permissionStateLineageRetired(state, lineageKey):
		return permissionStateGeneration{}, nil, permissionStateFailure(
			permissionstatev1.FailureIncomparableLineage,
		)
	default:
		state.retired[state.active] = struct{}{}
		state.active = lineageKey
		state.generation++
		cache.entries = make(map[string]permissionStateCacheEntry)
		cache.highWatermarks = make(map[string]permissionstatev1.EffectiveSnapshot)
	}
	cache.lineage = state
	generation := permissionStateGeneration{
		lineage:            state.generation,
		globalInvalidation: cache.globalInvalidation,
		userInvalidation:   cache.userInvalidations[userID],
	}

	entry, ok := cache.entries[key]
	if !ok || !cache.now().Before(entry.expiresAt) {
		return generation, nil, nil
	}
	satisfies, err := entry.snapshot.Version().Satisfies(floor)
	if err != nil {
		return permissionStateGeneration{}, nil, err
	}
	if !satisfies {
		return generation, nil, nil
	}
	copy := entry
	return generation, &copy, nil
}

func (cache *PermissionStateCache) publish(
	key string,
	userID string,
	generation permissionStateGeneration,
	floor permissionstatev1.Version,
	snapshot permissionstatev1.EffectiveSnapshot,
) error {
	lineageKey := string(floor.Lineage().Bytes())
	cache.mu.Lock()
	defer cache.mu.Unlock()

	state := cache.lineage
	if state.generation != generation.lineage || state.active != lineageKey {
		return permissionStateFailure(permissionstatev1.FailureIncomparableLineage)
	}
	if cache.globalInvalidation != generation.globalInvalidation ||
		cache.userInvalidations[userID] != generation.userInvalidation {
		return permissionStateFailure(permissionstatev1.FailureStaleSnapshot)
	}
	if highWatermark, ok := cache.highWatermarks[key]; ok {
		comparison, err := snapshot.Version().Compare(highWatermark.Version())
		if err != nil {
			return err
		}
		if comparison < 0 {
			return permissionStateFailure(permissionstatev1.FailureStaleSnapshot)
		}
		if comparison == 0 &&
			(!slices.Equal(snapshot.SystemPermissions(), highWatermark.SystemPermissions()) ||
				!slices.Equal(snapshot.CustomPermissions(), highWatermark.CustomPermissions())) {
			return permissionStateFailure(permissionstatev1.FailureAuthorityUnavailable)
		}
	}
	cache.highWatermarks[key] = snapshot
	cache.entries[key] = permissionStateCacheEntry{
		snapshot:  snapshot,
		expiresAt: cache.now().Add(cache.ttl),
	}
	return nil
}

func (cache *PermissionStateCache) invalidationGeneration(
	userID string,
) permissionStateGeneration {
	cache.mu.Lock()
	defer cache.mu.Unlock()
	return permissionStateGeneration{
		globalInvalidation: cache.globalInvalidation,
		userInvalidation:   cache.userInvalidations[userID],
	}
}

func permissionStateLineageRetired(state permissionStateLineage, lineage string) bool {
	_, retired := state.retired[lineage]
	return retired
}

func permissionStateSubjectKey(subject permissionstatev1.Subject) string {
	return subject.UserID() + "\x00" + subject.ProfileID()
}

func permissionStateFailure(category permissionstatev1.FailureCategory) error {
	return &permissionstatev1.Failure{Category: category}
}