John O'Keefe
a6723e6ac8
One ticker re-reading its interval from the settings registry each cycle (live tunable — the cache stays honest because writes go through the settings API path, AGENTS.md gotcha #2). Per tick: list feeds (first-admin identity, invariant 12), skip is_complete feeds (back off; explicit refresh still works), poll each (conditional GET via stored validators), and apply the subscriber auto-download policy union after each successful fetch so new episodes queue without waiting a tick. Schema/registry: podcast_poll_interval_seconds (default 3600s, floor 300s, category podcasts) — seeded with ON CONFLICT DO NOTHING per the idempotency rule.
79 lines
2.3 KiB
Go
79 lines
2.3 KiB
Go
// Podcast feed poller (plan §7 task 1.6): one periodic worker ticking
|
|
// on the podcast_poll_interval_seconds system setting (re-read each
|
|
// cycle — the settings API path keeps the registry cache honest),
|
|
// polling every non-complete feed and applying subscriber auto-download
|
|
// policies after each successful fetch (D6).
|
|
//
|
|
// RLS: detached work under the first-admin identity — the service's
|
|
// adminCtx supplies it (invariant 12). Back-off: itunes:complete feeds
|
|
// are skipped by the ticker (explicit refresh still works).
|
|
package services
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
|
|
"bookhoard/internal/database"
|
|
)
|
|
|
|
type PodcastPoller struct {
|
|
db *database.Queries
|
|
svc *PodcastService
|
|
worker *PodcastDownloadWorker
|
|
settings *database.SettingsRegistry
|
|
}
|
|
|
|
func NewPodcastPoller(db *database.Queries, svc *PodcastService, worker *PodcastDownloadWorker, settings *database.SettingsRegistry) *PodcastPoller {
|
|
return &PodcastPoller{db: db, svc: svc, worker: worker, settings: settings}
|
|
}
|
|
|
|
// Start launches the poll loop; exits when ctx is cancelled.
|
|
func (p *PodcastPoller) Start(ctx context.Context) {
|
|
go func() {
|
|
for {
|
|
interval := p.settings.PodcastPollInterval()
|
|
if interval < time.Minute {
|
|
interval = time.Minute // floor against bad settings rows
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(interval):
|
|
}
|
|
p.tick(ctx)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// tick polls every non-complete feed once.
|
|
func (p *PodcastPoller) tick(ctx context.Context) {
|
|
actx, err := p.svc.adminCtx(ctx)
|
|
if err != nil {
|
|
log.Printf("[PODCAST-POLL] no acting identity, skipping tick: %v", err)
|
|
return
|
|
}
|
|
feeds, err := p.db.ListPodcastFeedsForPolling(actx)
|
|
if err != nil {
|
|
log.Printf("[PODCAST-POLL] worklist: %v", err)
|
|
return
|
|
}
|
|
for _, feed := range feeds {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if feed.IsComplete {
|
|
continue // back off ended feeds (explicit refresh still works)
|
|
}
|
|
if err := p.svc.PollFeed(actx, &feed); err != nil {
|
|
log.Printf("[PODCAST-POLL] %s: %v", feed.Title, err)
|
|
continue
|
|
}
|
|
// Fresh episodes → apply the subscriber policy union (auto-
|
|
// download without waiting for the next tick).
|
|
if _, err := p.worker.ApplyAutoPolicies(actx, feed.ID); err != nil {
|
|
log.Printf("[PODCAST-POLL] auto-download policies for %s: %v", feed.Title, err)
|
|
}
|
|
}
|
|
}
|