package sfcrmappservice

import (
	"context"
	"errors"
	"io"
	"sync"
	"sync/atomic"
	"time"

	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
)

const appMetadataWatchRetryDelay = 100 * time.Millisecond

// RuntimeAppMetadataV3Watch is the narrow, transport-neutral watch surface
// required by the sfcrm AppMetadata snapshot. Runtime remains the authority
// that computes and receipts every immutable Catalog revision.
type RuntimeAppMetadataV3Watch interface {
	Recv() (catalogsemantic.AppMetadataViewV3, error)
	Cancel()
}

type RuntimeWatchAppMetadataViewV3Func func(
	context.Context,
	catalogsemantic.WatchRequest,
) (RuntimeAppMetadataV3Watch, error)

// RuntimeReconcileAppMetadataV3ReceiptsFunc reconciles one cached AppMetadata
// receipt with a later Runtime operation that was served from the same
// Catalog/schema/engine authority at a newer SecurityEpoch. It returns true
// only after Runtime has reissued the same immutable AppMetadata view at that
// exact epoch.
type RuntimeReconcileAppMetadataV3ReceiptsFunc func(
	context.Context,
	commonsemantic.Receipts,
	commonsemantic.Receipts,
) (bool, error)

// RuntimeAppMetadataV3Cache keeps exactly one immutable, revision-bound view
// for this AppInstallation. HTTP routes may share the snapshot because it is
// app-scoped rather than principal-scoped; caller permissions remain separate
// Runtime requests and are joined by CatalogRevision/SecurityEpoch receipts.
//
// The cache never guesses freshness with a TTL. Runtime's monotonic watch
// publishes the next complete revision. A caller that already requires a
// higher revision synchronously reads that minimum from Runtime before use.
type RuntimeAppMetadataV3Cache struct {
	get   RuntimeGetAppMetadataViewV3Func
	watch RuntimeWatchAppMetadataViewV3Func

	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup

	refreshMu sync.Mutex
	mu        sync.RWMutex
	view      catalogsemantic.AppMetadataViewV3
	closed    bool
	closeOnce sync.Once
	watchMu   sync.Mutex
	active    RuntimeAppMetadataV3Watch

	requests      atomic.Uint64
	fullReads     atomic.Uint64
	watchAttempts atomic.Uint64
	watchUpdates  atomic.Uint64
	watchFailures atomic.Uint64
}

type RuntimeAppMetadataV3CacheMetrics struct {
	Requests      uint64
	FullReads     uint64
	WatchAttempts uint64
	WatchUpdates  uint64
	WatchFailures uint64
}

func NewRuntimeAppMetadataV3Cache(
	initialCtx context.Context,
	get RuntimeGetAppMetadataViewV3Func,
	watch RuntimeWatchAppMetadataViewV3Func,
) (*RuntimeAppMetadataV3Cache, error) {
	if initialCtx == nil || get == nil || watch == nil {
		return nil, errors.New("sfcrm AppService: Runtime AppMetadata get and watch ports are required")
	}
	initial, err := get(initialCtx, catalogsemantic.GetRequest{})
	if err != nil {
		return nil, err
	}
	if err := catalogsemantic.ValidateAppMetadataViewV3(initial); err != nil {
		return nil, err
	}
	ctx, cancel := context.WithCancel(context.Background())
	cache := &RuntimeAppMetadataV3Cache{
		get: get, watch: watch, ctx: ctx, cancel: cancel, view: initial,
	}
	cache.fullReads.Store(1)
	cache.wg.Add(1)
	go cache.runWatch()
	return cache, nil
}

// Get returns an immutable snapshot by value. Callers must not mutate its
// component byte slices. All current sfcrm consumers decode or inspect them
// read-only, avoiding a multi-megabyte defensive copy on every HTTP route.
func (cache *RuntimeAppMetadataV3Cache) Get(
	ctx context.Context,
	request catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV3, error) {
	if cache == nil || ctx == nil {
		return catalogsemantic.AppMetadataViewV3{}, unavailableAppMetadata()
	}
	if err := catalogsemantic.ValidateGetRequest(ctx, request); err != nil {
		return catalogsemantic.AppMetadataViewV3{}, err
	}
	cache.requests.Add(1)
	if view, ok := cache.current(request.MinCatalogRevision); ok {
		return view, nil
	}

	// Collapse concurrent requests that observe the same revision transition.
	cache.refreshMu.Lock()
	defer cache.refreshMu.Unlock()
	if view, ok := cache.current(request.MinCatalogRevision); ok {
		return view, nil
	}
	view, err := cache.get(ctx, request)
	if err != nil {
		return catalogsemantic.AppMetadataViewV3{}, err
	}
	cache.fullReads.Add(1)
	if err := cache.store(view); err != nil {
		return catalogsemantic.AppMetadataViewV3{}, err
	}
	return cache.currentRequired(request.MinCatalogRevision)
}

