waybetter-for-x/apps/desktop/tests/liveCounts.test.ts

import assert from 'node:assert/strict';
import { test } from 'node:test';
import { createCountHold, filterLiveLine, filterLiveStream, isLivePipelineUrl, LIVE_QUIET_MS } from '../src/fixes/mutationGate/liveCounts.ts';
import { installMutationGate } from '../src/fixes/mutationGate/index.ts';

// Message shape as captured from X's live pipeline on 2026-09-28 (engagement counts only).
const POST = '2103816196386852995';
const push = (id: string, engagement: Record<string, unknown>) =>
  `data: ${JSON.stringify({ topic: `/tweet_engagement/${id}`, payload: { tweet_engagement: engagement } })}`;
const parse = (line: string) => JSON.parse(line.slice(5)) as { topic: string; payload: { tweet_engagement: Record<string, unknown> } };
const tick = () => new Promise(resolve => setImmediate(resolve));

test('live pipeline URL: only api.x.com / api.twitter.com event streams', () => {
  assert.equal(isLivePipelineUrl(`https://api.x.com/live_pipeline/events?topic=%2Ftweet_engagement%2F${POST}`), true);
  assert.equal(isLivePipelineUrl('https://api.twitter.com/live_pipeline/events?topic=x'), true);
  for (const url of ['https://x.com/live_pipeline/events', 'http://api.x.com/live_pipeline/events', 'https://api.x.com/1.1/live_pipeline/update_subscriptions',
    'https://evil.example/live_pipeline/events', '/live_pipeline/events', 'not a url /live_pipeline/events']) assert.equal(isLivePipelineUrl(url), false, url);
});

test('count hold: held while a request is in flight and for LIVE_QUIET_MS after the last one ends', () => {
  let now = 1000;
  const hold = createCountHold(() => now);
  assert.equal(hold.held('like', POST), false);
  hold.begin('like', POST);
  hold.begin('like', POST);
  now += 60_000;
  assert.equal(hold.held('like', POST), true, 'in flight, however long');
  assert.equal(hold.held('retweet', POST), false, 'per count kind');
  assert.equal(hold.held('like', '1'), false, 'per post');
  hold.end('like', POST);
  now += LIVE_QUIET_MS * 2;
  assert.equal(hold.held('like', POST), true, 'one request still in flight');
  hold.end('like', POST);
  now += LIVE_QUIET_MS - 1;
  assert.equal(hold.held('like', POST), true, 'quiet window after the last end');
  now += 1;
  assert.equal(hold.held('like', POST), false);
  assert.equal(hold.size(), 0, 'expired entries are dropped');
});

test('line filter: removes only the held count field of the held post', () => {
  const held = (kind: string, id: string) => kind === 'like' && id === POST;
  let drops = 0;
  const line = push(POST, { like_count: '3', retweet_count: '1', view_count_info: { count: '351', state: 'EnabledWithCount' } });
  const out = parse(filterLiveLine(line, held, () => drops++));
  assert.deepEqual(out, { topic: `/tweet_engagement/${POST}`, payload: { tweet_engagement: { retweet_count: '1', view_count_info: { count: '351', state: 'EnabledWithCount' } } } });
  assert.equal(drops, 1);
  // Everything else is returned byte for byte.
  for (const other of [push('42', { like_count: '9' }), push(POST, { view_count_info: { count: '1' } }), 'data: {"topic":"/system/config","payload":{"config":{}}}',
    'data: {not json', '', ':heartbeat', 'event: message', `${push(POST, { like_count: '3' })}x`, 'data: {"topic":"/tweet_engagement/abc","payload":{"tweet_engagement":{"like_count":"1"}}}']) {
    assert.equal(filterLiveLine(other, held), other);
  }
  // A CRLF stream keeps its line ending.
  assert.ok(filterLiveLine(`${push(POST, { like_count: '3' })}\r`, held).endsWith('}\r'));
  assert.deepEqual(parse(filterLiveLine(`${push(POST, { like_count: '3' })}\r`, held).trimEnd()).payload.tweet_engagement, {});
});

test('stream filter: messages split across chunks are filtered whole; cancel reaches the source', async () => {
  const enc = new TextEncoder();
  const text = `${push(POST, { like_count: '3', view_count_info: { count: '5' } })}\n\n${push('42', { like_count: '7' })}\n\n`;
  const cut = [0, 17, 18, 60, text.length - 3, text.length];
  const source = new ReadableStream<Uint8Array>({
    start(controller) { for (let i = 1; i < cut.length; i++) controller.enqueue(enc.encode(text.slice(cut[i - 1], cut[i]))); controller.close(); },
  });
  const out = await new Response(filterLiveStream(source, (kind, id) => kind === 'like' && id === POST)).text();
  const lines = out.split('\n');
  assert.deepEqual(parse(lines[0]!).payload.tweet_engagement, { view_count_info: { count: '5' } });
  assert.equal(lines[2], push('42', { like_count: '7' }));
  assert.equal(out.endsWith('\n\n'), true);
  let cancelled: unknown = null;
  const endless = new ReadableStream<Uint8Array>({ pull() { return new Promise(() => {}); }, cancel(reason) { cancelled = reason; } });
  await filterLiveStream(endless, () => false).cancel('bye');
  assert.equal(cancelled, 'bye');
});

