authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-25 13:13:11-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
loga973e43421c9829a7450a7fe06d5f0e81c1b98c1
tree86ceed1ec903d8137ea889bcf8cbb1086f90353c
parentc4ff23449bdee234293fd3e94ed38ac442670346

Validate inbound WebSocket push payloads before dispatch

- Decode `server.welcome`, terminal, and orchestration push events with Effect schemas - Drop malformed payloads with warning logs instead of forwarding invalid data - Add wsNativeApi tests for valid forwarding, caching, and malformed-payload handling

2 files changed, 212 insertions(+), 10 deletions(-)

apps/web/src/wsNativeApi.test.ts+182-3
...@@ -1,8 +1,28 @@...@@ -1,8 +1,28 @@
1import { CommandId, ORCHESTRATION_WS_METHODS, ProjectId, ThreadId } from "@t3tools/contracts";1import {
2import { describe, expect, it, vi } from "vitest";2 CommandId,
3 ORCHESTRATION_WS_CHANNELS,
4 ORCHESTRATION_WS_METHODS,
5 ProjectId,
6 ThreadId,
7 WS_CHANNELS,
8} from "@t3tools/contracts";
9import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
310
4const requestMock = vi.fn<(...args: Array<unknown>) => Promise<unknown>>();11const requestMock = vi.fn<(...args: Array<unknown>) => Promise<unknown>>();
5const subscribeMock = vi.fn<(...args: Array<unknown>) => () => void>(() => () => undefined);12const channelListeners = new Map<string, Set<(data: unknown) => void>>();
13const subscribeMock = vi.fn<(channel: string, listener: (data: unknown) => void) => () => void>(
14 (channel, listener) => {
15 const listeners = channelListeners.get(channel) ?? new Set<(data: unknown) => void>();
16 listeners.add(listener);
17 channelListeners.set(channel, listeners);
18 return () => {
19 listeners.delete(listener);
20 if (listeners.size === 0) {
21 channelListeners.delete(channel);
22 }
23 };
24 },
25);
626
7vi.mock("./wsTransport", () => {27vi.mock("./wsTransport", () => {
8 return {28 return {
...@@ -13,7 +33,166 @@ vi.mock("./wsTransport", () => {...@@ -13,7 +33,166 @@ vi.mock("./wsTransport", () => {
13 };33 };
14});34});
1535
36function emitPush(channel: string, data: unknown): void {
37 const listeners = channelListeners.get(channel);
38 if (!listeners) return;
39 for (const listener of listeners) {
40 listener(data);
41 }
42}
43
44beforeEach(() => {
45 vi.resetModules();
46 requestMock.mockReset();
47 subscribeMock.mockClear();
48 channelListeners.clear();
49});
50
51afterEach(() => {
52 vi.restoreAllMocks();
53});
54
16describe("wsNativeApi", () => {55describe("wsNativeApi", () => {
56 it("delivers and caches valid server.welcome payloads", async () => {
57 const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
58 const { createWsNativeApi, onServerWelcome } = await import("./wsNativeApi");
59
60 createWsNativeApi();
61 const listener = vi.fn();
62 onServerWelcome(listener);
63
64 const payload = { cwd: "/tmp/workspace", projectName: "t3-code" };
65 emitPush(WS_CHANNELS.serverWelcome, payload);
66
67 expect(listener).toHaveBeenCalledTimes(1);
68 expect(listener).toHaveBeenCalledWith(payload);
69
70 const lateListener = vi.fn();
71 onServerWelcome(lateListener);
72
73 expect(lateListener).toHaveBeenCalledTimes(1);
74 expect(lateListener).toHaveBeenCalledWith(payload);
75 expect(warnSpy).not.toHaveBeenCalled();
76 });
77
78 it("ignores invalid server.welcome payloads and keeps subscription active", async () => {
79 const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
80 const { createWsNativeApi, onServerWelcome } = await import("./wsNativeApi");
81
82 createWsNativeApi();
83 const listener = vi.fn();
84 onServerWelcome(listener);
85
86 emitPush(WS_CHANNELS.serverWelcome, { cwd: 42, projectName: "t3-code" });
87 emitPush(WS_CHANNELS.serverWelcome, { cwd: "/tmp/workspace", projectName: "t3-code" });
88
89 expect(listener).toHaveBeenCalledTimes(1);
90 expect(listener).toHaveBeenCalledWith({ cwd: "/tmp/workspace", projectName: "t3-code" });
91 expect(warnSpy).toHaveBeenCalledTimes(1);
92 expect(warnSpy).toHaveBeenCalledWith("Dropped inbound WebSocket push payload", {
93 reason: "decode-failed",
94 raw: { cwd: 42, projectName: "t3-code" },
95 issue: expect.stringContaining("SchemaError"),
96 });
97 });
98
99 it("forwards valid terminal and orchestration events", async () => {
100 const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
101 const { createWsNativeApi } = await import("./wsNativeApi");
102
103 const api = createWsNativeApi();
104 const onTerminalEvent = vi.fn();
105 const onDomainEvent = vi.fn();
106
107 api.terminal.onEvent(onTerminalEvent);
108 api.orchestration.onDomainEvent(onDomainEvent);
109
110 const terminalEvent = {
111 threadId: "thread-1",
112 terminalId: "terminal-1",
113 createdAt: "2026-02-24T00:00:00.000Z",
114 type: "output",
115 data: "hello",
116 } as const;
117 emitPush(WS_CHANNELS.terminalEvent, terminalEvent);
118
119 const orchestrationEvent = {
120 sequence: 1,
121 eventId: "event-1",
122 aggregateKind: "project",
123 aggregateId: "project-1",
124 occurredAt: "2026-02-24T00:00:00.000Z",
125 commandId: null,
126 causationEventId: null,
127 correlationId: null,
128 metadata: {},
129 type: "project.created",
130 payload: {
131 projectId: "project-1",
132 title: "Project",
133 workspaceRoot: "/tmp/workspace",
134 defaultModel: null,
135 scripts: [],
136 createdAt: "2026-02-24T00:00:00.000Z",
137 updatedAt: "2026-02-24T00:00:00.000Z",
138 },
139 } as const;
140 emitPush(ORCHESTRATION_WS_CHANNELS.domainEvent, orchestrationEvent);
141
142 expect(onTerminalEvent).toHaveBeenCalledTimes(1);
143 expect(onTerminalEvent).toHaveBeenCalledWith(terminalEvent);
144 expect(onDomainEvent).toHaveBeenCalledTimes(1);
145 expect(onDomainEvent).toHaveBeenCalledWith(orchestrationEvent);
146 expect(warnSpy).not.toHaveBeenCalled();
147 });
148
149 it("drops malformed terminal and orchestration push payloads", async () => {
150 const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
151 const { createWsNativeApi } = await import("./wsNativeApi");
152
153 const api = createWsNativeApi();
154 const onTerminalEvent = vi.fn();
155 const onDomainEvent = vi.fn();
156
157 api.terminal.onEvent(onTerminalEvent);
158 api.orchestration.onDomainEvent(onDomainEvent);
159
160 emitPush(WS_CHANNELS.terminalEvent, {
161 threadId: "thread-1",
162 terminalId: "",
163 createdAt: "2026-02-24T00:00:00.000Z",
164 type: "output",
165 data: "hello",
166 });
167 emitPush(ORCHESTRATION_WS_CHANNELS.domainEvent, {
168 sequence: -1,
169 type: "project.created",
170 });
171
172 expect(onTerminalEvent).not.toHaveBeenCalled();
173 expect(onDomainEvent).not.toHaveBeenCalled();
174 expect(warnSpy).toHaveBeenCalledTimes(2);
175 expect(warnSpy).toHaveBeenNthCalledWith(1, "Dropped inbound WebSocket push payload", {
176 reason: "decode-failed",
177 raw: {
178 threadId: "thread-1",
179 terminalId: "",
180 createdAt: "2026-02-24T00:00:00.000Z",
181 type: "output",
182 data: "hello",
183 },
184 issue: expect.stringContaining("SchemaError"),
185 });
186 expect(warnSpy).toHaveBeenNthCalledWith(2, "Dropped inbound WebSocket push payload", {
187 reason: "decode-failed",
188 raw: {
189 sequence: -1,
190 type: "project.created",
191 },
192 issue: expect.stringContaining("SchemaError"),
193 });
194 });
195
17 it("wraps orchestration dispatch commands in the command envelope", async () => {196 it("wraps orchestration dispatch commands in the command envelope", async () => {
18 requestMock.mockResolvedValue(undefined);197 requestMock.mockResolvedValue(undefined);
19 const { createWsNativeApi } = await import("./wsNativeApi");198 const { createWsNativeApi } = await import("./wsNativeApi");
apps/web/src/wsNativeApi.ts+30-7
...@@ -1,11 +1,14 @@...@@ -1,11 +1,14 @@
1import {1import {
2 OrchestrationEvent,
2 ORCHESTRATION_WS_CHANNELS,3 ORCHESTRATION_WS_CHANNELS,
3 ORCHESTRATION_WS_METHODS,4 ORCHESTRATION_WS_METHODS,
4 type NativeApi,5 type NativeApi,
6 TerminalEvent,
5 WS_CHANNELS,7 WS_CHANNELS,
6 WS_METHODS,8 WS_METHODS,
7 type WsWelcomePayload,9 WsWelcomePayload,
8} from "@t3tools/contracts";10} from "@t3tools/contracts";
11import { Cause, Schema } from "effect";
912
10import { showContextMenuFallback } from "./contextMenuFallback";13import { showContextMenuFallback } from "./contextMenuFallback";
11import { WsTransport } from "./wsTransport";14import { WsTransport } from "./wsTransport";
...@@ -14,6 +17,22 @@ let instance: { api: NativeApi; transport: WsTransport } | null = null;...@@ -14,6 +17,22 @@ let instance: { api: NativeApi; transport: WsTransport } | null = null;
14const welcomeListeners = new Set<(payload: WsWelcomePayload) => void>();17const welcomeListeners = new Set<(payload: WsWelcomePayload) => void>();
15let lastWelcome: WsWelcomePayload | null = null;18let lastWelcome: WsWelcomePayload | null = null;
1619
20const decodeAndWarnOnFailure = <T>(
21 schema: Schema.Schema<T> & { readonly DecodingServices: never },
22 raw: unknown,
23): T | null => {
24 const decoded = Schema.decodeUnknownExit(schema)(raw);
25 if (decoded._tag === "Failure") {
26 console.warn("Dropped inbound WebSocket push payload", {
27 reason: "decode-failed",
28 raw,
29 issue: Cause.pretty(decoded.cause),
30 });
31 return null;
32 }
33 return decoded.value;
34};
35
17/**36/**
18 * Subscribe to the server welcome message. If a welcome was already received37 * Subscribe to the server welcome message. If a welcome was already received
19 * before this call, the listener fires synchronously with the cached payload.38 * before this call, the listener fires synchronously with the cached payload.
...@@ -44,7 +63,8 @@ export function createWsNativeApi(): NativeApi {...@@ -44,7 +63,8 @@ export function createWsNativeApi(): NativeApi {
44 // Listen for server welcome and forward to registered listeners.63 // Listen for server welcome and forward to registered listeners.
45 // Also cache it so late subscribers (React effects) get it immediately.64 // Also cache it so late subscribers (React effects) get it immediately.
46 transport.subscribe(WS_CHANNELS.serverWelcome, (data) => {65 transport.subscribe(WS_CHANNELS.serverWelcome, (data) => {
47 const payload = data as WsWelcomePayload;66 const payload = decodeAndWarnOnFailure(WsWelcomePayload, data);
67 if (!payload) return;
48 lastWelcome = payload;68 lastWelcome = payload;
49 for (const listener of welcomeListeners) {69 for (const listener of welcomeListeners) {
50 try {70 try {
...@@ -76,7 +96,10 @@ export function createWsNativeApi(): NativeApi {...@@ -76,7 +96,10 @@ export function createWsNativeApi(): NativeApi {
76 restart: (input) => transport.request(WS_METHODS.terminalRestart, input),96 restart: (input) => transport.request(WS_METHODS.terminalRestart, input),
77 close: (input) => transport.request(WS_METHODS.terminalClose, input),97 close: (input) => transport.request(WS_METHODS.terminalClose, input),
78 onEvent: (callback) =>98 onEvent: (callback) =>
79 transport.subscribe(WS_CHANNELS.terminalEvent, callback as (data: unknown) => void),99 transport.subscribe(WS_CHANNELS.terminalEvent, (data) => {
100 const payload = decodeAndWarnOnFailure(TerminalEvent, data);
101 if (payload) callback(payload);
102 }),
80 },103 },
81 projects: {104 projects: {
82 searchEntries: (input) => transport.request(WS_METHODS.projectsSearchEntries, input),105 searchEntries: (input) => transport.request(WS_METHODS.projectsSearchEntries, input),
...@@ -135,10 +158,10 @@ export function createWsNativeApi(): NativeApi {...@@ -135,10 +158,10 @@ export function createWsNativeApi(): NativeApi {
135 replayEvents: (fromSequenceExclusive) =>158 replayEvents: (fromSequenceExclusive) =>
136 transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }),159 transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }),
137 onDomainEvent: (callback) =>160 onDomainEvent: (callback) =>
138 transport.subscribe(161 transport.subscribe(ORCHESTRATION_WS_CHANNELS.domainEvent, (data) => {
139 ORCHESTRATION_WS_CHANNELS.domainEvent,162 const payload = decodeAndWarnOnFailure(OrchestrationEvent, data);
140 callback as (data: unknown) => void,163 if (payload) callback(payload);
141 ),164 }),
142 },165 },
143 };166 };
144167