diff --git a/apps/poller/src/daemon.js b/apps/poller/src/daemon.js index d20ddb6..958f3a1 100644 --- a/apps/poller/src/daemon.js +++ b/apps/poller/src/daemon.js @@ -63,6 +63,7 @@ export const CRAWL_STALE_AFTER = '30m'; * sourceIntervalMs: number, * topicSearchIntervalMs: number, * topicsIntervalMs: number, + * trendsIntervalMs?: number, * statusMs?: number, * }, * run: Record Promise>, @@ -100,6 +101,9 @@ export function buildPollerDaemon({ createDaemon = defaultCreateDaemon, store, r if (c.clusterBackfill) add('cluster-backfill', c.clusterIntervalMs, run.clusterBackfill); if (c.cardEnabled) add('cards', c.cardIntervalMs, run.cards); add('alerts', c.alertIntervalMs, run.alerts); + // Outside the catch-up gate: a small read most hours and a sample once a day, + // and prod has run with CRAWL_CATCHUP=1 for weeks. + if (run.trends) add('trends', c.trendsIntervalMs ?? 3_600_000, run.trends); // The weekly digest: a no-op outside its slot. Optional so callers that build // a daemon without it (tests, older configs) keep working. if (run.newsletter) add('newsletter', c.newsletterIntervalMs ?? 600_000, run.newsletter); diff --git a/apps/poller/src/index.js b/apps/poller/src/index.js index 3856c9c..24e9abf 100644 --- a/apps/poller/src/index.js +++ b/apps/poller/src/index.js @@ -11,6 +11,7 @@ import { warmDirectoryCache, warmLiveStatsCache, warmPanelCaches, + trends, } from '@rssamplifier/db'; import { crawlDue, @@ -523,6 +524,14 @@ async function topicsTick() { log('topics', { topics, ms: Date.now() - started }); } +// Hourly, and a single small read on the 23 hours with no finished day to add. +async function trendsTick() { + if (stopping) return; + const started = Date.now(); + const days = await trends.rollupTrendDays(db); + if (days.length) log('trends', { days, ms: Date.now() - started }); +} + /** * Sweep the tables that grow by themselves. @@ -1245,6 +1254,7 @@ const daemon = buildPollerDaemon({ discoveryTopics: discoveryTopicsTick, discoveryNotify: discoveryNotifyTick, topics: topicsTick, + trends: trendsTick, clusterBackfill: backfillTick, cards: cardTick, alerts: alertTick, diff --git a/apps/web/src/app/api/trending/route.js b/apps/web/src/app/api/trending/route.js new file mode 100644 index 0000000..c6fac00 --- /dev/null +++ b/apps/web/src/app/api/trending/route.js @@ -0,0 +1,35 @@ +import { trends } from '@rssamplifier/db'; + +import { db } from '../../../lib/db.js'; + +export const dynamic = 'force-dynamic'; + +const clamp = (value, min, max, fallback) => { + const n = Math.round(Number(value)); + return Number.isFinite(n) && n > 0 ? Math.min(max, Math.max(min, n)) : fallback; +}; + +/** + * What the directory is writing about more than it was last week. + * + * Open, like the rest of the directory. Reads the daily rollup, so it is cheap, + * and the numbers only change when a day is rolled up, hence the hour of cache. + * `count` is feeds in a one in four sample of English blogs and news sites, so + * it is a measure of movement rather than a census. + */ +export async function GET(request) { + const params = new URL(request.url).searchParams; + const window = clamp(params.get('window'), 1, 14, 7); + const limit = clamp(params.get('limit'), 1, 200, 50); + const result = await trends.trendingTerms(db(), { window, limit }); + return Response.json( + { + source: 'rssamp', + window_days: window, + generated_at: result.generatedAt, + documents: result.feeds, + topics: result.topics, + }, + { headers: { 'cache-control': 'public, max-age=3600' } }, + ); +} diff --git a/packages/db/index.js b/packages/db/index.js index 92a07d0..514af26 100644 --- a/packages/db/index.js +++ b/packages/db/index.js @@ -27,3 +27,4 @@ export * as crawlSales from './src/crawl-sales.js'; export * as webrings from './src/webrings.js'; export * as newsletter from './src/newsletter.js'; export * as coinpay from './src/coinpay.js'; +export * as trends from './src/trends.js'; diff --git a/packages/db/migrations-pg/20261107020000_trend_terms.sql b/packages/db/migrations-pg/20261107020000_trend_terms.sql new file mode 100644 index 0000000..dd4e431 --- /dev/null +++ b/packages/db/migrations-pg/20261107020000_trend_terms.sql @@ -0,0 +1,22 @@ +-- What the directory is writing about, one day at a time. A trend is a +-- comparison of two weeks of titles, and feed_items is too large to scan for +-- that on a request, so the poller rolls each finished UTC day into these. + +-- One row per day rolled up. `feeds` is the denominator: a day with more feeds +-- in the sample is not a day when everything trended. +create table if not exists trend_days ( + day text primary key, + feeds bigint not null, + items bigint not null, + rolled_at text not null +); + +-- How many sampled feeds used a term in a title that day. Feeds, not items, so +-- a firehose posting a thousand times is one voice. +create table if not exists trend_terms ( + day text not null references trend_days (day) on delete cascade, + term text not null, + feeds bigint not null, + primary key (day, term) +); +create index if not exists trend_terms_term_idx on trend_terms (term); diff --git a/packages/db/src/trends.js b/packages/db/src/trends.js new file mode 100644 index 0000000..d5fa31e --- /dev/null +++ b/packages/db/src/trends.js @@ -0,0 +1,240 @@ +/** + * What the directory is writing about more than it was last week. + * + * Grouping two weeks of feed_items on a request does not finish (a plain count + * over the range timed out at 90 seconds on 2026-11-07), so each finished UTC + * day is rolled up once into trend_days and trend_terms and the API reads those. + * + * Three choices the shape of the data forces: + * + * - Feeds, not items. One npm firehose posts more than a thousand blogs, so a + * term counts once per feed per day. + * - A sample. One item in four is plenty to see a subject moving, and it keeps + * a day to about 70k rows. + * - Rates, not counts. The crawler's reach changes from week to week, so both + * sides are compared per feed sampled. + */ +import { nowIso } from './client.js'; + +const PAGE = 5000; +const SAMPLE_MOD = 4; +/** Below this many sampled feeds a term is one newsroom's vocabulary. */ +const MIN_DAY_FEEDS = 3; +const MIN_WINDOW_FEEDS = 10; +/** A term must be at least this much more common than the week before. */ +const MIN_LIFT = 1.5; +/** Smooths a term that was absent last week, so 12 from nothing is not infinite. */ +const PRIOR_FLOOR = 3; +const KEEP_DAYS = 30; +const DAY_MS = 86_400_000; + +const STOP = new Set( + `a about after again against all also am an and any are around as at away back be because been before being + best between big both but by can could day days did do does done down during each even every first for from + get gets getting go goes going good got great had has have having he her here hers him his how i if in inside + into is it its just know last latest less like live look looks made make makes many may me might more most much + must my near need needs never new news next no not now of off old on once one only onto or other our out over + own part people per place read really report reports right said same say says see she should show shows since + so some still such take takes than that the their them then there these they thing things think this those + three through time times to today together tomorrow too top two under until up update updates us use used using + very via video vs want wants was watch way ways we week weeks well were what when where which while who whose + why will win with within without work works world would year years yesterday yet you your + january february march april may june july august september october november december + monday tuesday wednesday thursday friday saturday sunday + episode podcast review blog post article page free online home guide tips`.split(/\s+/), +); + +// Feeds declare English and publish something else: on the sample day most of +// the false trends were Indonesian. A headline with one of these is not English +// enough to count, and a headline with none of ENGLISH cannot be shown to be. +const FOREIGN = new Set( + 'dan yang di dengan untuk ke dari pada ini itu de la el los las del para por una con und der die das ein eine le les des du et que em do da os um nao il di per sono het een van'.split( + ' ', + ), +); +const ENGLISH = new Set( + 'the of to in for and on with is at from by an as after how why what who are was will has have be that this it its over his her their'.split( + ' ', + ), +); + +const words = (title) => + (String(title ?? '') + .toLowerCase() + .replace(/&#?[a-z0-9]+;|https?:\S+/g, ' ') + .replace(/[‘’]/g, "'") + .match(/[a-z0-9][a-z0-9+#.'-]*[a-z0-9+#]|[a-z]/g) ?? [] + ).map((w) => w.replace(/'s$/, '')); + +export function isEnglishTitle(title) { + const list = words(title); + return list.some((w) => ENGLISH.has(w)) && !list.some((w) => FOREIGN.has(w)); +} + +const usable = (word) => + !!word && word.length >= 3 && word.length <= 24 && !STOP.has(word) && !/^[0-9.+#'-]+$/.test(word); + +/** Single words and adjacent pairs, the pairs being what names a subject. */ +export function titleTerms(title) { + const list = words(title); + const out = new Set(); + for (let i = 0; i < list.length; i++) { + if (!usable(list[i])) continue; + out.add(list[i]); + if (usable(list[i + 1])) out.add(`${list[i]} ${list[i + 1]}`); + } + return out; +} + +/** @param {Iterable<{ feed_id: string, title: string }>} rows */ +export function tallyTerms(rows) { + const byFeed = new Map(); + let items = 0; + for (const row of rows) { + if (!isEnglishTitle(row.title)) continue; + items++; + let terms = byFeed.get(row.feed_id); + if (!terms) byFeed.set(row.feed_id, (terms = new Set())); + for (const term of titleTerms(row.title)) terms.add(term); + } + const terms = new Map(); + for (const seen of byFeed.values()) { + for (const term of seen) terms.set(term, (terms.get(term) ?? 0) + 1); + } + return { feeds: byFeed.size, items, terms }; +} + +const isoDay = (ms) => new Date(ms).toISOString().slice(0, 10); + +/** Rolls one UTC day up, replacing any earlier result for it. */ +export async function rollupTrendDay(db, day) { + const from = `${day}T00`; + const to = `${isoDay(Date.parse(`${day}T00:00:00Z`) + DAY_MS)}T00`; + // Skips a backfill: a feed crawled for the first time brings years of posts + // with today's created_at, and those are not what anyone is writing today. + const oldest = isoDay(Date.parse(`${day}T00:00:00Z`) - 3 * DAY_MS); + + const rows = []; + let after = 0; + for (;;) { + const page = await db.execute({ + sql: `select i.rowid as rowid, i.feed_id, i.title + from feed_items i join feeds f on f.id = i.feed_id + where i.created_at >= ? and i.created_at < ? and i.rowid > ? + and i.rowid % ? = 0 + and i.published_at >= ? + and f.category in ('blog', 'news') + and (f.language is null or lower(f.language) like 'en%') + order by i.rowid + limit ?`, + args: [from, to, after, SAMPLE_MOD, oldest, PAGE], + }); + for (const row of page.rows) rows.push({ feed_id: row.feed_id, title: row.title }); + if (page.rows.length < PAGE) break; + after = Number(page.rows.at(-1).rowid); + } + + const { feeds, items, terms } = tallyTerms(rows); + const kept = [...terms].filter(([, n]) => n >= MIN_DAY_FEEDS); + await db.execute({ sql: 'delete from trend_days where day = ?', args: [day] }); + await db.execute({ + sql: 'insert into trend_days (day, feeds, items, rolled_at) values (?, ?, ?, ?)', + args: [day, feeds, items, nowIso()], + }); + for (let i = 0; i < kept.length; i += 500) { + const chunk = kept.slice(i, i + 500); + await db.execute({ + sql: `insert into trend_terms (day, term, feeds) values ${chunk.map(() => '(?, ?, ?)').join(', ')}`, + args: chunk.flatMap(([term, n]) => [day, term, n]), + }); + } + return { day, feeds, items, terms: kept.length }; +} + +/** + * Rolls up the finished days still missing, oldest first. The first run does a + * fortnight, a sample of about 70k rows a day; after that it is one day a day. + * + * @param {{ now?: number, days?: number }} [options] + */ +export async function rollupTrendDays(db, { now = Date.now(), days = 15 } = {}) { + const wanted = Array.from({ length: days }, (_, i) => isoDay(now - (days - i) * DAY_MS)); + const { rows } = await db.execute({ sql: 'select day from trend_days where day >= ?', args: [wanted[0]] }); + const have = new Set(rows.map((row) => row.day)); + const done = []; + for (const day of wanted.filter((d) => !have.has(d))) { + done.push(await rollupTrendDay(db, day)); + } + await db.execute({ sql: 'delete from trend_days where day < ?', args: [isoDay(now - KEEP_DAYS * DAY_MS)] }); + return done; +} + +/** + * Ranks terms from two windows. + * + * A subject shows up as a handful of overlapping terms (joshua wong, wong, + * activist joshua, wong sentenced), so walking down the ranking keeps the first + * and drops anything sharing a word with what is already kept. Pairs rank ahead + * of their own single words because they say what the subject is. + */ +export function rankTerms({ current, prior, currentFeeds, priorFeeds, limit = 50 }) { + if (!(currentFeeds > 0) || !(priorFeeds > 0)) return []; + const scored = []; + for (const [term, count] of current) { + if (count < MIN_WINDOW_FEEDS) continue; + const before = prior.get(term) ?? 0; + const scaledBefore = (before * currentFeeds) / priorFeeds; + const lift = count / (scaledBefore + PRIOR_FLOOR); + if (lift < MIN_LIFT) continue; + const pair = term.includes(' ') ? 1.5 : 1; + scored.push({ topic: term, count, prior_count: before, score: Math.round(lift * Math.log2(count) * pair * 100) / 100 }); + } + scored.sort((a, b) => b.score - a.score || b.count - a.count); + + const taken = new Set(); + const out = []; + for (const row of scored) { + const parts = row.topic.split(' '); + if (parts.some((w) => taken.has(w))) continue; + parts.forEach((w) => taken.add(w)); + out.push(row); + if (out.length >= limit) break; + } + return out; +} + +/** + * The current ranking: the latest `window` rolled days against the `window` + * before them. + * + * @param {{ window?: number, limit?: number }} [options] + */ +export async function trendingTerms(db, { window = 7, limit = 50 } = {}) { + const { rows: days } = await db.execute({ + sql: 'select day, feeds, items, rolled_at from trend_days order by day desc limit ?', + args: [window * 2], + }); + const empty = { topics: [], days: days.length, feeds: 0, generatedAt: days[0]?.rolled_at ?? null }; + if (days.length < 2) return empty; + // Fewer than two full windows splits what there is, rather than showing nothing. + const size = Math.min(window, Math.floor(days.length / 2)); + const recent = days.slice(0, size); + const earlier = days.slice(size, size * 2); + const sum = async (span) => { + const { rows } = await db.execute({ + sql: `select term, sum(feeds)::bigint as feeds from trend_terms + where day >= ? and day <= ? group by term`, + args: [span.at(-1).day, span[0].day], + }); + return new Map(rows.map((row) => [row.term, Number(row.feeds)])); + }; + const feedsIn = (span) => span.reduce((n, d) => n + Number(d.feeds), 0); + const topics = rankTerms({ + current: await sum(recent), + prior: await sum(earlier), + currentFeeds: feedsIn(recent), + priorFeeds: feedsIn(earlier), + limit, + }); + return { topics, days: days.length, feeds: feedsIn(recent), generatedAt: days[0].rolled_at }; +} diff --git a/packages/db/test/trends.test.js b/packages/db/test/trends.test.js new file mode 100644 index 0000000..6ab190f --- /dev/null +++ b/packages/db/test/trends.test.js @@ -0,0 +1,59 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +import { isEnglishTitle, rankTerms, tallyTerms, titleTerms } from '../src/trends.js'; + +test('a title gives its subjects and leaves out the filler', () => { + const terms = titleTerms("Joshua Wong's sentence is the longest yet"); + assert.ok(terms.has('joshua wong')); + assert.ok(terms.has('sentence')); + assert.ok(!terms.has('the')); + assert.ok(!terms.has('longest yet')); +}); + +test('feeds that declare English and publish something else are left out', () => { + assert.equal(isEnglishTitle('Kasus penyadapan dan dukungan Prabowo'), false); + assert.equal(isEnglishTitle('Listyo Sigit tunjukkan'), false, 'nothing shows it is English'); + assert.equal(isEnglishTitle('Democrats sweep the House in a blue wave'), true); +}); + +test('a firehose is one voice', () => { + const rows = [ + ...Array.from({ length: 500 }, () => ({ feed_id: 'npm', title: 'Release of the widget' })), + { feed_id: 'blog', title: 'Thoughts on the widget' }, + ]; + const { feeds, terms, items } = tallyTerms(rows); + assert.equal(items, 501); + assert.equal(feeds, 2); + assert.equal(terms.get('widget'), 2); +}); + +const rank = (current, prior, feeds = [1000, 1000]) => + rankTerms({ current: new Map(current), prior: new Map(prior), currentFeeds: feeds[0], priorFeeds: feeds[1] }); + +test('a rising subject ranks and a steady one does not', () => { + const out = rank( + [['midterm results', 60], ['music', 400]], + [['midterm results', 2], ['music', 390]], + ); + assert.deepEqual(out.map((t) => t.topic), ['midterm results']); + assert.equal(out[0].count, 60); + assert.equal(out[0].prior_count, 2); +}); + +test('one subject appears once, under its name', () => { + const out = rank( + [['joshua wong', 60], ['wong', 70], ['activist joshua', 40], ['wong sentenced', 30]], + [], + ); + assert.deepEqual(out.map((t) => t.topic), ['joshua wong']); +}); + +test('the rate is compared, so a bigger crawl does not make everything trend', () => { + assert.deepEqual(rank([['cats', 200]], [['cats', 100]], [2000, 1000]), []); +}); + +test('too few feeds is noise, and an empty window ranks nothing', () => { + assert.deepEqual(rank([['rare thing', 5]], []), []); + assert.deepEqual(rank([['subject', 50]], [], [0, 1000]), []); +});