tapir/internal/uploads/worker.go

package uploads

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

	"git.d-ma.be/mathias/tapir/internal/domain"
	"git.d-ma.be/mathias/tapir/internal/errkind"
	"git.d-ma.be/mathias/tapir/internal/progress"
)

// ErrStopped is returned by Submit once the worker has stopped.
var ErrStopped = errors.New("uploads: worker stopped")

// Converter turns the uploaded file into the 16 kHz mono PCM WAV whisper wants.
type Converter interface {
	ToWAV(ctx context.Context, src, dst string) error
}

// Transcriber turns a WAV into a timed transcription. The gateway chain
// (iguana primary, berget fallback) implements it.
type Transcriber interface {
	Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error)
}

// Planner is an optional extra on a Transcriber: it declares its own phases (for
// example one per engine) so the page can show them. A Transcriber that does not
// implement it gets a single "transcribing" phase from the worker. Whatever a Planner
// declares it reports itself, through progress.From(ctx) (ADR-035).
type Planner interface {
	Plan() []progress.Planned
}

// WorkerStore is the slice of the private transcript store the worker drives.
type WorkerStore interface {
	StartTranscribing(ctx context.Context, userID, id string) error
	CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error
	FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error
	SaveProgress(ctx context.Context, userID, id string, steps []domain.ProgressStep, at time.Time) error
	Heartbeat(ctx context.Context, userID, id string) error
}

// WorkerOption tunes a Worker.
type WorkerOption func(*Worker)

// WithHeartbeat sets how often a running job stamps its row alive and how often it logs
// which phase is still running. The defaults are 10 s and 5 min (ADR-035).
func WithHeartbeat(beat, still time.Duration) WorkerOption {
	return func(w *Worker) { w.beatEvery, w.stillEvery = beat, still }
}

// WithMaxDuration makes the worker measure the converted audio and fail a job whose real
// length exceeds the limit, or falls far short of what the file claimed. Zero (the
// default) measures nothing. The limit is first checked against the file's header, which
// can lie (ADR-035 review).
func WithMaxDuration(d time.Duration) WorkerOption { return func(w *Worker) { w.maxDuration = d } }

// WithLogger replaces slog.Default() as the worker's logger.
func WithLogger(l *slog.Logger) WorkerOption { return func(w *Worker) { w.log = l } }

const (
	defaultBeatEvery  = 10 * time.Second
	defaultStillEvery = 5 * time.Minute
)

// jobSink binds the store to one job so a progress.Tracker can persist through it.
type jobSink struct {
	st         WorkerStore
	userID, id string
}

func (s jobSink) SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error {
	return s.st.SaveProgress(ctx, s.userID, s.id, steps, at)
}

func (s jobSink) Beat(ctx context.Context) error { return s.st.Heartbeat(ctx, s.userID, s.id) }

// Worker transcribes accepted uploads in-process, ONE AT A TIME (ADR-033). It is
// not a second deployable: it shares the web process's lifetime, which is why it
// inherits ADR-018's single-replica assumption — the queue is process-local. The
// concurrency of one is deliberate: two simultaneous 90-minute jobs on iguana were
// never measured, and a Metal OOM there latches the whisper backend (infra#273).
type Worker struct {
	store   WorkerStore
	conv    Converter
	tr      Transcriber
	tmpDir  string
	timeout time.Duration

	log                   *slog.Logger
	beatEvery, stillEvery time.Duration
	maxDuration           time.Duration

	mu      sync.Mutex
	queue   []Job
	stopped bool
	wake    chan struct{}
}

// NewWorker wires a Worker. timeout bounds each job (conversion + transcription);
// a job that outlives it is failed rather than left spinning.
func NewWorker(s WorkerStore, c Converter, t Transcriber, tmpDir string, timeout time.Duration, opts ...WorkerOption) *Worker {
	w := &Worker{
		store: s, conv: c, tr: t, tmpDir: tmpDir, timeout: timeout, wake: make(chan struct{}, 1),
		beatEvery: defaultBeatEvery, stillEvery: defaultStillEvery,
	}
	for _, o := range opts {
		o(w)
	}
	return w
}

