package worker

import (
	"context"
	"errors"
	"log/slog"
	"sync"
	"time"

	"vidarchive/internal/models"
	"vidarchive/internal/service"
)

type Pool struct {
	downloadSvc *service.DownloadService
	workers     int
	queue       chan *models.Download
	ctx         context.Context
	cancel      context.CancelFunc
	wg          sync.WaitGroup
}

func New(downloadSvc *service.DownloadService, workers int) *Pool {
	ctx, cancel := context.WithCancel(context.Background())
	return &Pool{
		downloadSvc: downloadSvc,
		workers:     workers,
		queue:       make(chan *models.Download, 100),
		ctx:         ctx,
		cancel:      cancel,
	}
}

func (p *Pool) Start() {
	// Reset any downloads that were in progress during a previous run, and clear
	// the temp dirs they left behind so a re-run doesn't import a duplicate.
	if err := p.downloadSvc.ResetStalledDownloads(); err != nil {
		slog.Error("failed to reset stalled downloads", "err", err)
	}

	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go p.worker(i)
	}

	p.wg.Add(1)
	go p.queueChecker()
}

// Stop cancels running downloads and waits for the workers to return, so the
// process doesn't exit while a yt-dlp child is still writing to the temp dir.
func (p *Pool) Stop() {
	p.cancel()
	p.downloadSvc.CancelAll()
	p.wg.Wait()
}

// Submit enqueues a download without blocking. When the buffer is full the row
// stays "queued" in the database and the queue checker picks it up on a later
// tick, so a burst of submissions can't stall the HTTP handler that made them.
func (p *Pool) Submit(d *models.Download) {
	if p.ctx.Err() != nil {
		return
	}

	select {
	case p.queue <- d:
		slog.Debug("download handed to a worker", "download_id", d.ID)
	default:
		slog.Info("download left in database queue, worker buffer full", "download_id", d.ID)
	}
}

func (p *Pool) worker(id int) {
	defer p.wg.Done()
	slog.Debug("worker started", "worker", id)
	for {
		select {
		case d := <-p.queue:
			slog.Info("processing download", "worker", id, "download_id", d.ID)
			processed, err := p.downloadSvc.ExecuteDownload(p.ctx, d)
			switch {
			case errors.Is(err, service.ErrCancelled):
				slog.Info("download cancelled", "worker", id, "download_id", d.ID)
			case err != nil:
				slog.Error("download failed", "worker", id, "download_id", d.ID, "err", err)
			case processed:
				slog.Info("download completed", "worker", id, "download_id", d.ID)
			default:
				slog.Debug("download already claimed by another worker", "worker", id, "download_id", d.ID)
			}
		case <-p.ctx.Done():
			slog.Debug("worker stopped", "worker", id)
			return
		}
	}
}

func (p *Pool) queueChecker() {
	defer p.wg.Done()
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			p.checkQueue()
		case <-p.ctx.Done():
			return
		}
	}
}

func (p *Pool) checkQueue() {
	// Pull at most a channel's worth of the oldest queued downloads rather than
	// loading every queued row each tick. Anything beyond the buffer is picked up
	// on a later tick; duplicates are harmless (MarkStarted claims atomically).
	downloads, err := p.downloadSvc.GetQueued(cap(p.queue))
	if err != nil {
		slog.Error("queue check failed", "err", err)
		return
	}

	for _, d := range downloads {
		select {
		case p.queue <- d:
		default:
			return
		}
	}
}