Add real podcast subsystem: RSS feed refresh, bounded background downloads, restart recovery

2026-08-23 00:24:43 -05:00
parent 8d93bea384
commit fce44a2ffd
6 changed files with 809 additions and 0 deletions
@@ -0,0 +1,299 @@
+// Package podcasts implements a real podcast subsystem: RSS/Atom feed
+// refresh, episode metadata, enclosure downloads with bounded background
+// workers, and download-state recovery across restarts. Downloaded
+// episodes are registered as ordinary media files (via internal/library)
+// so they stream through exactly the same path/permission-checked code
+// as scanned music, rather than a separate ad hoc file server.
+package podcasts
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"git.noroutes.net/NoRoutes/vibetunes/internal/database"
+	"git.noroutes.net/NoRoutes/vibetunes/internal/idgen"
+	"git.noroutes.net/NoRoutes/vibetunes/internal/metadata"
+)
+
+const maxConcurrentDownloads = 2
+
+type Manager struct {
+	DB          *sql.DB
+	DataDir     string // episodes are stored under <DataDir>/podcasts/<channelID>/
+	FFprobePath string
+	HTTPClient  *http.Client
+	Log         *slog.Logger
+
+	sem chan struct{}
+}
+
+func NewManager(db *sql.DB, dataDir, ffprobePath string, log *slog.Logger) *Manager {
+	return &Manager{
+		DB:          db,
+		DataDir:     dataDir,
+		FFprobePath: ffprobePath,
+		HTTPClient:  &http.Client{Timeout: 60 * time.Second},
+		Log:         log,
+		sem:         make(chan struct{}, maxConcurrentDownloads),
+	}
+}
+
+// RecoverInterruptedDownloads resets episodes stuck in "downloading"
+// status back to "new" on startup — the process was killed mid-download,
+// so there is no partial state worth preserving, but the episode should
+// not be stuck forever as a phantom in-progress download.
+func (m *Manager) RecoverInterruptedDownloads() error {
+	stuck, err := database.ListPodcastEpisodesByStatus(m.DB, "downloading")
+	if err != nil {
+		return err
+	}
+	for _, ep := range stuck {
+		if err := database.SetPodcastEpisodeStatus(m.DB, ep.ID, "new"); err != nil {
+			return err
+		}
+	}
+	if len(stuck) > 0 {
+		m.Log.Info("recovered interrupted podcast downloads", "count", len(stuck))
+	}
+	return nil
+}
+
+// RefreshChannel fetches and parses a channel's feed, updating its
+// metadata and upserting every episode found (keyed by GUID, so episode
+// IDs and any completed downloads survive repeated refreshes).
+func (m *Manager) RefreshChannel(ctx context.Context, channelID string) error {
+	channel, err := database.GetPodcastChannel(m.DB, channelID)
+	if err != nil {
+		return err
+	}
+	_ = database.SetPodcastChannelStatus(m.DB, channelID, "downloading", "")
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, channel.URL, nil)
+	if err != nil {
+		_ = database.SetPodcastChannelStatus(m.DB, channelID, "error", err.Error())
+		return err
+	}
+	resp, err := m.HTTPClient.Do(req)
+	if err != nil {
+		_ = database.SetPodcastChannelStatus(m.DB, channelID, "error", err.Error())
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		errMsg := fmt.Sprintf("feed returned HTTP %d", resp.StatusCode)
+		_ = database.SetPodcastChannelStatus(m.DB, channelID, "error", errMsg)
+		return fmt.Errorf("%s", errMsg)
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) // 16MB cap: a feed XML document is never legitimately bigger
+	if err != nil {
+		_ = database.SetPodcastChannelStatus(m.DB, channelID, "error", err.Error())
+		return err
+	}
+
+	feed, err := parseRSS(body)
+	if err != nil {
+		_ = database.SetPodcastChannelStatus(m.DB, channelID, "error", "invalid feed: "+err.Error())
+		return err
+	}
+
+	if err := database.UpdatePodcastChannelMetadata(m.DB, channelID, feed.Channel.Title, feed.Channel.Description); err != nil {
+		return err
+	}
+	for _, item := range feed.Channel.Items {
+		if item.Enclosure.URL == "" {
+			continue // not a downloadable episode (e.g. a text-only post in a mixed feed)
+		}
+		guid := item.episodeGUID()
+		if guid == "" {
+			continue
+		}
+		if _, err := database.UpsertPodcastEpisode(m.DB, channelID, guid, item.Title, item.Description, item.PubDate, item.Enclosure.URL, parseDuration(item.Duration)); err != nil {
+			return err
+		}
+	}
+
+	return database.SetPodcastChannelStatus(m.DB, channelID, "completed", "")
+}
+
+// parseDuration handles the iTunes duration formats: plain seconds,
+// "MM:SS", or "HH:MM:SS".
+func parseDuration(s string) float64 {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return 0
+	}
+	if n, err := strconv.ParseFloat(s, 64); err == nil {
+		return n
+	}
+	parts := strings.Split(s, ":")
+	var total float64
+	for _, p := range parts {
+		n, err := strconv.Atoi(p)
+		if err != nil {
+			return 0
+		}
+		total = total*60 + float64(n)
+	}
+	return total
+}
+
+// EnsurePodcastFolder returns the music folder downloaded episodes are
+// registered under, creating one (kind="podcasts") on first use.
+func EnsurePodcastFolder(db *sql.DB, dataDir string) (*database.MusicFolder, error) {
+	folders, err := database.ListMusicFolders(db)
+	if err != nil {
+		return nil, err
+	}
+	for _, f := range folders {
+		if f.Kind == "podcasts" {
+			return f, nil
+		}
+	}
+	path := filepath.Join(dataDir, "podcasts")
+	if err := os.MkdirAll(path, 0o755); err != nil {
+		return nil, err
+	}
+	return database.CreateMusicFolder(db, "Podcasts", path, "podcasts")
+}
+
+// DownloadEpisode downloads an episode's enclosure and registers it as a
+// streamable media file. Bounded by Manager's semaphore so a burst of
+// download requests can't open unlimited concurrent connections.
+func (m *Manager) DownloadEpisode(ctx context.Context, episodeID string) error {
+	select {
+	case m.sem <- struct{}{}:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	defer func() { <-m.sem }()
+
+	ep, err := database.GetPodcastEpisode(m.DB, episodeID)
+	if err != nil {
+		return err
+	}
+	channel, err := database.GetPodcastChannel(m.DB, ep.ChannelID)
+	if err != nil {
+		return err
+	}
+	folder, err := EnsurePodcastFolder(m.DB, m.DataDir)
+	if err != nil {
+		return err
+	}
+
+	if err := database.SetPodcastEpisodeStatus(m.DB, episodeID, "downloading"); err != nil {
+		return err
+	}
+
+	ext := extensionFromURL(ep.EnclosureURL)
+	relPath := filepath.Join(sanitizeName(channel.Title, channel.ID), sanitizeName(ep.Title, ep.ID)+ext)
+	absPath := filepath.Join(folder.Path, relPath)
+	if err := os.MkdirAll(filepath.Dir(absPath), 0o755); err != nil {
+		_ = database.SetPodcastEpisodeStatus(m.DB, episodeID, "error")
+		return err
+	}
+
+	if err := m.downloadToFile(ctx, ep.EnclosureURL, absPath); err != nil {
+		_ = database.SetPodcastEpisodeStatus(m.DB, episodeID, "error")
+		return err
+	}
+
+	info, err := os.Stat(absPath)
+	if err != nil {
+		_ = database.SetPodcastEpisodeStatus(m.DB, episodeID, "error")
+		return err
+	}
+
+	prober := &metadata.Prober{FFprobePath: m.FFprobePath}
+	tags, perr := prober.Probe(ctx, absPath)
+	duration, bitrate, sampleRate, channels, format := ep.DurationSeconds, 0, 0, 0, strings.TrimPrefix(ext, ".")
+	if perr == nil {
+		duration, bitrate, sampleRate, channels, format = tags.DurationSeconds, tags.BitrateKbps, tags.SampleRateHz, tags.Channels, tags.Format
+	}
+
+	mf, err := database.CreateMediaFile(m.DB, folder.ID, filepath.ToSlash(relPath), info.Size(), info.ModTime().Unix(),
+		idgen.New(""), format, "song", duration, bitrate, sampleRate, channels)
+	if err != nil {
+		_ = database.SetPodcastEpisodeStatus(m.DB, episodeID, "error")
+		return err
+	}
+	if _, err := database.UpsertSong(m.DB, database.UpsertSongInput{MediaFileID: mf.ID, Title: ep.Title}); err != nil {
+		_ = database.SetPodcastEpisodeStatus(m.DB, episodeID, "error")
+		return err
+	}
+
+	return database.SetPodcastEpisodeMediaFile(m.DB, episodeID, mf.ID)
+}
+
+func (m *Manager) downloadToFile(ctx context.Context, sourceURL, destPath string) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, sourceURL, nil)
+	if err != nil {
+		return err
+	}
+	resp, err := m.HTTPClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("enclosure download returned HTTP %d", resp.StatusCode)
+	}
+
+	tmp := destPath + ".part"
+	out, err := os.Create(tmp)
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(out, resp.Body); err != nil {
+		out.Close()
+		os.Remove(tmp)
+		return err
+	}
+	if err := out.Close(); err != nil {
+		os.Remove(tmp)
+		return err
+	}
+	return os.Rename(tmp, destPath)
+}
+
+func extensionFromURL(rawURL string) string {
+	u, err := url.Parse(rawURL)
+	if err != nil {
+		return ".mp3"
+	}
+	ext := filepath.Ext(u.Path)
+	if ext == "" {
+		return ".mp3"
+	}
+	return ext
+}
+
+func sanitizeName(name, fallbackID string) string {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		return fallbackID
+	}
+	var b strings.Builder
+	for _, r := range name {
+		switch r {
+		case '/', '\\', '\x00':
+			b.WriteByte('_')
+		default:
+			b.WriteRune(r)
+		}
+	}
+	s := b.String()
+	if len(s) > 100 {
+		s = s[:100]
+	}
+	return s
+}
@@ -0,0 +1,240 @@
+package podcasts
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"git.noroutes.net/NoRoutes/vibetunes/internal/database"
+)
+
+const testFeedXML = `<?xml version="1.0" encoding="UTF-8"?>
+<rss version="2.0" xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd">
+<channel>
+  <title>Test Podcast</title>
+  <description>A podcast for testing</description>
+  <item>
+    <title>Episode One</title>
+    <description>The first episode</description>
+    <guid>episode-1-guid</guid>
+    <pubDate>Mon, 01 Jan 2024 00:00:00 GMT</pubDate>
+    <itunes:duration>125</itunes:duration>
+    <enclosure url="%s/audio.mp3" type="audio/mpeg" length="12345"/>
+  </item>
+  <item>
+    <title>Episode Two</title>
+    <description>The second episode</description>
+    <guid>episode-2-guid</guid>
+    <pubDate>Tue, 02 Jan 2024 00:00:00 GMT</pubDate>
+    <itunes:duration>2:05</itunes:duration>
+    <enclosure url="%s/audio.mp3" type="audio/mpeg" length="12345"/>
+  </item>
+</channel>
+</rss>`
+
+func TestParseRSS(t *testing.T) {
+	feed, err := parseRSS([]byte(fmt.Sprintf(testFeedXML, "http://x", "http://x")))
+	if err != nil {
+		t.Fatalf("parseRSS: %v", err)
+	}
+	if feed.Channel.Title != "Test Podcast" {
+		t.Errorf("Title = %q", feed.Channel.Title)
+	}
+	if len(feed.Channel.Items) != 2 {
+		t.Fatalf("Items = %d, want 2", len(feed.Channel.Items))
+	}
+	if feed.Channel.Items[0].episodeGUID() != "episode-1-guid" {
+		t.Errorf("guid = %q", feed.Channel.Items[0].episodeGUID())
+	}
+}
+
+func TestParseDuration(t *testing.T) {
+	cases := map[string]float64{
+		"125":     125,
+		"2:05":    125,
+		"1:02:05": 3725,
+		"":        0,
+		"bogus":   0,
+	}
+	for in, want := range cases {
+		if got := parseDuration(in); got != want {
+			t.Errorf("parseDuration(%q) = %v, want %v", in, got, want)
+		}
+	}
+}
+
+func newTestManager(t *testing.T) (*Manager, string) {
+	t.Helper()
+	dir := t.TempDir()
+	db, err := database.Open(filepath.Join(dir, "test.db"))
+	if err != nil {
+		t.Fatalf("database.Open: %v", err)
+	}
+	t.Cleanup(func() { db.Close() })
+	log := slog.New(slog.NewTextHandler(io.Discard, nil))
+	return NewManager(db, dir, "", log), dir
+}
+
+func TestRefreshChannel(t *testing.T) {
+	m, _ := newTestManager(t)
+
+	var feedURL string
+	mux := http.NewServeMux()
+	mux.HandleFunc("/feed.xml", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, testFeedXML, feedURL, feedURL)
+	})
+	mux.HandleFunc("/audio.mp3", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("fake mp3 bytes"))
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	feedURL = srv.URL
+
+	channel, err := database.CreatePodcastChannel(m.DB, srv.URL+"/feed.xml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := m.RefreshChannel(ctx, channel.ID); err != nil {
+		t.Fatalf("RefreshChannel: %v", err)
+	}
+
+	updated, err := database.GetPodcastChannel(m.DB, channel.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if updated.Title != "Test Podcast" {
+		t.Errorf("channel title = %q", updated.Title)
+	}
+	if updated.Status != "completed" {
+		t.Errorf("channel status = %q, want completed", updated.Status)
+	}
+
+	episodes, err := database.ListPodcastEpisodes(m.DB, channel.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(episodes) != 2 {
+		t.Fatalf("episodes = %d, want 2", len(episodes))
+	}
+}
+
+func TestRefreshChannelIsIdempotent(t *testing.T) {
+	m, _ := newTestManager(t)
+	var feedURL string
+	mux := http.NewServeMux()
+	mux.HandleFunc("/feed.xml", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, testFeedXML, feedURL, feedURL)
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	feedURL = srv.URL
+
+	channel, err := database.CreatePodcastChannel(m.DB, srv.URL+"/feed.xml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	if err := m.RefreshChannel(ctx, channel.ID); err != nil {
+		t.Fatalf("RefreshChannel (1st): %v", err)
+	}
+	episodes1, _ := database.ListPodcastEpisodes(m.DB, channel.ID)
+
+	if err := m.RefreshChannel(ctx, channel.ID); err != nil {
+		t.Fatalf("RefreshChannel (2nd): %v", err)
+	}
+	episodes2, _ := database.ListPodcastEpisodes(m.DB, channel.ID)
+
+	if len(episodes2) != len(episodes1) {
+		t.Fatalf("refresh duplicated episodes: %d -> %d", len(episodes1), len(episodes2))
+	}
+	if episodes1[0].ID != episodes2[0].ID {
+		t.Error("episode ID changed across refresh")
+	}
+}
+
+func TestDownloadEpisode(t *testing.T) {
+	if _, err := exec.LookPath("ffprobe"); err != nil {
+		t.Skip("ffprobe not installed")
+	}
+	m, _ := newTestManager(t)
+	var feedURL string
+	mux := http.NewServeMux()
+	mux.HandleFunc("/feed.xml", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, testFeedXML, feedURL, feedURL)
+	})
+	mux.HandleFunc("/audio.mp3", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("not really mp3 but bytes"))
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	feedURL = srv.URL
+
+	channel, err := database.CreatePodcastChannel(m.DB, srv.URL+"/feed.xml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	if err := m.RefreshChannel(ctx, channel.ID); err != nil {
+		t.Fatalf("RefreshChannel: %v", err)
+	}
+	episodes, _ := database.ListPodcastEpisodes(m.DB, channel.ID)
+
+	if err := m.DownloadEpisode(ctx, episodes[0].ID); err != nil {
+		t.Fatalf("DownloadEpisode: %v", err)
+	}
+
+	ep, err := database.GetPodcastEpisode(m.DB, episodes[0].ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ep.Status != "completed" {
+		t.Fatalf("episode status = %q, want completed", ep.Status)
+	}
+	if !ep.MediaFileID.Valid {
+		t.Fatal("expected episode to have a media_file_id after download")
+	}
+
+	mf, err := database.GetMediaFile(m.DB, ep.MediaFileID.String)
+	if err != nil {
+		t.Fatalf("GetMediaFile: %v", err)
+	}
+	if mf.SizeBytes == 0 {
+		t.Error("downloaded media file has zero size")
+	}
+}
+
+func TestRecoverInterruptedDownloads(t *testing.T) {
+	m, _ := newTestManager(t)
+	channel, err := database.CreatePodcastChannel(m.DB, "http://example.com/feed.xml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ep, err := database.UpsertPodcastEpisode(m.DB, channel.ID, "guid1", "Ep", "", "", "http://example.com/a.mp3", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := database.SetPodcastEpisodeStatus(m.DB, ep.ID, "downloading"); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := m.RecoverInterruptedDownloads(); err != nil {
+		t.Fatalf("RecoverInterruptedDownloads: %v", err)
+	}
+
+	got, err := database.GetPodcastEpisode(m.DB, ep.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != "new" {
+		t.Fatalf("status after recovery = %q, want new", got.Status)
+	}
+}
@@ -0,0 +1,57 @@
+package podcasts
+
+import "encoding/xml"
+
+// rssFeed is a pragmatic subset of RSS 2.0 (with the de facto iTunes
+// podcast extensions most feeds use) — enough to get channel metadata
+// and each episode's title, description, publish date, and audio
+// enclosure. Real podcast feeds vary a lot in which optional fields they
+// include, so every field here is best-effort.
+type rssFeed struct {
+	Channel rssChannel `xml:"channel"`
+}
+
+type rssChannel struct {
+	Title       string    `xml:"title"`
+	Description string    `xml:"description"`
+	Image       rssImage  `xml:"image"`
+	Items       []rssItem `xml:"item"`
+}
+
+type rssImage struct {
+	URL string `xml:"url"`
+}
+
+type rssItem struct {
+	GUID        rssGUID      `xml:"guid"`
+	Title       string       `xml:"title"`
+	Description string       `xml:"description"`
+	PubDate     string       `xml:"pubDate"`
+	Enclosure   rssEnclosure `xml:"enclosure"`
+	Duration    string       `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd duration"`
+}
+
+type rssGUID struct {
+	Value string `xml:",chardata"`
+}
+
+type rssEnclosure struct {
+	URL string `xml:"url,attr"`
+}
+
+func parseRSS(data []byte) (*rssFeed, error) {
+	var feed rssFeed
+	if err := xml.Unmarshal(data, &feed); err != nil {
+		return nil, err
+	}
+	return &feed, nil
+}
+
+// episodeGUID returns the item's GUID if present, falling back to its
+// enclosure URL — some minimal feeds omit <guid> entirely.
+func (i rssItem) episodeGUID() string {
+	if i.GUID.Value != "" {
+		return i.GUID.Value
+	}
+	return i.Enclosure.URL
+}