// TweetStream complete example: connect, reconnect, handle every event, and measure latency.
// Node.js 22.18 or later runs this TypeScript file directly. No dependencies.
//
//   TWEETSTREAM_API_KEY=YOUR_API_KEY node tweetstream-example.ts
//
// Optional settings:
//   TWEETSTREAM_ACCOUNTS=handle1,handle2               add these accounts before connecting
//   TWEETSTREAM_WS_URL=wss://ws-iad.tweetstream.io/ws  use the US endpoint

// The messages this program reads, limited to the fields it uses. Every field of every event:
// https://tweetstream.io/docs/asyncapi.json

// Every server message is a v1 envelope. Route on t (channel) and op (operation); d is the payload.
type Envelope = { v: 1; t: string; op: string; id?: string; ts: number; d: unknown };

// Posts name their author by handle. Some account events use username instead.
type Account = { id?: string; handle?: string; username?: string; platform?: string };

type PostKind = "post" | "quote" | "reply" | "retweet";

// A post that this one quotes, replies to, or reposts. It can reference another through subtweet.
type Reference = {
  type: Exclude<PostKind, "post">;
  tweetId: string;
  author?: Account;
  text?: string;
  subtweet?: Reference;
};

// tweet/content: a new post. author.platform is "twitter" when it is missing.
type PostContent = {
  tweetId: string;
  text: string;
  createdAt: number; // Unix milliseconds
  kind: PostKind;
  author: Account;
  ref?: Reference;
};

// tweet/update: only the fields that changed. textUpdateType "completion" delivers the full text
// of a post that arrived shortened; it is not an edit.
type PostUpdate = Partial<Omit<PostContent, "createdAt">> & {
  tweetId: string;
  platform?: string;
  textUpdateType?: "completion";
};

// tweet/meta: a complete snapshot of the tokens, markets, and image text found in a post.
type PostMeta = {
  tweetId: string;
  platform?: string;
  detected?: {
    tokens?: Array<{ symbol?: string; contract?: string; priceUsd?: number }>;
    prediction?: Array<{ title?: string; priceUsd?: number }>;
  };
  ocr?: { text: string };
};

// tweet/delete, tweet/pin, and tweet/unpin.
type PostEvent = { tweetId: string; platform?: string; author?: Account };

type Coin = { symbol?: string; address?: string };

// The payload of each t/op pair this program handles.
type Payloads = {
  "control/auth_pong": Record<string, never>;
  "tweet/content": PostContent;
  "tweet/update": PostUpdate;
  "tweet/meta": PostMeta;
  "tweet/delete": PostEvent;
  "tweet/pin": PostEvent;
  "tweet/unpin": PostEvent;
  "account/follow": { actor: Account; target: Account };
  "account/unfollow": { actor: Account; target: Account };
  "account/profile_update": { actor: Account; changes: Record<string, unknown> };
  "account/affiliate_update": { organization: Account; member: Account; action: string };
  "account/callout": { caller: Account; token: Coin };
  "account/callout_update": { updateOf: string; token?: Coin };
  "account/thesis": { author: Account; coin: Coin; thesis: { text: string } };
  "account/token_verified": { coin: Coin };
  "news/article": { source?: { name?: string }; title?: string };
  "charity/added": { name: string };
  "control/twitter_handles_result": {
    results: Array<{ input: string; state: string }>;
    error: string | null;
  };
};
type Pair = keyof Payloads;
type Handler<P> = (d: P, receivedAt: number) => void;

// What this program keeps for each post.
type StoredPost = {
  content: Partial<PostUpdate> | null;
  meta: PostMeta | null;
  deleted: boolean;
  pinned: boolean;
  hasContent: boolean;
};

const API_KEY = process.env.TWEETSTREAM_API_KEY?.trim();
const DASHBOARD = "https://tweetstream.io/dashboard?tab=api";
if (!API_KEY) {
  console.error(`Set TWEETSTREAM_API_KEY. Copy your key from ${DASHBOARD}`);
  process.exit(1);
}
const WS_URL = process.env.TWEETSTREAM_WS_URL || "wss://ws-global.tweetstream.io/ws";
const API_URL = "https://api.tweetstream.io";
const ACCOUNTS = (process.env.TWEETSTREAM_ACCOUNTS ?? "")
  .split(",")
  .map((handle) => handle.trim().replace(/^@/, ""))
  .filter(Boolean);
