package branch

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

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
)

func TestBranchSyncExecutorDoesNotHeadOfLineBlock(t *testing.T) {
	executor := newBranchSyncExecutor(2)
	slowRelease := make(chan struct{})
	fastStarted := make(chan struct{})
	done := make(chan branchSyncBatchResult, 1)

	go func() {
		done <- executor.Execute(context.Background(), []*entity.Branch{
			{ID: "slow"},
			{ID: "fast"},
		}, func(_ context.Context, branch *entity.Branch) error {
			if branch.ID == "slow" {
				<-slowRelease
				return nil
			}
			close(fastStarted)
			return nil
		})
	}()

	select {
	case <-fastStarted:
	case <-time.After(time.Second):
		t.Fatal("fast branch was blocked behind slow branch")
	}
	close(slowRelease)
	result := <-done
	require.Len(t, result.outcomes, 2)
	require.Zero(t, result.skipped)
}

func TestBranchSyncExecutorSharesConcurrencyLimitAcrossBatches(t *testing.T) {
	executor := newBranchSyncExecutor(2)
	started := make(chan struct{}, 4)
	release := make(chan struct{})
	syncBranch := func(ctx context.Context, _ *entity.Branch) error {
		started <- struct{}{}
		<-release
		return ctx.Err()
	}

	var wg sync.WaitGroup
	for _, branches := range [][]*entity.Branch{
		{{ID: "a"}, {ID: "b"}},
		{{ID: "c"}, {ID: "d"}},
	} {
		wg.Add(1)
		go func(batch []*entity.Branch) {
			defer wg.Done()
			executor.Execute(context.Background(), batch, syncBranch)
		}(branches)
	}

	<-started
	<-started
	select {
	case <-started:
		t.Fatal("executor exceeded its shared concurrency limit")
	case <-time.After(100 * time.Millisecond):
	}
	close(release)
	wg.Wait()
}

func TestBranchSyncExecutorSkipsBranchAlreadyInFlight(t *testing.T) {
	executor := newBranchSyncExecutor(2)
	started := make(chan struct{})
	release := make(chan struct{})
	firstDone := make(chan branchSyncBatchResult, 1)
	branch := &entity.Branch{ID: "same"}

	go func() {
		firstDone <- executor.Execute(context.Background(), []*entity.Branch{branch},
			func(context.Context, *entity.Branch) error {
				close(started)
				<-release
				return nil
			})
	}()
	<-started

	duplicate := executor.Execute(context.Background(), []*entity.Branch{branch},
		func(context.Context, *entity.Branch) error {
			t.Fatal("duplicate branch reconciliation started")
			return nil
		})
	require.Equal(t, 1, duplicate.skipped)
	require.Empty(t, duplicate.outcomes)

	close(release)
	<-firstDone
}

func TestBranchSyncExecutorTurnsPanicIntoErrorAndReleasesClaim(t *testing.T) {
	executor := newBranchSyncExecutor(1)
	branch := &entity.Branch{ID: "panics"}

	result := executor.Execute(context.Background(), []*entity.Branch{branch},
		func(context.Context, *entity.Branch) error {
			panic("boom")
		})

	require.Len(t, result.outcomes, 1)
	require.ErrorContains(t, result.outcomes[0].err, "panic while syncing branch panics: boom")

	retry := executor.Execute(context.Background(), []*entity.Branch{branch},
		func(context.Context, *entity.Branch) error { return nil })
	require.Zero(t, retry.skipped)
	require.NoError(t, retry.outcomes[0].err)
}