all repos — sz-chrome-remote-cdp @ master

pi-browser-harness, patched to drive a Chrome on another host over CDP, with castle_cdp_search web search

castle-search/src/cdp.ts (view raw)

 1
 2
 3
 4
 5
 6
 7
 8
 9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
/**
 * A minimal CDP client — just enough to open a page, navigate it, read the DOM,
 * and close it again.
 *
 * Deliberately dependency-free: Node 22 ships a global `WebSocket`, so this
 * needs no `ws`. That matters because an extension installed by `pi install`
 * gets its own `node_modules`, and fewer moving parts there is fewer things to
 * go wrong on a machine that is not the one it was written on.
 *
 * The browser is *not* ours. It is a shared Chrome someone may be looking at,
 * recycled hourly (`me.soarez.chrome-cdp-recycle`). Two consequences run
 * through this file:
 *
 *   1. The websocket URL is re-resolved from `/json/version` on every connect,
 *      never cached. Chrome re-mints its browser UUID on each launch, so a
 *      pinned ws:// URL breaks the first time the browser is recycled.
 *   2. Every page this opens is created in the background and closed again,
 *      including on error paths, so a failed search does not leave tabs behind
 *      in someone else's window.
 */

import { BrowserUnavailableError, SearchAbortedError, messageOf } from "./errors.ts";
import type { CdpTarget } from "./endpoint.ts";

/** How long a single CDP command may take before the socket is presumed dead. */
const COMMAND_TIMEOUT_MS = 15_000;
/** How long to wait for /json/version and for the websocket handshake. */
const CONNECT_TIMEOUT_MS = 5_000;

type CdpMessage = {
  id?: number;
  method?: string;
  params?: Record<string, unknown>;
  sessionId?: string;
  result?: Record<string, unknown>;
  error?: { code?: number; message?: string };
};

type Pending = {
  resolve: (value: Record<string, unknown>) => void;
  reject: (reason: Error) => void;
  timer: ReturnType<typeof setTimeout>;
};

/**
 * Reject if `signal` fires before `promise` settles, cleaning up the listener
 * either way. Node's own abort-aware APIs are not available here — the
 * websocket predates them — so aborts are raced in rather than plumbed through.
 */
export const raceAbort = <T>(promise: Promise<T>, signal: AbortSignal | undefined): Promise<T> => {
  if (!signal) return promise;
  if (signal.aborted) return Promise.reject(abortError(signal));
  return new Promise<T>((resolve, reject) => {
    const onAbort = (): void => reject(abortError(signal));
    signal.addEventListener("abort", onAbort, { once: true });
    promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort));
  });
};

const abortError = (signal: AbortSignal): Error => {
  const reason: unknown = signal.reason;
  if (reason instanceof Error && reason.name === "TimeoutError") {
    return new SearchAbortedError("search timed out");
  }
  return new SearchAbortedError("search was cancelled");
};

/** Ask the browser for its current websocket endpoint. Never cached — see above. */
const queryWebSocketUrl = async (target: CdpTarget, signal?: AbortSignal): Promise<string> => {
  const url = `http://${target.host}:${target.port}/json/version`;
  let payload: unknown;
  try {
    const res = await fetch(url, {
      signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(CONNECT_TIMEOUT_MS)]) : AbortSignal.timeout(CONNECT_TIMEOUT_MS),
    });
    if (!res.ok) {
      throw new BrowserUnavailableError(`${url} answered HTTP ${res.status}`);
    }
    payload = await res.json();
  } catch (e) {
    if (e instanceof BrowserUnavailableError) throw e;
    if (signal?.aborted) throw abortError(signal);
    throw new BrowserUnavailableError(
      `No CDP endpoint answered at ${url} (${messageOf(e)}). Is the browser running and reachable on the overlay?`,
    );
  }
  const ws = (payload as Record<string, unknown> | null)?.["webSocketDebuggerUrl"];
  if (typeof ws !== "string" || ws === "") {
    throw new BrowserUnavailableError(`${url} answered without a webSocketDebuggerUrl`);
  }
  return ws;
};

/** A live browser-level CDP connection. */
export class BrowserConnection {
  #ws: WebSocket;
  #nextId = 1;
  #pending = new Map<number, Pending>();
  #listeners = new Set<(msg: CdpMessage) => void>();
  #closedReason: string | null = null;

