package service_test

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

	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/service"

	"gorm.io/gorm"
)

func TestEnqueuePullRequestSerializesWithConcurrentUnstack(t *testing.T) {
	svc, tenantHandle, cleanup := setupTestService(t)
	defer cleanup()
	repo, authCtx, _, prs := setupAsyncMergeStack(t, svc, tenantHandle, "enqueue-unstack-serialization", 3)
	createMergeQueueRuleset(t, svc, authCtx, repo.FullName)
	stack, err := svc.GetPullRequestStackForPR(authCtx, prs[2].ID)
	if err != nil || stack == nil {
		t.Fatalf("GetPullRequestStackForPR = %+v, %v", stack, err)
	}

	var paused atomic.Bool
	resolvedTarget := make(chan struct{})
	resumeEnqueue := make(chan struct{})
	var releaseOnce sync.Once
	release := func() { releaseOnce.Do(func() { close(resumeEnqueue) }) }
	defer release()
	callbackName := "test:block_enqueue_after_queue_target_resolution"
	if err := tenantHandle.DB().Callback().Query().After("gorm:query").Register(callbackName, func(tx *gorm.DB) {
		if tx.Statement.Table == "rulesets" && paused.CompareAndSwap(false, true) {
			close(resolvedTarget)
			<-resumeEnqueue
		}
	}); err != nil {
		t.Fatal(err)
	}
	defer tenantHandle.DB().Callback().Query().Remove(callbackName)

	type enqueueResult struct {
		entries []db.MergeQueueEntry
		err     error
	}
	enqueueCh := make(chan enqueueResult, 1)
	go func() {
		entries, err := svc.EnqueuePullRequest(authCtx, prs[2].ID, prs[2].HeadSHA, false)
		enqueueCh <- enqueueResult{entries: entries, err: err}
	}()
	select {
	case <-resolvedTarget:
	case <-time.After(10 * time.Second):
		t.Fatal("enqueue did not resolve its Queue target")
	}

	type unstackResult struct {
		stack *service.PullRequestStackView
		err   error
	}
	unstackCh := make(chan unstackResult, 1)
	go func() {
		remaining, err := svc.UnstackPullRequestStack(authCtx, repo.FullName, stack.Number)
		unstackCh <- unstackResult{stack: remaining, err: err}
	}()

	var unstack unstackResult
	var unstackCompletedEarly bool
	select {
	case unstack = <-unstackCh:
		unstackCompletedEarly = true
	case <-time.After(250 * time.Millisecond):
		// The Queue submission holds the shared topology lock until its target
		// and entries commit, so unstack must still be waiting here.
	}
	release()

	var enqueue enqueueResult
	select {
	case enqueue = <-enqueueCh:
	case <-time.After(10 * time.Second):
		t.Fatal("enqueue did not finish")
	}
	if !unstackCompletedEarly {
		select {
		case unstack = <-unstackCh:
		case <-time.After(10 * time.Second):
			t.Fatal("unstack did not finish after enqueue released the topology lock")
		}
	}

	if enqueue.err != nil || len(enqueue.entries) != len(prs) {
		t.Fatalf("EnqueuePullRequest = %+v, %v; want one atomic %d-member Stack batch", enqueue.entries, enqueue.err, len(prs))
	}
	if unstack.err == nil || (!errors.Is(unstack.err, service.ErrValidation) && !errors.Is(unstack.err, service.ErrConflict)) {
		t.Fatalf("concurrent UnstackPullRequestStack = %+v, %v; want active-Stack rejection", unstack.stack, unstack.err)
	}
	current, err := svc.GetPullRequestStack(authCtx, repo.FullName, stack.Number)
	if err != nil || len(current.Entries) != len(prs) {
		t.Fatalf("Stack after concurrent enqueue/unstack = %+v, %v", current, err)
	}
	for i, entry := range enqueue.entries {
		if entry.PullRequestID != prs[i].ID {
			t.Fatalf("queue entry %d pull request = %d, want %d", i, entry.PullRequestID, prs[i].ID)
		}
	}
}