import { Injectable } from '@nestjs/common';
import * as ivm from 'isolated-vm';
import * as dns from 'node:dns/promises';
import * as net from 'node:net';
import { resolveRpc } from './planfi-rpc-catalog';

export interface SandboxInput {
  code: string;
  input?: unknown;
  limits: { timeoutMs: number; memoryMb: number };
  /**
   * Read-only key/value bag exposed to user code as `globalThis.env`. Copied
   * into the isolate and frozen at bootstrap — never the host `process.env`.
   */
  env?: Record<string, string>;
  /**
   * Exact hostnames (case-insensitive) the in-isolate `fetch` is allowed to
   * reach. Fail-closed: an empty/absent allowlist blocks every request.
   */
  fetchAllowlist?: string[];
  /**
   * When set, injects a `planfi` global into the isolate exposing
   * `planfi.call(...)`. The PlanFi base URL is the deployment env var
   * `SCHEDULED_JOBS_PLANFI_BASE_URL`; absent → `planfi` not injected.
   */
  planfi?: { baseUrl: string; writeMode?: 'safe' | 'real' };
}

export interface SandboxResult {
  status: 'SUCCESS' | 'FAILED' | 'TIMED_OUT';
  output?: unknown;
  logs: string[];
  error?: { message: string; stack?: string };
  durationMs: number;
}

/**
 * Serializable shape returned by the in-isolate `fetch`. Deliberately a plain
 * object with no streams/functions so it can cross the isolate boundary via
 * ExternalCopy.
 */
interface SandboxFetchResponse {
  ok: boolean;
  status: number;
  statusText: string;
  headers: Record<string, string>;
  body: string;
}

const LOG_CAP_BYTES = 256 * 1024; // 256 KB
const OUTPUT_CAP_BYTES = 512 * 1024; // 512 KB

// Hard ceilings for the host-side fetch.
const FETCH_TIMEOUT_CEILING_MS = 10_000;
const FETCH_BODY_CAP_BYTES = 1024 * 1024; // 1 MB

// planfi.call ceiling. Unlike fetch() — arbitrary external hosts, kept tight
// at 10s — planfi.call is locked same-origin to the operator-configured PlanFi
// API and legitimately runs heavy internal RPCs (segment materialization,
// usage sync). It may therefore spend the job's own timeout budget, capped at
// the maximum configurable job timeout (mirrors the bundle schema max).
const PLANFI_CALL_TIMEOUT_CEILING_MS = 900_000;

/** Effective per-call timeout for planfi.call: the job's timeout budget,
 *  falling back to the fetch default when absent, capped at 15 min. */
export function planfiCallTimeoutMs(jobTimeoutMs: number): number {
  return Math.min(
    Math.max(1, jobTimeoutMs || FETCH_TIMEOUT_CEILING_MS),
    PLANFI_CALL_TIMEOUT_CEILING_MS,
  );
}

// Per-run network limits.
const FETCH_MAX_COUNT = 50; // max total fetches per run
const FETCH_MAX_CONCURRENT = 4; // max in-flight at once

// Max redirect hops before we reject.
const REDIRECT_MAX_HOPS = 3;

// Hard ceiling on wall-clock time per run (ms). Even if the job's CPU quota is
// higher, the wall-clock timer fires at this ceiling and disposes the isolate.
const WALL_CLOCK_CEILING_MS = 120_000;

/**
 * Wall-clock budget for a run: timeoutMs * 3 clamped to the ceiling — but the
 * ceiling never truncates the job's own configured timeout (a 5-min job must
 * get 5 min of wall time, not 2). The result always stays inside the
 * processor's lock envelope (lockTtl = timeoutMs + 90s), so a long run cannot
 * outlive its queue lock.
 */
export function computeWallClockMs(timeoutMs: number): number {
  return Math.min(timeoutMs * 3, Math.max(WALL_CLOCK_CEILING_MS, timeoutMs));
}

// HTTP methods user code may request.
const ALLOWED_METHODS = new Set([
  'GET',
  'POST',
  'PUT',
  'PATCH',
  'DELETE',
  'HEAD',
]);

const WRITE_METHODS = new Set(['POST', 'PUT', 'PATCH', 'DELETE']);

// Headers that user code must NOT be able to inject (hop-by-hop / forwarding).
const BLOCKED_REQUEST_HEADERS = new Set([
  'host',
  'forwarded',
  'x-forwarded-for',
  'x-forwarded-host',
  'x-forwarded-proto',
  'x-forwarded-port',
  'x-real-ip',
  'connection',
  'keep-alive',
  'proxy-authenticate',
  'proxy-authorization',
  'te',
  'trailers',
  'transfer-encoding',
  'upgrade',
]);

function sanitizeToString(value: unknown): string {
  if (typeof value === 'string') return value;
  try {
    return String(value);
  } catch {
    return '<non-stringifiable>';
  }
}

// Log args: objects/arrays render as JSON instead of "[object Object]".
// Circular/non-serializable values fall back to String().
function formatLogArg(value: unknown): string {
  if (value !== null && typeof value === 'object') {
    try {
      const json = JSON.stringify(value);
      if (json !== undefined) return json;
    } catch {
      // fall through to sanitizeToString
    }
  }
  return sanitizeToString(value);
}

// Canonical timeout message thrown by isolated-vm when a script exceeds its
// budget. We match on this exact phrase instead of a generic /time|timeout/
// regex, which would falsely flag legitimate user errors mentioning 'runtime',
// 'datetime', 'timestamp', etc. Paired with the wall-clock timer for I/O loops.
const IVM_TIMEOUT_MESSAGE = /Script execution timed out/;