const TWITTER_EPOCH_MS = 1288834974657n;

const posts = new Map<string, StoredPost>(); // "platform:tweetId" -> stored post
const counts = new Map<string, number>(); // "t/op" -> number of events seen
const latencies: number[] = []; // publish-to-receipt ms for the most recent 10,000 new posts
let attempt = 0;
let hinted = false;

const log = (label: string, text: string) => console.log(`${label.padEnd(9)} ${text}`);
const at = (account: Account | undefined) =>
  `@${String(account?.handle ?? account?.username ?? "unknown").replace(/^@/, "")}`;
const clip = (text: unknown, max = 80) => {
  const flat = String(text ?? "").replace(/\s+/g, " ").trim();
  return JSON.stringify(flat.length > max ? `${flat.slice(0, max - 1)}…` : flat);
};

// Posts are keyed by (platform, tweetId). Content carries the platform on its author;
// later operations carry it on d.platform.
const platformOf = (d: PostEvent) => d.platform ?? d.author?.platform ?? "twitter";
function postFor(d: PostEvent, platform = platformOf(d)): StoredPost {
  const key = `${platform}:${d.tweetId}`;
  let post = posts.get(key);
  if (!post) {
    post = { content: null, deleted: false, hasContent: false, meta: null, pinned: false };
    posts.set(key, post);
    if (posts.size > 10_000) {
      const oldest = posts.keys().next().value;
      if (oldest !== undefined) posts.delete(oldest); // bound memory
    }
  }
  return post;
}

// Updates carry only the fields that changed, so merge them over what you stored.
const merge = (base: Partial<PostUpdate> | null, patch: Partial<PostUpdate> | null) => ({
  ...base,
  ...patch,
  author: { ...base?.author, ...patch?.author },
});

function publishedAtMs(platform: string, d: PostContent): number | null {
  // X post IDs encode their publish time: (id >> 22) + the Twitter epoch.
  if (platform === "twitter" && /^\d+$/.test(d.tweetId)) {
    return Number((BigInt(d.tweetId) >> 22n) + TWITTER_EPOCH_MS);
  }
  return typeof d.createdAt === "number" ? d.createdAt : null;
}

function referenceChain(ref: Reference | undefined): string {
  const links: string[] = [];
  // A referenced post can reference another one through ref.subtweet.
  for (let r = ref, depth = 0; r && depth < 8; r = r.subtweet, depth++) {
    links.push(`${r.type ?? "ref"} ${at(r.author)} ${r.tweetId} ${clip(r.text, 40)}`);
  }
  return links.join(" -> ");
}

function describeMeta(meta: PostMeta): string {
  const price = (value: number | undefined, prefix = "") => (value == null ? "" : ` ${prefix}${value}`);
  const tokens = meta.detected?.tokens ?? [];
  const markets = meta.detected?.prediction ?? [];
  const parts: string[] = [];
  if (tokens.length) {
    parts.push(`tokens ${tokens.map((t) => `${t.symbol ?? t.contract}${price(t.priceUsd, "$")}`).join(", ")}`);
  }
  if (markets.length) {
    parts.push(`markets ${markets.map((m) => `${clip(m.title, 40)}${price(m.priceUsd)}`).join(", ")}`);
  }
  if (meta.ocr?.text) parts.push(`ocr ${clip(meta.ocr.text, 40)}`);
  return parts.join(" | ") || "no detections";
}

// Delete, pin, and unpin change the stored post's state and keep its content.
function lifecycle(d: PostEvent, label: string, change: Partial<StoredPost>) {
  Object.assign(postFor(d), change);
  log(label, `${at(d.author)} ${d.tweetId}`);
}

