From 33540b1022506ff40e664cc13a6b9d63cc3e60e6 Mon Sep 17 00:00:00 2001 From: Rayhan Atmadja Date: Tue, 27 Oct 2026 15:25:03 +0700 Subject: [PATCH] fix: low latency polling (refetch + rate limit) --- CLAUDE.md | 59 ++++++++++++------- src/cmd/cli.ts | 4 +- src/engine/interceptor.ts | 64 ++++++++++++++++++--- src/engine/rate-limit.ts | 62 ++++++++++++++++++++ src/engine/stream.ts | 117 +++++++++++++++++++++++++++++--------- src/schemas/stream.ts | 4 +- 6 files changed, 251 insertions(+), 59 deletions(-) create mode 100644 src/engine/rate-limit.ts diff --git a/CLAUDE.md b/CLAUDE.md index 3baf183..4124f56 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,32 +4,49 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Commands -Package manager is pnpm (`packageManager: pnpm@10.33.0`). +Package manager is pnpm (`packageManager: pnpm@10.33.0`). `.nvmrc` pins Node 24; `engines` requires `>=22.12.0`. `pnpm install` runs `husky` through `prepare`, which installs the git hooks. -- `pnpm dev` — run the CLI directly via tsx (`src/cmd/cli.ts`) -- `pnpm watch` — run the CLI with tsx watch mode -- `pnpm check` — typecheck only (`tsc --noEmit`) -- `pnpm build` — compile to `dist/` (`tsc && tsc-alias`, needed to resolve `@/*` path aliases in output) -- `pnpm start` — run the compiled CLI from `dist/cmd/cli.js` +- `pnpm dev ` — run the CLI from source via tsx, e.g. `pnpm dev search -k "typescript" -l en` +- `pnpm watch ` — same, restarting on file change +- `pnpm check` — typecheck only (`tsc --noEmit`). It is also the pre-commit hook and the first CI step. +- `pnpm build` — `clean` + `tsc` + `tsc-alias`. `tsc-alias` is required because it rewrites the `@/*` imports in `dist/`. +- `pnpm start ` — run `dist/cmd/cli.js` +- Smoke test, as CI runs it: `node dist/cmd/cli.js --version` and `node dist/cmd/cli.js --help` -There is no test suite or lint config in this repo. +There is no test suite, no lint config, and therefore no single-test command. Verify with `pnpm check`. For changes to scraping, run against real X with a valid token. `parseTweetsFromResponse` (`src/engine/parser.ts`) is pure, so you can check it offline by passing it a saved SearchTimeline JSON body. -Example CLI invocation: -``` -pnpm dev search -k "typescript" --auth-token $AUTH_TOKEN -l en -``` -`AUTH_TOKEN` can be set via env var instead of `--auth-token`. `.env` is loaded automatically (`dotenv/config`) at CLI entry. +`AUTH_TOKEN` can come from the environment or from `--auth-token`. `.env` is loaded by `dotenv/config` at CLI entry and is gitignored. Never print or commit the token. + +## Git and CI + +- Commits must follow Conventional Commits. The `commit-msg` hook runs commitlint, and the PR-title workflow checks PR titles the same way. +- Merges to `main` run release-please (`release-please-config.json`). It owns the version bump and changelog, based on `feat`/`fix`/`perf`/`refactor` commits. Don't edit the version in `package.json` or `.release-please-manifest.json` by hand. +- The release workflow publishes to npm. It uses the Node version from `.nvmrc`. +- CI (`.github/workflows/ci.yml`) runs `pnpm install --frozen-lockfile`, `pnpm check`, `pnpm build`, the smoke test, and `npm pack --dry-run`, on Node 22, 24, and 26. ## Architecture -XStream is a CLI tool that scrapes X/Twitter search results by driving a real (stealth) headless browser rather than calling the X API directly. +Pipeline: `cmd/cli.ts` → `engine/stream.ts` (`XStreamer`) → `engine/browser.ts`, `engine/interceptor.ts`, `engine/parser.ts`, `engine/rate-limit.ts` → a `TweetSink`. + +**Entry (`src/cmd/cli.ts`).** Commander parses the flags. `StreamSchema` (zod, `src/schemas/stream.ts`) validates and coerces them into `StreamConfig`. Validation errors exit with code 2 and runtime errors with code 1. The sink is chosen here: `NdjsonStdoutSink` (`src/sinks/ndjson.ts`), or the file sink defined in `cli.ts`, which appends. The first SIGINT or SIGTERM calls `XStreamer.stop()` for a graceful shutdown. A second signal forces exit. + +**Session (`XStreamer.start` in `engine/stream.ts`).** It launches the browser, attaches the interceptor to the page, and navigates to the live search URL. If the URL ends up on `/login` or `/i/flow`, the token is invalid. The run then waits for `[data-testid="cellInnerDiv"]` and enters `pollLoop`. The `finally` block closes the browser and the sink. + +**Tweets come from two channels that share one listener.** +1. Passive: `XInterceptor` (`engine/interceptor.ts`) listens to every page `response` and matches `/i/api/graphql/` with `SearchTimeline`. It parses the body with `parser.ts`, which walks `data.search_by_raw_query.search_timeline.timeline.instructions[].entries[]`. It also saves the successful request as a `RequestTemplate` (URL, method, headers, body). Cookies, `sec-*` headers, and anything in `UNREPLAYABLE_HEADERS` are dropped. +2. Active: `pollLoop` replays that template with `page.evaluate(fetch(..., { credentials: "include" }))`. The page supplies the cookies and the `ct0` CSRF header, so Node never handles auth itself. The refetch's response goes through the same interceptor listener. The `refetch` result carries only status and headers, which feed rate limiting and failure counting. + +**Recovery.** A refetch status other than 200 or 429 counts as a failure. After `MAX_REFETCH_FAILURES` (3) in a row, `pollLoop` clears the template and reloads the page. The reload captures a fresh template from its SearchTimeline response. A 429 is not a failure. It is handled by pacing. + +**Pacing (`engine/rate-limit.ts`, `RateLimitTracker`).** It reads `x-rate-limit-{limit,remaining,reset}`. It spreads requests evenly over the remaining budget until reset, keeping 2 requests in reserve. On a 429 it blocks until the reset time. Without reset headers it backs off exponentially, from 60s up to 5 minutes. `--interval` is the floor, and `nextDelay` adds up to 10% jitter. + +**Dedup and output.** `handleTweets` drops IDs already in `seenIds`, which holds at most 10,000 entries and evicts the oldest. Each batch is sorted oldest first. Once `--max` is reached, the run stops. -Flow: `cmd/cli.ts` (Commander-based CLI, validates args, builds `StreamConfig` via the Zod schema in `schemas/stream.ts`) → `engine/stream.ts` (`XStreamer`, orchestrates the session) → `engine/browser.ts` (`launchBrowser`, launches `playwright-extra` Chromium with the stealth plugin and injects the `auth_token` cookie for `.x.com`) → `engine/interceptor.ts` (`XInterceptor`, attaches a Playwright `page.on("response")` listener) → `engine/parser.ts` (`parseTweetsFromResponse`, walks X's internal GraphQL `SearchTimeline` response shape to extract `Tweet` objects). +## Invariants and gotchas -Key points: -- Authentication is cookie-based: the caller supplies their own X `auth_token` cookie value (from a logged-in browser session), not an API key. -- Tweet extraction works by intercepting the browser's own network responses to X's internal GraphQL endpoint (`/i/api/graphql/...SearchTimeline`), not by scraping the DOM or calling a public API. Parsing logic in `parser.ts` is tightly coupled to that undocumented response shape (`data.search_by_raw_query.search_timeline.timeline.instructions[].entries[]...tweet_results.result`) and will break if X changes it. -- Path alias `@/*` maps to `src/*` (see `tsconfig.json`); `tsc-alias` rewrites these in the compiled `dist/` output since `tsc` alone does not. -- `XStreamer.start()` currently navigates to the search URL, waits for the feed to appear, and tears down the browser — the interceptor collects tweets into memory but nothing yet reads/exports `capturedTweets` from the CLI. -- `types.ts`'s `Crawler` interface declares fields (`output`, `max`, `batchSize`, `batchDelay`, `db`) not yet wired into `StreamConfig`/`XStreamer` — these represent planned but unimplemented options (result limits, batching/pagination via cursor, output destination). -- Each module (`browser.ts`, `interceptor.ts`, `parser.ts`, `stream.ts`) creates its own `tslog` `Logger` instance scoped by name; log level is `DEBUG` unless `NODE_ENV=prod`, in which case it's `INFO`. +- stdout carries only NDJSON tweets. Logs go to stderr through the tslog transport in `src/logger.ts`. Don't `console.log` or write to stdout anywhere else. +- Each module gets its logger from `createLogger(name)`. The level is DEBUG unless `NODE_ENV=prod`, which sets INFO. +- The project compiles to CommonJS (there is no `"type": "module"`). `cli.ts` uses `__dirname` and `browser.ts` uses `require.resolve`. Switching to ESM means changing both. +- `browser.ts` downloads Chromium on first run if the Playwright executable is missing. It runs through `playwright-extra` with the stealth plugin. +- `extractCursor` in `parser.ts` is exported but unused, so cursor pagination isn't wired up. +- `README.md` lags the code. Its `--interval` minimum and default (10 and 30) differ from `schemas/stream.ts` (2 and 5), and its description of fixed-interval page reloads predates the in-page refetch. Trust the source. diff --git a/src/cmd/cli.ts b/src/cmd/cli.ts index c2b54f4..266dec5 100644 --- a/src/cmd/cli.ts +++ b/src/cmd/cli.ts @@ -98,8 +98,8 @@ program .option("--no-headless", "Run with visible browser UI (default: headless)") .option( "--interval ", - "Polling interval between refetches, in seconds (min 10)", - "30", + "Minimum seconds between polls (min 2). Actual spacing adapts to X's rate limit.", + "5", ) .option("--max ", "Stop after emitting this many tweets") .option( diff --git a/src/engine/interceptor.ts b/src/engine/interceptor.ts index 71038c1..8d15a65 100644 --- a/src/engine/interceptor.ts +++ b/src/engine/interceptor.ts @@ -1,17 +1,63 @@ import { Tweet } from "@/types"; -import { Page } from "playwright"; +import { Page, Request as PWRequest } from "playwright"; import { createLogger } from "@/logger"; import { parseTweetsFromResponse } from "@/engine/parser"; +import { RateLimitTracker } from "./rate-limit"; + +export interface RequestTemplate { + url: string; + method: string; + headers: Record; + body: Buffer | null; +} + +const UNREPLAYABLE_HEADERS = new Set([ + "cookie", + "host", + "user-agent", + "referer", + "origin", + "content-length", + "connection", + "accept-encoding", +]); + +async function buildTemplate(request: PWRequest): Promise { + const rawHeaders = await request.allHeaders(); + const headers = Object.fromEntries( + Object.entries(rawHeaders).filter( + ([name]) => + !name.startsWith(":") && + !name.startsWith("sec-") && + !UNREPLAYABLE_HEADERS.has(name), + ), + ); + return { + url: request.url(), + method: request.method(), + headers, + body: request.postDataBuffer(), + }; +} export class XInterceptor { private isAttached: boolean = false; + private template: RequestTemplate | null = null; private readonly logger = createLogger("X Interceptor"); constructor( private readonly onTweets: (tweets: Tweet[]) => void, - private readonly onRateLimit?: () => void, + private readonly rateLimit: RateLimitTracker, ) {} + public getTemplate(): RequestTemplate | null { + return this.template; + } + + public invalidTemplate(): void { + this.template = null; + } + public attach(page: Page): void { if (this.isAttached) return; @@ -26,13 +72,14 @@ export class XInterceptor { const status = response.status(); const headers = response.headers(); - const remaining = headers["x-rate-limit-remaining"]; - const limit = headers["x-rate-limit-limit"]; - const reset = headers["x-rate-limit-reset"]; + this.rateLimit.update(headers, status); - if (remaining !== undefined) { + const snapshot = this.rateLimit.getSnapshot(); + if (snapshot) { this.logger.debug( - `Rate limit: ${remaining}/${limit}, resets ${reset}`, + `Rate limit snapshot: ${snapshot.remaining}/${snapshot.limit}, resets at ${new Date( + snapshot.resetAt, + ).toISOString()}`, ); } @@ -40,7 +87,6 @@ export class XInterceptor { this.logger.warn( `Rate limited (429) on SearchTimeline request: ${url}`, ); - this.onRateLimit?.(); return; } @@ -51,6 +97,8 @@ export class XInterceptor { return; } + this.template = await buildTemplate(response.request()); + const contentType = headers["content-type"] || ""; if (!contentType.includes("application/json")) { return; diff --git a/src/engine/rate-limit.ts b/src/engine/rate-limit.ts new file mode 100644 index 0000000..2ae554a --- /dev/null +++ b/src/engine/rate-limit.ts @@ -0,0 +1,62 @@ +const RESERVE_REQUESTS = 2; +const RESET_SLACK_MS = 1_000; +const HEADERLESS_429_BASE_MS = 60_000; +const HEADERLESS_429_MAX_MS = 5 * 60_000; + +export interface RateLimitSnapshot { + limit: number; + remaining: number; + resetAt: number; +} + +export class RateLimitTracker { + private snapshot: RateLimitSnapshot | null = null; + private blockedUntil = 0; + private headerless429Ms = HEADERLESS_429_BASE_MS; + + public update(headers: Record, status: number): void { + const limit = Number(headers["x-rate-limit-limit"]); + const remaining = Number(headers["x-rate-limit-remaining"]); + const reset = Number(headers["x-rate-limit-reset"]); + const hasHeaders = + Number.isFinite(limit) && + Number.isFinite(remaining) && + Number.isFinite(reset); + + if (hasHeaders) { + this.snapshot = { limit, remaining, resetAt: reset * 1000 }; + } + + if (status === 429) { + if (hasHeaders) { + this.blockedUntil = reset * 1000; + } else { + this.blockedUntil = Date.now() + this.headerless429Ms; + this.headerless429Ms = Math.min( + this.headerless429Ms * 2, + HEADERLESS_429_MAX_MS, + ); + } + } else if (status === 200) { + this.headerless429Ms = HEADERLESS_429_BASE_MS; + } + } + + public getSnapshot(): RateLimitSnapshot | null { + return this.snapshot; + } + + public nextDelay(minMs: number, now: number = Date.now()): number { + if (this.blockedUntil > now) { + return this.blockedUntil - now + RESET_SLACK_MS; + } + + const snapshot = this.snapshot; + if (!snapshot || snapshot.resetAt <= now) return minMs; + + const usable = snapshot.remaining - RESERVE_REQUESTS; + if (usable <= 0) return snapshot.resetAt - now + RESET_SLACK_MS; + + return Math.max(minMs, Math.ceil((snapshot.resetAt - now) / usable)); + } +} diff --git a/src/engine/stream.ts b/src/engine/stream.ts index 814fb95..367e244 100644 --- a/src/engine/stream.ts +++ b/src/engine/stream.ts @@ -4,10 +4,17 @@ import { Browser, BrowserContext, Page } from "playwright"; import { Logger } from "tslog"; import { createLogger } from "@/logger"; import { launchBrowser } from "./browser"; -import { XInterceptor } from "./interceptor"; +import { RequestTemplate, XInterceptor } from "./interceptor"; +import { RateLimitTracker } from "./rate-limit"; -const MAX_INTERVAL_MS = 5 * 60_000; const MAX_SEEN_IDS = 10_000; +const MAX_REFETCH_FAILURES = 3; +const JITTER_RATIO = 0.1; + +interface RefetchResult { + status: number; + headers: Record; +} function sleep(ms: number, signal: AbortSignal): Promise { return new Promise((resolve) => { @@ -33,17 +40,18 @@ export class XStreamer { private readonly interceptor: XInterceptor; private readonly seenIds: Set = new Set(); private readonly abortController = new AbortController(); - private intervalMs: number; private emittedCount = 0; + private readonly rateLimit = new RateLimitTracker(); + private readonly minIntervalMs: number; constructor(streamConfig: StreamConfig, sink: TweetSink) { this.streamConfig = streamConfig; this.sink = sink; - this.intervalMs = streamConfig.interval * 1000; + this.minIntervalMs = streamConfig.interval * 1000; this.logger = createLogger("XStreamer"); this.interceptor = new XInterceptor( (tweets) => this.handleTweets(tweets), - () => this.handleRateLimit(), + this.rateLimit, ); } @@ -109,26 +117,93 @@ export class XStreamer { private async pollLoop(page: Page): Promise { const signal = this.abortController.signal; + let failures = 0; while (!signal.aborted) { - await sleep(this.intervalMs, signal); + const delay = this.nextDelay(); + this.logger.debug(`Next poll in ${delay}ms`); + await sleep(delay, signal); if (signal.aborted) break; - await page - .reload({ waitUntil: "domcontentloaded", timeout: 60000 }) - .catch((error) => { - this.logger.warn(`Failed to reload search page: ${error}`); - }); - if (signal.aborted) break; + const template = this.interceptor.getTemplate(); + if (template) { + const result = await this.refetch(page, template); + this.rateLimit.update(result.headers, result.status); - await page - .waitForSelector('[data-testid="cellInnerDiv"]', { timeout: 30000 }) - .catch(() => { - this.logger.warn("Tweet feed not detected after reload."); + if (result.status === 200 || result.status === 429) { + failures = 0; + continue; + } + + failures++; + this.logger.warn( + `In-page refetch failed (status ${result.status}), attempt ${failures}/${MAX_REFETCH_FAILURES}.`, + ); + if (failures < MAX_REFETCH_FAILURES) continue; + } + + this.logger.info( + "Reloading search page to capture a fresh request template.", + ); + failures = 0; + this.interceptor.invalidTemplate(); + await this.reloadPage(page); + } + } + + private nextDelay(): number { + const base = this.rateLimit.nextDelay(this.minIntervalMs); + return Math.round(base * (1 + Math.random() * JITTER_RATIO)); + } + + private async refetch( + page: Page, + template: RequestTemplate, + ): Promise { + const payload = { + url: template.url, + method: template.method, + headers: template.headers, + body: template.body?.toString("utf8") ?? null, + }; + + try { + return await page.evaluate(async ({ url, method, headers, body }) => { + const csrf = /(?:^|;\s*)ct0=([^;]+)/.exec(document.cookie)?.[1]; + if (csrf) headers["x-csrf-token"] = csrf; + + const response = await fetch(url, { + method, + headers, + body: body ?? undefined, + credentials: "include", }); + await response.text(); // wait for the full body before resolving + return { + status: response.status, + headers: Object.fromEntries(response.headers.entries()), + }; + }, payload); + } catch (error) { + this.logger.warn(`In-page refetch threw: ${error}`); + return { status: 0, headers: {} }; } } + private async reloadPage(page: Page): Promise { + await page + .reload({ waitUntil: "domcontentloaded", timeout: 60000 }) + .catch((error) => { + this.logger.warn(`Failed to reload search page: ${error}`); + }); + + await page + .waitForSelector('[data-testid="cellInnerDiv"]', { timeout: 30000 }) + .catch(() => { + this.logger.warn("Tweet feed not detected after reload."); + }); + } + private assertAuthenticated(page: Page): void { const url = page.url(); if (url.includes("/login") || url.includes("/i/flow")) { @@ -176,16 +251,6 @@ export class XStreamer { this.seenIds.add(id); } - private handleRateLimit(): void { - const nextInterval = Math.min(this.intervalMs * 2, MAX_INTERVAL_MS); - if (nextInterval !== this.intervalMs) { - this.intervalMs = nextInterval; - this.logger.warn( - `Rate limited by X. Backing off polling interval to ${this.intervalMs}ms.`, - ); - } - } - private buildSearchUrl(keywords: string, lang?: string): string { const query = lang ? `${keywords} lang:${lang}` : keywords; const encoded = encodeURIComponent(query); diff --git a/src/schemas/stream.ts b/src/schemas/stream.ts index d8d9cb4..2178530 100644 --- a/src/schemas/stream.ts +++ b/src/schemas/stream.ts @@ -9,8 +9,8 @@ export const StreamSchema = z.object({ headless: z.boolean().default(true), interval: z.coerce .number() - .min(10, "Interval must be at least 10 seconds") - .default(30), + .min(2, "Interval must be at least 2 seconds") + .default(5), max: z.coerce.number().int().positive().optional(), output: z.string().default("stdout"), });