internal/uploads/worker.go

@@ -0,0 +1,191 @@
+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
+}