package runtimeworker

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

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

func TestSiblingLoadIndependentTimeoutBudgets(t *testing.T) {
	store, mutation, flow := workerScheduledFlowView(t)
	source, err := catalogsnapshot.RuntimeWorkBindingFromView(flow)
	if err != nil {
		t.Fatal(err)
	}
	const budget = immutableArtifactLoadTimeout
	const slowRead = budget * 3 / 5
	for _, tc := range []struct {
		name                       string
		snapshotDelay, viewDelay   time.Duration
		callerTimeout, wantElapsed time.Duration
		wantDeadline               bool
	}{
		{name: "combined reads exceed one budget", snapshotDelay: slowRead, viewDelay: slowRead, wantElapsed: 2 * slowRead},
		{name: "snapshot retains its timeout", snapshotDelay: budget + time.Second, wantElapsed: budget, wantDeadline: true},
		{name: "view retains its own timeout", snapshotDelay: slowRead, viewDelay: budget + time.Second, wantElapsed: slowRead + budget, wantDeadline: true},
		{name: "caller deadline does not cancel shared load", snapshotDelay: slowRead, viewDelay: slowRead, callerTimeout: 5 * time.Second, wantElapsed: 5 * time.Second, wantDeadline: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			synctest.Test(t, func(t *testing.T) {
				delayed := &delayedSiblingStore{Store: store, snapshotDelay: tc.snapshotDelay, viewDelay: tc.viewDelay}
				binder := commandBindingReuseBinder(t, delayed, flow)
				// Drain any detached read even when an enclosing deadline regresses.
				defer delayed.active.Wait()
				ctx := t.Context()
				if tc.callerTimeout != 0 {
					var cancel context.CancelFunc
					ctx, cancel = context.WithTimeout(ctx, tc.callerTimeout)
					defer cancel()
				}
				started := time.Now()
				view, err := binder.openSiblingProfileFromSource(ctx, source, mutation.Profile(), mutation.CapabilityID())
				elapsed := time.Since(started)
				t.Logf("elapsed=%s error=%v", elapsed, err)
				if tc.wantDeadline {
					if !errors.Is(err, context.DeadlineExceeded) || isPermanentMetadataBindingError(err) {
						t.Errorf("want retryable deadline, got %v", err)
					}
				} else if err != nil || view == nil || view.Digest() != mutation.Digest() {
					t.Errorf("independent read budgets must yield the exact published sibling: view=%v error=%v", view, err)
				}
				if elapsed != tc.wantElapsed {
					t.Errorf("elapsed = %s, want %s", elapsed, tc.wantElapsed)
				}
				if tc.callerTimeout != 0 {
					// A surviving caller joins the same load, not a restarted read.
					view, err = binder.openSiblingProfileFromSource(t.Context(), source, mutation.Profile(), mutation.CapabilityID())
					if err != nil || view == nil || view.Digest() != mutation.Digest() {
						t.Errorf("surviving caller: view=%v error=%v", view, err)
					}
					if elapsed := time.Since(started); elapsed != tc.snapshotDelay+tc.viewDelay {
						t.Errorf("shared load finished at %s, want %s", elapsed, tc.snapshotDelay+tc.viewDelay)
					}
				}
			})
		})
	}
}

// Delay only the transport; use real published artifacts and production
// projection, decoding, and verification. synctest advances the timers instantly.
type delayedSiblingStore struct {
	catalogsnapshot.Store
	snapshotDelay, viewDelay time.Duration
	active                   sync.WaitGroup
}

func (s *delayedSiblingStore) Get(ctx context.Context, kind catalogsnapshot.ArtifactKind, digest catalogsnapshot.Digest) (catalogsnapshot.Artifact, error) {
	s.active.Add(1)
	defer s.active.Done()
	delay := s.viewDelay
	if kind == catalogsnapshot.ArtifactCatalogSnapshot {
		delay = s.snapshotDelay
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return catalogsnapshot.Artifact{}, ctx.Err()
	case <-timer.C:
		return s.Store.Get(ctx, kind, digest)
	}
}