package worker

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

type testSource struct {
	mu sync.Mutex

	readyErr     error
	reconcileErr error
	claimErr     error
	nextErr      error
	reconciles   int
	claims       int
	queue        []Attempt
	next         time.Time
	onReconcile  func(*testSource)
	claim        func(context.Context, int) (ClaimBatch, error)
}

func (s *testSource) CheckReady(context.Context) error { return s.readyErr }

func (s *testSource) Reconcile(context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.reconciles++
	if s.onReconcile != nil {
		s.onReconcile(s)
	}
	return s.reconcileErr
}

func (s *testSource) Claim(ctx context.Context, limit int) (ClaimBatch, error) {
	if s.claim != nil {
		return s.claim(ctx, limit)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.claims++
	if s.claimErr != nil {
		return ClaimBatch{}, s.claimErr
	}
	if len(s.queue) == 0 {
		return ClaimBatch{}, nil
	}
	if limit > len(s.queue) {
		limit = len(s.queue)
	}
	attempts := append([]Attempt(nil), s.queue[:limit]...)
	s.queue = s.queue[limit:]
	return ClaimBatch{Attempts: attempts, More: len(s.queue) > 0}, nil
}

func (s *testSource) NextWorkAt(context.Context) (time.Time, bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.next, !s.next.IsZero(), s.nextErr
}

func (s *testSource) enqueue(attempts ...Attempt) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.queue = append(s.queue, attempts...)
}

func (s *testSource) counts() (reconciles, claims int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.reconciles, s.claims
}

type testGenerationProbe struct {
	generation atomic.Uint64
	reads      atomic.Uint64
}

func (p *testGenerationProbe) ReadGeneration(context.Context, uint64) (uint64, error) {
	p.reads.Add(1)
	return p.generation.Load(), nil
}

func testConfig() Config {
	return Config{
		Parallelism: 2, ClaimBatchSize: 2,
		ReconcileInterval: time.Hour, ShutdownTimeout: time.Second,
	}
}

func runController(t *testing.T, controller *Controller) (context.CancelFunc, <-chan error) {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- controller.Run(ctx) }()
	return cancel, done
}

func waitFor(t *testing.T, condition func() bool, description string) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for !condition() {
		if time.Now().After(deadline) {
			t.Fatalf("timed out waiting for %s", description)
		}
		time.Sleep(time.Millisecond)
	}
}

func stopController(t *testing.T, cancel context.CancelFunc, done <-chan error) {
	t.Helper()
	cancel()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("controller shutdown: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("controller did not stop")
	}
}

func TestControllerStartupReconcilesAndBoundsAttempts(t *testing.T) {
	release := make(chan struct{}, 4)
	var active atomic.Int64
	var maximum atomic.Int64
	var completed atomic.Int64
	makeAttempt := func(id string) Attempt {
		return Attempt{ID: id, Run: func(ctx context.Context) error {
			current := active.Add(1)
			for {
				observed := maximum.Load()
				if current <= observed || maximum.CompareAndSwap(observed, current) {
					break
				}
			}
			defer active.Add(-1)
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-release:
				completed.Add(1)
				return nil
			}
		}}
	}
	source := &testSource{queue: []Attempt{
		makeAttempt("1"), makeAttempt("2"), makeAttempt("3"), makeAttempt("4"),
	}}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool { return active.Load() == 2 }, "two active attempts")
	if maximum.Load() != 2 || controller.Snapshot().InFlight != 2 {
		t.Fatalf("maximum=%d snapshot=%+v, want exactly two in flight", maximum.Load(), controller.Snapshot())
	}
	release <- struct{}{}
	release <- struct{}{}
	waitFor(t, func() bool { return completed.Load() == 2 && active.Load() == 2 }, "continuation attempts")
	release <- struct{}{}
	release <- struct{}{}
	waitFor(t, func() bool {
		return completed.Load() == 4 && controller.Snapshot().Completed == 4
	}, "all attempt results observed")
	stopController(t, cancel, done)

	reconciles, _ := source.counts()
	snapshot := controller.Snapshot()
	if reconciles != 1 || snapshot.Claimed != 4 || snapshot.Completed != 4 || snapshot.AttemptFailures != 0 {
		t.Fatalf("reconciles=%d snapshot=%+v", reconciles, snapshot)
	}
}

