diff --git a/.env.example b/.env.example index a43449a..a72eade 100644 --- a/.env.example +++ b/.env.example @@ -46,3 +46,44 @@ CARD_BACKFILL_SECONDS=20 # Nonessential maintenance can be paused while a large first-crawl backlog owns # the database write path. The work is resumable when these are switched on. CLUSTER_BACKFILL=1 + +# --------------------------------------------------------------------- X / Twitter +# X publishes no feeds, so posts are collected through a provider and mirrored +# at /x/. Off by default: with this unset nothing is collected, existing +# X feeds keep serving what they hold, and no source is marked unhealthy for it. +X_ENABLED=false + +# Failover order. The provider never appears in a public URL, so this can change +# under a live subscriber without their reader noticing. +X_PRIMARY_PROVIDER=rsshub +X_FALLBACK_PROVIDERS=teapot,official + +# Self-hosted, alongside the app, and not exposed publicly. A provider whose +# base URL is unset is skipped rather than guessed at. +RSSHUB_BASE_URL= +RSSHUB_ACCESS_KEY= +TEAPOT_BASE_URL= + +# The official API: the only provider that costs money per request, hence the +# caps. 0 means unlimited. +X_API_BEARER_TOKEN= +X_API_DAILY_READS=0 +X_API_MONTHLY_READS=0 +X_API_MAX_RPM=0 + +# Logged-in X sessions for the unofficial providers, as JSON: +# [{"id":"x-1","authToken":"...","ct0":"..."}] +# +# These are a full login to an X account - whoever holds them can post as it and +# change its password. They live here rather than in a table so that a leaked +# database dump carries none of them; x_sessions holds health state only. Use +# dedicated accounts, and separate ones for production and development. +# +# The positional pair X_AUTH_TOKENS / X_CT0_TOKENS also works and is a trap: +# the lists are matched by index, so removing one dead account from the middle +# of the first and forgetting the second pairs every later token with the wrong +# cookie. Prefer the JSON form, which cannot express that. +X_SESSIONS= + +X_FETCH_TIMEOUT_MS=15000 +X_SESSION_COOLDOWN_SECONDS=900 diff --git a/README.md b/README.md index 1290c85..783c24f 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ packages/feed/ Feed discovery, RSS/Atom/JSON Feed parsing, OPML, SSRF guards packages/ingest/ Submit + crawl orchestration packages/db/ Turso/libSQL client, migrations and every query packages/notify/ Alerts — web push, email digests and webhooks +packages/social/ X and Reddit: canonical identity, and X's provider adapters ``` Everything outside the Next app is plain ESM with JSDoc types — no build step, so Docker stays @@ -56,6 +57,98 @@ pnpm --filter @rssamplifier/db migrate | `/discoveries/{id}` | Progress of one keyword run: what was added, and why the rest was not | | `/crawlstats` | Crawler and discovery queues, live (`/crawlstatus` redirects here) | | `/random` | Redirect to a random blog — the toolbar's ✦ | +| `/r` | Every subreddit and Reddit user in the directory | +| `/r/` | One community: `/r/programming`, `.rss` `.atom` `.json` `.md` | +| `/r/u/` | One Reddit user, under the same prefix | +| `/x` | Every X account, search and list in the directory | +| `/x/` | One timeline: `/x/OpenAI`, plus `/replies` and `/media` | +| `/x/list/` | One X list | +| `/x/search?q=` | An X search as a feed — X's own operators pass through | +| `/x/status` | Which provider is collecting X, and how it is doing | + +### The two social namespaces + +Reddit and X both live under a prefix of their own, and for the same reason from +opposite directions. Reddit publishes real RSS, so a subreddit resolves down the +ordinary path and lands as an untyped row at a slug of its own — which is how +50,099 of them ended up filed among the blogs. X publishes nothing at all, so +without a provider it is not submittable in the first place. + +`packages/social` answers one question for both: **what is the canonical identity +of this thing?** `@OpenAI`, `x.com/OpenAI` and `https://twitter.com/openai/` are +one source (`x:user:openai`, at `/x/OpenAI`); `/r/programming`, `/r/Programming/` +and `/r/programming/new/.rss` are one community (`r:sub:programming`, at +`/r/programming`). One identity means one row, which means **one polling job no +matter how many people subscribe** — the thing that matters most when the +upstream rate-limits per account. + +A social source is an ordinary row in `feeds`, not a parallel `sources` table. +That is what lets it inherit dedupe, backoff, interval learning, keyword +extraction, full-text search, alerts, sitemaps and all five syndication formats +without a line of X-specific code in any of them — and it is why an X post can +appear in `/topics/artificial-intelligence.rss` beside a blog post with nothing +to tell them apart. + +`/{slug}` still answers for both, and always will: it is the permanent identity +of a row and links already point at it. The `/r/` and `/x/` address is the +canonical one, which is what search engines are told. + +## Collecting X + +X has no feeds, so posts are collected through a provider and mirrored here. +**Which provider never appears in a public URL** — that is the whole design. +A reader subscribed to `/x/OpenAI.rss` through RSSHub is still subscribed when +the collection method is replaced under them. + +``` +RSSHub (primary) → Teapot (fallback) → official X API (paid) → cached items +``` + +Failover is per attempt and the order is fixed; three failures in a row set a +provider aside for a few minutes and one success clears it. Nothing here can +empty a feed: items are only ever written on success, so an outage leaves +yesterday's posts exactly where they were and the public route goes on serving +them. + +| Variable | Default | What | +| --- | --- | --- | +| `X_ENABLED` | `false` | The kill switch. Off means no collection; existing feeds keep serving. | +| `X_PRIMARY_PROVIDER` | `rsshub` | First provider tried. | +| `X_FALLBACK_PROVIDERS` | `teapot,official` | Tried in order after it. | +| `RSSHUB_BASE_URL` | — | A self-hosted RSSHub. Unset means the provider is skipped. | +| `RSSHUB_ACCESS_KEY` | — | If that instance requires one. | +| `TEAPOT_BASE_URL` | — | A Teapot or Nitter-shaped instance. | +| `X_API_BEARER_TOKEN` | — | Official API. Unset means that provider is skipped. | +| `X_API_DAILY_READS` | `0` | Spend cap, `0` for unlimited. Also `X_API_MONTHLY_READS`, `X_API_MAX_RPM`. | +| `X_SESSIONS` | — | JSON: `[{"id":"x-1","authToken":"…","ct0":"…"}]` | +| `X_FETCH_TIMEOUT_MS` | `15000` | Upstream deadline. | +| `X_SESSION_COOLDOWN_SECONDS` | `900` | How long a rate-limited session rests. | + +**Session credentials live in the environment and never in a table.** +`auth_token` and `ct0` are a full login to an X account — whoever holds them can +post as it and change its password. `x_sessions` and `x_provider_state` hold +health state only, so a leaked database dump carries no credentials, and nothing +renderable about a session is one. Use dedicated accounts, and separate ones for +production and development. + +§47 of the PRD also allows the positional pair `X_AUTH_TOKENS` / `X_CT0_TOKENS`. +It works, and it is a trap worth naming: the lists are matched by position, so +deleting one dead account from the middle of the first and forgetting the second +silently pairs every later token with the wrong cookie — a pool of sessions that +all authenticate as nobody. `X_SESSIONS` cannot express that state. + +Both unofficial providers keep their own logged-in sessions by default, so the +pool has nothing to hand them unless the deployment exposes a per-request +parameter (`RSSHUB_SESSION_PARAM`, `TEAPOT_SESSION_HEADER`). Nothing here guesses +at those names: guessing would mean putting a live cookie on a query string +somebody else's instance might log. + +**Not built:** the `/admin/x` buttons of §34 — disable a provider, clear a +cooldown, force a refresh. This codebase has no notion of an administrator to +guard them with, and shipping the levers before the lock is how a kill switch +becomes a way for anyone to turn collection off. `X_ENABLED` and +`X_PRIMARY_PROVIDER` cover the two that matter without a code deploy. +`/x/status` is the read-only half, and is where those buttons go. ## Agent endpoints diff --git a/apps/poller/package.json b/apps/poller/package.json index 4862ab5..10a658b 100644 --- a/apps/poller/package.json +++ b/apps/poller/package.json @@ -12,6 +12,7 @@ "@rssamplifier/discover": "workspace:*", "@rssamplifier/feed": "workspace:*", "@rssamplifier/ingest": "workspace:*", - "@rssamplifier/notify": "workspace:*" + "@rssamplifier/notify": "workspace:*", + "@rssamplifier/social": "workspace:*" } } diff --git a/apps/poller/src/index.js b/apps/poller/src/index.js index 161a211..884fbff 100644 --- a/apps/poller/src/index.js +++ b/apps/poller/src/index.js @@ -5,6 +5,7 @@ import { q, accounts, alerts, + social, takeWriteTally, warmStatsCache, warmDirectoryCache, @@ -24,6 +25,7 @@ import { import { runDueSources, discoverFromOwnTopics } from '@rssamplifier/discover'; import { findFeedCard } from '@rssamplifier/feed'; import { deliverAlerts, vapidConfig } from '@rssamplifier/notify'; +import { createXRuntime, xEnabled } from '@rssamplifier/social'; import { createRecorder, toEntry, writeFailure } from './log.js'; @@ -294,7 +296,12 @@ async function tick() { batchSize, concurrency, publishLog ? recorder.record : null, - { perHost }, + // `crawl` is handed to crawlFeed as-is. The X runtime travels here rather + // than being built per feed because it is the thing that *remembers*: + // which provider is in cooldown, which session is resting. Rebuilt per + // crawl it would be a system with no memory, rediscovering the same + // outage on every source in the batch. + { perHost, crawl: { xRuntime } }, ); if (crawled || failed) { // The backlog is the number worth watching: crawled/failed only say the @@ -578,6 +585,42 @@ try { process.exit(1); } +/** + * The X collection runtime, built once and shared by every crawl in this process. + * + * Built after the migration, because `hydrate()` reads `x_provider_state` and + * `x_sessions` — the tables that migration creates — to restore cooldowns + * across a redeploy. Without that a service that redeploys ten times in a day + * forgets ten outages and re-walks into each of them. + * + * `null` when X is switched off, and that is a first-class state rather than a + * failure: `crawlFeed` sees no runtime and reschedules its X sources without + * touching their health, so existing feeds keep serving what they hold and + * nothing is retired while the integration is off (§42's kill switch). + * + * Logged either way. "The X sources stopped updating" is the kind of thing that + * goes unnoticed for a week, and a boot line saying `x=false` is what turns + * that into a five-second answer — the same reason the push half of the alerts + * stack prints `push=true` here. + */ +const xRuntime = xEnabled(env) + ? await createXRuntime({ + env, + providerStore: social.providerStore(db), + sessionStore: social.sessionStore(db), + // Straight onto the same live log the crawl writes to, so a failover or a + // session cooldown appears on /crawlstats beside the crawl it affected + // rather than in a stream nobody has open (§35). + onEvent: publishLog ? (event, fields) => recorder.record(toEntry(event, fields)) : null, + }) + : null; + +log('x-runtime', { + enabled: Boolean(xRuntime), + providers: xRuntime ? xRuntime.registry.candidates().map((p) => p.name) : [], + sessions: xRuntime ? xRuntime.sessions.size : 0, +}); + /** * Key the items stored before the grouping column existed. * diff --git a/apps/web/next.config.mjs b/apps/web/next.config.mjs index d5caa50..a7f31c7 100644 --- a/apps/web/next.config.mjs +++ b/apps/web/next.config.mjs @@ -206,6 +206,43 @@ const nextConfig = { // The directory itself: what was added to it, newest first. { source: '/feed.:format(rss|atom|json|xml|md)', destination: '/api/directory/feed/:format' }, + // The two social namespaces. + // + // No playlist spellings, for the same reason /following has none: a + // timeline and a subreddit carry no enclosures, so an `.m3u` of one + // would be an empty file with a confident name. + // + // Ordered narrowest-first within each prefix, because `:username` and + // `:subreddit` match anything: every fixed address under /x has to be + // named before /x/:username, and /r/u before /r/:subreddit. A rewrite + // parameter never spans a slash, so the two-segment rules cannot be + // shadowed by the one-segment ones — but the fixed segments can be, and + // silently. + { + source: '/x/search.:format(rss|atom|json|xml|md)', + destination: '/api/x/search/feed/:format', + }, + { + source: '/x/list/:listId.:format(rss|atom|json|xml|md)', + destination: '/api/x/list/:listId/feed/:format', + }, + { + source: '/x/:username/:mode(replies|media).:format(rss|atom|json|xml|md)', + destination: '/api/x/:username/:mode/feed/:format', + }, + { + source: '/x/:username.:format(rss|atom|json|xml|md)', + destination: '/api/x/:username/feed/:format', + }, + { + source: '/r/u/:username.:format(rss|atom|json|xml|md)', + destination: '/api/r/u/:username/feed/:format', + }, + { + source: '/r/:subreddit.:format(rss|atom|json|xml|md)', + destination: '/api/r/:subreddit/feed/:format', + }, + // One category of it. The segments are the category pages' own paths, // duplicated from CATEGORIES in apps/web/src/lib/categories.js — this // file is evaluated before the workspace resolves, so it cannot import diff --git a/apps/web/package.json b/apps/web/package.json index 5e6136a..4bc53e3 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -19,6 +19,7 @@ "@rssamplifier/mail": "workspace:*", "@rssamplifier/notify": "workspace:*", "@rssamplifier/search": "workspace:*", + "@rssamplifier/social": "workspace:*", "@rssamplifier/translate": "workspace:*", "@simplewebauthn/browser": "^13.0.0", "@swc/helpers": "^0.5.23", diff --git a/apps/web/src/app/AddSocialSource.jsx b/apps/web/src/app/AddSocialSource.jsx new file mode 100644 index 0000000..e634e12 --- /dev/null +++ b/apps/web/src/app/AddSocialSource.jsx @@ -0,0 +1,68 @@ +import { siteUrl } from '../lib/db.js'; + +/** + * What `/r/somewhere` or `/x/somebody` shows when nobody has added it yet. + * + * A 404 would be the easy answer and the wrong one. The address is well formed, + * the thing at the other end almost certainly exists, and the visitor has + * already told us exactly what they want by typing it — so the page offers to + * add it rather than telling them they were wrong to ask. + * + * A plain `
` to `/api/submit`, like every other control on + * this site: it works with JavaScript off, and the endpoint answers an HTML + * caller with a 303 back to the source's own page. Nothing is fetched from X or + * Reddit while the visitor waits — the row is written, the poller collects on + * its next tick, and this page is replaced by the real one within the minute + * (§17, §37). + * + * @param {{ network: 'x'|'reddit', label: string, input: string, canonical: string }} props + * `input` is what gets submitted — the canonical upstream URL, not what was + * typed, so the source that gets created is the one this page is about. + */ +export default function AddSocialSource({ network, label, input, canonical }) { + const platform = network === 'x' ? 'X' : 'Reddit'; + + return ( +
+

{label}

+ +

+ Nobody has added this {platform} source to the directory yet. Add it and RSS Amplifier + will start collecting it — usually within a minute. +

+ + + + + + +

+ Once it is here, it will be at{' '} + + {siteUrl()} + {canonical} + {' '} + in every format this site publishes:{' '} + .rss, .atom, .json and .md. That + address does not change, whatever we have to do behind it to keep collecting. +

+ + {network === 'x' ? ( +

+ X publishes no feeds of its own, so this is collected on your behalf and mirrored here. + Protected accounts are not collected, and posts arrive as fast as we can read them + rather than in real time. +

+ ) : ( +

+ Reddit publishes its own feed for this, and we read it on a schedule and keep a copy — + so the address above works whether or not Reddit is answering right now. +

+ )} + +

+ Browse what is already here +

+
+ ); +} diff --git a/apps/web/src/app/SocialIndex.jsx b/apps/web/src/app/SocialIndex.jsx new file mode 100644 index 0000000..bed0d1f --- /dev/null +++ b/apps/web/src/app/SocialIndex.jsx @@ -0,0 +1,138 @@ +import { social } from '@rssamplifier/db'; +import { socialPathFor } from '@rssamplifier/social'; + +import { db, siteUrl } from '../lib/db.js'; +import ListFilter from './ListFilter.jsx'; +import { FILTER_FROM } from '../lib/listFilter.js'; + +/** + * The index of one network's sources — `/r` and `/x`. + * + * One component over a table of two labels, the way `/blogs` and `/podcasts` + * share `CategoryIndex`: two copies of a listing drift the moment one grows a + * feature, and this one is going to grow at least a sort. + * + * **It reports two numbers, not one, and that is the honest part.** The + * subreddit import put 50,099 communities into the directory and the crawler + * has read a fraction of them, so a page saying "50,099 subreddits" would be + * promising a directory that mostly does not exist yet. Saying how many have + * actually been read is the same distinction the MCP server draws with + * `freshness`, and for the same reason: a row being present is not evidence + * that anything is behind it. + */ + +/** How many sources a page of this listing holds. */ +const PER_PAGE = 100; + +/** + * @param {{ network: 'x'|'reddit', page?: number }} props + */ +export default async function SocialIndex({ network, page = 1 }) { + const client = db(); + const offset = (Math.max(1, page) - 1) * PER_PAGE; + + const [rows, counts] = await Promise.all([ + social.listSocialFeeds(client, network, { limit: PER_PAGE, offset }), + social.countSocialFeeds(client, network), + ]); + + const platform = network === 'x' ? 'X' : 'Reddit'; + const noun = network === 'x' ? 'accounts, searches and lists' : 'communities and users'; + const base = network === 'x' ? '/x' : '/r'; + + return ( +
+

{platform}

+ +

+ {counts.total.toLocaleString()} {platform} {noun} in the directory,{' '} + {counts.crawled.toLocaleString()} of which we have read at least once. Every one has a + page here and a feed in four formats, at an address that does not change. +

+ + {network === 'x' ? ( +

+ X publishes no feeds, so these are collected on your behalf and mirrored here — the + posts you read come out of this directory, never out of X.{' '} + Turn a search into a feed, or{' '} + see how collection is going. +

+ ) : ( +

+ Reddit publishes a feed for every community. We read them on a schedule and keep a + copy, so these addresses work whether or not Reddit is answering right now. +

+ )} + +
+ + + +
+ + {rows.length >= FILTER_FROM ? ( + + ) : null} + + {rows.length === 0 ? ( +

Nothing here yet. Add the first one above.

+ ) : ( + + )} + + {rows.length === PER_PAGE ? ( +

+ Next page +

+ ) : null} + +

+ Everything here also feeds the topic pages, mixed in with blogs, podcasts and the rest — + see topics. The directory’s own river is at{' '} + {siteUrl()}/feed.rss. +

+
+ ); +} + +/** + * `?page=` as a number, defaulting to the first. + * + * @param {unknown} raw + * @returns {number} + */ +export function pageNumber(raw) { + const parsed = Number(Array.isArray(raw) ? raw[0] : raw); + return Number.isFinite(parsed) && parsed >= 1 ? Math.floor(parsed) : 1; +} diff --git a/apps/web/src/app/api/r/[subreddit]/feed/[format]/route.js b/apps/web/src/app/api/r/[subreddit]/feed/[format]/route.js new file mode 100644 index 0000000..96292cf --- /dev/null +++ b/apps/web/src/app/api/r/[subreddit]/feed/[format]/route.js @@ -0,0 +1,36 @@ +import { riverFail } from '../../../../../../lib/river.js'; +import { redditTarget, socialRiver } from '../../../../../../lib/socialRiver.js'; + +export const dynamic = 'force-dynamic'; + +/** + * One subreddit, as a feed of ours. `/r/programming.rss` rewrites here. + * + * @param {Request} req + * @param {{ params: Promise<{ subreddit: string, format: string }> }} ctx + */ +export async function GET(req, { params }) { + const { subreddit, format } = await params; + const url = new URL(req.url); + + const target = redditTarget({ subreddit }); + // 400 rather than 404 (§41): the name cannot be a subreddit at all, so there + // is nothing to look for and nothing that adding it would fix. + if (!target) { + return riverFail( + format, + 400, + `not a subreddit: r/${subreddit}`, + 'Subreddit names are 3-21 characters of A-Z, 0-9 and underscore.', + ); + } + + return socialRiver({ + ref: target.ref, + canonical: target.canonical, + label: target.label, + format, + limit: url.searchParams.get('limit'), + req, + }); +} diff --git a/apps/web/src/app/api/r/u/[username]/feed/[format]/route.js b/apps/web/src/app/api/r/u/[username]/feed/[format]/route.js new file mode 100644 index 0000000..3caac18 --- /dev/null +++ b/apps/web/src/app/api/r/u/[username]/feed/[format]/route.js @@ -0,0 +1,37 @@ +import { riverFail } from '../../../../../../../lib/river.js'; +import { redditTarget, socialRiver } from '../../../../../../../lib/socialRiver.js'; + +export const dynamic = 'force-dynamic'; + +/** + * One Reddit user's posts. `/r/u/spez.rss` rewrites here. + * + * Under `/r/` rather than at a `/u/` of its own, so that one prefix holds all + * of Reddit — which is the whole point of having the namespace. + * + * @param {Request} req + * @param {{ params: Promise<{ username: string, format: string }> }} ctx + */ +export async function GET(req, { params }) { + const { username, format } = await params; + const url = new URL(req.url); + + const target = redditTarget({ username }); + if (!target) { + return riverFail( + format, + 400, + `not a Reddit username: ${username}`, + 'Usernames are 3-20 characters of A-Z, 0-9, underscore and hyphen.', + ); + } + + return socialRiver({ + ref: target.ref, + canonical: target.canonical, + label: target.label, + format, + limit: url.searchParams.get('limit'), + req, + }); +} diff --git a/apps/web/src/app/api/submit/route.js b/apps/web/src/app/api/submit/route.js index a130480..596f12d 100644 --- a/apps/web/src/app/api/submit/route.js +++ b/apps/web/src/app/api/submit/route.js @@ -341,7 +341,11 @@ export async function POST(req) { } const first = result?.accepted?.[0]; - const location = first ? `/${first.slug}` : '/submit?error=1'; + // The address the source lives at, which is `/{slug}` for a feed and + // `/r/programming` or `/x/OpenAI` for a social one. Both render the same + // page; sending somebody to the slug would land them at the address the + // namespace exists to replace. + const location = first ? (first.path ?? `/${first.slug}`) : '/submit?error=1'; return new Response(null, { status: 303, headers: { location } }); } finally { diff --git a/apps/web/src/app/api/x/[username]/[mode]/feed/[format]/route.js b/apps/web/src/app/api/x/[username]/[mode]/feed/[format]/route.js new file mode 100644 index 0000000..8ecea3f --- /dev/null +++ b/apps/web/src/app/api/x/[username]/[mode]/feed/[format]/route.js @@ -0,0 +1,46 @@ +import { riverFail } from '../../../../../../../lib/river.js'; +import { socialRiver, xTarget } from '../../../../../../../lib/socialRiver.js'; + +export const dynamic = 'force-dynamic'; + +/** The two tabs of an account that are feeds in their own right (§25, §7). */ +const MODES = new Set(['replies', 'media']); + +/** + * `/x/OpenAI/replies.rss` and `/x/OpenAI/media.rss`. + * + * A different source rather than a filter on the timeline: each has its own + * canonical ref and therefore its own row, which is what lets a reader + * subscribe to an account's posts and its replies at the same time without + * either one having to carry the other's items. + * + * @param {Request} req + * @param {{ params: Promise<{ username: string, mode: string, format: string }> }} ctx + */ +export async function GET(req, { params }) { + const { username, mode, format } = await params; + const url = new URL(req.url); + + if (!MODES.has(mode)) { + return riverFail(format, 404, `no such X feed: ${mode}`, 'Try /replies or /media.'); + } + + const target = xTarget({ username, mode }); + if (!target) { + return riverFail( + format, + 400, + `not an X handle: ${username}`, + 'Handles are 1-15 characters of A-Z, 0-9 and underscore.', + ); + } + + return socialRiver({ + ref: target.ref, + canonical: target.canonical, + label: target.label, + format, + limit: url.searchParams.get('limit'), + req, + }); +} diff --git a/apps/web/src/app/api/x/[username]/feed/[format]/route.js b/apps/web/src/app/api/x/[username]/feed/[format]/route.js new file mode 100644 index 0000000..44483eb --- /dev/null +++ b/apps/web/src/app/api/x/[username]/feed/[format]/route.js @@ -0,0 +1,39 @@ +import { riverFail } from '../../../../../../lib/river.js'; +import { socialRiver, xTarget } from '../../../../../../lib/socialRiver.js'; + +export const dynamic = 'force-dynamic'; + +/** + * One X account's timeline, as a feed of ours. `/x/OpenAI.rss` rewrites here. + * + * The address AC-2 is about. Which provider collected these posts — RSSHub, + * Teapot, the official API — appears nowhere in the URL, the document or the + * headers, so the collection method can be replaced under a live subscriber + * without their reader noticing. + * + * @param {Request} req + * @param {{ params: Promise<{ username: string, format: string }> }} ctx + */ +export async function GET(req, { params }) { + const { username, format } = await params; + const url = new URL(req.url); + + const target = xTarget({ username }); + if (!target) { + return riverFail( + format, + 400, + `not an X handle: ${username}`, + 'Handles are 1-15 characters of A-Z, 0-9 and underscore.', + ); + } + + return socialRiver({ + ref: target.ref, + canonical: target.canonical, + label: target.label, + format, + limit: url.searchParams.get('limit'), + req, + }); +} diff --git a/apps/web/src/app/api/x/list/[listId]/feed/[format]/route.js b/apps/web/src/app/api/x/list/[listId]/feed/[format]/route.js new file mode 100644 index 0000000..0f7896d --- /dev/null +++ b/apps/web/src/app/api/x/list/[listId]/feed/[format]/route.js @@ -0,0 +1,34 @@ +import { riverFail } from '../../../../../../../lib/river.js'; +import { socialRiver, xTarget } from '../../../../../../../lib/socialRiver.js'; + +export const dynamic = 'force-dynamic'; + +/** + * An X list, by id. `/x/list/123456789.rss` rewrites here. + * + * Only the numeric id, never `/:owner/:slug` — resolving a list's slug to its + * id means asking X, which is a request we would be making on behalf of an + * anonymous caller who has not yet subscribed to anything. §29 leaves that + * alias to a later phase for exactly that reason. + * + * @param {Request} req + * @param {{ params: Promise<{ listId: string, format: string }> }} ctx + */ +export async function GET(req, { params }) { + const { listId, format } = await params; + const url = new URL(req.url); + + const target = xTarget({ listId }); + if (!target) { + return riverFail(format, 400, `not an X list id: ${listId}`, 'A list id is 6-25 digits.'); + } + + return socialRiver({ + ref: target.ref, + canonical: target.canonical, + label: target.label, + format, + limit: url.searchParams.get('limit'), + req, + }); +} diff --git a/apps/web/src/app/api/x/search/feed/[format]/route.js b/apps/web/src/app/api/x/search/feed/[format]/route.js new file mode 100644 index 0000000..b83bc28 --- /dev/null +++ b/apps/web/src/app/api/x/search/feed/[format]/route.js @@ -0,0 +1,46 @@ +import { riverFail } from '../../../../../../lib/river.js'; +import { socialRiver, xTarget } from '../../../../../../lib/socialRiver.js'; + +export const dynamic = 'force-dynamic'; + +/** + * An X search. `/x/search.rss?q=bitcoin` rewrites here. + * + * The query stays in the query string rather than becoming a path segment, and + * that is the one design decision in this file. §5 shows a slugged form + * (`/x/search/artificial-intelligence.rss`) and §28 the query-string one; only + * the second can carry `from:OpenAI lang:en` without inventing an escaping + * scheme, and a subscription URL is a bad place to invent one. X's operator + * syntax is passed through whole and never reimplemented. + * + * The `?q=` survives the rewrite because a route handler's `req.url` is the URL + * the *client* asked for — the same reason `/following.rss?t=` works and a + * destination query string would not. See the note in next.config.mjs. + * + * @param {Request} req + * @param {{ params: Promise<{ format: string }> }} ctx + */ +export async function GET(req, { params }) { + const { format } = await params; + const url = new URL(req.url); + const query = (url.searchParams.get('q') ?? '').trim(); + + if (!query) { + return riverFail(format, 400, 'no query', 'Ask for /x/search.rss?q=your+search'); + } + + const target = xTarget({ query }); + if (!target) { + return riverFail(format, 400, `not a usable X search: ${query}`, 'Try a shorter query.'); + } + + return socialRiver({ + ref: target.ref, + canonical: target.canonical, + label: target.label, + query: target.query, + format, + limit: url.searchParams.get('limit'), + req, + }); +} diff --git a/apps/web/src/app/layout.jsx b/apps/web/src/app/layout.jsx index 9de37e5..d33bb4a 100644 --- a/apps/web/src/app/layout.jsx +++ b/apps/web/src/app/layout.jsx @@ -291,6 +291,15 @@ export default function RootLayout({ children }) { which is where somebody wonders who wrote the thing. */} Authors

+ {/* The two platform namespaces, on a line of their own rather than + appended to the browse row above. That row is the directory's + own categories — what a feed *is* — and these are two places + feeds come from, which is a different question. Keeping them + apart also stops the row growing a third arm every time a + platform is added. */} +

+ Platforms: Reddit · X +

Machine-readable: MCP server · CLI ·{' '} JSON API · OPML ·{' '} diff --git a/apps/web/src/app/r/[subreddit]/page.jsx b/apps/web/src/app/r/[subreddit]/page.jsx new file mode 100644 index 0000000..9c4248c --- /dev/null +++ b/apps/web/src/app/r/[subreddit]/page.jsx @@ -0,0 +1,64 @@ +import { notFound } from 'next/navigation'; +import { redditSource } from '@rssamplifier/social'; + +import { socialFeed, socialMetadata } from '../../../lib/socialPage.js'; +import AddSocialSource from '../../AddSocialSource.jsx'; +import FeedPage from '../../[slug]/page.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * A subreddit at the address people already know how to type. + * + * The page is `/{slug}`'s — literally, the same component with the same props — + * because a subreddit in this directory is a feed like any other and giving it + * a second, parallel page would be two things to keep in step for no gain. What + * `/r/` adds is the name: the canonical URL, the feed addresses, and a place + * for a community that is not in the directory yet to be added from. + * + * Rendering the component rather than redirecting to it is deliberate. A + * redirect would make `/{slug}` the address a reader ends up on and bookmarks, + * which is the opposite of the intent — see `socialPage.js` for how the two + * addresses are told apart without either breaking. + * + * @param {{ params: Promise<{ subreddit: string }> }} props + */ +export async function generateMetadata({ params }) { + const { subreddit } = await params; + const source = redditSource(`r/${subreddit}`); + if (!source) return { title: 'Not found', robots: { index: false, follow: false } }; + + return socialMetadata({ + feed: await socialFeed(source.ref), + canonical: source.path, + label: source.title, + network: 'reddit', + }); +} + +/** + * @param {{ params: Promise<{ subreddit: string }> }} props + */ +export default async function SubredditPage({ params }) { + const { subreddit } = await params; + + const source = redditSource(`r/${subreddit}`); + // Not a subreddit name at all. The other miss below — a real name nobody has + // added — gets an offer to add it; this one gets a 404, because there is + // nothing at the other end to add and a form here would submit nothing. + if (!source) notFound(); + + const feed = await socialFeed(source.ref); + if (!feed) { + return ( + + ); + } + + return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }) }); +} diff --git a/apps/web/src/app/r/page.jsx b/apps/web/src/app/r/page.jsx new file mode 100644 index 0000000..aa4b060 --- /dev/null +++ b/apps/web/src/app/r/page.jsx @@ -0,0 +1,27 @@ +import { siteUrl } from '../../lib/db.js'; +import SocialIndex, { pageNumber } from '../SocialIndex.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * @param {{ searchParams: Promise> }} props + */ +export async function generateMetadata({ searchParams }) { + const page = pageNumber((await searchParams).page); + + return { + title: page === 1 ? 'Reddit' : `Reddit · page ${page}`, + description: + 'Subreddits and Reddit users in the RSS Amplifier directory, each with a page and a feed at an address that does not change.', + alternates: { + canonical: page === 1 ? `${siteUrl()}/r` : `${siteUrl()}/r?page=${page}`, + }, + }; +} + +/** + * @param {{ searchParams: Promise> }} props + */ +export default async function RedditIndexPage({ searchParams }) { + return ; +} diff --git a/apps/web/src/app/r/u/[username]/page.jsx b/apps/web/src/app/r/u/[username]/page.jsx new file mode 100644 index 0000000..2379404 --- /dev/null +++ b/apps/web/src/app/r/u/[username]/page.jsx @@ -0,0 +1,55 @@ +import { notFound } from 'next/navigation'; +import { redditSource } from '@rssamplifier/social'; + +import { socialFeed, socialMetadata } from '../../../../lib/socialPage.js'; +import AddSocialSource from '../../../AddSocialSource.jsx'; +import FeedPage from '../../../[slug]/page.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * One Reddit user, at `/r/u/spez`. + * + * Under `/r/` rather than at `/u/`, so one prefix holds all of Reddit. It also + * keeps `/u/` free, which matters more than it sounds: `/{slug}` is the + * catch-all at the root of this site, and every prefix claimed is a slug taken + * away from the directory. + * + * @param {{ params: Promise<{ username: string }> }} props + */ +export async function generateMetadata({ params }) { + const { username } = await params; + const source = redditSource(`u/${username}`); + if (!source) return { title: 'Not found', robots: { index: false, follow: false } }; + + return socialMetadata({ + feed: await socialFeed(source.ref), + canonical: source.path, + label: source.title, + network: 'reddit', + }); +} + +/** + * @param {{ params: Promise<{ username: string }> }} props + */ +export default async function RedditUserPage({ params }) { + const { username } = await params; + + const source = redditSource(`u/${username}`); + if (!source) notFound(); + + const feed = await socialFeed(source.ref); + if (!feed) { + return ( + + ); + } + + return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }) }); +} diff --git a/apps/web/src/app/x/[username]/[mode]/page.jsx b/apps/web/src/app/x/[username]/[mode]/page.jsx new file mode 100644 index 0000000..1e1fcae --- /dev/null +++ b/apps/web/src/app/x/[username]/[mode]/page.jsx @@ -0,0 +1,56 @@ +import { notFound } from 'next/navigation'; +import { xSource } from '@rssamplifier/social'; + +import { socialFeed, socialMetadata } from '../../../../lib/socialPage.js'; +import AddSocialSource from '../../../AddSocialSource.jsx'; +import FeedPage from '../../../[slug]/page.jsx'; + +export const dynamic = 'force-dynamic'; + +/** The tabs that are feeds of their own: `/x/OpenAI/replies`, `/x/OpenAI/media`. */ +const MODES = { replies: 'with_replies', media: 'media' }; + +/** + * @param {{ username: string, mode: string }} params + */ +function sourceFor({ username, mode }) { + const tab = MODES[mode]; + return tab ? xSource(`https://x.com/${username}/${tab}`) : null; +} + +/** + * @param {{ params: Promise<{ username: string, mode: string }> }} props + */ +export async function generateMetadata({ params }) { + const source = sourceFor(await params); + if (!source) return { title: 'Not found', robots: { index: false, follow: false } }; + + return socialMetadata({ + feed: await socialFeed(source.ref), + canonical: source.path, + label: source.title, + network: 'x', + }); +} + +/** + * @param {{ params: Promise<{ username: string, mode: string }> }} props + */ +export default async function XModePage({ params }) { + const source = sourceFor(await params); + if (!source) notFound(); + + const feed = await socialFeed(source.ref); + if (!feed) { + return ( + + ); + } + + return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }) }); +} diff --git a/apps/web/src/app/x/[username]/page.jsx b/apps/web/src/app/x/[username]/page.jsx new file mode 100644 index 0000000..c187989 --- /dev/null +++ b/apps/web/src/app/x/[username]/page.jsx @@ -0,0 +1,54 @@ +import { notFound } from 'next/navigation'; +import { xSource } from '@rssamplifier/social'; + +import { socialFeed, socialMetadata } from '../../../lib/socialPage.js'; +import AddSocialSource from '../../AddSocialSource.jsx'; +import FeedPage from '../../[slug]/page.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * One X account, at `/x/OpenAI`. + * + * Same arrangement as `/r/[subreddit]`: the page is `/{slug}`'s, and what this + * route contributes is the name and the canonical tag. See that file for why + * it renders rather than redirects. + * + * @param {{ params: Promise<{ username: string }> }} props + */ +export async function generateMetadata({ params }) { + const { username } = await params; + const source = xSource(username); + if (!source) return { title: 'Not found', robots: { index: false, follow: false } }; + + return socialMetadata({ + feed: await socialFeed(source.ref), + canonical: source.path, + label: source.title, + network: 'x', + }); +} + +/** + * @param {{ params: Promise<{ username: string }> }} props + */ +export default async function XAccountPage({ params }) { + const { username } = await params; + + const source = xSource(username); + if (!source) notFound(); + + const feed = await socialFeed(source.ref); + if (!feed) { + return ( + + ); + } + + return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }) }); +} diff --git a/apps/web/src/app/x/list/[listId]/page.jsx b/apps/web/src/app/x/list/[listId]/page.jsx new file mode 100644 index 0000000..1fc7f06 --- /dev/null +++ b/apps/web/src/app/x/list/[listId]/page.jsx @@ -0,0 +1,50 @@ +import { notFound } from 'next/navigation'; +import { xSource } from '@rssamplifier/social'; + +import { socialFeed, socialMetadata } from '../../../../lib/socialPage.js'; +import AddSocialSource from '../../../AddSocialSource.jsx'; +import FeedPage from '../../../[slug]/page.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * An X list, at `/x/list/123456789`. + * + * @param {{ params: Promise<{ listId: string }> }} props + */ +export async function generateMetadata({ params }) { + const { listId } = await params; + const source = xSource(`https://x.com/i/lists/${listId}`); + if (!source) return { title: 'Not found', robots: { index: false, follow: false } }; + + return socialMetadata({ + feed: await socialFeed(source.ref), + canonical: source.path, + label: source.title, + network: 'x', + }); +} + +/** + * @param {{ params: Promise<{ listId: string }> }} props + */ +export default async function XListPage({ params }) { + const { listId } = await params; + + const source = xSource(`https://x.com/i/lists/${listId}`); + if (!source) notFound(); + + const feed = await socialFeed(source.ref); + if (!feed) { + return ( + + ); + } + + return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }) }); +} diff --git a/apps/web/src/app/x/page.jsx b/apps/web/src/app/x/page.jsx new file mode 100644 index 0000000..a31eff5 --- /dev/null +++ b/apps/web/src/app/x/page.jsx @@ -0,0 +1,27 @@ +import { siteUrl } from '../../lib/db.js'; +import SocialIndex, { pageNumber } from '../SocialIndex.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * @param {{ searchParams: Promise> }} props + */ +export async function generateMetadata({ searchParams }) { + const page = pageNumber((await searchParams).page); + + return { + title: page === 1 ? 'X' : `X · page ${page}`, + description: + 'X accounts, searches and lists as feeds you can subscribe to — collected by RSS Amplifier and served from here, never from X.', + alternates: { + canonical: page === 1 ? `${siteUrl()}/x` : `${siteUrl()}/x?page=${page}`, + }, + }; +} + +/** + * @param {{ searchParams: Promise> }} props + */ +export default async function XIndexPage({ searchParams }) { + return ; +} diff --git a/apps/web/src/app/x/search/page.jsx b/apps/web/src/app/x/search/page.jsx new file mode 100644 index 0000000..8f1dda8 --- /dev/null +++ b/apps/web/src/app/x/search/page.jsx @@ -0,0 +1,93 @@ +import { notFound } from 'next/navigation'; +import { xSource } from '@rssamplifier/social'; + +import { siteUrl } from '../../../lib/db.js'; +import { socialFeed, socialMetadata } from '../../../lib/socialPage.js'; +import AddSocialSource from '../../AddSocialSource.jsx'; +import FeedPage from '../../[slug]/page.jsx'; + +export const dynamic = 'force-dynamic'; + +/** + * An X search, at `/x/search?q=bitcoin`. + * + * The query stays in the query string for the reason given in the route + * handler: X's operator syntax is passed through whole, and `from:OpenAI + * lang:en` cannot survive being a path segment without an escaping scheme + * nobody should have to learn to subscribe to something. + * + * With no `?q=` this is a form rather than a 404 — somebody who typed + * `/x/search` was asking for the search page, and giving them one is a shorter + * path to what they wanted than an error. + * + * @param {{ searchParams: Promise> }} props + */ +export async function generateMetadata({ searchParams }) { + const query = String((await searchParams).q ?? '').trim(); + const source = query ? xSource(`https://x.com/search?q=${encodeURIComponent(query)}`) : null; + + if (!source) { + return { + title: 'Search X', + description: 'Turn an X search into a feed you can subscribe to.', + alternates: { canonical: `${siteUrl()}/x/search` }, + }; + } + + const [canonical, suffix] = source.path.split('?'); + + return socialMetadata({ + feed: await socialFeed(source.ref), + // The query belongs in the canonical URL: two searches are two pages, and + // collapsing them onto `/x/search` would tell a crawler they are one. + canonical: `${canonical}?${suffix}`, + label: source.title, + network: 'x', + }); +} + +/** + * @param {{ searchParams: Promise> }} props + */ +export default async function XSearchPage({ searchParams }) { + const query = String((await searchParams).q ?? '').trim(); + + if (!query) { + return ( +

