package worker

import (
	"context"
	"sync"

	"github.com/sirupsen/logrus"
	"go.uber.org/multierr"
)

// Worker defines the contract for backup execution strategies.
//
// Implementations handle a specific backup mode (e.g. MetricWorker for
// VictoriaMetrics export, FileWorker for cross-bucket replication), managing
// shard scheduling, checkpointing, and concurrent processing internally.
type Worker interface {
	// Run loads checkpoint state, schedules shards, and processes them
	// concurrently until all shards complete or an error occurs.
	Run(ctx context.Context) error

	// Close releases resources held by the worker (clients, connections).
	Close() error
}

// runWorker runs a bounded worker pool that processes shards concurrently.
//
// It enqueues all non-skipped shards into a channel, launches `concurrency`
// goroutines to consume them, and aggregates all processing errors. The
// labeled break ensures context cancellation stops the enqueue loop immediately.
func runWorker[S any](
	ctx context.Context,
	shards []S,
	concurrency int,
	shouldSkip func(index int, shard S) bool,
	process func(ctx context.Context, shard S) error,
) error {
	queue := make(chan S, 1)
	errCh := make(chan error, len(shards))

	workerCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	var wg sync.WaitGroup
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for shard := range queue {
				if err := process(workerCtx, shard); err != nil {
					select {
					case errCh <- err:
					default:
					}
				}
			}
		}()
	}

loop:
	for i, shard := range shards {
		if shouldSkip(i, shard) {
			logrus.Infof("Skipped completed shard: %+v", shard)
			continue
		}
		select {
		case queue <- shard:
		case <-workerCtx.Done():
			break loop
		}
	}
	close(queue)
	wg.Wait()

	close(errCh)
	var errs error
	for err := range errCh {
		errs = multierr.Append(errs, err)
	}
	return errs
}