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>
This commit is contained in:
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user