import { gunzipSync } from "node:zlib";
import { Agent, fetch as undiciFetch, type Dispatcher } from "undici";
import { assertUrlAllowed, safeLookup, UrlPolicyError, type FetchMeta, type Observation } from "@websensor/core";

/**
 * Generic HTTP fetcher: conditional GET (ETag / Last-Modified), manual redirect handling
 * with SSRF validation on every hop, size/time limits, and a dispatcher whose DNS lookup
 * only returns policy-approved addresses (DNS rebinding protection).
 */

export interface FetchOptions {
  method?: "GET" | "HEAD";
  etag?: string | null;
  lastModified?: string | null;
  headers?: Record<string, string>;
  timeoutMs?: number;
  maxBytes?: number;
  maxRedirects?: number;
  accept?: string;
  userAgent?: string;
  /** keep every response header (minus cookies' values) — used by the `headers` connector */
  keepAllHeaders?: boolean;
}

export class FetchError extends Error {
  constructor(
    public readonly code: string,
    message: string,
  ) {
    super(message);
    this.name = "FetchError";
  }
}

const DEFAULT_UA = process.env.WS_USER_AGENT ?? "WebSensorBot/0.1 (+https://www.websensor.io/bot; contact@websensor.io)";
const DEFAULT_TIMEOUT = 25_000;
const DEFAULT_MAX_BYTES = 12 * 1024 * 1024;

let agent: Dispatcher | null = null;
let agentH1: Dispatcher | null = null;
export function getDispatcher(h1only = false): Dispatcher {
  if (h1only) {
    if (!agentH1) agentH1 = new Agent({ connect: { lookup: safeLookup as never, timeout: 10_000 }, connections: 32, pipelining: 1, keepAliveTimeout: 15_000, headersTimeout: 25_000, bodyTimeout: 30_000, allowH2: false });
    return agentH1;
  }
  if (!agent) {
    agent = new Agent({
      connect: { lookup: safeLookup as never, timeout: 10_000 },
      connections: 64,
      pipelining: 1,
      keepAliveTimeout: 15_000,
      headersTimeout: 20_000,
      bodyTimeout: 30_000,
      allowH2: true,
    });
  }
  return agent;
}

/** Hosts where HTTP/2 misbehaved (NGHTTP2 stream errors, header timeouts) — pinned to HTTP/1.1. */
const h1Hosts = new Set<string>();

export async function closeDispatcher(): Promise<void> {
  if (agent) await agent.close();
  if (agentH1) await agentH1.close();
  agent = null;
  agentH1 = null;
}

/** A conventional browser identity, used only as a second attempt when the bot UA is refused (403). */
export const BROWSER_UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15";

function headerMap(h: Headers, all = false): Record<string, string> {
  const out: Record<string, string> = {};
  h.forEach((v, k) => {
    if (all || /^(content-type|content-length|etag|last-modified|cache-control|server|x-ratelimit-[a-z-]+|retry-after|date|age|via|cf-ray|x-cache|link)$/i.test(k)) out[k.toLowerCase()] = v;
  });
  if (all) {
    // undici joins multiple Set-Cookie into one; getSetCookie() keeps them apart.
    const sc = (h as unknown as { getSetCookie?: () => string[] }).getSetCookie?.();
    if (sc?.length) out["set-cookie"] = sc.join(", ");
  }
  return out;
}

