spb/social-runtime-crawler

1import type { Page, Response, WebSocket } from "playwright";2import { createLogger, newId, nowIso, type Platform } from "@src/shared";3import type { EventBus } from "@src/events";4import { classifyResponse, type CapturedResponse, type ClassifiedResponse, type ClassifierHints } from "./ResponseClassifier.ts";56const log = createLogger("network");78const MAX_BODY = 3 * 1024 * 1024; // 3 MB per JSON body9const BODY_TIMEOUT_MS = 8_000; // give up on bodies that keep streaming (long-poll)10const COLLECT_TIMEOUT_MS = 6_000; // a step never waits longer than this for in-flight bodies11const STREAMING_URL = /\/ajax\/bz|\/pull\?|\/ajax\/presence|\/rsrc\.php|\/ajax\/mercury|\/subscriptions|\/realtime|sse|event-stream/i;12const IGNORED_HOST = /doubleclick|googlesyndication|google-analytics|googletagmanager|facebook\.com\/tr|scorecardresearch|sentry|datadog|hotjar|adservice|\/log_event|\/ptracking|\/csi_204|\/generate_204|\/youtubei\/v1\/log|\/api\/stats/i;1314export interface NetworkObserverOptions {15  page: Page;16  bus: EventBus;17  sessionId: string;18  platform: Platform;19  hints: ClassifierHints;20  keepBodies?: boolean; // keep JSON bodies in memory for the current step21}2223/**24 * NetworkObserver (§10): hooks Playwright response/websocket events, captures JSON/GraphQL/media25 * responses and runs them through the classifier. Results are buffered per step and published on the bus.26 */27export class NetworkObserver {28  private stepBuffer: ClassifiedResponse[] = [];29  private currentStep = 0;30  private currentActionId?: string;31  private detached: (() => void)[] = [];32  private inflight = new Set<Promise<void>>();33  readonly seenShapeHashes = new Map<string, number>();34  public totalResponses = 0;3536  constructor(private readonly opts: NetworkObserverOptions) {}3738  attach(): void {39    const { page } = this.opts;40    const onResponse = (res: Response) => {41      const p = this.handleResponse(res).catch((e) => log.debug("response handling failed", { err: (e as Error).message }));42      this.inflight.add(p);43      p.finally(() => this.inflight.delete(p));44    };45    const onWs = (ws: WebSocket) => this.handleWebSocket(ws);46    page.on("response", onResponse);47    page.on("websocket", onWs);48    this.detached.push(() => page.off("response", onResponse), () => page.off("websocket", onWs));49  }5051  detach(): void {52    for (const d of this.detached) d();53    this.detached = [];54  }5556  /** Called by the engine before each action. */57  beginStep(step: number, actionId?: string): void {58    this.currentStep = step;59    this.currentActionId = actionId;60    this.stepBuffer = [];61  }6263  /** Wait (bounded) for in-flight bodies then return what this step produced. */64  async collectStep(): Promise<ClassifiedResponse[]> {65    await Promise.race([Promise.allSettled([...this.inflight]), new Promise((r) => setTimeout(r, COLLECT_TIMEOUT_MS))]);66    return [...this.stepBuffer];67  }6869  private async handleResponse(res: Response): Promise<void> {70    const req = res.request();71    const url = res.url();72    if (IGNORED_HOST.test(url) || STREAMING_URL.test(url)) return;73    if (/text\/event-stream/.test(res.headers()["content-type"] ?? "")) return;74    const resourceType = req.resourceType();75    if (["font", "stylesheet", "script"].includes(resourceType)) return;76    const headers = res.headers();77    const ct = headers["content-type"] ?? "";78    const isJsonLike = /json|javascript|text\/plain/.test(ct) && ["xhr", "fetch", "other"].includes(resourceType);79    const isMedia = /video|audio|mpegurl|dash/.test(ct) || /videoplayback|\.m3u8|\.mpd/.test(url);80    const isFragment = /text\/html/.test(ct) && resourceType !== "document";81    if (!isJsonLike && !isMedia && !isFragment && resourceType !== "image") return;82    if (resourceType === "image") {83      // Only count images from media CDNs (thumbnails) — keep metadata only.84      if (!/ytimg|redd\.it|redditmedia|fbcdn|cdninstagram|tiktokcdn|twimg|licdn/i.test(url)) return;85    }8687    const captured: CapturedResponse = {88      request_id: newId("req"),89      url,90      method: req.method(),91      status: res.status(),92      content_type: ct,93      resource_type: resourceType,94      body_size: Number(headers["content-length"] ?? 0),95      post_data: req.postData()?.slice(0, 4000) ?? undefined,96      captured_at: nowIso(),97      step: this.currentStep,98    };99    if (isJsonLike || isFragment) {100      try {101        // Some platforms keep responses streaming for minutes (long-poll, chunked GraphQL subscriptions):102        // never wait more than BODY_TIMEOUT for a body — metadata alone is still a useful observation.103        const buf = await Promise.race([res.body(), new Promise<null>((r) => setTimeout(() => r(null), BODY_TIMEOUT_MS))]);104        if (buf) {105          captured.body_size = buf.length;106          if (buf.length <= MAX_BODY) captured.body = buf.toString("utf8");107        } else {108          captured.content_type += "; streaming";109        }110      } catch {111        // body may be unavailable (redirects, preflight) — keep metadata only112      }113    }114    this.totalResponses++;115    const classified = classifyResponse(captured, this.opts.hints);116    if (classified.kind === "other" && !captured.body) return;117    this.stepBuffer.push(classified);118119    const fp = classified.fingerprint;120    const firstSeen = !this.seenShapeHashes.has(fp.response_shape_hash);121    this.seenShapeHashes.set(fp.response_shape_hash, (this.seenShapeHashes.get(fp.response_shape_hash) ?? 0) + 1);122123    this.opts.bus.emit({124      event_type: "NETWORK_RESPONSE_OBSERVED",125      platform: this.opts.platform,126      session_id: this.opts.sessionId,127      step: this.currentStep,128      payload: {129        request_id: captured.request_id,130        url: captured.url,131        method: captured.method,132        status: captured.status,133        kind: classified.kind,134        content_type: fp.content_type,135        body_size: captured.body_size,136        shape_hash: fp.response_shape_hash,137        hostname: fp.hostname,138        path_pattern: fp.path_pattern,139        graphql_operation: fp.graphql_operation,140        entity_count: classified.entities.length,141        entity_types: fp.observed_entity_types,142        confidence: classified.confidence,143      },144      provenance: [{ surface: "network", confidence: classified.confidence }],145      discovered_via: { action_id: this.currentActionId },146    });147148    if (firstSeen && classified.schema && (classified.entities.length > 0 || classified.schema.candidate_entity_types.length > 0)) {149      this.opts.bus.emit({150        event_type: "NETWORK_SCHEMA_DISCOVERED",151        platform: this.opts.platform,152        session_id: this.opts.sessionId,153        step: this.currentStep,154        payload: {155          fingerprint: fp,156          schema: {157            shape_hash: classified.schema.shape_hash,158            root_type: classified.schema.root_type,159            repeated_object_paths: classified.schema.repeated_object_paths,160            candidate_entity_types: classified.schema.candidate_entity_types,161            object_count: classified.schema.object_count,162            fields: classified.schema.fields163              .filter((f) => f.semantic[0]?.kind !== "unknown")164              .slice(0, 80)165              .map((f) => ({ path: f.path, semantic: f.semantic[0], types: f.types, example: f.examples[0] })),166          },167          sample_url: captured.url,168        },169        provenance: [{ surface: "network", confidence: classified.confidence }],170      });171    }172  }173174  private handleWebSocket(ws: WebSocket): void {175    const url = ws.url();176    let frames = 0;177    ws.on("framereceived", (frame) => {178      frames++;179      if (frames > 60) return; // chat/presence sockets are chatty: keep only the first frames as evidence180      const payload = typeof frame.payload === "string" ? frame.payload : frame.payload.toString("utf8");181      this.opts.bus.emit({182        event_type: "WEBSOCKET_FRAME_OBSERVED",183        platform: this.opts.platform,184        session_id: this.opts.sessionId,185        step: this.currentStep,186        payload: { url, size: payload.length, looks_json: /^\s*[[{]/.test(payload), preview: payload.slice(0, 200) },187        provenance: [{ surface: "network", confidence: 0.6 }],188      });189    });190  }191}192