package main

import (
	"context"
	"sync"
)

// Pool membatasi jumlah worker yang memproses job secara bersamaan.
// Job masuk lewat channel jobs, keluar sebagai Result lewat channel
// results; keduanya tanpa buffer sehingga pengirim, worker, dan pembaca
// hasil bertemu langsung tanpa antrean tersembunyi.
type Pool struct {
	jumlahWorker int
	fn           func(Job) (string, error)
	jobs         chan Job
	results      chan Result
	ctx          context.Context
	cancel       context.CancelFunc
	wg           sync.WaitGroup
	sekali       sync.Once
}

// New membuat pool ber-jumlahWorker worker yang memproses job dengan
// fungsi fn. Pool belum berjalan sampai Start dipanggil.
func New(jumlahWorker int, fn func(Job) (string, error)) *Pool {
	return &Pool{
		jumlahWorker: jumlahWorker,
		fn:           fn,
		jobs:         make(chan Job),
		results:      make(chan Result),
	}
}

// Start menjalankan seluruh worker plus satu goroutine pengawas.
// Context yang dibatalkan menghentikan semua worker di tengah pekerjaan;
// Shutdown tetap perlu dipanggil untuk jalur berhenti yang normal.
func (p *Pool) Start(ctx context.Context) {
	p.ctx, p.cancel = context.WithCancel(ctx)
	for i := 0; i < p.jumlahWorker; i++ {
		p.wg.Add(1)
		go worker(p.ctx, p.jobs, p.results, p.fn, &p.wg)
	}
	// Pengawas menutup channel hasil setelah semua worker keluar.
	// Pembaca hasil berhenti bereputar tepat pada saat itu — tanpa
	// pengawas, pembaca tidak pernah tahu pool sudah selesai.
	go p.awasi()
}

// awasi menunggu worker habis lalu menutup results dan melepas context.
// Goroutine inilah yang mencegah bocor: selesai → channel ditutup.
func (p *Pool) awasi() {
	p.wg.Wait()
	close(p.results)
	p.cancel()
}

// Submit mengirim satu job ke worker yang tersedia dan memblokir saat
// semua worker sedang sibuk (backpressure). Job dibuang begitu context
// pool sudah dibatalkan, jadi pemanggil harus berhenti mengirim setelah
// pembatalan terjadi.
func (p *Pool) Submit(job Job) {
	select {
	case p.jobs <- job:
	case <-p.ctx.Done():
		// Pool sudah berhenti; job tidak dikirim.
	}
}

// Results mengembalikan channel hasil. Channel ini ditutup otomatis
// setelah seluruh worker keluar, sehingga `for range` berakhir pasti.
func (p *Pool) Results() <-chan Result {
	return p.results
}

// Shutdown menutup channel job. Worker menghabiskan sisa job yang sudah
// masuk, keluar, lalu pengawas menutup channel hasil. Shutdown aman
// dipanggil berapa pun kali — hanya pemanggil pertama yang menutup.
func (p *Pool) Shutdown() {
	p.sekali.Do(func() {
		close(p.jobs)
	})
}