+

Search X

+

+ Any X search can be a feed. Type one below — X’s own operators work, so{' '} + from:OpenAI lang:en does what you would expect. +

+
+ + +
+

+ Browse the X sources already here +

+
+ ); + } + + // Cannot be null: the only thing that makes a search unparseable is an empty + // query, and that is the branch above. Guarded anyway, because the day + // somebody adds a length cap to the parser this is where it would surface. + const source = xSource(`https://x.com/search?q=${encodeURIComponent(query)}`); + if (!source) notFound(); + + const feed = await socialFeed(source.ref); + if (!feed) { + return ( + + ); + } + + return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }) }); +} diff --git a/apps/web/src/app/x/status/page.jsx b/apps/web/src/app/x/status/page.jsx new file mode 100644 index 0000000..0e37636 --- /dev/null +++ b/apps/web/src/app/x/status/page.jsx @@ -0,0 +1,144 @@ +import { social } from '@rssamplifier/db'; + +import { db, siteUrl } from '../../../lib/db.js'; + +export const dynamic = 'force-dynamic'; + +export const metadata = { + title: 'X collection status', + description: 'Which provider is collecting X posts, how it is doing, and how stale anything is.', + // A status board is not a page to rank; it is a page to check. + robots: { index: false, follow: true }, +}; + +/** + * How X collection is going (§32, §33, §34). + * + * **Read-only, and deliberately so.** §34 asks for a `/admin/x` with buttons to + * disable a provider, clear a cooldown and force a refresh. This codebase has + * no notion of an administrator at all — no role column, no admin route, no + * guard to hang one on — so those buttons would have to arrive with an + * authorisation system, and shipping the levers ahead of the lock is how a + * kill switch becomes a way for anybody to turn collection off. The + * environment already holds the two that matter: `X_ENABLED` stops collection + * entirely and `X_PRIMARY_PROVIDER` re-orders the stack, both without a deploy + * of code. The buttons are the part left undone, and this is where they go. + * + * **It is a lagging view, not a probe.** The web service never collects + * anything; the poller does, and writes what happened to `x_provider_state`. + * So this page reads that table and says when it was last written rather than + * asking a provider how it is right now — which is also the only way to render + * it without spending an upstream request per page view (§32). + * + * Nothing here is a secret. Session ids are names, not credentials; the tokens + * live in the environment and are not in the database to leak (AC-7). + */ +export default async function XStatusPage() { + const client = db(); + + const [providers, sessions, counts, stale] = await Promise.all([ + social.providerStates(client), + social.sessionStates(client), + social.countSocialFeeds(client, 'x'), + social.countStaleSocialFeeds(client, 'x'), + ]); + + return ( +
+