function truncateJson(value: unknown): unknown {
  let serialized: string;
  try {
    serialized = JSON.stringify(value);
  } catch {
    return '[non-serializable return value]';
  }

  // JSON.stringify returns the literal string `undefined` (i.e. the JS value
  // undefined, not a string) for undefined, functions and symbols. Accessing
  // .length on it would throw a TypeError. Treat that as "no output".
  if (serialized === undefined) {
    return undefined;
  }

  if (serialized.length > OUTPUT_CAP_BYTES) {
    return serialized.slice(0, OUTPUT_CAP_BYTES) + '…[truncado]';
  }
  return value;
}

/**
 * SSRF guard: returns true when `ip` falls in a private, loopback, link-local
 * or unique-local range that user code must never be able to reach.
 *
 * FIX-7: Strip IPv6 brackets before classifying; handle IPv4-mapped hex form
 * (`::ffff:7f00:1`) in addition to dotted form; fail-closed on anything
 * unclassifiable.
 *
 * Covers (per spec):
 *   IPv4 — 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, 127.0.0.0/8,
 *          169.254.0.0/16, 0.0.0.0/8
 *   IPv6 — ::1 (loopback), :: (unspecified), fc00::/7 (ULA),
 *          fe80::/10 (link-local), ::ffff:a.b.c.d / ::ffff:HHHH:HHHH
 */
