reddit: firehose keeps every /r/* feed current; fix nested NUL placeholders

apps/poller/src/daemon.js

@@ -84,6 +84,8 @@ export function buildPollerDaemon({ createDaemon = defaultCreateDaemon, store, r
   add('crawl', c.intervalMs, run.crawl, { staleAfter: CRAWL_STALE_AFTER });
   add('import', c.intervalMs, run.import);
   add('purge', 3_600_000, run.purge);
+  // Outside catch-up too: it is what keeps all of /r/* current.
+  if (c.redditFirehose && run.redditFirehose) add('reddit-firehose', c.redditFirehoseMs ?? 120_000, run.redditFirehose);
 
   // Everything the old tick() skipped below `if (catchupOnly) return`.
   if (!c.catchupOnly) {

apps/poller/src/index.js

@@ -24,6 +24,7 @@ import {
   drainImport,
   seedTopRings,
   verifyRingMembers,
+  createRedditFirehose,
 } from '@rssamplifier/ingest';
 import { runDueSources, discoverFromOwnTopics } from '@rssamplifier/discover';
 import { findFeedCard } from '@rssamplifier/feed';
@@ -525,6 +526,20 @@ async function topicsTick() {
 }
 
 // Hourly, and a single small read on the 23 hours with no finished day to add.
+// Every new post on Reddit, matched to the subreddits we hold: what keeps all
+// of /r/* current (see packages/ingest/src/redditFirehose.js). REDDIT_FIREHOSE=0
+// switches it off.
+const redditFirehoseEnabled = process.env.REDDIT_FIREHOSE !== '0';
+const redditFirehoseMs = Number(process.env.REDDIT_FIREHOSE_MS) || 120_000;
+const redditFirehose = createRedditFirehose();
+
+async function redditFirehoseTick() {
+  if (stopping) return;
+  const started = Date.now();
+  const res = await redditFirehose(db);
+  log('reddit-firehose', { ...res, ms: Date.now() - started });
+}
+
 async function trendsTick() {
   if (stopping) return;
   const started = Date.now();
@@ -1243,8 +1258,11 @@ const daemon = buildPollerDaemon({
     sourceIntervalMs,
     topicSearchIntervalMs,
     topicsIntervalMs,
+    redditFirehose: redditFirehoseEnabled,
+    redditFirehoseMs,
   },
   run: {
+    redditFirehose: redditFirehoseTick,
     crawl: tick,
     import: importTick,
     purge: purgeTick,

packages/db/src/queries.js

@@ -3035,6 +3035,30 @@ export async function requestRefresh(db, id) {
   });
 }
 
+/**
+ * Live rows for a set of social refs (`r:sub:rss`, `r:user:spez`), with the
+ * columns a crawl needs: the Reddit firehose matches every new post on the site
+ * against these and crawls only the rows it found.
+ *
+ * @param {Client} db
+ * @param {string[]} refs
+ * @returns {Promise<object[]>}
+ */
+export async function feedsBySocialRefs(db, refs) {
+  const out = [];
+  for (let i = 0; i < refs.length; i += 500) {
+    const chunk = refs.slice(i, i + 500);
+    if (chunk.length === 0) continue;
+    const { rows } = await db.execute({
+      sql: `select ${DUE_COLUMNS} from feeds
+            where social_ref in (${chunk.map(() => '?').join(',')}) and status <> 'dead'`,
+      args: chunk,
+    });
+    out.push(...rows);
+  }
+  return out;
+}
+
 /**
  * `requestRefresh`, but only when the stored copy is older than `maxAgeMinutes`
  * and the feed is not already at the front. For sources whose readers expect a

packages/ingest/index.js

@@ -8,6 +8,7 @@ export {
   refreshFeedKeywords,
 } from './src/crawl.js';
 export { importFeeds, importOpml } from './src/import.js';
+export { createRedditFirehose } from './src/redditFirehose.js';
 export { queueFeeds } from './src/queue.js';
 export { drainImport } from './src/drain.js';
 export {

packages/ingest/src/redditFirehose.js

@@ -0,0 +1,128 @@
+/**
+ * Every subreddit in the directory, current within minutes, from one stream.
+ *
+ * Asking fifty thousand subreddits one at a time every hour is ~14 requests a
+ * second, around the clock, to a volunteer-run mirror, and the per-host crawler
+ * managed ~7 an hour. Reddit as a whole produces ~25 posts a second, and Arctic
+ * Shift will page through all of them, oldest first, a hundred at a time — so
+ * reading *every* new post on Reddit is ~15 requests a minute. Each post is
+ * matched to the rows we hold (`r:sub:<subreddit>` and `r:user:<author>`) and
+ * every matched row goes through `crawlFeed` with its posts already in hand,
+ * so dedupe, keywords, FTS, alerts and scheduling are the crawl's own.
+ *
+ * The cursor lives in memory. A restart starts `lookbackSeconds` back, which is
+ * safe because items dedupe by guid; the per-source crawl and the on-view
+ * refresh cover anything older.
+ */
+
+import { q } from '@rssamplifier/db';
+import { fetchRedditFirehosePage, redditPostToItem, redditResult } from '@rssamplifier/social';
+
+import { crawlFeed } from './crawl.js';
+
+const PAGE = 100;
+
+/**
+ * @param {{
+ *   lookbackSeconds?: number,
+ *   maxPages?: number,
+ *   concurrency?: number,
+ *   env?: Record<string, string|undefined>,
+ *   fetchPage?: typeof fetchRedditFirehosePage,
+ *   crawl?: typeof crawlFeed,
+ *   now?: () => number,
+ * }} [opts]
+ */
+export function createRedditFirehose(opts = {}) {
+  const now = opts.now ?? (() => Date.now());
+  const fetchPage = opts.fetchPage ?? fetchRedditFirehosePage;
+  const crawl = opts.crawl ?? crawlFeed;
+  const maxPages = opts.maxPages ?? 80;
+  const concurrency = opts.concurrency ?? 6;
+
+  /** unix seconds; posts created after this are still to be read */
+  let cursor = Math.floor(now() / 1000) - (opts.lookbackSeconds ?? 900);
+  /** ids already handled near the cursor, since pages overlap by a second */
+  let seen = new Set();
+
+  /**
+   * One pass: read pages until caught up or `maxPages`, then store.
+   *
+   * @param {import('@rssamplifier/db/src/pg.js').PgClient} db
+   */
+  return async function runRedditFirehose(db) {
+    const posts = [];
+    let pages = 0;
+    for (; pages < maxPages; pages += 1) {
+      const page = await fetchPage(cursor, { env: opts.env });
+      for (const post of page) {
+        const id = String(post?.id ?? '');
+        if (!id || seen.has(id)) continue;
+        seen.add(id);
+        posts.push(post);
+      }
+      if (page.length === 0) break;
+
+      const last = Number(page[page.length - 1]?.created_utc) || cursor;
+      // Step back one second so a second split across two pages is read whole;
+      // `seen` drops the overlap. A page that is all one second (100+ posts in
+      // it) must move on, or the loop would ask the same page for ever.
+      cursor = last - 1 > cursor ? last - 1 : last;
+      if (page.length < PAGE) break;
+    }
+    if (seen.size > 20_000) seen = new Set([...seen].slice(-5_000));
+
+    // Group by the refs a row can have.
+    const byRef = new Map();
+    const push = (ref, post) => {
+      const list = byRef.get(ref);
+      if (list) list.push(post);
+      else byRef.set(ref, [post]);
+    };
+    for (const post of posts) {
+      if (post?.subreddit) push(`r:sub:${String(post.subreddit).toLowerCase()}`, post);
+      const author = String(post?.author ?? '');
+      if (author && author !== '[deleted]') push(`r:user:${author.toLowerCase()}`, post);
+    }
+
+    const rows = byRef.size ? await q.feedsBySocialRefs(db, [...byRef.keys()]) : [];
+
+    let crawled = 0;
+    let failed = 0;
+    let items = 0;
+    let next = 0;
+    const worker = async () => {
+      while (next < rows.length) {
+        const row = rows[next++];
+        const ref = String(row.social_ref);
+        const matched = (byRef.get(ref) ?? []).slice().sort((a, b) => b.created_utc - a.created_utc);
+        const mapped = matched.map(redditPostToItem).filter(Boolean);
+        if (mapped.length === 0) continue;
+        const [, mode, name] = /^r:(sub|user):(.+)$/.exec(ref) ?? [];
+        if (!mode) continue;
+        try {
+          const res = await crawl(db, row, {
+            x: async () => redditResult(row, { mode: /** @type {'sub'|'user'} */ (mode), name }, matched, mapped),
+          });
+          if (res?.ok) {
+            crawled += 1;
+            items += Number(res.newItems ?? 0);
+          } else failed += 1;
+        } catch {
+          failed += 1;
+        }
+      }
+    };
+    await Promise.all(Array.from({ length: Math.max(1, Math.min(concurrency, rows.length)) }, worker));
+
+    return {
+      pages,
+      posts: posts.length,
+      matched: rows.length,
+      crawled,
+      failed,
+      items,
+      lagSeconds: Math.max(0, Math.floor(now() / 1000) - cursor),
+    };
+  };
+}