// ---------- in the gate: X's XHR like/unlike + its streamed live pipeline ----------

function env() {
  const replies: (() => void)[] = [];
  class FakeXHR extends EventTarget {
    readyState = 0; status = 0; responseText = ''; responseType: XMLHttpRequestResponseType = ''; url = '';
    get response() { return this.responseText; }
    getResponseHeader() { return 'application/json'; }
    getAllResponseHeaders() { return 'content-type: application/json\r\n'; }
    open(_method: string, url: string) { this.url = url; this.readyState = 1; }
    send() {
      const want = this.url.endsWith('/FavoriteTweet');
      replies.push(() => {
        this.readyState = 4; this.status = 200;
        this.responseText = want ? '{"data":{"favorite_tweet":"Done"}}' : '{"data":{"unfavorite_tweet":"Done"}}';
        for (const type of ['readystatechange', 'load', 'loadend']) this.dispatchEvent(new Event(type));
      });
    }
    abort() {}
  }
  let pipe!: ReadableStreamDefaultController<Uint8Array>;
  const stream = new ReadableStream<Uint8Array>({ start(controller) { pipe = controller; } });
  const fetch = (async (input: unknown) => String(input).includes('/live_pipeline/events')
    ? new Response(stream, { status: 200, headers: { 'content-type': 'text/event-stream' } })
    : new Response('{}')) as typeof globalThis.fetch;
  const win = { fetch, location: { href: `https://x.com/playfishies/status/${POST}` }, XMLHttpRequest: FakeXHR as unknown as typeof XMLHttpRequest };
  const send = (op: 'FavoriteTweet' | 'UnfavoriteTweet', id = POST) => {
    const xhr = new win.XMLHttpRequest();
    xhr.open('POST', `/i/api/graphql/q/${op}`);
    xhr.send(JSON.stringify({ variables: { tweet_id: id } }));
  };
  const enc = new TextEncoder();
  const emit = (line: string) => pipe.enqueue(enc.encode(`${line}\n\n`));
  return { win, replies, send, emit };
}

async function reader(win: { fetch: typeof fetch }) {
  // X's own reader: fetch the event stream, split on "\n", keep "data:" lines.
  const response = await win.fetch(`https://api.x.com/live_pipeline/events?topic=%2Ftweet_engagement%2F${POST}`, { headers: { Accept: 'text/event-stream' } });
  assert.equal(response.ok, true);
  const got: string[] = [];
  const body = response.body!.getReader();
  const decoder = new TextDecoder();
  let rest = '';
  void (async () => {
    for (;;) {
      const { done, value } = await body.read();
      if (done) return;
      const lines = (rest += decoder.decode(value, { stream: true })).split('\n');
      rest = lines.pop()!;
      for (const line of lines) if (line.startsWith('data:')) got.push(line);
    }
  })();
  return got;
}

test('gate: a push sent while like→unlike is in flight never reaches X with the stale like_count (live 2026-09-28 case)', async () => {
  const e = env();
  const gate = installMutationGate(e.win);
  const got = await reader(e.win);
  e.emit(push(POST, { like_count: '2' }));
  await tick(); await tick();
  assert.deepEqual(parse(got[0]!).payload.tweet_engagement, { like_count: '2' }, 'idle: pushes pass through');
  // Not liked (2). Like, then unlike 40 ms later: the like is on the wire, the unlike held behind it.
  e.send('FavoriteTweet');
  e.send('UnfavoriteTweet');
  // X's server pushes the count after the FavoriteTweet ("3"), before the unlike reached it.
  e.emit(push(POST, { like_count: '3', view_count_info: { count: '351', state: 'EnabledWithCount' } }));
  e.emit(push('42', { like_count: '8' }));
  await tick(); await tick();
  assert.deepEqual(parse(got[1]!).payload.tweet_engagement, { view_count_info: { count: '351', state: 'EnabledWithCount' } });
  assert.equal(got[2], push('42', { like_count: '8' }), 'another post is untouched');
  while (e.replies.length) { e.replies.shift()!(); await tick(); }
  // Right after the last request settled, a late push is still stale.
  e.emit(push(POST, { like_count: '3' }));
  await tick(); await tick();
  assert.deepEqual(parse(got[3]!).payload.tweet_engagement, {});
  assert.equal(gate.live.liveDropped, 2);
  // Kill switch: the stream keeps flowing, unfiltered.
  gate.dispose();
  e.emit(push(POST, { like_count: '3' }));
  await tick(); await tick();
  assert.equal(got[4], push(POST, { like_count: '3' }));
});