feat(podcasts): download worker — state machine, idempotency, policy union (task 1.5)

One sequential worker (day one) over a DB-backed queue
(download_state='queued' IS the worklist — the episode list always
renders truth, D8/D19). State machine: pending → queued → downloading
→ downloaded | failed (+ download_error + download_attempts).

Idempotency: downloaded re-request is a no-op; a file already on disk
that probes cleanly re-marks the row without re-fetching
(crash-between-move-and-update reconciliation). Retry: manual path
resets attempts (human override), auto policies re-queue up to
maxAutoDownloadAttempts=5 then stay failed until a human retries.

Pipeline: enclosure fetch (same SSRF/size guards as the fetcher) →
.part temp → declared-size pre-check → MIME/size validation (empty /
over-cap / content-sniff refuses non-audio) → ffprobe reconcile +
embedded-cover extraction (scanner machinery reuse: runFFProbe,
extractEmbeddedAudioCover, AudioMetadataPayload) → atomic rename →
SUCCESS-ONLY row update (invariant 7: a failure never clears existing
episode data).

Policy union (D6): subscriber policies per feed — 'all' → everything
missing; 'newest N' → the N tail positions; 'off' contributes nothing;
auto-retry budget respected at queue time. DeleteFeedFiles for the
admin-only removal path (episode files + cover siblings + artwork).

internal/services/podcast_worker.go