export function isPrivateOrReservedIp(ip: string): boolean {
  // FIX-7: strip brackets from IPv6 literals (e.g. "[::1]" → "::1").
  let candidate = ip;
  if (candidate.startsWith('[') && candidate.endsWith(']')) {
    candidate = candidate.slice(1, -1);
  }

  const family = net.isIP(candidate);
  if (family === 4) {
    const parts = candidate.split('.').map((p) => Number(p));
    if (parts.length !== 4 || parts.some((n) => Number.isNaN(n))) {
      // Unparseable — treat as unsafe (fail-closed).
      return true;
    }
    const [a, b] = parts;
    if (a === 10) return true; // 10.0.0.0/8
    if (a === 172 && b >= 16 && b <= 31) return true; // 172.16.0.0/12
    if (a === 192 && b === 168) return true; // 192.168.0.0/16
    if (a === 127) return true; // 127.0.0.0/8 (loopback)
    if (a === 169 && b === 254) return true; // 169.254.0.0/16 (link-local)
    if (a === 0) return true; // 0.0.0.0/8 ("this network")
    return false;
  }

  if (family === 6) {
    let v6 = candidate.toLowerCase();
    // Strip a zone id (e.g. fe80::1%eth0) before classifying.
    const pct = v6.indexOf('%');
    if (pct !== -1) v6 = v6.slice(0, pct);

    if (v6 === '::1' || v6 === '::') return true; // loopback / unspecified

    // FIX-7: IPv4-mapped — dotted form  ::ffff:127.0.0.1
    const mappedDotted = v6.match(
      /^::ffff:(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$/,
    );
    if (mappedDotted) return isPrivateOrReservedIp(mappedDotted[1]);

    // FIX-7: IPv4-mapped — hex form  ::ffff:7f00:0001  (::ffff:HHHH:HHHH)
    const mappedHex = v6.match(/^::ffff:([0-9a-f]{1,4}):([0-9a-f]{1,4})$/);
    if (mappedHex) {
      const hi = parseInt(mappedHex[1], 16);
      const lo = parseInt(mappedHex[2], 16);
      const dotted = `${(hi >> 8) & 0xff}.${hi & 0xff}.${(lo >> 8) & 0xff}.${lo & 0xff}`;
      return isPrivateOrReservedIp(dotted);
    }

    const firstHextet = v6.split(':')[0];
    const high = parseInt(firstHextet || '0', 16);
    if (Number.isNaN(high)) return true; // fail-closed
    // fc00::/7 (ULA): first 7 bits are 1111110 → 0xfc00..0xfdff.
    if (high >= 0xfc00 && high <= 0xfdff) return true;
    // fe80::/10 (link-local): 0xfe80..0xfebf.
    if (high >= 0xfe80 && high <= 0xfebf) return true;
    return false;
  }

  // Not a literal IP we can classify — fail-closed.
  return true;
}

/**
 * Validates a single URL (after parsing) for protocol + allowlist + DNS/SSRF.
 * Extracted so the redirect loop can re-run the FULL guard on each hop.
 *
 * Throws with a user-safe message on any violation.
 */
async function guardUrl(
  url: URL,
  allowlist: string[],
  opts: { allowPrivate?: boolean } = {},
): Promise<void> {
  if (url.protocol !== 'http:' && url.protocol !== 'https:') {
    throw new Error(
      `fetch bloqueado: protocolo "${url.protocol}" não permitido (apenas http/https)`,
    );
  }

  // FIX-7: strip brackets from IPv6 host literals before checking allowlist.
  let hostname = url.hostname.toLowerCase();
  if (hostname.startsWith('[') && hostname.endsWith(']')) {
    hostname = hostname.slice(1, -1);
  }

  const normalizedAllowlist = (allowlist || []).map((h) =>
    String(h).toLowerCase(),
  );
  if (normalizedAllowlist.length === 0) {
    throw new Error(
      'fetch bloqueado: a allowlist de domínios do job está vazia (fail-closed)',
    );
  }
  if (!normalizedAllowlist.includes(hostname)) {
    throw new Error(
      `fetch bloqueado: domínio "${hostname}" fora da allowlist do job`,
    );
  }

  // SSRF guard: resolve the hostname and reject if ANY resolved address is
  // private/loopback/link-local/ULA. If the hostname is already a literal IP,
  // check it directly.
  //
  // SECURITY RESIDUAL (DNS rebinding TOCTOU): the IP validated by dns.lookup
  // may differ from the IP that the underlying TCP stack connects to if the
  // DNS entry changes between our check and fetch's connect (TTL 0 attack).
  // Full mitigation requires IP-pinning via a custom resolver dispatcher (e.g.
  // undici Agent with a connect.lookup hook that returns only the validated IP).
  // We do not add undici as a new runtime dependency here; instead the redirect
  // fix (#1) closes the main practical attack vector. A future hardening pass
  // should add undici (already transitive in Node 18+ environments) and pin
  // the resolved IP. Risk level: LOW in practice given the allowlist + redirect
  // fix + the short DNS window; HIGH in theory.
  if (net.isIP(hostname)) {
    if (!opts.allowPrivate && isPrivateOrReservedIp(hostname)) {
      throw new Error(
        `fetch bloqueado: IP "${hostname}" é privado/reservado (SSRF)`,
      );
    }
  } else {
    let addresses: { address: string }[];
    try {
      addresses = await dns.lookup(hostname, { all: true });
    } catch {
      throw new Error(`fetch bloqueado: falha ao resolver "${hostname}"`);
    }
    if (addresses.length === 0) {
      throw new Error(
        `fetch bloqueado: "${hostname}" não resolveu para nenhum endereço`,
      );
    }
    for (const { address } of addresses) {
      if (!opts.allowPrivate && isPrivateOrReservedIp(address)) {
        throw new Error(
          `fetch bloqueado: "${hostname}" resolve para IP privado/reservado (SSRF)`,
        );
      }
    }
  }
}

/**
 * Host-side implementation of the in-isolate `fetch`. Receives the URL and a
 * JSON-stringified init (both as plain strings across the boundary), enforces
 * the allowlist + SSRF guard, performs the request, and returns a fully
 * serializable {@link SandboxFetchResponse}.
 *
 * Security hardening applied (all fixes implemented in this function):
 *   FIX-1  Redirect SSRF: redirect:'manual' hard-set; Location header
 *          re-validated through guardUrl on every hop; max REDIRECT_MAX_HOPS.
 *   FIX-3  Network budget: fetch count + concurrency semaphore enforced via
 *          counters passed in from the run() call site.
 *   FIX-4  Response body OOM: streamed via getReader() with byte cap; stream
 *          cancelled as soon as FETCH_BODY_CAP_BYTES is exceeded.
 *   FIX-6  User init sanitization: method restricted to ALLOWED_METHODS;
 *          headers filtered to drop hop-by-hop/forwarding; body size capped;
 *          redirect and signal hard-set LAST (cannot be overridden by user).
 *   FIX-7  IPv6 bracket stripping + hex-form IPv4-mapped handled in guardUrl.
 *   FIX-8  Host errors sanitized to generic messages before crossing the
 *          isolate boundary.
 *
 * Kept as a module-level function (not a class method) so the SandboxRunner
 * structural type stays `{ run() }` — the processor's tests pass a `{ run }`
 * mock, which would otherwise stop being assignable — and so tests can mock
 * global fetch and dns without reaching into private scope.
 */
export async function hostFetch(
  rawUrl: string,
  rawInit: string,
  allowlist: string[],
  timeoutMs: number,
  fetchCounter: { count: number },
  concurrencyCounter: { inFlight: number },
  // ARFIX-02: per-run registry of in-flight AbortControllers so the run()
  // teardown / wall-clock dispose can cancel outstanding host fetches en masse.
  // Optional (defaults to a throwaway Set) so existing direct callers and the
  // `{ run }` structural-type mock stay assignable.
  activeControllers: Set<AbortController> = new Set(),
): Promise<SandboxFetchResponse> {
  // Yield to the microtask queue before any (possibly synchronous) throw.
  // isolated-vm's `apply({ result: { promise: true } })` only marshals a
  // rejection back into the isolate as a catchable error when the host
  // function rejects *asynchronously*; a synchronous throw before the first
  // await leaks as an unhandled host rejection instead. This guarantees every
  // policy violation surfaces as a rejected Promise the user code can try/catch.
  await Promise.resolve();

  // FIX-3: per-run fetch budget. The count budget is checked (and consumed)
  // BEFORE reserving a concurrency slot so an over-count attempt never reserves.
  fetchCounter.count += 1;
  if (fetchCounter.count > FETCH_MAX_COUNT) {
    throw new Error(
      'fetch bloqueado: limite de requisições por execução atingido',
    );
  }

  // ARFIX-03: reserve-before-validate. Increment the concurrency counter
  // synchronously (no await in between) so N concurrent calls cannot all pass a
  // check-then-reserve gap and exceed FETCH_MAX_CONCURRENT. Roll the reservation
  // back immediately if it pushed us over the cap.
  concurrencyCounter.inFlight += 1;
  if (concurrencyCounter.inFlight > FETCH_MAX_CONCURRENT) {
    concurrencyCounter.inFlight -= 1;
    throw new Error(
      'fetch bloqueado: limite de requisições simultâneas atingido',
    );
  }

  // From here on the reservation is held; the try/finally below guarantees the
  // decrement on EVERY exit path (including the URL/init parsing throws).
  let controller: AbortController | undefined;
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    let url: URL;
    try {
      url = new URL(rawUrl);
    } catch {
      throw new Error('fetch bloqueado: URL inválida');
    }

    // FIX-6: parse and sanitize user init — explicit allowlist, not spread.
    let parsedInit: Record<string, unknown> = {};
    try {
      parsedInit = rawInit
        ? (JSON.parse(rawInit) as Record<string, unknown>)
        : {};
    } catch {
      parsedInit = {};
    }

    // Method: restrict to known-safe set; default GET.
    const rawMethod =
      typeof parsedInit['method'] === 'string'
        ? parsedInit['method'].toUpperCase()
        : 'GET';
    const method = ALLOWED_METHODS.has(rawMethod) ? rawMethod : 'GET';

    // Headers: filter hop-by-hop / forwarding headers.
    const safeHeaders: Record<string, string> = {};
    if (parsedInit['headers'] && typeof parsedInit['headers'] === 'object') {
      for (const [k, v] of Object.entries(
        parsedInit['headers'] as Record<string, unknown>,
      )) {
        if (
          typeof k === 'string' &&
          typeof v === 'string' &&
          !BLOCKED_REQUEST_HEADERS.has(k.toLowerCase())
        ) {
          safeHeaders[k] = v;
        }
      }
    }

    // Body: only allow string values; cap at FETCH_BODY_CAP_BYTES.
    let body: string | undefined;
    if (typeof parsedInit['body'] === 'string') {
      body =
        parsedInit['body'].length > FETCH_BODY_CAP_BYTES
          ? parsedInit['body'].slice(0, FETCH_BODY_CAP_BYTES)
          : parsedInit['body'];
    }

    // Build the sanitized RequestInit (redirect and signal set LAST — cannot be
    // overridden by any user-supplied field).
    const effectiveTimeout = Math.min(
      Math.max(1, timeoutMs || FETCH_TIMEOUT_CEILING_MS),
      FETCH_TIMEOUT_CEILING_MS,
    );
    controller = new AbortController();
    // ARFIX-02: register so run() teardown can abort this in-flight fetch.
    activeControllers.add(controller);
    timer = setTimeout(() => controller!.abort(), effectiveTimeout);

    // FIX-1: Redirect SSRF — follow redirects manually so each hop is re-validated.
    let currentUrl = url;
    let hops = 0;

    // Validate the initial URL.
    await guardUrl(currentUrl, allowlist);

    let response: Response;
    while (true) {
      // Build init with redirect:'manual' hard-set LAST so user cannot override.
      const fetchInit: RequestInit = {
        method,
        headers: safeHeaders,
        ...(body !== undefined ? { body } : {}),
        // FIX-1: hard-set redirect LAST — no user value can override this.
        redirect: 'manual',
        signal: controller.signal,
      };

      try {
        response = await fetch(currentUrl.toString(), fetchInit);
      } catch (err) {
        if (err instanceof Error && err.name === 'AbortError') {
          throw new Error(
            `fetch bloqueado: timeout de ${effectiveTimeout}ms excedido`,
          );
        }
        // FIX-8: sanitize transport errors.
        throw new Error('fetch falhou: erro de rede');
      }

      // FIX-1: if this is a redirect, validate the Location before following.
      if (
        response.status >= 300 &&
        response.status < 400 &&
        response.headers.has('location')
      ) {
        hops += 1;
        if (hops > REDIRECT_MAX_HOPS) {
          throw new Error(
            `fetch bloqueado: limite de redirecionamentos (${REDIRECT_MAX_HOPS}) atingido`,
          );
        }

        const location = response.headers.get('location') as string;
        let nextUrl: URL;
        try {
          // Resolve relative redirects against the current URL.
          nextUrl = new URL(location, currentUrl.toString());
        } catch {
          throw new Error(
            'fetch bloqueado: Location de redirecionamento inválida',
          );
        }

        // FIX-1: re-run the FULL guard (protocol + allowlist + DNS/SSRF) on
        // the redirect target before following.
        await guardUrl(nextUrl, allowlist);

        currentUrl = nextUrl;
        // Continue the loop to follow the redirect.
        continue;
      }

      // Not a redirect — we have the final response.
      break;
    }

    // FIX-4: stream body with byte cap; cancel as soon as we exceed the limit.
    let bodyText: string;
    if (response.body) {
      const reader = response.body.getReader();
      const chunks: Uint8Array[] = [];
      let totalBytes = 0;
      let truncated = false;

      try {
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          if (value) {
            totalBytes += value.byteLength;
            if (totalBytes > FETCH_BODY_CAP_BYTES) {
              // Accept up to the cap from this chunk, then cancel.
              const remaining =
                FETCH_BODY_CAP_BYTES - (totalBytes - value.byteLength);
              if (remaining > 0) {
                chunks.push(value.slice(0, remaining));
              }
              truncated = true;
              await reader.cancel();
              break;
            }
            chunks.push(value);
          }
        }
      } catch {
        // Reader may throw if the response is aborted; that's fine — we already
        // have what we need.
      }

      const combined = new Uint8Array(
        chunks.reduce((acc, c) => acc + c.byteLength, 0),
      );
      let offset = 0;
      for (const chunk of chunks) {
        combined.set(chunk, offset);
        offset += chunk.byteLength;
      }
      bodyText = new TextDecoder().decode(combined);
      if (truncated) bodyText += '…[truncado]';
    } else {
      bodyText = '';
    }

    const headers: Record<string, string> = {};
    response.headers.forEach((value, key) => {
      headers[key] = value;
    });

    return {
      ok: response.ok,
      status: response.status,
      statusText: response.statusText,
      headers,
      body: bodyText,
    };
  } finally {
    if (timer !== undefined) clearTimeout(timer);
    // ARFIX-02: unregister this fetch's controller (no-op if never created).
    if (controller !== undefined) activeControllers.delete(controller);
    // ARFIX-03: release the reservation made before validation.
    concurrencyCounter.inFlight -= 1;
  }
}