// ReconcileReceipts closes the one freshness axis that CatalogRevision watches
// cannot observe. A security publication may advance SecurityEpoch without
// changing the immutable Catalog view, so a later Runtime operation can carry
// a newer epoch than the cached view. In that exact case, perform one collapsed
// full read and accept it only when Runtime reissues the same view identity and
// digest at the operation's exact authority. No other authority mismatch is
// repaired here.
func (cache *RuntimeAppMetadataV3Cache) ReconcileReceipts(
	ctx context.Context,
	expected commonsemantic.Receipts,
	actual commonsemantic.Receipts,
) (bool, error) {
	if cache == nil || ctx == nil ||
		!commonsemantic.ValidReceipts(expected, catalogsemantic.AppMetadataV3CapabilityName) ||
		actual.ProtocolVersion != expected.ProtocolVersion ||
		!sameAppMetadataAuthorityExceptSecurity(expected, actual) ||
		actual.SecurityEpoch <= expected.SecurityEpoch {
		return false, nil
	}

	cache.refreshMu.Lock()
	defer cache.refreshMu.Unlock()
	baseline, ok := cache.current(&expected.CatalogRevision)
	if !ok || !sameAppMetadataAuthorityExceptSecurity(baseline.Receipts, expected) {
		return false, nil
	}
	switch baseline.Receipts.SecurityEpoch {
	case actual.SecurityEpoch:
		return true, nil
	case expected.SecurityEpoch:
		// The caller still holds the exact immutable snapshot currently cached.
	default:
		return false, nil
	}

	minimum := expected.CatalogRevision
	fresh, err := cache.get(ctx, catalogsemantic.GetRequest{MinCatalogRevision: &minimum})
	if err != nil {
		return false, err
	}
	cache.fullReads.Add(1)
	if err := catalogsemantic.ValidateAppMetadataViewV3(fresh); err != nil {
		return false, err
	}
	if !sameAppMetadataV3Snapshot(baseline, fresh) ||
		!sameAppMetadataAuthority(fresh.Receipts, actual) {
		return false, nil
	}
	if err := cache.store(fresh); err != nil {
		return false, err
	}
	return true, nil
}

func (cache *RuntimeAppMetadataV3Cache) Metrics() RuntimeAppMetadataV3CacheMetrics {
	if cache == nil {
		return RuntimeAppMetadataV3CacheMetrics{}
	}
	return RuntimeAppMetadataV3CacheMetrics{
		Requests: cache.requests.Load(), FullReads: cache.fullReads.Load(),
		WatchAttempts: cache.watchAttempts.Load(), WatchUpdates: cache.watchUpdates.Load(),
		WatchFailures: cache.watchFailures.Load(),
	}
}

func (cache *RuntimeAppMetadataV3Cache) Close() {
	if cache == nil {
		return
	}
	cache.closeOnce.Do(func() {
		cache.mu.Lock()
		cache.closed = true
		cache.mu.Unlock()
		cache.cancel()
		cache.cancelActiveWatch()
		cache.wg.Wait()
	})
}

func (cache *RuntimeAppMetadataV3Cache) current(minimum *uint64) (catalogsemantic.AppMetadataViewV3, bool) {
	cache.mu.RLock()
	defer cache.mu.RUnlock()
	if cache.closed || cache.view.Lineage.CatalogRevision == 0 ||
		(minimum != nil && cache.view.Lineage.CatalogRevision < *minimum) {
		return catalogsemantic.AppMetadataViewV3{}, false
	}
	return cache.view, true
}

func (cache *RuntimeAppMetadataV3Cache) currentRequired(minimum *uint64) (catalogsemantic.AppMetadataViewV3, error) {
	if view, ok := cache.current(minimum); ok {
		return view, nil
	}
	return catalogsemantic.AppMetadataViewV3{}, unavailableAppMetadata()
}

func (cache *RuntimeAppMetadataV3Cache) store(view catalogsemantic.AppMetadataViewV3) error {
	if err := catalogsemantic.ValidateAppMetadataViewV3(view); err != nil {
		return err
	}
	cache.mu.Lock()
	defer cache.mu.Unlock()
	if cache.closed {
		return unavailableAppMetadata()
	}
	current := cache.view
	switch {
	case current.Lineage.CatalogRevision == 0:
		cache.view = view
	case view.Lineage.CatalogRevision < current.Lineage.CatalogRevision:
		return nil
	case view.Lineage.CatalogRevision == current.Lineage.CatalogRevision:
		if !sameAppMetadataV3Snapshot(current, view) ||
			!sameAppMetadataAuthorityExceptSecurity(current.Receipts, view.Receipts) {
			return &commonsemantic.Failure{Category: commonsemantic.FailureInternal}
		}
		if view.Receipts.SecurityEpoch >= current.Receipts.SecurityEpoch {
			cache.view = view
		}
	default:
		if view.OrgID != current.OrgID ||
			view.AppInstallationID != current.AppInstallationID ||
			view.AppID != current.AppID {
			return &commonsemantic.Failure{Category: commonsemantic.FailureInternal}
		}
		cache.view = view
	}
	return nil
}

