/*
 * SPEC: PJ2026-0104010803 Workbench事件流可见性 draft-2026-07-09-p0-kafka-authority.
 * Responsibility: debug-only SSE passthrough for HWLAB Kafka event envelopes.
 */
import { sendJson } from "./server-http-utils.ts";
import { openKafkaEventStream } from "./kafka-event-bridge.ts";
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { authenticateWorkbenchRead } from "./server-workbench-read-http.ts";

const CONTRACT_VERSION = "workbench-debug-kafka-sse-v1";
const DEFAULT_STREAM = "hwlab";
const ALLOWED_STREAMS = new Set(["stdio", "agentrun", "hwlab"]);

export async function handleWorkbenchKafkaSseDebugHttp(request, response, url, options = {}) {
  const route = routeSuffix(url.pathname);
  try {
    const auth = await authenticateWorkbenchRead(request, response, options);
    if (!auth) return;
    if (auth.actor?.role !== "admin") {
      return sendJson(response, 403, debugError("admin_required", "Only admin users can access raw Workbench Kafka debug streams."));
    }
    if (route === "") {
      if (request.method !== "GET") return methodNotAllowed(response, "GET");
      return sendJson(response, 200, describeKafkaSseDebug(url, options.env ?? process.env));
    }
    if (route === "/events") {
      if (request.method !== "GET") return methodNotAllowed(response, "GET");
      return openKafkaDebugSse(request, response, url, options);
    }
    return sendJson(response, 404, debugError("workbench_debug_kafka_sse_route_not_found", "Workbench debug Kafka SSE route is not implemented.", { route }));
  } catch (error) {
    options.logger?.warn?.({
      event: "workbench_debug_kafka_sse_failed",
      route,
      errorName: error?.name ?? "Error",
      message: error instanceof Error ? error.message : String(error ?? "unknown"),
      valuesRedacted: true
    });
    if (response.headersSent) return writeSse(response, "hwlab.kafka.error", { ok: false, error: errorMessagePayload(error) });
    return sendJson(response, 500, debugError("workbench_debug_kafka_sse_failed", "Workbench debug Kafka SSE request failed."));
  }
}

function describeKafkaSseDebug(url, env) {
  const stream = streamFromUrl(url);
  const requestedFilters = filtersFromUrl(url);
  const resolvedFilters = resolveKafkaDebugFilters(requestedFilters, { traceStore: defaultCodeAgentTraceStore });
  return {
    ok: true,
    contractVersion: CONTRACT_VERSION,
    stream,
    topic: topicForStream(stream, env),
    filters: requestedFilters,
    resolvedFilters,
    eventsRoute: `/v1/workbench/debug/kafka-sse/events${url.search || ""}`,
    valuesPrinted: false
  };
}

async function openKafkaDebugSse(request, response, url, options) {
  const env = options.env ?? process.env;
  const stream = streamFromUrl(url);
  const filters = filtersFromUrl(url);
  const resolvedFilters = resolveKafkaDebugFilters(filters, options);
  response.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store, no-transform",
    connection: "keep-alive",
    "x-accel-buffering": "no",
    "x-content-type-options": "nosniff"
  });
  if (typeof response.flushHeaders === "function") response.flushHeaders();
  let kafkaStream = null;
  let closed = false;
  const close = () => {
    if (closed) return;
    closed = true;
    void kafkaStream?.stop?.();
  };
  response.once("close", close);
  request.once?.("aborted", close);
  request.socket?.once?.("close", close);
  try {
    kafkaStream = await openKafkaEventStream({
      env,
      stream,
      fromBeginning: url.searchParams.get("fromBeginning") === "1" || url.searchParams.get("fromBeginning") === "true",
      ...resolvedFilters,
      kafkaFactory: options.kafkaFactory,
      onEvent: async (record) => {
        writeSse(response, "hwlab.kafka.event", {
          ok: true,
          contractVersion: CONTRACT_VERSION,
          stream,
          topic: record.topic,
          partition: record.partition,
          offset: record.offset,
          key: record.key,
          timestamp: record.timestamp,
          value: record.value,
          serverSentAt: new Date().toISOString(),
          valuesPrinted: false
        });
      }
    });
    if (closed) {
      await kafkaStream.stop?.();
      return;
    }
    writeSse(response, "hwlab.kafka.connected", {
      ok: true,
      contractVersion: CONTRACT_VERSION,
      consumerReady: true,
      groupId: kafkaStream.groupId,
      stream,
      topic: kafkaStream.topic ?? topicForStream(stream, env),
      filters,
      resolvedFilters,
      serverSentAt: new Date().toISOString(),
      valuesPrinted: false
    });
  } catch (error) {
    writeSse(response, "hwlab.kafka.error", { ok: false, error: errorMessagePayload(error), valuesPrinted: false });
  }
}