func TestControllerWakeAndReconcileSignalsAreDistinct(t *testing.T) {
	var completed atomic.Int64
	makeAttempt := func(id string) Attempt {
		return Attempt{ID: id, Run: func(context.Context) error {
			completed.Add(1)
			return nil
		}}
	}
	source := &testSource{}
	source.onReconcile = func(source *testSource) {
		if source.reconciles == 2 {
			source.queue = append(source.queue, makeAttempt("reconciled"))
		}
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, controller.Ready, "controller readiness")

	source.enqueue(makeAttempt("wake"))
	controller.Wake()
	waitFor(t, func() bool { return completed.Load() == 1 }, "wake attempt")
	reconciles, _ := source.counts()
	if reconciles != 1 {
		t.Fatalf("Wake performed reconciliation: got %d calls", reconciles)
	}

	controller.RequestReconcile()
	waitFor(t, func() bool { return completed.Load() == 2 }, "reconciled attempt")
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.WakeRequests != 1 || snapshot.ReconcileRequests != 1 || snapshot.Reconciliations != 2 {
		t.Fatalf("snapshot=%+v", snapshot)
	}
}

func TestControllerExactDueTimerActivatesWork(t *testing.T) {
	due := time.Now().Add(60 * time.Millisecond)
	var completed atomic.Bool
	attemptStarted := make(chan time.Time, 1)
	source := &testSource{next: due}
	source.onReconcile = func(source *testSource) {
		if source.reconciles != 1 {
			return
		}
		source.queue = nil
	}
	// Materialize the due attempt from the authoritative claim path, not from
	// the timer itself. The timer is only an execution opportunity.
	source.claimErr = nil
	controllerSource := &dueSource{testSource: source, due: due, completed: &completed, attemptStarted: attemptStarted}
	controller, err := NewController(controllerSource, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, func() bool {
		snapshot := controller.Snapshot()
		return completed.Load() && snapshot.Completed == 1
	}, "due attempt result")
	startedAt := <-attemptStarted
	if startedAt.Before(due) {
		t.Fatalf("attempt started at %s before authoritative due time %s", startedAt, due)
	}
	time.Sleep(10 * time.Millisecond)
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.NextWorkScheduled || snapshot.Completed != 1 ||
		snapshot.ClaimCalls != 2 || snapshot.NextWorkLookups != 2 {
		t.Fatalf("snapshot=%+v", snapshot)
	}
}

type dueSource struct {
	*testSource
	due            time.Time
	completed      *atomic.Bool
	attemptStarted chan<- time.Time
	once           atomic.Bool
}

func (s *dueSource) Claim(ctx context.Context, limit int) (ClaimBatch, error) {
	if !time.Now().Before(s.due) && s.once.CompareAndSwap(false, true) {
		s.mu.Lock()
		s.next = time.Time{}
		s.mu.Unlock()
		return ClaimBatch{Attempts: []Attempt{{ID: "due", Run: func(context.Context) error {
			s.attemptStarted <- time.Now()
			s.completed.Store(true)
			return nil
		}}}}, nil
	}
	return s.testSource.Claim(ctx, limit)
}

func TestControllerPeriodicReconcileRecoversLostWake(t *testing.T) {
	var completed atomic.Bool
	source := &testSource{}
	source.onReconcile = func(source *testSource) {
		if source.reconciles == 2 {
			source.queue = append(source.queue, Attempt{ID: "recovered", Run: func(context.Context) error {
				completed.Store(true)
				return nil
			}})
		}
	}
	config := testConfig()
	config.ReconcileInterval = 20 * time.Millisecond
	controller, err := NewController(source, nil, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, completed.Load, "periodic reconciliation recovery")
	stopController(t, cancel, done)
}

func TestControllerGenerationChangeReconciles(t *testing.T) {
	var completed atomic.Bool
	probe := &testGenerationProbe{}
	probe.generation.Store(1)
	source := &testSource{}
	source.onReconcile = func(source *testSource) {
		if source.reconciles == 2 {
			source.queue = append(source.queue, Attempt{ID: "generation", Run: func(context.Context) error {
				completed.Store(true)
				return nil
			}})
		}
	}
	config := testConfig()
	config.GenerationInterval = 10 * time.Millisecond
	controller, err := NewController(source, probe, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, controller.Ready, "controller readiness")
	probe.generation.Store(2)
	waitFor(t, completed.Load, "generation reconciliation")
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.GenerationChanges != 1 || snapshot.Reconciliations != 2 {
		t.Fatalf("snapshot=%+v", snapshot)
	}
}

