internal/uploads/worker.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"git.d-ma.be/mathias/tapir/internal/domain"
+	"git.d-ma.be/mathias/tapir/internal/progress"
 )
 
 // ErrStopped is returned by Submit once the worker has stopped.
@@ -26,13 +27,52 @@ 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 }
+}
+
+// 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
@@ -45,6 +85,9 @@ type Worker struct {
 	tmpDir  string
 	timeout time.Duration
 
+	log                   *slog.Logger
+	beatEvery, stillEvery time.Duration
+
 	mu      sync.Mutex
 	queue   []Job
 	stopped bool
@@ -53,8 +96,15 @@ type Worker 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)}
+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.
@@ -125,9 +175,17 @@ func (w *Worker) stop() {
 // 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) }()
-	log := slog.Default().With("upload", job.UploadID)
+	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 {
@@ -136,32 +194,75 @@ func (w *Worker) Process(ctx context.Context, job Job) {
 		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() {
+		if ctx.Err() == nil {
+			tr.Fail()
+		}
+		w.fail(ctx, job, log)
+	}
+
 	wav, err := w.tempWAV()
 	if err != nil {
 		log.Error("upload: temp wav", "err", err)
-		w.fail(ctx, job, log)
+		abort()
 		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", "err", err)
-		w.fail(ctx, job, log)
+		abort()
 		return
 	}
-	t, err := w.tr.Transcribe(jobCtx, wav)
+	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", "err", err, "elapsed", time.Since(start).Round(time.Second))
-		w.fail(ctx, job, log)
+		abort()
 		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)
-		w.fail(ctx, job, log)
+		abort()
 		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))
 }