import { randomUUID } from "node:crypto";
import { existsSync, mkdirSync, writeFileSync } from "node:fs";
import { resolve } from "node:path";
import { fileURLToPath } from "node:url";

import { afterAll, beforeAll, describe, expect, it } from "@effect/vitest";
import { Schema } from "effect";
import { unstable_dev, type Unstable_DevWorker } from "wrangler";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { ElicitRequestSchema } from "@modelcontextprotocol/sdk/types.js";
import { microsoftCatalog } from "@executor-js/plugin-openapi/providers/microsoft";

// ---------------------------------------------------------------------------
// End-to-end test for the Cloudflare host: boots the REAL worker on workerd via
// Miniflare (wrangler `unstable_dev`) with a local D1 + R2, dev-auth on. This is
// the only test that exercises the CF-specific stack together — D1 schema
// bring-up, the R2-backed blob seam (multi-MB spec storage), QuickJS-WASM
// execution, and the MCP envelope — through the actual HTTP surface.
// ---------------------------------------------------------------------------

const dir = fileURLToPath(new URL(".", import.meta.url));
const runId = randomUUID().slice(0, 8);

const ensureStaticAssets = () => {
  // CI runs from a fresh checkout with no `vite build`, so `./dist` (the SPA
  // assets dir wrangler.jsonc points `assets.directory` at) is absent and
  // `unstable_dev`'s assets validation aborts boot. These tests drive the
  // API/MCP surface (all `run_worker_first` paths), not the SPA.
  const distIndex = resolve(dir, "../dist/index.html");
  if (!existsSync(distIndex)) {
    mkdirSync(resolve(dir, "../dist"), { recursive: true });
    writeFileSync(distIndex, "<!doctype html><title>executor</title>");
  }
};

// Inline spec (no network); registers one tool, exercising the D1 write path.
const SPEC = JSON.stringify({
  openapi: "3.0.0",
  info: { title: "Test", version: "1.0.0" },
  servers: [{ url: "https://example.com" }],
  paths: {
    "/ping": {
      get: { operationId: "ping", responses: { "200": { description: "ok" } } },
    },
  },
});

const decodeUnknownJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));

interface JsonReadableResponse {
  readonly headers: { get(name: string): string | null };
  json(): Promise<unknown>;
  text(): Promise<string>;
}

const readMcpJson = async <A>(response: JsonReadableResponse): Promise<A> => {
  const contentType = response.headers.get("content-type") ?? "";
  if (contentType.includes("application/json")) {
    return (await response.json()) as A;
  }
  // Parse the SSE stream properly instead of grabbing the first `data:` line:
  // the server may legally interleave other events (e.g. the priming event a
  // tools/call stream emits so clients can reconnect) before the JSON-RPC
  // message. Only `message`-typed events (the SSE default) carry JSON-RPC.
  const text = await response.text();
  let responseMessage: unknown;
  for (const block of text.split("\n\n")) {
    let eventType = "message";
    let data = "";
    for (const line of block.replace(/\r/g, "").split("\n")) {
      if (line.startsWith("event:")) eventType = line.slice("event:".length).trim();
      if (line.startsWith("data:")) data += line.slice("data:".length).trimStart();
    }
    if (eventType !== "message" || data.length === 0) continue;
    const parsed = decodeUnknownJson(data);
    // Skip protocol-legal notifications; the tests want the response.
    if (
      typeof parsed === "object" &&
      parsed !== null &&
      ("result" in parsed || "error" in parsed)
    ) {
      responseMessage = parsed;
      break;
    }
  }
  expect(responseMessage, "a JSON-RPC response arrives on the SSE stream").toBeTruthy();
  return responseMessage as A;
};