export async function httpFetch(sensorId: string, urlStr: string, opts: FetchOptions = {}): Promise<Observation> {
  const started = Date.now();
  const method = opts.method ?? "GET";
  const maxRedirects = opts.maxRedirects ?? 5;
  let current = urlStr;
  let redirects = 0;
  const headers: Record<string, string> = {
    "user-agent": opts.userAgent ?? DEFAULT_UA,
    accept: opts.accept ?? "application/rss+xml, application/atom+xml, application/json, application/xml, text/html;q=0.9, text/plain;q=0.8, */*;q=0.5",
    "accept-language": "en-US,en;q=0.8,fr;q=0.5",
    "accept-encoding": "gzip, deflate, br",
    ...opts.headers,
  };
  if (opts.etag) headers["if-none-match"] = opts.etag;
  if (opts.lastModified) headers["if-modified-since"] = opts.lastModified;

  const fail = (code: string, message: string): Observation => ({
    sensorId,
    url: urlStr,
    fetchedAt: new Date(),
    notModified: false,
    error: { code, message },
    meta: { status: 0, url: urlStr, finalUrl: current, contentType: null, contentLength: 0, etag: null, lastModified: null, durationMs: Date.now() - started, redirects, method, headers: {} },
  });

  for (;;) {
    try {
      await assertUrlAllowed(current);
    } catch (e) {
      return fail("ssrf_blocked", e instanceof UrlPolicyError ? e.message : String(e));
    }
    const ac = new AbortController();
    const timer = setTimeout(() => ac.abort(), opts.timeoutMs ?? DEFAULT_TIMEOUT);
    let res: Response;
    try {
      const host = new URL(current).hostname;
      res = (await undiciFetch(current, { method, headers, redirect: "manual", signal: ac.signal, dispatcher: getDispatcher(h1Hosts.has(host)) } as never)) as unknown as Response;
    } catch (e) {
      clearTimeout(timer);
      const msg = e instanceof Error ? `${e.name}: ${e.message}${(e as { cause?: Error }).cause ? " — " + String((e as { cause?: Error }).cause?.message ?? (e as { cause?: unknown }).cause) : ""}` : String(e);
      if (/NGHTTP2|HTTP\/2/i.test(msg)) {
        const host = new URL(current).hostname;
        if (!h1Hosts.has(host)) {
          h1Hosts.add(host);
          continue; // retry this hop over HTTP/1.1
        }
      }
      const code = ac.signal.aborted ? "timeout" : /ENOTFOUND|EAI_AGAIN|getaddrinfo/i.test(msg) ? "dns" : /EBLOCKED|blocked/i.test(msg) ? "ssrf_blocked" : /CERT|TLS|SSL|certificate/i.test(msg) ? "tls" : /ECONNREFUSED|ECONNRESET|EHOSTUNREACH|ETIMEDOUT|socket/i.test(msg) ? "connection" : "fetch_failed";
      return fail(code, msg);
    }

    if ([301, 302, 303, 307, 308].includes(res.status)) {
      clearTimeout(timer);
      const loc = res.headers.get("location");
      if (!loc) return fail("redirect_without_location", `HTTP ${res.status} without Location`);
      if (++redirects > maxRedirects) return fail("too_many_redirects", `More than ${maxRedirects} redirects`);
      try {
        current = new URL(loc, current).toString();
      } catch {
        return fail("bad_redirect", `Invalid Location header: ${loc}`);
      }
      // conditional headers only apply to the original resource
      delete headers["if-none-match"];
      delete headers["if-modified-since"];
      continue;
    }

    const meta: FetchMeta = {
      status: res.status,
      url: urlStr,
      finalUrl: current,
      contentType: res.headers.get("content-type"),
      contentLength: 0,
      etag: res.headers.get("etag"),
      lastModified: res.headers.get("last-modified"),
      durationMs: 0,
      redirects,
      method,
      headers: headerMap(res.headers, opts.keepAllHeaders),
    };

    if (res.status === 304) {
      clearTimeout(timer);
      meta.durationMs = Date.now() - started;
      return { sensorId, url: urlStr, fetchedAt: new Date(), meta, notModified: true };
    }

    if (method === "HEAD") {
      clearTimeout(timer);
      meta.durationMs = Date.now() - started;
      meta.contentLength = Number(res.headers.get("content-length") ?? 0);
      return { sensorId, url: urlStr, fetchedAt: new Date(), meta, notModified: false };
    }

    const maxBytes = opts.maxBytes ?? DEFAULT_MAX_BYTES;
    const declared = Number(res.headers.get("content-length") ?? 0);
    if (declared > maxBytes) {
      clearTimeout(timer);
      return fail("too_large", `Content-Length ${declared} exceeds ${maxBytes}`);
    }
    const chunks: Uint8Array[] = [];
    let total = 0;
    try {
      const reader = res.body?.getReader();
      if (reader) {
        for (;;) {
          const { done, value } = await reader.read();
          if (done) break;
          total += value.byteLength;
          if (total > maxBytes) {
            await reader.cancel();
            clearTimeout(timer);
            return fail("too_large", `Body exceeded ${maxBytes} bytes`);
          }
          chunks.push(value);
        }
      }
    } catch (e) {
      clearTimeout(timer);
      return fail(ac.signal.aborted ? "timeout" : "body_read_failed", e instanceof Error ? e.message : String(e));
    }
    clearTimeout(timer);
    let body = Buffer.concat(chunks);
    // Some servers send gzip'd sitemaps as application/octet-stream without content-encoding.
    if (body.length > 2 && body[0] === 0x1f && body[1] === 0x8b) {
      try {
        body = gunzipSync(body);
      } catch {
        // keep as-is
      }
    }
    // UTF-16 with BOM (e.g. AWS Health `public/currentevents`) → transcode to UTF-8 so every parser downstream works.
    if (body.length > 2 && ((body[0] === 0xff && body[1] === 0xfe) || (body[0] === 0xfe && body[1] === 0xff))) {
      const le = body[0] === 0xff;
      const payload = body.subarray(2);
      const utf16 = le ? payload : Buffer.from(payload).swap16();
      body = Buffer.from(utf16.toString("utf16le"), "utf8");
    } else if (body.length > 3 && body[0] === 0xef && body[1] === 0xbb && body[2] === 0xbf) {
      body = body.subarray(3); // UTF-8 BOM
    }
    meta.contentLength = body.length;
    meta.durationMs = Date.now() - started;
    return { sensorId, url: urlStr, fetchedAt: new Date(), meta, body, notModified: false };
  }
}

