feat(workbench): integrate operational workflows and map state

2026-09-24 04:34:46 +08:00
parent 1752b9ef84
commit 436d5500f1
76 changed files with 5081 additions and 1937 deletions
@@ -0,0 +1,127 @@
+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;
+  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();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  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";
+}