PRA-only-demo/src/main/browser/transport.ts

import type { BrowserWindow } from "electron";
import WebSocket from "ws";
import { RuntimeError } from "../engine/types";

/**
 * CDP 传输层抽象。
 *
 * 上层 BrowserSession 的全部 CDP 通信都经由本接口：
 * - EmbeddedTransport：内置运行浏览器（Electron BrowserWindow.webContents.debugger）
 * - ExternalTransport：外部 Edge/Chrome 浏览器（WebSocket 直连调试端口）
 *
 * 两者只在传输方式上不同，CDP 协议域（DOM/Input/Page/Network/Runtime）完全一致。
 */
export type CdpEventHandler = (params: Record<string, unknown>) => void;

export interface CdpTransport {
  readonly kind: "embedded" | "external";
  /** 连接是否已失效（窗口销毁 / ws 断开） */
  readonly closed: boolean;
  send<T = unknown>(method: string, params?: Record<string, unknown>): Promise<T>;
  /** 订阅 CDP 事件（如 Page.loadEventFired、Network.responseReceived），handler 只收到对应 method 的 params */
  on(method: string, handler: CdpEventHandler): void;
  off(method: string, handler: CdpEventHandler): void;
  close(): Promise<void>;
}

export function mapCdpError(error: unknown, kind: "embedded" | "external"): RuntimeError {
  const message = (error as Error)?.message || String(error);
  if (/destroyed|Target closed|not attached|connection.*(closed|lost)|websocket/i.test(message)) {
    if (kind === "external") {
      return new RuntimeError("browser_disconnected", "外部浏览器连接已断开", {
        detail: message,
        recommendation: "确认浏览器仍处于打开状态；若被手动关闭请重新运行「打开网页」"
      });
    }
    return new RuntimeError("tab_unavailable", "浏览器连接已失效", { detail: message });
  }
  return new RuntimeError("browser_error", message, { recommendation: "重试或检查页面状态" });
}

/** 内置运行浏览器：包装 webContents.debugger */
export class EmbeddedTransport implements CdpTransport {
  readonly kind = "embedded" as const;
  private wrappers = new Map<CdpEventHandler, (event: unknown, method: string, params?: Record<string, unknown>) => void>();

  constructor(private win: BrowserWindow) {}

  get closed(): boolean {
    return this.win.isDestroyed();
  }

  async send<T = unknown>(method: string, params: Record<string, unknown> = {}): Promise<T> {
    if (this.win.isDestroyed()) {
      throw new RuntimeError("tab_unavailable", "运行浏览器已关闭", { recommendation: "重新运行流程" });
    }
    try {
      if (!this.win.webContents.debugger.isAttached()) this.win.webContents.debugger.attach("1.3");
      return (await this.win.webContents.debugger.sendCommand(method, params)) as T;
    } catch (error) {
      throw mapCdpError(error, "embedded");
    }
  }

  on(method: string, handler: CdpEventHandler): void {
    if (this.win.isDestroyed()) return;
    const wrapper = (_event: unknown, evtMethod: string, evtParams?: Record<string, unknown>) => {
      if (evtMethod === method) handler(evtParams ?? {});
    };
    this.wrappers.set(handler, wrapper);
    this.win.webContents.debugger.on("message", wrapper as never);
  }

  off(method: string, handler: CdpEventHandler): void {
    const wrapper = this.wrappers.get(handler);
    if (wrapper) {
      this.win.webContents.debugger.removeListener("message", wrapper as never);
      this.wrappers.delete(handler);
    }
  }

  async close(): Promise<void> {
    if (!this.win.isDestroyed() && this.win.webContents.debugger.isAttached()) {
      try {
        this.win.webContents.debugger.detach();
      } catch {
        // 窗口关闭过程中 detach 可能失败，忽略
      }
    }
  }
}

interface CdpMessage {
  id?: number;
  method?: string;
  params?: Record<string, unknown>;
  result?: unknown;
  error?: { message?: string; data?: unknown };
}

