/**
 * FU-053 Layer 1 (T001) — streaming-route carve-out verification probe
 * (architect-approved Shape 3, 2026-05-29).
 *
 * The companion probe `fu053-t001-verify.ts` substantiates the request-tx GUC
 * writer for ordinary routes. THIS probe substantiates the four carved-out
 * streaming routes that run OUTSIDE the request transaction inside
 * `streamingBypassContext`, doing all DB work in bounded `withTenantDbPhase`
 * phases. Run with:
 *
 *   npx tsx scripts/fu053-t001-streaming-verify.ts
 *
 * Coverage:
 *   1. Carve-out routing — the 4 enumerated routes enter bypass context; an
 *      excluded sibling (/executive-summary/view) does NOT.
 *   2. Fail-loud guard — raw `db` inside a bypass handler (no bounded phase)
 *      throws the FU-053 guard error instead of silently using the pool.
 *   3. Bounded-phase positive control — inside `withTenantDbPhase` the tenant
 *      GUC and `current_tenant_id()` resolve to the request tenant (RLS-ready).
 *   4. Concurrent isolation — many concurrent bypass requests for different
 *      tenants never bleed context across each other in their bounded phases.
 *   5. Pool non-starvation — N (> pool max) concurrent bypass "streams" hold
 *      NO pool client during the stream window, so an unrelated pooled query
 *      issued mid-stream completes promptly (the Blocker-1 regression guard).
 *   6. No GUC leak — after a bounded phase commits and releases, a subsequent
 *      pooled query with no context sees an empty tenant GUC.
 */
import { EventEmitter } from "node:events";
import { sql } from "drizzle-orm";
import { db, auditDb, pool } from "../server/db";
import { tenantMiddleware } from "../server/lib/tenant-middleware";
import { withTenantDbPhase } from "../server/lib/tenant-context";

let pass = 0;
let fail = 0;
const log = (label: string, ok: boolean, detail = "") => {
  console.log(`${ok ? "PASS" : "FAIL"} — ${label}${detail ? ": " + detail : ""}`);
  ok ? pass++ : fail++;
};

/** Minimal mock Express response built on EventEmitter (finish/close). */
class MockRes extends EventEmitter {
  statusCode = 200;
  headersSent = false;
  body: any = undefined;
  status(code: number) {
    this.statusCode = code;
    return this;
  }
  json(obj: any) {
    this.body = obj;
    this.headersSent = true;
    return this;
  }
}

/**
 * Drive one request through `tenantMiddleware` against a STREAMING route.
 * `method`+`fullPath` decide whether the carve-out matches. The middleware is
 * mounted at `/api`, so we split into baseUrl `/api` + the sub-path exactly as
 * Express would. The bypass branch runs `next()` synchronously inside
 * `streamingBypassContext`, so the handler promise is captured there and the
 * bypass context propagates to its async continuations.
 */
function driveStreaming<T>(
  method: string,
  fullPath: string,
  tenantId: string | undefined,
  handler: (req: any) => Promise<T>,
): Promise<{ result?: T; threw?: Error; res: MockRes }> {
  const req: any = {
    method,
    baseUrl: "/api",
    path: fullPath.replace(/^\/api/, ""),
    headers: tenantId ? { "x-tenant-id": tenantId } : {},
    session: undefined,
  };
  const res = new MockRes();
  let hp: Promise<T> | undefined;
  const next = () => {
    hp = handler(req);
  };
  return tenantMiddleware(req, res as any, next as any).then(async () => {
    if (!hp) return { res };
    try {
      return { result: await hp, res };
    } catch (e: any) {
      return { threw: e instanceof Error ? e : new Error(String(e)), res };
    }
  });
}

/**
 * Drive a request-tx (NON-bypass) route — used only for the excluded-sibling
 * contrast. Settles the request tx by emitting `finish` once the handler runs.
 */