function resolveKafkaDebugFilters(filters = {}, options = {}) {
  const traceId = textValue(filters.traceId);
  if (!traceId) return filters;
  const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
  const snapshot = typeof traceStore?.snapshot === "function" ? traceStore.snapshot(traceId) : null;
  const resolved = collectTraceLinkedIds(snapshot);
  const hasResolvedKafkaKey = Boolean(resolved.runId || resolved.commandId || resolved.sessionId);
  if (!hasResolvedKafkaKey) return filters;
  const resolvedSessionId = resolved.runId || resolved.commandId ? null : resolved.sessionId;
  return compactObject({
    sessionId: filters.sessionId || resolvedSessionId,
    runId: filters.runId || resolved.runId,
    commandId: filters.commandId || resolved.commandId
  });
}

function collectTraceLinkedIds(snapshot) {
  const out = { sessionId: null, runId: null, commandId: null };
  const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
  for (const event of events) collectIdsFromRecord(event, out);
  collectIdsFromRecord(snapshot?.lastEvent, out);
  return out;
}

function collectIdsFromRecord(record, out) {
  const value = record && typeof record === "object" && !Array.isArray(record) ? record : {};
  const agentRun = value.agentRun && typeof value.agentRun === "object" && !Array.isArray(value.agentRun) ? value.agentRun : {};
  const payload = value.payload && typeof value.payload === "object" && !Array.isArray(value.payload) ? value.payload : {};
  out.sessionId ||= textValue(value.sessionId ?? value.sourceSessionId ?? agentRun.sessionId ?? payload.sessionId);
  out.runId ||= textValue(value.runId ?? value.sourceRunId ?? agentRun.runId ?? payload.runId);
  out.commandId ||= textValue(value.commandId ?? value.sourceCommandId ?? agentRun.commandId ?? payload.commandId);
}

function writeSse(response, eventName, payload) {
  if (response.destroyed || response.writableEnded) return false;
  try {
    response.write(`event: ${eventName}\n`);
    const id = sseEventId(payload);
    if (id) response.write(`id: ${id}\n`);
    response.write(`data: ${JSON.stringify(payload)}\n\n`);
    return true;
  } catch {
    return false;
  }
}

function routeSuffix(pathname) {
  return String(pathname || "").replace(/^\/v1\/workbench\/debug\/kafka-sse/u, "");
}

function streamFromUrl(url) {
  const stream = String(url.searchParams.get("stream") || DEFAULT_STREAM).trim();
  return ALLOWED_STREAMS.has(stream) ? stream : DEFAULT_STREAM;
}

function filtersFromUrl(url) {
  return compactObject({
    traceId: safeId(url.searchParams.get("traceId") || url.searchParams.get("trace-id")),
    sessionId: safeId(url.searchParams.get("sessionId") || url.searchParams.get("session-id")),
    runId: safeId(url.searchParams.get("runId") || url.searchParams.get("run-id")),
    commandId: safeId(url.searchParams.get("commandId") || url.searchParams.get("command-id"))
  });
}

function topicForStream(stream, env) {
  if (stream === "stdio") return textValue(env.HWLAB_KAFKA_STDIO_TOPIC ?? env.AGENTRUN_KAFKA_STDIO_TOPIC) || "codex-stdio.raw.v1";
  if (stream === "agentrun") return textValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC) || "agentrun.event.v1";
  return textValue(env.HWLAB_KAFKA_EVENT_TOPIC) || "hwlab.event.v1";
}

function methodNotAllowed(response, allow) {
  response.setHeader("allow", allow);
  return sendJson(response, 405, debugError("method_not_allowed", `Method not allowed; expected ${allow}.`));
}

function debugError(code, message, extra = {}) {
  return { ok: false, error: { code, message, ...extra }, valuesPrinted: false };
}

function errorMessagePayload(error) {
  return { code: "kafka_sse_error", message: error instanceof Error ? error.message : String(error ?? "unknown") };
}

function safeId(value) {
  const text = textValue(value);
  return text && /^[A-Za-z0-9_.:-]{3,220}$/u.test(text) ? text : null;
}

function textValue(value) {
  const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim();
  return text.length > 0 ? text : null;
}

function compactObject(value) {
  return Object.fromEntries(Object.entries(value).filter(([, entry]) => textValue(entry)));
}

function sseEventId(payload) {
  const topic = textValue(payload?.topic);
  const partition = textValue(payload?.partition);
  const offset = textValue(payload?.offset);
  return topic && partition && offset ? `${topic}:${partition}:${offset}` : null;
}