func TestControllerAttemptFailureIsFatalAndPanicSafe(t *testing.T) {
	for _, tc := range []struct {
		name string
		run  func(context.Context) error
	}{
		{name: "error", run: func(context.Context) error { return errors.New("disposition failed") }},
		{name: "panic", run: func(context.Context) error { panic("boom") }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			source := &testSource{queue: []Attempt{{ID: tc.name, Run: tc.run}}}
			controller, err := NewController(source, nil, testConfig())
			if err != nil {
				t.Fatal(err)
			}
			done := make(chan error, 1)
			go func() { done <- controller.Run(context.Background()) }()
			select {
			case runErr := <-done:
				if runErr == nil {
					t.Fatal("controller succeeded after unsafe attempt failure")
				}
			case <-time.After(2 * time.Second):
				t.Fatal("controller did not fail")
			}
			if snapshot := controller.Snapshot(); snapshot.AttemptFailures != 1 {
				t.Fatalf("snapshot=%+v", snapshot)
			}
		})
	}
}

func TestControllerCancellationIsGracefulWhileAttemptStops(t *testing.T) {
	started := make(chan struct{})
	source := &testSource{queue: []Attempt{{ID: "blocking", Run: func(ctx context.Context) error {
		close(started)
		<-ctx.Done()
		// The domain durably releases or disposes the claim before reporting
		// success. Returning ctx.Err would represent an unsafe unknown outcome.
		return nil
	}}}}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("attempt did not start")
	}
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.InFlight != 0 || snapshot.Completed != 1 ||
		snapshot.AttemptFailures != 0 || snapshot.Unusable {
		t.Fatalf("graceful cancellation accounting=%+v", snapshot)
	}
}

func TestControllerCancellationDoesNotHideAttemptFailure(t *testing.T) {
	started := make(chan struct{})
	source := &testSource{queue: []Attempt{{ID: "unsafe-cancel", Run: func(ctx context.Context) error {
		close(started)
		<-ctx.Done()
		return ctx.Err()
	}}}}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("attempt did not start")
	}
	cancel()
	select {
	case runErr := <-done:
		if !errors.Is(runErr, context.Canceled) {
			t.Fatalf("cancellation hid attempt failure: %v", runErr)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("controller did not stop")
	}
	if snapshot := controller.Snapshot(); snapshot.AttemptFailures != 1 || snapshot.Completed != 0 || !snapshot.Unusable {
		t.Fatalf("unsafe cancellation accounting=%+v", snapshot)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
		t.Fatalf("unsafe controller reuse error=%v", runErr)
	}
}

