import type { Page, Response, WebSocket } from "playwright";
import { createLogger, newId, nowIso, type Platform } from "@src/shared";
import type { EventBus } from "@src/events";
import { classifyResponse, type CapturedResponse, type ClassifiedResponse, type ClassifierHints } from "./ResponseClassifier.ts";

const log = createLogger("network");

const MAX_BODY = 3 * 1024 * 1024; // 3 MB per JSON body
const BODY_TIMEOUT_MS = 8_000; // give up on bodies that keep streaming (long-poll)
const COLLECT_TIMEOUT_MS = 6_000; // a step never waits longer than this for in-flight bodies
const STREAMING_URL = /\/ajax\/bz|\/pull\?|\/ajax\/presence|\/rsrc\.php|\/ajax\/mercury|\/subscriptions|\/realtime|sse|event-stream/i;
const 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;

export interface NetworkObserverOptions {
  page: Page;
  bus: EventBus;
  sessionId: string;
  platform: Platform;
  hints: ClassifierHints;
  keepBodies?: boolean; // keep JSON bodies in memory for the current step
}

/**
 * NetworkObserver (§10): hooks Playwright response/websocket events, captures JSON/GraphQL/media
 * responses and runs them through the classifier. Results are buffered per step and published on the bus.
 */
export class NetworkObserver {
  private stepBuffer: ClassifiedResponse[] = [];
  private currentStep = 0;
  private currentActionId?: string;
  private detached: (() => void)[] = [];
  private inflight = new Set<Promise<void>>();
  readonly seenShapeHashes = new Map<string, number>();
  public totalResponses = 0;

  constructor(private readonly opts: NetworkObserverOptions) {}

  attach(): void {
    const { page } = this.opts;
    const onResponse = (res: Response) => {
      const p = this.handleResponse(res).catch((e) => log.debug("response handling failed", { err: (e as Error).message }));
      this.inflight.add(p);
      p.finally(() => this.inflight.delete(p));
    };
    const onWs = (ws: WebSocket) => this.handleWebSocket(ws);
    page.on("response", onResponse);
    page.on("websocket", onWs);
    this.detached.push(() => page.off("response", onResponse), () => page.off("websocket", onWs));
  }

  detach(): void {
    for (const d of this.detached) d();
    this.detached = [];
  }

  /** Called by the engine before each action. */
  beginStep(step: number, actionId?: string): void {
    this.currentStep = step;
    this.currentActionId = actionId;
    this.stepBuffer = [];
  }

  /** Wait (bounded) for in-flight bodies then return what this step produced. */
  async collectStep(): Promise<ClassifiedResponse[]> {
    await Promise.race([Promise.allSettled([...this.inflight]), new Promise((r) => setTimeout(r, COLLECT_TIMEOUT_MS))]);
    return [...this.stepBuffer];
  }

