package runtimeworker

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
)

func TestCommandSiblingConcurrentLoadSurvivesCallerCancellation(t *testing.T) {
	store, command, mutation, flow := commandBindingReuseViews(t, 0)
	counted := &countingArtifactStore{Store: store, gets: make(map[catalogsnapshot.ArtifactKind]int)}
	blocked := &blockingSiblingStore{
		Store: counted, digest: flow.Digest(), entered: make(chan struct{}), release: make(chan struct{}),
	}
	var release sync.Once
	unblock := func() { release.Do(func() { close(blocked.release) }) }
	t.Cleanup(unblock)
	binder := commandBindingReuseBinder(t, blocked, command)
	encoded := commandBindingReuseEncoded(t, command)
	ctx, cancelAll := context.WithTimeout(t.Context(), 10*time.Second)
	defer cancelAll()
	leaderCtx, cancelLeader := context.WithCancel(ctx)
	defer cancelLeader()
	leaderResult := make(chan error, 1)
	go func() {
		_, err := bindReuseCommand(leaderCtx, binder, encoded)
		leaderResult <- err
	}()
	select {
	case <-blocked.entered:
	case <-ctx.Done():
		t.Fatal("command never requested its published Flow sibling")
	}
	type result struct {
		ctx context.Context
		err error
	}
	const callers = 24
	results := make(chan result, callers)
	for range callers {
		go func() {
			bound, err := bindReuseCommand(ctx, binder, encoded)
			results <- result{ctx: bound, err: err}
		}()
	}
	cancelLeader()
	select {
	case err := <-leaderResult:
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("canceled caller error = %v", err)
		}
	case <-ctx.Done():
		t.Fatal("canceled caller remained blocked by shared loading")
	}
	unblock()
	for range callers {
		select {
		case got := <-results:
			if got.err != nil {
				t.Fatal(got.err)
			}
			assertBindingReuseViews(t, got.ctx, mutation, flow)
		case <-ctx.Done():
			t.Fatal("surviving callers did not complete")
		}
	}
	if got := counted.count(catalogsnapshot.ArtifactCatalogSnapshot); got != 1 {
		t.Fatalf("shared snapshot reads = %d, want 1", got)
	}
	if got := counted.count(catalogsnapshot.ArtifactRuntimeMetadataView); got != 3 {
		t.Fatalf("shared published view reads = %d, want 3", got)
	}
	if _, err := bindReuseCommand(leaderCtx, binder, encoded); !errors.Is(err, context.Canceled) {
		t.Fatalf("warm canceled command error = %v", err)
	}
}

// Only the artifact transport is delayed; decoding, source verification and
// command authority are production implementations over real published files.
type blockingSiblingStore struct {
	catalogsnapshot.Store
	digest  catalogsnapshot.Digest
	entered chan struct{}
	release chan struct{}
	once    sync.Once
}

func (s *blockingSiblingStore) Get(ctx context.Context, kind catalogsnapshot.ArtifactKind, digest catalogsnapshot.Digest) (catalogsnapshot.Artifact, error) {
	if kind == catalogsnapshot.ArtifactRuntimeMetadataView && digest == s.digest {
		s.once.Do(func() { close(s.entered) })
		select {
		case <-s.release:
		case <-ctx.Done():
			return catalogsnapshot.Artifact{}, ctx.Err()
		}
	}
	return s.Store.Get(ctx, kind, digest)
}

func TestSiblingReuseStillValidatesEveryAttempt(t *testing.T) {
	store, command, _, flow := commandBindingReuseViews(t, 0)
	binder := commandBindingReuseBinder(t, store, command)
	source, err := catalogsnapshot.RuntimeWorkBindingFromView(command)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := bindReuseCommand(t.Context(), binder, commandBindingReuseEncoded(t, command)); err != nil {
		t.Fatal(err)
	}
	for _, tc := range []struct {
		name string
		edit func(*catalogsnapshot.RuntimeWorkBinding)
	}{
		{"org", func(b *catalogsnapshot.RuntimeWorkBinding) { b.OrgID = "other-org" }},
		{"installation", func(b *catalogsnapshot.RuntimeWorkBinding) { b.AppInstallationID = "other-installation" }},
		{"revision", func(b *catalogsnapshot.RuntimeWorkBinding) { b.CatalogRevision++ }},
		{"schema", func(b *catalogsnapshot.RuntimeWorkBinding) { b.SchemaGeneration++ }},
		{"engine", func(b *catalogsnapshot.RuntimeWorkBinding) { b.EngineGeneration++ }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			invalid := source
			tc.edit(&invalid)
			_, err := binder.openSiblingProfileFromSource(t.Context(), invalid, flow.Profile(), flow.CapabilityID())
			if !isPermanentMetadataBindingError(err) {
				t.Fatalf("mismatched source must fail permanently, got %v", err)
			}
		})
	}
	if _, err := binder.openSiblingProfileFromSource(t.Context(), source, command.Profile(), flow.CapabilityID()); !isPermanentMetadataBindingError(err) {
		t.Fatalf("wrong expected profile accepted: %v", err)
	}
	// Compatibility is an execution constraint, not an immutable artifact fact.
	binder.compatibility.EngineGeneration++
	if _, err := binder.openSiblingProfileFromSource(t.Context(), source, flow.Profile(), flow.CapabilityID()); err == nil || isPermanentMetadataBindingError(err) {
		t.Fatalf("incompatible engine must fail retryably, got %v", err)
	}
	binder.compatibility.EngineGeneration--
	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	if _, err := binder.openSiblingProfileFromSource(ctx, source, flow.Profile(), flow.CapabilityID()); !errors.Is(err, context.Canceled) {
		t.Fatalf("warm canceled sibling error = %v", err)
	}
	if view, err := binder.openSiblingProfileFromSource(t.Context(), source, flow.Profile(), flow.CapabilityID()); err != nil || view.Digest() != flow.Digest() {
		t.Fatalf("bad attempts poisoned valid binding: view=%v err=%v", view, err)
	}
}