packages/ingest/test/reddit-firehose.test.js

@@ -0,0 +1,79 @@
+import assert from 'node:assert/strict';
+import { test, beforeEach, after } from 'node:test';
+import { q, social } from '@rssamplifier/db';
+import { connectTest } from '@rssamplifier/db/src/testdb.js';
+import { redditSource } from '@rssamplifier/social';
+
+import { createRedditFirehose } from '../src/redditFirehose.js';
+
+let db;
+
+beforeEach(async () => {
+  db?.close();
+  db = await connectTest();
+});
+
+after(async () => {
+  db?.close();
+});
+
+async function seedSub(name) {
+  const source = redditSource(`r/${name}`);
+  const stored = await social.upsertSocialSource(db, {
+    network: 'reddit',
+    ref: source.ref,
+    slug: source.slug,
+    title: source.title,
+    feedUrl: source.feedUrl,
+    siteUrl: source.feedUrl,
+  });
+  return String(stored.id);
+}
+
+const T0 = 1_794_000_000;
+const post = (id, subreddit, at) => ({
+  id,
+  title: `post ${id}`,
+  author: 'someone',
+  subreddit,
+  permalink: `/r/${subreddit}/comments/${id}/x/`,
+  created_utc: at,
+  is_self: true,
+  selftext: 'hello',
+  num_comments: 0,
+  removed_by_category: null,
+});
+
+test('every new post reaches the subreddit row it belongs to, and nothing else is stored', async () => {
+  const rss = await seedSub('rss');
+  const linux = await seedSub('linux');
+
+  // Two pages: a full one (forces a second read) then a short one (caught up).
+  const first = Array.from({ length: 100 }, (_, i) => post(`a${i}`, i % 2 ? 'RSS' : 'notours', T0 + i));
+  const second = [post('b1', 'linux', T0 + 200), post('b2', 'rss', T0 + 201)];
+  const asked = [];
+  const fetchPage = async (after) => {
+    asked.push(after);
+    return asked.length === 1 ? first : asked.length === 2 ? [first[99], ...second] : [];
+  };
+
+  const run = createRedditFirehose({ fetchPage, now: () => (T0 + 300) * 1000, lookbackSeconds: 300 });
+  const res = await run(db);
+
+  assert.equal(asked.length, 2, 'stopped once a page came back short');
+  assert.equal(asked[1], T0 + 98, 'the next page starts a second back so a split second is read whole');
+  assert.equal(res.posts, 102, 'the overlapping post is not counted twice');
+  assert.equal(res.matched, 2);
+
+  const { rows: counted } = await db.execute({ sql: 'select guid from feed_items where feed_id = ?', args: [rss] });
+  const guids = counted.map((r) => String(r.guid));
+  // 51 matched (r/RSS case-insensitively); ITEMS_PER_CRAWL keeps the newest 50,
+  // so only the oldest is dropped. Two-minute passes never meet that cap.
+  assert.equal(guids.length, 50);
+  assert.ok(guids.includes('t3_b2') && !guids.includes('t3_a1'));
+  const linuxItems = await q.itemsForFeed(db, linux, 100);
+  assert.deepEqual(linuxItems.map((r) => r.guid), ['t3_b1']);
+
+  const [row] = (await db.execute({ sql: 'select last_success_at from feeds where id = ?', args: [rss] })).rows;
+  assert.ok(row.last_success_at, 'the row is marked read');
+});

