diff --git a/database/schema/schema.sql b/database/schema/schema.sql index e84419ec..ab80855d 100644 --- a/database/schema/schema.sql +++ b/database/schema/schema.sql @@ -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_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_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; -- Create refresh_tokens table diff --git a/internal/database/queries/queries.sql b/internal/database/queries/queries.sql index 9372f5e5..59209fce 100644 --- a/internal/database/queries/queries.sql +++ b/internal/database/queries/queries.sql @@ -4943,7 +4943,7 @@ WHERE source_feed_id = $1 -- source filter keeps scan items out of podcast search. SELECT * FROM media_items 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 LIMIT 50; @@ -4957,9 +4957,43 @@ SELECT f.*, s.created_at AS subscribed_at FROM podcast_subscriptions s JOIN podcast_feeds f ON f.id = s.feed_id -WHERE s.user_id = $1 - AND ($2 = '' OR f.title ILIKE '%' || $2 || '%' - OR COALESCE(s.display_name, '') ILIKE '%' || $2 || '%' - OR COALESCE(f.host, '') ILIKE '%' || $2 || '%' - OR COALESCE(f.publisher, '') ILIKE '%' || $2 || '%') +WHERE s.user_id = sqlc.arg('user_id') + AND (sqlc.arg('search_query')::text = '' OR f.title ILIKE '%' || sqlc.arg('search_query')::text || '%' + OR COALESCE(s.display_name, '') ILIKE '%' || sqlc.arg('search_query')::text || '%' + OR COALESCE(f.host, '') ILIKE '%' || sqlc.arg('search_query')::text || '%' + OR COALESCE(f.publisher, '') ILIKE '%' || sqlc.arg('search_query')::text || '%') 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:). 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; diff --git a/internal/database/settings_registry.go b/internal/database/settings_registry.go index 7b33fcda..8c9fb847 100644 --- a/internal/database/settings_registry.go +++ b/internal/database/settings_registry.go @@ -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: "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: "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. @@ -188,6 +189,13 @@ func (r *SettingsRegistry) ScanPollInterval() time.Duration { 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. func (r *SettingsRegistry) AutoScanEnabled() bool { return r.getBool("auto_scan_enabled") } diff --git a/internal/services/podcast_poller.go b/internal/services/podcast_poller.go new file mode 100644 index 00000000..8868bf3d --- /dev/null +++ b/internal/services/podcast_poller.go @@ -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) + } + } +}