func sameAppMetadataAuthorityExceptSecurity(
	left commonsemantic.Receipts,
	right commonsemantic.Receipts,
) bool {
	return left.CatalogRevision == right.CatalogRevision &&
		left.SchemaGeneration == right.SchemaGeneration &&
		left.EngineGeneration == right.EngineGeneration
}

func sameAppMetadataAuthority(
	left commonsemantic.Receipts,
	right commonsemantic.Receipts,
) bool {
	return sameAppMetadataAuthorityExceptSecurity(left, right) &&
		left.SecurityEpoch == right.SecurityEpoch
}

func reconcileAppMetadataSecurityEpoch(
	ctx context.Context,
	reconcile RuntimeReconcileAppMetadataV3ReceiptsFunc,
	expected commonsemantic.Receipts,
	actual commonsemantic.Receipts,
) (bool, error) {
	if reconcile == nil || ctx == nil ||
		!sameAppMetadataAuthorityExceptSecurity(expected, actual) ||
		actual.SecurityEpoch <= expected.SecurityEpoch {
		return false, nil
	}
	return reconcile(ctx, expected, actual)
}

func sameAppMetadataV3Snapshot(
	left catalogsemantic.AppMetadataViewV3,
	right catalogsemantic.AppMetadataViewV3,
) bool {
	return left.OrgID == right.OrgID &&
		left.AppInstallationID == right.AppInstallationID &&
		left.AppID == right.AppID &&
		left.AppReleaseDigest == right.AppReleaseDigest &&
		left.Lineage.CatalogRevision == right.Lineage.CatalogRevision &&
		left.Lineage.SchemaGeneration == right.Lineage.SchemaGeneration &&
		left.Lineage.EngineGeneration == right.Lineage.EngineGeneration &&
		left.Lineage.SourceSnapshotDigest == right.Lineage.SourceSnapshotDigest &&
		left.Lineage.ViewDigest == right.Lineage.ViewDigest
}

func (cache *RuntimeAppMetadataV3Cache) runWatch() {
	defer cache.wg.Done()
	for cache.ctx.Err() == nil {
		view, ok := cache.current(nil)
		if !ok {
			return
		}
		after := view.Lineage.CatalogRevision
		stream, err := cache.watch(cache.ctx, catalogsemantic.WatchRequest{
			AfterCatalogRevision: &after,
		})
		cache.watchAttempts.Add(1)
		if err != nil {
			cache.watchFailures.Add(1)
			if !cache.waitRetry() {
				return
			}
			continue
		}
		if !cache.setActiveWatch(stream) {
			stream.Cancel()
			return
		}
		for cache.ctx.Err() == nil {
			next, recvErr := stream.Recv()
			if recvErr != nil {
				if cache.ctx.Err() == nil && !expectedAppMetadataWatchEnd(recvErr) {
					cache.watchFailures.Add(1)
				}
				stream.Cancel()
				cache.clearActiveWatch()
				if !errors.Is(recvErr, io.EOF) && cache.ctx.Err() == nil {
					// Reconnect from the last accepted revision. A transient
					// transport failure never replaces the immutable snapshot.
				}
				break
			}
			if err := cache.store(next); err != nil {
				cache.watchFailures.Add(1)
				stream.Cancel()
				cache.clearActiveWatch()
				break
			}
			cache.watchUpdates.Add(1)
		}
		cache.clearActiveWatch()
		if !cache.waitRetry() {
			return
		}
	}
}

func expectedAppMetadataWatchEnd(err error) bool {
	if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) ||
		errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	var failure *commonsemantic.Failure
	return errors.As(err, &failure) &&
		(failure.Category == commonsemantic.FailureCanceled ||
			failure.Category == commonsemantic.FailureDeadlineExceeded)
}

func (cache *RuntimeAppMetadataV3Cache) setActiveWatch(watch RuntimeAppMetadataV3Watch) bool {
	cache.watchMu.Lock()
	defer cache.watchMu.Unlock()
	if cache.ctx.Err() != nil {
		return false
	}
	cache.active = watch
	return true
}

func (cache *RuntimeAppMetadataV3Cache) clearActiveWatch() {
	cache.watchMu.Lock()
	defer cache.watchMu.Unlock()
	cache.active = nil
}

func (cache *RuntimeAppMetadataV3Cache) cancelActiveWatch() {
	cache.watchMu.Lock()
	watch := cache.active
	cache.active = nil
	cache.watchMu.Unlock()
	if watch != nil {
		watch.Cancel()
	}
}

func (cache *RuntimeAppMetadataV3Cache) waitRetry() bool {
	timer := time.NewTimer(appMetadataWatchRetryDelay)
	defer timer.Stop()
	select {
	case <-cache.ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func unavailableAppMetadata() error {
	return &commonsemantic.Failure{Category: commonsemantic.FailureUnavailable}
}