next-tjwater-frontend/src/features/workbench/api/workbench-sync-client.ts

import { z } from "zod";
import type { AccessTokenProvider } from "@/shared/auth/keycloak-auth";
import { env } from "@/shared/config/env";
import {
  parseOperationalTaskNotification,
  parseOperationalTaskSummary,
  type OperationalTaskNotification,
  type OperationalTaskSummary
} from "./operational-task-client";
import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client";
import {
  parseWorkbenchRuntimeSnapshot,
  type WorkbenchRuntimeSnapshot
} from "./workbench-runtime-client";

const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, "");

const snapshotEnvelopeSchema = z.object({
  schema: z.literal("workbench-sync-snapshot@1"),
  event_cursor: z.number().int().nonnegative(),
  runtime: z.unknown(),
  tasks: z.array(z.unknown()),
  notifications: z.array(z.unknown()),
  generated_at: z.string().datetime({ offset: true })
}).strict();

const eventSchema = z.object({
  schema: z.literal("workbench-event@1"),
  event_seq: z.number().int().positive(),
  event_id: z.string().uuid(),
  project_id: z.string().uuid(),
  topic: z.enum([
    "operational.task.changed",
    "operational.run.queued",
    "operational.run.changed",
    "operational.report.published",
    "operational.action.changed",
    "operational.notification.changed",
    "workbench.alert.changed",
    "workbench.work_order.changed",
    "analysis.run.changed"
  ]),
  aggregate_type: z.string().min(1),
  aggregate_id: z.string().min(1),
  aggregate_revision: z.number().int().positive(),
  occurred_at: z.string().datetime({ offset: true }),
  metadata: z.record(z.string(), z.unknown())
}).strict();

export type WorkbenchEvent = z.infer<typeof eventSchema>;

export type WorkbenchSyncSnapshot = {
  eventCursor: number;
  runtime: WorkbenchRuntimeSnapshot;
  tasks: OperationalTaskSummary[];
  notifications: OperationalTaskNotification[];
  generatedAt: string;
};

export async function fetchWorkbenchSyncSnapshot(
  getAccessToken?: AccessTokenProvider,
  signal?: AbortSignal
): Promise<WorkbenchSyncSnapshot> {
  const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, signal);
  const response = await fetch(`${SERVER_API_BASE_URL}/api/v1/workbench/sync-snapshot`, {
    headers: createWorkbenchHeaders(accessToken, project.project_id),
    signal
  });
  const payload: unknown = await response.json().catch(() => null);
  if (!response.ok) throw new Error(getServerError(payload, response.status, "工作台同步快照请求"));
  const envelope = snapshotEnvelopeSchema.parse(payload);
  return {
    eventCursor: envelope.event_cursor,
    runtime: parseWorkbenchRuntimeSnapshot(envelope.runtime),
    tasks: envelope.tasks.map(parseOperationalTaskSummary),
    notifications: envelope.notifications.map(parseOperationalTaskNotification),
    generatedAt: envelope.generated_at
  };
}

export async function consumeWorkbenchEvents(input: {
  after: number;
  getAccessToken?: AccessTokenProvider;
  onEvents: (events: WorkbenchEvent[]) => void;
  onOpen?: () => void;
  onReset: () => Promise<void>;
  signal: AbortSignal;
}): Promise<"closed" | "reset"> {
  const { project, accessToken } = await findOwnedWorkbenchProject(input.getAccessToken, input.signal);
  const headers = createWorkbenchHeaders(accessToken, project.project_id);
  headers.set("Accept", "text/event-stream");
  headers.set("Last-Event-ID", String(input.after));
  const response = await fetch(
    `${SERVER_API_BASE_URL}/api/v1/workbench/events?after=${input.after}`,
    { headers, signal: input.signal }
  );
  if (!response.ok || !response.body) {
    const payload: unknown = await response.json().catch(() => null);
    throw new Error(getServerError(payload, response.status, "工作台事件连接"));
  }
  const reader = response.body.getReader();
  input.onOpen?.();
  const decoder = new TextDecoder();
  let buffer = "";
  try {
    while (!input.signal.aborted) {
      const { value, done } = await reader.read();
      buffer += decoder.decode(value, { stream: !done });
      if (buffer.length > 1_048_576) throw new Error("工作台事件帧超过大小限制");
      const events: WorkbenchEvent[] = [];
      let boundary = buffer.search(/\r?\n\r?\n/);
      while (boundary >= 0) {
        const separator = buffer.slice(boundary).match(/^\r?\n\r?\n/)?.[0] ?? "\n\n";
        const block = buffer.slice(0, boundary);
        buffer = buffer.slice(boundary + separator.length);
        const lines = block.split(/\r?\n/);
        const eventName = lines.find((line) => line.startsWith("event:"))?.slice(6).trim();
        const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
        if (eventName === "reset_required") {
          await input.onReset();
          return "reset";
        }
        if (eventName === "workbench_event" && data) events.push(eventSchema.parse(JSON.parse(data)));
        boundary = buffer.search(/\r?\n\r?\n/);
      }
      if (events.length > 0) input.onEvents(events);
      if (done) return "closed";
    }
    return "closed";
  } finally {
    await reader.cancel().catch(() => undefined);
    reader.releaseLock();
  }
}