/**
 * Outbound JSON POST (webhook alerts). Same SSRF policy and dispatcher as every fetch; no redirects
 * are followed (a webhook that redirects is misconfigured), small response cap, short timeout.
 */
export async function postJson(urlStr: string, body: string, headers: Record<string, string> = {}, timeoutMs = 10_000): Promise<{ status: number; error?: string }> {
  try {
    await assertUrlAllowed(urlStr);
  } catch (e) {
    return { status: 0, error: e instanceof UrlPolicyError ? e.message : String(e) };
  }
  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), timeoutMs);
  try {
    const res = (await undiciFetch(urlStr, { method: "POST", body, headers: { "content-type": "application/json", "user-agent": DEFAULT_UA, ...headers }, redirect: "manual", signal: ac.signal, dispatcher: getDispatcher(true) } as never)) as unknown as Response;
    // drain (bounded) so the socket can be reused
    const reader = res.body?.getReader();
    let total = 0;
    if (reader) {
      for (;;) {
        const { done, value } = await reader.read();
        if (done) break;
        total += value.byteLength;
        if (total > 64 * 1024) {
          await reader.cancel();
          break;
        }
      }
    }
    return { status: res.status };
  } catch (e) {
    return { status: 0, error: ac.signal.aborted ? "timeout" : e instanceof Error ? e.message : String(e) };
  } finally {
    clearTimeout(timer);
  }
}

/** Retry-aware wrapper for transient failures (5xx, connection, timeout). */
export async function httpFetchWithRetry(sensorId: string, url: string, opts: FetchOptions = {}, retries = 1): Promise<Observation> {
  let last: Observation | null = null;
  for (let attempt = 0; attempt <= retries; attempt++) {
    const obs = await httpFetch(sensorId, url, opts);
    last = obs;
    if (!obs.error && obs.meta.status === 403 && !opts.userAgent) {
      // Some WAFs refuse unknown bot identities on public feeds; try once as a regular browser.
      const alt = await httpFetch(sensorId, url, { ...opts, userAgent: BROWSER_UA, headers: { ...opts.headers, "sec-fetch-mode": "navigate", "sec-fetch-dest": "document", "upgrade-insecure-requests": "1" } });
      if (alt.error || alt.meta.status === 403) return obs;
      return alt;
    }
    if (obs.error && ["timeout", "connection", "fetch_failed"].includes(obs.error.code) && !opts.userAgent && attempt === 0) {
      // Some edges (Akamai) silently reset unknown bot identities at the TLS/HTTP layer.
      const alt = await httpFetch(sensorId, url, { ...opts, userAgent: BROWSER_UA });
      if (!alt.error) return alt;
    }
    const transient = obs.error ? ["timeout", "connection", "body_read_failed", "fetch_failed"].includes(obs.error.code) : obs.meta.status >= 500 && obs.meta.status !== 501;
    if (!transient) return obs;
    if (attempt < retries) await new Promise((r) => setTimeout(r, 800 * (attempt + 1)));
  }
  return last!;
}