X collection status

+ +

+ {counts.total.toLocaleString()} X sources, {counts.crawled.toLocaleString()} collected at + least once, {stale.toLocaleString()} overdue by more than three of their own refresh + intervals. Overdue is judged on when we last read a source, never on when it + last posted — a quiet account is quiet, not broken. +

+ +

Providers

+ + {providers.length === 0 ? ( +

+ No provider has reported yet. Either collection is switched off (X_ENABLED) + or the poller has not run a crawl since this table was created. +

+ ) : ( + + + + + + + + + + + + + {providers.map((row) => ( + + + + + + + + + ))} + +
ProviderStatusLast successFailuresCooldown untilLast error
{String(row.provider)}{String(row.status ?? 'unknown')}{row.last_success_at ? String(row.last_success_at) : '—'}{Number(row.consecutive_failures ?? 0)}{row.cooldown_until ? String(row.cooldown_until) : '—'}{row.error_message ? String(row.error_message) : '—'}
+ )} + +

+ Which of these answered any particular post is not recorded against the post, and that is + the point: {siteUrl()}/x/OpenAI.rss is the same address whichever provider + filled it, so a subscriber never has to know one failed over to another. +

+ +

Sessions

+ + {sessions.length === 0 ? ( +

+ No X sessions are configured. The unofficial providers need a logged-in session to + collect anything; the official API provider does not. +

+ ) : ( + + + + + + + + + + + + + {sessions.map((row) => ( + + + + + + + + + ))} + +
SessionStatusLast usedFailuresCooldown untilLast error
{String(row.id)}{String(row.status ?? 'healthy')}{row.last_used_at ? String(row.last_used_at) : '—'}{Number(row.consecutive_failures ?? 0)}{row.cooldown_until ? String(row.cooldown_until) : '—'}{row.last_error ? String(row.last_error) : '—'}
+ )} + +

+ An expired session has had its credentials rejected and will not come back on + its own — a cookie that has been invalidated does not become valid again after a wait, so + it stays out until it is replaced in X_SESSIONS. A{' '} + rate_limited or cooldown session returns by itself when its + cooldown runs out. +

+ +

+ Back to X · Crawler status +

