feat(podcasts): RSS fetcher — parse, SSRF guards, upsert-without-delete (task 1.4)

The fetcher is the SOLE writer of feed/episode metadata (D9). Three
files, no new deps (stdlib encoding/xml + net/http):

podcast_rss.go — RSS 2.0 + iTunes + Podcasting-2.0 structs. iTunes
fields are namespace-qualified in tags (local names collide with core
RSS: title/subtitle/author/summary/explicit/image/duration);
podcast:* fields match on local name (collision-free, tolerant of both
declared namespace URIs). Description/author fallback chains per §5;
itunes:duration parses seconds/HH:MM:SS/MM:SS/prose; pubDate accepts
the RFC-822/1123 spellings feeds actually emit.

podcast_http.go — the security boundary for user-supplied URLs:
https-only on every hop (CheckRedirect), private/loopback/link-local/
multicast IP ranges blocked at DIAL time (Dialer.Control sees the
RESOLVED address — DNS rebinding into 127.0.0.1 is refused at the
socket), 10-redirect cap, hard body caps (25MiB feeds / 20MiB artwork /
1GiB enclosures), no proxy.

podcast_service.go — fetch → parse → upsert. Episodes upsert
metadata-only keyed on (source_feed_id, episode_guid); rows for items
that vanish from the XML are NEVER deleted (D14 search covers
historical episodes). itunes:new-feed-url + permanent HTTP redirects
rewrite feed_url in place (row/subscriptions/episodes stay put);
podcast:guid dedupe reuses the canonical row instead of forking.
Artwork downloads feed-level under <feed-slug>/artwork.<ext>.
AddFeed authorizes ONLY on explicit arguments (library visible +
allow_member_feeds for non-admins — the D20 scoping model); detached
work runs under the first-admin identity via adminCtx (invariant 12).
Feed layout: <download-root>/<feed-slug>/<pos:04d>-<slug>.<ext> so
ServeFile/Range streaming resolve unchanged.
This commit is contained in:
John O'Keefe committed 2026-10-29 16:03:22 -04:00
1 parent 9a148c2344
commit 85a9873a75
3 files changed
+1209

No files matched your search

