Bounded parallelism.

export function runUnbounded(jobs: Job[], work: Work): Promise<string[]> {
	return Promise.all(jobs.map(work));
}

type Pending = {
	run: () => Promise<unknown>;
	resolve: (value: unknown) => void;
	reject: (reason: unknown) => void;
};

// A small p-limit-style helper: active work is capped, while pending work waits in memory.
export function createLimiter(concurrency: number) {
	if (!Number.isInteger(concurrency) || concurrency < 1) {
		throw new RangeError('concurrency must be a positive integer');
	}
	let active = 0;
	const pending: Pending[] = [];

	function pump() {
		while (active < concurrency && pending.length > 0) {
			const item = pending.shift()!;
			active += 1;
			void item
				.run()
				.then(item.resolve, item.reject)
				.finally(() => {
					active -= 1;
					pump();
				});
		}
	}

	return function limit<T>(run: () => Promise<T>): Promise<T> {
		return new Promise<T>((resolve, reject) => {
			pending.push({
				run,
				resolve: (value) => resolve(value as T),
				reject
			});
			pump();
		});
	};
}

export async function runWithLimit(
	jobs: Job[],
	concurrency: number,
	work: Work
): Promise<string[]> {
	const limit = createLimiter(concurrency);
	return Promise.all(jobs.map((job) => limit(() => work(job))));
}

// A worker-pool style loop keeps a fixed number of consumers pulling from one job list.
export async function runWorkerPool(jobs: Job[], workers: number, work: Work): Promise<string[]> {
	if (!Number.isInteger(workers) || workers < 1) {
		throw new RangeError('workers must be a positive integer');
	}
	const results = new Array<string>(jobs.length);
	let next = 0;
	async function worker() {
		while (next < jobs.length) {
			const index = next;
			next += 1;
			results[index] = await work(jobs[index]);
		}
	}
	await Promise.all(Array.from({ length: Math.min(workers, jobs.length) }, worker));
	return results;
}

// One goroutine per job: every job is admitted at once.
func runUnbounded(ctx context.Context, jobs []Job, work Work) ([]string, error) {
	results := make([]string, len(jobs))
	errs := make([]error, len(jobs))
	var wg sync.WaitGroup
	for index, job := range jobs {
		wg.Add(1)
		go func() {
			defer wg.Done()
			results[index], errs[index] = work(ctx, job)
		}()
	}
	wg.Wait()
	return results, errors.Join(errs...)
}

// A p-limit-style semaphore: a goroutine per job, but only `limit` may hold a slot.
func runWithLimit(ctx context.Context, jobs []Job, limit int, work Work) ([]string, error) {
	if limit < 1 {
		return nil, fmt.Errorf("limit must be positive")
	}
	slots := make(chan struct{}, limit)
	results := make([]string, len(jobs))
	errs := make([]error, len(jobs))
	var wg sync.WaitGroup
	for index, job := range jobs {
		wg.Add(1)
		go func() {
			defer wg.Done()
			slots <- struct{}{}        // wait for a free slot
			defer func() { <-slots }() // give it back
			results[index], errs[index] = work(ctx, job)
		}()
	}
	wg.Wait()
	return results, errors.Join(errs...)
}

// A fixed number of workers pull job indexes from an explicit, bounded queue.
func runWorkerPool(ctx context.Context, jobs []Job, workers int, work Work) ([]string, error) {
	if workers < 1 {
		return nil, fmt.Errorf("workers must be positive")
	}
	queue := make(chan int, workers) // a full queue makes the producer wait
	results := make([]string, len(jobs))
	errs := make([]error, len(jobs))
	var wg sync.WaitGroup
	for range workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for index := range queue {
				results[index], errs[index] = work(ctx, jobs[index])
			}
		}()
	}
	for index := range jobs {
		queue <- index
	}
	close(queue)
	wg.Wait()
	return results, errors.Join(errs...)
}