feat(podcasts): feed poller + podcast_poll_interval_seconds setting (task 1.6)
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.
This commit is contained in:
1 parent
4288a86b2d
commit
a6723e6ac8
4 files changed
+129
-7
No files matched your search
@@ -100,7 +100,9 @@ INSERT INTO system_settings (setting_key, setting_value, description, setting_ty
|
|||||||
('sync_queue_interval_seconds', '5', 'How often the sync queue flushes', 'int', '1', '3600', TRUE, 'sync'),
|
('sync_queue_interval_seconds', '5', 'How often the sync queue flushes', 'int', '1', '3600', TRUE, 'sync'),
|
||||||
('sync_queue_batch_size', '50', 'Maximum items processed per sync queue flush', 'int', '1', '10000', TRUE, 'sync'),
|
('sync_queue_batch_size', '50', 'Maximum items processed per sync queue flush', 'int', '1', '10000', TRUE, 'sync'),
|
||||||
('worker_pool_size', '3', 'Number of background worker goroutines', 'int', '1', '100', TRUE, 'performance'),
|
('worker_pool_size', '3', 'Number of background worker goroutines', 'int', '1', '100', TRUE, 'performance'),
|
||||||
('worker_queue_cap', '100', 'Background worker job queue capacity', 'int', '1', '10000', TRUE, 'performance')
|
('worker_queue_cap', '100', 'Background worker job queue capacity', 'int', '1', '10000', TRUE, 'performance'),
|
||||||
|
-- podcasts
|
||||||
|
('podcast_poll_interval_seconds', '3600', 'How often to poll podcast feeds for new episodes (seconds; complete feeds back off)', 'int', '300', '604800', FALSE, 'podcasts')
|
||||||
ON CONFLICT (setting_key) DO NOTHING;
|
ON CONFLICT (setting_key) DO NOTHING;
|
||||||
|
|
||||||
-- Create refresh_tokens table
|
-- Create refresh_tokens table
|
||||||
|
|||||||
@@ -4943,7 +4943,7 @@ WHERE source_feed_id = $1
|
|||||||
-- source filter keeps scan items out of podcast search.
|
-- source filter keeps scan items out of podcast search.
|
||||||
SELECT * FROM media_items
|
SELECT * FROM media_items
|
||||||
WHERE source = 'podcast_episode'
|
WHERE source = 'podcast_episode'
|
||||||
AND ($1 = '' OR title ILIKE '%' || $1 || '%' OR description ILIKE '%' || $1 || '%')
|
AND (sqlc.arg('search_query')::text = '' OR title ILIKE '%' || sqlc.arg('search_query')::text || '%' OR description ILIKE '%' || sqlc.arg('search_query')::text || '%')
|
||||||
ORDER BY feed_position ASC NULLS LAST, date_published ASC NULLS LAST
|
ORDER BY feed_position ASC NULLS LAST, date_published ASC NULLS LAST
|
||||||
LIMIT 50;
|
LIMIT 50;
|
||||||
|
|
||||||
@@ -4957,9 +4957,43 @@ SELECT f.*,
|
|||||||
s.created_at AS subscribed_at
|
s.created_at AS subscribed_at
|
||||||
FROM podcast_subscriptions s
|
FROM podcast_subscriptions s
|
||||||
JOIN podcast_feeds f ON f.id = s.feed_id
|
JOIN podcast_feeds f ON f.id = s.feed_id
|
||||||
WHERE s.user_id = $1
|
WHERE s.user_id = sqlc.arg('user_id')
|
||||||
AND ($2 = '' OR f.title ILIKE '%' || $2 || '%'
|
AND (sqlc.arg('search_query')::text = '' OR f.title ILIKE '%' || sqlc.arg('search_query')::text || '%'
|
||||||
OR COALESCE(s.display_name, '') ILIKE '%' || $2 || '%'
|
OR COALESCE(s.display_name, '') ILIKE '%' || sqlc.arg('search_query')::text || '%'
|
||||||
OR COALESCE(f.host, '') ILIKE '%' || $2 || '%'
|
OR COALESCE(f.host, '') ILIKE '%' || sqlc.arg('search_query')::text || '%'
|
||||||
OR COALESCE(f.publisher, '') ILIKE '%' || $2 || '%')
|
OR COALESCE(f.publisher, '') ILIKE '%' || sqlc.arg('search_query')::text || '%')
|
||||||
ORDER BY COALESCE(s.display_name, f.title) ASC;
|
ORDER BY COALESCE(s.display_name, f.title) ASC;
|
||||||
|
|
||||||
|
-- name: GetNextQueuedEpisode :one
|
||||||
|
-- Sequential download worker's claim source (single worker, day one).
|
||||||
|
SELECT * FROM media_items
|
||||||
|
WHERE source = 'podcast_episode' AND download_state = 'queued'
|
||||||
|
ORDER BY updated_at ASC
|
||||||
|
LIMIT 1;
|
||||||
|
|
||||||
|
-- name: CountQueuedEpisodes :one
|
||||||
|
SELECT COUNT(*) FROM media_items
|
||||||
|
WHERE source = 'podcast_episode' AND download_state IN ('queued', 'downloading');
|
||||||
|
|
||||||
|
-- name: CountPodcastEpisodesForFeed :one
|
||||||
|
SELECT COUNT(*) FROM media_items
|
||||||
|
WHERE source_feed_id = $1 AND source = 'podcast_episode';
|
||||||
|
|
||||||
|
-- ============================================================================
|
||||||
|
-- GENERIC reader_settings KV (podcast order override, plan D5: key
|
||||||
|
-- podcast_order:<feed_id>). The pre-existing reader_settings queries are
|
||||||
|
-- hard-keyed to 'reader_settings'; these two are the generic form.
|
||||||
|
-- ============================================================================
|
||||||
|
|
||||||
|
-- name: GetUserSettingByKey :one
|
||||||
|
SELECT setting_value FROM reader_settings
|
||||||
|
WHERE user_id = $1 AND setting_key = $2;
|
||||||
|
|
||||||
|
-- name: UpsertUserSettingByKey :exec
|
||||||
|
INSERT INTO reader_settings (user_id, setting_key, setting_value)
|
||||||
|
VALUES ($1, $2, $3)
|
||||||
|
ON CONFLICT (user_id, setting_key)
|
||||||
|
DO UPDATE SET setting_value = EXCLUDED.setting_value, updated_at = NOW();
|
||||||
|
|
||||||
|
-- name: DeleteUserSettingByKey :exec
|
||||||
|
DELETE FROM reader_settings WHERE user_id = $1 AND setting_key = $2;
|
||||||
@@ -82,6 +82,7 @@ var SettingDefaults = []SettingDefault{
|
|||||||
{Key: "sync_queue_batch_size", Value: "50", Type: SettingTypeInt, Min: "1", Max: "10000", RequiresRestart: true, Category: "sync", Group: "Sync Queue", Description: "Maximum items processed per sync queue flush"},
|
{Key: "sync_queue_batch_size", Value: "50", Type: SettingTypeInt, Min: "1", Max: "10000", RequiresRestart: true, Category: "sync", Group: "Sync Queue", Description: "Maximum items processed per sync queue flush"},
|
||||||
{Key: "worker_pool_size", Value: "3", Type: SettingTypeInt, Min: "1", Max: "100", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Number of background worker goroutines"},
|
{Key: "worker_pool_size", Value: "3", Type: SettingTypeInt, Min: "1", Max: "100", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Number of background worker goroutines"},
|
||||||
{Key: "worker_queue_cap", Value: "100", Type: SettingTypeInt, Min: "1", Max: "10000", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Background worker job queue capacity"},
|
{Key: "worker_queue_cap", Value: "100", Type: SettingTypeInt, Min: "1", Max: "10000", RequiresRestart: true, Category: "performance", Group: "Worker Pool", Description: "Background worker job queue capacity"},
|
||||||
|
{Key: "podcast_poll_interval_seconds", Value: "3600", Type: SettingTypeInt, Min: "300", Max: "604800", Category: "podcasts", Group: "Podcasts", Description: "How often to poll podcast feeds for new episodes (seconds; complete feeds back off)"},
|
||||||
}
|
}
|
||||||
|
|
||||||
// defaultBy indexes SettingDefaults by key for O(1) lookup.
|
// defaultBy indexes SettingDefaults by key for O(1) lookup.
|
||||||
@@ -188,6 +189,13 @@ func (r *SettingsRegistry) ScanPollInterval() time.Duration {
|
|||||||
return time.Duration(r.getInt("scan_poll_interval_seconds")) * time.Second
|
return time.Duration(r.getInt("scan_poll_interval_seconds")) * time.Second
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PodcastPollInterval is the feed-poller tick (live setting; the poller
|
||||||
|
// re-reads it every cycle — writes go through the settings API path,
|
||||||
|
// AGENTS.md gotcha #2).
|
||||||
|
func (r *SettingsRegistry) PodcastPollInterval() time.Duration {
|
||||||
|
return time.Duration(r.getInt("podcast_poll_interval_seconds")) * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
// AutoScanEnabled reports whether auto-scanning is on.
|
// AutoScanEnabled reports whether auto-scanning is on.
|
||||||
func (r *SettingsRegistry) AutoScanEnabled() bool { return r.getBool("auto_scan_enabled") }
|
func (r *SettingsRegistry) AutoScanEnabled() bool { return r.getBool("auto_scan_enabled") }
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,78 @@
|
|||||||
|
// 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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user