function driveRequestTx<T>(
  method: string,
  fullPath: string,
  tenantId: string,
  handler: () => Promise<T>,
): Promise<{ result?: T; threw?: Error }> {
  const req: any = {
    method,
    baseUrl: "/api",
    path: fullPath.replace(/^\/api/, ""),
    headers: { "x-tenant-id": tenantId },
    session: undefined,
  };
  const res = new MockRes();
  let result: T | undefined;
  let threw: Error | undefined;
  const next = () => {
    void (async () => {
      try {
        result = await handler();
      } catch (e: any) {
        threw = e instanceof Error ? e : new Error(String(e));
      } finally {
        res.statusCode = 200;
        res.emit("finish");
      }
    })();
  };
  return tenantMiddleware(req, res as any, next as any).then(() => ({ result, threw }));
}

/** Read the tenant GUC through the active runner (proxy). */
async function readGuc(): Promise<string> {
  const r: any = await db.execute(
    sql`SELECT current_setting('app.current_tenant_id', true) AS tid`,
  );
  return r.rows?.[0]?.tid ?? "";
}

const BYPASS_ROUTES: Array<[string, string]> = [
  ["POST", "/api/conversations/42/messages"],
  ["GET", "/api/compliance/governance-pack/download"],
  ["GET", "/api/compliance/executive-summary/download"],
  ["GET", "/api/regulator/pilot-pack.zip"],
];

