feat(podcasts): real-time episode list — byte progress bars, direct swaps (issues #1/#3)
WHY: downloads needed manual refreshes to see progress; the 204 + HX-Trigger event indirection missed its listener, and the worker buffered whole enclosures so no progress existed to show. - Worker streams enclosures to disk through a counting loop (64KiB chunks, scheme/redirect/size guards preserved) with an in-memory per-episode progress snapshot (received/total; total from Content-Length, seeded by the feed's declared length). Deliberately transient — the DB row state remains the durable truth. - Episode-list renderer seam: the router injects a closure into PodcastHandler (handlers can't import templates — MediaDetail cycle) so htmx download/retry/Download All/refresh/order/policy actions answer with the REFRESHED partial directly (200, hx-target= #episode-list) — no events to miss. Non-HX callers keep JSON; tests without a renderer keep 204+trigger. - Polling tightens to 2s while downloads are active; downloading rows render a percentage bar + MB counter (quote-free inline style per templ gotcha #8) or an indeterminate bar when the server sends no length. - Issue #3: Download All (button AND the add-feed checkbox) flips the CALLER's subscription policy to 'all' — per-user, never global (D6: files shared, policies not) — and the auto-download select out-of-band-swaps so the dropdown shows what the user is getting. They can turn it down afterwards; the flip is honest communication, not coercion (user-approved 2026-10-29). PINs: Download All with HX answers 200 + episode-list + oob select and flips newest → all; add-feed download_all sets policy all.
This commit is contained in:
1 parent
ffc80f7ba7
commit
30766020c8
8 files changed
+545
-57
No files matched your search
@@ -31,7 +31,10 @@ import (
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"bookhoard/internal/database"
|
||||
@@ -48,13 +51,19 @@ 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),
|
||||
db: db,
|
||||
svc: svc,
|
||||
wake: make(chan struct{}, 1),
|
||||
progress: make(map[string]*episodeProgress),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -287,19 +296,20 @@ func (w *PodcastDownloadWorker) runPipeline(ctx context.Context, ep database.Med
|
||||
// Unprobeable stray: fall through and re-fetch over it.
|
||||
}
|
||||
|
||||
// ---- fetch to temp ----
|
||||
// ---- 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) }
|
||||
|
||||
res, err := guardFetch(ctx, ep.EnclosureUrl.String, "", "", maxEnclosureCapBytes)
|
||||
if err != nil {
|
||||
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 fmt.Errorf("download failed: %w", classifyFetchError(err))
|
||||
}
|
||||
if err := os.WriteFile(tmpPath, res.Body, 0o644); err != nil {
|
||||
cleanup()
|
||||
return fmt.Errorf("write: %w", err)
|
||||
return fetchErr
|
||||
}
|
||||
_, _ = received, total
|
||||
|
||||
// ---- validation ----
|
||||
info, err := os.Stat(tmpPath)
|
||||
@@ -328,6 +338,139 @@ func (w *PodcastDownloadWorker) runPipeline(ctx context.Context, ep database.Med
|
||||
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 {
|
||||
|
||||
Reference in new issue
Block a user