// Submit queues a job. It never blocks. It satisfies Submitter.
func (w *Worker) Submit(j Job) error {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.stopped {
		return ErrStopped
	}
	w.queue = append(w.queue, j)
	select {
	case w.wake <- struct{}{}:
	default:
	}
	return nil
}

// Run processes queued jobs until ctx ends, then deletes the audio of anything
// still queued and refuses new jobs. The rows of those jobs stay queued or
// transcribing and are failed as interrupted by the start-up sweep
// (store.FailInterruptedUploads): the audio is not retained, so they cannot resume.
func (w *Worker) Run(ctx context.Context) {
	defer w.stop()
	for {
		job, ok := w.next(ctx)
		if !ok {
			return
		}
		w.Process(ctx, job)
	}
}

// next blocks for the next job. Cancellation wins over a non-empty queue: a job
// still queued at shutdown must not be STARTED (it would begin a long conversion
// with an already-cancelled context); stop() deletes its audio instead.
func (w *Worker) next(ctx context.Context) (Job, bool) {
	for {
		if ctx.Err() != nil {
			return Job{}, false
		}
		w.mu.Lock()
		if len(w.queue) > 0 {
			j := w.queue[0]
			w.queue = w.queue[1:]
			w.mu.Unlock()
			return j, true
		}
		w.mu.Unlock()
		select {
		case <-w.wake:
		case <-ctx.Done():
			return Job{}, false
		}
	}
}

func (w *Worker) stop() {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.stopped = true
	for _, j := range w.queue {
		_ = os.Remove(j.Path)
	}
	w.queue = nil
}

// Process runs one job: row -> transcribing, convert, transcribe, store. The
// uploaded file and the converted WAV are deleted on every path — the audio is
// never retained (ADR-033). A failure is recorded with a TYPED kind; a shutdown
// mid-job is not a failure and leaves the row for the start-up sweep.
//
// Every phase is reported through one progress.Tracker, which saves it to the row for
// the page and logs it, and a heartbeat stamps the row on its own while the job runs
// (ADR-035). Nothing logged or saved here carries transcript text or the file's name.
func (w *Worker) Process(ctx context.Context, job Job) {
	defer func() { _ = os.Remove(job.Path) }()
	base := w.log
	if base == nil {
		base = slog.Default()
	}
	log := base.With("upload", job.UploadID)
	start := time.Now()

	if err := w.store.StartTranscribing(ctx, job.UserID, job.UploadID); err != nil {
		log.Error("upload: start transcribing", "err", err)
		w.fail(ctx, job, log, domain.UploadErrTranscriptionFailed)
		return
	}

	plan := []progress.Planned{{Key: "received", Label: "Received"}, {Key: "audio", Label: "Preparing the audio"}}
	planner, planned := w.tr.(Planner)
	if planned {
		plan = append(plan, planner.Plan()...)
	} else {
		plan = append(plan, progress.Planned{Key: "transcribe", Label: "Transcribing"})
	}
	plan = append(plan, progress.Planned{Key: "save", Label: "Saving the transcript"})
	tr := progress.NewTracker(ctx, jobSink{st: w.store, userID: job.UserID, id: job.UploadID}, log, plan)
	tr.Start()
	tr.Begin("received", 0)
	tr.Done("received")

	hbCtx, stopBeat := context.WithCancel(ctx)
	hbDone := make(chan struct{})
	go func() {
		defer close(hbDone)
		tr.Heartbeat(hbCtx, w.beatEvery, w.stillEvery)
	}()
	defer func() { stopBeat(); <-hbDone }() // no heartbeat outlives its job

	// abort records where the job stopped. A shutdown (the parent context ended) is not
	// a failure: the row is left for the start-up sweep, so the step stays as it was.
	abort := func(kind domain.UploadErrorKind) {
		if ctx.Err() == nil {
			tr.FailRunning()
		}
		w.fail(ctx, job, log, kind)
	}

	wav, err := w.tempWAV()
	if err != nil {
		log.Error("upload: temp wav", "err", err)
		abort(domain.UploadErrTranscriptionFailed)
		return
	}
	defer func() { _ = os.Remove(wav) }()

	jobCtx, cancel := context.WithTimeout(ctx, w.timeout)
	defer cancel()
	tr.Begin("audio", 0)
	if err := w.conv.ToWAV(jobCtx, job.Path, wav); err != nil {
		log.Error("upload: convert", "kind", errkind.Of(err))
		abort(domain.UploadErrTranscriptionFailed)
		return
	}
	if kind, bad := w.audioLengthProblem(wav, job.Duration); bad {
		abort(kind)
		log.Error("upload: converted audio length rejected", "kind", string(kind))
		return
	}
	tr.Done("audio")

	if !planned {
		tr.Begin("transcribe", 0)
	}
	t, err := w.tr.Transcribe(progress.With(jobCtx, tr), wav)
	if err != nil {
		log.Error("upload: transcribe", "kind", errkind.Of(err), "elapsed", time.Since(start).Round(time.Second))
		abort(domain.UploadErrTranscriptionFailed)
		return
	}
	if !planned {
		tr.Done("transcribe")
	}

	tr.Begin("save", 0)
	if err := w.store.CompleteUpload(ctx, job.UserID, job.UploadID, t); err != nil {
		log.Error("upload: store transcript", "err", err)
		abort(domain.UploadErrTranscriptionFailed)
		return
	}
	tr.Done("save")
	tr.Finish()
	log.Info("upload transcribed", "endpoint", t.Endpoint, "audio_s", t.AudioSeconds,
		"segments", len(t.Segments), "elapsed", time.Since(start).Round(time.Second))
}

