250 lines
13 KiB
TypeScript
250 lines
13 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
import type { InvestigationPresentationCommand, InvestigationPresentationCursor } from "../api/investigation-presentation";
|
|
import type { WorkbenchEvent } from "../api/workbench-sync-client";
|
|
import { InvestigationPresentationFollower } from "./investigation-presentation-follower";
|
|
import { WorkbenchEventChannel } from "./workbench-event-channel";
|
|
|
|
const taskId = "50000000-0000-4000-8000-000000000001";
|
|
const runId = "55000000-0000-4000-8000-000000000001";
|
|
const nextRunId = "55000000-0000-4000-8000-000000000002";
|
|
const highlight: InvestigationPresentationCommand = { action: "highlight_features", layer: "scada", feature_ids: ["MP01"], effect: "pulse" };
|
|
const openChart: InvestigationPresentationCommand = {
|
|
action: "open_chart", chart_id: "pressure", evidence_refs: ["reading-1"],
|
|
chart: { title: "压力", chart_type: "line", x_data: ["10:00"], series: [{ name: "MP01", data: [0.4] }] }
|
|
};
|
|
const closeChart: InvestigationPresentationCommand = { action: "close_chart", chart_id: "pressure" };
|
|
|
|
function cursor(eventCursor: number, generation = 1): InvestigationPresentationCursor {
|
|
return { task_id: taskId, execution_generation: generation, event_cursor: eventCursor,
|
|
active_run: { run_id: generation === 1 ? runId : nextRunId, generation, status: "running" } };
|
|
}
|
|
|
|
function event(sequence: number, command = highlight, generation = 1): WorkbenchEvent {
|
|
return { schema: "workbench-event@1", event_seq: sequence, event_id: "60000000-0000-4000-8000-000000000001",
|
|
project_id: "10000000-0000-4000-8000-000000000001", topic: "operational.presentation.requested",
|
|
aggregate_type: "operational_presentation", aggregate_id: taskId, aggregate_revision: 1,
|
|
occurred_at: "2026-10-10T01:00:00Z", metadata: {
|
|
task_id: taskId, run_id: generation === 1 ? runId : nextRunId, execution_generation: generation, command
|
|
} };
|
|
}
|
|
|
|
function deferred<T>() {
|
|
let resolve!: (value: T) => void;
|
|
const promise = new Promise<T>((done) => { resolve = done; });
|
|
return { promise, resolve };
|
|
}
|
|
|
|
function setup(overrides: Partial<ConstructorParameters<typeof InvestigationPresentationFollower>[0]> = {}) {
|
|
const channel = new WorkbenchEventChannel();
|
|
const execute = vi.fn(async (_command: InvestigationPresentationCommand, _signal: AbortSignal) => {});
|
|
const fetchCursor = vi.fn(async (_signal: AbortSignal) => cursor(10));
|
|
const fetchIdentity = vi.fn(async (_signal: AbortSignal) => cursor(10));
|
|
const clear = vi.fn();
|
|
const options = { taskId, subscribe: channel.subscribe, execute, fetchCursor, fetchIdentity,
|
|
clear, stopPulse: vi.fn(), onStatus: vi.fn() };
|
|
const follower = new InvestigationPresentationFollower({ ...options, ...overrides });
|
|
return { channel, follower, ...options };
|
|
}
|
|
|
|
describe("live investigation presentation following", () => {
|
|
it("discards queued map commands while retaining chart commands and later new map events", async () => {
|
|
const painting = deferred<void>();
|
|
const execute = vi.fn((command: InvestigationPresentationCommand) =>
|
|
command.action === "open_chart" ? painting.promise : Promise.resolve());
|
|
const test = setup({ execute });
|
|
test.follower.start();
|
|
test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
|
test.channel.publish({ type: "events", events: [event(11, openChart), event(12), event(13, closeChart)] });
|
|
test.follower.discardPendingMapCommands();
|
|
painting.resolve();
|
|
await vi.waitFor(() => expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["open_chart", "close_chart"]));
|
|
test.channel.publish({ type: "events", events: [event(14)] });
|
|
await vi.waitFor(() => expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["open_chart", "close_chart", "highlight_features"]));
|
|
} finally { painting.resolve(); test.follower.stop(); }
|
|
});
|
|
|
|
it("listens before fetching the baseline and preserves events arriving during cursor lookup", async () => {
|
|
const baseline = deferred<InvestigationPresentationCursor>();
|
|
const order: string[] = [];
|
|
const test = setup();
|
|
test.channel.publish({ type: "open" });
|
|
const fetchCursor = vi.fn(() => {
|
|
order.push("cursor");
|
|
test.channel.publish({ type: "events", events: [event(8), event(11)] });
|
|
return baseline.promise;
|
|
});
|
|
const follower = new InvestigationPresentationFollower({ ...test,
|
|
subscribe: (listener) => { order.push("listen"); return test.channel.subscribe(listener); }, fetchCursor });
|
|
follower.start();
|
|
try {
|
|
expect(order).toEqual(["listen", "cursor"]);
|
|
expect(test.execute).not.toHaveBeenCalled();
|
|
baseline.resolve(cursor(10));
|
|
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
|
expect(test.execute.mock.calls[0]?.[0]).toEqual(highlight);
|
|
} finally { follower.stop(); }
|
|
});
|
|
|
|
it("executes same-batch chart commands in sequence and waits for open before close", async () => {
|
|
const painted = deferred<void>();
|
|
const execute = vi.fn((command: InvestigationPresentationCommand) => command.action === "open_chart" ? painted.promise : Promise.resolve());
|
|
const test = setup({ execute });
|
|
test.follower.start();
|
|
test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
|
test.channel.publish({ type: "events", events: [event(12, closeChart), event(11, openChart)] });
|
|
expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["open_chart"]);
|
|
painted.resolve();
|
|
await vi.waitFor(() => expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["open_chart", "close_chart"]));
|
|
} finally { painted.resolve(); test.follower.stop(); }
|
|
});
|
|
|
|
it("verifies a new generation without losing its first command or fetching a new baseline", async () => {
|
|
const fetchIdentity = vi.fn(async () => cursor(10, 2));
|
|
const test = setup({ fetchIdentity });
|
|
test.follower.start();
|
|
test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
|
test.channel.publish({ type: "events", events: [event(11, highlight, 2)] });
|
|
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
|
expect(fetchIdentity).toHaveBeenCalledTimes(1);
|
|
expect(test.fetchCursor).toHaveBeenCalledTimes(1);
|
|
expect(test.execute.mock.calls[0]?.[0]).toEqual(highlight);
|
|
} finally { test.follower.stop(); }
|
|
});
|
|
|
|
it("retains a command while identity lookup is stale and applies it once the task snapshot catches up", async () => {
|
|
const test = setup();
|
|
test.follower.start();
|
|
test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
|
test.channel.publish({ type: "events", events: [event(11, highlight, 2)] });
|
|
await vi.waitFor(() => expect(test.fetchIdentity).toHaveBeenCalledTimes(1));
|
|
expect(test.execute).not.toHaveBeenCalled();
|
|
test.follower.updateIdentity(cursor(10, 2));
|
|
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
|
expect(test.fetchCursor).toHaveBeenCalledTimes(1);
|
|
} finally { test.follower.stop(); }
|
|
});
|
|
|
|
it("aborts on disconnect and uses a fresh reconnect baseline without replaying old commands", async () => {
|
|
const oldDisplay = deferred<void>();
|
|
const execute = vi.fn(async (_command: InvestigationPresentationCommand, _signal: AbortSignal) => {})
|
|
.mockImplementationOnce(() => oldDisplay.promise);
|
|
const fetchCursor = vi.fn(async () => cursor(10)).mockResolvedValueOnce(cursor(10)).mockResolvedValueOnce(cursor(20));
|
|
const test = setup({ execute, fetchCursor });
|
|
test.follower.start();
|
|
test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
|
test.channel.publish({ type: "events", events: [event(11)] });
|
|
const oldSignal = execute.mock.calls[0]?.[1];
|
|
test.channel.publish({ type: "disconnect" });
|
|
expect(oldSignal?.aborted).toBe(true);
|
|
const clearCount = test.clear.mock.calls.length;
|
|
test.channel.publish({ type: "events", events: [event(12, openChart)] });
|
|
test.channel.publish({ type: "open" });
|
|
await vi.waitFor(() => expect(fetchCursor).toHaveBeenCalledTimes(2));
|
|
test.channel.publish({ type: "events", events: [event(12, openChart), event(20, openChart), event(21, closeChart)] });
|
|
oldDisplay.resolve();
|
|
await vi.waitFor(() => expect(execute.mock.calls.map(([command]) => command.action)).toEqual(["highlight_features", "close_chart"]));
|
|
expect(test.clear.mock.calls.length).toBe(clearCount);
|
|
expect(execute.mock.calls[1]?.[1].aborted).toBe(false);
|
|
} finally { oldDisplay.resolve(); test.follower.stop(); }
|
|
});
|
|
});
|
|
|
|
|
|
it("buffers live events while saved history is loading and drains only after that boundary", async () => {
|
|
const ready = deferred<void>();
|
|
const loadHistory = vi.fn(async () => ready.promise);
|
|
const test = setup({ loadHistory });
|
|
test.follower.start(); test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(loadHistory).toHaveBeenCalledTimes(1));
|
|
test.channel.publish({ type: "events", events: [event(11, openChart)] });
|
|
expect(test.execute).not.toHaveBeenCalled();
|
|
ready.resolve();
|
|
await vi.waitFor(() => expect(test.execute).toHaveBeenCalledTimes(1));
|
|
} finally { test.follower.stop(); }
|
|
});
|
|
|
|
it("loads saved history without live following and aborts an unfinished history load on stop", async () => {
|
|
const ready = deferred<void>();
|
|
const loadHistory = vi.fn(async (_cursor: InvestigationPresentationCursor, _signal: AbortSignal) => ready.promise);
|
|
const subscribe = vi.fn();
|
|
const test = setup({ follow: false, loadHistory, subscribe });
|
|
test.follower.start();
|
|
await vi.waitFor(() => expect(loadHistory).toHaveBeenCalledTimes(1));
|
|
expect(subscribe).not.toHaveBeenCalled();
|
|
test.follower.stop();
|
|
expect(loadHistory.mock.calls[0][1].aborted).toBe(true);
|
|
ready.resolve();
|
|
expect(test.execute).not.toHaveBeenCalled();
|
|
});
|
|
|
|
|
|
it("reads saved history even before the live event stream opens", async () => {
|
|
const loadHistory = vi.fn(async () => undefined);
|
|
const test = setup({ loadHistory });
|
|
test.follower.start();
|
|
try {
|
|
await vi.waitFor(() => expect(loadHistory).toHaveBeenCalledTimes(1));
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("history"));
|
|
expect(test.execute).not.toHaveBeenCalled();
|
|
} finally { test.follower.stop(); }
|
|
});
|
|
|
|
|
|
it("keeps successful results across disconnect and failed reconnect, clearing only for a new identity", async () => {
|
|
let visible = "saved chart";
|
|
const clear = vi.fn(() => { visible = ""; });
|
|
const fetchCursor = vi.fn(async () => cursor(10));
|
|
const test = setup({ clear, fetchCursor });
|
|
test.follower.start(); test.channel.publish({ type: "open" });
|
|
try {
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("live"));
|
|
visible = "saved chart";
|
|
clear.mockClear();
|
|
test.channel.publish({ type: "disconnect" });
|
|
expect(visible).toBe("saved chart");
|
|
expect(clear).not.toHaveBeenCalled();
|
|
fetchCursor.mockRejectedValueOnce(new Error("offline"));
|
|
test.channel.publish({ type: "open" });
|
|
await vi.waitFor(() => expect(test.onStatus).toHaveBeenCalledWith("error", "offline"));
|
|
expect(visible).toBe("saved chart");
|
|
fetchCursor.mockResolvedValueOnce(cursor(20, 2));
|
|
test.channel.publish({ type: "open" });
|
|
await vi.waitFor(() => expect(clear).toHaveBeenCalledTimes(1));
|
|
expect(visible).toBe("");
|
|
} finally { test.follower.stop(); }
|
|
});
|
|
|
|
it("clears a changed generation before hydrating its saved results and ignores a stale connection", async () => {
|
|
const stale = deferred<InvestigationPresentationCursor>();
|
|
const order: string[] = [];
|
|
const fetchCursor = vi.fn(async () => cursor(10))
|
|
.mockResolvedValueOnce(cursor(10))
|
|
.mockImplementationOnce(() => stale.promise)
|
|
.mockResolvedValueOnce(cursor(30, 2));
|
|
const test = setup({ fetchCursor, clear: () => order.push("clear"), loadHistory: async (next) => { order.push(`history:${next.execution_generation}`); } });
|
|
test.follower.start();
|
|
try {
|
|
await vi.waitFor(() => expect(order).toContain("history:1"));
|
|
order.length = 0;
|
|
test.channel.publish({ type: "open" });
|
|
test.channel.publish({ type: "disconnect" });
|
|
test.channel.publish({ type: "open" });
|
|
await vi.waitFor(() => expect(order).toContain("history:2"));
|
|
expect(order).toEqual(["clear", "history:2"]);
|
|
stale.resolve(cursor(20));
|
|
await Promise.resolve();
|
|
expect(order).toEqual(["clear", "history:2"]);
|
|
} finally { stale.resolve(cursor(20)); test.follower.stop(); }
|
|
});
|