import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js";
import { trace } from "@opentelemetry/api";
import { db } from "@/lib/db";
import { createMcpServerForEndpoint } from "@/lib/mcp/handler";
import { recordMcpInvalidRequest } from "@/lib/mcp/metrics";

// SSE connections are long-lived by design (MCP Streamable HTTP spec).
// p99 ≈ 299 s is the reverse-proxy idle timeout, not an app bug.
// Keep-alive comments sent every 30 s prevent Traefik/nginx from dropping idle connections.
// maxDuration tells Next.js/Vercel the expected maximum connection duration.
export const maxDuration = 300;

function withSseKeepAlive(response: Response, intervalMs = 30_000): Response {
  if (!response.body) return response;
  const contentType = response.headers.get("content-type") ?? "";
  if (!contentType.includes("text/event-stream")) return response;

  const encoder = new TextEncoder();
  const ping = encoder.encode(": ping\n\n");
  const { readable, writable } = new TransformStream<Uint8Array, Uint8Array>();
  const writer = writable.getWriter();

  const pump = async () => {
    const reader = response.body!.getReader();
    const pingTimer = setInterval(() => {
      writer.write(ping).catch(() => clearInterval(pingTimer));
    }, intervalMs);
    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        await writer.write(value);
      }
    } finally {
      clearInterval(pingTimer);
      writer.close().catch(() => {});
    }
  };

  pump().catch(() => writer.close().catch(() => {}));

  return new Response(readable, { status: response.status, headers: response.headers });
}

async function handleMcpRequest(
  request: Request,
  apiKey: string
): Promise<Response> {
  // Explicitly tag the active OTel span with the route pattern.
  // The http-level requestHook in instrumentation.node.ts also sets this, but in
  // production (no src/app/ present) normalizeRoute falls back to the raw URL which
  // embeds the actual API key. Setting it here on the framework-level span ensures
  // SigNoz always aggregates all MCP traffic under a single canonical route.
  trace.getActiveSpan()?.setAttribute("http.route", "/api/mcp/[apiKey]");

  // Look up endpoint by API key
  const endpoint = await db.mcpEndpoint.findUnique({
    where: { apiKey },
    include: {
      permissions: { select: { action: true } },
      serviceConnection: true,
    },
  });

  if (!endpoint) {
    recordMcpInvalidRequest("unauthorized");
    return new Response(JSON.stringify({ error: "Invalid API key" }), {
      status: 401,
      headers: { "Content-Type": "application/json" },
    });
  }

  if (!endpoint.isActive) {
    recordMcpInvalidRequest("unauthorized");
    return new Response(
      JSON.stringify({ error: "Endpoint is deactivated" }),
      { status: 403, headers: { "Content-Type": "application/json" } }
    );
  }

  // Rate limit check — atomic upsert to avoid races and remove auditLog.count hot-path query.
  // Bucket key = (endpointId, current UTC minute). Count is incremented before execution.
  const windowStart = new Date(Math.floor(Date.now() / 60_000) * 60_000);
  const result = await db.$queryRaw<{ count: number }[]>`
    INSERT INTO "rate_limit_bucket" ("endpointId", "windowStart", "count")
    VALUES (${endpoint.id}, ${windowStart}, 1)
    ON CONFLICT ("endpointId", "windowStart")
    DO UPDATE SET "count" = "rate_limit_bucket"."count" + 1
    RETURNING "count"
  `;
  const currentCount = Number(result[0]?.count ?? 1);

  if (currentCount > endpoint.rateLimitPerMinute) {
    recordMcpInvalidRequest("rate_limited");
    return new Response(JSON.stringify({ error: "Rate limit exceeded" }), {
      status: 429,
      headers: { "Content-Type": "application/json" },
    });
  }

  const allowedActions = endpoint.permissions.map((p) => p.action);
  const server = createMcpServerForEndpoint(
    endpoint.id,
    endpoint.name,
    allowedActions,
    endpoint.serviceConnectionId
  );

  // Create stateless transport for this request
  const transport = new WebStandardStreamableHTTPServerTransport({
    sessionIdGenerator: undefined,
  });

  await server.connect(transport);

  // Clone before the transport consumes the body so we can log it on 400.
  const requestClone = request.clone();
  const response = await transport.handleRequest(request);

  if (response.status === 400) {
    recordMcpInvalidRequest("invalid_format");
    const bodyPreview = await requestClone.text().catch(() => "<unreadable>");
    console.log(
      JSON.stringify({
        event: "mcp.invalid_request",
        route: "/api/mcp/[apiKey]",
        status: 400,
        body_preview: bodyPreview.slice(0, 500),
      })
    );
  }

  return withSseKeepAlive(response);
}

export async function GET(
  request: Request,
  { params }: { params: Promise<{ apiKey: string }> }
) {
  const { apiKey } = await params;
  return handleMcpRequest(request, apiKey);
}

export async function POST(
  request: Request,
  { params }: { params: Promise<{ apiKey: string }> }
) {
  const { apiKey } = await params;
  return handleMcpRequest(request, apiKey);
}

export async function DELETE(
  request: Request,
  { params }: { params: Promise<{ apiKey: string }> }
) {
  const { apiKey } = await params;
  return handleMcpRequest(request, apiKey);
}