  private async handleResponse(res: Response): Promise<void> {
    const req = res.request();
    const url = res.url();
    if (IGNORED_HOST.test(url) || STREAMING_URL.test(url)) return;
    if (/text\/event-stream/.test(res.headers()["content-type"] ?? "")) return;
    const resourceType = req.resourceType();
    if (["font", "stylesheet", "script"].includes(resourceType)) return;
    const headers = res.headers();
    const ct = headers["content-type"] ?? "";
    const isJsonLike = /json|javascript|text\/plain/.test(ct) && ["xhr", "fetch", "other"].includes(resourceType);
    const isMedia = /video|audio|mpegurl|dash/.test(ct) || /videoplayback|\.m3u8|\.mpd/.test(url);
    const isFragment = /text\/html/.test(ct) && resourceType !== "document";
    if (!isJsonLike && !isMedia && !isFragment && resourceType !== "image") return;
    if (resourceType === "image") {
      // Only count images from media CDNs (thumbnails) — keep metadata only.
      if (!/ytimg|redd\.it|redditmedia|fbcdn|cdninstagram|tiktokcdn|twimg|licdn/i.test(url)) return;
    }

    const captured: CapturedResponse = {
      request_id: newId("req"),
      url,
      method: req.method(),
      status: res.status(),
      content_type: ct,
      resource_type: resourceType,
      body_size: Number(headers["content-length"] ?? 0),
      post_data: req.postData()?.slice(0, 4000) ?? undefined,
      captured_at: nowIso(),
      step: this.currentStep,
    };
    if (isJsonLike || isFragment) {
      try {
        // Some platforms keep responses streaming for minutes (long-poll, chunked GraphQL subscriptions):
        // never wait more than BODY_TIMEOUT for a body — metadata alone is still a useful observation.
        const buf = await Promise.race([res.body(), new Promise<null>((r) => setTimeout(() => r(null), BODY_TIMEOUT_MS))]);
        if (buf) {
          captured.body_size = buf.length;
          if (buf.length <= MAX_BODY) captured.body = buf.toString("utf8");
        } else {
          captured.content_type += "; streaming";
        }
      } catch {
        // body may be unavailable (redirects, preflight) — keep metadata only
      }
    }
    this.totalResponses++;
    const classified = classifyResponse(captured, this.opts.hints);
    if (classified.kind === "other" && !captured.body) return;
    this.stepBuffer.push(classified);

    const fp = classified.fingerprint;
    const firstSeen = !this.seenShapeHashes.has(fp.response_shape_hash);
    this.seenShapeHashes.set(fp.response_shape_hash, (this.seenShapeHashes.get(fp.response_shape_hash) ?? 0) + 1);

    this.opts.bus.emit({
      event_type: "NETWORK_RESPONSE_OBSERVED",
      platform: this.opts.platform,
      session_id: this.opts.sessionId,
      step: this.currentStep,
      payload: {
        request_id: captured.request_id,
        url: captured.url,
        method: captured.method,
        status: captured.status,
        kind: classified.kind,
        content_type: fp.content_type,
        body_size: captured.body_size,
        shape_hash: fp.response_shape_hash,
        hostname: fp.hostname,
        path_pattern: fp.path_pattern,
        graphql_operation: fp.graphql_operation,
        entity_count: classified.entities.length,
        entity_types: fp.observed_entity_types,
        confidence: classified.confidence,
      },
      provenance: [{ surface: "network", confidence: classified.confidence }],
      discovered_via: { action_id: this.currentActionId },
    });

    if (firstSeen && classified.schema && (classified.entities.length > 0 || classified.schema.candidate_entity_types.length > 0)) {
      this.opts.bus.emit({
        event_type: "NETWORK_SCHEMA_DISCOVERED",
        platform: this.opts.platform,
        session_id: this.opts.sessionId,
        step: this.currentStep,
        payload: {
          fingerprint: fp,
          schema: {
            shape_hash: classified.schema.shape_hash,
            root_type: classified.schema.root_type,
            repeated_object_paths: classified.schema.repeated_object_paths,
            candidate_entity_types: classified.schema.candidate_entity_types,
            object_count: classified.schema.object_count,
            fields: classified.schema.fields
              .filter((f) => f.semantic[0]?.kind !== "unknown")
              .slice(0, 80)
              .map((f) => ({ path: f.path, semantic: f.semantic[0], types: f.types, example: f.examples[0] })),
          },
          sample_url: captured.url,
        },
        provenance: [{ surface: "network", confidence: classified.confidence }],
      });
    }
  }

  private handleWebSocket(ws: WebSocket): void {
    const url = ws.url();
    let frames = 0;
    ws.on("framereceived", (frame) => {
      frames++;
      if (frames > 60) return; // chat/presence sockets are chatty: keep only the first frames as evidence
      const payload = typeof frame.payload === "string" ? frame.payload : frame.payload.toString("utf8");
      this.opts.bus.emit({
        event_type: "WEBSOCKET_FRAME_OBSERVED",
        platform: this.opts.platform,
        session_id: this.opts.sessionId,
        step: this.currentStep,
        payload: { url, size: payload.length, looks_json: /^\s*[[{]/.test(payload), preview: payload.slice(0, 200) },
        provenance: [{ surface: "network", confidence: 0.6 }],
      });
    });
  }
}