package invocation

import (
	"bytes"
	"context"
	"errors"
	"log/slog"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
)

type fakeScheduleDispatcherRunner struct {
	mu           sync.Mutex
	calls        int
	inFlight     int
	maxInFlight  int
	started      chan int
	releaseFirst chan struct{}
	outcomes     []ScheduleDispatchOutcome
	errors       []error
}

type synchronizedLogBuffer struct {
	mu     sync.Mutex
	buffer bytes.Buffer
}

func (buffer *synchronizedLogBuffer) Write(value []byte) (int, error) {
	buffer.mu.Lock()
	defer buffer.mu.Unlock()
	return buffer.buffer.Write(value)
}

func (buffer *synchronizedLogBuffer) String() string {
	buffer.mu.Lock()
	defer buffer.mu.Unlock()
	return buffer.buffer.String()
}

func (buffer *synchronizedLogBuffer) contains(value string) bool {
	buffer.mu.Lock()
	defer buffer.mu.Unlock()
	return bytes.Contains(buffer.buffer.Bytes(), []byte(value))
}

func (runner *fakeScheduleDispatcherRunner) RunOnce(context.Context) (ScheduleDispatchOutcome, error) {
	runner.mu.Lock()
	runner.calls++
	call := runner.calls
	runner.inFlight++
	if runner.inFlight > runner.maxInFlight {
		runner.maxInFlight = runner.inFlight
	}
	runner.mu.Unlock()

	if runner.started != nil {
		runner.started <- call
	}
	if call == 1 && runner.releaseFirst != nil {
		<-runner.releaseFirst
	}

	runner.mu.Lock()
	runner.inFlight--
	var outcome ScheduleDispatchOutcome
	if call <= len(runner.outcomes) {
		outcome = runner.outcomes[call-1]
	}
	var err error
	if call <= len(runner.errors) {
		err = runner.errors[call-1]
	}
	runner.mu.Unlock()
	return outcome, err
}

func (runner *fakeScheduleDispatcherRunner) snapshot() (calls int, maxInFlight int) {
	runner.mu.Lock()
	defer runner.mu.Unlock()
	return runner.calls, runner.maxInFlight
}

func TestScheduleWorkerRunsImmediatelyWithoutOverlappingPasses(t *testing.T) {
	runner := &fakeScheduleDispatcherRunner{
		started:      make(chan int, 4),
		releaseFirst: make(chan struct{}),
	}
	worker := NewScheduleWorker(runner, ScheduleWorkerOptions{PollInterval: time.Millisecond})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	worker.Start(ctx)
	worker.Start(ctx)
	require.Equal(t, 1, receiveScheduleWorkerCall(t, runner.started))

	// Several poll intervals can elapse while the first pass is blocked, but
	// the worker must not start an overlapping pass.
	time.Sleep(10 * time.Millisecond)
	calls, maxInFlight := runner.snapshot()
	require.Equal(t, 1, calls)
	require.Equal(t, 1, maxInFlight)

	close(runner.releaseFirst)
	require.Equal(t, 2, receiveScheduleWorkerCall(t, runner.started))
	cancel()
	require.Eventually(t, func() bool {
		calls, maxInFlight = runner.snapshot()
		return calls >= 2 && maxInFlight == 1
	}, time.Second, time.Millisecond)
}

func TestScheduleWorkerLogsCandidateErrorsAndContinuesPolling(t *testing.T) {
	var logs synchronizedLogBuffer
	logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
	runner := &fakeScheduleDispatcherRunner{
		started: make(chan int, 4),
		outcomes: []ScheduleDispatchOutcome{
			{DueListed: 2, Attempted: 2, Created: 1, Errors: []error{errors.New("candidate one failed")}},
			{DueListed: 1, Attempted: 1, Replayed: 1},
		},
	}
	worker := NewScheduleWorker(runner, ScheduleWorkerOptions{Logger: logger, PollInterval: time.Millisecond})
	ctx, cancel := context.WithCancel(context.Background())

	worker.Start(ctx)
	require.Equal(t, 1, receiveScheduleWorkerCall(t, runner.started))
	require.Equal(t, 2, receiveScheduleWorkerCall(t, runner.started))
	cancel()
	require.Eventually(t, func() bool {
		return logs.contains("candidate one failed")
	}, time.Second, time.Millisecond)
	require.Contains(t, logs.String(), "Workflow Invocation schedule candidate failed")
	require.Contains(t, logs.String(), "Workflow Invocation schedule pass completed")
}

func TestScheduleWorkerContinuesAfterPassFailure(t *testing.T) {
	runner := &fakeScheduleDispatcherRunner{
		started: make(chan int, 4),
		errors:  []error{errors.New("list failed")},
	}
	worker := NewScheduleWorker(runner, ScheduleWorkerOptions{PollInterval: time.Millisecond})
	ctx, cancel := context.WithCancel(context.Background())

	worker.Start(ctx)
	require.Equal(t, 1, receiveScheduleWorkerCall(t, runner.started))
	require.Equal(t, 2, receiveScheduleWorkerCall(t, runner.started))
	cancel()
}

func TestScheduleWorkerRunOnceAndNilSafety(t *testing.T) {
	want := ScheduleDispatchOutcome{DueListed: 1, Attempted: 1, Created: 1}
	runner := &fakeScheduleDispatcherRunner{outcomes: []ScheduleDispatchOutcome{want}}
	worker := NewScheduleWorker(runner, ScheduleWorkerOptions{})

	got, err := worker.RunOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, want, got)

	var nilWorker *ScheduleWorker
	got, err = nilWorker.RunOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, ScheduleDispatchOutcome{}, got)
	nilWorker.Start(context.Background())
	nilWorker.Run(context.Background())

	workerWithoutDispatcher := NewScheduleWorker(nil, ScheduleWorkerOptions{})
	workerWithoutDispatcher.Run(context.Background())
}

func receiveScheduleWorkerCall(t *testing.T, calls <-chan int) int {
	t.Helper()
	select {
	case call := <-calls:
		return call
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for schedule worker pass")
		return 0
	}
}