(async () => {
  // ── Test 1: carve-out routing ────────────────────────────────────────────
  console.log("\n=== Test 1: carve-out routing — 4 enumerated routes bypass ===");
  for (const [method, path] of BYPASS_ROUTES) {
    // Inside a bypass handler raw `db` MUST throw; that throw is our proof the
    // route entered streamingBypassContext (and not the request-tx path).
    const { threw } = await driveStreaming(method, path, "tenant_route", async () => {
      await db.execute(sql`SELECT 1`);
      return "no-throw";
    });
    log(`${method} ${path} → bypass context active`,
      !!threw && /FU-053 T001: raw `db`/.test(threw.message),
      threw ? "guard threw as expected" : "did NOT enter bypass");
  }

  // ── Test 1b: excluded sibling does NOT bypass ────────────────────────────
  console.log("\n=== Test 1b: excluded sibling /executive-summary/view ===");
  {
    const { result, threw } = await driveRequestTx(
      "GET",
      "/api/compliance/executive-summary/view",
      "tenant_view",
      async () => readGuc(),
    );
    log("/executive-summary/view runs in request tx (raw db works, GUC set)",
      !threw && result === "tenant_view",
      threw ? `threw: ${threw.message}` : `guc='${result}'`);
  }

  // ── Test 2: fail-loud guard ──────────────────────────────────────────────
  console.log("\n=== Test 2: fail-loud — raw db in bypass handler throws ===");
  {
    const { threw } = await driveStreaming(
      "GET",
      "/api/regulator/pilot-pack.zip",
      "tenant_guard",
      async () => {
        await db.select().from(sql`(SELECT 1) AS t` as any);
        return "reached";
      },
    );
    log("raw db.select inside bypass handler throws the FU-053 guard",
      !!threw && /withTenantDbPhase/.test(threw.message),
      threw ? "threw" : "did NOT throw");
  }

  // ── Test 3: bounded-phase positive control + RLS-readiness ───────────────
  console.log("\n=== Test 3: withTenantDbPhase positive + current_tenant_id() ===");
  {
    const { result, threw } = await driveStreaming(
      "GET",
      "/api/regulator/pilot-pack.zip",
      "tenant_phase",
      async (req) =>
        withTenantDbPhase(req, async () => {
          const guc = await readGuc();
          let fn = "";
          try {
            const r: any = await db.execute(sql`SELECT current_tenant_id() AS t`);
            fn = r.rows?.[0]?.t ?? "";
          } catch (e: any) {
            fn = `__ERR__:${e?.message || e}`;
          }
          return { guc, fn };
        }),
    );
    log("bounded phase sees the request tenant via GUC", !threw && result?.guc === "tenant_phase",
      threw ? `threw: ${threw.message}` : `guc='${result?.guc}'`);
    log("current_tenant_id() returns the request tenant (RLS-ready for T003)",
      result?.fn === "tenant_phase",
      `current_tenant_id()='${result?.fn}'`);
  }

  // ── Test 4: concurrent isolation across bypass phases ────────────────────
  console.log("\n=== Test 4: concurrent bypass isolation ===");
  {
    const seen: Record<string, string> = {};
    const mk = (t: string) =>
      driveStreaming("GET", "/api/regulator/pilot-pack.zip", t, async (req) =>
        withTenantDbPhase(req, async () => {
          await new Promise((r) => setTimeout(r, 20));
          const g = await readGuc();
          seen[t] = g;
          return g;
        }),
      );
    await Promise.all([
      mk("ten_a"), mk("ten_b"), mk("ten_c"), mk("ten_d"), mk("ten_e"),
    ]);
    const ok = Object.entries(seen).every(([t, g]) => t === g) &&
      Object.keys(seen).length === 5;
    log("each concurrent bypass phase sees only its own tenant", ok, JSON.stringify(seen));
  }

  // ── Test 5: pool non-starvation (Blocker-1 regression guard) ─────────────
  console.log("\n=== Test 5: pool non-starvation under concurrent streams ===");
  {
    const poolMax = (pool as any).options?.max ?? 10;
    const concurrency = poolMax + 5; // exceed the pool deliberately
    const STREAM_MS = 1500;

    let inStream = 0;
    const allInStream = new Promise<void>((resolve) => {
      const check = setInterval(() => {
        if (inStream >= concurrency) {
          clearInterval(check);
          resolve();
        }
      }, 5);
    });

    const streams = Array.from({ length: concurrency }, (_, i) =>
      driveStreaming("GET", "/api/regulator/pilot-pack.zip", `stream_${i}`, async (req) => {
        // Phase A: short bounded DB phase (holds a client briefly, then releases).
        const a = await withTenantDbPhase(req, async () => readGuc());
        // Stream window: hold the response open, holding NO pool client.
        inStream++;
        await new Promise((r) => setTimeout(r, STREAM_MS));
        // Phase B: short bounded DB phase again.
        const b = await withTenantDbPhase(req, async () => readGuc());
        return { a, b };
      }),
    );

    // Wait until every stream is in its (client-free) stream window, then fire
    // an unrelated pooled query and time it. If streams pinned clients across
    // the stream (the old in-stream-query behavior), concurrency > poolMax would
    // make this query block ~STREAM_MS. Shape 3 holds no client → it's prompt.
    await allInStream;
    const t0 = Date.now();
    await auditDb.execute(sql`SELECT 1`);
    const latency = Date.now() - t0;

    const results = await Promise.all(streams);
    const isolated = results.every((r, i) =>
      r.result?.a === `stream_${i}` && r.result?.b === `stream_${i}`);

    log(`mid-stream pooled query is prompt with ${concurrency} streams (pool max ${poolMax})`,
      latency < 600, `latency=${latency}ms`);
    log("every stream's bounded phases saw its own tenant", isolated);
  }

  // ── Test 6: no GUC leak after phase release ──────────────────────────────
  console.log("\n=== Test 6: no GUC leak after bounded phase ===");
  {
    await driveStreaming("GET", "/api/regulator/pilot-pack.zip", "tenant_leak", async (req) =>
      withTenantDbPhase(req, async () => readGuc()),
    );
    const r: any = await auditDb.execute(
      sql`SELECT current_setting('app.current_tenant_id', true) AS tid`,
    );
    const tid = r.rows?.[0]?.tid ?? "";
    log("pooled query after phase sees empty GUC (no leak)", tid === "" || tid == null,
      `got '${tid}'`);
  }

  console.log(`\n──────── FU-053 T001 streaming probe: ${pass} passed, ${fail} failed ────────`);
  process.exit(fail === 0 ? 0 : 1);
})().catch((err) => {
  console.error("PROBE CRASHED:", err);
  process.exit(2);
});