/** 外部 Edge/Chrome：WebSocket 直连页面级 CDP 端点（webSocketDebuggerUrl） */
export class ExternalTransport implements CdpTransport {
  readonly kind = "external" as const;
  private ws: WebSocket | null = null;
  private nextId = 1;
  private pending = new Map<number, { resolve: (value: unknown) => void; reject: (error: Error) => void }>();
  private handlers = new Map<string, Set<CdpEventHandler>>();
  private closedFlag = false;
  private readonly openPromise: Promise<void>;

  private constructor(
    wsUrl: string,
    readonly description: string
  ) {
    this.openPromise = new Promise<void>((resolve, reject) => {
      const ws = new WebSocket(wsUrl, { maxPayload: 256 * 1024 * 1024 });
      this.ws = ws;
      ws.on("open", () => resolve());
      ws.on("message", (data: unknown) => this.handleMessage(String(data)));
      ws.on("error", (error: Error) => {
        this.closedFlag = true;
        this.failAllPending(new Error(`外部浏览器连接异常：${error.message}`));
        reject(error);
      });
      ws.on("close", () => {
        this.closedFlag = true;
        this.failAllPending(new Error("外部浏览器连接已关闭"));
      });
    });
  }

  static async connect(wsUrl: string, description: string, timeoutMs = 10000): Promise<ExternalTransport> {
    const transport = new ExternalTransport(wsUrl, description);
    const timeout = new Promise<never>((_, reject) =>
      setTimeout(() => reject(new RuntimeError("browser_disconnected", `连接外部浏览器超时：${description}`)), timeoutMs)
    );
    await Promise.race([transport.openPromise, timeout]);
    return transport;
  }

  get closed(): boolean {
    return this.closedFlag || !this.ws || this.ws.readyState !== this.ws.OPEN;
  }

  async send<T = unknown>(method: string, params: Record<string, unknown> = {}): Promise<T> {
    if (this.closed) {
      throw new RuntimeError("browser_disconnected", "外部浏览器连接已断开", {
        recommendation: "确认浏览器未被手动关闭，然后重新运行「打开网页」"
      });
    }
    const id = this.nextId++;
    const ws = this.ws as WebSocket;
    const reply = new Promise<unknown>((resolve, reject) => {
      this.pending.set(id, { resolve, reject });
      try {
        ws.send(JSON.stringify({ id, method, params }));
      } catch (error) {
        this.pending.delete(id);
        reject(error as Error);
      }
    });
    try {
      return (await reply) as T;
    } catch (error) {
      throw mapCdpError(error, "external");
    }
  }

  on(method: string, handler: CdpEventHandler): void {
    if (!this.handlers.has(method)) this.handlers.set(method, new Set());
    this.handlers.get(method)?.add(handler);
  }

  off(method: string, handler: CdpEventHandler): void {
    this.handlers.get(method)?.delete(handler);
  }

  async close(): Promise<void> {
    const ws = this.ws;
    this.closedFlag = true;
    this.failAllPending(new Error("外部浏览器连接已主动关闭"));
    if (!ws) return;
    await new Promise<void>((resolve) => {
      if (ws.readyState === ws.CLOSED) return resolve();
      ws.once("close", () => resolve());
      try {
        ws.close();
      } catch {
        resolve();
      }
      setTimeout(() => resolve(), 1500);
    });
  }

  private handleMessage(raw: string): void {
    let message: CdpMessage;
    try {
      message = JSON.parse(raw) as CdpMessage;
    } catch {
      return;
    }
    if (typeof message.id === "number") {
      const entry = this.pending.get(message.id);
      if (!entry) return;
      this.pending.delete(message.id);
      if (message.error) {
        entry.reject(new Error(message.error.message || "CDP 调用失败"));
      } else {
        entry.resolve(message.result);
      }
      return;
    }
    if (message.method) {
      const handlers = this.handlers.get(message.method);
      if (!handlers) return;
      const params = message.params ?? {};
      for (const handler of handlers) {
        try {
          handler(params);
        } catch {
          // 单个监听器异常不影响其他监听器
        }
      }
    }
  }

  private failAllPending(error: Error): void {
    for (const [id, entry] of this.pending) {
      entry.reject(error);
      this.pending.delete(id);
    }
  }
}