351 lines
10 KiB
Go
351 lines
10 KiB
Go
// 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"
|
|
"git.noroutes.net/NoRoutes/vibetunes/internal/netguard"
|
|
)
|
|
|
|
const (
|
|
maxConcurrentDownloads = 2
|
|
// maxEnclosureSize bounds a single episode download so a malicious
|
|
// or misbehaving feed can't exhaust disk space; legitimate podcast
|
|
// episodes are essentially never larger than this.
|
|
maxEnclosureSize = 4 << 30 // 4GB
|
|
)
|
|
|
|
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{}
|
|
}
|
|
|
|
// NewManager builds a podcast manager whose HTTP client refuses to
|
|
// connect to non-public (loopback/LAN/link-local/cloud-metadata)
|
|
// addresses — feed URLs and episode enclosure URLs are attacker- or
|
|
// admin-supplied and must never turn VibeSonic into an internal
|
|
// network proxy. Set allowPrivateNetworks only for deployments that
|
|
// intentionally host feeds on a private network.
|
|
func NewManager(db *sql.DB, dataDir, ffprobePath string, log *slog.Logger, allowPrivateNetworks bool) *Manager {
|
|
return &Manager{
|
|
DB: db,
|
|
DataDir: dataDir,
|
|
FFprobePath: ffprobePath,
|
|
HTTPClient: netguard.NewSafeHTTPClient(60*time.Second, allowPrivateNetworks),
|
|
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", "")
|
|
|
|
if err := netguard.ValidateURL(channel.URL); err != nil {
|
|
_ = database.SetPodcastChannelStatus(m.DB, channelID, "error", err.Error())
|
|
return err
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
// DownloadChannel downloads every episode in a channel that is not already
|
|
// complete or in progress. Downloads are run in feed order and still pass
|
|
// through the manager's concurrency limiter.
|
|
func (m *Manager) DownloadChannel(ctx context.Context, channelID string) error {
|
|
episodes, err := database.ListPodcastEpisodes(m.DB, channelID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, episode := range episodes {
|
|
if episode.Status == "completed" || episode.Status == "downloading" {
|
|
continue
|
|
}
|
|
if err := m.DownloadEpisode(ctx, episode.ID); err != nil {
|
|
if m.Log != nil {
|
|
m.Log.Warn("podcast episode download failed", "episode", episode.ID, "err", err)
|
|
}
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (m *Manager) downloadToFile(ctx context.Context, sourceURL, destPath string) error {
|
|
if err := netguard.ValidateURL(sourceURL); err != nil {
|
|
return err
|
|
}
|
|
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
|
|
}
|
|
n, err := io.Copy(out, io.LimitReader(resp.Body, maxEnclosureSize+1))
|
|
if err != nil {
|
|
out.Close()
|
|
os.Remove(tmp)
|
|
return err
|
|
}
|
|
if n > maxEnclosureSize {
|
|
out.Close()
|
|
os.Remove(tmp)
|
|
return fmt.Errorf("enclosure exceeds maximum allowed size of %d bytes", maxEnclosureSize)
|
|
}
|
|
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
|
|
}
|