package cli

import (
	"context"
	"errors"
	"runtime"
	"sync"
	"sync/atomic"
)

// pool is a bounded worker pool for concurrent, independent tasks (§14). A
// single caller feeds work via submit; up to concurrency tasks run in flight at
// once, bounded by the sem token held from submit until the task returns.
//
// The pool is deliberately upload-agnostic: a task is any func(ctx) error, so
// the same machinery serves push uploads today and prune deletes later. Task
// wording, error identity, and any success side-effect live in the closure.
//
// submit is the only place that logs a per-action line, and it is only ever
// called from one goroutine (the caller feeding work), so output needs no
// locking; only the shared counters are guarded. Tasks run concurrently, so any
// state a task closure touches beyond its own arguments is the caller's
// responsibility to synchronize.
type workerPool struct {
	ctx    context.Context
	cancel context.CancelCauseFunc
	sem    chan struct{}
	wg     sync.WaitGroup

	// failFast cancels the pool on the first failure so in-flight tasks are
	// abandoned and queued ones never start (the push-asset policy, §8.3). When
	// false, every submitted task runs and all failures are collected (the
	// commit/prune-delete policy).
	failFast bool

	mtx       sync.Mutex
	errs      []error
	succeeded int64
}

// newWorkerPool creates a new worker pool to process tasks concurrently. If `concurrency`
// is zero, the number of parallel jobs is aut-detected based on the number of CPUs.
func newWorkerPool(ctx context.Context, concurrency uint, failFast bool) *workerPool {
	if concurrency == 0 {
		concurrency = 2 * uint(runtime.NumCPU())
	}
	ctx, cancel := context.WithCancelCause(ctx)
	return &workerPool{
		ctx:      ctx,
		cancel:   cancel,
		sem:      make(chan struct{}, concurrency),
		failFast: failFast,
	}
}

// submit blocks until a worker slot is free, then runs task in the background.
// It reports false when the pool has been cancelled (fail-fast or the parent
// context), signalling the caller to stop feeding work. action is the text that
// will be logged before the task is executed.
func (p *workerPool) submit(action string, task func(ctx context.Context) error) bool {
	select {
	case p.sem <- struct{}{}:
	case <-p.ctx.Done():
		return false
	}

	logAction(action)

	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		defer func() { <-p.sem }()

		if err := task(p.ctx); err != nil {
			if !errors.Is(err, context.Canceled) {
				p.mtx.Lock()
				p.errs = append(p.errs, err)
				p.mtx.Unlock()
			}
			if p.failFast {
				p.cancel(CancellationError("operation aborted"))
			}
			return
		}
		atomic.AddInt64(&p.succeeded, 1)
	}()
	return true
}

// cancelled reports whether the pool's context is already done (fail-fast
// triggered or the parent context cancelled). A feeder loop can poll this to
// stop producing/buffering work promptly, instead of only discovering the
// cancellation on its next submit.
func (p *workerPool) cancelled() bool {
	select {
	case <-p.ctx.Done():
		return true
	default:
		return false
	}
}

// wait blocks until all in-flight tasks finish, then releases the pool's
// context. It returns the number of succeeded tasks and the errors the
// workers collected.
func (p *workerPool) wait() (int, []error) {
	p.wg.Wait()
	p.cancel(nil)
	return int(p.succeeded), p.errs
}