From 1910d3dfb5284b1b033b64ed332744b7fb2d1461 Mon Sep 17 00:00:00 2001
From: Mathias <mthbqv@gmail.com>
Date: Thu, 1 Oct 2026 16:40:13 +0200
Subject: [PATCH] feat(uploads): Intake -- spool, probe, validate, queue; clean
 up on every failure (ADR-033)

Bounded read (cap+1 byte tells 'at the limit' from 'over'); a file that fails
validation creates no row; a prober fault is our error, not the user's bad
file; a row whose job cannot be submitted is failed, never left queued.

Mutation-checked: cap off-by-one, missing cleanup and deleting the handed-over
file each turn a test red.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
---
 internal/uploads/intake.go | 115 +++++++++++++++++++++++++++++++++++++
 1 file changed, 115 insertions(+)
 create mode 100644 internal/uploads/intake.go

diff --git a/internal/uploads/intake.go b/internal/uploads/intake.go
new file mode 100644
index 0000000..d33cba5
--- /dev/null
+++ b/internal/uploads/intake.go
@@ -0,0 +1,115 @@
+package uploads
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+
+	"git.d-ma.be/mathias/tapir/internal/domain"
+)
+
+// ErrUnreadable is what a Prober returns when the FILE is the problem (not media,
+// truncated, damaged), as opposed to the prober itself failing. Accept turns it
+// into a user-facing rejection; any other prober error is our fault and surfaces
+// as an internal error — a missing ffprobe must not read as "your file is bad".
+var ErrUnreadable = errors.New("uploads: file is not readable media")
+
+// Prober inspects a file on disk. The ffprobe adapter implements it.
+type Prober interface {
+	Probe(ctx context.Context, path string) (Probe, error)
+}
+
+// Store is the slice of the private transcript store the intake needs.
+type Store interface {
+	CreateUpload(ctx context.Context, userID, title string) (string, error)
+	FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error
+}
+
+// Job is one accepted upload awaiting transcription. Path is a temp file the
+// receiver owns: it must delete it when finished (the audio is never retained).
+type Job struct {
+	UserID   string
+	UploadID string
+	Path     string
+}
+
+// Submitter hands accepted jobs to the in-process worker (ADR-033).
+type Submitter interface {
+	Submit(Job) error
+}
+
+// Intake validates an uploaded recording and, if it passes, queues it.
+type Intake struct {
+	store  Store
+	prober Prober
+	sub    Submitter
+	limits Limits
+	tmpDir string
+}
+
+// NewIntake wires an Intake. tmpDir is where uploads are spooled; it must exist.
+func NewIntake(s Store, p Prober, sub Submitter, l Limits, tmpDir string) *Intake {
+	return &Intake{store: s, prober: p, sub: sub, limits: l, tmpDir: tmpDir}
+}
+
+// Accept spools src to disk (bounded by MaxBytes), probes it, validates it, records
+// a queued upload and submits the job. It returns the upload id. A file that fails
+// validation yields a *RejectError and creates NO row — rejection is synchronous
+// and leaves nothing behind. On success the temp file is handed to the job and the
+// worker deletes it; on every failure path Accept deletes it itself.
+func (in *Intake) Accept(ctx context.Context, userID, title string, src io.Reader) (string, error) {
+	f, err := os.CreateTemp(in.tmpDir, "upload-*")
+	if err != nil {
+		return "", fmt.Errorf("uploads: spool: %w", err)
+	}
+	path := f.Name()
+	handedOver := false
+	defer func() {
+		if !handedOver {
+			_ = os.Remove(path)
+		}
+	}()
+
+	// Read one byte past the cap so "exactly at the limit" and "over" are told apart.
+	n, err := io.Copy(f, io.LimitReader(src, in.limits.MaxBytes+1))
+	if cerr := f.Close(); err == nil {
+		err = cerr
+	}
+	if err != nil {
+		return "", fmt.Errorf("uploads: receive: %w", err)
+	}
+	if n > in.limits.MaxBytes {
+		return "", &RejectError{Reason: ReasonTooLarge, Message: fmt.Sprintf(
+			"This file is larger than the %d MB limit.", in.limits.MaxBytes>>20)}
+	}
+
+	probe, err := in.prober.Probe(ctx, path)
+	if errors.Is(err, ErrUnreadable) {
+		return "", &RejectError{
+			Reason:  ReasonUnreadable,
+			Message: "This file could not be read as audio; it may be damaged or not a recording.",
+		}
+	}
+	if err != nil {
+		return "", fmt.Errorf("uploads: probe: %w", err)
+	}
+	if err := Validate(probe, in.limits); err != nil {
+		return "", err
+	}
+
+	id, err := in.store.CreateUpload(ctx, userID, title)
+	if err != nil {
+		return "", fmt.Errorf("uploads: create: %w", err)
+	}
+	if err := in.sub.Submit(Job{UserID: userID, UploadID: id, Path: path}); err != nil {
+		// A row that will never run must not sit queued: that is the stuck spinner.
+		if ferr := in.store.FailUpload(ctx, userID, id, domain.UploadErrInterrupted); ferr != nil {
+			return "", fmt.Errorf("uploads: submit: %w (and failing the row: %v)", err, ferr)
+		}
+		return "", fmt.Errorf("uploads: submit: %w", err)
+	}
+	handedOver = true
+	return id, nil
+}