packages/social/index.js

@@ -97,6 +97,7 @@ export {
   arcticGet,
   commentTree,
   fetchRedditComments,
+  fetchRedditFirehosePage,
   fetchRedditListing,
   fetchRedditPost,
   isShowable as isShowableRedditPost,
@@ -105,4 +106,4 @@ export {
   redditPostId,
 } from './src/reddit/arctic.js';
 
-export { fetchRedditSource } from './src/reddit/fetch.js';
+export { fetchRedditSource, redditResult } from './src/reddit/fetch.js';

packages/social/src/reddit/arctic.js

@@ -97,6 +97,22 @@ export async function fetchRedditListing(spec, opts = {}) {
   return Array.isArray(data) ? data : [];
 }
 
+/**
+ * One page of every new post on Reddit, oldest first, created after `after`
+ * (unix seconds). The firehose: about 25 posts a second across the whole site,
+ * so a hundred-post page per ~4 seconds of Reddit keeps every subreddit we hold
+ * current for ~15 requests a minute, against ~14 a second to ask fifty
+ * thousand subreddits one by one every hour.
+ *
+ * @param {number} after unix seconds
+ * @param {Parameters<typeof arcticGet>[2]} [opts]
+ * @returns {Promise<object[]>}
+ */
+export async function fetchRedditFirehosePage(after, opts = {}) {
+  const data = await arcticGet('/api/posts/search', { after: Math.floor(after), sort: 'asc', limit: 100 }, opts);
+  return Array.isArray(data) ? data : [];
+}
+
 /**
  * One post by id, or null when the archive has never seen it.
  *
@@ -360,7 +376,16 @@ function inline(raw) {
     .replace(/(^|[^*])\*([^*\s][^*]*)\*/g, '$1<em>$2</em>')
     .replace(/~~([^~]+)~~/g, '<del>$1</del>');
 
-  return s.replace(/\u0000(\d+)\u0000/g, (_, n) => tokens[Number(n)]);
+  // Tokens nest (a code span inside a link label is kept inside the link's
+  // token), so restore until none are left. One pass leaked the inner
+  // placeholder, NULs and all, and Postgres refused the whole subreddit's write
+  // ("invalid byte sequence for encoding UTF8: 0x00") on every crawl.
+  const PLACEHOLDER = /\u0000(\d+)\u0000/g;
+  for (let pass = 0; pass < 8 && PLACEHOLDER.test(s); pass += 1) {
+    PLACEHOLDER.lastIndex = 0;
+    s = s.replace(PLACEHOLDER, (_, n) => tokens[Number(n)] ?? '');
+  }
+  return s.replace(/\u0000/g, '');
 }
 
 function escapeHtml(s) {