diff --git a/bun.lock b/bun.lock index 2342190..6cb30b2 100644 --- a/bun.lock +++ b/bun.lock @@ -11,7 +11,7 @@ "@profullstack/openprofile": "^0.2.0", "@profullstack/player": "0.16.0", "@profullstack/profile": "0.1.6", - "@profullstack/scraper-daemon": "0.2.2", + "@profullstack/scraper-daemon": "0.3.0", "@profullstack/submit-feed": "^0.1.3", "@profullstack/throttle": "^0.4.1", "@profullstack/x402-gateway": "^0.16.1", @@ -204,7 +204,7 @@ "@profullstack/profile": ["@profullstack/profile@0.1.6", "", { "peerDependencies": { "@libsql/client": ">=0.6.0", "postgres": "^3.4.0", "react": ">=18" }, "optionalPeers": ["@libsql/client", "postgres", "react"], "bin": { "profile": "src/cli.js" } }, "sha512-xkn0P3ANz151QJdlBwcBXkHk813qW9qAmuTZKcuvpEZYkyHDC9J9KfvpVSRepfjECzz4ONGHWO5Zwc4BHpfK8g=="], - "@profullstack/scraper-daemon": ["@profullstack/scraper-daemon@0.2.2", "", { "peerDependencies": { "bullmq": ">=6", "ioredis": ">=5" }, "optionalPeers": ["bullmq", "ioredis"], "bin": { "scraper-daemon": "bin/scraper-daemon.js" } }, "sha512-V/+WT7q6VOedkkepYszctkXdfvWGZbmEZMi9Q7nejO35pg8kPVwR0W53GjswdsAXM/AqA+wMCfB5/oQo1Ihifw=="], + "@profullstack/scraper-daemon": ["@profullstack/scraper-daemon@0.3.0", "", { "peerDependencies": { "bullmq": ">=6", "ioredis": ">=5" }, "optionalPeers": ["bullmq", "ioredis"], "bin": { "scraper-daemon": "bin/scraper-daemon.js" } }, "sha512-C/COc8Oef+/8/Bdu/uyvTZbeIO0rMm3q7wEJNzDqcuoZ+cEs6OEGzrKwCkA1N7zqyi3NASCAytfeFxJiiaG9TQ=="], "@profullstack/submit-feed": ["@profullstack/submit-feed@0.1.3", "", { "dependencies": { "fast-xml-parser": "^5.2.5" } }, "sha512-RMH294Q1qFJ6L01pU5q/yowshx1Xpu4gfGC2OuNaxyZ5jySqM9FvgqvVuWH+G18711SqzNq+DL8dvbAaBgBIvA=="], diff --git a/package.json b/package.json index be1483d..3c84ec0 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "p0dcasters", - "version": "0.4.2", + "version": "0.5.0", "private": true, "type": "module", "scripts": { @@ -18,7 +18,7 @@ "@profullstack/openprofile": "^0.2.0", "@profullstack/player": "0.16.0", "@profullstack/profile": "0.1.6", - "@profullstack/scraper-daemon": "0.2.2", + "@profullstack/scraper-daemon": "0.3.0", "@profullstack/submit-feed": "^0.1.3", "@profullstack/throttle": "^0.4.1", "@profullstack/x402-gateway": "^0.16.1", diff --git a/scripts/daemon-lib.mjs b/scripts/daemon-lib.mjs index 2b305d0..bdd4ccf 100644 --- a/scripts/daemon-lib.mjs +++ b/scripts/daemon-lib.mjs @@ -78,6 +78,7 @@ export async function headStamp(fetchImpl, url = DUMP_URL, { signal, timeoutMs = * @param {{acquire(): Promise, release(): void}} deps.lock the shell's flock * @param {(s: {job: string, lastOk: number|null, lastError: string|null}) => Promise} [deps.alert] * @param {() => Promise} [deps.newsletter] one pass of the weekly digest (src/lib/newsletter.ts) + * @param {object} [deps.crawl] the show crawler (src/lib/feed-crawl.ts): { lib, q, concurrency } * @param {object} [deps.bin] { node, bun, python } * @param {string} [deps.repo] the checkout the scripts live in */ @@ -94,6 +95,7 @@ export function buildDaemon(deps) { lock, alert, newsletter, + crawl = null, console: out = console, now = Date.now, minRows = 10_000, @@ -352,8 +354,9 @@ export function buildDaemon(deps) { redis, store, // Two: the directory pipeline runs for hours and must not hold the weekly - // digest back. Each task still runs one at a time on its own. - concurrency: 2, + // digest back. Each task still runs one at a time on its own. The show + // crawler's reads come on top, so it never waits behind either. + concurrency: 2 + (crawl ? crawl.concurrency + 1 : 0), onRun, onStall: async (s) => { out.error(`[directory] stalled: no success since ${s.lastOk ? new Date(s.lastOk).toISOString() : "start"}${s.lastError ? `: ${s.lastError}` : ""}`); @@ -380,5 +383,52 @@ export function buildDaemon(deps) { }, }); } + if (crawl) addShowCrawler(daemon, crawl, out); return daemon; } + +/** Feeds read per tier at once from one host: anchor.fm alone is 116k shows. */ +export const SHOWS_PER_HOST = 4; +export const SHOWS_JOB = "shows"; + +/** + * The show crawler: every listed show's feed on its own rhythm (see + * src/lib/feed-crawl.ts), plus a job that keeps feed_state in step with the + * podcasts table the dump reload swaps in. + * + * @param {ReturnType} daemon + * @param {{ lib: typeof import('../src/lib/feed-crawl.ts'), q: Function, concurrency: number, tiers?: string[] }} crawl + */ +export function addShowCrawler(daemon, { lib, q, concurrency, tiers }, out = console) { + const TIERS = tiers ?? ["hot", "active", "slow", "dormant", "archived"]; + daemon.job({ + name: "shows-seed", + every: "30m", + timeout: "20m", + async run() { + const seeded = await lib.seedFeedState(q); + // After a reload the fresh rows carry the dump's dates; put back anything newer we read. + const reapplied = await lib.applyToPodcasts(q); + return { ...seeded, reapplied }; + }, + }); + daemon.feeds({ + name: SHOWS_JOB, + concurrency, + perHost: SHOWS_PER_HOST, + timeout: "1m", + staleAfter: "2h", + due: (tier, limit) => lib.dueFeeds(q, TIERS.indexOf(tier), limit), + run: async (feed, ctx) => { + const r = await lib.readFeed(feed, ctx.signal); + return { dates: r.dates, changed: !r.notModified, crawl: r }; + }, + save: (feed, plan, result) => lib.saveCrawl(q, feed, plan, result.crawl), + fail: (feed, f) => lib.saveFailure(q, feed, f), + previous: (feed) => feed.minutes, + fails: (feed) => feed.fails, + // Ten failures in a row and a show is read once a quarter, like an archived one. + retireAfter: 10, + }); + out.log?.(`[p0dcasters.com] show crawler on: ${concurrency} at once, ${SHOWS_PER_HOST} per host`); +} diff --git a/scripts/daemon.mjs b/scripts/daemon.mjs index 2b82f5d..c06483e 100755 --- a/scripts/daemon.mjs +++ b/scripts/daemon.mjs @@ -29,6 +29,7 @@ import { buildDaemon, pathsFor } from "./daemon-lib.mjs"; // Bun runs the TypeScript directly: the same sender the web app's admin route uses. import { runWeeklyNewsletter, slotFor, slotFrom, sqlStore } from "../src/lib/newsletter.ts"; import { sendMail } from "../src/lib/auth/mail.ts"; +import * as feedCrawl from "../src/lib/feed-crawl.ts"; const REPO = join(dirname(fileURLToPath(import.meta.url)), ".."); const P = pathsFor(); @@ -234,7 +235,9 @@ if (!c) { console.error("p0dcasters daemon: no working DATABASE_URL; exiting (systemd retries)"); process.exit(1); } -const sql = new SQL(c.url, { max: 2 }); +// Two for the directory and the digest, plus room for the show crawler's writes. +const CRAWL_CONCURRENCY = Number(process.env.CRAWL_CONCURRENCY ?? 24); +const sql = new SQL(c.url, { max: CRAWL_CONCURRENCY > 0 ? 8 : 2 }); // scraper-daemon 0.1.1+ passes state as text both ways, so Bun's SQL needs no help here. const store = pgStore((text, params) => sql.unsafe(text, params)); @@ -298,6 +301,11 @@ if (process.argv.includes("--status")) { process.exit(0); } +/* ---------- the show crawler (src/lib/feed-crawl.ts) ---------- */ + +const crawlQuery = (text, params = []) => sql.unsafe(text, params); +if (CRAWL_CONCURRENCY > 0) await feedCrawl.ensureFeedState(crawlQuery); + const daemon = buildDaemon({ createDaemon, store, @@ -310,6 +318,7 @@ const daemon = buildDaemon({ lock, alert, newsletter: () => newsletterPass(), + crawl: CRAWL_CONCURRENCY > 0 ? { lib: feedCrawl, q: crawlQuery, concurrency: CRAWL_CONCURRENCY } : null, repo: REPO, }); diff --git a/src/app/api/episodes/[slug]/refresh/route.ts b/src/app/api/episodes/[slug]/refresh/route.ts new file mode 100644 index 0000000..4c05e32 --- /dev/null +++ b/src/app/api/episodes/[slug]/refresh/route.ts @@ -0,0 +1,48 @@ +import { revalidatePath } from "next/cache"; +import { db, one } from "@/lib/db"; +import { episodeCache, fetchEpisodes } from "@/lib/feed"; + +export const dynamic = "force-dynamic"; + +/** A show read this recently by a refresh is answered from what we have. */ +const RECENT_MS = 2 * 60_000; +const lastRefresh = new Map(); + +/** + * Read a show's feed now: the Refresh button on a show page. + * + * The page's episodes come straight from the feed, so dropping the cached copy + * and reading it again is the whole refresh as far as the reader can see. The + * show is also moved to the front of the crawler's queue (feed_state), so the + * directory row every list sorts by is rewritten by the daemon within seconds. + */ +export async function POST(_req: Request, { params }: { params: Promise<{ slug: string }> }) { + const { slug } = await params; + const show = await one<{ slug: string; feed_url: string }>("SELECT slug, feed_url FROM podcasts WHERE slug = ?", [slug]); + if (!show) return Response.json({ error: "not found" }, { status: 404 }); + + const last = lastRefresh.get(show.feed_url) ?? 0; + if (Date.now() - last < RECENT_MS) { + return Response.json({ ok: true, state: "recent" }, { headers: { "cache-control": "no-store" } }); + } + lastRefresh.set(show.feed_url, Date.now()); + if (lastRefresh.size > 5000) lastRefresh.clear(); + + const before = episodeCache.get(show.feed_url); + episodeCache.delete(show.feed_url); + const episodes = await fetchEpisodes(show.feed_url); + const newest = (list: { pubdate: number }[] | undefined) => (list ?? []).reduce((m, e) => Math.max(m, e.pubdate), 0); + const fresh = before ? episodes.filter((e) => e.pubdate > newest(before)).length : null; + + try { + await db().execute({ sql: "UPDATE feed_state SET next_at = '1970-01-01T00:00:00Z' WHERE feed_url = ?", args: [show.feed_url] }); + } catch { + // No crawler state yet (first deploy): the page is still current. + } + revalidatePath(`/podcast/${show.slug}`); + + return Response.json( + { ok: episodes.length > 0, state: episodes.length ? "read" : "failed", episodes: episodes.length, newEpisodes: fresh, newest: newest(episodes) || null }, + { headers: { "cache-control": "no-store" } }, + ); +} diff --git a/src/app/podcast/[slug]/page.tsx b/src/app/podcast/[slug]/page.tsx index ae352e6..1fd36e5 100644 --- a/src/app/podcast/[slug]/page.tsx +++ b/src/app/podcast/[slug]/page.tsx @@ -12,6 +12,7 @@ import FollowButton from "@/components/FollowButton"; import { LatestButton, ShowEpisodes } from "@/components/ShowEpisodes"; import PlaylistButtons from "@/components/PlaylistButtons"; import ShareButton from "@/components/ShareButton"; +import RefreshButton from "@/components/RefreshButton"; import { nixampUrl, playlistUrl } from "@/lib/playlist"; import { profileUrl } from "@/lib/openprofile/generate"; import { cadence, languageName, titleCase, clamp, safeImage, normalizeLang } from "@/lib/format"; @@ -192,6 +193,7 @@ export default async function Show({ params }: { params: Promise<{ slug: string + RSS feed diff --git a/src/components/RefreshButton.tsx b/src/components/RefreshButton.tsx new file mode 100644 index 0000000..23f923d --- /dev/null +++ b/src/components/RefreshButton.tsx @@ -0,0 +1,35 @@ +"use client"; + +import { useRouter } from "next/navigation"; +import { useState } from "react"; + +/** + * Read the show's feed now instead of waiting for the crawler: POSTs to + * /api/episodes//refresh, then re-renders the page with what came back. + */ +export default function RefreshButton({ slug }: { slug: string }) { + const router = useRouter(); + const [state, setState] = useState<"idle" | "busy" | "done" | "same" | "failed">("idle"); + + const refresh = async () => { + setState("busy"); + try { + const res = await fetch(`/api/episodes/${encodeURIComponent(slug)}/refresh`, { method: "POST" }); + const body = (await res.json().catch(() => ({}))) as { ok?: boolean; state?: string; newEpisodes?: number | null }; + if (!body.ok) return setState("failed"); + setState(body.newEpisodes ? "done" : "same"); + router.refresh(); + } catch { + setState("failed"); + } finally { + window.setTimeout(() => setState("idle"), 2500); + } + }; + + const label = { idle: "↻ Refresh", busy: "Reading…", done: "New episodes", same: "Up to date", failed: "Feed did not answer" }[state]; + return ( + + ); +} diff --git a/src/lib/feed-crawl.ts b/src/lib/feed-crawl.ts new file mode 100644 index 0000000..e1ecd35 --- /dev/null +++ b/src/lib/feed-crawl.ts @@ -0,0 +1,225 @@ +// The show crawler: keeps the directory's "newest episode" current between dumps. +// +// The directory itself comes from the Podcast Index dump every few days, and a +// show page reads its episodes live, but every list on the site (latest, a +// category, search) sorts by podcasts.newest_pubdate, which is only as fresh as +// the dump. This reads each show's feed on its own rhythm, with the house cadence +// (@profullstack/scraper-daemon/cadence): fifteen minutes for a show posting +// hourly, every few hours around a weekly show's release day, up to a week for a +// quiet one, and a show silent six months is archived (read once a quarter). +// +// State lives in feed_state, keyed by feed URL, because the loader rebuilds the +// podcasts table wholesale on every dump; applyToPodcasts() puts what the crawler +// learned back onto the new rows. +// +// Every function takes `q(text, params)`: Postgres with $n placeholders (Bun SQL's +// unsafe() in the daemon), so this file has no driver of its own. +import { tierSql } from "@profullstack/scraper-daemon/cadence"; +import { closeTruncatedFeed, MAX_FEED_BYTES, readCappedText } from "./feed-body"; +import { parseFeed, type Episode } from "./feed"; + +export type Query = (text: string, params?: unknown[]) => Promise[]>; + +export type FeedRow = { + key: string; + url: string; + minutes: number; + fails: number; + etag: string | null; + last_modified: string | null; + dates: number[]; +}; + +const UA = "p0dcasters/1.0 (+https://p0dcasters.com)"; +const TIER = tierSql("minutes"); +/** The last this-many publish dates are kept, so a 304 can still be scheduled on rhythm. */ +const KEEP_DATES = 30; + +export const FEED_STATE_DDL = [ + `CREATE TABLE IF NOT EXISTS feed_state( + feed_url text PRIMARY KEY, + minutes integer NOT NULL DEFAULT 60, + next_at timestamptz NOT NULL DEFAULT now(), + fails integer NOT NULL DEFAULT 0, + last_error text, + etag text, + last_modified text, + checked_at timestamptz, + ok_at timestamptz, + dates jsonb NOT NULL DEFAULT '[]', + newest_pubdate bigint, + episode_count integer, + latest_audio text, + latest_duration integer + )`, + // Must stay exactly tierSql("minutes"): dueFeeds filters on that expression. + `CREATE INDEX IF NOT EXISTS feed_state_due_tier ON feed_state((${TIER}), next_at)`, +]; + +export async function ensureFeedState(q: Query): Promise { + for (const s of FEED_STATE_DDL) await q(s); +} + +/** + * Give every listed show a row, and drop rows for shows no longer listed. + * + * A new show starts on a guess from what the dump knows (per_week and the newest + * episode) and is spread over the next six hours, so a reload of 300k new rows is + * not 300k reads in one minute; the first real read replaces the guess. + */ +export async function seedFeedState(q: Query): Promise<{ added: number; dropped: number }> { + const added = await q( + `INSERT INTO feed_state(feed_url, minutes, next_at, newest_pubdate) + SELECT p.feed_url, s.minutes, + now() + random() * least(s.minutes, 360) * interval '1 minute', + p.newest_pubdate + FROM podcasts p + CROSS JOIN LATERAL (SELECT CASE + WHEN p.newest_pubdate < extract(epoch FROM now() - interval '180 days') THEN 129600 + WHEN coalesce(p.per_week, 0) <= 0 THEN 1440 + ELSE least(10079, greatest(15, round(10080 / p.per_week / 4)))::int + END AS minutes) s + ON CONFLICT (feed_url) DO NOTHING + RETURNING 1`, + ); + const dropped = await q( + `DELETE FROM feed_state s WHERE NOT EXISTS (SELECT 1 FROM podcasts p WHERE p.feed_url = s.feed_url) RETURNING 1`, + ); + return { added: added.length, dropped: dropped.length }; +} + +/** Due shows of one tier (0 hot .. 4 archived), oldest due first. */ +export async function dueFeeds(q: Query, tierIndex: number, limit: number): Promise { + const rows = await q( + `SELECT feed_url, minutes, fails, etag, last_modified, dates + FROM feed_state WHERE ${TIER} = $1 AND next_at <= now() + ORDER BY next_at LIMIT $2`, + [tierIndex, limit], + ); + return rows.map((r) => ({ + key: String(r.feed_url), + url: String(r.feed_url), + minutes: Number(r.minutes), + fails: Number(r.fails ?? 0), + etag: (r.etag as string | null) ?? null, + last_modified: (r.last_modified as string | null) ?? null, + dates: Array.isArray(r.dates) ? (r.dates as number[]) : typeof r.dates === "string" ? (JSON.parse(r.dates) as number[]) : [], + })); +} + +export type CrawlResult = { + notModified: boolean; + episodes: Episode[]; + dates: number[]; + etag: string | null; + lastModified: string | null; +}; + +/** One conditional read of a feed. Throws on anything but 200/304, so the daemon's error ladder applies. */ +export async function readFeed(feed: FeedRow, signal?: AbortSignal): Promise { + const headers: Record = { "user-agent": UA, accept: "application/rss+xml, application/xml, */*" }; + if (feed.etag) headers["if-none-match"] = feed.etag; + if (feed.last_modified) headers["if-modified-since"] = feed.last_modified; + const timeout = AbortSignal.timeout(20_000); + const res = await fetch(feed.url, { headers, signal: signal ? AbortSignal.any([signal, timeout]) : timeout, redirect: "follow" }); + if (res.status === 304) { + res.body?.cancel().catch(() => {}); + return { notModified: true, episodes: [], dates: feed.dates.map((s) => s * 1000), etag: feed.etag, lastModified: feed.last_modified }; + } + if (res.status === 429 || res.status === 503) { + res.body?.cancel().catch(() => {}); + const after = Number(res.headers.get("retry-after")); + throw Object.assign(new Error(`HTTP ${res.status}`), { retryAfterMs: Number.isFinite(after) && after > 0 ? after * 1000 : 30 * 60_000 }); + } + if (!res.ok) { + res.body?.cancel().catch(() => {}); + throw new Error(`HTTP ${res.status}`); + } + const body = await readCappedText(res, MAX_FEED_BYTES); + const xml = body.truncated ? closeTruncatedFeed(body.text) : body.text; + if (!xml) throw new Error(`feed over ${MAX_FEED_BYTES} bytes with no complete item`); + const episodes = parseFeed(xml); + return { + notModified: false, + episodes, + dates: episodes.filter((e) => e.pubdate > 0).map((e) => e.pubdate * 1000), + etag: res.headers.get("etag"), + lastModified: res.headers.get("last-modified"), + }; +} + +/** The newest dated episode, which is what the directory sorts by. */ +export function newestEpisode(episodes: Episode[]): Episode | null { + let best: Episode | null = null; + for (const e of episodes) if (e.pubdate > 0 && (!best || e.pubdate > best.pubdate)) best = e; + return best; +} + +/** Write the next read and what was learned, to feed_state and to the live podcasts row. */ +export async function saveCrawl( + q: Query, + feed: FeedRow, + plan: { minutes: number; nextAt: number }, + result: CrawlResult, +): Promise { + const newest = newestEpisode(result.episodes); + const dates = [...new Set(result.dates.map((t) => Math.floor(t / 1000)))].sort((a, b) => b - a).slice(0, KEEP_DATES); + await q( + `UPDATE feed_state SET minutes = $1, next_at = to_timestamp($2), fails = 0, last_error = NULL, + etag = $3, last_modified = $4, checked_at = now(), ok_at = now(), dates = $5::text::jsonb, + newest_pubdate = coalesce($6, newest_pubdate), + episode_count = coalesce($7, episode_count), + latest_audio = coalesce($8, latest_audio), + latest_duration = coalesce($9, latest_duration) + WHERE feed_url = $10`, + [ + plan.minutes, + plan.nextAt / 1000, + result.etag, + result.lastModified, + JSON.stringify(dates), + newest?.pubdate ?? null, + result.notModified ? null : result.episodes.length, + newest?.source ?? null, + newest?.duration ?? null, + feed.url, + ], + ); + if (newest) { + // Only ever forward: a feed that trimmed its archive must not make a show look older. + await q( + `UPDATE podcasts SET newest_pubdate = $1, latest_audio = $2, latest_duration = coalesce($3, latest_duration), + episode_count = greatest(episode_count, $4) + WHERE feed_url = $5 AND newest_pubdate < $6`, + [newest.pubdate, newest.source, newest.duration, result.episodes.length, feed.url, newest.pubdate], + ); + } +} + +export async function saveFailure( + q: Query, + feed: FeedRow, + f: { error: string; held: boolean; fails: number | null; nextAt: number; retire: boolean }, +): Promise { + await q( + `UPDATE feed_state SET next_at = to_timestamp($1), checked_at = now(), last_error = $2, + fails = CASE WHEN $3 THEN fails ELSE fails + 1 END, + minutes = CASE WHEN $4 THEN 129600 ELSE minutes END + WHERE feed_url = $5`, + [f.nextAt / 1000, f.error.slice(0, 500), f.held, f.retire, feed.url], + ); +} + +/** After a reload swapped in fresh podcasts rows: re-apply anything newer the crawler already saw. */ +export async function applyToPodcasts(q: Query): Promise { + const rows = await q( + `UPDATE podcasts p SET newest_pubdate = s.newest_pubdate, + latest_audio = coalesce(s.latest_audio, p.latest_audio), + latest_duration = coalesce(s.latest_duration, p.latest_duration), + episode_count = greatest(p.episode_count, coalesce(s.episode_count, 0)) + FROM feed_state s + WHERE s.feed_url = p.feed_url AND s.newest_pubdate > p.newest_pubdate + RETURNING 1`, + ); + return rows.length; +} diff --git a/test/feed-crawl.test.mjs b/test/feed-crawl.test.mjs new file mode 100644 index 0000000..87c32ee --- /dev/null +++ b/test/feed-crawl.test.mjs @@ -0,0 +1,109 @@ +import { describe, expect, test, afterAll, beforeAll } from "bun:test"; +import { SQL } from "bun"; +import { createDaemon } from "@profullstack/scraper-daemon"; +import * as lib from "../src/lib/feed-crawl.ts"; +import { addShowCrawler } from "../scripts/daemon-lib.mjs"; + +const url = process.env.TEST_DATABASE_URL; +const NOW = Math.floor(Date.now() / 1000); +const DAY = 86_400; + +describe.skipIf(!url)("feed_state against Postgres", () => { + let sql; + let q; + const db = `p0d_crawl_${process.pid}`; + beforeAll(async () => { + const admin = new SQL(url); + await admin.unsafe(`CREATE DATABASE ${db}`); + await admin.close(); + sql = new SQL(url.replace(/\/[^/]*$/, `/${db}`)); + q = (t, p = []) => sql.unsafe(t, p); + await q(`CREATE TABLE podcasts(feed_url text, slug text, newest_pubdate bigint NOT NULL, per_week double precision, + episode_count bigint NOT NULL DEFAULT 0, latest_audio text, latest_duration bigint)`); + await q(`INSERT INTO podcasts VALUES + ('https://feeds.example/weekly.xml', 'weekly', ${NOW - 9 * DAY}, 1, 20, 'old.mp3', 100), + ('https://feeds.example/daily.xml', 'daily', ${NOW - DAY}, 7, 300, 'd.mp3', 100), + ('https://feeds.example/gone.xml', 'gone', ${NOW - 200 * DAY}, 1, 5, 'g.mp3', 100)`); + await lib.ensureFeedState(q); + }); + afterAll(async () => { + await sql?.close(); + const admin = new SQL(url); + await admin.unsafe(`DROP DATABASE IF EXISTS ${db} WITH (FORCE)`); + await admin.close(); + }); + + test("seeding guesses a cadence from the dump and archives long-silent shows", async () => { + expect(await lib.seedFeedState(q)).toEqual({ added: 3, dropped: 0 }); + const rows = Object.fromEntries((await q(`SELECT feed_url, minutes FROM feed_state`)).map((r) => [r.feed_url, r.minutes])); + expect(rows["https://feeds.example/weekly.xml"]).toBe(2520); + expect(rows["https://feeds.example/daily.xml"]).toBe(360); + expect(rows["https://feeds.example/gone.xml"]).toBe(129600); + expect(await lib.seedFeedState(q)).toEqual({ added: 0, dropped: 0 }); + }); + + test("due shows come back per tier, and a crawl moves the directory row forward", async () => { + await q(`UPDATE feed_state SET next_at = now() - interval '1 minute'`); + const slow = await lib.dueFeeds(q, 2, 10); + expect(slow.map((f) => f.url)).toEqual(["https://feeds.example/weekly.xml"]); + expect(await lib.dueFeeds(q, 4, 10)).toHaveLength(1); + + const feed = slow[0]; + const episodes = [ + { id: "a", title: "new", audio: "new.mp3", source: "new.mp3", pubdate: NOW - 3600, duration: 3000, description: "", image: null, link: null }, + { id: "b", title: "old", audio: "old.mp3", source: "old.mp3", pubdate: NOW - 9 * DAY, duration: 100, description: "", image: null, link: null }, + ]; + await lib.saveCrawl(q, feed, { minutes: 2520, nextAt: Date.now() + 2520 * 60_000 }, { notModified: false, episodes, dates: episodes.map((e) => e.pubdate * 1000), etag: '"v1"', lastModified: null }); + const [p] = await q(`SELECT newest_pubdate, latest_audio, episode_count FROM podcasts WHERE slug = 'weekly'`); + expect(Number(p.newest_pubdate)).toBe(NOW - 3600); + expect(p.latest_audio).toBe("new.mp3"); + const [s] = await q(`SELECT etag, dates, fails FROM feed_state WHERE feed_url = $1`, [feed.url]); + expect(s.etag).toBe('"v1"'); + expect(await lib.dueFeeds(q, 2, 10)).toHaveLength(0); + }); + + test("a reload that swaps in older rows gets the crawler's dates back", async () => { + await q(`UPDATE podcasts SET newest_pubdate = ${NOW - 9 * DAY} WHERE slug = 'weekly'`); + expect(await lib.applyToPodcasts(q)).toBe(1); + const [p] = await q(`SELECT newest_pubdate FROM podcasts WHERE slug = 'weekly'`); + expect(Number(p.newest_pubdate)).toBe(NOW - 3600); + }); + + test("failures count, a 429 does not, and retiring parks the show for a quarter", async () => { + await lib.saveFailure(q, { url: "https://feeds.example/daily.xml" }, { error: "HTTP 500", held: false, fails: 1, nextAt: Date.now(), retire: false }); + await lib.saveFailure(q, { url: "https://feeds.example/daily.xml" }, { error: "HTTP 429", held: true, fails: null, nextAt: Date.now(), retire: false }); + let [s] = await q(`SELECT fails, minutes FROM feed_state WHERE feed_url = 'https://feeds.example/daily.xml'`); + expect(s.fails).toBe(1); + await lib.saveFailure(q, { url: "https://feeds.example/daily.xml" }, { error: "HTTP 500", held: false, fails: 10, nextAt: Date.now(), retire: true }); + [s] = await q(`SELECT fails, minutes FROM feed_state WHERE feed_url = 'https://feeds.example/daily.xml'`); + expect(s.minutes).toBe(129600); + }); + + test("shows dropped from the directory lose their state", async () => { + await q(`DELETE FROM podcasts WHERE slug = 'gone'`); + expect(await lib.seedFeedState(q)).toEqual({ added: 0, dropped: 1 }); + }); +}); + +describe("addShowCrawler", () => { + test("reads due shows hot first and writes the plan back", async () => { + const saved = []; + const fake = { + seedFeedState: async () => ({ added: 0, dropped: 0 }), + applyToPodcasts: async () => 0, + dueFeeds: async (_q, tier, limit) => + tier === 0 && !saved.length ? [{ key: "a", url: "https://a.example/rss", minutes: 60, fails: 0, dates: [] }].slice(0, limit) : [], + readFeed: async () => ({ notModified: false, episodes: [], dates: [Date.now() - 1_800_000, Date.now() - 3_600_000], etag: null, lastModified: null }), + saveCrawl: async (_q, feed, plan) => void saved.push({ feed, plan }), + saveFailure: async () => {}, + }; + const d = createDaemon({ log: () => {}, tickMs: 60_000 }); + addShowCrawler(d, { lib: fake, q: async () => [], concurrency: 4 }, { log: () => {} }); + d.start(); + await new Promise((r) => setTimeout(r, 50)); + await d.stop(); + expect(saved).toHaveLength(1); + expect(saved[0].plan.minutes).toBe(15); + expect(saved[0].plan.tier).toBe("hot"); + }); +});