const handlers: { [P in Pair]: Handler<Payloads[P]> } = {
  "control/auth_pong": () => log("auth", "ok, waiting for events"),
  "tweet/content": (d, receivedAt) => {
    const platform = d.author?.platform ?? "twitter";
    const post = postFor(d, platform);
    post.content = merge(d, post.content); // an update that arrived first still wins
    let latency = "";
    const publishedAt = publishedAtMs(platform, d);
    if (!post.hasContent && publishedAt !== null) {
      latencies.push(receivedAt - publishedAt);
      if (latencies.length > 10_000) latencies.shift();
      latency = `${receivedAt - publishedAt} ms  `;
    }
    post.hasContent = true;
    const where = platform === "twitter" ? "" : ` (${platform})`;
    log("post", `${at(d.author)}${where} ${d.tweetId}  ${latency}${clip(post.content.text)}`);
    if (d.ref) log("", `chain: ${referenceChain(d.ref)}`);
  },
  "tweet/update": (d) => {
    const post = postFor(d);
    post.content = merge(post.content, d); // a ref in the update replaces the stored one
    const who = `${at(post.content.author)} ${d.tweetId}`;
    const changed = Object.keys(d).filter((k) => !["platform", "receivedAt", "tweetId"].includes(k));
    if (d.ref) log("update", `${who} chain: ${referenceChain(d.ref)}`);
    else if (d.textUpdateType === "completion") log("update", `${who} full text ${clip(d.text)}`);
    else log("update", `${who} changed ${changed.join(", ")}`);
  },
  "tweet/meta": (d) => {
    const post = postFor(d);
    post.meta = d; // each meta is a complete snapshot: replace it, never merge
    log("meta", `${at(post.content?.author)} ${d.tweetId}  ${describeMeta(d)}`);
  },
  "tweet/delete": (d) => lifecycle(d, "delete", { deleted: true }),
  "tweet/pin": (d) => lifecycle(d, "pin", { pinned: true }),
  "tweet/unpin": (d) => lifecycle(d, "unpin", { pinned: false }),
  "account/follow": (d) => log("follow", `${at(d.actor)} followed ${at(d.target)}`),
  "account/unfollow": (d) => log("unfollow", `${at(d.actor)} unfollowed ${at(d.target)}`),
  "account/profile_update": (d) =>
    log("profile", `${at(d.actor)} changed ${Object.keys(d.changes ?? {}).join(", ")}`),
  "account/affiliate_update": (d) =>
    log("affiliate", `${at(d.organization)} ${d.action} ${at(d.member)}`),
  "account/callout": (d) =>
    log("callout", `${at(d.caller)} called ${d.token?.symbol} ${d.token?.address}`),
  "account/callout_update": (d) => log("callout", `update to ${d.updateOf}: ${d.token?.symbol}`),
  "account/thesis": (d) =>
    log("thesis", `${at(d.author)} on ${d.coin?.symbol}: ${clip(d.thesis?.text, 60)}`),
  "account/token_verified": (d) => log("verified", `${d.coin?.symbol ?? d.coin?.address}`),
  "news/article": (d) => log("news", `${d.source?.name}: ${clip(d.title)}`),
  "charity/added": (d) => log("charity", d.name),
  "control/twitter_handles_result": (d) =>
    log("handles", (d.results ?? []).map((r) => `${r.input} ${r.state}`).join(", ") || String(d.error)),
};
const isPair = (pair: string): pair is Pair => Object.hasOwn(handlers, pair);

function onMessage(data: unknown, receivedAt: number) {
  let event: Partial<Envelope> | null;
  try {
    event = JSON.parse(String(data));
  } catch {
    log("skip", "frame is not JSON");
    return;
  }
  if (event?.v !== 1 || typeof event.t !== "string" || typeof event.op !== "string") return;
  const pair = `${event.t}/${event.op}`;
  counts.set(pair, (counts.get(pair) ?? 0) + 1);
  if (!isPair(pair)) return; // unknown pairs are counted, then ignored
  // The payload is JSON from the server; the types above describe what each pair carries.
  const handle = handlers[pair] as Handler<unknown>;
  try {
    handle(event.d ?? {}, receivedAt);
  } catch (error) {
    log("error", `${pair}: ${error}`); // one malformed event must not stop the program
  }
}

function printSummary() {
  const sorted = [...latencies].sort((a, b) => a - b);
  const pct = (p: number) => sorted[Math.max(0, Math.ceil(p * sorted.length) - 1)];
  const timing = sorted.length
    ? `p50 ${pct(0.5)} ms  p95 ${pct(0.95)} ms  (n=${sorted.length})`
    : "no posts yet";
  const seen = [...counts].map(([pair, n]) => `${pair} ${n}`).join(", ") || "none";
  log("summary", `latency ${timing} | events ${seen}`);
}

