import * as OtelTracer from "@effect/opentelemetry/Tracer";
import { Effect, Layer, Predicate } from "effect";

import {
  McpAuthProvider,
  jsonRpcErrorBody,
  mcpModernDisabledResponse,
  defaultMcpResource,
  UNAVAILABLE_RETRY_AFTER_SECONDS,
  type AuthOutcome,
  type McpModernServerBuilder,
  type McpResource,
} from "@executor-js/host-mcp";
import { requestBodyFromRequest } from "@executor-js/host-mcp/tool-server";
import {
  currentPropagationHeaders,
  readArtifactsEnabled,
  readElicitationMode,
  withMcpResponseHeaders,
  withPropagationHeaders,
  withVerifiedIdentityHeaders,
} from "@executor-js/cloudflare/mcp/do-headers";
import type { McpSessionProps } from "@executor-js/cloudflare/mcp/agent-durable-object";
import {
  classifyMcpProtocolEra,
  makeMcpModernRequestRouter,
  mcpCorsPreflightResponse,
  requireMcpRequestStateKey,
} from "@executor-js/cloudflare/mcp/modern-request-router";
import { mcpExecutionOwnerDirectoryFromNamespace } from "@executor-js/cloudflare/mcp/execution-owner-directory";
import { createMcpSessionStub, mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub";

import { wrapMcpSseResponse } from "../observability/memory-metrics";
import { WorkerTelemetryLive } from "../observability/telemetry";
import { cloudMcpAuth } from "./auth-provider";
import { parseTraceparent } from "./traceparent";

const DEAD_SESSION_CACHE_TTL_MS = 5 * 60 * 1_000;
const DEAD_SESSION_CACHE_MAX_ENTRIES = 4_096;
const deadSessionExpiries = new Map<string, number>();
const timedOutSessionIds = new Set<string>();

type DeadSessionReason = "not_found" | "timed_out";

const isDeadSessionCached = (sessionId: string, now = Date.now()): boolean => {
  const expiry = deadSessionExpiries.get(sessionId);
  if (expiry === undefined) return false;
  if (expiry > now) return true;
  deadSessionExpiries.delete(sessionId);
  timedOutSessionIds.delete(sessionId);
  return false;
};

const cacheDeadSession = (sessionId: string, reason: DeadSessionReason, now = Date.now()): void => {
  deadSessionExpiries.delete(sessionId);
  if (deadSessionExpiries.size >= DEAD_SESSION_CACHE_MAX_ENTRIES) {
    const oldestSessionId = deadSessionExpiries.keys().next().value;
    if (oldestSessionId !== undefined) {
      deadSessionExpiries.delete(oldestSessionId);
      timedOutSessionIds.delete(oldestSessionId);
    }
  }
  deadSessionExpiries.set(sessionId, now + DEAD_SESSION_CACHE_TTL_MS);
  if (reason === "timed_out") timedOutSessionIds.add(sessionId);
  else timedOutSessionIds.delete(sessionId);
};

const cachedDeadSessionMessage = (sessionId: string): string =>
  timedOutSessionIds.has(sessionId) ? "Session timed out, please reconnect" : "Session not found";

/** Test-only access to reset and verify the isolate-local dead-session cache. */
export const cloudDeadSessionCacheForTest = {
  clear: (): void => {
    deadSessionExpiries.clear();
    timedOutSessionIds.clear();
  },
  remember: (sessionId: string, now?: number): void =>
    cacheDeadSession(sessionId, "not_found", now),
  has: isDeadSessionCached,
  size: (): number => deadSessionExpiries.size,
};

const jsonRpcResponse = (
  status: number,
  code: number,
  message: string,
  challenge?: string,
): Response =>
  challenge === undefined
    ? jsonRpcErrorBody(status, code, message)
    : jsonRpcErrorBody(status, code, message, { challenge });

/**
 * A dead session id answers by request method. POST/DELETE keep the 404 that
 * tells a compliant client to re-initialize. A standalone GET gets 405: the
 * v1 SDK treats that as "no SSE stream offered" and stops retrying quietly,
 * which breaks the reconnect loops of pre-cutover always-on deployments —
 * their GET-404 path never re-initialized, it just retried forever.
 */
const deadSessionResponse = (method: string, message: string): Response =>
  method === "GET"
    ? new Response(JSON.stringify({ jsonrpc: "2.0", error: { code: -32001, message }, id: null }), {
        status: 405,
        headers: {
          "content-type": "application/json",
          allow: "POST, DELETE",
          "access-control-allow-origin": "*",
        },
      })
    : jsonRpcResponse(404, -32001, message);

const renderAuthError = (
  auth: McpAuthProvider["Service"],
  request: Request,
  outcome: Exclude<AuthOutcome, { readonly _tag: "Authenticated" }>,
): Response => {
  if (Predicate.isTagged(outcome, "Unauthorized")) {
    return jsonRpcResponse(
      401,
      -32001,
      "Unauthorized",
      outcome.challenge ?? `Bearer resource_metadata="${auth.resourceMetadataUrl(request)}"`,
    );
  }
  if (Predicate.isTagged(outcome, "Forbidden")) {
    return jsonRpcResponse(403, outcome.code ?? -32001, outcome.message);
  }
  // Unavailable: a transient auth-infra failure (JWKS blip OR a WorkOS
  // membership-lookup 429/5xx/timeout). Both are retryable, so advertise a
  // Retry-After so the client (and any polite retry layer) backs off instead of
  // hammering (same rendering as the shared envelope's Unavailable branch).
  // Crucially, this path NEVER reaches the session-destroy branch below — a
  // transient failure must not condemn a live session.
  //
  // Note this 503 shares JSON-RPC code -32001 with the terminated-session 404
  // ("Session timed out, please reconnect"); that is intentional — -32001 is
  // the generic auth/session envelope code, and the HTTP STATUS is the
  // discriminator clients act on: 503 = retry the SAME session id, 404 = the
  // id is dead, reconnect.
  return jsonRpcErrorBody(503, -32001, outcome.message, {
    retryAfterSeconds: UNAVAILABLE_RETRY_AFTER_SECONDS,
  });
};

const authenticate = (request: Request, authProvider: Layer.Layer<McpAuthProvider>) =>
  Effect.gen(function* () {
    const auth = yield* McpAuthProvider;
    const outcome = yield* auth.authenticate(request);
    return { auth, outcome };
  }).pipe(Effect.provide(authProvider));

// The earlier shared envelope ran the MCP auth path inside the Effect app, whose
// HttpMiddleware provided the OTEL tracer — that is where the `mcp.request`
// span (client fingerprint, rpc method, auth outcome) exported from. This
// handler dispatches from the raw worker entry instead, so a bare
// `Effect.runPromise` leaves every span in Effect's no-op default tracer and
// they silently never export. Run each program under the worker telemetry
// layer, parented to the edge `http.server` span server.ts stamps onto the
// forwarded request's traceparent — which also makes `currentPropagationHeaders`
// (via Effect.currentParentSpan) ferry that same trace into the session DO
// instead of letting the DO start a fresh root per request.
const runTraced = <A>(request: Request, program: Effect.Effect<A>): Promise<A> => {
  const parsed = parseTraceparent(
    request.headers.get("traceparent"),
    request.headers.get("tracestate"),
  );
  return Effect.runPromise(
    (parsed ? OtelTracer.withSpanContext(program, parsed) : program).pipe(
      Effect.provide(WorkerTelemetryLive),
    ),
  );
};

type TraceCloudMcpRequest = (
  request: Request,
  env: Env,
  ctx: ExecutionContext,
  handle: (tracedRequest: Request) => Promise<Response>,
) => Promise<Response>;

interface CloudMcpAgentHandlerOptions {
  readonly makeModernServerBuilder: (
    session: McpSessionProps["session"],
  ) => McpModernServerBuilder["Service"];
  readonly authProvider?: Layer.Layer<McpAuthProvider>;
  readonly traceRequest?: TraceCloudMcpRequest;
}

// The MCP resource the request targets. `server.ts` routes both the bare `/mcp`
// and `/mcp/toolkits/<slug>` to this handler (`prepareMcpOrgScope` strips the org
// selector but keeps the toolkit segment), so a session minted on a toolkit path
// scopes its tool catalog to that toolkit.
const resourceFromPath = (request: Request): McpResource => {
  const segments = new URL(request.url).pathname.split("/").filter((s) => s.length > 0);
  if (segments.length === 3 && segments[0] === "mcp" && segments[1] === "toolkits" && segments[2]) {
    return { kind: "toolkit", slug: segments[2] };
  }
  return defaultMcpResource;
};

const propsForPrincipal = (
  request: Request,
  principal: Extract<AuthOutcome, { readonly _tag: "Authenticated" }>["principal"],
  resource: McpResource,
): Effect.Effect<McpSessionProps> =>
  Effect.gen(function* () {
    const propagation = yield* currentPropagationHeaders(request);
    return {
      session: {
        organizationId: principal.organizationId,
        userId: principal.accountId,
        elicitationMode: readElicitationMode(request),
        artifactsEnabled: readArtifactsEnabled(request),
        resource,
        webOrigin: new URL(request.url).origin,
      },
      propagation,
    };
  });

/** Build the cloud worker's authenticated legacy/modern MCP request handler. */
export const makeCloudMcpAgentHandler = (options: CloudMcpAgentHandlerOptions) => {
  const authProvider = options.authProvider ?? cloudMcpAuth;
  const traceRequest = options.traceRequest ?? ((request, _env, _ctx, handle) => handle(request));
  const modern = makeMcpModernRequestRouter();
  const ALLOWED_METHODS = new Set(["GET", "POST", "DELETE", "OPTIONS"]);

  const handle = async (request: Request, env: Env, ctx: ExecutionContext): Promise<Response> => {
    if (request.method === "OPTIONS") {
      return mcpCorsPreflightResponse(request.headers.get("access-control-request-headers"));
    }
    // Preserve the old envelope's JSON-RPC 405 before authenticating, so
    // unsupported methods never reach the session engine.
    if (!ALLOWED_METHODS.has(request.method)) {
      return jsonRpcResponse(405, -32001, "Method not allowed");
    }
    const sessionId = request.headers.get("mcp-session-id");

    const { auth, outcome } = await runTraced(request, authenticate(request, authProvider));
    if (!Predicate.isTagged(outcome, "Authenticated")) {
      // Destroying a live session on auth grounds requires a POSITIVE
      // determination that access is genuinely gone — only `Forbidden` carries
      // that (valid bearer, org absent/revoked). `Unavailable` (transient WorkOS
      // / JWKS failure) and `Unauthorized` (retry with a fresh token) must leave
      // the session intact, so the condemn path is gated on `Forbidden` alone.
      if (Predicate.isTagged(outcome, "Forbidden") && sessionId) {
        const session = mcpSessionStub(env.MCP_SESSION, sessionId);
        await Effect.runPromise(
          Effect.ignore(
            session ? Effect.tryPromise(() => session._cf_scheduleDestroy()) : Effect.void,
          ),
        );
      }
      return renderAuthError(auth, request, outcome);
    }

    const parsedBody = await Effect.runPromise(requestBodyFromRequest(request));
    const era = await classifyMcpProtocolEra(request, parsedBody);
    if (era === "modern") {
      if (env.MCP_2026_07_28_ENABLED === "false") {
        return mcpModernDisabledResponse();
      }
      const resource = resourceFromPath(request);
      const props = await runTraced(
        request,
        propsForPrincipal(request, outcome.principal, resource),
      );
      (ctx as ExecutionContext & { props?: McpSessionProps }).props = props;
      const forwarded = withVerifiedIdentityHeaders(
        request,
        {
          accountId: outcome.principal.accountId,
          organizationId: outcome.principal.organizationId,
        },
        resource,
      );
      return modern.fetch({
        request: forwarded,
        parsedBody,
        principal: outcome.principal,
        resource,
        props,
        requestStateSigningKey: requireMcpRequestStateKey(env.MCP_REQUEST_STATE_KEY),
        builder: options.makeModernServerBuilder(props.session),
        sessions: env.MCP_SESSION,
        executionOwners: mcpExecutionOwnerDirectoryFromNamespace(env.MCP_EXECUTION_OWNER),
      });
    }

    if (!sessionId && request.method === "DELETE") {
      // Matches the old envelope's contract (@modelcontextprotocol/sdk's
      // `WebStandardStreamableHTTPServerTransport.handleDeleteRequest`): 200,
      // not 204 — see e2e/cloud/mcp-protocol.test.ts.
      return new Response(null, {
        status: 200,
        headers: { "access-control-allow-origin": "*" },
      });
    }

    const existingSession = sessionId ? mcpSessionStub(env.MCP_SESSION, sessionId) : null;
    if (sessionId && !existingSession) {
      return deadSessionResponse(request.method, "Session not found");
    }
    if (existingSession && sessionId) {
      const owner = await existingSession.validateMcpSessionOwner({
        accountId: outcome.principal.accountId,
        organizationId: outcome.principal.organizationId,
      });
      if (owner === "not_found") {
        cacheDeadSession(sessionId, "not_found");
        return deadSessionResponse(request.method, "Session not found");
      }
      if (owner === "terminated") {
        // DELETE-condemned but the deferred destroy alarm hasn't wiped storage
        // yet. Same envelope as the post-destroy race below: the client must
        // treat the id as dead and reconnect.
        cacheDeadSession(sessionId, "timed_out");
        return deadSessionResponse(request.method, "Session timed out, please reconnect");
      }
      if (owner === "forbidden") {
        return jsonRpcResponse(403, -32003, "MCP session does not belong to the current bearer");
      }
    }

    const resource = resourceFromPath(request);
    const propagation = await runTraced(request, currentPropagationHeaders(request));
    const forwarded = withPropagationHeaders(
      withVerifiedIdentityHeaders(
        request,
        {
          accountId: outcome.principal.accountId,
          organizationId: outcome.principal.organizationId,
        },
        resource,
      ),
      propagation,
    );
    const target = existingSession ?? createMcpSessionStub(env.MCP_SESSION).stub;
    let response: Response;
    // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary: a condemned DO abort can reject its direct fetch
    try {
      response = await target.fetch(forwarded);
    } catch (error) {
      // `_cf_scheduleDestroy` (called above via DELETE) marks the DO
      // condemned and schedules its alarm; the alarm's storage wipe then
      // `ctx.abort("destroyed")`s the isolate. A request that lands after the
      // alarm has already fired — same DO, same tick budget as the DELETE in
      // tests — throws that abort reason out of `stub.fetch` instead of the
      // DO ever getting to answer. Map it to the old envelope's reconnect
      // error for a dead session (e2e/cloud/mcp-protocol.test.ts expects the
      // client to be told to reconnect, matching a timed-out session).
      // oxlint-disable-next-line executor/no-unknown-error-message -- adapter boundary: the abort reason is a plain runtime Error whose message IS the signal
      if (Predicate.isError(error) && error.message === "destroyed") {
        if (sessionId) cacheDeadSession(sessionId, "timed_out");
        return deadSessionResponse(request.method, "Session timed out, please reconnect");
      }
      // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary: rethrow anything that isn't the condemned-DO abort to the Workers runtime unchanged
      throw error;
    }
    return withMcpResponseHeaders(wrapMcpSseResponse(request, env, response));
  };

  return async (request: Request, env: Env, ctx: ExecutionContext): Promise<Response> => {
    const sessionId = request.headers.get("mcp-session-id");
    const cacheEligible = request.method !== "OPTIONS" && ALLOWED_METHODS.has(request.method);
    if (cacheEligible && sessionId && isDeadSessionCached(sessionId)) {
      return Effect.runPromise(
        Effect.gen(function* () {
          const { auth, outcome } = yield* authenticate(request, authProvider);
          return Predicate.isTagged(outcome, "Authenticated")
            ? deadSessionResponse(request.method, cachedDeadSessionMessage(sessionId))
            : renderAuthError(auth, request, outcome);
        }).pipe(Effect.withTracerEnabled(false)),
      );
    }
    return traceRequest(request, env, ctx, (tracedRequest) => handle(tracedRequest, env, ctx));
  };
};