package sfcrmappservice

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

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

type appMetadataWatchResult struct {
	view catalogsemantic.AppMetadataViewV3
	err  error
}

type appMetadataWatchProbe struct {
	results  chan appMetadataWatchResult
	canceled chan struct{}
	once     sync.Once
}

func newAppMetadataWatchProbe() *appMetadataWatchProbe {
	return &appMetadataWatchProbe{
		results: make(chan appMetadataWatchResult, 4), canceled: make(chan struct{}),
	}
}

func (watch *appMetadataWatchProbe) Recv() (catalogsemantic.AppMetadataViewV3, error) {
	select {
	case <-watch.canceled:
		return catalogsemantic.AppMetadataViewV3{}, io.EOF
	case result := <-watch.results:
		return result.view, result.err
	}
}

func (watch *appMetadataWatchProbe) Cancel() {
	watch.once.Do(func() { close(watch.canceled) })
}

func appMetadataCacheTestView(t *testing.T, revision uint64) catalogsemantic.AppMetadataViewV3 {
	t.Helper()
	view := searchTestView(t, []searchObjectSpec{{
		APIName: "Account", KeyPrefix: "001", Label: "Account",
	}})
	view.Lineage.CatalogRevision = revision
	view.Receipts.CatalogRevision = revision
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(view)
	if err := catalogsemantic.ValidateAppMetadataViewV3(view); err != nil {
		t.Fatalf("AppMetadata cache test view is invalid: %v", err)
	}
	return view
}

