bookhoard/internal/services/podcast_worker.go

// Podcast download worker (plan §7 / D6 / D8 / D19; pool since §0 item
// 4): a download-manager-style pool 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).
// Each queue ticket (download_queue_pos, a SEQUENCE nextval at
// enqueue) makes the enqueue loop's ordering the download order.
//
// Concurrency: up to podcast_download_concurrency episodes in flight
// server-wide (default 4, admin-tunable live 1–16). Fair share: the
// claim prefers the feed least-recently-served, so several books
// download in parallel while chapters within a book land in ticket
// order (start listening while the rest catch up). Bandwidth: an
// optional shared token bucket (podcast_download_speed_mbps, 0 =
// unlimited) paces all workers together.
//
// 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"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"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
	settings *database.SettingsRegistry
	wake     chan struct{}

	// Fair-share bookkeeping (§0 item 4): feed UUID string → last
	// time the pool served it. Claim prefers the smallest value;
	// never-served feeds (missing key) go first. Guarded by fairMu.
	fairMu     sync.Mutex
	lastServed map[string]time.Time

	// Shared bandwidth bucket across all pool workers (§0 item 4).
	limiter bandwidthLimiter

	// Live byte-progress slots for in-flight downloads (issue #1).
	// Keyed by episode UUID string; guarded by progMu.
	progMu   sync.Mutex
	progress map[string]*episodeProgress
}

func NewPodcastDownloadWorker(db *database.Queries, svc *PodcastService) *PodcastDownloadWorker {
	return &PodcastDownloadWorker{
		db:         db,
		svc:        svc,
		wake:       make(chan struct{}, 1),
		lastServed: make(map[string]time.Time),
		progress:   make(map[string]*episodeProgress),
	}
}

// SetSettings wires the live settings registry (admin tunables:
// concurrency + bandwidth cap). Without it the worker runs on the
// registered defaults (4 in flight, unlimited speed) — the test
// harness constructs workers without settings.
func (w *PodcastDownloadWorker) SetSettings(r *database.SettingsRegistry) {
	w.settings = r
}

// concurrency re-reads the live setting each drain cycle.
func (w *PodcastDownloadWorker) concurrency() int {
	if w.settings == nil {
		return 4 // registered default
	}
	if n := w.settings.PodcastDownloadConcurrency(); n >= 1 {
		return n
	}
	return 4
}

// refillLimiter re-syncs the shared bucket's rate (bytes/sec) from the
// live setting; 0 = unlimited.
func (w *PodcastDownloadWorker) refillLimiter() {
	if w.settings == nil {
		return
	}
	w.limiter.setRate(int64(w.settings.PodcastDownloadSpeedMbps() * 125000)) // Mbps → bytes/sec
}

// 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)
		}
	}()
}

// DrainNow runs one bounded pool round synchronously — the test hook
// and the future admin "retry now" seam (normally the wake channel
// drives rounds).
func (w *PodcastDownloadWorker) DrainNow(ctx context.Context) {
	w.drain(ctx)
}

// drain runs one pool round: up to concurrency() claim-loop goroutines
// each claim (fair-share) → download → settle, until the queue is
// empty. The cap setting is re-read every round — live-tunable.
func (w *PodcastDownloadWorker) drain(ctx context.Context) {
	if ctx.Err() != nil {
		return
	}
	w.refillLimiter()
	limit := w.concurrency()
	var wg sync.WaitGroup
	for i := 0; i < limit; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				if ctx.Err() != nil {
					return
				}
				ep, ok := w.claimNext(ctx)
				if !ok {
					return // queue drained (or nothing claimable)
				}
				w.runOne(ctx, ep)
			}
		}()
	}
	wg.Wait()
}

// claimNext atomically claims the next queued episode, preferring the
// feed least-recently-served (fair share — books proceed in parallel)
// and the earliest queue ticket within a feed (chapters land in
// enqueue order). Returns ok=false when nothing is claimable.
func (w *PodcastDownloadWorker) claimNext(ctx context.Context) (database.MediaItems, bool) {
	actx, err := w.svc.adminCtx(ctx)
	if err != nil {
		return database.MediaItems{}, false
	}
	cands, err := w.db.ListQueuedPodcastEpisodes(actx)
	if err != nil || len(cands) == 0 {
		return database.MediaItems{}, false
	}

	// Group per feed preserving ticket order (the SQL scan is ordered).
	perFeed := make(map[pgtype.UUID][]pgtype.UUID, 8)
	for _, c := range cands {
		perFeed[c.SourceFeedID] = append(perFeed[c.SourceFeedID], c.ID)
	}

	w.fairMu.Lock()
	defer w.fairMu.Unlock()
	pickFeed := pickFeedToServe(perFeed, w.lastServed)
	if !pickFeed.Valid {
		return database.MediaItems{}, false
	}

	// Claim the head ticket of the chosen feed; if another claimant
	// won it meanwhile (retry raced), fall through the feed's tickets.
	for _, id := range perFeed[pickFeed] {
		ep, err := w.db.ClaimQueuedPodcastEpisode(actx, id)
		if err != nil {
			continue // claimed elsewhere / state moved on
		}
		w.lastServed[pickFeed.String()] = time.Now()
		return ep, true
	}
	return database.MediaItems{}, false
}

