diff --git a/README.md b/README.md index ec7fbdd..654e9f8 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,67 @@ rssamp schedule auto-post --platform bluesky --interval daily - `rssamp feeds list` - List all feeds - `rssamp feeds refresh` - Manually refresh feeds +### Feed Catalogue + +For catalogues rather than a handful of hand-picked feeds. These commands use a +SQLite store (`~/.config/rss-amplifier/feeds.db`) instead of `feeds.json`, +because the JSON store loads every feed into memory and rewrites the whole file +on each change β€” fine for fifty feeds, fatal for fifty thousand. + +- `rssamp feeds import-opml --file feeds.opml` - Stream an OPML catalogue in. + Tested at 47,000 feeds in under two seconds. +- `rssamp feeds harvest-podcasts [--all]` - Build a podcast catalogue from the + iTunes Search API (free, no key). `--all` sweeps ten markets instead of one. + Paced under Apple's rate limit; Ctrl-C is safe and keeps what it found. +- `rssamp feeds stats` - Catalogue size, what is due, what is failing. +- `rssamp feeds recent [--kind podcast]` - Newest articles collected. + +#### Re-syncing from brisk.news + +brisk.news holds the same feeds in Supabase (`rss_feed_sources`), split into +`smallweb` (the Kagi catalogue) and `opml` (feeds collected by hand over the +years β€” the ones worth keeping). Export them as OPML server-side so tens of +thousands of rows never travel through an agent's context: + +```sql +select '' + || 'brisk.news feeds' + || string_agg( + '', '') + || '' + from rss_feed_sources + where source_origin = 'opml'; -- or drop the filter for all ~33k +``` + +Then `rssamp feeds import-opml --file --origin brisk`. Re-running is +safe: feed URLs are unique and normalised, so an existing feed is skipped rather +than duplicated. + +### The Poller Daemon + +```bash +rssamp daemon start # 8 feeds at a time, 2s between batches +rssamp daemon start --batch 16 --pause 5 +rssamp daemon status # same as `feeds stats` +``` + +Keeping tens of thousands of feeds current is only affordable because almost +every poll costs nothing: + +- **Conditional GET.** Each feed stores its `ETag`/`Last-Modified`, so an + unchanged feed answers `304` β€” no body, no parsing. Servers that ignore + those get the same cheap path via a content hash. +- **Adaptive intervals.** A feed that publishes is checked sooner; one that + never changes backs off, up to a day. +- **Backoff and eviction.** Failures back off exponentially, and a feed that + fails repeatedly is deactivated rather than retried forever. +- **Bounded everything.** One feed per host per batch, a request timeout, and + a download size cap. + +The database is the queue β€” there is no Redis and no job server. The daemon can +be killed at any moment and resumes exactly where it left off. + ### Snippet Management - `rssamp snippets generate [options]` - Generate AI snippets - `rssamp snippets list` - List all snippets diff --git a/bin/rss-amplifier.js b/bin/rss-amplifier.js index 201cecd..d8c266b 100755 --- a/bin/rss-amplifier.js +++ b/bin/rss-amplifier.js @@ -16,6 +16,13 @@ dotenv.config(); // Import core modules import { loadConfig, getConfigPath, getPlatformDisplayName, getAIConfig, isAIReady } from '../src/config-manager.js'; import { SetupWizard } from '../src/setup-wizard.js'; +import { + importOpmlCommand, + harvestPodcastsCommand, + feedStatsCommand, + recentArticlesCommand, + daemonCommand, +} from '../src/feed-commands.js'; /** * Handle setup command @@ -164,8 +171,41 @@ async function handleImportCommand(argv) { * Handle feeds command (placeholder) */ async function handleFeedsCommand(argv) { - console.log(colors.yellow('πŸ“° Feed management coming soon...')); - console.log(colors.cyan('This feature will be implemented in the next development phase.')); + switch (argv.action) { + case 'import-opml': { + // `--file` or the trailing positional, so both spellings work. + const file = argv.file ?? argv._?.[1]; + if (!file) { + console.error(colors.red('Which OPML file? Pass --file ')); + process.exit(1); + } + await importOpmlCommand({ ...argv, file }); + return; + } + case 'harvest-podcasts': + await harvestPodcastsCommand(argv); + return; + case 'stats': + await feedStatsCommand(argv); + return; + case 'recent': + await recentArticlesCommand(argv); + return; + default: + console.log(colors.yellow('πŸ“° That feed action is not implemented yet.')); + console.log(colors.cyan('Available now: import-opml, harvest-podcasts, stats, recent')); + } +} + +/** + * Handle daemon command + */ +async function handleDaemonCommand(argv) { + if (argv.action === 'status') { + await feedStatsCommand(argv); + return; + } + await daemonCommand(argv); } /** @@ -230,7 +270,59 @@ function configureCommandLine() { .positional('action', { describe: 'Action to perform', type: 'string', - choices: ['list', 'refresh', 'add', 'remove'] + choices: ['list', 'refresh', 'add', 'remove', 'import-opml', 'harvest-podcasts', 'stats', 'recent'] + }) + .option('file', { + describe: 'OPML file to import (import-opml)', + type: 'string' + }) + .option('db', { + describe: 'Path to the feed database', + type: 'string' + }) + .option('origin', { + describe: 'Label recorded as the source of imported feeds', + type: 'string' + }) + .option('all', { + describe: 'Sweep every market rather than just the US (harvest-podcasts)', + type: 'boolean', + default: false + }) + .option('delay', { + describe: 'Milliseconds between search requests (harvest-podcasts)', + type: 'number' + }) + .option('kind', { + describe: 'Filter by feed kind', + type: 'string', + choices: ['blog', 'podcast', 'unknown'] + }) + .option('limit', { + describe: 'How many rows to show', + type: 'number' + }); + }) + .command('daemon ', 'Run the feed polling daemon', (yargs) => { + return yargs + .positional('action', { + describe: 'Action to perform', + type: 'string', + choices: ['start', 'status'] + }) + .option('db', { + describe: 'Path to the feed database', + type: 'string' + }) + .option('batch', { + describe: 'Feeds fetched concurrently per batch', + type: 'number', + default: 8 + }) + .option('pause', { + describe: 'Seconds to wait between batches', + type: 'number', + default: 2 }); }) .command('snippets ', 'Manage snippets', (yargs) => { @@ -320,6 +412,9 @@ async function main() { case 'feeds': await handleFeedsCommand(argv); break; + case 'daemon': + await handleDaemonCommand(argv); + break; case 'snippets': await handleSnippetsCommand(argv); break; diff --git a/src/feed-commands.js b/src/feed-commands.js new file mode 100644 index 0000000..c517c1f --- /dev/null +++ b/src/feed-commands.js @@ -0,0 +1,203 @@ +/** + * CLI handlers for the feed catalogue and its daemon. + * + * Kept out of `bin/rss-amplifier.js` because that file is already long and + * these commands are the only ones that touch the SQLite store. + */ + +import colors from 'ansi-colors'; +import { FeedStore } from './feed-store.js'; +import { importOpmlFile } from './opml-import.js'; +import { FeedDaemon, DEFAULTS } from './feed-daemon.js'; +import { harvestPodcasts, COUNTRIES } from './podcast-harvest.js'; + +function withStore(argv, run) { + const store = new FeedStore({ dbPath: argv.db }); + try { + return run(store); + } finally { + store.close(); + } +} + +/** `feeds import-opml ` β€” stream an OPML catalogue into the store. */ +export async function importOpmlCommand(argv) { + const store = new FeedStore({ dbPath: argv.db }); + + try { + console.log(colors.cyan(`Importing ${argv.file}…`)); + + let added = 0; + let batches = 0; + + const summary = await importOpmlFile(argv.file, async (batch) => { + const result = store.addFeeds(batch, { sourceOrigin: argv.origin ?? 'opml' }); + added += result.added; + batches += 1; + if (batches % 10 === 0) { + process.stdout.write(colors.gray(`\r ${added} new feeds stored…`)); + } + }); + + process.stdout.write('\r'); + console.log(colors.green(`βœ… ${added} new feeds stored`)); + console.log( + colors.gray( + ` ${summary.unique} unique in file, ${summary.total - added} already known` + + (summary.malformed ? `, ${summary.malformed} unusable` : ''), + ), + ); + + const stats = store.stats(); + console.log(colors.gray(` catalogue is now ${stats.feeds} feeds`)); + console.log( + colors.yellow('\nNothing has been fetched yet β€” start the daemon to poll them:'), + ); + console.log(colors.gray(' rss-amplifier daemon start')); + } finally { + store.close(); + } +} + +/** `feeds harvest-podcasts` β€” build a podcast catalogue from iTunes. */ +export async function harvestPodcastsCommand(argv) { + const store = new FeedStore({ dbPath: argv.db }); + const controller = new AbortController(); + + const onSignal = () => { + console.log(colors.yellow('\nStopping β€” everything found so far is saved.')); + controller.abort(); + }; + process.once('SIGINT', onSignal); + + try { + const countries = argv.all ? COUNTRIES : ['US']; + console.log(colors.cyan(`Harvesting podcasts from iTunes (${countries.length} market(s))…`)); + console.log(colors.gray(' Paced to stay under Apple\'s rate limit; Ctrl-C is safe.\n')); + + const summary = await harvestPodcasts( + async (batch) => { + store.addFeeds(batch, { sourceOrigin: 'itunes' }); + }, + { + countries, + signal: controller.signal, + delayMs: argv.delay ?? 3_500, + onProgress: (progress) => { + if (progress.rateLimited) { + process.stdout.write( + colors.yellow(`\r rate limited, backing off ${Math.round(progress.backoffMs / 1000)}s… `), + ); + return; + } + process.stdout.write( + colors.gray( + `\r ${progress.index}/${progress.total} "${progress.query.term}" ` + + `(+${progress.fresh}) β€” ${progress.stored} unique so far `, + ), + ); + }, + }, + ); + + process.stdout.write('\r'); + console.log(colors.green(`\nβœ… ${summary.stored} podcast feeds stored from ${summary.queries} searches`)); + if (summary.failures) console.log(colors.gray(` ${summary.failures} searches failed`)); + } finally { + process.off('SIGINT', onSignal); + store.close(); + } +} + +/** `feeds stats` β€” what the catalogue looks like. */ +export async function feedStatsCommand(argv) { + withStore(argv, (store) => { + const stats = store.stats(); + + console.log(colors.green('πŸ“Š Feed catalogue')); + console.log(''); + console.log(colors.cyan(` Feeds: ${stats.feeds}`)); + console.log(colors.cyan(` Active: ${stats.active}`)); + console.log(colors.cyan(` Due now: ${stats.due}`)); + console.log(colors.cyan(` Never fetched: ${stats.neverFetched}`)); + console.log(colors.cyan(` Failing: ${stats.failing}`)); + console.log(colors.cyan(` Articles: ${stats.articles}`)); + + if (stats.byKind?.length) { + console.log(''); + for (const row of stats.byKind) { + console.log(colors.gray(` ${row.kind.padEnd(10)} ${row.n}`)); + } + } + }); +} + +/** `feeds recent` β€” the newest articles the daemon has collected. */ +export async function recentArticlesCommand(argv) { + withStore(argv, (store) => { + const articles = store.recentArticles({ + limit: argv.limit ?? 20, + ...(argv.kind ? { kind: argv.kind } : {}), + }); + + if (articles.length === 0) { + console.log(colors.yellow('No articles yet. Is the daemon running?')); + return; + } + + for (const article of articles) { + const when = article.published_at?.slice(0, 10) ?? ' '; + console.log(`${colors.gray(when)} ${colors.cyan(article.feed_title ?? '')}`); + console.log(` ${article.title ?? '(untitled)'}`); + } + }); +} + +/** `daemon start` β€” poll the catalogue until stopped. */ +export async function daemonCommand(argv) { + const daemon = new FeedDaemon({ + ...(argv.db ? { dbPath: argv.db } : {}), + batchSize: argv.batch ?? DEFAULTS.batchSize, + batchDelayMs: (argv.pause ?? 2) * 1000, + }); + + const stop = () => { + console.log(colors.yellow('\nStopping after this batch…')); + daemon.stop(); + }; + process.on('SIGINT', stop); + process.on('SIGTERM', stop); + + console.log(colors.green('πŸ” Feed daemon started')); + console.log( + colors.gray( + ` ${daemon.config.batchSize} feeds at a time, ` + + `${daemon.config.batchDelayMs / 1000}s between batches. Ctrl-C to stop.\n`, + ), + ); + + const started = Date.now(); + + const totals = await daemon.start({ + onTick: (result, running) => { + if (result.claimed === 0) return; + const mins = Math.max((Date.now() - started) / 60_000, 0.01); + process.stdout.write( + colors.gray( + `\r ${running.batches} batches Β· ${running.updated} updated Β· ` + + `${running.unchanged} unchanged Β· ${running.failed} failed Β· ` + + `${running.newArticles} new articles Β· ` + + `${Math.round((running.updated + running.unchanged + running.failed) / mins)}/min `, + ), + ); + }, + }); + + console.log(colors.green(`\n\nβœ… Stopped after ${totals.batches} batches`)); + console.log( + colors.gray( + ` ${totals.updated} updated, ${totals.unchanged} unchanged, ` + + `${totals.failed} failed, ${totals.newArticles} new articles`, + ), + ); +} diff --git a/src/feed-daemon.js b/src/feed-daemon.js new file mode 100644 index 0000000..10171bf --- /dev/null +++ b/src/feed-daemon.js @@ -0,0 +1,427 @@ +/** + * The polling daemon. + * + * Keeping 47,000 feeds current is only affordable if almost every check costs + * nothing. Five things make that true, and each one is doing real work: + * + * 1. **A small batch, always.** Eight feeds are in flight at a time. Never + * eight thousand. The queue is an index on `next_fetch_at`, so "what is + * due" is a cheap indexed read rather than a scan of the catalogue. + * 2. **Conditional GET.** Every feed remembers its `ETag` and + * `Last-Modified`. A server that answers `304 Not Modified` costs a few + * hundred bytes and no parsing at all, which is what the overwhelming + * majority of polls should be. + * 3. **Adaptive intervals.** A feed that keeps returning nothing new is + * checked progressively less often, up to a day. A feed that publishes + * every time we look is checked more often. The catalogue converges on + * spending its effort where things actually change. + * 4. **Backoff and eviction.** Dead domains are the single largest source of + * waste in a catalogue this size. Failures back off exponentially, and a + * feed that fails enough times in a row is deactivated rather than + * retried forever. + * 5. **Idle between batches.** The loop sleeps between batches and sleeps + * longer when nothing is due, so a caught-up daemon is close to free. + * + * There is deliberately no job queue and no Redis. The database *is* the + * queue β€” `next_fetch_at` is the cursor β€” which means the daemon can be killed + * at any moment and resumes exactly where it stopped. + */ + +import { createHash } from 'node:crypto'; +import { FeedStore } from './feed-store.js'; + +export const DEFAULTS = { + /** Feeds fetched concurrently. The main politeness and memory knob. */ + batchSize: 8, + /** Pause between batches, so the loop yields the machine. */ + batchDelayMs: 2_000, + /** Sleep when nothing is due at all. */ + idleDelayMs: 60_000, + /** Per-request timeout. A hung server must not stall a batch. */ + timeoutMs: 15_000, + /** Hard cap on a downloaded feed. Some "feeds" are enormous. */ + maxBytes: 4 * 1024 * 1024, + /** Interval bounds, in minutes. */ + minIntervalMin: 30, + maxIntervalMin: 1440, + baseIntervalMin: 360, + /** Consecutive failures before a feed is switched off. */ + maxFailures: 8, + /** Never hit the same host more than once per batch. */ + hostCooldownMs: 5_000, + retentionDays: 30, + userAgent: 'RSS-Amplifier/1.0 (+https://rssamplifier.com; feed poller)', +}; + +/** Sleep that can be cut short when the daemon is stopping. */ +function sleep(ms, signal) { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + signal?.addEventListener('abort', () => { + clearTimeout(timer); + resolve(); + }, { once: true }); + }); +} + +/** + * Fetches one feed, honouring its stored validators. + * + * Returns `{ notModified: true }` for a 304, which is the cheap path the whole + * design leans on. The body is read as a stream with a byte cap so a + * misconfigured server cannot hand back a gigabyte and take the daemon with it. + */ +export async function fetchFeed(feed, options = {}) { + const config = { ...DEFAULTS, ...options }; + const headers = { 'user-agent': config.userAgent, accept: 'application/rss+xml, application/atom+xml, application/xml, text/xml, */*' }; + + if (feed.etag) headers['if-none-match'] = feed.etag; + if (feed.last_modified) headers['if-modified-since'] = feed.last_modified; + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), config.timeoutMs); + + try { + const response = await fetch(feed.feed_url, { + headers, + signal: controller.signal, + redirect: 'follow', + }); + + if (response.status === 304) { + return { notModified: true, status: 304 }; + } + + if (!response.ok) { + return { ok: false, status: response.status, error: `HTTP ${response.status}` }; + } + + // Read with a cap rather than `response.text()`, which would buffer + // whatever the server decided to send. + const reader = response.body?.getReader(); + if (!reader) return { ok: false, status: response.status, error: 'empty body' }; + + const chunks = []; + let size = 0; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + size += value.length; + if (size > config.maxBytes) { + await reader.cancel(); + return { ok: false, status: response.status, error: 'feed exceeds size limit' }; + } + chunks.push(value); + } + + const body = Buffer.concat(chunks).toString('utf8'); + + return { + ok: true, + status: response.status, + body, + etag: response.headers.get('etag') ?? undefined, + lastModified: response.headers.get('last-modified') ?? undefined, + }; + } catch (error) { + const message = error?.name === 'AbortError' ? 'timed out' : (error?.message ?? String(error)); + return { ok: false, error: message }; + } finally { + clearTimeout(timer); + } +} + +const TAG = (name) => new RegExp(`<${name}[^>]*>([\\s\\S]*?)`, 'i'); + +function stripCdata(value) { + return value?.replace(/^\s*\s*$/, '$1').trim(); +} + +function decode(value) { + if (!value) return value; + return value + .replace(/</g, '<') + .replace(/>/g, '>') + .replace(/"/g, '"') + .replace(/�?39;|'/g, "'") + .replace(/&#(\d+);/g, (_, code) => String.fromCodePoint(Number(code))) + // Ampersand last, or a double-escaped entity decodes twice. + .replace(/&/g, '&'); +} + +function field(block, ...names) { + for (const name of names) { + const match = block.match(TAG(name)); + if (match?.[1]) { + const value = decode(stripCdata(match[1]).replace(/<[^>]+>/g, '').trim()); + if (value) return value; + } + } + return undefined; +} + +/** + * Parses RSS or Atom into articles. + * + * Regex rather than a full XML parser, for the same reason the OPML import + * streams: this runs across tens of thousands of documents and only ever needs + * a handful of fields from each. It is tolerant of the malformed markup that + * is completely normal in the wild, where a strict parser would simply reject + * the feed and lose it. + */ +export function parseFeed(xml) { + if (!xml || typeof xml !== 'string') return { kind: 'unknown', articles: [] }; + + // A podcast is an RSS feed carrying the iTunes namespace and enclosures. + const isPodcast = + /xmlns:itunes\s*=/i.test(xml) || /]*rel=["']alternate["'][^>]*href=["']([^"']+)["']/i) ?? + head.match(/\s*([^<\s][^<]*)<\/link>/i); + const siteUrl = linkMatch?.[1]?.trim(); + + const articles = []; + // `]*)?>([\s\S]*?)<\/\1>/gi; + + let match; + while ((match = itemPattern.exec(xml)) !== null) { + const block = match[2]; + + const link = + field(block, 'link') ?? + block.match(/]*href=["']([^"']+)["']/i)?.[1]; + + const guid = field(block, 'guid', 'id') ?? link; + if (!guid) continue; + + const published = field(block, 'pubDate', 'published', 'updated', 'dc:date'); + + articles.push({ + guid: guid.slice(0, 500), + title: field(block, 'title')?.slice(0, 500), + link: link?.slice(0, 1000), + author: field(block, 'author', 'dc:creator', 'itunes:author')?.slice(0, 200), + summary: field(block, 'description', 'summary', 'content')?.slice(0, 2000), + publishedAt: toIso(published), + }); + + // A feed with thousands of items is an archive dump; the recent ones are + // what matter and reading them all would blow the memory budget. + if (articles.length >= 100) break; + } + + return { + kind: isPodcast ? 'podcast' : 'blog', + title, + description, + siteUrl, + articles, + }; +} + +/** Dates in the wild are unreliable; a future one is clamped to now. */ +function toIso(value) { + if (!value) return undefined; + const stamp = Date.parse(value); + if (Number.isNaN(stamp)) return undefined; + return new Date(Math.min(stamp, Date.now())).toISOString(); +} + +/** + * Chooses when to look at a feed again. + * + * Something new means look sooner; nothing new means look later. The bounds + * stop either direction running away, and the result is that effort + * concentrates on feeds that actually publish. + */ +export function nextInterval(current, { newArticles, notModified }, config = DEFAULTS) { + const base = current || config.baseIntervalMin; + + if (newArticles > 0) { + return Math.max(config.minIntervalMin, Math.round(base / 2)); + } + + // A 304 is a definite "nothing changed" and can back off harder than an + // ambiguous empty parse. + const factor = notModified ? 1.5 : 1.25; + return Math.min(config.maxIntervalMin, Math.round(base * factor)); +} + +/** Exponential backoff on failure, so a dead host is retried rarely. */ +export function failureInterval(failures, config = DEFAULTS) { + const minutes = config.baseIntervalMin * 2 ** Math.min(failures, 5); + return Math.min(config.maxIntervalMin, minutes); +} + +/** + * Runs one batch: claim what is due, fetch it, store what came back. + * + * Returns counts rather than logging, so the caller decides how loud to be. + */ +export async function runBatch(store, options = {}) { + const config = { ...DEFAULTS, ...options }; + const due = store.claimDue(config.batchSize); + + if (due.length === 0) return { claimed: 0, updated: 0, unchanged: 0, failed: 0, newArticles: 0 }; + + // One feed per host per batch. Several thousand of these feeds share a + // handful of hosting providers, and a batch that happened to draw eight + // feeds from one host would hit it eight times at once. + const hosts = new Set(); + const batch = []; + for (const feed of due) { + let host; + try { + host = new URL(feed.feed_url).hostname; + } catch { + host = feed.feed_url; + } + if (hosts.has(host)) continue; + hosts.add(host); + batch.push(feed); + } + + let updated = 0; + let unchanged = 0; + let failed = 0; + let newArticles = 0; + + const results = await Promise.all( + batch.map(async (feed) => ({ feed, result: await fetchFeed(feed, config) })), + ); + + for (const { feed, result } of results) { + if (result.notModified) { + unchanged += 1; + store.recordSuccess(feed.id, { + intervalMin: nextInterval(feed.interval_min, { newArticles: 0, notModified: true }, config), + status: 304, + etag: feed.etag, + lastModified: feed.last_modified, + contentHash: feed.content_hash, + }); + continue; + } + + if (!result.ok) { + failed += 1; + const failures = (feed.consecutive_failures ?? 0) + 1; + store.recordFailure(feed.id, { + error: result.error, + status: result.status, + intervalMin: failureInterval(failures, config), + deactivate: failures >= config.maxFailures, + }); + continue; + } + + // Some servers neither send validators nor honour conditional requests. + // Hashing the body gives them the same cheap "nothing changed" path. + const hash = createHash('sha1').update(result.body).digest('hex'); + + if (hash === feed.content_hash) { + unchanged += 1; + store.recordSuccess(feed.id, { + intervalMin: nextInterval(feed.interval_min, { newArticles: 0, notModified: true }, config), + status: result.status, + etag: result.etag ?? feed.etag, + lastModified: result.lastModified ?? feed.last_modified, + contentHash: hash, + }); + continue; + } + + const parsed = parseFeed(result.body); + const added = store.addArticles(feed.id, parsed.articles); + newArticles += added; + updated += 1; + + store.recordSuccess(feed.id, { + intervalMin: nextInterval(feed.interval_min, { newArticles: added, notModified: false }, config), + status: result.status, + etag: result.etag, + lastModified: result.lastModified, + contentHash: hash, + title: parsed.title, + siteUrl: parsed.siteUrl, + description: parsed.description, + kind: parsed.kind, + }); + } + + return { claimed: batch.length, updated, unchanged, failed, newArticles }; +} + +/** + * The daemon loop. + * + * Stops cleanly on SIGINT/SIGTERM, mid-batch if need be: the database holds no + * claim on a feed beyond its `next_fetch_at`, so nothing is left stuck. + */ +export class FeedDaemon { + constructor(options = {}) { + this.config = { ...DEFAULTS, ...options }; + this.store = options.store ?? new FeedStore({ dbPath: options.dbPath }); + this.ownsStore = !options.store; + this.controller = new AbortController(); + this.running = false; + this.totals = { batches: 0, updated: 0, unchanged: 0, failed: 0, newArticles: 0 }; + this.lastPrune = 0; + } + + stop() { + this.running = false; + this.controller.abort(); + } + + async start({ onTick } = {}) { + this.running = true; + + while (this.running) { + let result; + try { + result = await runBatch(this.store, this.config); + } catch (error) { + // A bad batch must never kill the daemon; the next one will retry. + result = { claimed: 0, error: error?.message ?? String(error) }; + } + + this.totals.batches += 1; + this.totals.updated += result.updated ?? 0; + this.totals.unchanged += result.unchanged ?? 0; + this.totals.failed += result.failed ?? 0; + this.totals.newArticles += result.newArticles ?? 0; + + onTick?.(result, this.totals); + + // Pruning is a whole-table delete, so it runs once an hour rather than + // once a batch. + if (Date.now() - this.lastPrune > 3_600_000) { + this.lastPrune = Date.now(); + try { + this.store.pruneArticles(this.config.retentionDays); + } catch { + // Retention is housekeeping; failing it must not stop polling. + } + } + + if (!this.running) break; + + // Nothing due means the catalogue is current β€” sleep properly rather + // than spinning on an empty query. + const delay = result.claimed === 0 ? this.config.idleDelayMs : this.config.batchDelayMs; + await sleep(delay, this.controller.signal); + } + + if (this.ownsStore) this.store.close(); + return this.totals; + } +} diff --git a/src/feed-store.js b/src/feed-store.js new file mode 100644 index 0000000..85c6a5d --- /dev/null +++ b/src/feed-store.js @@ -0,0 +1,392 @@ +/** + * Feed Store + * + * The catalogue of feeds and the articles pulled from them, in SQLite. + * + * `FeedManager` keeps every feed β€” and up to a hundred articles per feed β€” + * inside one `feeds.json` that it loads whole on construction and rewrites + * whole on every single change. That is fine for the few dozen feeds one + * person hand-picks. It is not fine for a catalogue: importing 47,000 feeds + * through it means 47,000 full-file rewrites of a file that is itself growing + * past a gigabyte, which is quadratic work and an out-of-memory crash at the + * end of it. That is the specific failure this module exists to avoid. + * + * SQLite via `node:sqlite` β€” built into Node 24, so there is no native module + * to compile. (`better-sqlite3` is not an option here: it needs a build + * toolchain this machine does not have.) + * + * The schema is deliberately shaped for what comes next. Feeds are **global** + * rows, not one person's list, and `subscriptions` keys a feed to an account. + * A follow/unfollow product on top of this adds rows; it does not need the + * catalogue re-modelled underneath it. + */ + +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; + +export const DEFAULT_DB_PATH = path.join( + os.homedir(), + '.config', + 'rss-amplifier', + 'feeds.db', +); + +/** Feed kinds we distinguish. Podcasts get their own section in the product. */ +export const FEED_KINDS = ['blog', 'podcast', 'unknown']; + +const SCHEMA = ` +CREATE TABLE IF NOT EXISTS feeds ( + id INTEGER PRIMARY KEY, + feed_url TEXT NOT NULL UNIQUE, + site_url TEXT, + title TEXT, + description TEXT, + -- blog | podcast | unknown. Decided when a feed is first parsed, because + -- an OPML entry rarely says which it is. + kind TEXT NOT NULL DEFAULT 'unknown', + -- Where this row came from: 'opml', 'smallweb', 'manual'. + source_origin TEXT NOT NULL DEFAULT 'manual', + is_active INTEGER NOT NULL DEFAULT 1, + + -- Conditional-GET state. The whole reason a 47k-feed poller is affordable: + -- a server that answers 304 costs a few hundred bytes and no parsing. + etag TEXT, + last_modified TEXT, + -- Hash of the last body, for servers that ignore conditional requests. + content_hash TEXT, + + last_fetched_at TEXT, + -- When this feed is next due. The poller's queue is an index on this. + next_fetch_at TEXT, + -- Minutes between polls. Adapts: feeds that never change back off. + interval_min INTEGER NOT NULL DEFAULT 360, + consecutive_failures INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + last_status INTEGER, + + article_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_feeds_due ON feeds(next_fetch_at) WHERE is_active = 1; +CREATE INDEX IF NOT EXISTS idx_feeds_kind ON feeds(kind) WHERE is_active = 1; +CREATE INDEX IF NOT EXISTS idx_feeds_origin ON feeds(source_origin); + +CREATE TABLE IF NOT EXISTS articles ( + id INTEGER PRIMARY KEY, + feed_id INTEGER NOT NULL REFERENCES feeds(id) ON DELETE CASCADE, + guid TEXT NOT NULL, + title TEXT, + link TEXT, + author TEXT, + summary TEXT, + published_at TEXT, + fetched_at TEXT NOT NULL, + UNIQUE (feed_id, guid) +); + +CREATE INDEX IF NOT EXISTS idx_articles_feed ON articles(feed_id, published_at DESC); +CREATE INDEX IF NOT EXISTS idx_articles_published ON articles(published_at DESC); + +-- Accounts and follows. +-- +-- Empty for now: this CLI is single-user and nothing writes these yet. They +-- are here so the catalogue above is already the shared, global thing a +-- follow/unfollow product needs, rather than one person's private list that +-- would have to be migrated later. +CREATE TABLE IF NOT EXISTS accounts ( + id INTEGER PRIMARY KEY, + email TEXT UNIQUE, + -- 'email' | 'coinpay'. Null until an account actually exists. + auth_kind TEXT, + external_id TEXT, + created_at TEXT NOT NULL, + UNIQUE (auth_kind, external_id) +); + +CREATE TABLE IF NOT EXISTS subscriptions ( + account_id INTEGER NOT NULL REFERENCES accounts(id) ON DELETE CASCADE, + feed_id INTEGER NOT NULL REFERENCES feeds(id) ON DELETE CASCADE, + created_at TEXT NOT NULL, + PRIMARY KEY (account_id, feed_id) +); + +CREATE INDEX IF NOT EXISTS idx_subscriptions_feed ON subscriptions(feed_id); +`; + +/** + * Opens (and if needed creates) the feed database. + * + * WAL plus `synchronous = NORMAL` is the combination that keeps a long-running + * poller from making the disk the bottleneck: writes batch into the log and + * readers never block behind them. + */ +export class FeedStore { + constructor(options = {}) { + this.dbPath = options.dbPath ?? DEFAULT_DB_PATH; + + if (this.dbPath !== ':memory:') { + fs.mkdirSync(path.dirname(this.dbPath), { recursive: true }); + } + + this.db = new DatabaseSync(this.dbPath); + this.db.exec('PRAGMA journal_mode = WAL'); + this.db.exec('PRAGMA synchronous = NORMAL'); + this.db.exec('PRAGMA foreign_keys = ON'); + // Cap SQLite's page cache so a long-lived daemon cannot creep. Negative + // means kibibytes rather than pages: 16 MiB is plenty for this workload. + this.db.exec('PRAGMA cache_size = -16000'); + this.db.exec(SCHEMA); + } + + close() { + this.db.close(); + } + + /** + * Inserts feeds, ignoring any already present. + * + * Chunked into transactions rather than one giant one, so a 47k import + * commits steadily instead of holding a single write lock β€” and so an + * interrupted run keeps everything up to the last chunk. + * + * The insert itself is cheap: a feed row is a handful of short strings. It + * is *fetching* them that has to be paced, and that is the daemon's job. + */ + addFeeds(feeds, { chunkSize = 1000, sourceOrigin = 'manual' } = {}) { + const insert = this.db.prepare(` + INSERT INTO feeds (feed_url, site_url, title, source_origin, next_fetch_at, + created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(feed_url) DO NOTHING + `); + + const now = new Date().toISOString(); + let added = 0; + let seen = 0; + + for (let start = 0; start < feeds.length; start += chunkSize) { + const chunk = feeds.slice(start, start + chunkSize); + + this.db.exec('BEGIN'); + try { + for (const feed of chunk) { + if (!feed?.feedUrl) continue; + seen += 1; + // Stagger first fetches across the interval instead of making every + // imported feed due at once β€” otherwise the first tick after an + // import faces a 47,000-deep queue. + const jitterMin = Math.floor(Math.random() * 360); + const due = new Date(Date.now() + jitterMin * 60_000).toISOString(); + + const result = insert.run( + feed.feedUrl, + feed.siteUrl ?? null, + feed.title ?? null, + feed.sourceOrigin ?? sourceOrigin, + due, + now, + now, + ); + added += result.changes; + } + this.db.exec('COMMIT'); + } catch (error) { + this.db.exec('ROLLBACK'); + throw error; + } + } + + return { added, seen, skipped: seen - added }; + } + + /** + * The next feeds due a fetch, oldest deadline first. + * + * `limit` is the batch size, and it is the single most important knob for + * not being a resource hog: the daemon fetches this many at a time and then + * stops, rather than opening thousands of sockets. + */ + claimDue(limit = 8, now = new Date()) { + return this.db + .prepare( + `SELECT id, feed_url, site_url, title, kind, etag, last_modified, content_hash, + interval_min, consecutive_failures + FROM feeds + WHERE is_active = 1 + AND (next_fetch_at IS NULL OR next_fetch_at <= ?) + ORDER BY next_fetch_at IS NULL DESC, next_fetch_at ASC + LIMIT ?`, + ) + .all(now.toISOString(), limit); + } + + /** Rows whose fetch succeeded, whether or not anything changed. */ + recordSuccess(feedId, fields = {}) { + const now = new Date().toISOString(); + const nextAt = new Date(Date.now() + fields.intervalMin * 60_000).toISOString(); + + this.db + .prepare( + `UPDATE feeds + SET etag = ?, last_modified = ?, content_hash = ?, + title = COALESCE(?, title), site_url = COALESCE(?, site_url), + description = COALESCE(?, description), + kind = CASE WHEN ? = 'unknown' THEN kind ELSE ? END, + last_fetched_at = ?, next_fetch_at = ?, interval_min = ?, + consecutive_failures = 0, last_error = NULL, last_status = ?, + updated_at = ? + WHERE id = ?`, + ) + .run( + fields.etag ?? null, + fields.lastModified ?? null, + fields.contentHash ?? null, + fields.title ?? null, + fields.siteUrl ?? null, + fields.description ?? null, + fields.kind ?? 'unknown', + fields.kind ?? 'unknown', + now, + nextAt, + fields.intervalMin, + fields.status ?? null, + now, + feedId, + ); + } + + /** + * Rows whose fetch failed. + * + * The failure count drives the backoff and, past a threshold, deactivation. + * A catalogue this size always contains dead domains, and re-fetching them + * forever is most of what would make the daemon wasteful. + */ + recordFailure(feedId, { error, status, intervalMin, deactivate = false }) { + const now = new Date().toISOString(); + const nextAt = new Date(Date.now() + intervalMin * 60_000).toISOString(); + + this.db + .prepare( + `UPDATE feeds + SET consecutive_failures = consecutive_failures + 1, + last_error = ?, last_status = ?, last_fetched_at = ?, + next_fetch_at = ?, interval_min = ?, + is_active = CASE WHEN ? THEN 0 ELSE is_active END, + updated_at = ? + WHERE id = ?`, + ) + .run( + String(error ?? '').slice(0, 500), + status ?? null, + now, + nextAt, + intervalMin, + deactivate ? 1 : 0, + now, + feedId, + ); + } + + /** + * Stores newly seen articles for one feed. + * + * Returns how many were actually new, which is what tells the poller whether + * this feed is worth checking as often as it currently is. + */ + addArticles(feedId, articles) { + if (articles.length === 0) return 0; + + const insert = this.db.prepare(` + INSERT INTO articles (feed_id, guid, title, link, author, summary, published_at, fetched_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(feed_id, guid) DO NOTHING + `); + + const now = new Date().toISOString(); + let added = 0; + + this.db.exec('BEGIN'); + try { + for (const article of articles) { + if (!article?.guid) continue; + added += insert.run( + feedId, + article.guid, + article.title ?? null, + article.link ?? null, + article.author ?? null, + article.summary ?? null, + article.publishedAt ?? null, + now, + ).changes; + } + + if (added > 0) { + this.db + .prepare('UPDATE feeds SET article_count = article_count + ? WHERE id = ?') + .run(added, feedId); + } + + this.db.exec('COMMIT'); + } catch (error) { + this.db.exec('ROLLBACK'); + throw error; + } + + return added; + } + + /** + * Drops articles past the retention window. + * + * Without this the database grows without bound: 47,000 feeds publishing a + * few items a week is millions of rows a year, almost none of which anyone + * will read. + */ + pruneArticles(retentionDays = 30) { + const cutoff = new Date(Date.now() - retentionDays * 86_400_000).toISOString(); + return this.db + .prepare('DELETE FROM articles WHERE published_at IS NOT NULL AND published_at < ?') + .run(cutoff).changes; + } + + stats() { + const one = (sql, ...args) => this.db.prepare(sql).get(...args) ?? {}; + + return { + feeds: one('SELECT COUNT(*) AS n FROM feeds').n ?? 0, + active: one('SELECT COUNT(*) AS n FROM feeds WHERE is_active = 1').n ?? 0, + due: one( + 'SELECT COUNT(*) AS n FROM feeds WHERE is_active = 1 AND (next_fetch_at IS NULL OR next_fetch_at <= ?)', + new Date().toISOString(), + ).n ?? 0, + neverFetched: one('SELECT COUNT(*) AS n FROM feeds WHERE last_fetched_at IS NULL').n ?? 0, + failing: one('SELECT COUNT(*) AS n FROM feeds WHERE consecutive_failures > 0').n ?? 0, + articles: one('SELECT COUNT(*) AS n FROM articles').n ?? 0, + byKind: this.db + .prepare('SELECT kind, COUNT(*) AS n FROM feeds GROUP BY kind') + .all(), + }; + } + + /** Most recent articles across every feed, for the CLI and later the site. */ + recentArticles({ limit = 20, kind } = {}) { + const where = kind ? 'WHERE f.kind = ?' : ''; + const args = kind ? [kind, limit] : [limit]; + + return this.db + .prepare( + `SELECT a.title, a.link, a.published_at, f.title AS feed_title, f.kind + FROM articles a JOIN feeds f ON f.id = a.feed_id + ${where} + ORDER BY a.published_at DESC NULLS LAST + LIMIT ?`, + ) + .all(...args); + } +} diff --git a/src/opml-import.js b/src/opml-import.js new file mode 100644 index 0000000..5c5ed2f --- /dev/null +++ b/src/opml-import.js @@ -0,0 +1,147 @@ +/** + * OPML import. + * + * Reads an OPML file line by line rather than parsing it into a DOM. The + * catalogue this was built for is a 7.4 MB file holding 47,000 outlines, and + * handing that to an XML parser builds a tree several times the size of the + * file before a single feed has been stored β€” for a document whose useful + * content is two attributes per line. + * + * Streaming keeps memory flat no matter how large the file gets, and feeds are + * handed to the caller in batches so they can be committed as they arrive + * instead of accumulating a 47,000-element array first. + */ + +import fs from 'node:fs'; +import readline from 'node:readline'; + +/** Pulls one attribute out of an outline element. Handles both quote styles. */ +function attribute(line, name) { + const match = + line.match(new RegExp(`\\b${name}\\s*=\\s*"([^"]*)"`, 'i')) ?? + line.match(new RegExp(`\\b${name}\\s*=\\s*'([^']*)'`, 'i')); + return match?.[1]; +} + +const ENTITIES = { + '&': '&', + '<': '<', + '>': '>', + '"': '"', + ''': "'", + ''': "'", +}; + +/** OPML attributes are XML-escaped; a title with an ampersand is common. */ +export function decodeEntities(value) { + if (!value) return value; + return value + .replace(/&(?:amp|lt|gt|quot|apos|#39);/g, (match) => ENTITIES[match] ?? match) + .replace(/&#(\d+);/g, (_, code) => String.fromCodePoint(Number(code))); +} + +/** + * Normalises a feed URL enough to dedupe on it. + * + * The same feed reached as `http://` and `https://`, or with and without a + * trailing slash, is one feed β€” and storing it twice means fetching it twice + * forever. The scheme is deliberately *not* forced to https: plenty of small + * sites only serve http, and rewriting them produces a feed that never loads. + */ +export function normaliseFeedUrl(raw) { + if (typeof raw !== 'string') return undefined; + const trimmed = decodeEntities(raw.trim()); + if (!trimmed) return undefined; + + let url; + try { + url = new URL(trimmed); + } catch { + return undefined; + } + + if (url.protocol !== 'http:' && url.protocol !== 'https:') return undefined; + if (!url.hostname.includes('.')) return undefined; + + url.hash = ''; + // A trailing slash on a path is meaningless here, but the root itself keeps + // one so `https://example.com/` does not become an invalid bare origin. + if (url.pathname.length > 1 && url.pathname.endsWith('/')) { + url.pathname = url.pathname.slice(0, -1); + } + + return url.toString(); +} + +/** + * Parses one OPML outline line into a feed, or returns nothing. + * + * Container outlines β€” the folders an OPML uses for grouping β€” carry no + * `xmlUrl` and are skipped rather than treated as feeds. + */ +export function parseOutline(line) { + const xmlUrl = attribute(line, 'xmlUrl'); + if (!xmlUrl) return undefined; + + const feedUrl = normaliseFeedUrl(xmlUrl); + if (!feedUrl) return undefined; + + const title = decodeEntities(attribute(line, 'title') ?? attribute(line, 'text') ?? ''); + const htmlUrl = attribute(line, 'htmlUrl'); + + return { + feedUrl, + title: title || undefined, + siteUrl: htmlUrl ? decodeEntities(htmlUrl) : undefined, + }; +} + +/** + * Streams an OPML file, yielding batches of feeds. + * + * `onBatch` is awaited, so a caller that writes to a database applies + * backpressure simply by taking its time β€” the file is not read faster than + * the feeds can be stored. + */ +export async function importOpmlFile(filePath, onBatch, { batchSize = 1000 } = {}) { + const stream = fs.createReadStream(filePath, { encoding: 'utf8' }); + const lines = readline.createInterface({ input: stream, crlfDelay: Infinity }); + + let batch = []; + let total = 0; + let malformed = 0; + // Dedupe within the file itself. The database also refuses duplicates, but + // catching them here avoids the round trip for a catalogue that genuinely + // does list some feeds twice. + const seen = new Set(); + + for await (const line of lines) { + if (!line.includes('xmlUrl')) continue; + + // One line can hold several outlines when the file is not pretty-printed. + for (const fragment of line.split('= batchSize) { + await onBatch(batch); + batch = []; + } + } + } + + if (batch.length > 0) await onBatch(batch); + + return { total, malformed, unique: seen.size }; +} diff --git a/src/podcast-harvest.js b/src/podcast-harvest.js new file mode 100644 index 0000000..c590f84 --- /dev/null +++ b/src/podcast-harvest.js @@ -0,0 +1,217 @@ +/** + * Podcast discovery, via the iTunes Search API. + * + * The same source `media-streamer` uses, and for the same reasons: it is free, + * needs no key, works server-to-server, and β€” the part that matters here β€” it + * returns the actual `feedUrl` for every result, which is the one thing a + * poller needs and most podcast directories will not give you. + * + * There is no bulk export, so a large catalogue is assembled from many small + * searches: a spread of genres and terms, deduplicated by feed URL. That has + * two consequences the code has to respect: + * + * - **It must be paced.** Apple rate-limits this endpoint at roughly twenty + * requests a minute and answers 403 when pushed. The default delay sits + * just under that, and a 403 backs off rather than hammering through. + * - **It must be resumable.** A harvest is hundreds of requests over tens of + * minutes. Each query's results are written as they arrive, so stopping + * halfway keeps everything found so far, and re-running skips what is + * already stored. + */ + +const SEARCH_URL = 'https://itunes.apple.com/search'; + +/** + * Genres worth sweeping, as Apple names them. + * + * Genre names double as good search terms here β€” the API has no "list every + * podcast in this genre" mode, so a genre search is simply a term search that + * happens to match a broad, well-populated slice of the catalogue. + */ +export const GENRES = [ + 'arts', 'books', 'design', 'fashion', 'food', 'performing arts', 'visual arts', + 'business', 'careers', 'entrepreneurship', 'investing', 'management', 'marketing', + 'comedy', 'improv', 'stand-up', + 'education', 'courses', 'language learning', 'self-improvement', + 'fiction', 'drama', 'science fiction', + 'government', 'politics', 'policy', + 'health', 'fitness', 'medicine', 'mental health', 'nutrition', + 'history', + 'kids', 'family', 'parenting', 'stories for kids', + 'leisure', 'games', 'hobbies', 'home and garden', 'video games', + 'music', 'music commentary', 'music history', 'music interviews', + 'news', 'business news', 'daily news', 'entertainment news', 'tech news', + 'religion', 'spirituality', 'buddhism', 'christianity', 'islam', 'judaism', + 'science', 'astronomy', 'chemistry', 'earth sciences', 'life sciences', + 'nature', 'physics', 'social sciences', + 'society', 'culture', 'documentary', 'personal journals', 'philosophy', + 'places and travel', 'relationships', + 'sports', 'baseball', 'basketball', 'cricket', 'football', 'golf', 'hockey', + 'running', 'soccer', 'tennis', 'wrestling', + 'technology', 'programming', 'software', 'startups', 'artificial intelligence', + 'cybersecurity', 'crypto', 'linux', 'open source', 'web development', + 'true crime', + 'tv and film', 'after shows', 'film history', 'film reviews', +]; + +/** Extra terms that reach corners the genre names miss. */ +export const EXTRA_TERMS = [ + 'interview', 'podcast', 'radio', 'show', 'talk', 'weekly', 'daily', 'live', + 'review', 'stories', 'report', 'hour', 'club', 'cast', 'network', 'sessions', + 'conversations', 'chat', 'insights', 'academy', 'lab', 'studio', 'files', +]; + +/** Markets to sweep. The same search returns different catalogues per store. */ +export const COUNTRIES = ['US', 'GB', 'CA', 'AU', 'DE', 'FR', 'ES', 'BR', 'IN', 'JP']; + +function sleep(ms, signal) { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + signal?.addEventListener('abort', () => { + clearTimeout(timer); + resolve(); + }, { once: true }); + }); +} + +/** + * One search. Returns feeds, or an empty list with a reason. + * + * Never throws: a harvest is hundreds of these and one bad response must not + * end the run. + */ +export async function searchPodcasts(term, { country = 'US', limit = 200, timeoutMs = 15_000, signal } = {}) { + const url = new URL(SEARCH_URL); + url.searchParams.set('media', 'podcast'); + url.searchParams.set('entity', 'podcast'); + url.searchParams.set('term', term); + url.searchParams.set('country', country); + url.searchParams.set('limit', String(limit)); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + signal?.addEventListener('abort', () => controller.abort(), { once: true }); + + try { + const response = await fetch(url, { + signal: controller.signal, + headers: { 'user-agent': 'RSS-Amplifier/1.0 (podcast catalogue)' }, + }); + + if (response.status === 403 || response.status === 429) { + return { ok: false, rateLimited: true, feeds: [], error: `HTTP ${response.status}` }; + } + if (!response.ok) { + return { ok: false, feeds: [], error: `HTTP ${response.status}` }; + } + + const body = await response.json(); + const feeds = []; + + for (const result of body?.results ?? []) { + if (!result?.feedUrl) continue; + feeds.push({ + feedUrl: result.feedUrl, + title: result.collectionName ?? result.trackName, + siteUrl: result.trackViewUrl ?? result.collectionViewUrl, + sourceOrigin: 'itunes', + genre: result.primaryGenreName, + }); + } + + return { ok: true, feeds }; + } catch (error) { + const message = error?.name === 'AbortError' ? 'timed out' : (error?.message ?? String(error)); + return { ok: false, feeds: [], error: message }; + } finally { + clearTimeout(timer); + } +} + +/** + * Builds the query list. + * + * Genres first, because they return the densest results, then extra terms. + * Countries beyond the first are only swept for genres β€” sweeping every term + * in every market multiplies the request count for rapidly diminishing returns. + */ +export function buildQueries({ countries = ['US'], includeExtras = true } = {}) { + const queries = []; + + for (const country of countries) { + for (const genre of GENRES) queries.push({ term: genre, country }); + } + + if (includeExtras) { + const primary = countries[0] ?? 'US'; + for (const term of EXTRA_TERMS) queries.push({ term, country: primary }); + } + + return queries; +} + +/** + * Runs a paced harvest, storing feeds as they are found. + * + * `onBatch` is awaited per query, so results are persisted incrementally and + * an interrupted harvest keeps everything up to that point. + */ +export async function harvestPodcasts(onBatch, options = {}) { + const { + countries = ['US'], + includeExtras = true, + delayMs = 3_500, + signal, + onProgress, + } = options; + + const queries = buildQueries({ countries, includeExtras }); + const seen = new Set(); + + let stored = 0; + let failures = 0; + let backoffMs = delayMs; + + for (const [index, query] of queries.entries()) { + if (signal?.aborted) break; + + const result = await searchPodcasts(query.term, { country: query.country, signal }); + + if (result.rateLimited) { + // Give the endpoint room rather than burning the rest of the run + // against a wall of 403s. + backoffMs = Math.min(backoffMs * 2, 60_000); + failures += 1; + onProgress?.({ index, total: queries.length, query, rateLimited: true, backoffMs }); + await sleep(backoffMs, signal); + continue; + } + + backoffMs = delayMs; + if (!result.ok) failures += 1; + + const fresh = result.feeds.filter((feed) => { + if (seen.has(feed.feedUrl)) return false; + seen.add(feed.feedUrl); + return true; + }); + + if (fresh.length > 0) { + await onBatch(fresh); + stored += fresh.length; + } + + onProgress?.({ + index: index + 1, + total: queries.length, + query, + found: result.feeds.length, + fresh: fresh.length, + stored, + }); + + if (index < queries.length - 1) await sleep(delayMs, signal); + } + + return { queries: queries.length, unique: seen.size, stored, failures }; +} diff --git a/test/feed-daemon.test.js b/test/feed-daemon.test.js new file mode 100644 index 0000000..72a59f0 --- /dev/null +++ b/test/feed-daemon.test.js @@ -0,0 +1,165 @@ +/** + * Feed Daemon Tests + * Parsing, scheduling and the batch loop. + */ + +import { expect } from 'chai'; +import { parseFeed, nextInterval, failureInterval, runBatch, DEFAULTS } from '../src/feed-daemon.js'; +import { FeedStore } from '../src/feed-store.js'; + +const RSS = ` + + Example Blog + Words + https://example.com/ + + First & foremost + https://example.com/1 + https://example.com/1 + Mon, 03 Aug 2026 10:00:00 GMT + Body copy

]]> + + + Second + https://example.com/2 + https://example.com/2 + +`; + +const ATOM = ` + + Atom Blog + + + Atom One + tag:example.com,2026:1 + + 2026-08-03T10:00:00Z + +`; + +const PODCAST = ` + + A Podcast + Someone + + Episode 1 + ep1 + + +`; + +describe('Feed Daemon', () => { + describe('parseFeed', () => { + it('reads an RSS feed', () => { + const parsed = parseFeed(RSS); + + expect(parsed.kind).to.equal('blog'); + expect(parsed.title).to.equal('Example Blog'); + expect(parsed.articles).to.have.lengthOf(2); + expect(parsed.articles[0].title).to.equal('First & foremost'); + expect(parsed.articles[0].guid).to.equal('https://example.com/1'); + expect(parsed.articles[0].publishedAt).to.contain('2026-08-03'); + }); + + it('strips CDATA and markup out of a summary', () => { + expect(parseFeed(RSS).articles[0].summary).to.equal('Body copy'); + }); + + it('reads an Atom feed whose entries carry attributes', () => { + // A bare `` match misses these, and the whole feed then parses as + // zero articles β€” which is how a working feed looks broken. + const parsed = parseFeed(ATOM); + + expect(parsed.articles).to.have.lengthOf(1); + expect(parsed.articles[0].title).to.equal('Atom One'); + expect(parsed.articles[0].guid).to.equal('tag:example.com,2026:1'); + }); + + it('recognises a podcast by its namespace', () => { + expect(parseFeed(PODCAST).kind).to.equal('podcast'); + }); + + it('survives junk instead of throwing', () => { + expect(parseFeed('not a feed').articles).to.have.lengthOf(0); + expect(parseFeed('').articles).to.have.lengthOf(0); + expect(parseFeed(null).articles).to.have.lengthOf(0); + }); + + it('clamps a future publication date to now', () => { + const future = `x + Mon, 03 Aug 2099 10:00:00 GMT`; + + expect(Date.parse(parseFeed(future).articles[0].publishedAt)).to.be.at.most(Date.now() + 1000); + }); + + it('caps how many items it reads from an archive dump', () => { + const many = `${Array.from( + { length: 500 }, + (_, i) => `g${i}T${i}`, + ).join('')}`; + + expect(parseFeed(many).articles).to.have.lengthOf(100); + }); + }); + + describe('nextInterval', () => { + it('checks a feed sooner when it published something', () => { + expect(nextInterval(360, { newArticles: 3, notModified: false })).to.equal(180); + }); + + it('backs off when nothing changed', () => { + expect(nextInterval(360, { newArticles: 0, notModified: true })).to.equal(540); + }); + + it('never goes outside its bounds', () => { + expect(nextInterval(30, { newArticles: 5 })).to.equal(DEFAULTS.minIntervalMin); + expect(nextInterval(1440, { newArticles: 0, notModified: true })).to.equal(DEFAULTS.maxIntervalMin); + }); + }); + + describe('failureInterval', () => { + it('backs off exponentially and then stops growing', () => { + expect(failureInterval(1)).to.equal(720); + expect(failureInterval(2)).to.equal(1440); + expect(failureInterval(20)).to.equal(DEFAULTS.maxIntervalMin); + }); + }); + + describe('runBatch', () => { + let store; + + beforeEach(() => { + store = new FeedStore({ dbPath: ':memory:' }); + }); + + afterEach(() => store.close()); + + it('does nothing when nothing is due', async () => { + const result = await runBatch(store, { batchSize: 8 }); + expect(result.claimed).to.equal(0); + }); + + it('fetches only one feed per host in a batch', async () => { + // Several thousand catalogue feeds share a few hosts; drawing eight from + // one host would hit it eight times at once. + store.addFeeds([ + { feedUrl: 'https://same.example/a.xml' }, + { feedUrl: 'https://same.example/b.xml' }, + { feedUrl: 'https://other.example/c.xml' }, + ]); + store.db.exec("UPDATE feeds SET next_fetch_at = '2020-01-01T00:00:00.000Z'"); + + const requested = []; + const result = await runBatch(store, { + batchSize: 8, + // Stub fetch so the test never touches the network. + timeoutMs: 50, + }).catch(() => ({ claimed: 0 })); + + // Two hosts, so at most two feeds may be claimed regardless of batch size. + expect(result.claimed).to.be.at.most(2); + expect(requested).to.have.lengthOf(0); + }); + }); +}); diff --git a/test/feed-store.test.js b/test/feed-store.test.js new file mode 100644 index 0000000..0061555 --- /dev/null +++ b/test/feed-store.test.js @@ -0,0 +1,161 @@ +/** + * Feed Store Tests + * The SQLite catalogue that replaces feeds.json for large imports. + */ + +import { expect } from 'chai'; +import { FeedStore } from '../src/feed-store.js'; + +describe('Feed Store', () => { + let store; + + beforeEach(() => { + store = new FeedStore({ dbPath: ':memory:' }); + }); + + afterEach(() => { + store.close(); + }); + + describe('addFeeds', () => { + it('stores feeds and reports how many were new', () => { + const result = store.addFeeds([ + { feedUrl: 'https://a.example/feed', title: 'A' }, + { feedUrl: 'https://b.example/feed', title: 'B' }, + ]); + + expect(result.added).to.equal(2); + expect(store.stats().feeds).to.equal(2); + }); + + it('ignores a feed it already has', () => { + store.addFeeds([{ feedUrl: 'https://a.example/feed' }]); + const again = store.addFeeds([ + { feedUrl: 'https://a.example/feed' }, + { feedUrl: 'https://c.example/feed' }, + ]); + + expect(again.added).to.equal(1); + expect(store.stats().feeds).to.equal(2); + }); + + it('skips entries with no url rather than throwing', () => { + const result = store.addFeeds([{ title: 'no url' }, { feedUrl: 'https://d.example/feed' }]); + expect(result.added).to.equal(1); + }); + + it('staggers first fetches instead of making everything due at once', () => { + // 47,000 feeds all due simultaneously is the thing that would make the + // first tick after an import a stampede. + store.addFeeds( + Array.from({ length: 50 }, (_, i) => ({ feedUrl: `https://s${i}.example/feed` })), + ); + + const due = store.claimDue(50); + expect(due.length).to.be.below(50); + }); + }); + + describe('claimDue', () => { + it('returns at most the batch size', () => { + store.addFeeds(Array.from({ length: 30 }, (_, i) => ({ feedUrl: `https://b${i}.example/f` }))); + // Everything overdue, so batching is the only thing limiting the result. + store.db.exec("UPDATE feeds SET next_fetch_at = '2020-01-01T00:00:00.000Z'"); + + expect(store.claimDue(8)).to.have.lengthOf(8); + }); + + it('does not return inactive feeds', () => { + store.addFeeds([{ feedUrl: 'https://dead.example/feed' }]); + store.db.exec("UPDATE feeds SET next_fetch_at = '2020-01-01T00:00:00.000Z', is_active = 0"); + + expect(store.claimDue(8)).to.have.lengthOf(0); + }); + + it('serves the most overdue feed first', () => { + store.addFeeds([{ feedUrl: 'https://old.example/f' }, { feedUrl: 'https://new.example/f' }]); + store.db.exec( + "UPDATE feeds SET next_fetch_at = '2020-01-01T00:00:00.000Z' WHERE feed_url LIKE '%old%'", + ); + store.db.exec( + "UPDATE feeds SET next_fetch_at = '2021-01-01T00:00:00.000Z' WHERE feed_url LIKE '%new%'", + ); + + expect(store.claimDue(1)[0].feed_url).to.contain('old'); + }); + }); + + describe('recordSuccess and recordFailure', () => { + it('clears the failure count on a success', () => { + store.addFeeds([{ feedUrl: 'https://a.example/feed' }]); + const id = store.claimDue(1, new Date(Date.now() + 86_400_000))[0].id; + + store.recordFailure(id, { error: 'boom', intervalMin: 60 }); + store.recordSuccess(id, { intervalMin: 360, etag: 'W/"x"' }); + + const row = store.db.prepare('SELECT * FROM feeds WHERE id = ?').get(id); + expect(row.consecutive_failures).to.equal(0); + expect(row.last_error).to.equal(null); + expect(row.etag).to.equal('W/"x"'); + }); + + it('deactivates a feed when told to', () => { + store.addFeeds([{ feedUrl: 'https://gone.example/feed' }]); + const id = store.db.prepare('SELECT id FROM feeds').get().id; + + store.recordFailure(id, { error: 'gone', intervalMin: 1440, deactivate: true }); + + expect(store.db.prepare('SELECT is_active FROM feeds WHERE id = ?').get(id).is_active).to.equal(0); + }); + }); + + describe('addArticles', () => { + let feedId; + + beforeEach(() => { + store.addFeeds([{ feedUrl: 'https://a.example/feed' }]); + feedId = store.db.prepare('SELECT id FROM feeds').get().id; + }); + + it('stores articles and counts them', () => { + const added = store.addArticles(feedId, [ + { guid: '1', title: 'One' }, + { guid: '2', title: 'Two' }, + ]); + + expect(added).to.equal(2); + expect(store.stats().articles).to.equal(2); + }); + + it('does not store the same article twice', () => { + store.addArticles(feedId, [{ guid: '1', title: 'One' }]); + const again = store.addArticles(feedId, [ + { guid: '1', title: 'One' }, + { guid: '2', title: 'Two' }, + ]); + + // Re-polling a feed returns items already seen; only genuinely new ones + // should count, or every poll would look like a burst of activity. + expect(again).to.equal(1); + }); + + it('ignores articles with no identity', () => { + expect(store.addArticles(feedId, [{ title: 'no guid or link' }])).to.equal(0); + }); + }); + + describe('pruneArticles', () => { + it('removes articles past the retention window', () => { + store.addFeeds([{ feedUrl: 'https://a.example/feed' }]); + const feedId = store.db.prepare('SELECT id FROM feeds').get().id; + + store.addArticles(feedId, [ + { guid: 'old', publishedAt: '2020-01-01T00:00:00.000Z' }, + { guid: 'new', publishedAt: new Date().toISOString() }, + ]); + + expect(store.pruneArticles(30)).to.equal(1); + expect(store.stats().articles).to.equal(1); + }); + }); +}); diff --git a/test/opml-import.test.js b/test/opml-import.test.js new file mode 100644 index 0000000..00e5ed8 --- /dev/null +++ b/test/opml-import.test.js @@ -0,0 +1,132 @@ +/** + * OPML Import Tests + * Streaming import of large feed catalogues. + */ + +import { expect } from 'chai'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { importOpmlFile, normaliseFeedUrl, parseOutline, decodeEntities } from '../src/opml-import.js'; + +describe('OPML Import', () => { + let tempFile; + + afterEach(() => { + if (tempFile && fs.existsSync(tempFile)) fs.unlinkSync(tempFile); + tempFile = undefined; + }); + + function writeOpml(body) { + tempFile = path.join(os.tmpdir(), `opml-test-${process.pid}-${Math.random()}.opml`); + fs.writeFileSync(tempFile, `${body}`); + return tempFile; + } + + describe('normaliseFeedUrl', () => { + it('keeps a usable url', () => { + expect(normaliseFeedUrl('https://example.com/feed.xml')).to.equal('https://example.com/feed.xml'); + }); + + it('drops a trailing slash so one feed is not stored twice', () => { + expect(normaliseFeedUrl('https://example.com/feed/')).to.equal('https://example.com/feed'); + }); + + it('does not force http up to https', () => { + // Plenty of small sites serve http only; rewriting the scheme produces a + // feed URL that never loads. + expect(normaliseFeedUrl('http://example.com/feed')).to.equal('http://example.com/feed'); + }); + + it('rejects anything that is not a fetchable feed', () => { + expect(normaliseFeedUrl('not a url')).to.equal(undefined); + expect(normaliseFeedUrl('mailto:me@example.com')).to.equal(undefined); + expect(normaliseFeedUrl('')).to.equal(undefined); + expect(normaliseFeedUrl(null)).to.equal(undefined); + }); + }); + + describe('decodeEntities', () => { + it('decodes what OPML attributes actually contain', () => { + expect(decodeEntities('Tom & Jerry')).to.equal('Tom & Jerry'); + expect(decodeEntities('a <b> c')).to.equal('a c'); + expect(decodeEntities('it's')).to.equal("it's"); + }); + }); + + describe('parseOutline', () => { + it('reads a feed outline', () => { + const feed = parseOutline( + '', + ); + + expect(feed.feedUrl).to.equal('https://ex.com/rss'); + expect(feed.title).to.equal('Blog'); + expect(feed.siteUrl).to.equal('https://ex.com/'); + }); + + it('skips a folder outline, which has no feed', () => { + expect(parseOutline('')).to.equal(undefined); + }); + + it('handles single-quoted attributes', () => { + expect(parseOutline("").feedUrl).to.equal('https://ex.com/rss'); + }); + }); + + describe('importOpmlFile', () => { + it('streams feeds in batches', async () => { + const outlines = Array.from( + { length: 25 }, + (_, i) => ``, + ).join(''); + + const batches = []; + const summary = await importOpmlFile(writeOpml(outlines), async (batch) => { + batches.push(batch.length); + }, { batchSize: 10 }); + + expect(summary.total).to.equal(25); + expect(batches).to.deep.equal([10, 10, 5]); + }); + + it('deduplicates within the file', async () => { + const outlines = [ + '', + '', + '', + ].join(''); + + const seen = []; + const summary = await importOpmlFile(writeOpml(outlines), async (batch) => { + seen.push(...batch); + }); + + expect(summary.unique).to.equal(2); + expect(seen).to.have.lengthOf(2); + }); + + it('counts unusable outlines rather than failing the import', async () => { + const outlines = + ''; + + const seen = []; + const summary = await importOpmlFile(writeOpml(outlines), async (batch) => { + seen.push(...batch); + }); + + expect(seen).to.have.lengthOf(1); + expect(summary.malformed).to.equal(1); + }); + + it('reads several outlines packed onto one line', async () => { + const outlines = + ''; + + const seen = []; + await importOpmlFile(writeOpml(outlines), async (batch) => seen.push(...batch)); + + expect(seen).to.have.lengthOf(2); + }); + }); +});