@@ -0,0 +1,484 @@
+// Podcast download worker (plan §7 / D6 / D8 / D19): ONE sequential
+// worker (day one) driving the episode state machine
+//
+//	pending → queued → downloading → downloaded
+//	                              ↘ failed (+ download_error)
+//
+// queued rows are the DB-backed worklist — the state is persisted so
+// the episode list ALWAYS renders truth (no in-memory queue to lose).
+//
+// Idempotency: a download request for an episode whose file already
+// exists and probes cleanly re-marks it downloaded without re-fetching
+// (crash-between-move-and-update reconciliation); a downloaded episode
+// re-request is a full no-op.
+//
+// Retry semantics: manual retry (same endpoint as download) resets
+// download_attempts; auto-download policies re-queue failed episodes on
+// later polls up to maxAutoDownloadAttempts, then stay failed until a
+// human retries.
+//
+// RLS: the worker runs detached — every statement executes under the
+// first-admin identity (invariant 12), same contract as the fetcher.
+package services
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"bookhoard/internal/database"
+
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+// maxAutoDownloadAttempts caps POLICY-driven re-queues of a failing
+// episode; a human retry always resets the budget.
+const maxAutoDownloadAttempts = 5
+
+// PodcastDownloadWorker processes queued episode downloads.
+type PodcastDownloadWorker struct {
+	db  *database.Queries
+	svc *PodcastService
+	wake chan struct{}
+}
+
+func NewPodcastDownloadWorker(db *database.Queries, svc *PodcastService) *PodcastDownloadWorker {
+	return &PodcastDownloadWorker{
+		db:   db,
+		svc:  svc,
+		wake: make(chan struct{}, 1),
+	}
+}
+
+// Wake nudges the loop (new work queued).
+func (w *PodcastDownloadWorker) Wake() {
+	select {
+	case w.wake <- struct{}{}:
+	default:
+	}
+}
+
+// Start launches the loop; it exits when ctx is cancelled. Called once
+// from route registration (scanner background-tasks precedent).
+func (w *PodcastDownloadWorker) Start(ctx context.Context) {
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-w.wake:
+			case <-time.After(5 * time.Second):
+				// Poll fallback: a queued row written by another process
+				// (or a crashed worker's leftovers) still gets picked up.
+			}
+			w.drain(ctx)
+		}
+	}()
+}
+
+// drain processes queued episodes until the queue is empty.
+func (w *PodcastDownloadWorker) drain(ctx context.Context) {
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		acted, err := w.processNext(ctx)
+		if err != nil {
+			log.Printf("[PODCAST-DL] worker step: %v", err)
+			return
+		}
+		if !acted {
+			return
+		}
+	}
+}
+
+// QueueEpisode marks one episode queued (manual download / retry path).
+// Idempotent by design: downloaded → no-op. The per-feed lock
+// serializes against polls rewriting the same row.
+func (w *PodcastDownloadWorker) QueueEpisode(ctx context.Context, episodeID, feedID pgtype.UUID) error {
+	actx, err := w.svc.adminCtx(ctx)
+	if err != nil {
+		return err
+	}
+	lock := w.svc.feedLock(feedID.String())
+	lock.Lock()
+	defer lock.Unlock()
+
+	ep, err := w.db.GetPodcastEpisode(actx, episodeID)
+	if err != nil {
+		return errNotFound("episode not found")
+	}
+	if ep.DownloadState == "downloaded" {
+		return nil // idempotent no-op
+	}
+	// Manual path resets the attempt budget (SetPodcastEpisodePending)
+	// then queues — a human retry always overrides auto-retry capping.
+	if err := w.db.SetPodcastEpisodePending(actx, episodeID); err != nil {
+		return err
+	}
+	if _, err := w.db.SetPodcastEpisodeDownloadQueued(actx, episodeID); err != nil {
+		return err
+	}
+	w.Wake()
+	return nil
+}
+
+// QueueFeedDownloads implements "Download All" (D8): queue exactly the
+// non-downloaded set — pending AND failed rows (retry is the same path).
+func (w *PodcastDownloadWorker) QueueFeedDownloads(ctx context.Context, feedID pgtype.UUID) (int, error) {
+	actx, err := w.svc.adminCtx(ctx)
+	if err != nil {
+		return 0, err
+	}
+	lock := w.svc.feedLock(feedID.String())
+	lock.Lock()
+	defer lock.Unlock()
+
+	eps, err := w.db.ListUndownloadedEpisodes(actx, feedID)
+	if err != nil {
+		return 0, err
+	}
+	queued := 0
+	for _, ep := range eps {
+		if _, err := w.db.SetPodcastEpisodeDownloadQueued(actx, ep.ID); err == nil {
+			queued++
+		}
+	}
+	w.Wake()
+	return queued, nil
+}
+
+// ApplyAutoPolicies unions subscriber policies for a feed (D6: most
+// aggressive wins for what ARRIVES — deletion stays admin-only) and
+// queues the target set, respecting the auto-retry attempt cap.
+func (w *PodcastDownloadWorker) ApplyAutoPolicies(ctx context.Context, feedID pgtype.UUID) (int, error) {
+	actx, err := w.svc.adminCtx(ctx)
+	if err != nil {
+		return 0, err
+	}
+	subs, err := w.db.ListFeedSubscribers(actx, feedID)
+	if err != nil {
+		return 0, err
+	}
+	wantAll := false
+	newestN := 0
+	for _, sub := range subs {
+		switch sub.AutoDownload {
+		case "all":
+			wantAll = true
+		case "newest":
+			if n := int(sub.AutoDownloadCount.Int32); n > newestN {
+				newestN = n
+			}
+		}
+	}
+	if !wantAll && newestN == 0 {
+		return 0, nil // nobody's policy asks for anything
+	}
+
+	lock := w.svc.feedLock(feedID.String())
+	lock.Lock()
+	defer lock.Unlock()
+
+	eps, err := w.db.ListUndownloadedEpisodes(actx, feedID)
+	if err != nil {
+		return 0, err
+	}
+
+	// ListUndownloadedEpisodes is native-order (oldest first); "newest
+	// N" takes the TAIL (highest positions = latest releases).
+	targets := eps
+	if !wantAll && newestN > 0 && len(eps) > newestN {
+		targets = eps[len(eps)-newestN:]
+	}
+
+	queued := 0
+	for _, ep := range targets {
+		// Auto-retry budget: only re-queue within the cap. A failed
+		// episode past the cap waits for a human (its row renders the
+		// reason + retry button — D8/D19).
+		if ep.DownloadState == "failed" && int(ep.DownloadAttempts) >= maxAutoDownloadAttempts {
+			continue
+		}
+		if _, err := w.db.SetPodcastEpisodeDownloadQueued(actx, ep.ID); err == nil {
+			queued++
+		}
+	}
+	if queued > 0 {
+		w.Wake()
+	}
+	return queued, nil
+}
+
+// processNext claims and processes one queued episode. Returns false
+// when the queue was empty.
+func (w *PodcastDownloadWorker) processNext(ctx context.Context) (bool, error) {
+	actx, err := w.svc.adminCtx(ctx)
+	if err != nil {
+		return false, err
+	}
+	ep, err := w.db.GetNextQueuedEpisode(actx)
+	if err != nil {
+		return false, nil // empty queue (or read failure — logged upstream)
+	}
+
+	_ = w.db.SetPodcastEpisodeDownloading(actx, ep.ID)
+
+	if err := w.runPipeline(actx, ep); err != nil {
+		// failed/rejected: state + reason persist; metadata/file facts
+		// untouched (invariant 7 — write-after-success only).
+		_ = w.db.MarkPodcastEpisodeFailed(actx, database.MarkPodcastEpisodeFailedParams{
+			ID:            ep.ID,
+			DownloadError: pgtypeText(err.Error()),
+			Column3:       false, // auto failure: never resets the attempt budget
+		})
+		log.Printf("[PODCAST-DL] episode %s failed: %v", ep.ID, err)
+	}
+	return true, nil
+}
+
+// runPipeline: fetch enclosure (SSRF/size guards) → temp file → MIME/
+// size validation → ffprobe reconcile → move into place → SUCCESS-only
+// row update.
+func (w *PodcastDownloadWorker) runPipeline(ctx context.Context, ep database.MediaItems) error {
+	if !ep.EnclosureUrl.Valid || ep.EnclosureUrl.String == "" {
+		return fmt.Errorf("episode has no enclosure URL")
+	}
+	if !ep.SourceFeedID.Valid {
+		return fmt.Errorf("episode has no feed reference")
+	}
+
+	feed, err := w.db.GetPodcastFeed(ctx, ep.SourceFeedID)
+	if err != nil {
+		return fmt.Errorf("feed lookup: %w", errNotFound("feed not found"))
+	}
+	root, err := w.svc.downloadRootAbs(ctx, &feed)
+	if err != nil {
+		return err
+	}
+
+	// Declared-size pre-check: reject hopeless downloads before
+	// transferring a byte (validation rejections render like failures,
+	// with their reason — D8).
+	if ep.FileSize.Valid && ep.FileSize.Int64 > maxEnclosureCapBytes {
+		return fmt.Errorf("rejected: enclosure declares %d bytes (cap %d)", ep.FileSize.Int64, maxEnclosureCapBytes)
+	}
+
+	feedDir := filepath.Join(root, filepath.Dir(ep.FilePath))
+	finalPath := filepath.Join(root, ep.FilePath)
+	if err := os.MkdirAll(feedDir, 0o755); err != nil {
+		return fmt.Errorf("download dir: %w", err)
+	}
+
+	// Idempotency reconciliation: file already on disk and probes?
+	if _, err := os.Stat(finalPath); err == nil {
+		if payload := probeEpisodeAudio(finalPath); payload != nil {
+			return w.markDownloaded(ctx, ep, &feed, finalPath, payload)
+		}
+		// Unprobeable stray: fall through and re-fetch over it.
+	}
+
+	// ---- fetch to temp ----
+	tmpPath := finalPath + ".part-" + randomSuffix()
+	cleanup := func() { _ = os.Remove(tmpPath) }
+
+	res, err := guardFetch(ctx, ep.EnclosureUrl.String, "", "", maxEnclosureCapBytes)
+	if err != nil {
+		cleanup()
+		return fmt.Errorf("download failed: %w", classifyFetchError(err))
+	}
+	if err := os.WriteFile(tmpPath, res.Body, 0o644); err != nil {
+		cleanup()
+		return fmt.Errorf("write: %w", err)
+	}
+
+	// ---- validation ----
+	info, err := os.Stat(tmpPath)
+	if err != nil || info.Size() == 0 {
+		cleanup()
+		return fmt.Errorf("rejected: empty download")
+	}
+	if info.Size() > maxEnclosureCapBytes {
+		cleanup()
+		return fmt.Errorf("rejected: enclosure exceeds %d byte cap", maxEnclosureCapBytes)
+	}
+	if !looksLikeAudioFile(tmpPath) {
+		cleanup()
+		return fmt.Errorf("rejected: downloaded file does not look like audio media")
+	}
+
+	// ---- ffprobe reconcile + embedded cover (scanner machinery reuse) ----
+	payload := probeEpisodeAudio(tmpPath)
+
+	// ---- move into place (crash-safe: rename is atomic on-same-fs) ----
+	if err := os.Rename(tmpPath, finalPath); err != nil {
+		cleanup()
+		return fmt.Errorf("finalize: %w", err)
+	}
+
+	return w.markDownloaded(ctx, ep, &feed, finalPath, payload)
+}
+
+// markDownloaded is the SUCCESS-ONLY row write (invariant 7): the file
+// is on disk and (when ffprobe is available) probed before this runs.
+func (w *PodcastDownloadWorker) markDownloaded(ctx context.Context, ep database.MediaItems, feed *database.PodcastFeeds, absPath string, payload *AudioMetadataPayload) error {
+	info, err := os.Stat(absPath)
+	if err != nil {
+		return fmt.Errorf("post-move stat: %w", err)
+	}
+
+	// Embedded cover extraction (optional enrichment; scanner precedent).
+	coverRel := pgtype.Text{}
+	base := strings.TrimSuffix(absPath, filepath.Ext(absPath))
+	if coverAbs := extractEmbeddedAudioCover(absPath, base); coverAbs != "" {
+		if root, err := w.svc.downloadRootAbs(ctx, feed); err == nil {
+			if rel, err := filepath.Rel(root, coverAbs); err == nil {
+				coverRel = pgtypeText(rel)
+			}
+		}
+	}
+
+	var chapterCount pgtype.Int4
+	if payload != nil && len(payload.Chapters) > 0 {
+		chapterCount = pgtypeInt4(int32(len(payload.Chapters)))
+	}
+
+	_, err = w.db.MarkPodcastEpisodeDownloaded(ctx, database.MarkPodcastEpisodeDownloadedParams{
+		ID:           ep.ID,
+		FilePath:     ep.FilePath,
+		FileSize:     pgtypeInt8(info.Size()),
+		MimeType:     pgtypeText(strings.TrimSpace(pickMimeType(ep, absPath))),
+		CoverImagePath: coverRel,
+		AudioMetadata: audioPayloadJSONB(payload),
+		ChapterCount: chapterCount,
+	})
+	if err != nil {
+		return fmt.Errorf("mark downloaded: %w", err)
+	}
+	log.Printf("[PODCAST-DL] downloaded %q (%s) for feed %s", ep.Title, ep.FilePath, feed.Title)
+	return nil
+}
+
+// probeEpisodeAudio shells out to ffprobe and builds the audio_metadata
+// payload (scanner machinery reuse; nil when ffprobe is absent — the
+// episode still downloads, facts degrade to declared values).
+func probeEpisodeAudio(path string) *AudioMetadataPayload {
+	report := runFFProbe(path)
+	if report == nil {
+		return nil
+	}
+	payload := &AudioMetadataPayload{}
+	if report.Format.Duration > 0 {
+		payload.DurationSeconds = float64(report.Format.Duration)
+	}
+	if report.Format.BitRate > 0 {
+		payload.BitRate = int64(report.Format.BitRate)
+	}
+	if len(report.Streams) > 0 {
+		profile := report.Streams[0].CodecName
+		if p := report.Streams[0].Profile; p != "" && p != "unknown" {
+			profile += " / " + p
+		}
+		payload.CodecProfile = profile
+	}
+	for _, ch := range report.Chapters {
+		payload.Chapters = append(payload.Chapters, AudioChapter{
+			Index: ch.ID,
+			Title: ch.Tags.Title,
+			Start: float64(ch.StartTime),
+			End:   float64(ch.EndTime),
+		})
+	}
+	return payload
+}
+
+// looksLikeAudioFile sniffs content: an audio container starts with
+// known magic bytes, or (ID3v1/v2, ADTS) has recognizable heads. HTML
+// error pages and JSON never pass.
+func looksLikeAudioFile(path string) bool {
+	f, err := os.Open(path)
+	if err != nil {
+		return false
+	}
+	defer func() { _ = f.Close() }()
+	head := make([]byte, 16)
+	n, _ := io.ReadFull(f, head)
+	if n < 4 {
+		return false
+	}
+	h := head[:n]
+	switch {
+	// MP3: ID3 tag or MPEG frame sync.
+	case string(h[:3]) == "ID3":
+		return true
+	case h[0] == 0xFF && h[1]&0xE0 == 0xE0:
+		return true
+	// MP4/M4A/M4B: ftyp box.
+	case string(h[4:8]) == "ftyp":
+		return true
+	// FLAC
+	case string(h[:4]) == "fLaC":
+		return true
+	// OGG/Opus
+	case string(h[:4]) == "OggS":
+		return true
+	// WAV/RIFF
+	case string(h[:4]) == "RIFF":
+		return true
+	// AAC ADTS (sync word)
+	case h[0] == 0xFF && (h[1]&0xF6) == 0xF0:
+		return true
+	}
+	return false
+}
+
+// pickMimeType prefers the feed-declared type; falls back by extension.
+func pickMimeType(ep database.MediaItems, absPath string) string {
+	if ep.MimeType.Valid && strings.HasPrefix(ep.MimeType.String, "audio/") {
+		return ep.MimeType.String
+	}
+	switch strings.ToLower(filepath.Ext(absPath)) {
+	case ".m4a", ".m4b":
+		return "audio/mp4"
+	case ".mp3":
+		return "audio/mpeg"
+	case ".flac":
+		return "audio/flac"
+	case ".ogg", ".oga", ".opus":
+		return "audio/ogg"
+	case ".wav":
+		return "audio/wav"
+	case ".aac":
+		return "audio/aac"
+	}
+	return "audio/mpeg"
+}
+
+// audioPayloadJSONB marshals the probed payload for the JSONB column
+// (nil payload → NULL: keep the column honest, never an empty object).
+func audioPayloadJSONB(p *AudioMetadataPayload) []byte {
+	if p == nil {
+		return nil
+	}
+	b, err := json.Marshal(p)
+	if err != nil {
+		return nil
+	}
+	return b
+}
+
+// randomSuffix disambiguates concurrent temp files (single worker, but
+// a crashed worker's stale .part files must never collide).
+func randomSuffix() string {
+	sum := sha256.Sum256([]byte(time.Now().Format(time.RFC3339Nano) + fmt.Sprint(os.Getpid())))
+	return hex.EncodeToString(sum[:4])
+}