// pickFeedToServe chooses the feed least-recently-served (never-served
// first); among never-served feeds the first encountered wins (any is
// equally fair). Unit-testable seam for the fair-share rule.
func pickFeedToServe(perFeed map[pgtype.UUID][]pgtype.UUID, lastServed map[string]time.Time) pgtype.UUID {
	var pick pgtype.UUID
	var best time.Time
	haveNever := false
	for feed := range perFeed {
		last, seen := lastServed[feed.String()]
		if !seen {
			if !haveNever {
				pick, haveNever = feed, true
			}
			continue
		}
		if haveNever {
			continue // a never-served feed already won
		}
		if !pick.Valid || last.Before(best) {
			pick, best = feed, last
		}
	}
	return pick
}

// runOne downloads a claimed episode and settles the row either way
// (failed/rejected: state + reason persist; metadata/file facts
// untouched — invariant 7, write-after-success only).
func (w *PodcastDownloadWorker) runOne(ctx context.Context, ep database.MediaItems) {
	actx, err := w.svc.adminCtx(ctx)
	if err != nil {
		return
	}
	if err := w.runPipeline(actx, ep); err != nil {
		_ = 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)
	}
}

// AdminCtx exposes the worker's first-admin identity (tests use it for
// worker-scope DB statements).
func (w *PodcastDownloadWorker) AdminCtx(ctx context.Context) (context.Context, error) {
	return w.svc.adminCtx(ctx)
}

// 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
}

// QueueFeedDownloadsAs is Download All honoring the CALLER's listening
// order (§0 item 5b): episodes queue in the feed's effective order (the
// caller's override, feed-native fallback). Tickets (§0 item 4) make
// that queue order the download order — "getting the last chapter
// first doesn't help me". userID zero = no override (feed order).
func (w *PodcastDownloadWorker) QueueFeedDownloadsAs(ctx context.Context, userID, 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
	}

	order := "native"
	if userID.Valid {
		order = w.orderOverrideFor(actx, userID, feedID)
	}
	// THE order helper (§0 item 5a) — same semantics as the sequence,
	// reader chain, and page list; no per-surface sort drift.
	OrderEpisodes(eps, order,
		func(ep database.MediaItems) int32 { return ep.FeedPosition.Int32 },
		func(ep database.MediaItems) string {
			if ep.DatePublished.Valid {
				return ep.DatePublished.Time.Format("2006-01-02")
			}
			return ""
		},
	)

	queued := 0
	for _, ep := range eps {
		if _, err := w.db.SetPodcastEpisodeDownloadQueued(actx, ep.ID); err == nil {
			queued++
		}
	}
	w.Wake()
	return queued, nil
}

// orderOverrideFor reads a user's podcast_order override via the
// reader_settings KV (the D5 storage; native when absent/invalid).
func (w *PodcastDownloadWorker) orderOverrideFor(ctx context.Context, userID, feedID pgtype.UUID) string {
	row, err := w.db.GetUserSettingByKey(ctx, database.GetUserSettingByKeyParams{
		UserID: userID, SettingKey: "podcast_order:" + feedID.String(),
	})
	if err != nil {
		return "native"
	}
	var v struct {
		Order string `json:"order"`
	}
	if json.Unmarshal(row, &v) != nil || !ValidPodcastEpisodeOrder(v.Order) {
		return "native"
	}
	return v.Order
}

// 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
}

// 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 (streamed with byte-progress tracking) ----
	// Enclosures stream straight to disk through a counting loop — no
	// whole-body buffering — so the episode list can render a real
	// progress bar while the worker runs (issue #1).
	tmpPath := finalPath + ".part-" + randomSuffix()
	cleanup := func() { _ = os.Remove(tmpPath) }

	received, total, fetchErr := w.streamEnclosure(ctx, ep, tmpPath)
	w.endProgress(ep.ID) // terminal either way; the row state is the truth
	if fetchErr != nil {
		cleanup()
		return fetchErr
	}
	_, _ = received, total

	// ---- 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)
}

// ---- byte-progress tracking (issue #1) ----
// In-memory, deliberately NOT persisted: a downloading row renders a
// bar from this snapshot while the worker holds it; the DB row's state
// remains the durable truth (queued/downloading/downloaded).

type episodeProgress struct {
	Received int64
	Total    int64 // 0 = unknown (no Content-Length, no declared length)
}

// beginProgress registers a download slot (total 0 = unknown until the
// response's Content-Length arrives).
func (w *PodcastDownloadWorker) beginProgress(episodeID pgtype.UUID, total int64) {
	w.progMu.Lock()
	defer w.progMu.Unlock()
	w.progress[episodeID.String()] = &episodeProgress{Total: total}
}

// addProgress accumulates streamed bytes.
func (w *PodcastDownloadWorker) addProgress(episodeID pgtype.UUID, n int64) {
	w.progMu.Lock()
	defer w.progMu.Unlock()
	if p, ok := w.progress[episodeID.String()]; ok {
		p.Received += n
		if p.Total == 0 {
			// Total unknown at start; keep received-only (bar renders
			// indeterminate).
		}
	}
}