// POST /api/add-account answers with one row per handle.
type AddAccountResponse = {
  results?: Array<{ input: string; state: string; message?: string }>;
  error?: string;
};

async function addAccounts() {
  const response = await fetch(`${API_URL}/api/add-account`, {
    body: JSON.stringify({ accounts: ACCOUNTS }),
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    method: "POST",
    signal: AbortSignal.timeout(15_000),
  });
  const body = (await response.json().catch(() => ({}))) as AddAccountResponse;
  // One row per handle, even on 207 (some failed) or 400 (all failed). Each row says why.
  const rows = Array.isArray(body.results) ? body.results : [];
  for (const row of rows) {
    log("accounts", `${row.input} ${row.state}${row.message ? `: ${row.message}` : ""}`);
  }
  if (!rows.length && !response.ok) {
    log("accounts", `add failed (${response.status}): ${body.error ?? "unknown error"}`);
  }
}

// GET /api/me reports which pool the key connects to and how many connections each has open.
type ConnectionPool = { count: number; limit: number };
type MeResponse = {
  credentialScope?: "standard" | "ultra_speed";
  websocket?: ConnectionPool; // standard keys
  ultraSpeed?: { websocket?: ConnectionPool } | null; // Ultra Speed keys
};

async function explainRefusedConnection() {
  // Node's built-in WebSocket hides the HTTP status of a refused upgrade, so ask the REST API.
  const headers = { Authorization: `Bearer ${API_KEY}` };
  const response = await fetch(`${API_URL}/api/me`, {
    headers,
    signal: AbortSignal.timeout(10_000),
  }).catch(() => null);
  if (response?.status === 401) {
    console.error(`API key rejected (401). Copy a current key from ${DASHBOARD}`);
    process.exit(1);
  }
  if (response?.status === 403) {
    console.error("Streaming is not active for this key (403). Start a trial or renew your plan.");
    process.exit(1);
  }
  const me = (response?.ok ? await response.json().catch(() => null) : null) as MeResponse | null;
  const pool = me?.credentialScope === "ultra_speed" ? me.ultraSpeed?.websocket : me?.websocket;
  if (pool && pool.count >= pool.limit) {
    log("limit", `all ${pool.limit} connections are in use; close another client`);
  }
}

function connect() {
  const socket = new WebSocket(WS_URL, ["tweetstream.v1", `tweetstream.auth.token.${API_KEY}`]);
  let openedAt: number | null = null; // set while this socket is open
  socket.addEventListener("open", () => {
    openedAt = Date.now();
    log("connected", WS_URL);
    socket.send(JSON.stringify({ op: "auth_ping" })); // answered by control/auth_pong
    setTimeout(() => {
      if (openedAt !== null && !latencies.length && !hinted) {
        hinted = true;
        log("hint", "posts appear when a monitored account posts. Add accounts in the");
        log("hint", "dashboard, or run again with TWEETSTREAM_ACCOUNTS=handle1,handle2");
      }
    }, 120_000);
  });
  // Take the receipt time before any parsing so latency excludes your own processing.
  socket.addEventListener("message", (event) => onMessage(event.data, Date.now()));
  let settled = false; // some runtimes fire both error and close; reconnect only once
  const reconnect = async (reason: string) => {
    if (settled) return;
    settled = true;
    if (openedAt === null) await explainRefusedConnection();
    else if (Date.now() - openedAt >= 30_000) attempt = 0; // healthy session: reset backoff
    openedAt = null;
    const delay = Math.min(30_000, 1_000 * 2 ** attempt) + Math.floor(Math.random() * 250);
    attempt = Math.min(attempt + 1, 5);
    log("closed", `${reason}; reconnecting in ${(delay / 1000).toFixed(1)} s`);
    setTimeout(connect, delay);
  };
  // Node's built-in WebSocket reports a refused or failed handshake with error and no close.
  socket.addEventListener("error", () => openedAt === null && reconnect("connection failed"));
  socket.addEventListener("close", (event) => reconnect(`code ${event.code}`));
}

process.on("SIGINT", () => {
  printSummary();
  process.exit(0);
});
setInterval(printSummary, 60_000);
if (ACCOUNTS.length) {
  addAccounts()
    .catch((error: Error) => log("accounts", `add failed: ${error.message}`))
    .finally(connect);
} else {
  connect();
}