Skip to docs content
Browse and search
On this page

Build a production consumer

Route known operations, reconnect with backoff, and make every side effect safe to repeat.

Run the 5-minute quickstart
Open in chat
ChatGPTClaudeCursor
Reviewed against product code on September 2, 2026

Connect

Use wss://ws-iad.tweetstream.io/ws in the USA and wss://ws-global.tweetstream.io/ws elsewhere. The server responds with tweetstream.v1; it does not echo the authentication-token protocol. The example reconnects with exponential backoff. Short sessions keep increasing the delay; 30 seconds of uptime resets it.

Connect with Node.js

typescript
import WebSocket from "ws";
 
type StreamEvent = {
  t?: string;
  op?: string;
  d?: {
    author?: { handle?: string };
    detected?: unknown;
    text?: string;
  };
};
 
const apiKey = process.env.TWEETSTREAM_API_KEY;
if (!apiKey) {
  throw new Error("Missing TWEETSTREAM_API_KEY");
}
 
let retry = 0;
let healthyTimer: ReturnType<typeof setTimeout> | undefined;
let reconnectTimer: ReturnType<typeof setTimeout> | undefined;
 
function scheduleReconnect(reason: string) {
  if (reconnectTimer) return;
 
  const delayMs = Math.min(30_000, 1_000 * 2 ** retry) + Math.floor(Math.random() * 500);
  retry += 1;
  console.warn(`Reconnecting in ${delayMs}ms: ${reason}`);
  reconnectTimer = setTimeout(() => {
    reconnectTimer = undefined;
    connect();
  }, delayMs);
}
 
function connect() {
  const ws = new WebSocket("wss://ws-global.tweetstream.io/ws", [
    "tweetstream.v1",
    `tweetstream.auth.token.${apiKey}`,
  ]);
 
  ws.on("open", () => {
    if (reconnectTimer) clearTimeout(reconnectTimer);
    reconnectTimer = undefined;
    healthyTimer = setTimeout(() => {
      retry = 0;
      healthyTimer = undefined;
    }, 30_000);
    console.log("TweetStream connected");
  });
 
  ws.on("message", (raw) => {
    const event = JSON.parse(raw.toString()) as StreamEvent;
 
    if (event.t === "tweet" && event.op === "content") {
      const tweet = event.d;
      console.log(tweet?.author?.handle, tweet?.text);
    }
 
    if (event.t === "tweet" && event.op === "meta") {
      console.log("enrichment", event.d?.detected);
    }
  });
 
  ws.on("close", (code, reason) => {
    if (healthyTimer) clearTimeout(healthyTimer);
    healthyTimer = undefined;
    scheduleReconnect(`close ${code}: ${reason.toString()}`);
  });
 
  ws.on("unexpected-response", (_request, response) => {
    console.error("Connection rejected", response.statusCode, response.statusMessage);
    response.resume();
    if (response.statusCode === 429 || response.statusCode === 503) {
      scheduleReconnect(`HTTP ${response.statusCode}`);
      return;
    }
    process.exitCode = 1;
  });
 
  ws.on("error", (error) => {
    console.error("WebSocket error", error.message);
  });
}
 
connect();

Prepare for production

  • Reconnect with backoff after close or network error.
  • Connect only to the documented TweetStream endpoints and use the authentication methods above. TweetStream handles routing; no infrastructure-specific headers are required.
  • Use (platform, tweetId) as the post key, not the envelope id. Read platform from author.platform on content and d.platform on later operations.
  • On older operations without d.platform, use an included author platform or default to twitter.
  • Apply every tweet/update idempotently. If it includes ref, replace the stored reference with that snapshot and follow ref.subtweet for the next referenced post.
  • Fingerprint the full envelope before dropping an exact replay.
  • Treat meta as late-arriving enrichment for a tweet you may already have routed.
  • Store receipts and projections in indexed durable storage. Define a retention or archival policy for your replay window and storage budget; never remove pending work.
  • Log failed actions without payload data and retry them with backoff. Keep the same receipt ID as the idempotency key.
  • Track active WebSocket connection count against your plan limit.

Measure signal latency

Record local receipt time at the start of the message callback. For X/Twitter events, decode the tweet snowflake timestamp to measure publication to receipt. Keep the host clock synced with NTP for meaningful wall-clock results. Use a monotonic clock for processing segments after receipt.

  • Report the watchlist, consumer region, UTC test window, sample size, p50, and p95.
  • Record reconnects, missing events, clock-sync status, and every exclusion rule.
  • Report cold-start and post-reconnect samples separately instead of folding them into warm-path results.
  • Compare results only when publication point, receipt point, geography, sample, and percentile use the same boundary.
SegmentStartStop
Publication to receiptX snowflake timestampLocal time captured at the top of the socket callback
Receipt to decisionLocal socket receiptStrategy and risk decision ready
Decision to venue acknowledgementRisk-approved decisionSeparate venue response or rejection

Measure snowflake latency

typescript
type ContentEvent = {
  t?: string;
  op?: string;
  d?: {
    tweetId?: string;
  };
};
 
const TWITTER_EPOCH_MS = 1_288_834_974_657n;
 
function tweetIdToTimestampMs(tweetId: string) {
  const id = BigInt(tweetId);
  return Number((id >> 22n) + TWITTER_EPOCH_MS);
}
 
function measureSnowflakeLatency(tweetId: string, arrivedAtMs: number) {
  const tweetedAtMs = tweetIdToTimestampMs(tweetId);
  return arrivedAtMs - tweetedAtMs;
}
 
ws.on("message", (raw) => {
  const arrivedAtMs = Date.now();
  const event = JSON.parse(raw.toString()) as ContentEvent;
  const tweetId = event.d?.tweetId;
 
  if (event.t === "tweet" && event.op === "content" && tweetId) {
    console.log("publication-to-receipt ms", measureSnowflakeLatency(tweetId, arrivedAtMs));
  }
});