// setProgressTotal records the authoritative total once known.
func (w *PodcastDownloadWorker) setProgressTotal(episodeID pgtype.UUID, total int64) {
	w.progMu.Lock()
	defer w.progMu.Unlock()
	if p, ok := w.progress[episodeID.String()]; ok && total > 0 {
		p.Total = total
	}
}

// endProgress drops the slot (terminal state reached — the row itself
// now tells the story).
func (w *PodcastDownloadWorker) endProgress(episodeID pgtype.UUID) {
	w.progMu.Lock()
	defer w.progMu.Unlock()
	delete(w.progress, episodeID.String())
}

// EpisodeProgress reports the live byte counts for a downloading
// episode (ok=false when nothing is in flight for it).
func (w *PodcastDownloadWorker) EpisodeProgress(episodeID pgtype.UUID) (received, total int64, ok bool) {
	w.progMu.Lock()
	defer w.progMu.Unlock()
	p, ok := w.progress[episodeID.String()]
	if !ok {
		return 0, 0, false
	}
	return p.Received, p.Total, true
}

// ---- shared bandwidth bucket (§0 item 4) ----
// One leaky-bucket-with-reservations limiter across ALL pool workers:
// each caller reserves the next slot of bandwidth and sleeps outside
// the lock — reservation order = transmission order, and N workers
// share one cap instead of N caps.

type bandwidthLimiter struct {
	mu   sync.Mutex
	rate int64     // bytes per second; <=0 = unlimited
	next time.Time // when the currently-reserved budget runs out
}

func (l *bandwidthLimiter) setRate(bytesPerSec int64) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.rate = bytesPerSec
}

// take reserves n bytes of bandwidth and returns how long the caller
// must wait before sending them (0 when free flow).
func (l *bandwidthLimiter) take(n int64) time.Duration {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.rate <= 0 {
		return 0
	}
	now := time.Now()
	if l.next.Before(now) {
		l.next = now
	}
	d := time.Duration(float64(n) / float64(l.rate) * float64(time.Second))
	start := l.next // my window begins when earlier reservations end
	l.next = start.Add(d)
	return start.Sub(now)
}

// streamEnclosure performs the SSRF-guarded enclosure download straight
// to tmpPath while updating the progress slot. Returns bytes received
// and the authoritative total (0 when unknown). Scheme and size guards
// match guardFetch; redirects re-dial through the same client so the
// IP guard re-applies per hop.
func (w *PodcastDownloadWorker) streamEnclosure(ctx context.Context, ep database.MediaItems, tmpPath string) (int64, int64, error) {
	u, err := url.Parse(ep.EnclosureUrl.String)
	if err != nil {
		return 0, 0, fmt.Errorf("download failed: invalid enclosure URL: %w", err)
	}
	if err := guardURLScheme(u); err != nil {
		return 0, 0, fmt.Errorf("download failed: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, ep.EnclosureUrl.String, nil)
	if err != nil {
		return 0, 0, fmt.Errorf("download failed: %w", err)
	}
	req.Header.Set("User-Agent", "Bookhoard/1.0 (podcast fetcher; +https://github.com/bookhoard)")

	// Declared total (feed's enclosure length) seeds the bar until the
	// response's Content-Length replaces it.
	declared := int64(0)
	if ep.FileSize.Valid {
		declared = ep.FileSize.Int64
	}
	w.beginProgress(ep.ID, declared)

	resp, err := podcastHTTPClientFactory().Do(req)
	if err != nil {
		return 0, declared, fmt.Errorf("download failed: %w", classifyFetchError(err))
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		return 0, declared, fmt.Errorf("download failed: HTTP %d for enclosure", resp.StatusCode)
	}

	total := resp.ContentLength // -1 when unknown
	if total < 0 {
		total = declared
	}
	w.setProgressTotal(ep.ID, total)

	out, err := os.Create(tmpPath)
	if err != nil {
		return 0, total, fmt.Errorf("write: %w", err)
	}
	defer func() { _ = out.Close() }()

	buf := make([]byte, 64<<10)
	var received int64
	for {
		n, readErr := resp.Body.Read(buf)
		if n > 0 {
			received += int64(n)
			if received > maxEnclosureCapBytes {
				return received, total, fmt.Errorf("rejected: enclosure exceeds %d byte cap", maxEnclosureCapBytes)
			}
			// Shared bandwidth cap: pace against the server-wide
			// bucket (0 rate = unlimited, take returns 0).
			if wait := w.limiter.take(int64(n)); wait > 0 {
				select {
				case <-ctx.Done():
					return received, total, fmt.Errorf("download failed: cancelled")
				case <-time.After(wait):
				}
			}
			w.addProgress(ep.ID, int64(n))
			if _, werr := out.Write(buf[:n]); werr != nil {
				return received, total, fmt.Errorf("write: %w", werr)
			}
		}
		if readErr == io.EOF {
			break
		}
		if readErr != nil {
			return received, total, fmt.Errorf("download failed: %w", readErr)
		}
	}
	return received, total, nil
}

// 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])
}