+190
View File
@@ -0,0 +1,190 @@
// SSRF-guarded HTTP fetching for the podcast fetcher (plan §7).
//
// This is the code path that turns USER-SUPPLIED URLs into server
// fetches — treat it as a security boundary, not a convenience wrapper:
//
// - https-only schemes (feed URL, enclosure, artwork, EVERY redirect hop)
// - private/loopback/link-local/multicast/unspecified IP ranges blocked
// at DIAL time via Dialer.Control — the check sees the RESOLVED
// address, so a DNS name that rebinding-resolves to 127.0.0.1 (or a
// redirect into the internal network) is refused at the socket, not
// merely at URL-parse time
// - hard response size caps on every body (feed XML, artwork, enclosures)
//
// Go's redirect follower re-dials through the same dialer per hop, so
// the IP guard re-applies automatically; the scheme guard re-applies
// via CheckRedirect.
package services
import (
"context"
"crypto/tls"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"syscall"
"time"
)
// Response size caps (bytes). Feeds are XML; artwork is an image;
// enclosures are audio episodes — the enclosure cap is deliberately the
// large one (audiobook chapters run long) but still bounded.
const (
maxFeedSizeBytes = 25 << 20 // 25 MiB
maxArtworkSizeBytes = 20 << 20 // 20 MiB
maxEnclosureCapBytes = 1 << 30 // 1 GiB
)
// ErrBlockedPrivateAddress marks SSRF-guard rejections (dial-time IP or
// scheme checks). Handlers surface it as a validation failure, and the
// download worker records it as the download_error text.
var ErrBlockedPrivateAddress = errors.New("address blocked: private, loopback, or otherwise non-public")
// podcastHTTPClient builds the shared guard client. Timeout is generous
// (podcast servers are slow) but bounded; enclosures stream through a
// LimitReader so a lying Content-Length cannot exhaust disk.
func podcastHTTPClient() *http.Client {
dialer := &net.Dialer{
Timeout: 15 * time.Second,
Control: func(network, address string, _ syscall.RawConn) error {
return guardDialAddress(address)
},
}
transport := &http.Transport{
DialContext: dialer.DialContext,
TLSHandshakeTimeout: 15 * time.Second,
ResponseHeaderTimeout: 60 * time.Second,
// Enclosure downloads are long; idle conn reuse still applies.
IdleConnTimeout: 90 * time.Second,
TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12},
Proxy: nil, // never route through a user-controllable proxy
}
return &http.Client{
Transport: transport,
Timeout: 15 * time.Minute, // outer bound (biggest enclosures on slow lines)
CheckRedirect: func(req *http.Request, via []*http.Request) error {
if len(via) >= 10 {
return fmt.Errorf("stopped after 10 redirects")
}
if err := guardURLScheme(req.URL); err != nil {
return err
}
return nil
},
}
}
// guardDialAddress validates the RESOLVED host:port at dial time.
func guardDialAddress(address string) error {
host, _, err := net.SplitHostPort(address)
if err != nil {
return fmt.Errorf("dial guard: %w", err)
}
ip := net.ParseIP(host)
if ip == nil {
return fmt.Errorf("dial guard: unparseable address %q", host)
}
if !isPublicIP(ip) {
return ErrBlockedPrivateAddress
}
return nil
}
// guardURLScheme enforces https on any URL we are about to touch.
func guardURLScheme(u *url.URL) error {
if u == nil || u.Scheme != "https" {
return fmt.Errorf("only https feed/enclosure URLs are supported")
}
return nil
}
// isPublicIP reports whether ip is a routable public unicast address.
// Everything loopback/private/link-local/multicast/unspecified —
// including IPv4-mapped IPv6 spellings of those — is rejected.
func isPublicIP(ip net.IP) bool {
if ip == nil {
return false
}
// Normalize IPv4-mapped IPv6 (::ffff:10.0.0.1) to its v4 form so
// the v4 range checks below actually fire.
if ip4 := ip.To4(); ip4 != nil {
ip = ip4
}
if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() ||
ip.IsLinkLocalMulticast() || ip.IsInterfaceLocalMulticast() ||
ip.IsMulticast() || ip.IsUnspecified() {
return false
}
return true
}
// guardFetch performs a GET with the scheme/IP guards, conditional-GET
// validators, and a hard body cap. Returns the body bytes, the FINAL
// URL after redirects (new-feed-url / migration detection), validators
// to persist, and whether the server answered 304 Not Modified.
type fetchResult struct {
Body []byte
FinalURL string
ETag string
LastModified string
NotModified bool
StatusCode int
}
func guardFetch(ctx context.Context, rawURL, etag, lastModified string, maxSize int64) (*fetchResult, error) {
u, err := url.Parse(rawURL)
if err != nil {
return nil, fmt.Errorf("invalid URL: %w", err)
}
if err := guardURLScheme(u); err != nil {
return nil, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
if err != nil {
return nil, err
}
// A recognizable UA; some podcast hosts 403 empty/default agents.
req.Header.Set("User-Agent", "Bookhoard/1.0 (podcast fetcher; +https://github.com/bookhoard)")
req.Header.Set("Accept", "*/*")
if etag != "" {
req.Header.Set("If-None-Match", etag)
}
if lastModified != "" {
req.Header.Set("If-Modified-Since", lastModified)
}
resp, err := podcastHTTPClient().Do(req)
if err != nil {
// Unwrap redirect/scheme guard errors so callers can classify.
return nil, err
}
defer func() { _ = resp.Body.Close() }()
res := &fetchResult{
FinalURL: resp.Request.URL.String(),
ETag: resp.Header.Get("ETag"),
LastModified: resp.Header.Get("Last-Modified"),
StatusCode: resp.StatusCode,
}
if resp.StatusCode == http.StatusNotModified {
res.NotModified = true
return res, nil
}
if resp.StatusCode != http.StatusOK {
return res, fmt.Errorf("fetch %s: HTTP %d", rawURL, resp.StatusCode)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, maxSize+1))
if err != nil {
return res, fmt.Errorf("fetch %s: %w", rawURL, err)
}
if int64(len(body)) > maxSize {
return res, fmt.Errorf("fetch %s: response exceeds %d byte cap", rawURL, maxSize)
}
res.Body = body
return res, nil
}
+257
View File
@@ -0,0 +1,257 @@
// RSS 2.0 + iTunes + Podcasting-2.0 feed parsing for the podcast
// fetcher (plan podcast-integration.md §5/§7). Stdlib encoding/xml only
// — no new deps.
//
// Namespace strategy: iTunes fields MUST be namespace-qualified in the
// struct tags (the itunes URI) because their local names collide with
// core RSS ("title", "subtitle", "author", "summary", "explicit",
// "image", "duration"). Podcasting-2.0 fields (podcast:guid,
// podcast:transcript, ...) have collision-free local names, so they
// match on local name — encoding/xml matches Name.Local when the tag
// carries no namespace, which also tolerates both declared namespace
// URIs the ecosystem uses.
//
// Out of scope at parse time (plan §5): podcast:value, liveItem,
// soundbite, alternateEnclosure, socialInteract, remoteItem/podroll,
// locked/block, txt, chat. lastBuildDate/ttl are poll hints only.
package services
import (
"encoding/xml"
"fmt"
"strconv"
"strings"
"time"
)
// itunesNS is the canonical iTunes podcast namespace URI.
const itunesNS = "http://www.itunes.com/dtds/podcast-1.0.dtd"
// PodcastFeedXML is one parsed RSS document.
type PodcastFeedXML struct {
XMLName xml.Name `xml:"rss"`
Channel PodcastChannelXML `xml:"channel"`
}
type PodcastChannelXML struct {
Title string `xml:"title"`
Link string `xml:"link"`
Language string `xml:"language"`
Description string `xml:"description"`
// iTunes channel namespace.
ITunesAuthor string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd author"`
ITunesOwner struct {
Name string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd name"`
Email string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd email"`
} `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd owner"`
ITunesImage struct {
Href string `xml:"href,attr"`
} `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd image"`
ITunesExplicit string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd explicit"`
ITunesType string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd type"`
ITunesCategory struct {
Text string `xml:"text,attr"`
Subcategory struct {
Text string `xml:"text,attr"`
} `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd category"`
} `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd category"`
ITunesNewFeedURL string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd new-feed-url"`
ITunesComplete string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd complete"`
// Podcasting 2.0 (matched on local name; see file comment).
PodcastGUID string `xml:"guid"`
PodcastMedium string `xml:"medium"`
LastBuildDate string `xml:"lastBuildDate"`
Items []PodcastItemXML `xml:"item"`
}
type PodcastItemXML struct {
GUID string `xml:"guid"`
Title string `xml:"title"`
ITunesTitle string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd title"`
Subtitle string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd subtitle"`
// Description resolution order (plan §5): description >
// itunes:summary > content:encoded.
Description string `xml:"description"`
ITunesSummary string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd summary"`
ContentEncoded string `xml:"encoded"`
// Item author: itunes:author preferred, then RSS author, then dc:creator.
ITunesAuthor string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd author"`
Author string `xml:"author"`
Creator string `xml:"http://purl.org/dc/elements/1.1/ creator"`
ITunesImage struct {
Href string `xml:"href,attr"`
} `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd image"`
ITunesExplicit string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd explicit"`
ITunesDuration string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd duration"`
ITunesEpisode string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd episode"`
ITunesSeason string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd season"`
PubDate string `xml:"pubDate"`
Enclosure struct {
URL string `xml:"url,attr"`
Length int64 `xml:"length,attr"`
Type string `xml:"type,attr"`
} `xml:"enclosure"`
// Podcasting 2.0: store-now, feature-later URLs.
Transcript struct {
URL string `xml:"url,attr"`
} `xml:"transcript"`
Chapters struct {
URL string `xml:"url,attr"`
} `xml:"chapters"`
Persons []struct {
Href string `xml:"href,attr"`
Role string `xml:"role,attr"`
} `xml:"person"`
}
// ParsePodcastFeed decodes a feed document. Returns an error when the
// XML is unparseable or the document is not an RSS channel.
func ParsePodcastFeed(data []byte) (*PodcastFeedXML, error) {
var feed PodcastFeedXML
if err := xml.Unmarshal(data, &feed); err != nil {
return nil, fmt.Errorf("feed parse: %w", err)
}
if feed.Channel.Title == "" && len(feed.Channel.Items) == 0 {
return nil, fmt.Errorf("feed parse: not an RSS channel (no title, no items)")
}
return &feed, nil
}
// EpisodeTitle prefers itunes:title over the raw RSS title (plan §5).
func (it *PodcastItemXML) EpisodeTitle() string {
if t := strings.TrimSpace(it.ITunesTitle); t != "" {
return t
}
return strings.TrimSpace(it.Title)
}
// EpisodeDescription resolves the description fallback chain.
func (it *PodcastItemXML) EpisodeDescription() string {
for _, c := range []string{it.Description, it.ITunesSummary, it.ContentEncoded} {
if t := strings.TrimSpace(c); t != "" {
return t
}
}
return ""
}
// EpisodeAuthor resolves the author fallback chain.
func (it *PodcastItemXML) EpisodeAuthor() string {
for _, c := range []string{it.ITunesAuthor, it.Author, it.Creator} {
if t := strings.TrimSpace(c); t != "" {
return t
}
}
return ""
}
// EpisodeExplicit maps the itunes:explicit spelling variants to bool.
func (it *PodcastItemXML) EpisodeExplicit() bool {
return parseExplicit(it.ITunesExplicit)
}
func parseExplicit(v string) bool {
switch strings.ToLower(strings.TrimSpace(v)) {
case "true", "yes", "explicit", "1":
return true
}
return false
}
// HostsAndGuests splits podcast:person entries by role (plan §5).
func (it *PodcastItemXML) HostsAndGuests() (hosts []string, guests []string) {
for _, p := range it.Persons {
name := p.Href
if name == "" {
continue
}
switch strings.ToLower(p.Role) {
case "guest":
guests = append(guests, name)
default: // host and unlabeled roles surface as hosts
hosts = append(hosts, name)
}
}
return hosts, guests
}
// parseITunesDuration accepts the three spellings the spec allows —
// plain seconds, HH:MM:SS, MM:SS — plus a best-effort prose fallback
// ("58 minutes"). Returns 0 for anything unparseable (the column stays
// NULL; ffprobe reconciles after download anyway).
func parseITunesDuration(v string) int {
v = strings.TrimSpace(v)
if v == "" {
return 0
}
if secs, err := strconv.Atoi(v); err == nil {
return secs
}
parts := strings.Split(v, ":")
if len(parts) <= 3 {
total := 0
ok := true
for _, p := range parts {
n, err := strconv.Atoi(strings.TrimSpace(p))
if err != nil {
ok = false
break
}
total = total*60 + n
}
if ok {
return total
}
}
// Prose best-effort: "<N> minutes/hours/seconds".
lower := strings.ToLower(v)
fields := strings.Fields(lower)
for i, f := range fields {
if n, err := strconv.Atoi(f); err == nil && i+1 < len(fields) {
switch {
case strings.HasPrefix(fields[i+1], "hour"):
return n * 3600
case strings.HasPrefix(fields[i+1], "minute"):
return n * 60
case strings.HasPrefix(fields[i+1], "second"):
return n
}
}
}
return 0
}
// parseFeedTime parses the RFC-822/1123 variants feeds actually emit.
func parseFeedTime(v string) (t time.Time, ok bool) {
v = strings.TrimSpace(v)
if v == "" {
return time.Time{}, false
}
layouts := []string{
time.RFC1123Z,
time.RFC1123,
time.RFC822Z,
time.RFC822,
"Mon, 2 Jan 2006 15:04:05 -0700",
"Mon, 2 Jan 2006 15:04:05 MST",
"2 Jan 2006 15:04:05 -0700",
"2 Jan 2006 15:04:05 MST",
time.ANSIC,
time.UnixDate,
}
for _, layout := range layouts {
if t, err := time.Parse(layout, v); err == nil {
return t, true
}
}
return time.Time{}, false
}
+762
View File
@@ -0,0 +1,762 @@
// Podcast fetcher (plan podcast-integration.md §7): fetch → parse →
// upsert-without-delete. The fetcher is the SOLE writer of feed/episode
// metadata (D9) — user-facing podcast writes are limited to
// podcast_subscriptions rows and the reader_settings order override.
//
// RLS identity (invariant 12): fetches run detached from any request,
// so every governed statement runs under the FIRST-ADMIN identity
// (scanner fallback precedent). With the D20 admin escape that identity
// sees and updates every feed/episode row. NEVER a bypass flag, never
// the bootstrap superuser.
//
// Upsert semantics: episodes are created/updated METADATA-ONLY, keyed
// on (source_feed_id, episode_guid); rows for items that vanish from
// the XML are NEVER deleted (history accumulates — D14 search covers
// gone episodes). Download state is owned by the download worker; the
// upsert cannot clobber it (guarded in the query itself).
package services
import (
"context"
"errors"
"fmt"
"log"
"net/url"
"os"
"path/filepath"
"regexp"
"strconv"
"strings"
"sync"
"time"
"bookhoard/internal/database"
"bookhoard/internal/rls"
"github.com/jackc/pgx/v5/pgtype"
)
// PodcastError is a status-carrying service error: handlers map it to
// the HTTP response directly (services stay echo-free).
type PodcastError struct {
Status int
Message string
}
func (e *PodcastError) Error() string { return e.Message }
func errNotFound(msg string) error { return &PodcastError{Status: 404, Message: msg} }
func errForbidden(msg string) error { return &PodcastError{Status: 403, Message: msg} }
func errBadRequest(msg string) error { return &PodcastError{Status: 400, Message: msg} }
// PodcastService carries the fetcher/worker dependencies. One instance
// per server (handlers + poller share it so the per-feed locks dedupe).
type PodcastService struct {
db *database.Queries
adminMu sync.Mutex
adminID pgtype.UUID
adminOnce bool
// Per-feed mutexes: a poll and a download for the same feed must
// not race (e.g. both writing the episode row). Keyed by feed UUID.
feedLocksMu sync.Mutex
feedLocks map[string]*sync.Mutex
}
func NewPodcastService(db *database.Queries) *PodcastService {
return &PodcastService{
db: db,
feedLocks: make(map[string]*sync.Mutex),
}
}
// feedLock returns the per-feed mutex (creates on first use).
func (s *PodcastService) feedLock(feedID string) *sync.Mutex {
s.feedLocksMu.Lock()
defer s.feedLocksMu.Unlock()
if m, ok := s.feedLocks[feedID]; ok {
return m
}
m := &sync.Mutex{}
s.feedLocks[feedID] = m
return m
}
// adminIdentity resolves the acting identity for detached podcast work:
// the first admin, resolved once per service lifetime. Errors surface
// to the caller (no admin → no fetch — same contract as the scanner).
func (s *PodcastService) adminIdentity() (pgtype.UUID, error) {
s.adminMu.Lock()
defer s.adminMu.Unlock()
if !s.adminOnce && !s.adminID.Valid {
id, err := s.db.GetFirstAdmin(context.Background())
if err != nil {
return pgtype.UUID{}, fmt.Errorf("podcasts: no admin identity available for detached work: %w", err)
}
s.adminID = id
s.adminOnce = true
}
if !s.adminID.Valid {
return pgtype.UUID{}, fmt.Errorf("podcasts: no admin identity available for detached work")
}
return s.adminID, nil
}
// adminCtx wraps ctx with the first-admin identity (invariant 12).
func (s *PodcastService) adminCtx(ctx context.Context) (context.Context, error) {
id, err := s.adminIdentity()
if err != nil {
return nil, err
}
return rls.WithUser(ctx, id), nil
}
// AddFeedResult carries what the add-feed flow did (the handler turns
// it into the response / serial feed hint).
type AddFeedResult struct {
Feed *database.PodcastFeeds
// Created is true when this call created the feed row; false when
// an existing row was reused (URL dedupe or guid migration match).
Created bool
// Subscribed is true when the requesting user's subscription was
// created by this call (false = already subscribed).
Subscribed bool
// EpisodeCount is the number of episode rows the fetch upserted.
EpisodeCount int
// FetchError, when set, means the feed row exists but the fetch
// failed — the subscription stands, the error surfaces to the user.
FetchError string
}
// AddFeed implements POST /api/podcasts/feeds (plan §8): fetch+parse the
// user-supplied URL, create-or-reuse the feed row, upsert metadata-only
// episodes, subscribe the creator. Authorization on explicit arguments
// (D20 scoping model): the target library must be visible to the actor
// and, for non-admins, allow member feeds.
func (s *PodcastService) AddFeed(ctx context.Context, userID pgtype.UUID, isAdmin bool, libraryID pgtype.UUID, rawFeedURL string) (*AddFeedResult, error) {
// ---- explicit-argument authorization (the ONLY Go-level gate) ----
lib, err := s.db.GetLibrary(ctx, libraryID)
if err != nil {
// Invisible-or-nonexistent: 404 semantics (existence leak).
return nil, errNotFound("library not found")
}
if !isAdmin {
if lib.TypeName != LibraryTypePodcasts {
return nil, errNotFound("library not found")
}
if !lib.AllowMemberFeeds {
return nil, errForbidden("this library does not allow members to add feeds")
}
}
feedURL := strings.TrimSpace(rawFeedURL)
if feedURL == "" {
return nil, errBadRequest("feed URL is required")
}
// The fetch/upsert runs under the first-admin identity (detached-
// equivalent); the request context never carries special powers into
// feed writes.
actx, err := s.adminCtx(ctx)
if err != nil {
return nil, err
}
res := &AddFeedResult{}
// ---- fetch + parse (SSRF-guarded; plan §7) ----
fetched, err := guardFetch(ctx, feedURL, "", "", maxFeedSizeBytes)
if err != nil {
return nil, fmt.Errorf("could not fetch feed: %w", classifyFetchError(err))
}
parsed, err := ParsePodcastFeed(fetched.Body)
if err != nil {
return nil, fmt.Errorf("could not parse feed: %w", err)
}
// Follow itunes:new-feed-url once at add time so the row is born
// pointing at the publisher's current home (migration semantics).
finalURL := fetched.FinalURL
if neu := strings.TrimSpace(parsed.Channel.ITunesNewFeedURL); neu != "" && neu != finalURL {
if retried, err := guardFetch(ctx, neu, "", "", maxFeedSizeBytes); err == nil {
if reparsed, err := ParsePodcastFeed(retried.Body); err == nil {
parsed = reparsed
finalURL = retried.FinalURL
}
}
}
title := parsed.Channel.Title
if title == "" {
title = "Untitled feed"
}
// ---- create-or-reuse (UNIQUE(library_id, feed_url) + guid dedupe) ----
feed, err := s.db.CreatePodcastFeed(actx, database.CreatePodcastFeedParams{
LibraryID: libraryID,
FeedUrl: finalURL,
Title: truncate(title, 500),
CreatedByAdminID: pgtype.UUID{}, // fetcher owns feed metadata; creator attribution is not user metadata
})
if err != nil {
return nil, fmt.Errorf("could not create feed: %w", err)
}
res.Feed = &feed
res.Created = true
// Guid migration dedupe: a row with the same canonical podcast_guid
// under a different URL reuses THAT row (do not fork the feed).
if guid := strings.TrimSpace(parsed.Channel.PodcastGUID); guid != "" {
if feed.PodcastGuid.String == "" || feed.PodcastGuid.String != guid {
if existing, err := s.db.GetPodcastFeedByGuid(actx, database.GetPodcastFeedByGuidParams{
LibraryID: feed.LibraryID,
PodcastGuid: pgtypeText(guid),
}); err == nil && existing.ID != feed.ID {
// Attach the subscriber to the canonical row; drop the
// freshly created stub (nothing subscribed to it yet).
_, _ = s.db.DeletePodcastFeed(actx, feed.ID)
feed = existing
res.Feed = &existing
// Re-point the canonical row at the fetched URL if it moved.
if feedUrlForComparison(existing.FeedUrl) != feedUrlForComparison(finalURL) {
_ = s.db.UpdatePodcastFeedRedirect(actx, database.UpdatePodcastFeedRedirectParams{
ID: existing.ID,
FeedUrl: finalURL,
})
}
}
}
}
if _, err := s.db.SubscribePodcastFeed(actx, database.SubscribePodcastFeedParams{
UserID: userID,
FeedID: feed.ID,
}); err == nil {
res.Subscribed = true
}
// ---- metadata + episodes ----
if err := s.applyFetch(actx, &feed, parsed, fetched); err != nil {
// Feed exists + subscribed, but the metadata pass failed: not
// fatal for the request; the poller reconciles later.
res.FetchError = err.Error()
return res, nil
}
updated, err := s.db.GetPodcastFeed(actx, feed.ID)
if err == nil {
res.Feed = &updated
}
return res, nil
}
// RefreshFeed implements POST /api/podcasts/feeds/:id/refresh: a manual
// poll. The per-feed lock serializes against poller ticks and downloads.
func (s *PodcastService) RefreshFeed(ctx context.Context, feedID pgtype.UUID) error {
actx, err := s.adminCtx(ctx)
if err != nil {
return err
}
feed, err := s.db.GetPodcastFeed(actx, feedID)
if err != nil {
return errNotFound("feed not found")
}
return s.PollFeed(actx, &feed)
}
// PollFeed fetches + reconciles ONE feed (caller supplies the acting
// identity in ctx). Safe to call concurrently per feed — the per-feed
// lock serializes poll/download.
func (s *PodcastService) PollFeed(ctx context.Context, feed *database.PodcastFeeds) error {
lock := s.feedLock(feed.ID.String())
lock.Lock()
defer lock.Unlock()
// Complete feeds back off: they are fetched only on explicit
// refresh (the poller skips them entirely — this path is manual).
fetched, err := guardFetch(ctx, feed.FeedUrl, feed.Etag.String, feed.LastModified.String, maxFeedSizeBytes)
if err != nil {
if feed.LastFetchedAt.Valid {
// A previously-good feed with a transient failure: keep the
// old state; the poller retries next tick.
return fmt.Errorf("fetch %s: %w", feed.Title, classifyFetchError(err))
}
return fmt.Errorf("fetch %s: %w", feed.Title, classifyFetchError(err))
}
if fetched.NotModified {
_, _ = s.db.UpdatePodcastFeedFetch(ctx, database.UpdatePodcastFeedFetchParams{
ID: feed.ID,
Etag: pgtypeText(feed.Etag.String),
LastModified: pgtypeText(feed.LastModified.String),
LastBuildDate: feed.LastBuildDate,
})
return nil
}
parsed, err := ParsePodcastFeed(fetched.Body)
if err != nil {
return fmt.Errorf("parse %s: %w", feed.Title, err)
}
// itunes:new-feed-url: the publisher MOVED the feed. Rewrite the
// stored URL (row, subscriptions and episodes stay put) and refetch
// the new location once so this poll applies its content.
if neu := strings.TrimSpace(parsed.Channel.ITunesNewFeedURL); neu != "" && neu != feed.FeedUrl {
_ = s.db.UpdatePodcastFeedRedirect(ctx, database.UpdatePodcastFeedRedirectParams{ID: feed.ID, FeedUrl: neu})
if retried, err := guardFetch(ctx, neu, "", "", maxFeedSizeBytes); err == nil {
if reparsed, err := ParsePodcastFeed(retried.Body); err == nil {
parsed = reparsed
fetched = retried
}
}
} else if fetched.FinalURL != "" && feedUrlForComparison(fetched.FinalURL) != feedUrlForComparison(feed.FeedUrl) {
// A permanent HTTP redirect chain moved the feed; persist the
// final URL so conditional validators stop bouncing.
_ = s.db.UpdatePodcastFeedRedirect(ctx, database.UpdatePodcastFeedRedirectParams{ID: feed.ID, FeedUrl: fetched.FinalURL})
}
if err := s.applyFetch(ctx, feed, parsed, fetched); err != nil {
return err
}
return nil
}
// applyFetch writes channel metadata, upserts episodes metadata-only,
// touches validators, and downloads artwork. Write-after-success rule
// (invariant 7): failures here never clear existing rows.
func (s *PodcastService) applyFetch(ctx context.Context, feed *database.PodcastFeeds, parsed *PodcastFeedXML, fetched *fetchResult) error {
// --- channel metadata (fetcher-owned) ---
title := parsed.Channel.Title
if title == "" {
title = feed.Title // never blank out a title on a broken doc
}
desc := stripHTML(parsed.Channel.Description)
host := parsed.Channel.ITunesAuthor
publisher := parsed.Channel.ITunesOwner.Name
tagSet := map[string]bool{}
if cat := strings.TrimSpace(parsed.Channel.ITunesCategory.Text); cat != "" {
tagSet[cat] = true
}
if sub := strings.TrimSpace(parsed.Channel.ITunesCategory.Subcategory.Text); sub != "" {
tagSet[sub] = true
}
tags := sortedStringSet(tagSet)
feedType := strings.TrimSpace(parsed.Channel.ITunesType)
if feedType != "serial" && feedType != "episodic" {
feedType = "" // unknown → column stays NULL
}
medium := strings.TrimSpace(parsed.Channel.PodcastMedium)
if _, err := s.db.UpdatePodcastFeedMetadata(ctx, database.UpdatePodcastFeedMetadataParams{
ID: feed.ID,
Title: truncate(title, 500),
Description: pgtypeText(desc),
Host: pgtypeText(truncate(host, 500)),
Publisher: pgtypeText(truncate(publisher, 500)),
OwnerName: pgtypeText(truncate(parsed.Channel.ITunesOwner.Name, 500)),
OwnerEmail: pgtypeText(truncate(parsed.Channel.ITunesOwner.Email, 500)),
Language: pgtypeText(truncate(parsed.Channel.Language, 50)),
Tags: tags,
Explicit: parseExplicit(parsed.Channel.ITunesExplicit),
FeedType: pgtypeText(feedType),
Medium: pgtypeText(truncate(medium, 20)),
IsComplete: parseExplicit(parsed.Channel.ITunesComplete),
PodcastGuid: pgtypeText(strings.TrimSpace(parsed.Channel.PodcastGUID)),
}); err != nil {
return fmt.Errorf("feed metadata update: %w", err)
}
// itunes:complete flips the poller's back-off flag (complete feeds
// poll rarely); explicit refresh still works.
// --- episodes (metadata-only upsert, positions from THIS parse) ---
root, err := s.downloadRootAbs(ctx, feed)
if err != nil {
return fmt.Errorf("download root: %w", err)
}
feedSlug := feedSlug(title, feed.ID)
count := 0
for i := range parsed.Channel.Items {
if err := s.upsertEpisode(ctx, feed, root, feedSlug, &parsed.Channel.Items[i], i+1); err != nil {
// One bad item must not sink the batch.
log.Printf("[PODCAST] episode upsert failed for feed %s item %d: %v", feed.ID, i+1, err)
continue
}
count++
}
// --- validators + fetch timestamp ---
var lastBuildDate pgtype.Timestamptz
if t, ok := parseFeedTime(parsed.Channel.LastBuildDate); ok {
lastBuildDate = pgtypeTimestamptz(t)
}
if _, err := s.db.UpdatePodcastFeedFetch(ctx, database.UpdatePodcastFeedFetchParams{
ID: feed.ID,
Etag: pgtypeText(fetched.ETag),
LastModified: pgtypeText(fetched.LastModified),
LastBuildDate: lastBuildDate,
}); err != nil {
return fmt.Errorf("fetch touch: %w", err)
}
// --- artwork (best-effort; feed renders fine without it) ---
s.refreshArtwork(ctx, feed, root, parsed)
return nil
}
// upsertEpisode maps one XML item onto a media_items row (plan §5 field
// inventory). Positions are 1-based, renumbered per fetch; file_path is
// the PLANNED final layout — the upsert only applies it to rows that
// are not yet downloaded (query-side guard).
func (s *PodcastService) upsertEpisode(ctx context.Context, feed *database.PodcastFeeds, downloadRoot, feedSlug string, item *PodcastItemXML, position int) error {
title := item.EpisodeTitle()
if title == "" {
title = fmt.Sprintf("Episode %d", position)
}
guid := strings.TrimSpace(item.GUID)
if guid == "" {
// GUID-less feeds exist; fall back to a stable identity from the
// enclosure URL (the next-best immutable per-episode key).
if item.Enclosure.URL != "" {
guid = "enc:" + item.Enclosure.URL
} else {
return fmt.Errorf("item %q has neither guid nor enclosure; skipping", title)
}
}
ext := enclosureExtension(item.Enclosure.URL, item.Enclosure.Type)
relPath := episodeRelPath(feedSlug, position, title, ext)
var published pgtype.Date
if t, ok := parseFeedTime(item.PubDate); ok {
published = pgtypeDate(t)
}
authors := []string{}
if a := item.EpisodeAuthor(); a != "" {
authors = append(authors, truncate(a, 255))
}
hosts, guests := item.HostsAndGuests()
_, err := s.db.UpsertPodcastEpisode(ctx, database.UpsertPodcastEpisodeParams{
LibraryID: feed.LibraryID,
Title: truncate(title, 255),
Subtitle: pgtypeText(truncate(item.Subtitle, 500)),
Description: pgtypeText(stripHTML(item.EpisodeDescription())),
Authors: authors,
AuthorsSearch: lowerSearchSlice(authors),
FilePath: relPath,
FileSize: pgtypeInt8(item.Enclosure.Length),
MimeType: pgtypeText(truncate(item.Enclosure.Type, 100)),
CoverImagePath: pgtypeText(artworkRelPath(feedSlug, artworkExtFromURL(item.ITunesImage.Href))),
Explicit: item.EpisodeExplicit(),
DatePublished: published,
Language: pgtypeText(truncate(feed.Language.String, 10)),
Tags: feed.Tags,
Genre: pgtypeText(strings.Join(feed.Tags, ", ")),
SourceFeedID: feed.ID,
EpisodeGuid: pgtypeText(truncate(guid, 500)),
FeedPosition: pgtypeInt4(int32(position)),
SeasonNumber: pgtypeInt4FromStr(item.ITunesSeason),
EpisodeNumber: pgtypeInt4FromStr(item.ITunesEpisode),
EnclosureUrl: pgtypeText(item.Enclosure.URL),
DeclaredDurationSeconds: pgtypeInt4(int32(parseITunesDuration(item.ITunesDuration))),
TranscriptUrl: pgtypeText(item.Transcript.URL),
ChaptersUrl: pgtypeText(item.Chapters.URL),
Hosts: truncateSlice(hosts, 255),
Guests: truncateSlice(guests, 255),
})
_ = downloadRoot // layout is relative; abs root resolves at download time
return err
}
// refreshArtwork downloads the channel image once per changed URL.
// Artwork is feed-level: episodes point at the same relative path.
func (s *PodcastService) refreshArtwork(ctx context.Context, feed *database.PodcastFeeds, downloadRoot string, parsed *PodcastFeedXML) {
imgURL := strings.TrimSpace(parsed.Channel.ITunesImage.Href)
if imgURL == "" {
return
}
target := filepath.Join(downloadRoot, artworkRelPath(feedSlug(feed.Title, feed.ID), artworkExtFromURL(imgURL)))
if feed.ArtworkPath.Valid {
if existing := filepath.Join(downloadRoot, feed.ArtworkPath.String); existing == target {
if _, err := os.Stat(existing); err == nil {
return // unchanged
}
}
}
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return
}
res, err := guardFetch(ctx, imgURL, "", "", maxArtworkSizeBytes)
if err != nil || len(res.Body) == 0 {
log.Printf("[PODCAST] artwork fetch failed for feed %s: %v", feed.ID, err)
return
}
if err := os.WriteFile(target, res.Body, 0o644); err != nil {
log.Printf("[PODCAST] artwork write failed for feed %s: %v", feed.ID, err)
return
}
rel, err := filepath.Rel(downloadRoot, target)
if err != nil {
return
}
_ = s.db.UpdatePodcastFeedArtwork(ctx, database.UpdatePodcastFeedArtworkParams{
ID: feed.ID,
ArtworkPath: pgtypeText(rel),
})
}
// DeleteFeedFiles removes a feed's episode files + artwork from the
// download root (admin feed-removal path; ?files=false skips it).
// Best-effort per file — the row cascade proceeds regardless.
func (s *PodcastService) DeleteFeedFiles(ctx context.Context, feed *database.PodcastFeeds, episodes []database.MediaItems) {
root, err := s.downloadRootAbs(ctx, feed)
if err != nil {
return
}
for _, ep := range episodes {
if ep.FilePath != "" {
_ = os.Remove(filepath.Join(root, ep.FilePath))
// Embedded-cover sibling (extractEmbeddedAudioCover convention).
base := filepath.Join(root, strings.TrimSuffix(ep.FilePath, filepath.Ext(ep.FilePath)))
_ = os.Remove(base + ".cover.jpg")
}
}
if feed.ArtworkPath.Valid {
_ = os.Remove(filepath.Join(root, feed.ArtworkPath.String))
}
// The feed directory itself: remove when empty (other feeds never
// share a slug — the feed-id suffix guarantees that).
_ = os.Remove(filepath.Join(root, feedSlug(feed.Title, feed.ID)))
}
// downloadRootAbs resolves the podcasts library's single download root
// (purpose='podcast') as an absolute filesystem path.
func (s *PodcastService) downloadRootAbs(ctx context.Context, feed *database.PodcastFeeds) (string, error) {
folders, err := s.db.GetPodcastLibraryFolder(ctx, feed.LibraryID)
if err != nil || len(folders) == 0 {
return "", fmt.Errorf("library %s has no download root registered", feed.LibraryID)
}
return folders[0].FolderPath, nil
}
// episodeRelPath is the pinned episode layout (plan §6):
// <feed-slug>/<feed_position:04d>-<slug>.<ext> — relative to the
// download root. getFullFilePath/ServeFile/Range streaming resolve it
// unchanged.
func episodeRelPath(feedSlug string, position int, title, ext string) string {
return fmt.Sprintf("%s/%04d-%s%s", feedSlug, position, slugify(title, 80), ext)
}
// artworkRelPath is the feed-level artwork location (feed-level, so
// every episode row can point at the same file).
func artworkRelPath(feedSlug, ext string) string {
return fmt.Sprintf("%s/artwork%s", feedSlug, ext)
}
// feedSlug derives the stable per-feed directory name: slugged title +
// short feed-id suffix so two identically-titled feeds never collide.
func feedSlug(title string, feedID pgtype.UUID) string {
base := slugify(title, 60)
suffix := "feed"
if feedID.Valid {
s := feedID.String()
if len(s) >= 8 {
suffix = s[:8]
}
}
if base == "" {
base = "feed"
}
return base + "-" + suffix
}
var slugStrip = regexp.MustCompile(`[^a-z0-9]+`)
// slugify lowercases, strips non-alphanumerics to dashes, and clamps
// the length (file-name-safe; unicode letters drop — file layout is
// an internal key, never display text).
func slugify(s string, maxLen int) string {
s = strings.ToLower(strings.TrimSpace(s))
s = slugStrip.ReplaceAllString(s, "-")
s = strings.Trim(s, "-")
if len(s) > maxLen {
s = strings.TrimRight(s[:maxLen], "-")
}
return s
}
// enclosureExtension picks a file extension from the URL, then the
// MIME type, then a safe default.
func enclosureExtension(rawURL, mimeType string) string {
if u, err := urlParse(rawURL); err == nil {
if ext := strings.ToLower(filepath.Ext(u.Path)); ext == ".mp3" || ext == ".m4a" || ext == ".m4b" ||
ext == ".flac" || ext == ".ogg" || ext == ".oga" || ext == ".opus" || ext == ".wav" || ext == ".aac" {
return ext
}
}
switch {
case strings.Contains(mimeType, "mp4") || strings.Contains(mimeType, "m4a"):
return ".m4a"
case strings.Contains(mimeType, "mpeg") || strings.Contains(mimeType, "mp3"):
return ".mp3"
case strings.Contains(mimeType, "flac"):
return ".flac"
case strings.Contains(mimeType, "ogg"):
return ".ogg"
case strings.Contains(mimeType, "wav"):
return ".wav"
}
return ".mp3"
}
// artworkExtFromURL keeps the source extension for artwork (jpg/png/webp).
func artworkExtFromURL(rawURL string) string {
if u, err := urlParse(rawURL); err == nil {
switch strings.ToLower(filepath.Ext(u.Path)) {
case ".png":
return ".png"
case ".webp":
return ".webp"
case ".gif":
return ".gif"
}
}
return ".jpg"
}
// stripHTML removes tags and collapses whitespace — feed descriptions
// are HTML; storage wants text (the TUI-ish surfaces render it plain).
func stripHTML(s string) string {
if s == "" {
return ""
}
var b strings.Builder
inTag := false
for _, r := range s {
switch {
case r == '<':
inTag = true
case r == '>':
inTag = false
case !inTag:
b.WriteRune(r)
}
}
out := b.String()
// Collapse runs of whitespace.
out = strings.Join(strings.Fields(out), " ")
return strings.TrimSpace(out)
}
// truncate clamps a string to max runes (VARCHAR safety at the DB edge).
func truncate(s string, max int) string {
s = strings.TrimSpace(s)
runes := []rune(s)
if len(runes) <= max {
return s
}
return string(runes[:max])
}
// truncateSlice clamps each element and drops empties.
func truncateSlice(in []string, max int) []string {
out := make([]string, 0, len(in))
for _, s := range in {
if t := truncate(s, max); t != "" {
out = append(out, t)
}
}
return out
}
// lowerSearchSlice mirrors authors into authors_search (scanner convention).
func lowerSearchSlice(in []string) []string {
out := make([]string, 0, len(in))
for _, s := range in {
out = append(out, strings.ToLower(s))
}
return out
}
// sortedStringSet → deterministic TEXT[] writes.
func sortedStringSet(set map[string]bool) []string {
out := make([]string, 0, len(set))
for k := range set {
if strings.TrimSpace(k) == "" {
continue
}
out = append(out, strings.TrimSpace(k))
}
// tiny insertion sort — tag sets are 1–5 entries
for i := 1; i < len(out); i++ {
for j := i; j > 0 && out[j] < out[j-1]; j-- {
out[j], out[j-1] = out[j-1], out[j]
}
}
return out
}
// classifyFetchError maps guard/HTTP errors to user-facing text (the
// add-feed dialog and the download_error column both render it).
func classifyFetchError(err error) error {
if errors.Is(err, ErrBlockedPrivateAddress) {
return fmt.Errorf("blocked: private/loopback addresses are not fetchable")
}
return err
}
// urlParse local helper so podcast files don't each import net/url.
func urlParse(raw string) (*url.URL, error) {
return url.Parse(raw)
}
// ---- pgtype helpers shared by the podcast files ----
func pgtypeText(s string) pgtype.Text {
if strings.TrimSpace(s) == "" {
return pgtype.Text{}
}
return pgtype.Text{String: s, Valid: true}
}
func pgtypeInt4(v int32) pgtype.Int4 {
if v == 0 {
return pgtype.Int4{}
}
return pgtype.Int4{Int32: v, Valid: true}
}
func pgtypeInt4FromStr(s string) pgtype.Int4 {
n, err := strconv.Atoi(strings.TrimSpace(s))
if err != nil || n <= 0 {
return pgtype.Int4{}
}
return pgtype.Int4{Int32: int32(n), Valid: true}
}
func pgtypeInt8(v int64) pgtype.Int8 {
if v <= 0 {
return pgtype.Int8{}
}
return pgtype.Int8{Int64: v, Valid: true}
}
func pgtypeTimestamptz(t time.Time) pgtype.Timestamptz {
return pgtype.Timestamptz{Time: t, Valid: true}
}
func pgtypeDate(t time.Time) pgtype.Date {
return pgtype.Date{Time: t.UTC().Truncate(24 * time.Hour), Valid: true}
}
// feedUrlForComparison normalizes trailing slashes for the redirect check.
func feedUrlForComparison(u string) string {
return strings.TrimRight(strings.TrimSpace(u), "/")
}