cmd/tapir/uploads.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"path/filepath"
+
+	"git.d-ma.be/mathias/tapir/internal/adapters/audio"
+	"git.d-ma.be/mathias/tapir/internal/adapters/transcribe"
+	"git.d-ma.be/mathias/tapir/internal/config"
+	"git.d-ma.be/mathias/tapir/internal/uploads"
+)
+
+// uploadsStore is the store surface the upload wiring needs. *store.Store
+// satisfies it.
+type uploadsStore interface {
+	uploads.Store
+	uploads.WorkerStore
+	FailInterruptedUploads(ctx context.Context) (int, error)
+}
+
+// buildUploads wires audio upload and private transcription (ADR-033). It returns
+// nils when uploads are disabled — never a typed-nil Intake, which would mount the
+// /uploads routes over nothing. When enabled it, in order: prepares a private spool
+// directory, deletes any audio a crashed run left in it, fails the rows such a run
+// left queued or transcribing (the audio is not retained, so they can never
+// resume), and only then builds the intake and the worker. The caller runs the
+// worker. The fallback endpoint is a cloud service; an empty
+// TAPIR_TRANSCRIBE_FALLBACK_MODEL keeps private audio on the homelab.
+func buildUploads(ctx context.Context, cfg config.Config, st uploadsStore, log *slog.Logger) (*uploads.Intake, *uploads.Worker, error) {
+	if !cfg.UploadsEnabled {
+		return nil, nil, nil
+	}
+	if cfg.GatewayURL == "" {
+		return nil, nil, fmt.Errorf("uploads: TAPIR_UPLOADS_ENABLED is set but there is no TAPIR_GATEWAY_URL to transcribe through")
+	}
+
+	if err := os.MkdirAll(cfg.UploadTmpDir, 0o700); err != nil {
+		return nil, nil, fmt.Errorf("uploads: spool dir: %w", err)
+	}
+	removeSpooledAudio(cfg.UploadTmpDir, log)
+
+	n, err := st.FailInterruptedUploads(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("uploads: fail interrupted rows: %w", err)
+	}
+	if n > 0 {
+		log.Warn("uploads: marked interrupted by a restart (audio is not retained; they must be re-uploaded)", "count", n)
+	}
+
+	// No overall client timeout: a 90 minute recording takes minutes. The deadline
+	// is the worker's per-job context.
+	client := &http.Client{}
+	primary := transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeModel, client)
+	var fallback transcribe.Backend
+	if cfg.TranscribeFallbackModel != "" && cfg.TranscribeFallbackModel != cfg.TranscribeModel {
+		fallback = transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeFallbackModel, client)
+	}
+
+	worker := uploads.NewWorker(st, &audio.FFmpeg{Bin: cfg.FFmpegBin}, transcribe.NewChain(primary, fallback),
+		cfg.UploadTmpDir, cfg.TranscribeTimeout)
+	intake := uploads.NewIntake(st, &audio.FFprobe{Bin: cfg.FFprobeBin}, worker,
+		uploadLimits(cfg), cfg.UploadTmpDir)
+	return intake, worker, nil
+}
+
+func uploadLimits(cfg config.Config) uploads.Limits {
+	return uploads.Limits{MaxBytes: cfg.UploadMaxBytes, MaxDuration: cfg.UploadMaxDuration}
+}
+
+// removeSpooledAudio deletes the spool files this process family creates
+// (upload-*, wav-*) and nothing else.
+func removeSpooledAudio(dir string, log *slog.Logger) {
+	for _, pat := range []string{"upload-*", "wav-*"} {
+		matches, _ := filepath.Glob(filepath.Join(dir, pat))
+		for _, m := range matches {
+			if err := os.Remove(m); err != nil {
+				log.Warn("uploads: remove leftover audio", "path", m, "err", err)
+			}
+		}
+	}
+}