moteva/frontend/src/infrastructure/designGateway/events.ts

import type { AgentMessage, ExtractNodeTextResponse, Project } from "@/domain/design";
import { authHeaders } from "@/infrastructure/authGateway";
import { createApiFailure } from "@/infrastructure/userMessages";

import type { ProjectEventHandlers, ProjectEventsOptions } from "./types";

export function projectEvents(projectId: string, handlers: ProjectEventHandlers, options: ProjectEventsOptions = {}) {
  const controller = new AbortController();
  let settled = false;
  const readProject = (data: string) => {
    const payload = JSON.parse(data) as { project: Project };
    return payload.project;
  };
  const close = () => {
    settled = true;
    controller.abort();
  };
  const onEvent = (event: string, data: string) => {
    if (event === "project") {
      handlers.onProject?.(readProject(data));
      return;
    }
    if (event === "thinking") {
      handlers.onThinking?.(JSON.parse(data) as AgentMessage);
      return;
    }
    if (event === "text_extraction") {
      handlers.onTextExtraction?.(JSON.parse(data) as ExtractNodeTextResponse);
      return;
    }
    if (event === "done") {
      settled = true;
      handlers.onDone?.(readProject(data));
      controller.abort();
      return;
    }
    if (event === "timeout") {
      settled = true;
      handlers.onTimeout?.(readProject(data));
      controller.abort();
      return;
    }
    if (event === "stream_error") {
      settled = true;
      const payload = data ? (JSON.parse(data) as { error?: string }) : {};
      handlers.onError?.(new Error(payload.error ?? "Event stream failed"));
      controller.abort();
    }
  };

  void streamProjectEvents(projectEventsUrl(projectId, options), controller.signal, onEvent).catch((error: Error) => {
    if (settled || controller.signal.aborted) return;
    settled = true;
    handlers.onError?.(error);
  });

  return close;
}

function projectEventsUrl(projectId: string, options: ProjectEventsOptions) {
  if (options.eventsUrl) return options.eventsUrl;
  const params = new URLSearchParams();
  if (options.threadId) params.set("threadId", options.threadId);
  const query = params.toString();
  return `/api/projects/${projectId}/events${query ? `?${query}` : ""}`;
}

async function streamProjectEvents(url: string, signal: AbortSignal, onEvent: (event: string, data: string) => void) {
  const headers = new Headers();
  Object.entries(authHeaders()).forEach(([key, value]) => headers.set(key, value));
  const response = await fetch(url, {
    headers,
    signal,
    cache: "no-store",
    credentials: "omit"
  });
  if (!response.ok) {
    const payload = await response.json().catch(() => ({ error: response.statusText }));
    throw createApiFailure("design", response, payload);
  }
  if (!response.body) {
    throw new Error("Event stream failed");
  }

  const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = "";
  try {
    for (;;) {
      const { value, done } = await reader.read();
      if (done) break;
      buffer += value;
      const parts = buffer.split(/\r?\n\r?\n/);
      buffer = parts.pop() ?? "";
      parts.forEach((part) => emitServerSentEvent(part, onEvent));
    }
    if (buffer.trim()) {
      emitServerSentEvent(buffer, onEvent);
    }
  } finally {
    reader.releaseLock();
  }
}

function emitServerSentEvent(raw: string, onEvent: (event: string, data: string) => void) {
  let event = "message";
  const data: string[] = [];
  raw.split(/\r?\n/).forEach((line) => {
    if (line.startsWith("event:")) {
      event = line.slice("event:".length).trim();
      return;
    }
    if (line.startsWith("data:")) {
      data.push(line.slice("data:".length).trimStart());
    }
  });
  if (data.length > 0) {
    onEvent(event, data.join("\n"));
  }
}

export const eventOperations = {
  projectEvents
};