tapir/internal/uploads/worker.go

package uploads

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

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

// 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)
}

// 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
}

// 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

	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) *Worker {
	return &Worker{store: s, conv: c, tr: t, tmpDir: tmpDir, timeout: timeout, wake: make(chan struct{}, 1)}
}

// 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.
func (w *Worker) Process(ctx context.Context, job Job) {
	defer func() { _ = os.Remove(job.Path) }()
	log := slog.Default().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)
		return
	}

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

	jobCtx, cancel := context.WithTimeout(ctx, w.timeout)
	defer cancel()
	if err := w.conv.ToWAV(jobCtx, job.Path, wav); err != nil {
		log.Error("upload: convert", "err", err)
		w.fail(ctx, job, log)
		return
	}
	t, err := w.tr.Transcribe(jobCtx, wav)
	if err != nil {
		log.Error("upload: transcribe", "err", err, "elapsed", time.Since(start).Round(time.Second))
		w.fail(ctx, job, log)
		return
	}
	if err := w.store.CompleteUpload(ctx, job.UserID, job.UploadID, t); err != nil {
		log.Error("upload: store transcript", "err", err)
		w.fail(ctx, job, log)
		return
	}
	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) {
	if ctx.Err() != nil {
		return
	}
	if err := w.store.FailUpload(ctx, job.UserID, job.UploadID, domain.UploadErrTranscriptionFailed); 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
}