package branch

import (
	"context"
	"fmt"
	"sync"

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

type branchSyncFunc func(context.Context, *entity.Branch) error

type branchSyncOutcome struct {
	branch *entity.Branch
	err    error
}

type branchSyncBatchResult struct {
	outcomes []branchSyncOutcome
	skipped  int
}

type branchSyncExecutor struct {
	slots    chan struct{}
	inFlight sync.Map
}

func newBranchSyncExecutor(maxConcurrent int) *branchSyncExecutor {
	return &branchSyncExecutor{slots: make(chan struct{}, maxConcurrent)}
}

func (e *branchSyncExecutor) Execute(
	ctx context.Context,
	branches []*entity.Branch,
	syncBranch branchSyncFunc,
) branchSyncBatchResult {
	claimedBranches := make([]*entity.Branch, 0, len(branches))
	result := branchSyncBatchResult{}
	for _, branch := range branches {
		if _, loaded := e.inFlight.LoadOrStore(branch.ID, struct{}{}); loaded {
			result.skipped++
			continue
		}
		claimedBranches = append(claimedBranches, branch)
	}

	result.outcomes = make([]branchSyncOutcome, len(claimedBranches))
	var wg sync.WaitGroup
	for i, branch := range claimedBranches {
		result.outcomes[i].branch = branch
		wg.Add(1)
		go func(index int, branch *entity.Branch) {
			defer wg.Done()
			defer e.inFlight.Delete(branch.ID)
			result.outcomes[index].err = e.executeOne(ctx, branch, syncBranch)
		}(i, branch)
	}
	wg.Wait()
	return result
}

func (e *branchSyncExecutor) executeOne(
	ctx context.Context,
	branch *entity.Branch,
	syncBranch branchSyncFunc,
) (err error) {
	select {
	case e.slots <- struct{}{}:
		defer func() { <-e.slots }()
	case <-ctx.Done():
		return ctx.Err()
	}
	defer func() {
		if recovered := recover(); recovered != nil {
			err = fmt.Errorf("panic while syncing branch %s: %v", branch.ID, recovered)
		}
	}()
	return syncBranch(ctx, branch)
}