// fail records a transcription failure — unless the PARENT context ended, which
// is a shutdown, not a failure: the row is left for the start-up sweep.
func (w *Worker) fail(ctx context.Context, job Job, log *slog.Logger, kind domain.UploadErrorKind) {
	if ctx.Err() != nil {
		return
	}
	if err := w.store.FailUpload(ctx, job.UserID, job.UploadID, kind); err != nil {
		log.Error("upload: record failure", "err", err)
	}
}

func (w *Worker) tempWAV() (string, error) {
	f, err := os.CreateTemp(w.tmpDir, "wav-*")
	if err != nil {
		return "", fmt.Errorf("uploads: temp wav: %w", err)
	}
	name := f.Name()
	if err := f.Close(); err != nil {
		_ = os.Remove(name)
		return "", fmt.Errorf("uploads: temp wav: %w", err)
	}
	return name, nil
}

const (
	// tooLongSlack is how far over the limit the converted audio may be before the claim
	// in the file's header is called a lie: ffmpeg rounds.
	tooLongSlack = 30 * time.Second
	// A file counts as truncated when it converts to under 80% of the length it claimed
	// AND more than two minutes short. Probe estimates for variable-bitrate files are
	// imprecise; refusing real recordings would be worse than the problem.
	incompleteBelow     = 0.8
	incompleteShortfall = 2 * time.Minute
)

// audioLengthProblem measures the converted WAV (16 kHz mono 16-bit PCM is 32,000 bytes a
// second after a 44-byte header) against the limit and against what the file claimed.
func (w *Worker) audioLengthProblem(wav string, claimed time.Duration) (domain.UploadErrorKind, bool) {
	if w.maxDuration <= 0 {
		return "", false
	}
	info, err := os.Stat(wav)
	if err != nil {
		return "", false
	}
	got := time.Duration(float64(max(info.Size()-44, 0)) / 32000 * float64(time.Second))
	switch {
	case got > w.maxDuration+tooLongSlack:
		return domain.UploadErrTooLong, true
	case float64(got) < incompleteBelow*float64(claimed) && claimed-got > incompleteShortfall:
		return domain.UploadErrIncomplete, true
	}
	return "", false
}