From 4288a86b2d67e67935088af2c8c8c5be219002b0 Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Thu, 29 Oct 2026 16:03:22 -0400 Subject: [PATCH] =?UTF-8?q?feat(podcasts):=20download=20worker=20=E2=80=94?= =?UTF-8?q?=20state=20machine,=20idempotency,=20policy=20union=20(task=201?= =?UTF-8?q?.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 484 ++++++++++++++++++++++++++++ 1 file changed, 484 insertions(+) create mode 100644 internal/services/podcast_worker.go diff --git a/internal/services/podcast_worker.go b/internal/services/podcast_worker.go new file mode 100644 index 00000000..d8b41b8a --- /dev/null +++ b/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]) +}