describe("cloudflare host e2e (workerd/miniflare)", () => {
  let worker: Unstable_DevWorker;

  beforeAll(async () => {
    ensureStaticAssets();

    worker = await unstable_dev(resolve(dir, "worker.ts"), {
      config: resolve(dir, "../wrangler.jsonc"),
      ip: "127.0.0.1",
      local: true,
      persist: false,
      experimental: { disableExperimentalWarning: true },
      vars: {
        EXECUTOR_SECRET_KEY: "test-secret-key-0123456789abcdef",
        ENABLE_DEV_AUTH: "true",
      },
    });
  }, 120_000);

  afterAll(async () => {
    await worker?.stop();
  });

  it("executes TypeScript via /api/executions (QuickJS on workerd)", async () => {
    const res = await worker.fetch("/api/executions", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ code: "export default 6 * 7" }),
    });
    expect(res.status).toBe(200);
    const body = (await res.json()) as {
      status: string;
      text: string;
      isError: boolean;
    };
    expect(body.status).toBe("completed");
    expect(body.isError).toBe(false);
    expect(body.text).toBe("42");
  }, 60_000);

  it("adds a LARGE OpenAPI source — exercises the R2 blob seam (~1MB spec) + createMany batching (>100 tools)", async () => {
    // Synthesize a spec big enough to (a) far exceed D1's per-value cap if it
    // were inlined — proving the spec text really lands in the R2 blob seam —
    // and (b) derive >100 tools (past D1's 100 bound-param createMany limit).
    const paths: Record<string, unknown> = {};
    for (let i = 0; i < 250; i++) {
      paths[`/op${i}`] = {
        get: {
          operationId: `op${i}`,
          summary: `operation ${i}`,
          description: "d".repeat(4000), // padding -> ~1MB total spec
          responses: { "200": { description: "ok" } },
        },
      };
    }
    const largeSpec = JSON.stringify({
      openapi: "3.0.0",
      info: { title: "Large", version: "1.0.0" },
      servers: [{ url: "https://example.com" }],
      paths,
    });
    expect(largeSpec.length).toBeGreaterThan(900_000);

    const slug = `largeapi-${runId}`;
    const add = await worker.fetch("/api/openapi/specs", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({
        spec: { kind: "blob", value: largeSpec },
        slug,
        description: "Large API",
        baseUrl: "https://example.com",
      }),
    });
    expect(add.status).toBe(200);
    const added = (await add.json()) as { toolCount: number };
    expect(added.toolCount).toBe(250);

    // Reads back the catalog row whose config points at the R2 spec blob.
    const got = await worker.fetch(`/api/openapi/integrations/${slug}`);
    expect(got.status).toBe(200);
    const integration = (await got.json()) as { slug: string } | null;
    expect(integration?.slug).toBe(slug);
  }, 90_000);

  it("adds an OpenAPI source and reads it back (D1 write + read path)", async () => {
    const slug = `testapi-${runId}`;
    const add = await worker.fetch("/api/openapi/specs", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({
        spec: { kind: "blob", value: SPEC },
        slug,
        description: "Test API",
        baseUrl: "https://example.com",
      }),
    });
    expect(add.status).toBe(200);
    const added = (await add.json()) as { toolCount: number; slug: string };
    expect(added.toolCount).toBeGreaterThan(0);

    const got = await worker.fetch(`/api/openapi/integrations/${slug}`);
    expect(got.status).toBe(200);
    const integration = (await got.json()) as { slug: string } | null;
    expect(integration?.slug).toBe(slug);
  }, 60_000);

  it("exposes Microsoft catalog presets through the OpenAPI integration catalog", async () => {
    const microsoftFilesPreset = microsoftCatalog.find(
      (preset) => preset.defaultSlug === "microsoft_files",
    );
    expect(microsoftFilesPreset).toBeTruthy();

    const removedProviderRoute = await worker.fetch("/api/microsoft/graph", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({}),
    });
    expect(removedProviderRoute.status).toBe(404);

    const slug = `microsoft-files-${runId}`;
    const add = await worker.fetch("/api/openapi/specs", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({
        spec: { kind: "blob", value: SPEC },
        slug,
        name: microsoftFilesPreset?.name,
        description: microsoftFilesPreset?.summary,
        family: microsoftFilesPreset?.family,
        authenticationTemplate: microsoftFilesPreset?.authTemplate,
      }),
    });
    expect(add.status).toBe(200);

    const res = await worker.fetch(`/api/integrations/${slug}`);
    expect(res.status).toBe(200);
    const integration = (await res.json()) as {
      readonly slug: string;
      readonly kind: string;
      readonly family?: string;
      readonly authMethods: readonly { readonly template: string; readonly kind: string }[];
    };
    expect(integration.slug).toBe(slug);
    expect(integration.kind).toBe("openapi");
    expect(integration.family).toBe("microsoft");
    expect(integration.authMethods.some((method) => method.template === "azureAdDelegated")).toBe(
      true,
    );
  });

  it("gates the API when dev-auth is on but treats the request as the dev admin", async () => {
    // dev-auth means the request is the fixed dev admin; a gated route resolves
    // to the principal. There is no scope stack in v2 — account/me is the
    // identity-backed read that the API gate protects.
    const res = await worker.fetch("/api/account/me");
    expect(res.status).toBe(200);
    const me = (await res.json()) as {
      user: { id: string };
      organization: { id: string };
    };
    expect(me.user.id).toBe("dev");
  });

  it("lists tools on a follow-up request after a fresh initialize (DO session survives across requests)", async () => {
    // The production regression: `initialize` creates the session, then a
    // SEPARATE `tools/list` request must find it. With the old in-process store a
    // second Worker isolate never saw the session and this returned "Not
    // connected"; the MCP-session Durable Object (id == session id) routes the
    // follow-up back to the same isolate, so the tool list comes through.
    const accept = "application/json, text/event-stream";
    const rpc = (sessionId: string | null, body: unknown) =>
      worker.fetch("/mcp", {
        method: "POST",
        headers: {
          "content-type": "application/json",
          accept,
          ...(sessionId ? { "mcp-session-id": sessionId } : {}),
        },
        body: JSON.stringify(body),
      });

    const init = await rpc(null, {
      jsonrpc: "2.0",
      id: 1,
      method: "initialize",
      params: {
        protocolVersion: "2025-03-26",
        capabilities: {},
        clientInfo: { name: "test", version: "1" },
      },
    });
    expect(init.status).toBe(200);
    const sessionId = init.headers.get("mcp-session-id");
    expect(sessionId).toBeTruthy();

    await rpc(sessionId, {
      jsonrpc: "2.0",
      method: "notifications/initialized",
    });

    const list = await rpc(sessionId, {
      jsonrpc: "2.0",
      id: 2,
      method: "tools/list",
    });
    expect(list.status).toBe(200);
    const listed = await readMcpJson<{
      result?: { tools?: ReadonlyArray<{ name: string }> };
    }>(list);
    const toolNames = listed.result?.tools?.map((t) => t.name) ?? [];
    expect(toolNames).toContain("execute");
  }, 60_000);

  it("serves toolkit MCP sessions and rejects cross-resource session reuse", async () => {
    const createToolkit = await worker.fetch("/api/toolkits", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({
        owner: "org",
        name: `Cloudflare Toolkit ${runId}`,
        slug: `cloudflare-toolkit-${runId}`,
      }),
    });
    expect(createToolkit.status).toBe(200);
    const toolkit = (await createToolkit.json()) as { id: string; slug: string };

    const addConnection = await worker.fetch(`/api/toolkits/${toolkit.id}/connections`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ pattern: "executor.*" }),
    });
    expect(addConnection.status).toBe(200);

    const accept = "application/json, text/event-stream";
    const toolkitPath = `/mcp/toolkits/${toolkit.slug}`;
    const rpc = (path: string, sessionId: string | null, body: unknown) =>
      worker.fetch(path, {
        method: "POST",
        headers: {
          "content-type": "application/json",
          accept,
          ...(sessionId ? { "mcp-session-id": sessionId } : {}),
        },
        body: JSON.stringify(body),
      });

    const init = await rpc(toolkitPath, null, {
      jsonrpc: "2.0",
      id: 1,
      method: "initialize",
      params: {
        protocolVersion: "2025-03-26",
        capabilities: {},
        clientInfo: { name: "toolkit-route-test", version: "1" },
      },
    });
    expect(init.status).toBe(200);
    const sessionId = init.headers.get("mcp-session-id");
    expect(sessionId).toBeTruthy();

    await rpc(toolkitPath, sessionId, {
      jsonrpc: "2.0",
      method: "notifications/initialized",
    });

    const list = await rpc(toolkitPath, sessionId, {
      jsonrpc: "2.0",
      id: 2,
      method: "tools/list",
    });
    expect(list.status).toBe(200);
    const listed = await readMcpJson<{
      result?: { tools?: ReadonlyArray<{ name: string }> };
    }>(list);
    expect(listed.result?.tools?.map((tool) => tool.name)).toContain("execute");

    const reusedOnDefault = await rpc("/mcp", sessionId, {
      jsonrpc: "2.0",
      id: 3,
      method: "tools/list",
    });
    expect(reusedOnDefault.status).toBe(403);

    const reusedOnOtherToolkit = await rpc(`${toolkitPath}-other`, sessionId, {
      jsonrpc: "2.0",
      id: 4,
      method: "tools/list",
    });
    expect(reusedOnOtherToolkit.status).toBe(403);
    for (const method of ["GET", "DELETE"]) {
      const response = await worker.fetch("/mcp", {
        method,
        headers: { accept, "mcp-session-id": sessionId! },
      });
      expect(response.status).toBe(403);
    }
    const stillUsable = await rpc(toolkitPath, sessionId, {
      jsonrpc: "2.0",
      id: 5,
      method: "tools/list",
    });
    expect(stillUsable.status).toBe(200);
    const deleted = await worker.fetch(toolkitPath, {
      method: "DELETE",
      headers: { accept, "mcp-session-id": sessionId! },
    });
    expect(deleted.status).toBe(204);
  }, 60_000);

  it("serves streamable HTTP GET only for initialized sessions", async () => {
    const missing = await worker.fetch("/mcp", {
      method: "GET",
      headers: { accept: "text/event-stream" },
    });
    expect(missing.status).toBe(400);
    await missing.text();

    const init = await worker.fetch("/mcp", {
      method: "POST",
      headers: {
        "content-type": "application/json",
        accept: "application/json, text/event-stream",
      },
      body: JSON.stringify({
        jsonrpc: "2.0",
        id: 1,
        method: "initialize",
        params: {
          protocolVersion: "2025-03-26",
          capabilities: {},
          clientInfo: { name: "test", version: "1" },
        },
      }),
    });
    expect(init.status).toBe(200);
    const sessionId = init.headers.get("mcp-session-id");
    expect(sessionId).toBeTruthy();
    await init.text();

    const stream = await worker.fetch("/mcp", {
      method: "GET",
      headers: {
        accept: "text/event-stream",
        "mcp-session-id": sessionId!,
      },
    });
    expect(stream.status).toBe(200);
    expect(stream.headers.get("content-type")).toContain("text/event-stream");
    await stream.body?.cancel();
  }, 60_000);

  it("invokes the execute tool over MCP (initialize → tools/call → QuickJS)", async () => {
    const accept = "application/json, text/event-stream";
    const rpc = (sessionId: string | null, body: unknown) =>
      worker.fetch("/mcp", {
        method: "POST",
        headers: {
          "content-type": "application/json",
          accept,
          ...(sessionId ? { "mcp-session-id": sessionId } : {}),
        },
        body: JSON.stringify(body),
      });

    const init = await rpc(null, {
      jsonrpc: "2.0",
      id: 1,
      method: "initialize",
      params: {
        protocolVersion: "2025-03-26",
        capabilities: {},
        clientInfo: { name: "test", version: "1" },
      },
    });
    expect(init.status).toBe(200);
    const sessionId = init.headers.get("mcp-session-id");
    expect(sessionId).toBeTruthy();

    await rpc(sessionId, {
      jsonrpc: "2.0",
      method: "notifications/initialized",
    });

    const call = await rpc(sessionId, {
      jsonrpc: "2.0",
      id: 2,
      method: "tools/call",
      params: { name: "execute", arguments: { code: "export default 6 * 7" } },
    });
    expect(call.status).toBe(200);
    const result = await readMcpJson<{
      result?: { structuredContent?: { result?: number } };
    }>(call);
    expect(result.result?.structuredContent?.result).toBe(42);
  }, 60_000);

  it("delivers native elicitation on the approval-gated tool call stream", async () => {
    const client = new Client(
      { name: "native-elicitation-test", version: "1.0.0" },
      { capabilities: { elicitation: { form: {}, url: {} } } },
    );
    let receivedElicitation = false;
    client.setRequestHandler(ElicitRequestSchema, async () => {
      receivedElicitation = true;
      return { action: "accept" as const, content: {} };
    });
    const transport = new StreamableHTTPClientTransport(
      new URL("/mcp?elicitation_mode=native", `http://${worker.address}:${worker.port}`),
    );
    await client.connect(transport);

    const result = await client.callTool(
      {
        name: "execute",
        arguments: {
          code: [
            "return await tools.executor.coreTools.policies.create({",
            '  owner: "org",',
            `  pattern: "native-elicitation-${runId}.*",`,
            '  action: "require_approval"',
            "});",
          ].join("\n"),
        },
      },
      undefined,
      { timeout: 5_000 },
    );

    expect(receivedElicitation).toBe(true);
    expect(result.isError).toBeFalsy();
    await client.close();
  }, 60_000);

  it("resumes a model-paused execution from a fresh MCP session", async () => {
    const accept = "application/json, text/event-stream";
    const rpc = (sessionId: string | null, body: unknown) =>
      worker.fetch("/mcp?elicitation_mode=model", {
        method: "POST",
        headers: {
          "content-type": "application/json",
          accept,
          ...(sessionId ? { "mcp-session-id": sessionId } : {}),
        },
        body: JSON.stringify(body),
      });
    const initialize = async (id: number): Promise<string> => {
      const init = await rpc(null, {
        jsonrpc: "2.0",
        id,
        method: "initialize",
        params: {
          protocolVersion: "2025-03-26",
          capabilities: {},
          clientInfo: { name: "test", version: "1" },
        },
      });
      expect(init.status).toBe(200);
      const sessionId = init.headers.get("mcp-session-id");
      expect(sessionId).toBeTruthy();
      await rpc(sessionId, {
        jsonrpc: "2.0",
        method: "notifications/initialized",
      });
      return sessionId!;
    };

    const sessionA = await initialize(1);
    const execute = await rpc(sessionA, {
      jsonrpc: "2.0",
      id: 2,
      method: "tools/call",
      params: {
        name: "execute",
        arguments: {
          code: [
            "return await tools.executor.coreTools.policies.create({",
            '  owner: "org",',
            '  pattern: "microsoft.*",',
            '  action: "require_approval"',
            "});",
          ].join("\n"),
        },
      },
    });
    expect(execute.status).toBe(200);
    const paused = await readMcpJson<{
      result?: {
        structuredContent?: {
          status?: string;
          executionId?: string;
        };
      };
    }>(execute);
    expect(paused.result?.structuredContent?.status).toBe("waiting_for_interaction");
    const executionId = paused.result?.structuredContent?.executionId;
    expect(executionId).toBeTruthy();

    const sessionB = await initialize(3);
    expect(sessionB).not.toBe(sessionA);
    const resume = await rpc(sessionB, {
      jsonrpc: "2.0",
      id: 4,
      method: "tools/call",
      params: {
        name: "resume",
        arguments: { executionId, action: "accept", content: "{}" },
      },
    });
    expect(resume.status).toBe(200);
    const resumed = await readMcpJson<{
      result?: {
        isError?: boolean;
        structuredContent?: {
          status?: string;
          recovery?: string;
          result?: unknown;
        };
      };
    }>(resume);
    expect(resumed.result?.structuredContent?.status).not.toBe("execution_not_found");
    expect(resumed.result?.structuredContent?.recovery, JSON.stringify(resumed.result)).not.toBe(
      "re_execute",
    );
    expect(resumed.result?.isError).toBeFalsy();
    expect(resumed.result?.structuredContent?.status).toBe("completed");
  }, 60_000);
});

describe("cloudflare host configuration errors", () => {
  let worker: Unstable_DevWorker;

  beforeAll(async () => {
    ensureStaticAssets();
    worker = await unstable_dev(resolve(dir, "worker.ts"), {
      config: resolve(dir, "../wrangler.jsonc"),
      ip: "127.0.0.1",
      local: true,
      persist: false,
      experimental: { disableExperimentalWarning: true },
      vars: {
        EXECUTOR_SECRET_KEY: "test-secret-key-0123456789abcdef",
      },
    });
  }, 120_000);

  afterAll(async () => {
    await worker?.stop();
  });

  it("returns an actionable response when Cloudflare Access is not configured", async () => {
    for (const path of ["/api/account/me", "/mcp"]) {
      const response = await worker.fetch(path);

      expect(response.status).toBe(503);
      expect(response.headers.get("cache-control")).toBe("no-store");
      await expect(response.text()).resolves.toBe(
        "Cloudflare Access is not configured. Set ACCESS_TEAM_DOMAIN and ACCESS_AUD before serving requests.\n",
      );
    }
  });
});