package workqueue

import (
	"context"
	"sync"
)

type Serializer[T any] struct {
	Sources  []T
	PerBatch int // default 1
}

func NewSerializer[T any](sources []T, perBatch int) Serializer[T] {
	return Serializer[T]{
		Sources:  sources,
		PerBatch: perBatch,
	}
}

func (s Serializer[T]) Do(ctx context.Context, fn func(ctx context.Context, job []T)) {
	if s.PerBatch <= 0 {
		s.PerBatch = 1
	}

	for index := 0; index < len(s.Sources); index += s.PerBatch {
		end := min(index+s.PerBatch, len(s.Sources))

		job := s.Sources[index:end]
		fn(ctx, job)
	}
}

type Parallelizer[T any] struct {
	Sources     []T
	PerBatch    int // default 1
	Concurrency int // default 1
}

func NewParallelizer[T any](sources []T, concurrency, perBatch int) Parallelizer[T] {
	return Parallelizer[T]{
		Sources:     sources,
		Concurrency: concurrency,
		PerBatch:    perBatch,
	}
}

func (p Parallelizer[T]) Do(ctx context.Context, fn func(ctx context.Context, job []T)) {
	if p.PerBatch <= 0 {
		p.PerBatch = 1
	}
	if p.Concurrency <= 0 {
		p.Concurrency = 1
	}

	// build work queue
	jobCount := len(p.Sources)/p.PerBatch + 1
	jobCh := make(chan []T, jobCount)
	for index := 0; index < len(p.Sources); index += p.PerBatch {
		end := min(index+p.PerBatch, len(p.Sources))
		jobCh <- p.Sources[index:end]
	}
	close(jobCh)

	// start workers
	wg := sync.WaitGroup{}
	wg.Add(p.Concurrency)
	for i := 0; i < p.Concurrency; i++ {
		go func() {
			defer wg.Done()

			for job := range jobCh {
				select {
				case <-ctx.Done():
					return
				default:
					fn(ctx, job)
				}
			}
		}()
	}

	wg.Wait()
}