func TestRuntimeAppMetadataV3CacheReadsOnceAndFollowsMonotonicWatch(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	initial := appMetadataCacheTestView(t, 17)
	next := appMetadataCacheTestView(t, 18)
	var getCalls atomic.Int64
	watchRequests := make(chan catalogsemantic.WatchRequest, 4)
	watches := make(chan *appMetadataWatchProbe, 4)
	cache, err := NewRuntimeAppMetadataV3Cache(
		ctx,
		func(context.Context, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV3, error) {
			getCalls.Add(1)
			return initial, nil
		},
		func(_ context.Context, request catalogsemantic.WatchRequest) (RuntimeAppMetadataV3Watch, error) {
			watch := newAppMetadataWatchProbe()
			watchRequests <- request
			watches <- watch
			return watch, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	defer cache.Close()

	request := <-watchRequests
	if request.AfterCatalogRevision == nil || *request.AfterCatalogRevision != 17 {
		t.Fatalf("watch starts after %#v, want revision 17", request.AfterCatalogRevision)
	}
	for index := 0; index < 20; index++ {
		view, err := cache.Get(ctx, catalogsemantic.GetRequest{})
		if err != nil || view.Lineage.CatalogRevision != 17 {
			t.Fatalf("cached get %d = revision %d, err %v", index, view.Lineage.CatalogRevision, err)
		}
	}
	if got := getCalls.Load(); got != 1 {
		t.Fatalf("Runtime full gets = %d, want startup get only", got)
	}

	watch := <-watches
	watch.results <- appMetadataWatchResult{view: next}
	eventually(t, time.Second, func() bool {
		view, err := cache.Get(ctx, catalogsemantic.GetRequest{})
		return err == nil && view.Lineage.CatalogRevision == 18
	})
	if got := getCalls.Load(); got != 1 {
		t.Fatalf("Runtime full gets after watch update = %d, want 1", got)
	}
	watch.Cancel()
	request = <-watchRequests
	if request.AfterCatalogRevision == nil || *request.AfterCatalogRevision != 18 {
		t.Fatalf("reconnected watch starts after %#v, want revision 18", request.AfterCatalogRevision)
	}
}

func TestRuntimeAppMetadataV3CacheSynchronouslyCatchesRequiredRevision(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	initial := appMetadataCacheTestView(t, 17)
	next := appMetadataCacheTestView(t, 19)
	var getCalls atomic.Int64
	watch := newAppMetadataWatchProbe()
	cache, err := NewRuntimeAppMetadataV3Cache(
		ctx,
		func(_ context.Context, request catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV3, error) {
			call := getCalls.Add(1)
			if call == 1 {
				return initial, nil
			}
			if request.MinCatalogRevision == nil || *request.MinCatalogRevision != 19 {
				t.Fatalf("refresh minimum = %#v, want 19", request.MinCatalogRevision)
			}
			return next, nil
		},
		func(context.Context, catalogsemantic.WatchRequest) (RuntimeAppMetadataV3Watch, error) {
			return watch, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	defer cache.Close()
	minimum := uint64(19)
	view, err := cache.Get(ctx, catalogsemantic.GetRequest{
		MinCatalogRevision: &minimum,
	})
	if err != nil || view.Lineage.CatalogRevision != 19 {
		t.Fatalf("minimum get = revision %d, err %v", view.Lineage.CatalogRevision, err)
	}
	if got := getCalls.Load(); got != 2 {
		t.Fatalf("Runtime full gets = %d, want startup + one revision catch-up", got)
	}
}

func TestRuntimeAppMetadataV3CacheRejectsSameRevisionIdentityDrift(t *testing.T) {
	initial := appMetadataCacheTestView(t, 17)
	drifted := initial
	drifted.AppReleaseDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	drifted.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(drifted)
	cache := &RuntimeAppMetadataV3Cache{view: initial}
	if err := cache.store(drifted); err == nil {
		t.Fatal("same-revision AppMetadata identity drift unexpectedly accepted")
	}
	view, ok := cache.current(nil)
	if !ok || view.Lineage.ViewDigest != initial.Lineage.ViewDigest {
		t.Fatal("identity drift replaced the last verified snapshot")
	}
}

func TestRuntimeAppMetadataV3CacheReconcilesIndependentSecurityEpoch(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	initial := appMetadataCacheTestView(t, 17)
	initial.Receipts.SecurityEpoch = 3
	if err := catalogsemantic.ValidateAppMetadataViewV3(initial); err != nil {
		t.Fatal(err)
	}
	fresh := initial
	fresh.Receipts.SecurityEpoch = 4
	var getCalls atomic.Int64
	refreshStarted := make(chan struct{})
	releaseRefresh := make(chan struct{})
	watch := newAppMetadataWatchProbe()
	cache, err := NewRuntimeAppMetadataV3Cache(
		ctx,
		func(context.Context, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV3, error) {
			if getCalls.Add(1) == 1 {
				return initial, nil
			}
			select {
			case <-refreshStarted:
			default:
				close(refreshStarted)
			}
			<-releaseRefresh
			return fresh, nil
		},
		func(context.Context, catalogsemantic.WatchRequest) (RuntimeAppMetadataV3Watch, error) {
			return watch, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	defer cache.Close()

	actual := initial.Receipts
	actual.CapabilityVersion = "runtime.test.v1"
	actual.SecurityEpoch = 4
	type result struct {
		reconciled bool
		err        error
	}
	results := make(chan result, 8)
	start := make(chan struct{})
	for index := 0; index < cap(results); index++ {
		go func() {
			<-start
			reconciled, reconcileErr := cache.ReconcileReceipts(ctx, initial.Receipts, actual)
			results <- result{reconciled: reconciled, err: reconcileErr}
		}()
	}
	close(start)
	<-refreshStarted
	close(releaseRefresh)
	for index := 0; index < cap(results); index++ {
		result := <-results
		if result.err != nil || !result.reconciled {
			t.Fatalf("concurrent reconcile %d = %t, err %v", index, result.reconciled, result.err)
		}
	}
	view, err := cache.Get(ctx, catalogsemantic.GetRequest{})
	if err != nil || view.Receipts.SecurityEpoch != 4 {
		t.Fatalf("cached epoch = %d, err %v, want 4", view.Receipts.SecurityEpoch, err)
	}
	if got := getCalls.Load(); got != 2 {
		t.Fatalf("Runtime full gets = %d, want startup + one security refresh", got)
	}

	// A concurrent caller that observed the old receipt reuses the completed
	// refresh instead of issuing another multi-megabyte AppMetadata read.
	reconciled, err := cache.ReconcileReceipts(ctx, initial.Receipts, actual)
	if err != nil || !reconciled {
		t.Fatalf("collapsed reconcile = %t, err %v", reconciled, err)
	}
	if got := getCalls.Load(); got != 2 {
		t.Fatalf("Runtime full gets after collapsed reconcile = %d, want 2", got)
	}
}

func TestRuntimeAppMetadataV3CacheRejectsReconcileAuthorityDriftWithoutPublishing(t *testing.T) {
	for _, test := range []struct {
		name   string
		mutate func(*catalogsemantic.AppMetadataViewV3)
	}{
		{
			name: "schema generation",
			mutate: func(view *catalogsemantic.AppMetadataViewV3) {
				view.Lineage.SchemaGeneration++
				view.Receipts.SchemaGeneration++
			},
		},
		{
			name: "engine generation",
			mutate: func(view *catalogsemantic.AppMetadataViewV3) {
				view.Lineage.EngineGeneration++
				view.Receipts.EngineGeneration++
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			initial := appMetadataCacheTestView(t, 17)
			initial.Receipts.SecurityEpoch = 3
			fresh := initial
			fresh.Receipts.SecurityEpoch = 4
			test.mutate(&fresh)
			fresh.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(fresh)
			if err := catalogsemantic.ValidateAppMetadataViewV3(fresh); err != nil {
				t.Fatal(err)
			}
			var getCalls atomic.Int64
			watch := newAppMetadataWatchProbe()
			cache, err := NewRuntimeAppMetadataV3Cache(
				ctx,
				func(context.Context, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV3, error) {
					if getCalls.Add(1) == 1 {
						return initial, nil
					}
					return fresh, nil
				},
				func(context.Context, catalogsemantic.WatchRequest) (RuntimeAppMetadataV3Watch, error) {
					return watch, nil
				},
			)
			if err != nil {
				t.Fatal(err)
			}
			defer cache.Close()

			actual := initial.Receipts
			actual.CapabilityVersion = "runtime.test.v1"
			actual.SecurityEpoch = 4
			reconciled, err := cache.ReconcileReceipts(ctx, initial.Receipts, actual)
			if err != nil || reconciled {
				t.Fatalf("authority-drift reconcile = %t, err %v", reconciled, err)
			}
			cached, ok := cache.current(nil)
			if !ok || !sameAppMetadataV3Snapshot(cached, initial) ||
				cached.Receipts.SecurityEpoch != initial.Receipts.SecurityEpoch {
				t.Fatalf("authority drift replaced verified cache: %#v", cached.Receipts)
			}
			if got := getCalls.Load(); got != 2 {
				t.Fatalf("Runtime full gets = %d, want startup + rejected reconcile", got)
			}
		})
	}
}

func TestRuntimeAppMetadataV3CacheDoesNotRegressSecurityEpoch(t *testing.T) {
	newer := appMetadataCacheTestView(t, 17)
	newer.Receipts.SecurityEpoch = 4
	older := newer
	older.Receipts.SecurityEpoch = 3
	cache := &RuntimeAppMetadataV3Cache{view: newer}
	if err := cache.store(older); err != nil {
		t.Fatal(err)
	}
	view, ok := cache.current(nil)
	if !ok || view.Receipts.SecurityEpoch != 4 {
		t.Fatalf("cached epoch = %d, want monotonic epoch 4", view.Receipts.SecurityEpoch)
	}
}

func eventually(t *testing.T, timeout time.Duration, condition func() bool) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if condition() {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatal(errors.New("condition did not become true before deadline"))
}