func TestControllerCancellationDoesNotHideUnknownClaimOutcome(t *testing.T) {
	started := make(chan struct{})
	source := &testSource{}
	source.claim = func(ctx context.Context, _ int) (ClaimBatch, error) {
		close(started)
		<-ctx.Done()
		return ClaimBatch{}, ctx.Err()
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("claim did not start")
	}
	cancel()
	select {
	case runErr := <-done:
		if !errors.Is(runErr, context.Canceled) {
			t.Fatalf("unknown claim outcome was hidden: %v", runErr)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("controller did not stop")
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable {
		t.Fatalf("unknown claim outcome left controller reusable: %+v", snapshot)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
		t.Fatalf("unsafe controller reuse error=%v", runErr)
	}
}

func TestControllerClaimErrorWithUnknownOutcomePreventsReuse(t *testing.T) {
	claimFailure := errors.New("claim response lost")
	source := &testSource{claimErr: claimFailure}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, claimFailure) {
		t.Fatalf("claim error=%v", runErr)
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable {
		t.Fatalf("unknown claim outcome left controller reusable: %+v", snapshot)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
		t.Fatalf("unsafe controller reuse error=%v", runErr)
	}
}

func TestControllerKnownUncommittedClaimCancellationIsReusable(t *testing.T) {
	started := make(chan struct{})
	var claimCalls atomic.Int64
	source := &testSource{}
	source.claim = func(ctx context.Context, _ int) (ClaimBatch, error) {
		if claimCalls.Add(1) == 1 {
			close(started)
			<-ctx.Done()
			return ClaimBatch{}, errors.Join(ErrSourceOperationNotCommitted, ctx.Err())
		}
		return ClaimBatch{}, nil
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("claim did not start")
	}
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.Unusable {
		t.Fatalf("known uncommitted claim disabled controller: %+v", snapshot)
	}

	cancel, done = runController(t, controller)
	waitFor(t, func() bool { return controller.Snapshot().ClaimCalls >= 2 }, "claim after safe restart")
	stopController(t, cancel, done)
	if snapshot := controller.Snapshot(); snapshot.Unusable || snapshot.ReadinessChecks != 2 {
		t.Fatalf("safe restart accounting=%+v", snapshot)
	}
}

func TestControllerRejectsContradictoryUncommittedClaimResult(t *testing.T) {
	var attempted atomic.Bool
	source := &testSource{}
	source.claim = func(context.Context, int) (ClaimBatch, error) {
		return ClaimBatch{Attempts: []Attempt{{ID: "contradiction", Run: func(context.Context) error {
			attempted.Store(true)
			return nil
		}}}}, ErrSourceOperationNotCommitted
	}
	controller, err := NewController(source, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrSourceOperationNotCommitted) {
		t.Fatalf("contradictory claim error=%v", runErr)
	}
	if attempted.Load() {
		t.Fatal("controller executed an attempt from a contradictory claim result")
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable || snapshot.Claimed != 0 || snapshot.InFlight != 0 {
		t.Fatalf("contradictory claim result accounting=%+v", snapshot)
	}
}

func TestControllerRejectsInvalidClaimBatch(t *testing.T) {
	for _, tc := range []struct {
		name  string
		batch ClaimBatch
	}{
		{name: "empty continuation", batch: ClaimBatch{More: true}},
		{name: "missing id", batch: ClaimBatch{Attempts: []Attempt{{Run: func(context.Context) error { return nil }}}}},
		{name: "missing run", batch: ClaimBatch{Attempts: []Attempt{{ID: "missing-run"}}}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			source := &testSource{}
			source.claim = func(context.Context, int) (ClaimBatch, error) { return tc.batch, nil }
			controller, err := NewController(source, nil, testConfig())
			if err != nil {
				t.Fatal(err)
			}
			if err := controller.Run(context.Background()); err == nil {
				t.Fatal("invalid claim batch was accepted")
			}
			if snapshot := controller.Snapshot(); !snapshot.Unusable || snapshot.Claimed != 0 || snapshot.InFlight != 0 {
				t.Fatalf("invalid claim batch did not fail closed: %+v", snapshot)
			}
			if runErr := controller.Run(context.Background()); !errors.Is(runErr, ErrControllerUnusable) {
				t.Fatalf("invalid claim batch allowed reuse: %v", runErr)
			}
		})
	}
}

func TestControllerShutdownTimeoutPreventsUnsafeReuse(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	source := &testSource{queue: []Attempt{{ID: "ignores-cancellation", Run: func(context.Context) error {
		close(started)
		<-release
		return nil
	}}}}
	config := testConfig()
	config.ShutdownTimeout = 20 * time.Millisecond
	controller, err := NewController(source, nil, config)
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("attempt did not start")
	}
	cancel()
	if err := <-done; !errors.Is(err, ErrShutdownTimeout) {
		t.Fatalf("shutdown error=%v", err)
	}
	if err := controller.Run(context.Background()); !errors.Is(err, ErrControllerUnusable) {
		t.Fatalf("reuse error=%v", err)
	}
	if snapshot := controller.Snapshot(); !snapshot.Unusable || snapshot.InFlight != 1 {
		t.Fatalf("incomplete shutdown is not observable: %+v", snapshot)
	}
	close(release)
}

func TestControllerRejectsConcurrentRun(t *testing.T) {
	controller, err := NewController(&testSource{}, nil, testConfig())
	if err != nil {
		t.Fatal(err)
	}
	cancel, done := runController(t, controller)
	waitFor(t, controller.Ready, "controller readiness")
	if err := controller.Run(context.Background()); !errors.Is(err, ErrAlreadyRunning) {
		t.Fatalf("concurrent run error=%v", err)
	}
	stopController(t, cancel, done)
}

func TestNewControllerRejectsIncompleteConfiguration(t *testing.T) {
	source := &testSource{}
	if _, err := NewController(nil, nil, testConfig()); err == nil {
		t.Fatal("nil source accepted")
	}
	if _, err := NewController(source, nil, Config{}); err == nil {
		t.Fatal("empty configuration accepted")
	}
	config := testConfig()
	config.GenerationInterval = time.Second
	if _, err := NewController(source, nil, config); err == nil {
		t.Fatal("generation interval without probe accepted")
	}
}