/** A chamada HTTP "crua" (escape hatch avançado de `planfi.call({path,...})`). */
interface PlanfiRawCall {
  method?: string;
  path: string;
  query?: Record<string, string>;
  body?: string;
  headers?: Record<string, string>;
}

/**
 * Serializable input for the in-isolate `planfi.call`. The isolate sends this as
 * a JSON string across the boundary. Dois modos:
 *  - RPC (primário): `{ rpc: "customer.list", params: {...} }` → resolvido pelo catálogo.
 *  - Cru (avançado): `{ raw: { path, method, query, body, headers } }`.
 */
interface PlanfiCallInput {
  rpc?: string;
  params?: Record<string, unknown>;
  raw?: PlanfiRawCall;
}

/**
 * Host-side implementation of `planfi.call`. Behaves like `hostFetch` but:
 *   - Always targets the configured PlanFi base URL (no allowlist check needed
 *     for the origin itself, but origin-escape from the `path` param is blocked).
 *   - Injects `X-Planfi-Service-Key` from `env.PLANFI_KEY` (hard-set last,
 *     cannot be overridden by user-supplied headers).
 *   - Reuses ALL existing hardening: SSRF guard, budget/concurrency counters,
 *     redirect:manual + re-validation, streaming body cap, sanitized errors.
 *
 * Security properties:
 *   1. Origin-escape: the resolved URL's host MUST equal the baseUrl host.
 *      This prevents `path: "//evil.com/x"` or `path: "https://evil.com/x"`.
 *   2. SSRF: the SSRF private-IP guard in `guardUrl` runs on the baseUrl host;
 *      if the PlanFi base resolves to a private IP it is rejected.
 *   3. Auth override: `X-Planfi-Service-Key` is set LAST, overwriting any
 *      user-supplied header with the same name.
 *   4. Shared budget: fetchCounter + concurrencyCounter are the SAME objects
 *      used by `hostFetch`, so total network activity is bounded together.
 */
