wayBetter for X

apps/desktop/src/observer/network.ts

@@ -0,0 +1,215 @@
+import { parseRelationshipObservations, type RelationshipSnapshot } from './relationships.ts';
+
+export const OBSERVATION_EVENT = 'x-workspace:relationship-observation';
+export const MAX_RESPONSE_BYTES = 64 * 1_024;
+export const MAX_OBSERVATION_READERS = 4;
+export const MAX_OBSERVATION_READ_MS = 1_500;
+
+export type PageObservation = RelationshipSnapshot & {
+  accountId: null;
+  requestEpoch: number;
+  requestStartedAt: string;
+  observedAt: string;
+  source: 'x-page-response:unverified-schema';
+};
+export type ResponseObserverHandle = {
+  installed: boolean;
+  reason?: string;
+  advanceEpoch(): number;
+  dispose(): void;
+};
+type PageWindow = Window & typeof globalThis;
+type Options = { onObservation?: (event: PageObservation) => void };
+const installations = new WeakMap<Window, ResponseObserverHandle>();
+
+export function isSupportedPage(target: Window): boolean {
+  try {
+    return target.top === target && target.location.origin === 'https://x.com';
+  } catch {
+    return false;
+  }
+}
+
+// Chat (XChat, end-to-end encrypted) and DM pages: nothing is read there (no DM capture).
+const CHAT_PATH = /^\/(?:messages|i\/chat)(?:\/|$)/i;
+
+function eligibleUrl(input: unknown, base: string): boolean {
+  try {
+    const raw = typeof input === 'string' ? input : input instanceof URL ? input.href
+      : input && typeof input === 'object' && 'url' in input ? String(input.url) : '';
+    // Cheap reject first: images, video, WASM, scripts and telemetry never parse a URL.
+    if (!raw.includes('/i/api/')) return false;
+    const url = new URL(raw, base);
+    return url.origin === 'https://x.com' && url.pathname.startsWith('/i/api/');
+  } catch {
+    return false;
+  }
+}
+function jsonType(contentType: string | null): boolean {
+  return Boolean(contentType && /^application\/(?:[\w.-]+\+)?json(?:\s*;|$)/i.test(contentType));
+}
+function declaredTooLarge(length: string | null): boolean {
+  return length !== null && (!/^\d+$/.test(length) || Number(length) > MAX_RESPONSE_BYTES);
+}
+
+/** Reads at most 64 KiB from a clone; never drains the original response. */
+async function readSmallJson(response: Response, signal: AbortSignal): Promise<unknown> {
+  if (signal.aborted) return;
+  if (!jsonType(response.headers.get('content-type')) || declaredTooLarge(response.headers.get('content-length'))) return;
+  if (!response.body) return;
+  const clone = response.clone();
+  const reader = clone.body!.getReader();
+  const cancel = () => { void reader.cancel().catch(() => {}); };
+  signal.addEventListener('abort', cancel, { once: true });
+  const decoder = new TextDecoder();
+  let text = '';
+  let bytes = 0;
+  try {
+    while (true) {
+      const chunk = await reader.read();
+      if (signal.aborted) return;
+      if (chunk.done) break;
+      bytes += chunk.value.byteLength;
+      if (bytes > MAX_RESPONSE_BYTES) return;
+      text += decoder.decode(chunk.value, { stream: true });
+    }
+    text += decoder.decode();
+    return JSON.parse(text);
+  } catch {
+    return;
+  } finally {
+    // Awaiting cancellation of one tee branch can wait for the other branch forever.
+    signal.removeEventListener('abort', cancel);
+    cancel();
+    reader.releaseLock();
+  }
+}
+
+/** Opt-in helper. It observes existing calls only and creates no network requests. */
+export function installResponseObserver(target: Window = window, options: Options = {}): ResponseObserverHandle {
+  const existing = installations.get(target);
+  if (existing) return existing;
+  if (!isSupportedPage(target)) return { installed: false, reason: 'Only the top-level https://x.com page is supported.', advanceEpoch: () => 0, dispose() {} };
+  const page = target as PageWindow;
+  let epoch = 0;
+  let active = true;
+  const activeReaders = new Set<AbortController>();
+  let pageUrl = page.location.href;
+  const stamp = () => {
+    if (page.location.href !== pageUrl) {
+      pageUrl = page.location.href;
+      epoch++;
+      for (const reader of activeReaders) reader.abort();
+    }
+    return { requestEpoch: epoch, requestStartedAt: new Date().toISOString() };
+  };
+  const emit = (body: unknown, started: ReturnType<typeof stamp>) => {
+    if (!active || started.requestEpoch !== epoch) return;
+    for (const relation of parseRelationshipObservations(body)) {
+      const event: PageObservation = {
+        ...relation, ...started, accountId: null,
+        observedAt: new Date().toISOString(), source: 'x-page-response:unverified-schema',
+      };
+      // Subscriber failure must never affect X's fetch/XHR callbacks.
+      try { options.onObservation?.(event); } catch { /* Consumer error is isolated. */ }
+      try { target.dispatchEvent(new page.CustomEvent(OBSERVATION_EVENT, { detail: event })); } catch { /* No raw data fallback. */ }
+    }
+  };
+
+  const originalFetch = page.fetch;
+  const onChat = () => CHAT_PATH.test(page.location.pathname ?? '');
+  const fetchWrapper: typeof fetch = function (this: unknown, ...args: Parameters<typeof fetch>) {
+    // The same Promise and Response reach X. Rejections and synchronous errors are unchanged.
+    if (!eligibleUrl(args[0], page.location.href) || onChat()) return Reflect.apply(originalFetch, this, args) as ReturnType<typeof fetch>;
+    const started = stamp();
+    const result = Reflect.apply(originalFetch, this, args) as ReturnType<typeof fetch>;
+    void result.then(response => {
+      if (active && started.requestEpoch === epoch && activeReaders.size < MAX_OBSERVATION_READERS && response.ok && (!response.url || eligibleUrl(response.url, page.location.href))) {
+        const controller = new AbortController();
+        activeReaders.add(controller);
+        const timeout = setTimeout(() => controller.abort(), MAX_OBSERVATION_READ_MS);
+        void readSmallJson(response, controller.signal).then(body => emit(body, started)).catch(() => {}).finally(() => {
+          clearTimeout(timeout);
+          activeReaders.delete(controller);
+        });
+      }
+    }, () => {}).catch(() => {});
+    return result;
+  };
+  let fetchPatched = false;
+  try { page.fetch = fetchWrapper; fetchPatched = page.fetch === fetchWrapper; } catch { /* Readonly runtime. */ }
+
+  const prototype = page.XMLHttpRequest?.prototype;
+  const originalOpen = prototype?.open;
+  const originalSend = prototype?.send;
+  const states = new WeakMap<XMLHttpRequest, { url: string; cleanup?: () => void }>();
+  const pendingListeners = new Set<() => void>();
+  const openWrapper = function (this: XMLHttpRequest, ...args: unknown[]) {
+    const result = Reflect.apply(originalOpen!, this, args);
+    states.get(this)?.cleanup?.();
+    states.set(this, { url: String(args[1]) });
+    return result;
+  } as XMLHttpRequest['open'];
+  const sendWrapper: XMLHttpRequest['send'] = function (this: XMLHttpRequest, ...args) {
+    const state = states.get(this);
+    if (!state || !eligibleUrl(state.url, page.location.href) || onChat()) return Reflect.apply(originalSend!, this, args);
+    const started = stamp();
+    const cleanup = () => {
+      this.removeEventListener('load', onLoad);
+      this.removeEventListener('loadend', cleanup);
+      pendingListeners.delete(cleanup);
+      if (state.cleanup === cleanup) state.cleanup = undefined;
+    };
+    const onLoad = () => {
+      try {
+        if (!active || this.status < 200 || this.status >= 300) return;
+        if (this.responseURL && !eligibleUrl(this.responseURL, page.location.href)) return;
+        if (!jsonType(this.getResponseHeader('content-type'))) return;
+        const length = this.getResponseHeader('content-length');
+        if (declaredTooLarge(length)) return;
+        if (this.responseType === '' || this.responseType === 'text') {
+          const text = this.responseText;
+          if (text.length > MAX_RESPONSE_BYTES || new TextEncoder().encode(text).byteLength > MAX_RESPONSE_BYTES) return;
+          emit(JSON.parse(text), started);
+        } else if (this.responseType === 'json' && length !== null) {
+          // A parsed response has no cheap byte length. Without a declared limit, skip it.
+          emit(this.response, started);
+        }
+      } catch { /* Invalid JSON or an unavailable response is ignored. */ }
+    };
+    state.cleanup = cleanup;
+    pendingListeners.add(cleanup);
+    this.addEventListener('load', onLoad);
+    this.addEventListener('loadend', cleanup, { once: true });
+    try { return Reflect.apply(originalSend!, this, args); }
+    catch (error) { cleanup(); throw error; }
+  };
+  let xhrPatched = false;
+  if (prototype) {
+    try {
+      prototype.open = openWrapper;
+      prototype.send = sendWrapper;
+      xhrPatched = prototype.open === openWrapper && prototype.send === sendWrapper;
+    } catch { /* Fetch still works if XHR cannot be patched. */ }
+  }
+  const handle: ResponseObserverHandle = {
+    installed: fetchPatched || xhrPatched,
+    ...(!fetchPatched && !xhrPatched ? { reason: 'Page request hooks are unavailable.' } : {}),
+    advanceEpoch() {
+      for (const reader of activeReaders) reader.abort();
+      return ++epoch;
+    },
+    dispose() {
+      if (!active) return;
+      active = false;
+      for (const reader of activeReaders) reader.abort();
+      if (page.fetch === fetchWrapper) page.fetch = originalFetch;
+      if (prototype?.open === openWrapper) prototype.open = originalOpen!;
+      if (prototype?.send === sendWrapper) prototype.send = originalSend!;
+      for (const cleanup of pendingListeners) cleanup();
+      installations.delete(target);
+    },
+  };
+  installations.set(target, handle);
+  return handle;
+}