coda/internal/worker/worker.go

package worker

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sync"
	"time"

	"coda/internal/config"
	"coda/internal/models"
	"coda/internal/queue"
	"coda/internal/scheduler"
	"coda/internal/store"
	"coda/internal/transcoder"
	"coda/internal/validator"
)

type Rescanner interface {
	Rescan(ctx context.Context, jobPath string) error
}

type NoopRescanner struct{}

func (NoopRescanner) Rescan(_ context.Context, _ string) error {
	return nil
}

type Pool struct {
	cfg        *config.Service
	store      *store.Store
	queue      *queue.Queue
	scheduler  *scheduler.Scheduler
	transcoder *transcoder.Transcoder
	validator  *validator.Validator
	Rescanner  Rescanner
	Notifier   Notifier
	wg         sync.WaitGroup
}

func New(
	cfg *config.Service,
	st *store.Store,
	q *queue.Queue,
	sched *scheduler.Scheduler,
	tr *transcoder.Transcoder,
	val *validator.Validator,
	rescanner Rescanner,
	notifier Notifier,
) *Pool {
	if notifier == nil {
		notifier = NoopNotifier{}
	}
	return &Pool{
		cfg:        cfg,
		store:      st,
		queue:      q,
		scheduler:  sched,
		transcoder: tr,
		validator:  val,
		Rescanner:  rescanner,
		Notifier:   notifier,
	}
}

func (p *Pool) Start(ctx context.Context) {
	cfg := p.cfg.Get()
	workers := cfg.Workers
	if workers < 1 {
		workers = 1
	}

	slog.Info("starting worker pool", "workers", workers)

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

func (p *Pool) Wait() {
	p.wg.Wait()
}

func (p *Pool) run(ctx context.Context) {
	defer p.wg.Done()

	for {
		select {
		case <-ctx.Done():
			return
		case jobID, ok := <-p.queue.Chan():
			if !ok {
				return
			}
			p.process(ctx, jobID)
		}
	}
}

func (p *Pool) process(ctx context.Context, id string) {
	logger := slog.With("job_id", id)
	startedAt := time.Now()

	if err := p.scheduler.WaitUntilOpen(ctx); err != nil {
		logger.Warn("job cancelled before window opened", "error", err)
		return
	}

	if err := p.store.Jobs.UpdateStatus(ctx, id, models.StatusProcessing); err != nil {
		logger.Error("failed to update status to processing", "error", err)
		return
	}

	job, err := p.store.Jobs.Get(ctx, id)
	if err != nil {
		logger.Error("failed to get job", "error", err)
		return
	}

	logger = slog.With("job_id", job.ID, "artist", job.Artist, "album", job.Album)

	flacFiles, err := filepath.Glob(filepath.Join(job.Path, "*.flac"))
	if err != nil {
		p.finishJob(ctx, logger, job, models.StatusError, time.Since(startedAt), fmt.Sprintf("glob flac files: %v", err))
		return
	}
	flacFilesLower, err := filepath.Glob(filepath.Join(job.Path, "*.FLAC"))
	if err != nil {
		p.finishJob(ctx, logger, job, models.StatusError, time.Since(startedAt), fmt.Sprintf("glob FLAC files: %v", err))
		return
	}
	flacFiles = append(flacFiles, flacFilesLower...)

	if len(flacFiles) == 0 {
		p.finishJob(ctx, logger, job, models.StatusError, time.Since(startedAt), "no FLAC files found")
		return
	}

	logger.Info("starting job", "path", job.Path, "flac_count", len(flacFiles))

	if err := p.store.Jobs.UpdateProgress(ctx, id, 0, len(flacFiles)); err != nil {
		logger.Warn("failed to set files total", "error", err)
	}

	var mp3Paths []string
	for i, flac := range flacFiles {
		logger.Info("converting file", "file", filepath.Base(flac))
		mp3, err := p.transcoder.ConvertFile(ctx, flac)
		if err != nil {
			p.finishJob(ctx, logger, job, models.StatusError, time.Since(startedAt), fmt.Sprintf("convert %s: %v", filepath.Base(flac), err))
			return
		}
		mp3Paths = append(mp3Paths, mp3)
		if err := p.store.Jobs.UpdateProgress(ctx, id, i+1, len(flacFiles)); err != nil {
			logger.Warn("failed to update progress", "error", err)
		}
	}

	if err := p.validator.Validate(ctx, mp3Paths); err != nil {
		p.finishJob(ctx, logger, job, models.StatusError, time.Since(startedAt), fmt.Sprintf("validation failed: %v", err))
		return
	}

	cfg := p.cfg.Get()
	if !cfg.DryRun {
		for _, flac := range flacFiles {
			if err := os.Remove(flac); err != nil {
				logger.Warn("failed to remove flac", "path", flac, "error", err)
			}
		}

		if p.Rescanner != nil {
			if err := p.Rescanner.Rescan(ctx, job.Path); err != nil {
				logger.Warn("rescanner failed", "error", err)
			}
		}
	}

	p.finishJob(ctx, logger, job, models.StatusDone, time.Since(startedAt), "")
}

func (p *Pool) finishJob(ctx context.Context, logger *slog.Logger, job *models.Job, status models.JobStatus, duration time.Duration, errMsg string) {
	durationMs := duration.Milliseconds()
	result := *job
	result.Status = status
	result.DurationMs = durationMs
	result.ErrorMsg = errMsg

	if p.Notifier != nil {
		if err := p.Notifier.Notify(ctx, &result); err != nil {
			logger.Warn("notifier failed", "error", err)
		}
	}

	if err := p.store.Jobs.Finish(ctx, job.ID, status, durationMs, errMsg); err != nil {
		logger.Error("failed to finish job", "error", err)
	}

	if status == models.StatusError {
		logger.Error("job failed", "error_msg", errMsg, "duration_ms", durationMs)
	} else {
		logger.Info("job completed", "duration_ms", durationMs)
	}
}