+
+ ); +} diff --git a/apps/web/src/lib/feedRiver.js b/apps/web/src/lib/feedRiver.js index 86dba0c..f62c6da 100644 --- a/apps/web/src/lib/feedRiver.js +++ b/apps/web/src/lib/feedRiver.js @@ -89,5 +89,9 @@ export async function feedRiver({ slug: rawSlug, format: rawFormat, limit: rawLi items, filename: slug, src: 'feed', + // The request, so an unchanged river can answer 304 (§18). Every other + // caller of riverResponse is free to leave this off and keep the behaviour + // it had; passing it is what opts a surface in. + req, }); } diff --git a/apps/web/src/lib/river.js b/apps/web/src/lib/river.js index ae67b04..50e70ba 100644 --- a/apps/web/src/lib/river.js +++ b/apps/web/src/lib/river.js @@ -21,6 +21,8 @@ * advertiser is metered for reach that never left the building. */ +import { createHash } from 'node:crypto'; + import { SYNDICATION_FORMATS, adSlotsFor, buildSyndication, interleaveAds, playable } from '@rssamplifier/feed'; import { fetchFeedAds } from './feedAds.js'; @@ -89,11 +91,34 @@ export async function riverResponse({ filename, src, maxAge = 300, + req = null, }) { // A playlist can only carry files. Filtering here rather than in every query // keeps the surfaces from each inventing their own idea of what is playable. const rows = spec.media ? items.filter(playable) : items; + // The validator, computed from what was stored rather than from the document. + // + // Two reasons it is not a hash of the body, and both matter. A body carries + // sponsored items chosen per request, so hashing it would mint a new ETag on + // every call and no reader would ever see a 304 — the header would be + // decoration. And computing it here, *before* the ad fetch, is what lets an + // unchanged river answer without paying for one: a 304 sends no document, so + // no ad was delivered, so no impression should be metered. Fetching one and + // discarding it would bill an advertiser for reach that never left the + // building, which is the same rule the ad count above follows. + const etag = riverEtag(format, channel, rows); + if (notModified(req, etag)) { + return new Response(null, { + status: 304, + headers: { + etag, + 'access-control-allow-origin': '*', + 'cache-control': `public, max-age=${maxAge}, s-maxage=${maxAge}, stale-while-revalidate=3600`, + }, + }); + } + // Sponsored items, one in ten. Never in a playlist: a sponsored line has // nothing for a player to open, and VLC handed one shows an error. // @@ -109,11 +134,62 @@ export async function riverResponse({ 'content-type': spec.type, 'content-disposition': `inline; filename="${riverFilename(filename, format)}"`, 'access-control-allow-origin': '*', + etag, 'cache-control': `public, max-age=${maxAge}, s-maxage=${maxAge}, stale-while-revalidate=3600`, }, }); } +/** + * A weak validator for a river. + * + * Weak — `W/"…"` — because it is deliberately not byte-for-byte: two responses + * carrying this tag hold the same posts but may hold different sponsored items. + * That is exactly what a weak validator is defined to mean, and claiming a + * strong one would be a lie that a range request could catch us in. + * + * The inputs are the format, the channel's own address and every item's + * identity and date. Anything that changes a byte of the rendered river changes + * one of those — except the ads, which is the point. + * + * @param {string} format + * @param {{ selfUrl?: string, title?: string }} channel + * @param {object[]} rows + * @returns {string} + */ +export function riverEtag(format, channel, rows) { + const hash = createHash('sha1'); + hash.update(`${format}\n${channel?.selfUrl ?? ''}\n${channel?.title ?? ''}\n${rows.length}`); + + for (const row of rows) { + hash.update(`\n${row?.id ?? row?.guid ?? ''} ${row?.published_at ?? ''}`); + } + + return `W/"${hash.digest('base64url').slice(0, 27)}"`; +} + +/** + * Does the caller already hold this exact river? + * + * `If-None-Match` may carry a list, and `*` matches anything we have. The weak + * prefix comes off both sides before comparing — RFC 9110 calls that weak + * comparison, and it is the only comparison a weak tag supports. + * + * @param {Request|null} req + * @param {string} etag + * @returns {boolean} + */ +export function notModified(req, etag) { + const header = req?.headers?.get?.('if-none-match'); + if (!header) return false; + + const mine = etag.replace(/^W\//, ''); + return header + .split(',') + .map((value) => value.trim().replace(/^W\//, '')) + .some((value) => value === '*' || value === mine); +} + /** * A filename a reader or a player will not be embarrassed by. * diff --git a/apps/web/src/lib/sitemap.js b/apps/web/src/lib/sitemap.js index 7aa7682..284cb68 100644 --- a/apps/web/src/lib/sitemap.js +++ b/apps/web/src/lib/sitemap.js @@ -38,6 +38,16 @@ export const STATIC_PAGES = [ { path: '/reels', changefreq: 'daily', priority: '0.7' }, { path: '/topics', changefreq: 'daily', priority: '0.8' }, { path: '/authors', changefreq: 'daily', priority: '0.8' }, + // The two platform namespaces, alongside the categories for the same reason: + // they are entry points into the directory that are about something. + // + // Only the index of each. The individual sources are already in the blog + // chunks under their `/{slug}` URL, and listing them a second time under + // `/r/…` and `/x/…` is exactly the duplicate a sitemap should not volunteer — + // the canonical tag on each page is what tells a crawler which of the two + // addresses to keep, and it does not need the sitemap's help to do it. + { path: '/r', changefreq: 'daily', priority: '0.8' }, + { path: '/x', changefreq: 'daily', priority: '0.8' }, { path: '/search', changefreq: 'daily', priority: '0.8' }, { path: '/submit', changefreq: 'weekly', priority: '0.7' }, { path: '/signup', changefreq: 'monthly', priority: '0.6' }, diff --git a/apps/web/src/lib/socialPage.js b/apps/web/src/lib/socialPage.js new file mode 100644 index 0000000..8fc788c --- /dev/null +++ b/apps/web/src/lib/socialPage.js @@ -0,0 +1,64 @@ +import { social } from '@rssamplifier/db'; + +import { db, siteUrl } from './db.js'; +import { feedAlternates } from './subscribe.js'; + +/** + * The plumbing shared by every `/r/…` and `/x/…` page. + * + * Each of those pages is the *same page* as `/{slug}` — same component, same + * items, same controls — reached by a different name. So the only work here is + * turning a canonical ref into the slug that page wants, and writing the + * metadata that says `/r/programming` rather than `/r-programming` is where + * this lives. + * + * That canonical tag is the substantive half. Both addresses render, because + * `/{slug}` is the permanent identity of a row and every link already pointing + * at one has to keep working. Telling crawlers which of the two is the real one + * is what stops that from being a duplicate-content problem, and what gets + * `/r/programming` into a search index instead of `/r-programming`. + */ + +/** + * @param {string} ref + * @returns {Promise} + */ +export async function socialFeed(ref) { + return social.feedBySocialRef(db(), ref); +} + +/** + * Metadata for a social source's page. + * + * @param {{ feed: object|null, canonical: string, label: string, network: string }} args + * @returns {object} + */ +export function socialMetadata({ feed, canonical, label, network }) { + const url = `${siteUrl()}${canonical}`; + + if (!feed) { + return { + title: label, + description: `${label} is not in the RSS Amplifier directory yet.`, + alternates: { canonical: url }, + // Nothing to index until somebody adds it. Without this, every mistyped + // handle on the internet is a thin page inviting a crawler to keep it. + robots: { index: false, follow: true }, + }; + } + + return { + title: String(feed.title ?? label), + description: String( + feed.description ?? `${label}, mirrored by the RSS Amplifier directory.`, + ), + alternates: { + canonical: url, + // The same four formats the rewrites serve. No playlists: a timeline and + // a subreddit carry no enclosures, so announcing an `.m3u` would be + // advertising an empty file. + types: feedAlternates(url, String(feed.title ?? label)), + }, + other: { 'x-social-network': network }, + }; +} diff --git a/apps/web/src/lib/socialRiver.js b/apps/web/src/lib/socialRiver.js new file mode 100644 index 0000000..95b5057 --- /dev/null +++ b/apps/web/src/lib/socialRiver.js @@ -0,0 +1,157 @@ +import { q, social } from '@rssamplifier/db'; +import { redditSource, xSource } from '@rssamplifier/social'; + +import { db, siteUrl } from './db.js'; +import { + riverFail, + riverFormat, + riverItem, + riverLimit, + riverResponse, + unsupportedFormat, +} from './river.js'; + +/** + * A social source, as a feed of ours, at the address people already know. + * + * `/r/programming.rss` and `/x/OpenAI.rss` are the same machinery as + * `/{slug}.rss` — the same items, the same renderer, the same ads — differing + * in exactly one way: the source is found by its canonical ref rather than by + * its slug, and the document says `/r/programming` is its own address. + * + * That last part is the entire contract with a subscriber. §5 and AC-2 ask that + * these URLs never change, and the thing most likely to change underneath them + * is which provider collected the posts — so the provider appears in neither + * the address nor the document. A reader subscribed today through RSSHub is + * subscribed tomorrow through Teapot without noticing, because there is nothing + * in what they hold that could tell them. + * + * **Nothing here touches X or Reddit.** The items come out of our database; + * the crawler put them there minutes ago. That is what makes AC-4 true by + * construction rather than by a cache: a hundred requests are a hundred reads + * of one row's items, and the upstream sees none of them. It is also what makes + * AC-5 true — an upstream that is down changes nothing about this path, because + * this path never asked it anything. + */ + +/** + * Serve one social source in one format. + * + * @param {{ + * ref: string, + * canonical: string, + * format: string, + * limit?: unknown, + * req?: Request, + * label?: string, + * query?: string|null, + * }} args `canonical` is the path this document lives at — `/r/programming` — + * without an extension. `query` is the query string a search feed keeps, and + * it goes *after* the extension: `/x/search.rss?q=bitcoin`, never + * `/x/search?q=bitcoin.rss`. Getting that the wrong way round produces a + * document whose stated address is a different search, which is the sort of + * thing a reader only discovers when their subscription drifts. + * @returns {Promise} + */ +export async function socialRiver({ + ref, + canonical, + format: rawFormat, + limit: rawLimit, + req = null, + label = null, + query = null, +}) { + const { format, spec } = riverFormat(rawFormat); + if (!spec) return unsupportedFormat(format); + + const client = db(); + const feed = await social.feedBySocialRef(client, ref); + + const suffix = query ? `?${query}` : ''; + + if (!feed) { + // A 404 with a way forward. Most misses here are a real account nobody has + // added yet rather than a typo, and the page at `canonical` is the one that + // offers to add it — so the hint points at a working next step instead of + // at the front door. + return riverFail( + format, + 404, + `not in the directory: ${label ?? ref}`, + `Add it at ${siteUrl()}${canonical}${suffix}`, + ); + } + + const page = `${siteUrl()}${canonical}`; + const rows = await q.itemsForFeed(client, String(feed.id), riverLimit(rawLimit)); + + const channel = { + title: String(feed.title ?? label ?? ref), + description: String( + feed.description ?? `${label ?? ref}, mirrored by the RSS Amplifier directory.`, + ), + link: `${page}${suffix}`, + selfUrl: `${page}.${format}${suffix}`, + language: feed.language ? String(feed.language) : undefined, + }; + + return riverResponse({ + format, + spec, + channel, + items: rows.map((row) => riverItem(row)), + // The stem of the downloaded filename. `rssamplifier-r-programming.rss` + // rather than the row's slug, so what lands in a Downloads folder matches + // the URL it was fetched from. + filename: canonical.replace(/^\//, '').replace(/\//g, '-'), + src: 'social', + req, + }); +} + +/** + * The canonical ref and path for a `/r/…` request, or null if it is not one. + * + * Parsing rather than pattern-matching, so `/r/Programming.rss` and + * `/r/programming.rss` reach the same source: Reddit's own URLs are + * case-insensitive, and two rows for one community is exactly what §38 exists + * to prevent. + * + * @param {{ subreddit?: string, username?: string }} params + * @returns {{ ref: string, canonical: string, label: string }|null} + */ +export function redditTarget(params) { + const source = params.username + ? redditSource(`u/${params.username}`) + : redditSource(`r/${params.subreddit}`); + + if (!source) return null; + return { ref: source.ref, canonical: source.path, label: source.title }; +} + +/** + * The same for `/x/…`, across all five modes. + * + * @param {{ username?: string, mode?: string, listId?: string, query?: string }} params + * @returns {{ ref: string, canonical: string, label: string, query: string|null }|null} + */ +export function xTarget(params) { + const input = params.query + ? `https://x.com/search?q=${encodeURIComponent(params.query)}` + : params.listId + ? `https://x.com/i/lists/${params.listId}` + : params.mode === 'replies' + ? `https://x.com/${params.username}/with_replies` + : params.mode === 'media' + ? `https://x.com/${params.username}/media` + : `https://x.com/${params.username ?? ''}`; + + const source = xSource(input); + if (!source) return null; + + // A search's path carries its query, and the two have to be handed back apart + // so the extension can go between them — see the note on `query` above. + const [canonical, query = null] = source.path.split('?'); + return { ref: source.ref, canonical, label: source.title, query }; +} diff --git a/packages/db/index.js b/packages/db/index.js index 278f1f4..c7a1b49 100644 --- a/packages/db/index.js +++ b/packages/db/index.js @@ -18,3 +18,4 @@ export * as extracts from './src/extracts.js'; export * as apikeys from './src/apikeys.js'; export * as authors from './src/authors.js'; export * as alerts from './src/alerts.js'; +export * as social from './src/social.js'; diff --git a/packages/db/migrations/20260829120956_social_sources.sql b/packages/db/migrations/20260829120956_social_sources.sql new file mode 100644 index 0000000..acef829 --- /dev/null +++ b/packages/db/migrations/20260829120956_social_sources.sql @@ -0,0 +1,168 @@ +-- Social sources: X and Reddit get an identity and a namespace of their own. +-- +-- Two columns on `feeds` rather than a `sources` table beside it. The PRD +-- sketches the second (§20) and it would be the right shape for a system that +-- did not already have one: `feeds` + `feed_items` is exactly the sources/items +-- pair that section describes, already carrying dedupe, scheduling, backoff, +-- keyword extraction, full-text search, alerts and syndication. A parallel pair +-- would need every one of those written a second time, and §30's "topic code +-- must not contain X-specific provider logic" would become a rule somebody has +-- to remember rather than a fact about the schema. +-- +-- `source_kind` was the obvious place to put this and is deliberately not used. +-- It carries `check (source_kind in ('feed', 'scraped'))`, and SQLite cannot +-- alter a CHECK constraint — widening it means rebuilding a table that now has +-- thirty-odd columns, a dozen indexes, foreign keys from six others and FTS +-- triggers, against 300k rows on a database with a single writer. The two new +-- columns say the same thing and cost nothing. + +-- Which platform, or null for the ordinary web. Left unconstrained on purpose: +-- the next network to get a namespace should be a migration that adds rows, not +-- one that rebuilds a table for the reason above. +alter table feeds add column social_network text; + +-- Our canonical identity for the source: `r:sub:programming`, `x:user:openai`. +-- This is the column that makes §37/§38 true — a thousand readers asking for +-- @OpenAI collapse onto one row here, and therefore onto one polling job. +alter table feeds add column social_ref text; + +-- Per-source render toggles (§6.3): includeReplies, includeReposts, +-- includeQuotes. JSON because they are read as a set and never queried on. +alter table feeds add column social_config text; + +-- One row per canonical source. Partial, so the 300k feeds that are not social +-- do not each occupy an index entry for a null. +create unique index if not exists feeds_social_ref_idx + on feeds (social_ref) where social_ref is not null; + +-- Listing a network's sources: /r and /x, and the status page's counts. +create index if not exists feeds_social_network_idx + on feeds (social_network, created_at desc) where social_network is not null; + +-- --------------------------------------------------------------------------- +-- Backfill: the Reddit sources that are already here. +-- +-- A bulk import put 50,099 subreddits in the directory, each at a slug of its +-- own among the blogs — see the comment on markHostThrottled in queries.js, +-- where the same import is why one host can be 41% of the crawl queue. They are +-- not moved, renamed or deleted: they gain an identity, which is what lets +-- /r/programming answer, and their own /{slug} keeps working for every link +-- already pointing at it. +-- +-- The name is extracted rather than matched, because the stored URLs come from +-- an OPML file and take every shape Reddit serves: with and without `www.`, on +-- `old.`, with `.rss`, with a sort segment, with a query string. A view holds +-- that arithmetic once — it is unpleasant enough written out that a second copy +-- would be a second place to get it subtly wrong. +create view if not exists _social_backfill_raw as +with candidates as ( + select + id, + feed_url, + case + when instr(feed_url, 'reddit.com/r/') > 0 then 'sub' + when instr(feed_url, 'reddit.com/user/') > 0 then 'user' + end as mode, + case + when instr(feed_url, 'reddit.com/r/') > 0 + then substr(feed_url, instr(feed_url, '/r/') + 3) + when instr(feed_url, 'reddit.com/user/') > 0 + then substr(feed_url, instr(feed_url, '/user/') + 6) + end as tail + from feeds + where social_ref is null + and (instr(feed_url, 'reddit.com/r/') > 0 or instr(feed_url, 'reddit.com/user/') > 0) +) +select + id, + feed_url, + mode, + -- Everything up to whichever of `/`, `.` or `?` comes first. The `|| c` + -- makes instr() certain to find each one, so there is no zero to special-case. + substr(tail, 1, min(instr(tail || '/', '/'), instr(tail || '.', '.'), instr(tail || '?', '?')) - 1) as name +from candidates; + +-- The same, filtered to names that really are names. A subreddit is 3-21 of +-- [A-Za-z0-9_] and a username 3-20 of that plus `-`; anything else extracted +-- from that position is not one, and guessing would file somebody's blog under +-- a community that does not exist. +create view if not exists _social_backfill as +select + id, + mode, + name, + -- Is this URL already the document Reddit publishes, rather than a sort tab + -- or an `old.` mirror of it? Used only to decide who wins a collision. + case + when feed_url = 'https://www.reddit.com/r/' || name || '/.rss' then 1 + when feed_url = 'https://www.reddit.com/user/' || name || '/.rss' then 1 + else 0 + end as canonical +from _social_backfill_raw +where name is not null + and ( + (mode = 'sub' and length(name) between 3 and 21 and name not glob '*[^A-Za-z0-9_]*') + or + (mode = 'user' and length(name) between 3 and 20 and name not glob '*[^A-Za-z0-9_-]*') + ); + +-- Two passes, and the order is the whole reason there are two. +-- +-- The mapping is many-to-one: `/r/x/.rss` and `/r/x/new/.rss` are two rows and +-- one community, and the unique index above is what says so. `update or ignore` +-- means the loser keeps a null ref and stays an ordinary feed rather than +-- failing this migration — a duplicate row is a tidiness problem, and aborting +-- a deploy over one would be the worse trade. +-- +-- But *which* row loses would otherwise be decided by rowid, which is to say by +-- the order an OPML file happened to list them in. Claiming the canonical +-- spellings first makes it deterministic and picks the better row: /r/programming +-- ends up backed by the feed Reddit publishes for that community, not by +-- whichever sort tab was imported first. +update or ignore feeds +set social_network = 'reddit', + social_ref = (select 'r:' || b.mode || ':' || lower(b.name) + from _social_backfill b where b.id = feeds.id) +where id in (select id from _social_backfill where canonical = 1); + +update or ignore feeds +set social_network = 'reddit', + social_ref = (select 'r:' || b.mode || ':' || lower(b.name) + from _social_backfill b where b.id = feeds.id) +where id in (select id from _social_backfill); + +drop view if exists _social_backfill; +drop view if exists _social_backfill_raw; + +-- --------------------------------------------------------------------------- +-- Provider and session health (§20, §32). +-- +-- Note what is absent from both tables: there is no token column, and there is +-- no room for one. X session credentials are a full login to an account, and +-- they live in the environment (`X_SESSIONS`) precisely so that a leaked +-- database dump — the likeliest way any of this escapes — carries none of them. +-- These tables hold the state that has to survive a redeploy: which provider is +-- in cooldown, which session is expired, and why. See §36 and AC-7. + +create table if not exists x_provider_state ( + provider text primary key, + status text not null default 'unknown', + last_success_at text, + last_failure_at text, + consecutive_failures integer not null default 0, + cooldown_until text, + -- A message, truncated by the writer. Never a URL with a query string: a + -- provider URL can carry a session token, which is what redact() in + -- providers/http.js exists to strip before anything reaches here or a log. + error_message text +); + +create table if not exists x_sessions ( + -- The id from X_SESSIONS. A name, not a secret. + id text primary key, + status text not null default 'healthy', + cooldown_until text, + last_used_at text, + consecutive_failures integer not null default 0, + last_error text +); diff --git a/packages/db/src/queries.js b/packages/db/src/queries.js index a0582f5..bc2f7d3 100644 --- a/packages/db/src/queries.js +++ b/packages/db/src/queries.js @@ -2383,7 +2383,8 @@ export async function recentlyCrawled(db, limit = 20) { /** The columns a crawl needs off a feed row. Shared by both due queries. */ const DUE_COLUMNS = `id, slug, title, feed_url, error_count, fetch_interval_minutes, source_kind, item_count, last_published_at, - http_etag, http_last_modified, content_hash, change_log`; + http_etag, http_last_modified, content_hash, change_log, + social_network, social_ref, social_config`; /** * The share of a tick reserved for hand-submitted feeds. diff --git a/packages/db/src/social.js b/packages/db/src/social.js new file mode 100644 index 0000000..2894662 --- /dev/null +++ b/packages/db/src/social.js @@ -0,0 +1,317 @@ +/** + * Queries for social sources — X and Reddit — and for the health tables the X + * provider stack writes to. + * + * Its own module rather than more of `queries.js`, which is already 3,900 + * lines: nothing here is read by the ordinary crawl path, and a caller that + * imports `social` is announcing what it is about to do. + * + * The shape of the deal with `queries.js` is worth stating, because it is what + * keeps this feature from spreading. A social source is a row in `feeds` like + * any other, so every query about *what a source published* — items, topics, + * search, alerts, the river — is already written and is not repeated here. What + * is here is only the part that is genuinely new: finding a row by its + * canonical ref, creating one, and listing a network. + */ + +import { newId, nowIso } from './client.js'; + +/** + * @typedef {import('@libsql/client').Client} Client + */ + +/** + * The one source behind a canonical ref. + * + * This is what `/r/programming` and `/x/OpenAI` resolve through, and it is the + * query that makes §39 true: ten thousand subscribers to `@OpenAI` are ten + * thousand calls to *this*, all landing on one row, and none of them reach X. + * + * @param {Client} db + * @param {string} ref + * @returns {Promise} + */ +export async function feedBySocialRef(db, ref) { + const key = String(ref ?? ''); + if (!key) return null; + + const result = await db.execute({ + sql: 'select * from feeds where social_ref = ? limit 1', + args: [key], + }); + + return result.rows[0] ?? null; +} + +/** + * Every source on one network, newest first. + * + * @param {Client} db + * @param {string} network + * @param {{ limit?: number, offset?: number }} [opts] + * @returns {Promise} + */ +export async function listSocialFeeds(db, network, opts = {}) { + const limit = Math.max(1, Math.min(Number(opts.limit) || 50, 200)); + const offset = Math.max(0, Number(opts.offset) || 0); + + const result = await db.execute({ + sql: `select id, slug, title, description, social_ref, social_network, site_url, image_url, + item_count, status, last_success_at, last_published_at, created_at + from feeds + where social_network = ? + order by item_count desc, created_at desc + limit ? offset ?`, + args: [String(network), limit, offset], + }); + + return result.rows; +} + +/** + * How many sources one network has, and how many of them have ever been read. + * + * Two numbers rather than one because they are very far apart here: the + * subreddit import put 50,099 rows in and the crawler has read a fraction of + * them, so a bare count on `/r` would promise a directory that is mostly + * unread. See the `freshness` note in the MCP server's instructions for the + * same distinction made to agents. + * + * @param {Client} db + * @param {string} network + * @returns {Promise<{ total: number, crawled: number }>} + */ +export async function countSocialFeeds(db, network) { + const result = await db.execute({ + sql: `select count(*) as total, + sum(case when last_success_at is not null then 1 else 0 end) as crawled + from feeds where social_network = ?`, + args: [String(network)], + }); + + const row = result.rows[0] ?? {}; + return { total: Number(row.total ?? 0), crawled: Number(row.crawled ?? 0) }; +} + +/** + * Create a social source, or hand back the one that is already there. + * + * The whole of §37/§38 lives in the `on conflict do nothing` and the read after + * it. Two people submitting `@OpenAI` a second apart must not create two rows, + * and the race is not hypothetical — the submit path is public and unauthenticated. + * The unique index on `social_ref` is the arbiter; this function just declines + * to argue with it. + * + * Inserted as `pending`, like every other new feed: the crawler picks it up on + * its next tick and the first fetch happens on the poller, never on the web + * request that created it (§17). The submitter is shown a page that fills in. + * + * @param {Client} db + * @param {{ + * network: string, ref: string, slug: string, title: string, feedUrl: string, + * siteUrl?: string|null, config?: object|null, priority?: number, + * }} source + * @returns {Promise<{ id: string, slug: string, created: boolean }>} + */ +export async function upsertSocialSource(db, source) { + const existing = await feedBySocialRef(db, source.ref); + if (existing) { + return { id: String(existing.id), slug: String(existing.slug), created: false }; + } + + const id = newId(); + const now = nowIso(); + + await db.execute({ + sql: `insert into feeds + (id, slug, feed_url, site_url, title, description, categories, category, status, + error_count, fetch_interval_minutes, next_fetch_at, item_count, + created_at, updated_at, source_kind, + social_network, social_ref, social_config, priority) + values (?, ?, ?, ?, ?, null, '[]', 'blog', 'pending', + 0, ?, ?, 0, ?, ?, 'feed', ?, ?, ?, ?) + on conflict do nothing`, + args: [ + id, + source.slug, + source.feedUrl, + source.siteUrl ?? null, + source.title, + // X sources are polled more often than the 60-minute default, because a + // timeline is the one thing in this directory where an hour old is + // visibly stale. §17's "active source: 5 minutes" is the ceiling the + // crawler's own interval learning then works down from. + source.network === 'x' ? 5 : 60, + now, + now, + now, + source.network, + source.ref, + source.config ? JSON.stringify(source.config) : null, + source.priority ?? 1, + ], + }); + + // Read back rather than trusting the insert. `do nothing` is silent about + // whether it did, and the row that is there may be one another request + // created in the microseconds between the check above and this insert. + const row = await feedBySocialRef(db, source.ref); + if (!row) { + // The slug collided rather than the ref — a different source already holds + // this name. The caller retries with a suffixed slug. + return { id: '', slug: '', created: false }; + } + + return { id: String(row.id), slug: String(row.slug), created: String(row.id) === id }; +} + +/** + * Change a source's render toggles (§6.3). + * + * @param {Client} db + * @param {string} id + * @param {object} config + */ +export async function setSocialConfig(db, id, config) { + await db.execute({ + sql: 'update feeds set social_config = ?, updated_at = ? where id = ?', + args: [JSON.stringify(config ?? {}), nowIso(), String(id)], + }); +} + +/** + * Every X source that is due, for the provider status page's "stale" count. + * + * Staleness is judged on `last_success_at` — when we last *read* the source — + * and never on `last_published_at`, because an account that has not posted for + * a month is quiet, not broken (§33). Confusing the two would light the status + * board up red for a directory working perfectly. + * + * @param {Client} db + * @param {number} [multiplier] how many refresh intervals late counts as stale + * @returns {Promise} + */ +export async function countStaleSocialFeeds(db, network, multiplier = 3) { + const result = await db.execute({ + sql: `select count(*) as stale + from feeds + where social_network = ? + and status <> 'dead' + and last_success_at is not null + and julianday('now') - julianday(last_success_at) + > (fetch_interval_minutes * ?) / 1440.0`, + args: [String(network), Number(multiplier) || 3], + }); + + return Number(result.rows[0]?.stale ?? 0); +} + +// --------------------------------------------------------------------------- +// Provider and session health. +// +// Both pairs are the `{ load, save }` shape `XRegistry` and `XSessionPool` +// accept, so the runtime is wired with two object literals and neither of those +// classes ever sees a database client. That is what keeps @rssamplifier/social +// testable without one. + +/** + * @param {Client} db + * @returns {{ load: () => Promise, save: (state: object) => Promise }} + */ +export function providerStore(db) { + return { + async load() { + const result = await db.execute('select * from x_provider_state'); + return result.rows; + }, + + async save(state) { + await db.execute({ + sql: `insert into x_provider_state + (provider, status, last_success_at, last_failure_at, + consecutive_failures, cooldown_until, error_message) + values (?, ?, ?, ?, ?, ?, ?) + on conflict (provider) do update set + status = excluded.status, + last_success_at = excluded.last_success_at, + last_failure_at = excluded.last_failure_at, + consecutive_failures = excluded.consecutive_failures, + cooldown_until = excluded.cooldown_until, + error_message = excluded.error_message`, + args: [ + String(state.provider), + String(state.status ?? 'unknown'), + state.last_success_at ?? null, + state.last_failure_at ?? null, + Number(state.consecutive_failures ?? 0), + state.cooldown_until ?? null, + state.error_message ? String(state.error_message).slice(0, 200) : null, + ], + }); + }, + }; +} + +/** + * @param {Client} db + * @returns {{ load: () => Promise, save: (state: object) => Promise }} + */ +export function sessionStore(db) { + return { + async load() { + const result = await db.execute('select * from x_sessions'); + return result.rows; + }, + + async save(state) { + await db.execute({ + sql: `insert into x_sessions + (id, status, cooldown_until, last_used_at, consecutive_failures, last_error) + values (?, ?, ?, ?, ?, ?) + on conflict (id) do update set + status = excluded.status, + cooldown_until = excluded.cooldown_until, + last_used_at = excluded.last_used_at, + consecutive_failures = excluded.consecutive_failures, + last_error = excluded.last_error`, + args: [ + String(state.id), + String(state.status ?? 'healthy'), + state.cooldown_until ?? null, + state.last_used_at ?? null, + Number(state.consecutive_failures ?? 0), + // Truncated here as well as at the writer, because this column is + // rendered on a status page and a 4KB provider stack trace on it is + // both useless and a way to leak a URL. Never a token: see redact(). + state.last_error ? String(state.last_error).slice(0, 200) : null, + ], + }); + }, + }; +} + +/** + * Provider health as the status page wants it, without the registry. + * + * The web app has no X runtime of its own — it never collects anything — so it + * reads the table the poller writes. That is also why the page can be honest + * about being a lagging view rather than a live probe. + * + * @param {Client} db + * @returns {Promise} + */ +export async function providerStates(db) { + const result = await db.execute('select * from x_provider_state order by provider'); + return result.rows; +} + +/** + * @param {Client} db + * @returns {Promise} + */ +export async function sessionStates(db) { + const result = await db.execute( + 'select id, status, cooldown_until, last_used_at, consecutive_failures, last_error from x_sessions order by id', + ); + return result.rows; +} diff --git a/packages/feed/src/slug.js b/packages/feed/src/slug.js index f9853e8..41eedef 100644 --- a/packages/feed/src/slug.js +++ b/packages/feed/src/slug.js @@ -28,6 +28,12 @@ const RESERVED = new Set([ 'lives', 'reels', 'topics', + // The social namespaces. A feed slugged 'r' or 'x' would still be served — + // Next puts a static segment ahead of [slug] — but its own page would be + // unreachable behind /r/… and /x/…, which is the same failure the categories + // above are listed for. + 'r', + 'x', // The people index, for the same reason as the categories above: a feed // slugged 'authors' would still be served, but only Next's static segment // would answer and the blog's own page would be unreachable. diff --git a/packages/ingest/package.json b/packages/ingest/package.json index d2bf21e..0624c9f 100644 --- a/packages/ingest/package.json +++ b/packages/ingest/package.json @@ -18,6 +18,7 @@ "@rssamplifier/feed": "workspace:*", "@rssamplifier/db": "workspace:*", "@rssamplifier/mail": "workspace:*", - "@rssamplifier/search": "workspace:*" + "@rssamplifier/search": "workspace:*", + "@rssamplifier/social": "workspace:*" } } diff --git a/packages/ingest/src/cadence.js b/packages/ingest/src/cadence.js index 8425857..8b820ae 100644 --- a/packages/ingest/src/cadence.js +++ b/packages/ingest/src/cadence.js @@ -37,6 +37,26 @@ import { createHash } from 'node:crypto'; /** Never re-read a feed more often than this, however fast it posts. */ export const MIN_INTERVAL = 60; +/** + * The floor for a source we collect through a provider rather than fetch. + * + * An hour is the right floor for the open web: a blog post that appears + * fifty-five minutes late is not late, and the floor is what stops 368,000 + * feeds from being asked more often than they can possibly have changed. + * + * A timeline is the one thing in this directory where it is plainly wrong. + * §17 asks for five minutes on an active source and that is what a reader + * expects of a social feed — an hour-old timeline reads as broken rather than + * as cached. It is affordable for the same reason it would not be for the + * directory at large: there are dozens of X sources, not hundreds of thousands, + * and one upstream refresh serves every subscriber to it (§39). + * + * Every function below takes this as a parameter rather than reading it, so the + * floor is a property of the *source* and there is exactly one place — the + * crawl — that decides which one a given row gets. + */ +export const SOCIAL_MIN_INTERVAL = 5; + /** * Never wait longer than this, however dead a feed looks. * @@ -177,10 +197,10 @@ export function newestPublished(items, now = Date.now()) { * @returns {number|null} minutes, or null when the document carries fewer than * two believable dates */ -export function intervalFromDates(items, now = Date.now()) { +export function intervalFromDates(items, now = Date.now(), floor = MIN_INTERVAL) { const times = publishedTimes(items, now); if (times.length < 2) return null; - return scheduleFrom(times, now); + return scheduleFrom(times, now, floor); } /** @@ -197,7 +217,7 @@ export function intervalFromDates(items, now = Date.now()) { * @param {number} now epoch ms * @returns {number} minutes, between MIN_INTERVAL and MAX_INTERVAL */ -function scheduleFrom(times, now) { +function scheduleFrom(times, now, floor = MIN_INTERVAL) { const silence = Math.max(0, (now - times[0]) / 60_000); // The typical gap, not the mean. A blog that posted forty times during one @@ -210,7 +230,7 @@ function scheduleFrom(times, now) { // archive dumped in one go, or a generator that stamps every entry with the // build time. There is no cadence to infer, so schedule it on its silence // alone, which is the only real evidence available. - if (spacing.length === 0) return clamp(silence / 4, MIN_INTERVAL, MAX_INTERVAL); + if (spacing.length === 0) return clamp(silence / 4, floor, MAX_INTERVAL); const rhythm = median(spacing); @@ -219,14 +239,14 @@ function scheduleFrom(times, now) { // through the next one — the directory's freshness promise is about how long // a post can sit unseen, and this is the term that bounds it. if (silence <= rhythm * QUIET_MULTIPLE) { - return clamp(rhythm / 2, MIN_INTERVAL, MAX_INTERVAL); + return clamp(rhythm / 2, floor, MAX_INTERVAL); } // Gone quiet relative to its own history. Schedule on the silence instead, // which makes the back-off self-scaling: the longer a feed stays dead the // less often it is asked, without a table of thresholds to maintain and // without ever quite giving up on it. - return clamp(silence / 4, MIN_INTERVAL, MAX_INTERVAL); + return clamp(silence / 4, floor, MAX_INTERVAL); } /** @@ -411,10 +431,10 @@ export function recordChange(raw, changed, now = Date.now()) { * @returns {number|null} minutes, or null when the log holds nothing usable and * the caller should fall back to the ladder */ -export function intervalFromChanges(raw, now = Date.now()) { +export function intervalFromChanges(raw, now = Date.now(), floor = MIN_INTERVAL) { const times = changeTimes(raw, now); if (times.length === 0) return null; - return scheduleFrom(times, now); + return scheduleFrom(times, now, floor); } /** @@ -433,9 +453,9 @@ export function intervalFromChanges(raw, now = Date.now()) { * @param {unknown} current the feed's `fetch_interval_minutes` * @returns {number|null} null when there was nothing to compute either */ -export function neverSooner(computed, current) { +export function neverSooner(computed, current, floor = MIN_INTERVAL) { if (computed === null || computed === undefined) return null; const held = Number(current); - if (!Number.isFinite(held) || held <= 0) return clamp(computed, MIN_INTERVAL, MAX_INTERVAL); - return clamp(Math.max(computed, held), MIN_INTERVAL, MAX_INTERVAL); + if (!Number.isFinite(held) || held <= 0) return clamp(computed, floor, MAX_INTERVAL); + return clamp(Math.max(computed, held), floor, MAX_INTERVAL); } diff --git a/packages/ingest/src/crawl.js b/packages/ingest/src/crawl.js index cd27322..eee2210 100644 --- a/packages/ingest/src/crawl.js +++ b/packages/ingest/src/crawl.js @@ -1,5 +1,6 @@ import { resolveFeed, scrapeFeed, feedTopics } from '@rssamplifier/feed'; import { q, authors } from '@rssamplifier/db'; +import { fetchXSource } from '@rssamplifier/social'; import { prepareCredits } from './enrich.js'; import { @@ -10,6 +11,8 @@ import { contentSignature, recordChange, neverSooner, + MIN_INTERVAL as FLOOR_DEFAULT, + SOCIAL_MIN_INTERVAL, } from './cadence.js'; /** Backoff ladder in minutes, indexed by consecutive error count. */ @@ -178,22 +181,66 @@ export function topicsFrom(feed = {}, storedItems = []) { * exercised end to end against a local server without this seam. * @returns {Promise<{ ok: boolean, newItems: number, error?: string }>} */ +/** + * Collect an X source, or decline politely if there is nothing to collect with. + * + * The declining is the point. The X runtime is built once at boot by whoever + * runs the crawl, and a process that has not built one — a test, a script, a + * deploy where `X_ENABLED` is off — must not treat that as the *source* + * failing. `markCrawlFailure` retires a feed after ten consecutive failures, so + * a poller started without X configured would quietly kill every X source in + * the directory over a few hours and leave no trace of why. + * + * So it returns a throttle instead: come back in an hour, change nothing about + * the feed's health. That is also the correct behaviour for the kill switch of + * §42 — turning X off must not damage what has already been collected, and the + * public routes go on serving it (§40, AC-5). + * + * @param {object} feed + * @param {{ x?: Function, xRuntime?: object }} opts + */ +async function collectSocial(feed, opts) { + const runtime = opts.xRuntime ?? null; + if (!runtime) { + return { ok: false, throttled: true, retryAfter: 3600, error: 'x-runtime-unavailable' }; + } + + return (opts.x ?? fetchXSource)(feed, { runtime }); +} + export async function crawlFeed(db, feed, opts = {}) { const id = String(feed.id); const scraped = feed.source_kind === 'scraped'; + // The third way in. A feed is fetched, a scraped source is read off a page, + // and a social source is collected through a provider — three methods, one + // return shape, and everything past this point is identical for all three. + // That is what keeps X out of the rest of the pipeline entirely: dedupe, + // interval learning, keyword extraction, credits, FTS and syndication never + // learn that it exists (§30, AC-8). + const social = feed.social_network === 'x' ? 'x' : null; + + // A provider-backed source polls on a five-minute floor rather than an hour's + // — see SOCIAL_MIN_INTERVAL. The floor is passed to every scheduling call + // below rather than read from a global, so this row's cadence is decided here + // and nowhere else. + const floor = social ? SOCIAL_MIN_INTERVAL : FLOOR_DEFAULT; + // What the server told us last time, sent back so it can answer "still the // same" without sending the document again. Scraped sources are excluded: what // is fetched there is a page of prose whose validators describe the page, and // a marketing site that has not changed its header is not evidence that the // posts extracted from it have not. - const conditional = scraped - ? {} - : { etag: feed.http_etag ?? null, lastModified: feed.http_last_modified ?? null }; + const conditional = + scraped || social + ? {} + : { etag: feed.http_etag ?? null, lastModified: feed.http_last_modified ?? null }; - const resolved = scraped - ? await (opts.scrape ?? scrapeFeed)(String(feed.feed_url)) - : await (opts.resolve ?? resolveFeed)(String(feed.feed_url), conditional); + const resolved = social + ? await collectSocial(feed, opts) + : scraped + ? await (opts.scrape ?? scrapeFeed)(String(feed.feed_url)) + : await (opts.resolve ?? resolveFeed)(String(feed.feed_url), conditional); // The publisher says nothing has changed. This is the cheapest and the most // trustworthy answer the crawler can get: no body was sent, nothing is parsed, @@ -206,7 +253,7 @@ export async function crawlFeed(db, feed, opts = {}) { // feed resting at the ceiling is not dragged back by being checked. if (resolved.notModified) { const minutes = - neverSooner(intervalFromChanges(feed.change_log), feed.fetch_interval_minutes) ?? + neverSooner(intervalFromChanges(feed.change_log, undefined, floor), feed.fetch_interval_minutes, floor) ?? Number(feed.fetch_interval_minutes) ?? MIN_INTERVAL; await q.markUnchanged(db, id, minutes, { @@ -290,10 +337,11 @@ export async function crawlFeed(db, feed, opts = {}) { // and never shorten it; a crawl that saw one recomputes freely, which is what // lets an abandoned feed that starts publishing again accelerate on its first // new post. - const dated = intervalFromDates(resolved.feed.items); - const observed = intervalFromChanges(changeLog); + const dated = intervalFromDates(resolved.feed.items, undefined, floor); + const observed = intervalFromChanges(changeLog, undefined, floor); const interval = - dated ?? (contentsChanged ? observed : neverSooner(observed, feed.fetch_interval_minutes)); + dated ?? + (contentsChanged ? observed : neverSooner(observed, feed.fetch_interval_minutes, floor)); // When this publisher last published, as distinct from when we last read // them. Stored on the feed row so a page can say "current, and dormant since diff --git a/packages/ingest/src/submit.js b/packages/ingest/src/submit.js index bcd1056..d8d9490 100644 --- a/packages/ingest/src/submit.js +++ b/packages/ingest/src/submit.js @@ -1,5 +1,6 @@ import { resolveFeed, scrapeFeed, normalizeUrl, parseOpml, uniqueSlug } from '@rssamplifier/feed'; -import { q } from '@rssamplifier/db'; +import { q, social } from '@rssamplifier/db'; +import { socialSourceFrom } from '@rssamplifier/social'; import { queueFeeds } from './queue.js'; import { refreshFeedKeywords } from './crawl.js'; @@ -79,6 +80,18 @@ export async function claimSlug(db, title, feedUrl) { * @returns {Promise<{ ok: true, slug: string, existing: boolean } | { ok: false, url: string, error: string }>} */ export async function submitOne(db, input) { + // Asked first, before the URL is even normalised, and that order is the whole + // difference between `/r/programming` and a subreddit filed among the blogs. + // + // Reddit publishes real RSS, so `https://www.reddit.com/r/programming/` + // resolves perfectly well down the ordinary path and lands as an untyped row + // at a slug of its own — which is exactly how 50,099 of them got here. X + // resolves to nothing at all, so without this it is simply not submittable. + // Recognising both up here means one answer to "what is this?" rather than a + // special case in each caller. + const source = socialSourceFrom(input); + if (source) return submitSocial(db, source); + const url = normalizeUrl(input); if (!url) return { ok: false, url: String(input), error: 'invalid-url' }; @@ -156,6 +169,59 @@ export async function submitOne(db, input) { return { ok: true, slug: inserted.slug, existing: false }; } +/** + * Accept a social source: claim its identity, queue its first collection. + * + * Nothing is fetched here, unlike `submitOne`'s ordinary path, and that is + * deliberate on a public endpoint that anybody may call. §37 is about exactly + * this: feed creation is the cheapest way to make somebody else's server do + * work, and an X source in particular would make it *our* upstream and *our* + * session paying for it. So a submission writes one row and leaves; the poller + * collects on its next tick, expedited by `priority` into the express lane, and + * the submitter lands on a page that fills in within the minute. + * + * Idempotent by canonical ref rather than by URL, which is the stronger claim: + * `@OpenAI`, `x.com/OpenAI` and `https://twitter.com/openai/` are one source + * here where they would be three feed rows anywhere else. + * + * @param {import('@libsql/client').Client} db + * @param {ReturnType} source + * @returns {Promise<{ ok: true, slug: string, existing: boolean } | { ok: false, url: string, error: string }>} + */ +async function submitSocial(db, source) { + const existing = await social.feedBySocialRef(db, source.ref); + if (existing) return { ok: true, slug: String(existing.slug), existing: true }; + + // The canonical slug first, then the collision-avoiding one. `r-programming` + // is a better name than `programming-2` for a row whose public address is + // /r/programming, and it is only unavailable if something already holds it. + const taken = await q.takenSlugs(db, source.slug); + const slug = taken.has(source.slug) + ? uniqueSlug(source.slug, source.feedUrl, (candidate) => taken.has(candidate)) + : source.slug; + + const stored = await social.upsertSocialSource(db, { + network: source.network, + ref: source.ref, + slug, + title: source.title, + feedUrl: source.feedUrl, + siteUrl: source.siteUrl, + priority: 1, + }); + + if (!stored.id) { + // The ref was free and the slug was not, or another request took both + // between the two statements above. Either way there is a row now. + const raced = await social.feedBySocialRef(db, source.ref); + return raced + ? { ok: true, slug: String(raced.slug), existing: true } + : { ok: false, url: source.feedUrl, error: 'slug-taken' }; + } + + return { ok: true, slug: stored.slug, existing: !stored.created, path: source.path }; +} + /** * Accept a list of URLs. * @@ -173,7 +239,12 @@ export async function submitMany(db, urls) { for (const url of urls.slice(0, MAX_BATCH)) { const res = await submitOne(db, url); - if (res.ok) accepted.push({ slug: res.slug, existing: res.existing }); + // `path` travels with the slug so a caller can redirect to the address a + // source actually lives at. For an ordinary feed that is `/{slug}` and the + // field is absent; for a social source it is `/r/programming` or + // `/x/OpenAI`, and sending somebody to the slug instead would land them on + // the same page at the address the namespace exists to replace. + if (res.ok) accepted.push({ slug: res.slug, existing: res.existing, path: res.path ?? null }); else rejected.push({ url: res.url, error: res.error }); } diff --git a/packages/ingest/test/social-crawl.test.js b/packages/ingest/test/social-crawl.test.js new file mode 100644 index 0000000..b9e0259 --- /dev/null +++ b/packages/ingest/test/social-crawl.test.js @@ -0,0 +1,284 @@ +import assert from 'node:assert/strict'; +import { test, beforeEach, after } from 'node:test'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { connect, migrate, q, social } from '@rssamplifier/db'; +import { normalizeXFeed, xSource } from '@rssamplifier/social'; + +import { crawlFeed } from '../src/crawl.js'; +import { submitOne } from '../src/submit.js'; +import { SOCIAL_MIN_INTERVAL, MIN_INTERVAL } from '../src/cadence.js'; + +/** + * The seam where a provider-collected source meets the ordinary crawler. + * + * The claim this whole design rests on is that an X source is a feed like any + * other from the moment it is collected — same dedupe, same scheduling, same + * storage, same everything downstream. These tests are that claim written down, + * because it is the sort of thing that is true when it is written and quietly + * stops being true two refactors later. + * + * The other half is what a *failure* must not do. `markCrawlFailure` retires a + * feed after ten consecutive failures, so anything that mistakes a rate limit, + * a provider outage or a missing runtime for a broken source would delete the + * whole X directory over an afternoon and leave no clue why. + */ + +let dir; +let db; + +beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), 'rsa-social-crawl-')); + db = connect({ url: `file:${join(dir, 'test.db')}` }); + await migrate(db); +}); + +after(async () => { + if (dir) await rm(dir, { recursive: true, force: true }); +}); + +/** @param {string[]} ids */ +function xPosts(ids) { + return ids.map((id, index) => ({ + id, + url: `https://x.com/OpenAI/status/${id}`, + text: `Post ${id}`, + createdAt: new Date(Date.now() - (index + 1) * 3_600_000).toISOString(), + author: { username: 'OpenAI', displayName: 'OpenAI Research' }, + })); +} + +/** A stand-in for `fetchXSource`, returning the same contract. */ +function collector(ids) { + return async (feed) => ({ + ok: true, + feedUrl: feed.feed_url, + feed: normalizeXFeed(xPosts(ids), { + spec: { mode: 'user', username: 'OpenAI' }, + url: String(feed.feed_url), + displayName: 'OpenAI Research', + }), + }); +} + +/** Create the row the crawler would be handed, and read it back. */ +async function seedXSource() { + const source = xSource('@OpenAI'); + const stored = await social.upsertSocialSource(db, { + network: 'x', + ref: source.ref, + slug: source.slug, + title: source.title, + feedUrl: source.url, + siteUrl: source.url, + }); + + const { rows } = await db.execute({ + sql: 'select * from feeds where id = ?', + args: [stored.id], + }); + return rows[0]; +} + +test('an X source is collected through the provider, never fetched', async () => { + const feed = await seedXSource(); + let resolved = 0; + + const result = await crawlFeed(db, feed, { + x: collector(['1', '2', '3']), + xRuntime: {}, + // If either of these is ever reached, the dispatch is wrong: there is no + // document at https://x.com/OpenAI to fetch or to scrape. + resolve: async () => { + resolved += 1; + throw new Error('the ordinary fetch must never see an X source'); + }, + scrape: async () => { + resolved += 1; + throw new Error('the scraper must never see an X source'); + }, + }); + + assert.equal(result.ok, true); + assert.equal(result.newItems, 3); + assert.equal(resolved, 0); + + const items = await q.itemsForFeed(db, String(feed.id), 50); + assert.deepEqual( + items.map((row) => row.guid).sort(), + ['x:1', 'x:2', 'x:3'], + ); +}); + +test('the same posts arriving again are stored once (AC-3)', async () => { + const feed = await seedXSource(); + + await crawlFeed(db, feed, { x: collector(['1', '2', '3']), xRuntime: {} }); + + const { rows: after } = await db.execute({ + sql: 'select * from feeds where id = ?', + args: [feed.id], + }); + + const second = await crawlFeed(db, after[0], { + x: collector(['1', '2', '3', '4']), + xRuntime: {}, + }); + + assert.equal(second.newItems, 1); + const items = await q.itemsForFeed(db, String(feed.id), 50); + assert.equal(items.length, 4); +}); + +test('an X source polls on the five-minute floor, not the hourly one (§17)', async () => { + const feed = await seedXSource(); + + // Posts an hour apart, which on the ordinary floor would still round up to 60. + await crawlFeed(db, feed, { x: collector(['1', '2', '3']), xRuntime: {} }); + + const { rows } = await db.execute({ + sql: 'select fetch_interval_minutes from feeds where id = ?', + args: [feed.id], + }); + + const interval = Number(rows[0].fetch_interval_minutes); + assert.ok(interval >= SOCIAL_MIN_INTERVAL, `interval ${interval}`); + assert.ok(interval < MIN_INTERVAL, `an X source should be able to go below ${MIN_INTERVAL}`); +}); + +test('a rate limit moves the schedule and touches no health column (§16)', async () => { + const feed = await seedXSource(); + await crawlFeed(db, feed, { x: collector(['1', '2']), xRuntime: {} }); + + const before = ( + await db.execute({ sql: 'select * from feeds where id = ?', args: [feed.id] }) + ).rows[0]; + + const result = await crawlFeed(db, before, { + xRuntime: {}, + x: async () => ({ ok: false, throttled: true, retryAfter: 300, error: 'rate-limited' }), + }); + + assert.equal(result.ok, false); + assert.equal(result.throttled, true); + + const after = (await db.execute({ sql: 'select * from feeds where id = ?', args: [feed.id] })) + .rows[0]; + + // The three columns that would eventually retire the feed. + assert.equal(Number(after.error_count), Number(before.error_count)); + assert.equal(after.status, before.status); + assert.equal(after.last_success_at, before.last_success_at); + + // And the items are exactly where they were — which is the whole of the + // stale-cache fallback (§40, AC-5). + const items = await q.itemsForFeed(db, String(feed.id), 50); + assert.equal(items.length, 2); +}); + +test('no X runtime is a reschedule, not a verdict on the source', async () => { + const feed = await seedXSource(); + + const result = await crawlFeed(db, feed, {}); + + assert.equal(result.ok, false); + assert.equal(result.throttled, true); + assert.equal(result.error, 'x-runtime-unavailable'); + + const after = (await db.execute({ sql: 'select * from feeds where id = ?', args: [feed.id] })) + .rows[0]; + assert.equal(Number(after.error_count), 0); + assert.notEqual(after.status, 'dead'); +}); + +test('the crawler is handed the columns it needs to dispatch on', async () => { + await seedXSource(); + const due = await q.dueFeeds(db, 10); + const row = due.find((feed) => feed.social_ref === 'x:user:openai'); + + assert.ok(row, 'an X source must appear in the due queue'); + assert.equal(row.social_network, 'x'); + assert.ok('social_config' in row); +}); + +test('an ordinary feed is untouched by any of this', async () => { + const { id } = await q.insertFeed(db, { + slug: 'a-blog', + feed_url: 'https://example.com/feed.xml', + title: 'A blog', + }); + + const { rows } = await db.execute({ sql: 'select * from feeds where id = ?', args: [id] }); + + let asked = 0; + const result = await crawlFeed(db, rows[0], { + xRuntime: {}, + x: async () => { + throw new Error('the X collector must never see an ordinary feed'); + }, + resolve: async () => { + asked += 1; + return { + ok: true, + feedUrl: 'https://example.com/feed.xml', + feed: { + title: 'A blog', + description: '', + siteUrl: 'https://example.com', + categories: [], + kind: 'blog', + items: [ + { + guid: 'p1', + url: 'https://example.com/p1', + title: 'A post', + summary: '', + contentHtml: '', + publishedAt: new Date().toISOString(), + categories: [], + audio: null, + }, + ], + }, + }; + }, + }); + + assert.equal(result.ok, true); + assert.equal(asked, 1); +}); + +test('submitting an X URL creates one source however it is spelled (§38)', async () => { + const first = await submitOne(db, 'https://twitter.com/OpenAI'); + const second = await submitOne(db, '@OpenAI'); + const third = await submitOne(db, 'https://x.com/openai/'); + + assert.equal(first.ok, true); + assert.equal(first.existing, false); + assert.equal(second.existing, true); + assert.equal(third.existing, true); + assert.equal(second.slug, first.slug); + + // And the caller is told where it lives, so a redirect lands on /x/OpenAI + // rather than on the slug the namespace exists to replace. + assert.equal(first.path, '/x/OpenAI'); + + const { rows } = await db.execute("select count(*) as n from feeds where social_network = 'x'"); + assert.equal(Number(rows[0].n), 1); +}); + +test('submitting a subreddit files it under Reddit rather than among the blogs', async () => { + const result = await submitOne(db, 'https://www.reddit.com/r/programming/.rss'); + + assert.equal(result.ok, true); + assert.equal(result.path, '/r/programming'); + + const row = await social.feedBySocialRef(db, 'r:sub:programming'); + assert.ok(row); + assert.equal(row.social_network, 'reddit'); + // Nothing was fetched: a public endpoint must not be a way to make our + // upstream do work (§37). The poller collects it on its next tick. + assert.equal(row.status, 'pending'); +}); diff --git a/packages/social/index.js b/packages/social/index.js new file mode 100644 index 0000000..d0fdc22 --- /dev/null +++ b/packages/social/index.js @@ -0,0 +1,56 @@ +/** + * Social sources: X/Twitter, which publishes no feeds, and Reddit, which does. + * + * The two are in one package because they answer the same question — "what is + * the canonical identity of a thing on a platform, and where does it live on + * this site?" — and differ only in how much work the answer takes. Reddit needs + * a URL rewritten; X needs three providers, a session pool and a normaliser. + */ + +export { + X_MODES, + parseXInput, + xRef, + xUrl, + xPath, + xSlug, + xTitle, + xSource, + xSpecFromRef, +} from './src/x/canonical.js'; + +export { normalizeXFeed, normalizeXPost } from './src/x/normalize.js'; + +export { + XError, + XRateLimited, + XAuthFailed, + XUnavailable, + XNoSuchSource, + classifyResponse, + retryAfterSeconds, +} from './src/x/errors.js'; + +export { XSessionPool, sessionsFromEnv, SESSION_STATES } from './src/x/sessions.js'; +export { XRegistry } from './src/x/registry.js'; +export { XBudget } from './src/x/providers/official.js'; +export { rsshubProvider } from './src/x/providers/rsshub.js'; +export { teapotProvider } from './src/x/providers/teapot.js'; +export { officialProvider } from './src/x/providers/official.js'; +export { postsFromRss } from './src/x/providers/fromRss.js'; + +export { createXRuntime, fetchXSource, xEnabled, readConfig } from './src/x/fetch.js'; + +export { + parseRedditInput, + redditRef, + redditFeedUrl, + redditSiteUrl, + redditPath, + redditSlug, + redditTitle, + redditSource, + redditSpecFromRef, +} from './src/reddit/canonical.js'; + +export { socialSourceFrom, socialPathFor, SOCIAL_NETWORKS } from './src/identify.js'; diff --git a/packages/social/package.json b/packages/social/package.json new file mode 100644 index 0000000..21497a5 --- /dev/null +++ b/packages/social/package.json @@ -0,0 +1,20 @@ +{ + "name": "@rssamplifier/social", + "version": "0.1.0", + "description": "Social sources: X/Twitter provider adapters, Reddit canonicalisation", + "type": "module", + "main": "index.js", + "exports": { + ".": "./index.js" + }, + "files": [ + "index.js", + "src" + ], + "scripts": { + "test": "node --test test/*.test.js" + }, + "dependencies": { + "@rssamplifier/feed": "workspace:*" + } +} diff --git a/packages/social/src/identify.js b/packages/social/src/identify.js new file mode 100644 index 0000000..e1bfd0e --- /dev/null +++ b/packages/social/src/identify.js @@ -0,0 +1,102 @@ +/** + * One question asked of both platforms: "is this URL one of ours?" + * + * The submit path, the importer and the crawler all need to recognise a social + * URL before they know which platform it belongs to, and none of them should + * have to try each parser in turn and remember the order. So the ordering lives + * here, once. + * + * It matters that this runs *before* the ordinary feed resolver. Reddit does + * publish RSS, so `https://www.reddit.com/r/programming/` resolves perfectly + * well as a plain feed — and lands as an untyped row at a slug of its own, + * which is exactly the 50,099-row outcome `reddit/canonical.js` describes. The + * difference between a subreddit at `/r/programming` and a subreddit filed + * among the blogs is entirely a matter of who looks at the URL first. + */ + +import { xSource } from './x/canonical.js'; +import { redditSource } from './reddit/canonical.js'; + +/** The networks that get a namespace of their own. */ +export const SOCIAL_NETWORKS = Object.freeze(['x', 'reddit']); + +/** + * Recognise a social source, or say it is not one. + * + * @param {unknown} input anything a person or an importer might supply + * @returns {{ + * network: 'x'|'reddit', + * ref: string, + * slug: string, + * title: string, + * path: string, + * feedUrl: string, + * siteUrl: string|null, + * }|null} + */ +export function socialSourceFrom(input) { + const reddit = redditSource(input); + if (reddit) { + return { + network: 'reddit', + ref: reddit.ref, + slug: reddit.slug, + title: reddit.title, + path: reddit.path, + feedUrl: reddit.feedUrl, + siteUrl: reddit.siteUrl, + }; + } + + const x = xSource(input); + if (x) { + return { + network: 'x', + ref: x.ref, + slug: x.slug, + title: x.title, + path: x.path, + // For X there is no document at this address and nothing ever fetches it. + // It is here because `feeds.feed_url` is `not null unique` and is the + // column every other surface reads to show a human where a feed came + // from — see the header of x/canonical.js. + feedUrl: x.url, + siteUrl: x.url, + }; + } + + return null; +} + +/** + * Where a stored row lives on this site, from its own columns. + * + * The fallback is `/{slug}`, which is every non-social feed and also any social + * row whose ref predates this code — so a caller can use this everywhere + * without checking whether a feed is social first. + * + * @param {{ social_network?: string|null, social_ref?: string|null, slug?: string }} feed + * @returns {string} + */ +export function socialPathFor(feed) { + const ref = feed?.social_ref ? String(feed.social_ref) : null; + const slug = String(feed?.slug ?? ''); + + if (ref?.startsWith('r:')) { + const [, mode, name] = ref.split(':'); + if (name) return mode === 'user' ? `/r/u/${name}` : `/r/${name}`; + } + + if (ref?.startsWith('x:')) { + const separator = ref.indexOf(':', 2); + const mode = ref.slice(2, separator); + const rest = ref.slice(separator + 1); + if (mode === 'user') return `/x/${rest}`; + if (mode === 'replies') return `/x/${rest}/replies`; + if (mode === 'media') return `/x/${rest}/media`; + if (mode === 'list') return `/x/list/${rest}`; + if (mode === 'search') return `/x/search?q=${encodeURIComponent(rest)}`; + } + + return `/${slug}`; +} diff --git a/packages/social/src/reddit/canonical.js b/packages/social/src/reddit/canonical.js new file mode 100644 index 0000000..3a0ccbc --- /dev/null +++ b/packages/social/src/reddit/canonical.js @@ -0,0 +1,224 @@ +/** + * Reddit, named the way Reddit names itself. + * + * Unlike X, Reddit needs no provider and no session: every subreddit publishes + * a real RSS document at `/r//.rss` and the ordinary crawler reads it. + * So there is nothing to adapt here — only something to *address*. + * + * That is the whole point of this file. On 2026-08-29 a bulk import put 50,099 + * subreddits into the directory, 41% of the entire crawl queue, and every one + * of them landed at a slug of its own alongside the blogs (`/programming`, + * `/askhistorians`). Two things are wrong with that. A subreddit is not a blog + * and a directory that files it as one is lying about its own contents; and + * `r/programming` has an obvious address that people already know how to type, + * which we were not serving. + * + * So `r:sub:programming` becomes the canonical identity, `/r/programming` the + * canonical URL, and the feed's own `/{slug}` page keeps working and points at + * it. Nothing is renamed and nothing is deleted: existing links survive, and + * the new address is the one search engines are told about. + * + * See `../x/canonical.js` for the same job on a platform that publishes no + * feeds at all — the two files share a shape on purpose. + */ + +/** Reddit's rule for a subreddit name: 3–21 of `[A-Za-z0-9_]`. */ +const SUBREDDIT = /^[A-Za-z0-9_]{3,21}$/; + +/** And for a username: 3–20, plus `-`, which subreddits may not contain. */ +const USERNAME = /^[A-Za-z0-9_-]{3,20}$/; + +/** Every host that is Reddit, including the ones the crawler will have stored. */ +const REDDIT_HOSTS = new Set([ + 'reddit.com', + 'www.reddit.com', + 'old.reddit.com', + 'new.reddit.com', + 'np.reddit.com', + 'i.reddit.com', + 'm.reddit.com', + 'amp.reddit.com', +]); + +/** + * Sort tabs, which are a view of a subreddit rather than a different one. + * + * `/r/programming/new/.rss` and `/r/programming/.rss` are the same community, + * and treating them as two sources would poll Reddit twice for one thing. The + * sort is dropped rather than preserved: a directory subscribes to a community, + * not to an ordering of it. + */ +const SORTS = new Set(['new', 'hot', 'top', 'rising', 'controversial', 'best', 'gilded']); + +/** + * Read whatever a person pasted and say which Reddit source they meant. + * + * @param {unknown} input + * @returns {{ mode: 'sub'|'user', name: string }|null} + */ +export function parseRedditInput(input) { + const raw = String(input ?? '').trim(); + if (!raw) return null; + + // The shorthands people actually type, before anything tries to parse a URL: + // `r/programming`, `/r/programming`, `u/spez`, `/u/spez`. + const short = /^\/?(r|u|user)\/([A-Za-z0-9_-]{3,21})\/?$/i.exec(raw); + if (short) { + const name = short[2]; + if (short[1].toLowerCase() === 'r') { + return SUBREDDIT.test(name) ? { mode: 'sub', name } : null; + } + return USERNAME.test(name) ? { mode: 'user', name } : null; + } + + const withScheme = /^https?:\/\//i.test(raw) ? raw : `https://${raw.replace(/^\/+/, '')}`; + + let url; + try { + url = new URL(withScheme); + } catch { + return null; + } + + if (!REDDIT_HOSTS.has(url.hostname.toLowerCase())) return null; + + // `.rss`, `.json` and a trailing `/` are all spellings of the same path. + const segments = url.pathname + .replace(/\.(rss|json|xml)$/i, '') + .split('/') + .filter(Boolean); + + if (segments.length < 2) return null; + + const kind = segments[0].toLowerCase(); + const name = segments[1]; + + if (kind === 'r') { + if (!SUBREDDIT.test(name)) return null; + // A third segment is either a sort we drop or a specific post we decline — + // a permalink is something to read, not something to subscribe to. + const tail = segments[2]?.toLowerCase(); + if (tail && !SORTS.has(tail)) return null; + return { mode: 'sub', name }; + } + + if (kind === 'u' || kind === 'user') { + return USERNAME.test(name) ? { mode: 'user', name } : null; + } + + return null; +} + +/** + * Our key for the source. Case-folded, because Reddit's own URLs are + * case-insensitive and `/r/Programming` and `/r/programming` are one community. + * + * @param {{ mode: string, name: string }} spec + * @returns {string|null} + */ +export function redditRef(spec) { + if (!spec?.name) return null; + if (spec.mode === 'sub') return `r:sub:${spec.name.toLowerCase()}`; + if (spec.mode === 'user') return `r:user:${spec.name.toLowerCase()}`; + return null; +} + +/** + * The RSS document Reddit actually publishes — this one *is* fetched, unlike + * an X source's canonical URL. + * + * `www.` rather than `old.` deliberately: the old host is a compatibility + * shim Reddit has said it will retire, and a directory that pins 50,000 feeds + * to it inherits that deadline. + * + * @param {{ mode: string, name: string }} spec + * @returns {string|null} + */ +export function redditFeedUrl(spec) { + if (!spec?.name) return null; + if (spec.mode === 'sub') return `https://www.reddit.com/r/${spec.name}/.rss`; + if (spec.mode === 'user') return `https://www.reddit.com/user/${spec.name}/.rss`; + return null; +} + +/** The human page on Reddit's side. */ +export function redditSiteUrl(spec) { + if (!spec?.name) return null; + if (spec.mode === 'sub') return `https://www.reddit.com/r/${spec.name}/`; + if (spec.mode === 'user') return `https://www.reddit.com/user/${spec.name}/`; + return null; +} + +/** + * Where it lives on this site. A user goes under `/r/u/…` rather than `/u/…` + * so that one prefix holds all of Reddit — which is the whole ask. + * + * @param {{ mode: string, name: string }} spec + * @returns {string|null} + */ +export function redditPath(spec) { + if (!spec?.name) return null; + if (spec.mode === 'sub') return `/r/${spec.name}`; + if (spec.mode === 'user') return `/r/u/${spec.name}`; + return null; +} + +/** A title for a source whose first crawl has not landed yet. */ +export function redditTitle(spec) { + if (spec?.mode === 'sub') return `r/${spec.name}`; + if (spec?.mode === 'user') return `u/${spec.name} on Reddit`; + return 'Reddit'; +} + +/** The directory slug, on the same rules as an X source's. */ +export function redditSlug(spec) { + const ref = redditRef(spec); + if (!ref) return null; + return ref + .replace(/^r:sub:/, 'r-') + .replace(/^r:user:/, 'r-u-') + .replace(/[^a-z0-9-]+/g, '-') + .replace(/-{2,}/g, '-') + .replace(/^-+|-+$/g, ''); +} + +/** + * Everything a source row needs, from one pasted string. + * + * @param {unknown} input + * @returns {{ + * mode: string, name: string, ref: string, feedUrl: string, siteUrl: string, + * path: string, slug: string, title: string + * }|null} + */ +export function redditSource(input) { + const spec = parseRedditInput(input); + if (!spec) return null; + + const ref = redditRef(spec); + const feedUrl = redditFeedUrl(spec); + const path = redditPath(spec); + const slug = redditSlug(spec); + if (!ref || !feedUrl || !path || !slug) return null; + + return { + ...spec, + ref, + feedUrl, + siteUrl: redditSiteUrl(spec), + path, + slug, + title: redditTitle(spec), + }; +} + +/** + * Rebuild the spec from a stored ref. + * + * @param {unknown} ref + * @returns {{ mode: 'sub'|'user', name: string }|null} + */ +export function redditSpecFromRef(ref) { + const match = /^r:(sub|user):([A-Za-z0-9_-]{3,21})$/.exec(String(ref ?? '')); + return match ? { mode: /** @type {'sub'|'user'} */ (match[1]), name: match[2] } : null; +} diff --git a/packages/social/src/x/canonical.js b/packages/social/src/x/canonical.js new file mode 100644 index 0000000..ddfa17d --- /dev/null +++ b/packages/social/src/x/canonical.js @@ -0,0 +1,341 @@ +/** + * What an X source *is*, before anybody has fetched anything. + * + * Every other source in this directory has an identity handed to it: a feed's + * identity is the URL its document lives at, and two submissions of the same + * URL collide on `feeds.feed_url` without anyone having to think about it. X + * has no such document, so the identity has to be constructed — and constructed + * the same way every time, or a thousand readers subscribing to @OpenAI become + * a thousand separate polling jobs against a platform that rate-limits per + * account (see §37/§38 of the PRD, and `markHostThrottled` in queries.js for + * what that costs when it goes wrong on a smaller platform). + * + * So this module is the whole of the answer to "are these two requests the same + * source?", and it is deliberately the only place that decides. It produces + * two strings per source: + * + * - a **ref** (`x:user:openai`) — our own key, lowercase and free of anything a + * URL parser could disagree about. This is what the unique index is on. + * - a **URL** (`https://x.com/OpenAI`) — the canonical public address of the + * thing on X's side. It goes in `feeds.feed_url` because that column is + * `not null unique` and every surface in this codebase expects a feed to have + * an http(s) address it could show a human. Nothing ever fetches it: the + * crawler routes an X source to a provider instead. It is an identifier that + * happens to also be a working link, which is the best kind. + * + * The display casing is preserved separately (`username`), because @OpenAI is + * how the account writes its own name and lowercasing it in the page title + * would be us correcting a publisher's spelling of themselves. + */ + +/** The five things a reader can point us at. Mirrors `XFeedMode` in the PRD. */ +export const X_MODES = Object.freeze(['user', 'replies', 'media', 'search', 'list']); + +/** + * X's own rule for a handle: 1–15 of `[A-Za-z0-9_]`. + * + * Worth pinning rather than accepting anything short, because this string is + * interpolated into an upstream provider's path. A handle that cannot contain a + * slash or a dot cannot walk out of the route it was put in. + */ +const HANDLE = /^[A-Za-z0-9_]{1,15}$/; + +/** A list id is a snowflake: digits, and long enough not to be a typo. */ +const LIST_ID = /^[0-9]{6,25}$/; + +/** + * Hosts that mean X. `twitter.com` is not a legacy alias to be tidied away — + * it is still what most links in the wild say, and what most people paste. + */ +const X_HOSTS = new Set([ + 'x.com', + 'www.x.com', + 'mobile.x.com', + 'twitter.com', + 'www.twitter.com', + 'mobile.twitter.com', + 'm.twitter.com', + 'nitter.net', +]); + +/** + * Path segments that are X's own furniture rather than somebody's handle. + * + * `https://x.com/search?q=…` and `https://x.com/i/lists/123` are real addresses + * whose first segment looks exactly like a username, and reading them as one + * would create a source called @search that can never return a post. The list + * is short on purpose: it names the paths this module actually routes plus the + * few reserved words that would otherwise be silently accepted as accounts. + */ +const NOT_A_HANDLE = new Set([ + 'i', + 'search', + // Ours rather than X's. `/x/list/…` and `/x/status` are fixed segments on + // this site, so an account genuinely named @list or @status could be stored + // and then never addressed — a row nothing can reach. Refusing it up front is + // the smaller loss, and it is two handles. + 'list', + 'status', + 'home', + 'explore', + 'notifications', + 'messages', + 'settings', + 'compose', + 'intent', + 'hashtag', + 'login', + 'signup', + 'about', + 'tos', + 'privacy', +]); + +/** + * Read whatever a person pasted and say which X source they meant. + * + * Accepts, per §6.2 and §7: a bare handle, an @handle, a profile URL on any of + * the hosts above, the `/with_replies` and `/media` tabs, a search URL, and a + * list URL. Returns null for anything else — including a link to a single post, + * which is a thing to read rather than a thing to subscribe to. + * + * @param {unknown} input + * @returns {{ mode: string, username?: string, query?: string, listId?: string }|null} + */ +export function parseXInput(input) { + const raw = String(input ?? '').trim(); + if (!raw) return null; + + // Bare handle or @handle. Checked before the URL parse because `OpenAI` is + // not a URL and `new URL()` on it throws rather than declining. + const bare = raw.replace(/^@/, ''); + if (HANDLE.test(bare) && !raw.includes('/') && !raw.includes(':')) { + return { mode: 'user', username: bare }; + } + + // `r/`-style shorthand has no X equivalent, but `x/OpenAI` and `@x.com` + // handles do turn up in pasted text, so a scheme-less URL gets one. + const withScheme = /^https?:\/\//i.test(raw) ? raw : `https://${raw.replace(/^\/+/, '')}`; + + let url; + try { + url = new URL(withScheme); + } catch { + return null; + } + + if (!X_HOSTS.has(url.hostname.toLowerCase())) return null; + + const segments = url.pathname.split('/').filter(Boolean); + if (segments.length === 0) return null; + + const first = segments[0].toLowerCase(); + + // A search, in either of the two spellings X itself uses. + if (first === 'search') { + const query = url.searchParams.get('q') ?? url.searchParams.get('query') ?? ''; + return query.trim() ? { mode: 'search', query: query.trim() } : null; + } + + // A list: /i/lists/:id, and the older /:owner/lists/:slug which we cannot + // resolve to an id without asking X, so it is declined rather than guessed. + if (first === 'i') { + const listId = segments[1]?.toLowerCase() === 'lists' ? segments[2] : null; + return listId && LIST_ID.test(listId) ? { mode: 'list', listId } : null; + } + + if (NOT_A_HANDLE.has(first)) return null; + if (!HANDLE.test(segments[0])) return null; + + const username = segments[0]; + const tab = segments[1]?.toLowerCase(); + + // A post URL (/:user/status/:id) is deliberately *not* a source. Somebody + // pasting one wants to read that post, and turning it into a subscription to + // the whole account is a different thing from what they asked for. + if (tab === 'status' || tab === 'statuses') return null; + + if (tab === 'with_replies' || tab === 'replies') return { mode: 'replies', username }; + if (tab === 'media' || tab === 'photo') return { mode: 'media', username }; + if (!tab) return { mode: 'user', username }; + + // Any other tab (/likes, /following, /highlights) is a page about the account + // rather than a feed of it. + return null; +} + +/** + * Our key for a source. Two requests that produce the same ref are the same + * upstream collector, and the unique index on `feeds.social_ref` enforces it. + * + * A search is keyed on its *exact* query text rather than a slug of it, because + * `from:OpenAI lang:en` and `from:openai lang:en` are the same search to X but + * `bitcoin` and `bitcoin ETF` are not, and no normalisation is safe across + * an operator syntax we deliberately do not reimplement (§28). + * + * @param {{ mode: string, username?: string, query?: string, listId?: string }} spec + * @returns {string|null} + */ +export function xRef(spec) { + if (!spec) return null; + switch (spec.mode) { + case 'user': + return spec.username ? `x:user:${spec.username.toLowerCase()}` : null; + case 'replies': + return spec.username ? `x:replies:${spec.username.toLowerCase()}` : null; + case 'media': + return spec.username ? `x:media:${spec.username.toLowerCase()}` : null; + case 'search': + return spec.query ? `x:search:${spec.query.trim().toLowerCase()}` : null; + case 'list': + return spec.listId ? `x:list:${spec.listId}` : null; + default: + return null; + } +} + +/** + * The canonical address on X's side — what goes in `feeds.feed_url`, what the + * page links out to, and what an item's `link` is relative to. + * + * @param {{ mode: string, username?: string, query?: string, listId?: string }} spec + * @returns {string|null} + */ +export function xUrl(spec) { + if (!spec) return null; + switch (spec.mode) { + case 'user': + return `https://x.com/${spec.username}`; + case 'replies': + return `https://x.com/${spec.username}/with_replies`; + case 'media': + return `https://x.com/${spec.username}/media`; + case 'search': + return `https://x.com/search?q=${encodeURIComponent(spec.query)}&f=live`; + case 'list': + return `https://x.com/i/lists/${spec.listId}`; + default: + return null; + } +} + +/** + * Where the source lives on *this* site. + * + * The public URL a reader subscribes to, and the one thing in this file that + * must never change when the collection method does (AC-2). A provider name + * appears nowhere in it. + * + * @param {{ mode: string, username?: string, query?: string, listId?: string }} spec + * @returns {string|null} path, no extension — `.rss`/`.atom`/`.json` append + */ +export function xPath(spec) { + if (!spec) return null; + switch (spec.mode) { + case 'user': + return `/x/${spec.username}`; + case 'replies': + return `/x/${spec.username}/replies`; + case 'media': + return `/x/${spec.username}/media`; + case 'search': + // The query rides in the query string rather than the path. §5 shows a + // slugged form and §28 the query-string one; only the second can carry + // `from:OpenAI lang:en` without inventing an escaping scheme, and a + // reader's subscription URL is not the place to invent one. + return `/x/search?q=${encodeURIComponent(spec.query)}`; + case 'list': + return `/x/list/${spec.listId}`; + default: + return null; + } +} + +/** + * A title for the source, used when the first crawl has not yet learned the + * account's display name. + * + * @param {{ mode: string, username?: string, query?: string, listId?: string }} spec + * @returns {string} + */ +export function xTitle(spec) { + switch (spec?.mode) { + case 'user': + return `@${spec.username} on X`; + case 'replies': + return `@${spec.username} on X — replies`; + case 'media': + return `@${spec.username} on X — media`; + case 'search': + return `X search: ${spec.query}`; + case 'list': + return `X list ${spec.listId}`; + default: + return 'X'; + } +} + +/** + * The directory slug for an X source. + * + * X sources keep a slug like every other feed, because `/{slug}` is the + * permanent identity of a row in this directory and half the site's internals + * (the reader, alerts, the queue, sitemaps) address a feed that way. `/x/…` is + * the *canonical* public address on top of it — see the canonical link on the + * feed page — not a replacement for the row's own name. + * + * @param {{ mode: string, username?: string, query?: string, listId?: string }} spec + * @returns {string|null} + */ +export function xSlug(spec) { + const ref = xRef(spec); + if (!ref) return null; + return ref + .replace(/^x:/, 'x-') + .replace(/:/g, '-') + .replace(/[^a-z0-9-]+/g, '-') + .replace(/-{2,}/g, '-') + .replace(/^-+|-+$/g, ''); +} + +/** + * Everything a source row needs, from one pasted string. + * + * @param {unknown} input + * @returns {{ + * mode: string, username?: string, query?: string, listId?: string, + * ref: string, url: string, path: string, slug: string, title: string + * }|null} + */ +export function xSource(input) { + const spec = parseXInput(input); + if (!spec) return null; + + const ref = xRef(spec); + const url = xUrl(spec); + const path = xPath(spec); + const slug = xSlug(spec); + if (!ref || !url || !path || !slug) return null; + + return { ...spec, ref, url, path, slug, title: xTitle(spec) }; +} + +/** + * Rebuild the spec from a stored ref, for the crawler — which holds a row, not + * the string somebody once pasted. + * + * @param {unknown} ref + * @returns {{ mode: string, username?: string, query?: string, listId?: string }|null} + */ +export function xSpecFromRef(ref) { + const raw = String(ref ?? ''); + const match = /^x:([a-z]+):([\s\S]+)$/.exec(raw); + if (!match) return null; + + const [, mode, rest] = match; + if (!X_MODES.includes(mode)) return null; + + if (mode === 'search') return { mode, query: rest }; + if (mode === 'list') return LIST_ID.test(rest) ? { mode, listId: rest } : null; + return HANDLE.test(rest) ? { mode, username: rest } : null; +} diff --git a/packages/social/src/x/errors.js b/packages/social/src/x/errors.js new file mode 100644 index 0000000..3f6d5cb --- /dev/null +++ b/packages/social/src/x/errors.js @@ -0,0 +1,137 @@ +/** + * Telling apart the four things that go wrong upstream (§16, §40). + * + * They look alike from the outside and want opposite responses, which is why + * they are types rather than strings. Getting this wrong is expensive in a + * specific way: `markCrawlFailure` walks a backoff ladder and retires a feed at + * ten consecutive failures, so recording a rate limit as a failure would retire + * every X source we have within a day of X getting busy — the platform-scale + * version of the mistake `markThrottled` exists to prevent for ordinary hosts. + * + * - **XRateLimited** — come back later. Nothing is broken; the schedule moves + * and no health column is touched. + * - **XAuthFailed** — this *session* is broken. Take it out of rotation and try + * another one; the provider and the source are both fine. + * - **XUnavailable** — this *provider* is broken. Fail over; the source is fine. + * - **XNoSuchSource** — the account or list does not exist, or is protected. + * The only one of the four that is genuinely about the source, and the only + * one that should ever count against its health. + * + * A protected account is deliberately in the last group and deliberately not + * retried harder: we do not attempt private timelines (§4, §42). + */ + +export class XError extends Error { + /** + * @param {string} message + * @param {{ provider?: string, sessionId?: string, status?: number, cause?: unknown }} [meta] + */ + constructor(message, meta = {}) { + super(message); + this.name = new.target.name; + this.provider = meta.provider ?? null; + this.sessionId = meta.sessionId ?? null; + this.status = meta.status ?? null; + if (meta.cause !== undefined) this.cause = meta.cause; + } +} + +export class XRateLimited extends XError { + /** + * @param {string} message + * @param {{ retryAfter?: number|null }} [meta] + */ + constructor(message, meta = {}) { + super(message, meta); + /** Seconds the server asked for, when it said. */ + this.retryAfter = meta.retryAfter ?? null; + } +} + +export class XAuthFailed extends XError {} +export class XUnavailable extends XError {} +export class XNoSuchSource extends XError {} + +/** + * What an HTTP response from an upstream provider means. + * + * The status codes are the reliable half. The body sniffing below is the + * unreliable half and is treated as such — it only ever *upgrades* a generic + * failure into a specific one, never the reverse, because every unofficial + * provider phrases these differently and a phrase we do not recognise must + * still fail safely as "provider unavailable" rather than silently as success. + * + * @param {{ status: number, headers?: Headers, body?: string, provider?: string, sessionId?: string }} res + * @returns {XError|null} null when the response is fine + */ +export function classifyResponse(res) { + const meta = { provider: res.provider, sessionId: res.sessionId, status: res.status }; + const body = String(res.body ?? '').slice(0, 2000); + const lower = body.toLowerCase(); + + if (res.status === 429) { + return new XRateLimited('rate-limited', { + ...meta, + retryAfter: retryAfterSeconds(res.headers?.get?.('retry-after')), + }); + } + + if (res.status === 401 || res.status === 403) { + // 403 is ambiguous on purpose upstream: it is both "your session is no + // longer valid" and "this account is protected". The body decides, and when + // it says nothing the session is blamed — because retrying a good session + // against a protected account costs one wasted request, while retiring a + // good session costs every source that shares it. + if (/protected|private account|not authorized to view/.test(lower)) { + return new XNoSuchSource('protected-account', meta); + } + return new XAuthFailed(`auth-failed-${res.status}`, meta); + } + + if (res.status === 404) return new XNoSuchSource('no-such-source', meta); + + if (res.status === 503 && res.headers?.get?.('retry-after')) { + return new XRateLimited('unavailable-retry-after', { + ...meta, + retryAfter: retryAfterSeconds(res.headers.get('retry-after')), + }); + } + + if (res.status >= 500 || res.status === 0) { + return new XUnavailable(`upstream-${res.status}`, meta); + } + + if (res.status >= 400) return new XUnavailable(`upstream-${res.status}`, meta); + + // A 200 that is really a failure. RSSHub in particular answers 200 with an + // error document when its own upstream refused, and an unrecognised error + // page parses to zero items — which the caller would otherwise read as "this + // account posted nothing", the quietest possible way to lose a feed. + if (/rate ?limit|too many requests/.test(lower)) { + return new XRateLimited('rate-limited-body', meta); + } + if (/could not authenticate|bad authentication|login required|checkpoint|denied by /.test(lower)) { + return new XAuthFailed('auth-failed-body', meta); + } + + return null; +} + +/** + * `Retry-After` in either of its two spellings. + * + * @param {string|null|undefined} header + * @returns {number|null} seconds + */ +export function retryAfterSeconds(header) { + if (!header) return null; + + const raw = String(header).trim(); + const seconds = Number(raw); + if (Number.isFinite(seconds) && seconds >= 0) return Math.min(seconds, 86_400); + + const when = Date.parse(raw); + if (Number.isNaN(when)) return null; + + return Math.max(0, Math.min(Math.round((when - Date.now()) / 1000), 86_400)); +} diff --git a/packages/social/src/x/fetch.js b/packages/social/src/x/fetch.js new file mode 100644 index 0000000..16e1887 --- /dev/null +++ b/packages/social/src/x/fetch.js @@ -0,0 +1,184 @@ +/** + * The one function the crawler calls, and the only one it needs. + * + * `crawlFeed` already knows how to hold two ingestion methods apart — a feed it + * fetches and parses, and a page it scrapes — and both hand back the same + * shape: `{ ok, feedUrl, feed }`, or a failure that says whether it was a + * throttle. This returns that same shape from a stack of X providers, which is + * what lets an X source travel every line of the existing pipeline: dedupe, + * interval learning, keyword extraction, author credit, FTS, alerts, sitemaps. + * + * The alternative — the `sources` and `items` tables the PRD sketches in §20 — + * would be a second copy of all of that, and §30's "topic code must not contain + * X-specific provider logic" would then be a rule to enforce rather than a + * property of the design. + * + * **A failure here never empties a feed.** Nothing in this file deletes an item, + * and `crawlFeed` writes items only on success, so an outage leaves yesterday's + * posts exactly where they were and the public route keeps serving them (§40, + * AC-5). That is the whole of the stale-cache fallback: there is no cache to + * fall back to, because the database was always the thing being served. + */ + +import { xSpecFromRef } from './canonical.js'; +import { normalizeXFeed } from './normalize.js'; +import { XRegistry } from './registry.js'; +import { XSessionPool, sessionsFromEnv } from './sessions.js'; + +/** + * How long to wait after an anomaly, in minutes. Long enough that a hundred + * queued sources do not all rediscover the same upstream problem inside a tick, + * short enough that a real recovery is picked up within the hour. + */ +const ANOMALY_MINUTES = 20; + +/** + * Build the runtime once, at boot. + * + * The registry and the pool both carry state that only means something when it + * accumulates — a provider's failure streak, a session's cooldown — so a fresh + * one per crawl would be a system with no memory, rediscovering every outage on + * every feed. + * + * @param {{ + * env?: Record, + * providerStore?: object, + * sessionStore?: object, + * onEvent?: (event: string, fields: object) => void, + * }} [opts] + */ +export async function createXRuntime(opts = {}) { + const env = opts.env ?? process.env; + + const registry = new XRegistry({ env, store: opts.providerStore ?? null }); + const sessions = new XSessionPool(sessionsFromEnv(env), { + store: opts.sessionStore ?? null, + cooldownSeconds: Number(env.X_SESSION_COOLDOWN_SECONDS) || undefined, + }); + + await Promise.all([registry.hydrate(), sessions.hydrate()]); + + return { registry, sessions, onEvent: opts.onEvent ?? (() => {}) }; +} + +/** + * Is the X integration switched on at all? + * + * `X_ENABLED=false` is the kill switch §42 asks for: it stops collection dead + * without touching a route, so every existing `/x/…` feed keeps serving what it + * already has and nothing new is fetched. + * + * @param {Record} [env] + */ +export function xEnabled(env = process.env) { + return String(env.X_ENABLED ?? 'false').toLowerCase() !== 'false'; +} + +/** + * Collect one X source, in the shape `crawlFeed` expects. + * + * @param {{ + * social_ref?: string, feed_url?: string, social_config?: string|null, + * item_count?: number, + * }} feed the row + * @param {{ + * runtime: Awaited>, + * limit?: number, + * signal?: AbortSignal, + * }} opts + * @returns {Promise<{ ok: boolean, feedUrl?: string, feed?: object, error?: string, + * throttled?: boolean, retryAfter?: number|null }>} + */ +export async function fetchXSource(feed, opts) { + const spec = xSpecFromRef(feed?.social_ref); + if (!spec) return { ok: false, error: 'invalid-x-ref' }; + + const { registry, sessions, onEvent } = opts.runtime; + const config = readConfig(feed?.social_config); + + let result; + try { + result = await registry.fetch( + { + mode: spec.mode, + username: spec.username, + query: spec.query, + listId: spec.listId, + limit: opts.limit ?? 50, + }, + { sessions, onEvent, signal: opts.signal }, + ); + } catch (error) { + // A rate limit is a schedule instruction, not evidence about the account. + // Returning it as `throttled` routes it to `markThrottled`, which moves + // `next_fetch_at` and leaves every health column alone — the same treatment + // an ordinary publisher's 429 gets, and for the same reason (§16). + if (error?.name === 'XRateLimited') { + return { ok: false, throttled: true, retryAfter: error.retryAfter ?? null, error: 'rate-limited' }; + } + // A deleted, suspended or protected account is the one failure that is + // genuinely about the source, so it is the one that counts against it. + return { ok: false, error: String(error?.message ?? 'x-fetch-failed').slice(0, 200) }; + } + + const posts = result.posts ?? []; + + // An account that has always been empty is a real thing; an account that had + // posts yesterday and none today is almost always an upstream that answered + // 200 with a page it could not fill. The second is treated as a throttle + // rather than as news, because believing it would let one bad response + // decide, through the content signature, that this feed is now unchanging + // and worth crawling once a day (§16, "empty-result anomalies"). + if (posts.length === 0 && Number(feed?.item_count ?? 0) > 0) { + onEvent('x.fetch.failed', { provider: result.provider, error: 'empty-result' }); + return { ok: false, throttled: true, retryAfter: ANOMALY_MINUTES * 60, error: 'empty-result' }; + } + + return { + ok: true, + feedUrl: feed.feed_url, + feed: normalizeXFeed(posts, { + spec, + url: String(feed.feed_url), + includeReplies: config.includeReplies, + includeReposts: config.includeReposts, + includeQuotes: config.includeQuotes, + displayName: result.displayName ?? null, + avatarUrl: result.avatarUrl ?? null, + }), + }; +} + +/** + * The per-source toggles of §6.3, with the PRD's defaults. + * + * Applied while the collected posts are turned into items, which means they + * decide what is *stored*: a source with `includeReposts: false` never has a + * repost in `feed_items`, and changing the toggle takes effect from the next + * crawl rather than retroactively. + * + * That is a deliberate limitation and the alternative was considered. Storing + * everything and filtering at render would let one row serve both a with- and a + * without-reposts view, but only if `feed_items` carried a column saying which + * items were reposts — a schema change for a toggle almost nobody moves, on the + * largest table in the database. Replies, the one split that people do want + * both of, do not need it: `/x/:user` and `/x/:user/replies` are different refs + * and therefore different rows, so both exist at once. + * + * @param {string|null|undefined} raw + */ +export function readConfig(raw) { + const defaults = { includeReplies: false, includeReposts: true, includeQuotes: true }; + if (!raw) return defaults; + + try { + const parsed = JSON.parse(String(raw)); + return { + includeReplies: parsed?.includeReplies ?? defaults.includeReplies, + includeReposts: parsed?.includeReposts ?? defaults.includeReposts, + includeQuotes: parsed?.includeQuotes ?? defaults.includeQuotes, + }; + } catch { + return defaults; + } +} diff --git a/packages/social/src/x/normalize.js b/packages/social/src/x/normalize.js new file mode 100644 index 0000000..8a282ee --- /dev/null +++ b/packages/social/src/x/normalize.js @@ -0,0 +1,373 @@ +/** + * An X post, written as one of our items. + * + * This is the seam the whole feature turns on. Above it, three providers each + * return posts in whatever shape they happen to speak; below it, nothing in the + * codebase knows that X exists — the crawler stores these rows, the topic + * river merges them with blog posts, and `buildSyndication` renders them into + * RSS, Atom, JSON Feed, Markdown and playlists without a single branch for + * where they came from (§30, AC-8). + * + * The target shape is `parseFeed`'s, not the database's. That is deliberate: + * `crawlFeed` takes a parsed document, and handing it one means an X source + * travels the same code path as a blog — dedupe, interval learning, keyword + * extraction, author credit, FTS indexing — rather than needing its own copy of + * each. The cost is that this file has to speak camelCase; the alternative is a + * second ingestion pipeline, which is a far larger cost. + * + * Three shapes need care, because each one is a post that is partly about + * *another* post: + * + * - a **repost** carries no text of its own, so an item built from its own + * fields is blank. The original's content is rendered under a line naming who + * reposted it (§26). + * - a **quote** is two posts in one item, and both halves have to survive or + * the item reads as a non-sequitur (§27). + * - a **reply** is a post with a parent we may never have seen. It keeps its own + * canonical URL and says nothing about what it is replying to (§25). + */ + +import { summarize } from '@rssamplifier/feed'; + +import { xTitle } from './canonical.js'; + +/** How much of a post becomes its title before an ellipsis. */ +const TITLE_CHARS = 110; + +/** + * Turn a provider's posts into a feed document. + * + * @param {import('./types.js').XPost[]} posts + * @param {{ + * spec: { mode: string, username?: string, query?: string, listId?: string }, + * url: string, + * includeReplies?: boolean, + * includeReposts?: boolean, + * includeQuotes?: boolean, + * displayName?: string|null, + * avatarUrl?: string|null, + * }} context + * @returns {{ title: string, description: string, siteUrl: string, language: null, + * imageUrl: string|null, categories: string[], kind: string, items: object[] }} + */ +export function normalizeXFeed(posts, context) { + const { + spec, + url, + includeReplies = spec.mode === 'replies', + includeReposts = true, + includeQuotes = true, + displayName = null, + avatarUrl = null, + } = context; + + const kept = (Array.isArray(posts) ? posts : []).filter((post) => + keep(post, { includeReplies, includeReposts, includeQuotes, mode: spec.mode }), + ); + + // A quoted post that also arrived in its own right is one post, not two + // (§27). The quote carries the whole of the quoted text already, so the + // standalone copy is the one to drop — dropping the quote instead would lose + // the commentary, which is the half somebody followed this account for. + const quoted = new Set(kept.map((post) => post.quotedPost?.id).filter(Boolean)); + const deduped = kept.filter((post) => !quoted.has(post.id) || post.quotedPostId); + + return { + title: channelTitle(spec, displayName), + description: channelDescription(spec, displayName), + siteUrl: url, + // X states no language on a timeline, and guessing one from the posts would + // put a label on the feed that the publisher never claimed. + language: null, + imageUrl: avatarUrl ?? null, + categories: [], + // A timeline is writing, and `blog` is what this directory calls writing. + // Not `news`: `isNewsroom` wants two independent signals before it moves a + // feed out of blogs, and "posts often" is only one of them. + kind: 'blog', + items: deduped.map((post) => normalizeXPost(post)).filter(Boolean), + }; +} + +/** + * One post as one item. + * + * @param {import('./types.js').XPost} post + * @returns {object|null} + */ +export function normalizeXPost(post) { + if (!post?.id) return null; + + const source = post.repostOf ?? post; + const text = String(source.text ?? '').trim(); + const author = post.author?.username ? `@${post.author.username}` : null; + + return { + // `x:`, never the URL (§19). A URL changes when a handle does — X + // serves /anyone/status/:id for the same post — so a URL-keyed dedupe + // re-ingests an account's whole timeline the day it renames itself. + guid: `x:${post.id}`, + url: post.url ?? postUrl(post), + title: itemTitle(post, source, text), + summary: summarize(plainSummary(post, source, text), 400), + contentHtml: itemHtml(post, source, text), + author: post.author?.displayName + ? `${post.author.displayName} (${author})` + : (author ?? null), + publishedAt: post.createdAt ?? null, + // The first image, so a card and a thumbnail have something to show. Video + // contributes its preview frame rather than nothing. + imageUrl: firstImage(source) ?? firstImage(post) ?? null, + categories: hashtags(text), + // X carries no enclosures. Video exists but it is served from a signed, + // short-lived URL that no podcast client could still play tomorrow, so + // nothing is attached and the media is rendered inline instead (§24). + audio: null, + }; +} + +/** + * Should this post be in the document at all? + * + * @param {import('./types.js').XPost} post + * @param {{ includeReplies: boolean, includeReposts: boolean, includeQuotes: boolean, mode: string }} opts + * @returns {boolean} + */ +function keep(post, opts) { + if (!post?.id) return false; + + // The replies *feed* is the one place a reply is the point (§25). Everywhere + // else the default is off, because an account's replies are mostly one half + // of a conversation and read as fragments without the other half. + if (post.replyToId && !opts.includeReplies && opts.mode !== 'replies') return false; + if (post.repostOfId && !opts.includeReposts) return false; + if (post.quotedPostId && !opts.includeQuotes) return false; + + return true; +} + +/** + * A title for something that has none. + * + * X posts have no titles, and every format we render wants one — a reader's + * list view is titles and nothing else. So the first line of the post becomes + * it, prefixed with the handle, which is what §19 specifies and what every + * other Twitter-to-RSS bridge has converged on for the same reason: in a topic + * river a bare fragment of prose gives no clue who said it. + * + * @param {import('./types.js').XPost} post + * @param {import('./types.js').XPost} source the reposted original, or the post + * @param {string} text + * @returns {string} + */ +function itemTitle(post, source, text) { + const who = post.author?.username ?? 'x'; + const reposted = post.repostOfId && source.author?.username; + + const body = collapse(text); + const clipped = body.length > TITLE_CHARS ? `${body.slice(0, TITLE_CHARS).trimEnd()}…` : body; + + if (reposted) { + return `${who} reposted @${source.author.username}: ${clipped || '(media)'}`; + } + return `${who}: ${clipped || '(media)'}`; +} + +/** + * The item's prose, for a summary and for search — media markup excluded, since + * an `` tag in a search index is noise. + */ +function plainSummary(post, source, text) { + const parts = []; + if (post.repostOfId && source.author?.username) { + parts.push(`Reposted @${source.author.username}:`); + } + parts.push(text); + + const quote = post.quotedPost; + if (quote) { + const handle = quote.author?.username ? `@${quote.author.username}` : 'a post'; + parts.push(`Quoting ${handle}: ${String(quote.text ?? '').trim()}`); + } + + return parts.filter(Boolean).join('\n\n'); +} + +/** + * The rendered item body. + * + * Escaped rather than sanitised, because none of this is markup to begin with: + * a post is plain text, and the only tags in the output are ones this function + * wrote. The one exception is a link, which is built from a URL we escape into + * an href rather than from anything the post supplied as HTML. + */ +function itemHtml(post, source, text) { + const blocks = []; + + if (post.repostOfId && source.author?.username) { + blocks.push( + `

${escapeHtml(post.author?.username ?? 'They')} reposted ` + + `@${escapeHtml( + source.author.username, + )}:

`, + ); + } + + if (text) blocks.push(`

${linkify(text)}

`); + + blocks.push(...mediaHtml(source)); + // A repost's own media is the original's, but a quote-with-media carries its + // own, so both are offered and duplicates are collapsed by the caller's set. + if (source !== post) blocks.push(...mediaHtml(post)); + + const quote = post.quotedPost; + if (quote) { + const handle = quote.author?.username; + const cite = handle + ? `@${escapeHtml(handle)}` + : 'a post'; + const link = quote.url ?? (quote.id ? `https://x.com/i/status/${quote.id}` : null); + blocks.push( + '
' + + `

${cite}

` + + `

${linkify(String(quote.text ?? '').trim())}

` + + mediaHtml(quote).join('') + + (link ? `

${escapeHtml(link)}

` : '') + + '
', + ); + } + + return blocks.join('\n') || '

(no text)

'; +} + +/** + * Images and video previews, as markup a reader will actually render. + * + * Video gets its poster frame wrapped in a link to the post rather than a + * `