package runtimeworker

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

	"github.com/nexuscrm/persistence"
)

type countedPlatformCommandObservationSource struct {
	calls atomic.Int64
	state persistence.RuntimePlatformCommandExecutionState
}

func (s *countedPlatformCommandObservationSource) Observe(context.Context) (persistence.RuntimePlatformCommandExecutionState, error) {
	s.calls.Add(1)
	return s.state, nil
}

type blockingPlatformCommandObservationSource struct {
	started chan struct{}
}

func platformCommandObservationTestOwner() persistence.RuntimeWorkOwnership {
	return persistence.RuntimeWorkOwnership{
		WorkFamily:          persistence.RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "runtime-platform-command-test",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		OwnershipEpoch:      7,
	}
}

func platformCommandObservationTestScope() persistence.RuntimeWorkScope {
	return persistence.RuntimeWorkScope{
		OrgID:               "org-1",
		AppInstallationID:   "installation-1",
		RuntimeCapabilityID: "capability-1",
	}
}

func (s *blockingPlatformCommandObservationSource) Observe(ctx context.Context) (persistence.RuntimePlatformCommandExecutionState, error) {
	select {
	case s.started <- struct{}{}:
	default:
	}
	<-ctx.Done()
	return persistence.RuntimePlatformCommandExecutionState{}, ctx.Err()
}

func newPlatformCommandObservationTestProcessor(
	t *testing.T,
	store *platformCommandStoreProbe,
	factory platformCommandObservationFactory,
) *platformCommandProcessor {
	t.Helper()
	registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
		"runtime.test": {1: &platformCommandExecutorProbe{}},
	}, nil)
	if err != nil {
		t.Fatal(err)
	}
	processor, err := newPlatformCommandProcessor(
		store,
		factory,
		platformCommandBinderProbe{view: commandExecutionTestView(t)},
		registry,
		0,
	)
	if err != nil {
		t.Fatal(err)
	}
	return processor
}

func TestPlatformCommandNegativeHintSkipsClaimAndNeverOpensObservation(t *testing.T) {
	store := &platformCommandStoreProbe{}
	var opens atomic.Int64
	processor := newPlatformCommandObservationTestProcessor(t, store, func() (platformCommandObservationSource, func() error, error) {
		opens.Add(1)
		return &countedPlatformCommandObservationSource{}, func() error { return nil }, nil
	})
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	if store.hintCalls != 1 || store.claimCalls != 0 || opens.Load() != 0 {
		t.Fatalf("hint/claim/observation opens = %d/%d/%d", store.hintCalls, store.claimCalls, opens.Load())
	}
}

func TestPlatformCommandHintFailureFallsThroughToFencedClaim(t *testing.T) {
	store := &platformCommandStoreProbe{hintErr: errors.New("hint unavailable")}
	processor := newPlatformCommandObservationTestProcessor(t, store, platformCommandTestObservationFactory())
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.runPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	if store.hintCalls != 1 || store.claimCalls != 1 {
		t.Fatalf("hint/claim calls = %d/%d, want 1/1", store.hintCalls, store.claimCalls)
	}
}

func TestPlatformCommandObservationHasNoStartupReadAndReusesOneSource(t *testing.T) {
	store := &platformCommandStoreProbe{}
	source := &countedPlatformCommandObservationSource{state: persistence.RuntimePlatformCommandExecutionState{
		QueueDepth: 4, PoisonDepth: 1, OldestAge: 3 * time.Second,
	}}
	var opens atomic.Int64
	var releases atomic.Int64
	processor := newPlatformCommandObservationTestProcessor(t, store, func() (platformCommandObservationSource, func() error, error) {
		opens.Add(1)
		return source, func() error { releases.Add(1); return nil }, nil
	})
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	stop := worker.startObservationLoopWithTiming(t.Context(), time.Hour, time.Second)
	stop()
	if opens.Load() != 0 {
		t.Fatalf("startup observation opens = %d, want zero", opens.Load())
	}
	if _, err := processor.Observe(t.Context()); err != nil {
		t.Fatal(err)
	}
	if _, err := processor.Observe(t.Context()); err != nil {
		t.Fatal(err)
	}
	if err := processor.closeObservation(); err != nil {
		t.Fatal(err)
	}
	if opens.Load() != 1 || source.calls.Load() != 2 || releases.Load() != 1 {
		t.Fatalf("opens/calls/releases = %d/%d/%d", opens.Load(), source.calls.Load(), releases.Load())
	}
}

func TestBlockedPlatformCommandObservationCannotBlockBusinessClaim(t *testing.T) {
	store := &platformCommandStoreProbe{hasWork: true}
	source := &blockingPlatformCommandObservationSource{started: make(chan struct{}, 1)}
	processor := newPlatformCommandObservationTestProcessor(t, store, func() (platformCommandObservationSource, func() error, error) {
		return source, func() error { return nil }, nil
	})
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	stop := worker.startObservationLoopWithTiming(t.Context(), time.Millisecond, time.Second)
	select {
	case <-source.started:
	case <-time.After(time.Second):
		stop()
		t.Fatal("observation did not start")
	}
	processed := make(chan error, 1)
	go func() { processed <- worker.runPass(t.Context()) }()
	select {
	case err := <-processed:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(100 * time.Millisecond):
		t.Fatal("blocked observation also blocked the business claim")
	}
	stop()
	if store.claimCalls != 1 {
		t.Fatalf("claim calls = %d, want one", store.claimCalls)
	}
}

func TestPlatformCommandObservationTimeoutMarksSnapshotStale(t *testing.T) {
	store := &platformCommandStoreProbe{}
	source := &blockingPlatformCommandObservationSource{started: make(chan struct{}, 1)}
	processor := newPlatformCommandObservationTestProcessor(t, store, func() (platformCommandObservationSource, func() error, error) {
		return source, func() error { return nil }, nil
	})
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	worker.refreshObservationWithTimeout(t.Context(), 10*time.Millisecond)
	snapshot := worker.snapshot()
	if snapshot.ObservationFailures != 1 || !snapshot.ObservationStale || snapshot.ObservationDuration <= 0 {
		t.Fatalf("observation snapshot = %+v", snapshot)
	}
	if err := processor.closeObservation(); err != nil {
		t.Fatal(err)
	}
}