bookhoard/internal/services/podcast_worker.go

// 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"
	"net/http"
	"net/url"
	"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
	wake chan struct{}

	// 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),
		progress: make(map[string]*episodeProgress),
	}
}

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

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