  readonly target: CdpTarget;

  private constructor(ws: WebSocket, target: CdpTarget) {
    this.#ws = ws;
    this.target = target;

    ws.addEventListener("message", (ev: MessageEvent) => {
      let msg: CdpMessage;
      try {
        msg = JSON.parse(String(ev.data)) as CdpMessage;
      } catch {
        return;
      }
      if (msg.id !== undefined) {
        const pending = this.#pending.get(msg.id);
        if (!pending) return;
        this.#pending.delete(msg.id);
        clearTimeout(pending.timer);
        if (msg.error) {
          pending.reject(new BrowserUnavailableError(`CDP error: ${msg.error.message ?? "unknown"}`));
        } else {
          pending.resolve(msg.result ?? {});
        }
        return;
      }
      for (const listener of [...this.#listeners]) listener(msg);
    });

    const fail = (reason: string): void => {
      this.#closedReason ??= reason;
      for (const [, pending] of this.#pending) {
        clearTimeout(pending.timer);
        pending.reject(new BrowserUnavailableError(reason));
      }
      this.#pending.clear();
    };
    ws.addEventListener("close", () => fail("CDP connection closed (the browser may have been recycled)"));
    ws.addEventListener("error", () => fail("CDP connection failed"));
  }

  /** Resolve the endpoint afresh and dial it. */
  static async connect(target: CdpTarget, signal?: AbortSignal): Promise<BrowserConnection> {
    const wsUrl = await queryWebSocketUrl(target, signal);
    const ws = new WebSocket(wsUrl);
    try {
      await raceAbort(
        new Promise<void>((resolve, reject) => {
          const timer = setTimeout(
            () => reject(new BrowserUnavailableError(`timed out opening ${wsUrl}`)),
            CONNECT_TIMEOUT_MS,
          );
          ws.addEventListener(
            "open",
            () => {
              clearTimeout(timer);
              resolve();
            },
            { once: true },
          );
          ws.addEventListener(
            "error",
            () => {
              clearTimeout(timer);
              reject(new BrowserUnavailableError(`could not open ${wsUrl}`));
            },
            { once: true },
          );
        }),
        signal,
      );
    } catch (e) {
      try {
        ws.close();
      } catch {
        // Already failed; nothing useful to do with a second failure.
      }
      throw e;
    }
    return new BrowserConnection(ws, target);
  }

  /** False once the socket has closed or failed — the reconnect trigger. */
  get isOpen(): boolean {
    return this.#closedReason === null && this.#ws.readyState === 1 /* OPEN */;
  }

  send(method: string, params: Record<string, unknown> = {}, sessionId?: string): Promise<Record<string, unknown>> {
    if (!this.isOpen) {
      return Promise.reject(new BrowserUnavailableError(this.#closedReason ?? "CDP connection is not open"));
    }
    const id = this.#nextId++;
    const payload: CdpMessage = sessionId ? { id, method, params, sessionId } : { id, method, params };
    return new Promise<Record<string, unknown>>((resolve, reject) => {
      const timer = setTimeout(() => {
        this.#pending.delete(id);
        reject(new BrowserUnavailableError(`CDP command ${method} timed out after ${COMMAND_TIMEOUT_MS}ms`));
      }, COMMAND_TIMEOUT_MS);
      this.#pending.set(id, { resolve, reject, timer });
      try {
        this.#ws.send(JSON.stringify(payload));
      } catch (e) {
        this.#pending.delete(id);
        clearTimeout(timer);
        reject(new BrowserUnavailableError(`could not send ${method}: ${messageOf(e)}`));
      }
    });
  }

  /** Subscribe to unsolicited protocol events. Returns an unsubscribe function. */
  onEvent(listener: (msg: CdpMessage) => void): () => void {
    this.#listeners.add(listener);
    return () => this.#listeners.delete(listener);
  }

  close(): void {
    this.#closedReason ??= "CDP connection closed by this extension";
    this.#listeners.clear();
    try {
      this.#ws.close();
    } catch {
      // Closing an already-dead socket is not a failure worth reporting.
    }
  }
}

/**
 * One throwaway page.
 *
 * Created in the background so it does not steal focus from whoever is using
 * the browser, and closed by {@link close} on every path.
 */
export class PageSession {
  readonly #conn: BrowserConnection;
  readonly #targetId: string;
  readonly #sessionId: string;
  #closed = false;

  private constructor(conn: BrowserConnection, targetId: string, sessionId: string) {
    this.#conn = conn;
    this.#targetId = targetId;
    this.#sessionId = sessionId;
  }

  static async open(conn: BrowserConnection, signal?: AbortSignal): Promise<PageSession> {
    if (signal?.aborted) throw abortError(signal);

    // Deliberately NOT raced against the abort signal. raceAbort abandons the
    // promise, it cannot cancel the command — and this command's side effect is
    // a tab. Abandoning it would leave a tab open in a browser someone else is
    // using, with nothing left holding its targetId to close it. It is a fast
    // command already bounded by COMMAND_TIMEOUT_MS, so waiting it out costs
    // little and guarantees we own whatever it created.
    const created = await conn.send("Target.createTarget", { url: "about:blank", background: true });
    const targetId = created["targetId"];
    if (typeof targetId !== "string") {
      throw new BrowserUnavailableError("Target.createTarget did not return a targetId");
    }

    // From here on the page exists, so every failure — including a late abort —
    // must still close it.
    try {
      if (signal?.aborted) throw abortError(signal);
      const attached = await raceAbort(
        conn.send("Target.attachToTarget", { targetId, flatten: true }),
        signal,
      );
      const sessionId = attached["sessionId"];
      if (typeof sessionId !== "string") {
        throw new BrowserUnavailableError("Target.attachToTarget did not return a sessionId");
      }
      const page = new PageSession(conn, targetId, sessionId);
      await raceAbort(conn.send("Page.enable", {}, sessionId), signal);
      return page;
    } catch (e) {
      await conn.send("Target.closeTarget", { targetId }).catch(() => {});
      throw e;
    }
  }

  /** Navigate and wait for the load event, or for `signal` to fire. */
  async navigate(url: string, signal?: AbortSignal): Promise<void> {
    // Subscribed before navigating, so a page that loads faster than the
    // Page.navigate reply cannot slip its load event past us.
    let off: () => void = () => {};
    const loaded = new Promise<void>((resolve) => {
      off = this.#conn.onEvent((msg) => {
        if (msg.sessionId === this.#sessionId && msg.method === "Page.loadEventFired") resolve();
      });
    });

    try {
      const nav = await raceAbort(this.#conn.send("Page.navigate", { url }, this.#sessionId), signal);
      const errorText = nav["errorText"];
      if (typeof errorText === "string" && errorText !== "") {
        throw new BrowserUnavailableError(`navigation to ${url} failed: ${errorText}`);
      }

      try {
        await raceAbort(loaded, signal);
      } catch (e) {
        // Stop the in-flight load so the page is not still fetching while we
        // tear it down; the tab is closed either way, this just makes it quicker.
        await this.#conn.send("Page.stopLoading", {}, this.#sessionId).catch(() => {});
        throw e;
      }
    } finally {
      // Unsubscribing here rather than inside the listener: on the failure paths
      // the listener never fires, and a subscription left on a connection that
      // outlives many searches accumulates.
      off();
    }
  }

  /** Evaluate an expression in the page and return its value by value. */
  async evaluate<T>(expression: string, signal?: AbortSignal): Promise<T> {
    const res = await raceAbort(
      this.#conn.send(
        "Runtime.evaluate",
        { expression, returnByValue: true, awaitPromise: true },
        this.#sessionId,
      ),
      signal,
    );
    const exception = res["exceptionDetails"] as { text?: string; exception?: { description?: string } } | undefined;
    if (exception) {
      const text = exception.exception?.description ?? exception.text ?? "unknown error";
      throw new BrowserUnavailableError(`page script failed: ${text}`);
    }
    return (res["result"] as { value?: T } | undefined)?.value as T;
  }

  /**
   * Close the tab. Idempotent and never throws: it runs on error paths, where a
   * second failure would mask the one worth reporting.
   */
  async close(): Promise<void> {
    if (this.#closed) return;
    this.#closed = true;
    await this.#conn.send("Target.closeTarget", { targetId: this.#targetId }).catch(() => {});
  }
}