export async function hostPlanfiCall(
  rawInit: string,
  baseUrl: string,
  planfiKey: string,
  timeoutMs: number,
  fetchCounter: { count: number },
  concurrencyCounter: { inFlight: number },
  writeMode?: 'safe' | 'real',
  // ARFIX-02: per-run registry of in-flight AbortControllers (see hostFetch).
  // Optional so existing direct callers stay assignable.
  activeControllers: Set<AbortController> = new Set(),
): Promise<unknown> {
  // Yield first — same reasoning as hostFetch: ensures async rejection so the
  // isolate can catch it via `promise: true`.
  await Promise.resolve();

  if (!planfiKey) {
    throw new Error(
      'planfi.call sem credencial: configure a credencial dos Jobs Agendados (SCHEDULED_JOBS_INTERNAL_PLANFI_KEY no executor) ou um segredo PLANFI_KEY no job',
    );
  }

  // FIX-3: shared network budget. The count budget is consumed first; the
  // concurrency reservation (ARFIX-03) happens later, AFTER the dry-run early
  // return, so a dry-run never reserves/leaks a slot.
  fetchCounter.count += 1;
  if (fetchCounter.count > FETCH_MAX_COUNT) {
    throw new Error(
      'planfi.call bloqueado: limite de requisições por execução atingido',
    );
  }

  let payload: PlanfiCallInput;
  try {
    payload = JSON.parse(rawInit) as PlanfiCallInput;
  } catch {
    payload = { raw: { path: '/' } };
  }

  // Normaliza os dois modos (RPC via catálogo / chamada crua) numa única forma.
  const isRpc = typeof payload.rpc === 'string';
  let rpcUnwrap: string | undefined;
  let call: PlanfiRawCall;
  if (isRpc) {
    const resolved = resolveRpc(payload.rpc as string, payload.params);
    rpcUnwrap = resolved.unwrap;
    call = {
      method: resolved.method,
      path: resolved.path,
      query: resolved.query,
      body:
        resolved.body !== undefined ? JSON.stringify(resolved.body) : undefined,
      headers:
        resolved.body !== undefined
          ? { 'Content-Type': 'application/json' }
          : undefined,
    };
  } else {
    call = payload.raw ?? { path: '/' };
  }

  const {
    method: rawMethod = 'GET',
    path,
    query,
    body: rawBody,
    headers: rawHeaders,
  } = call;

  // Method allowlist — same as fetch.
  const method = ALLOWED_METHODS.has(rawMethod.toUpperCase())
    ? rawMethod.toUpperCase()
    : 'GET';

  // Dry-run "modo seguro": não executa escrita — devolve resposta simulada,
  // preservando o shape esperado (RPC → dados; cru → tipo-Response).
  // ARFIX-03: returns BEFORE the concurrency reservation below, so a dry-run
  // never reserves or decrements a slot.
  if (writeMode === 'safe' && WRITE_METHODS.has(method)) {
    const simulated = { __simulated: true, method, path };
    return isRpc
      ? simulated
      : {
          ok: true,
          status: 200,
          statusText: 'OK (simulado)',
          headers: {},
          body: JSON.stringify(simulated),
        };
  }

  // ARFIX-03: reserve-before-validate. Increment synchronously (no await before
  // this point since the method/dry-run checks above are sync) and roll back if
  // it pushes us over the cap. The try/finally below guarantees the decrement on
  // every exit path, including the origin-escape and path-invalid throws that
  // previously happened before the (later) increment.
  concurrencyCounter.inFlight += 1;
  if (concurrencyCounter.inFlight > FETCH_MAX_CONCURRENT) {
    concurrencyCounter.inFlight -= 1;
    throw new Error(
      'planfi.call bloqueado: limite de requisições simultâneas atingido',
    );
  }

  let controller: AbortController | undefined;
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    // Build the URL. Must stay on the same origin as baseUrl.
    let resolvedUrl: URL;
    try {
      resolvedUrl = new URL(path, baseUrl);
    } catch {
      throw new Error('planfi.call bloqueado: path inválido');
    }

    // Origin-escape guard: the resolved URL's host must match the base host.
    let baseHost: string;
    try {
      baseHost = new URL(baseUrl).hostname.toLowerCase();
    } catch {
      throw new Error('planfi.call bloqueado: baseUrl inválida');
    }
    if (resolvedUrl.hostname.toLowerCase() !== baseHost) {
      throw new Error(
        `planfi.call bloqueado: path tentou escapar do host configurado (${baseHost})`,
      );
    }

    // Append query string.
    if (query && typeof query === 'object') {
      for (const [k, v] of Object.entries(query)) {
        if (typeof k === 'string' && typeof v === 'string') {
          resolvedUrl.searchParams.append(k, v);
        }
      }
    }

    // Guard: protocolo + allowlist (= [baseHost]). O bloqueio de IP privado NÃO
    // se aplica aqui: planfi.call é travado same-origin na baseUrl que o
    // OPERADOR configurou (SCHEDULED_JOBS_PLANFI_BASE_URL) — só consegue alcançar
    // esse host, não é SSRF. Por isso allowPrivate:true (necessário p/ rodar
    // contra o core em localhost no dev; em prod a base é pública e isto é no-op).
    await guardUrl(resolvedUrl, [baseHost], { allowPrivate: true });

    // Headers: filter hop-by-hop / forwarding headers from user input.
    const safeHeaders: Record<string, string> = {};
    if (rawHeaders && typeof rawHeaders === 'object') {
      for (const [k, v] of Object.entries(rawHeaders)) {
        if (
          typeof k === 'string' &&
          typeof v === 'string' &&
          !BLOCKED_REQUEST_HEADERS.has(k.toLowerCase())
        ) {
          safeHeaders[k] = v;
        }
      }
    }
    // Hard-set the auth header LAST so user cannot override/strip it.
    safeHeaders['X-Planfi-Service-Key'] = planfiKey;

    // Body: only allow strings; cap at FETCH_BODY_CAP_BYTES.
    let body: string | undefined;
    if (typeof rawBody === 'string') {
      body =
        rawBody.length > FETCH_BODY_CAP_BYTES
          ? rawBody.slice(0, FETCH_BODY_CAP_BYTES)
          : rawBody;
    }

    const effectiveTimeout = planfiCallTimeoutMs(timeoutMs);
    controller = new AbortController();
    // ARFIX-02: register so run() teardown can abort this in-flight call.
    activeControllers.add(controller);
    timer = setTimeout(() => controller!.abort(), effectiveTimeout);

    // FIX-1: Manual redirect handling — re-validate each hop through guardUrl.
    let currentUrl = resolvedUrl;
    let hops = 0;

    let response: Response;
    while (true) {
      const fetchInit: RequestInit = {
        method,
        headers: safeHeaders,
        ...(body !== undefined ? { body } : {}),
        redirect: 'manual',
        signal: controller.signal,
      };

      try {
        response = await fetch(currentUrl.toString(), fetchInit);
      } catch (err) {
        if (err instanceof Error && err.name === 'AbortError') {
          throw new Error(
            `planfi.call bloqueado: timeout de ${effectiveTimeout}ms excedido`,
          );
        }
        throw new Error('planfi.call falhou: erro de rede');
      }

      if (
        response.status >= 300 &&
        response.status < 400 &&
        response.headers.has('location')
      ) {
        hops += 1;
        if (hops > REDIRECT_MAX_HOPS) {
          throw new Error(
            `planfi.call bloqueado: limite de redirecionamentos (${REDIRECT_MAX_HOPS}) atingido`,
          );
        }

        const location = response.headers.get('location') as string;
        let nextUrl: URL;
        try {
          nextUrl = new URL(location, currentUrl.toString());
        } catch {
          throw new Error(
            'planfi.call bloqueado: Location de redirecionamento inválida',
          );
        }

        // Each redirect hop is also constrained to the base host.
        if (nextUrl.hostname.toLowerCase() !== baseHost) {
          throw new Error(
            `planfi.call bloqueado: redirecionamento para host não autorizado (${nextUrl.hostname})`,
          );
        }

        await guardUrl(nextUrl, [baseHost]);
        currentUrl = nextUrl;
        continue;
      }

      break;
    }

    // FIX-4: stream body with byte cap.
    let bodyText: string;
    if (response.body) {
      const reader = response.body.getReader();
      const chunks: Uint8Array[] = [];
      let totalBytes = 0;
      let truncated = false;

      try {
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          if (value) {
            totalBytes += value.byteLength;
            if (totalBytes > FETCH_BODY_CAP_BYTES) {
              const remaining =
                FETCH_BODY_CAP_BYTES - (totalBytes - value.byteLength);
              if (remaining > 0) {
                chunks.push(value.slice(0, remaining));
              }
              truncated = true;
              await reader.cancel();
              break;
            }
            chunks.push(value);
          }
        }
      } catch {
        // Reader may throw if aborted; fine — we have what we need.
      }

      const combined = new Uint8Array(
        chunks.reduce((acc, c) => acc + c.byteLength, 0),
      );
      let offset = 0;
      for (const chunk of chunks) {
        combined.set(chunk, offset);
        offset += chunk.byteLength;
      }
      bodyText = new TextDecoder().decode(combined);
      if (truncated) bodyText += '…[truncado]';
    } else {
      bodyText = '';
    }

    const headers: Record<string, string> = {};
    response.headers.forEach((value, key) => {
      headers[key] = value;
    });

    // RPC mode: devolve o JSON parseado (com unwrap), lançando em erro HTTP —
    // ergonomia de "chamada de função". Modo cru mantém a resposta tipo-Response.
    if (isRpc) {
      let parsed: unknown = null;
      if (bodyText) {
        try {
          parsed = JSON.parse(bodyText);
        } catch {
          if (!response.ok) {
            throw new Error(`planfi.call falhou: HTTP ${response.status}`);
          }
          throw new Error('planfi.call: resposta não-JSON da API PlanFi');
        }
      }
      if (!response.ok) {
        const detail =
          parsed &&
          typeof parsed === 'object' &&
          typeof (parsed as { message?: unknown }).message === 'string'
            ? (parsed as { message: string }).message
            : `HTTP ${response.status}`;
        throw new Error(`planfi.call falhou: ${detail}`);
      }
      if (
        rpcUnwrap &&
        parsed &&
        typeof parsed === 'object' &&
        rpcUnwrap in (parsed as Record<string, unknown>)
      ) {
        return (parsed as Record<string, unknown>)[rpcUnwrap];
      }
      return parsed;
    }

    return {
      ok: response.ok,
      status: response.status,
      statusText: response.statusText,
      headers,
      body: bodyText,
    };
  } finally {
    if (timer !== undefined) clearTimeout(timer);
    // ARFIX-02: unregister this call's controller (no-op if never created).
    if (controller !== undefined) activeControllers.delete(controller);
    // ARFIX-03: release the reservation made before validation.
    concurrencyCounter.inFlight -= 1;
  }
}