func TestCommandSiblingRequiresPublicationAndCanRetry(t *testing.T) {
	store, command, mutation, flow := commandBindingReuseViews(t, 0)
	partial, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	copyBindingReuseArtifacts(t, partial, store, command, mutation)
	binder := commandBindingReuseBinder(t, partial, command)
	encoded := commandBindingReuseEncoded(t, command)
	if _, err := bindReuseCommand(t.Context(), binder, encoded); err == nil || isPermanentMetadataBindingError(err) {
		t.Fatalf("unpublished sibling must fail retryably, got %v", err)
	}
	copyBindingReuseArtifacts(t, partial, store, flow)
	for _, candidate := range []*mutationArtifactBinder{binder, commandBindingReuseBinder(t, partial, command)} {
		ctx, err := bindReuseCommand(t.Context(), candidate, encoded)
		if err != nil {
			t.Fatal(err)
		}
		assertBindingReuseViews(t, ctx, mutation, flow)
	}
}

func TestCommandSiblingExactSnapshotIsolationAndEviction(t *testing.T) {
	shared, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	type fixture struct {
		command, mutation, flow *catalogsnapshot.RuntimeMetadataView
	}
	var fixtures []fixture
	for width := range 9 {
		store, command, mutation, flow := commandBindingReuseViews(t, width)
		copyBindingReuseArtifacts(t, shared, store, command, mutation, flow)
		fixtures = append(fixtures, fixture{command, mutation, flow})
	}
	counted := &countingArtifactStore{Store: shared, gets: make(map[catalogsnapshot.ArtifactKind]int)}
	binder := commandBindingReuseBinder(t, counted, fixtures[0].command)
	open := func(index int) {
		f := fixtures[index]
		ctx, err := bindReuseCommand(t.Context(), binder, commandBindingReuseEncoded(t, f.command))
		if err != nil {
			t.Fatal(err)
		}
		assertBindingReuseViews(t, ctx, f.mutation, f.flow)
	}
	// Same installation, capability and numeric revision; only content digests
	// distinguish these snapshots. Returning the latest one would be incorrect.
	open(0)
	open(1)
	reads := counted.count(catalogsnapshot.ArtifactRuntimeMetadataView)
	open(0)
	if got := counted.count(catalogsnapshot.ArtifactRuntimeMetadataView); got != reads {
		t.Fatalf("resident exact snapshot was not reused: reads %d -> %d", reads, got)
	}
	for index := 2; index < len(fixtures); index++ {
		open(index)
	}
	reads = counted.count(catalogsnapshot.ArtifactRuntimeMetadataView)
	open(0)
	if got := counted.count(catalogsnapshot.ArtifactRuntimeMetadataView); got != reads+3 {
		t.Fatalf("all three evicted views must be reopened through publication verification: reads %d -> %d", reads, got)
	}
}

func bindReuseCommand(ctx context.Context, binder *mutationArtifactBinder, encoded string) (context.Context, error) {
	authority, err := (commandViewBinderAdapter{binder: binder}).OpenCommand(ctx, encoded, RuntimeFlowCommandKind, RuntimeFlowCommandPayloadFormatVersionV3)
	if err != nil {
		return nil, err
	}
	return authority.Bind(ctx)
}

func assertBindingReuseViews(t testing.TB, ctx context.Context, mutation, flow *catalogsnapshot.RuntimeMetadataView) {
	t.Helper()
	gotMutation, ok := metadata.RuntimeMetadataViewFromContext(ctx)
	if !ok || gotMutation.Digest() != mutation.Digest() {
		t.Fatal("execution did not bind the exact authorized mutation sibling")
	}
	gotFlow, ok := metadata.RuntimeFlowWorkViewFromContext(ctx)
	if !ok || gotFlow.Digest() != flow.Digest() {
		t.Fatal("execution did not bind the exact authorized Flow sibling")
	}
}

func copyBindingReuseArtifacts(t testing.TB, dst, src catalogsnapshot.Store, views ...*catalogsnapshot.RuntimeMetadataView) {
	t.Helper()
	for _, view := range views {
		for kind, digest := range map[catalogsnapshot.ArtifactKind]catalogsnapshot.Digest{
			catalogsnapshot.ArtifactCatalogSnapshot:     view.SourceSnapshotDigest(),
			catalogsnapshot.ArtifactRuntimeMetadataView: view.Digest(),
		} {
			artifact, err := src.Get(t.Context(), kind, digest)
			if err != nil {
				t.Fatal(err)
			}
			if err := dst.Put(t.Context(), artifact); err != nil {
				t.Fatal(err)
			}
		}
	}
}