diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message-payload-guards/index.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message-payload-guards/index.ts new file mode 100644 index 00000000000..3ab759d6ce9 --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message-payload-guards/index.ts @@ -0,0 +1,33 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import type { MessageReceiveDataTypeMap } from '../interfaces/message-data-type-map.interface'; +import { OP } from '../interfaces/message-operator.interface'; + +import { isListUpdateNoteJobsPayload } from './job'; + +export type MessagePayloadGuard = (value: unknown) => boolean; + +type ReceiveOP = keyof MessageReceiveDataTypeMap; + +/** + * Runtime payload guards are registered only for OPs with a demonstrated + * payload-shape failure. Add new guards when a concrete runtime failure + * shows that validation is needed. + */ +const MESSAGE_PAYLOAD_GUARDS: Partial> = { + [OP.LIST_UPDATE_NOTE_JOBS]: isListUpdateNoteJobsPayload +}; + +export const getMessagePayloadGuard = (op: ReceiveOP): MessagePayloadGuard | undefined => { + return MESSAGE_PAYLOAD_GUARDS[op]; +}; diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message-payload-guards/job.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message-payload-guards/job.ts new file mode 100644 index 00000000000..f3edecd093b --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message-payload-guards/job.ts @@ -0,0 +1,47 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +const isRecord = (value: unknown): value is Record => typeof value === 'object' && value !== null; + +const isJobUpdate = (value: unknown): boolean => { + if (!isRecord(value)) { + return false; + } + + if (typeof value.noteId !== 'string') { + return false; + } + + if (typeof value.isRemoved !== 'boolean') { + return false; + } + + if (value.isRemoved) { + return true; + } + + return typeof value.noteName === 'string'; +}; + +export const isListUpdateNoteJobsPayload = (value: unknown): boolean => { + if (!isRecord(value)) { + return false; + } + + const noteRunningJobs = value.noteRunningJobs; + + if (!isRecord(noteRunningJobs)) { + return false; + } + + return Array.isArray(noteRunningJobs.jobs) && noteRunningJobs.jobs.every(isJobUpdate); +}; diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts new file mode 100644 index 00000000000..256c30029e0 --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.spec.ts @@ -0,0 +1,136 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { describe, expect, it, vi } from 'vitest'; + +import type { MessageReceiveDataTypeMap } from './interfaces/message-data-type-map.interface'; +import { OP } from './interfaces/message-operator.interface'; +import type { WebSocketMessage } from './interfaces/websocket-message.interface'; +import { Message } from './message'; + +const asReceivedMessage = (message: unknown): WebSocketMessage => + message as WebSocketMessage; + +describe('Message.receive', () => { + it('passes a non-removal job update with noteName', () => { + const message = new Message(); + const listener = vi.fn(); + const data = { + noteRunningJobs: { + jobs: [ + { + noteId: 'note-1', + noteName: 'Test Note', + isRemoved: false + } + ] + } + }; + + message.receive(OP.LIST_UPDATE_NOTE_JOBS).subscribe(listener); + + message.shortCircuit( + asReceivedMessage({ + op: OP.LIST_UPDATE_NOTE_JOBS, + data + }) + ); + + expect(listener).toHaveBeenCalledWith(data); + }); + + it('passes a partial removal payload without noteName', () => { + const message = new Message(); + const listener = vi.fn(); + const data = { + noteRunningJobs: { + jobs: [ + { + noteId: 'note-1', + isRemoved: true + } + ] + } + }; + + message.receive(OP.LIST_UPDATE_NOTE_JOBS).subscribe(listener); + + message.shortCircuit( + asReceivedMessage({ + op: OP.LIST_UPDATE_NOTE_JOBS, + data + }) + ); + + expect(listener).toHaveBeenCalledWith(data); + }); + + it('filters a non-removal job update without noteName', () => { + const message = new Message(); + const listener = vi.fn(); + + message.receive(OP.LIST_UPDATE_NOTE_JOBS).subscribe(listener); + + message.shortCircuit( + asReceivedMessage({ + op: OP.LIST_UPDATE_NOTE_JOBS, + data: { + noteRunningJobs: { + jobs: [ + { + noteId: 'note-1', + isRemoved: false + } + ] + } + } + }) + ); + + expect(listener).not.toHaveBeenCalled(); + }); + + it('filters a payload without a jobs array', () => { + const message = new Message(); + const listener = vi.fn(); + + message.receive(OP.LIST_UPDATE_NOTE_JOBS).subscribe(listener); + + message.shortCircuit( + asReceivedMessage({ + op: OP.LIST_UPDATE_NOTE_JOBS, + data: { + noteRunningJobs: {} + } + }) + ); + + expect(listener).not.toHaveBeenCalled(); + }); + + it('keeps existing behavior for an OP without a guard', () => { + const message = new Message(); + const listener = vi.fn(); + const data = {}; + + message.receive(OP.NOTE).subscribe(listener); + + message.shortCircuit( + asReceivedMessage({ + op: OP.NOTE, + data + }) + ); + + expect(listener).toHaveBeenCalledWith(data); + }); +}); diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts index 0f070c6354f..6274c97f5ba 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/message.ts @@ -30,6 +30,8 @@ import { } from './interfaces/message-paragraph.interface'; import { WebSocketMessage } from './interfaces/websocket-message.interface'; +import { getMessagePayloadGuard } from './message-payload-guards'; + export type ArgumentsType = T extends (...args: infer U) => void ? U : never; export type SendArgumentsType = MessageSendDataTypeMap[K] extends undefined @@ -175,6 +177,15 @@ export class Message { receive(op: K): Observable[K]> { return this.received$.pipe( filter(message => message.op === op), + filter(message => { + const guard = getMessagePayloadGuard(op); + + if (!guard) { + return true; + } + + return guard(message.data); + }), map(message => message.data) ) as Observable[K]>; } diff --git a/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts b/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts new file mode 100644 index 00000000000..b5dcf91563a --- /dev/null +++ b/zeppelin-web-angular/src/app/core/message-listener/message-listener.spec.ts @@ -0,0 +1,85 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { Subject } from 'rxjs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { Message, OP, MessageReceiveDataTypeMap } from '@zeppelin/sdk'; + +import { MessageListener, MessageListenersManager } from './message-listener'; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('MessageListener', () => { + it('logs handler errors with the OP and keeps the subscription active', () => { + const received$ = new Subject(); + const messageService = { + receive: vi.fn(() => received$.asObservable()) + } as unknown as Message; + + const error = new Error('boom'); + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}); + + class TestComponent extends MessageListenersManager { + calls = 0; + + handleNote(_data: MessageReceiveDataTypeMap[OP.NOTE]): void { + this.calls++; + + if (this.calls === 1) { + throw error; + } + } + } + + const descriptor = Object.getOwnPropertyDescriptor(TestComponent.prototype, 'handleNote')!; + + MessageListener(OP.NOTE)(TestComponent.prototype, 'handleNote', descriptor); + + const component = new TestComponent(messageService); + const data = {} as MessageReceiveDataTypeMap[OP.NOTE]; + + received$.next(data); + received$.next(data); + + expect(component.calls).toBe(2); + expect(consoleError).toHaveBeenCalledWith(`Failed to handle WebSocket OP ${String(OP.NOTE)}`, error); + }); + + it('passes received data to the handler', () => { + const received$ = new Subject(); + const messageService = { + receive: vi.fn(() => received$.asObservable()) + } as unknown as Message; + + class TestComponent extends MessageListenersManager { + receivedData?: MessageReceiveDataTypeMap[OP.NOTE]; + + handleNote(data: MessageReceiveDataTypeMap[OP.NOTE]): void { + this.receivedData = data; + } + } + + const descriptor = Object.getOwnPropertyDescriptor(TestComponent.prototype, 'handleNote')!; + + MessageListener(OP.NOTE)(TestComponent.prototype, 'handleNote', descriptor); + + const component = new TestComponent(messageService); + const data = {} as MessageReceiveDataTypeMap[OP.NOTE]; + + received$.next(data); + + expect(component.receivedData).toBe(data); + }); +}); diff --git a/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts b/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts index 6487124ecc7..c05a06f5d55 100644 --- a/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts +++ b/zeppelin-web-angular/src/app/core/message-listener/message-listener.ts @@ -49,8 +49,12 @@ export function MessageListener(op: K this.__zeppelinMessageListeners$__.add( this.messageService.receive(op).subscribe(data => { - // @ts-ignore - oldValue.apply(this, [data]); + try { + // @ts-ignore + oldValue.apply(this, [data]); + } catch (error) { + console.error(`Failed to handle WebSocket OP ${String(op)}`, error); + } }) ); }; diff --git a/zeppelin-web-angular/vitest.shell.config.mts b/zeppelin-web-angular/vitest.shell.config.mts index 0c035c2b2b6..531f42e24ba 100644 --- a/zeppelin-web-angular/vitest.shell.config.mts +++ b/zeppelin-web-angular/vitest.shell.config.mts @@ -11,9 +11,15 @@ */ // vite is pinned in package.json: the React remote keeps its own lockfile and drifted to a different minor. +import path from 'node:path'; import { defineConfig } from 'vitest/config'; export default defineConfig({ + resolve: { + alias: { + '@zeppelin/sdk': path.resolve(__dirname, './projects/zeppelin-sdk/src/public-api.ts') + } + }, // oxc does not apply the decorator options from tsconfig.base.json to specs, // which src/tsconfig.json excludes. Undeclared, a decorated spec fails to // parse with "Invalid or unexpected token".