@Injectable()
export class SandboxRunner {
  async run(input: SandboxInput): Promise<SandboxResult> {
    const start = Date.now();
    const timeoutMs = input.limits.timeoutMs;

    // Capture logs with size cap. Hoisted outside the try so that partial logs
    // produced before a failure/timeout are still returned (SBX-3).
    const logs: string[] = [];
    let logBytesUsed = 0;
    let logTruncated = false;

    // The Isolate is created inside the try so that an invalid memoryLimit
    // (e.g. < 8 MB) does not throw before the try and reject run(), which would
    // violate the contract of always resolving with a SandboxResult (SBX-4).
    // isolated-vm requires memoryLimit >= 8 MB.
    let isolate: ivm.Isolate | undefined;

    // FIX-3: per-run fetch counters shared across all hostFetch calls.
    const fetchCounter = { count: 0 };
    const concurrencyCounter = { inFlight: 0 };

    // ARFIX-02: registry of AbortControllers for fetches/planfi.calls in flight.
    // Shared with the host functions; aborted en masse on wall-clock dispose and
    // on the run()-level teardown so no host fetch keeps a socket/slot alive
    // after the run is considered done.
    const activeControllers = new Set<AbortController>();
    const abortActiveControllers = (): void => {
      for (const controller of activeControllers) {
        try {
          controller.abort();
        } catch {
          // abort() is idempotent/safe; never let one failure block the rest.
        }
      }
    };

    // FIX-3: wall-clock deadline. The ivm CPU timeout only bounds CPU ticks;
    // a job doing `while(true){ await fetch(...) }` burns no CPU between awaits
    // so it would run forever. A real-time deadline fires regardless.
    //
    // The wall-clock budget is timeoutMs * 3 clamped to WALL_CLOCK_CEILING_MS,
    // but never below the job's own timeoutMs (see computeWallClockMs). The
    // multiplier gives pure CPU-bound loops time for the ivm CPU timeout
    // (which fires after `timeoutMs` of actual CPU time) to win. For
    // I/O-looping jobs (e.g. while(true){await fetch(…)}) the ivm CPU quota is
    // never consumed, so the wall-clock timer is what terminates the run.
    const wallClockMs = computeWallClockMs(timeoutMs);
    let wallClockFired = false;
    let wallClockTimer: ReturnType<typeof setTimeout> | undefined;

    try {
      const memoryLimit = Math.max(8, Math.floor(input.limits.memoryMb || 128));
      isolate = new ivm.Isolate({ memoryLimit });

      // Capture a reference to the isolate for the wall-clock timer closure
      // before it might be disposed.
      const isolateRef = isolate;

      wallClockTimer = setTimeout(() => {
        wallClockFired = true;
        // ARFIX-02: cancel any in-flight host fetch BEFORE/AS we dispose the
        // isolate so it does not keep a socket + concurrency slot alive for up
        // to FETCH_TIMEOUT_CEILING_MS after the run is over. Additive and safe
        // even if the isolate is already gone.
        abortActiveControllers();
        if (isolateRef && !isolateRef.isDisposed) {
          isolateRef.dispose();
        }
      }, wallClockMs);

      const context = await isolate.createContext();
      const jail = context.global;

      // Set global self-reference so code can access global scope
      await jail.set('global', jail.derefInto());

      await jail.set(
        'log',
        new ivm.Callback((...args: unknown[]) => {
          if (logTruncated) return;
          const line = args.map(formatLogArg).join(' ');
          const lineBytes = Buffer.byteLength(line, 'utf8');
          if (logBytesUsed + lineBytes > LOG_CAP_BYTES) {
            logs.push('…[truncado]');
            logTruncated = true;
            return;
          }
          logs.push(line);
          logBytesUsed += lineBytes;
        }),
      );

      // Inject input as a deep copy into the isolate
      await jail.set(
        'input',
        new ivm.ExternalCopy(input.input ?? null).copyInto(),
      );

      // P2: read-only env. Copy the plain object into the isolate; it is frozen
      // in the bootstrap below. We never expose the host process.env — only this
      // job-scoped bag (defaults to {} when absent).
      await jail.set('env', new ivm.ExternalCopy(input.env ?? {}).copyInto());

      // P2: host fetch. Registered as ivm.Reference pointing to an async host
      // function. The bootstrap wraps it inside an IIFE: the wrapper function
      // is assigned to globalThis.fetch and the raw Reference is deleted from
      // the isolate global, so user code can only call the safe wrapper.
      //
      // FIX-5: __hostFetch (the raw ivm.Reference) is deleted by the bootstrap
      // IIFE immediately after the wrapper captures it. No raw host handle
      // remains reachable by user code after bootstrap runs.
      // Using ivm.Reference (not ivm.Callback) is correct for async cross-
      // boundary calls: .apply() with result:{promise:true,copy:true} marshals
      // the resolved value back into the isolate. ivm.Callback with async:true
      // returns the Promise object itself, which cannot be cloned.
      const allowlist = input.fetchAllowlist ?? [];
      const fetchEnabled = Array.isArray(input.fetchAllowlist);
      if (fetchEnabled) {
        const hostFetchRef = new ivm.Reference(
          (rawUrl: string, rawInit: string): Promise<SandboxFetchResponse> =>
            hostFetch(
              rawUrl,
              rawInit,
              allowlist,
              timeoutMs,
              fetchCounter,
              concurrencyCounter,
              activeControllers,
            ),
        );
        await jail.set('__hostFetch', hostFetchRef);
      }

      // planfi.call: injected ONLY when input.planfi?.baseUrl is set (mirrors the
      // fetch pattern — fail-closed by default). The host-side function is wrapped
      // the same way as __hostFetch: captured by the bootstrap IIFE, deleted from
      // the global, and exposed only as the safe `planfi.call` wrapper.
      const planfiEnabled = !!input.planfi?.baseUrl;
      if (planfiEnabled) {
        const planfiBaseUrl = input.planfi!.baseUrl;
        const hostPlanfiRef = new ivm.Reference(
          (rawInit: string): Promise<unknown> => {
            // Read PLANFI_KEY from the job's env at call time (not at setup time)
            // so key rotation is respected without restarting the job.
            // Fallback to the system-wide key if the job doesn't provide one.
            const planfiKey =
              (input.env ?? {})['PLANFI_KEY'] ||
              process.env.SCHEDULED_JOBS_INTERNAL_PLANFI_KEY ||
              '';
            return hostPlanfiCall(
              rawInit,
              planfiBaseUrl,
              planfiKey,
              timeoutMs,
              fetchCounter,
              concurrencyCounter,
              input.planfi!.writeMode,
              activeControllers,
            );
          },
        );
        await jail.set('__hostPlanfiCall', hostPlanfiRef);
      }

      // Bootstrap: freeze env (read-only) and, when enabled, define the async
      // fetch wrapper. The bootstrap captures __hostFetch via an IIFE, assigns
      // the safe wrapper to globalThis.fetch, and immediately deletes
      // __hostFetch so the raw Reference is unreachable by user code.
      //
      // FIX-5: after the IIFE runs, typeof globalThis.__hostFetch === 'undefined'.
      const bootstrap = `
        if (typeof globalThis.env !== 'undefined' && globalThis.env !== null) {
          Object.freeze(globalThis.env);
        }
        ${
          fetchEnabled
            ? `(function () {
                 const _hf = globalThis.__hostFetch;
                 delete globalThis.__hostFetch;
                 globalThis.fetch = function (url, init) {
                   return _hf.apply(
                     undefined,
                     [String(url), JSON.stringify(init || {})],
                     { result: { promise: true, copy: true } },
                   );
                 };
               })();`
            : ''
        }
        ${
          planfiEnabled
            ? `(function () {
                 const _hpc = globalThis.__hostPlanfiCall;
                 delete globalThis.__hostPlanfiCall;
                 globalThis.planfi = {
                   call: function (methodOrOpts, params) {
                     var payload =
                       typeof methodOrOpts === 'string'
                         ? { rpc: methodOrOpts, params: params || {} }
                         : { raw: methodOrOpts || {} };
                     return _hpc.apply(
                       undefined,
                       [JSON.stringify(payload)],
                       { result: { promise: true, copy: true } },
                     );
                   },
                 };
               })();`
            : ''
        }
      `;
      const bootstrapScript = await isolate.compileScript(bootstrap);
      await bootstrapScript.run(context);

      // Wrap user code to support top-level await and capture return value
      const wrappedCode = `(async () => { ${input.code} })()`;
      const script = await isolate.compileScript(wrappedCode);

      const rawOutput: unknown = await script.run(context, {
        timeout: timeoutMs,
        promise: true,
        copy: true,
      });

      const durationMs = Date.now() - start;

      return {
        status: 'SUCCESS',
        output: truncateJson(rawOutput),
        logs,
        durationMs,
      };
    } catch (err: unknown) {
      const durationMs = Date.now() - start;
      const message = sanitizeToString(
        err instanceof Error ? err.message : err,
      );

      // Detect timeout via the canonical isolated-vm message or the wall-clock
      // timer firing. ARFIX-01: do NOT classify by wall-clock duration
      // (durationMs >= timeoutMs) — durationMs is wall-clock, not CPU, so a run
      // that does legitimate slow async work past the CPU budget and then throws
      // a real error would be misreported as TIMED_OUT with its stack discarded.
      // The two signals below already cover every genuine timeout: a CPU-bound
      // loop trips IVM_TIMEOUT_MESSAGE; an I/O-looping job trips wallClockFired.
      const isTimeout = wallClockFired || IVM_TIMEOUT_MESSAGE.test(message);

      if (isTimeout) {
        return {
          status: 'TIMED_OUT',
          // Return any partial logs produced before the timeout (SBX-3).
          logs,
          error: {
            message: wallClockFired
              ? 'Execução encerrada: timeout de parede atingido'
              : message,
          },
          durationMs,
        };
      }

      const stack =
        err instanceof Error && err.stack
          ? sanitizeToString(err.stack)
          : undefined;

      return {
        status: 'FAILED',
        // Return any partial logs produced before the failure (SBX-3).
        logs,
        error: { message, stack },
        durationMs,
      };
    } finally {
      if (wallClockTimer !== undefined) {
        clearTimeout(wallClockTimer);
      }
      // ARFIX-02: cancel any host fetch still in flight on normal teardown so it
      // does not keep a socket/concurrency slot alive after run() settles.
      abortActiveControllers();
      if (isolate && !isolate.isDisposed) {
        isolate.dispose();
      }
    }
  }
}