diff --git a/app/components/Screenshot.vue b/app/components/Screenshot.vue index cfd6b476..06264627 100644 --- a/app/components/Screenshot.vue +++ b/app/components/Screenshot.vue @@ -44,7 +44,7 @@ async function takeScreenshot(): Promise { params, }, { - // `blob` is a wslink attachment: the viewer sends its key, the client receives the binary in its place + // `blob` is the image file content, received as binary (Uint8Array) through wslink response_function: async ({ blob }) => { if (screenshot_type.value === "file") { fileDownload(blob, `${current_filename}.${output_extension.value}`); diff --git a/app/composables/clipping_planes_widget_entries.ts b/app/composables/clipping_planes_widget_entries.ts index bd03b96b..11dbd627 100644 --- a/app/composables/clipping_planes_widget_entries.ts +++ b/app/composables/clipping_planes_widget_entries.ts @@ -47,6 +47,11 @@ function asWidgetFactory(widget: PlaneWidget): vtkAbstractWidgetFactory; } +function asWidgetHandle(widget: unknown): WidgetHandle { + // oxlint-disable-next-line no-unsafe-type-assertion -- vtk.js WidgetManager.addWidget is untyped for our custom WidgetHandle shape; narrowed here. + return widget as WidgetHandle; +} + interface WidgetEntryManagerParams { planes: Ref; debouncedApply: (...args: readonly unknown[]) => void; @@ -136,10 +141,7 @@ function useWidgetEntryManager({ planes, debouncedApply }: WidgetEntryManagerPar if (!widgetEntries[idx]) { // oxlint-disable-next-line no-unsafe-call no-unsafe-type-assertion -- vtk.js has no types for ImplicitPlaneWidget's factory; narrowed here. const planeWidget = vtkImplicitPlaneWidget() as PlaneWidget; - const widgetHandle = widgetManager.addWidget( - asWidgetFactory(planeWidget), - // oxlint-disable-next-line no-unsafe-type-assertion -- vtk.js WidgetManager.addWidget is untyped for our custom WidgetHandle shape; narrowed here. - ) as unknown as WidgetHandle; + const widgetHandle = asWidgetHandle(widgetManager.addWidget(asWidgetFactory(planeWidget))); widgetEntries.push(createWidgetEntry(planeWidget, widgetHandle, idx)); } const entry = widgetEntries[idx]; diff --git a/app/stores/camera_manager.ts b/app/stores/camera_manager.ts index 19f52406..49968e1e 100644 --- a/app/stores/camera_manager.ts +++ b/app/stores/camera_manager.ts @@ -1,10 +1,10 @@ -import { type Table, liveQuery } from "dexie"; -import type { Observable } from "rxjs"; +import type { Table } from "dexie"; import { useObservable } from "@vueuse/rxjs"; // Local imports import type { CameraOptions } from "@ogw_internal/stores/hybrid_viewer/vtk_types.js"; import { database } from "@ogw_internal/database/database.js"; +import { liveObservable } from "@ogw_front/utils/live_observable"; import { useViewerStore } from "@ogw_front/stores/viewer"; import viewer_schemas from "@geode/opengeodeweb-viewer/opengeodeweb_viewer_typed_schemas.js"; @@ -26,16 +26,11 @@ export const useCameraManagerStore = defineStore("camera_manager", () => { >; function refAllCameraPositions(): Ref { - // Dexie's liveQuery() returns Dexie's own minimal Observable shape, not an - // Actual rxjs Observable instance (useObservable's declared parameter type); - // The two are structurally close enough at runtime (vueuse only calls - // `.subscribe`) but not identical, hence the cast. return useObservable( - liveQuery(async () => { + liveObservable(async () => { const positions = await camera_positions_db.toArray(); return positions; - // oxlint-disable-next-line no-unsafe-type-assertion -- trusted vueuse/Dexie Observable boundary; see comment above. - }) as unknown as Observable, + }), { initialValue: [] as CameraPositionRecord[] }, ); } diff --git a/app/stores/data.ts b/app/stores/data.ts index 3bfd72da..cd2058ec 100644 --- a/app/stores/data.ts +++ b/app/stores/data.ts @@ -1,13 +1,13 @@ // oxlint-disable eslint/max-lines // Third party imports -import { type Table, liveQuery } from "dexie"; -import type { Observable } from "rxjs"; +import type { Table } from "dexie"; import { useObservable } from "@vueuse/rxjs"; import viewer_schemas from "@geode/opengeodeweb-viewer/opengeodeweb_viewer_typed_schemas.js"; // Local imports import { type ModelComponentRecord, useDataMesh } from "./data_helpers/mesh.js"; import { database } from "@ogw_internal/database/database.js"; +import { liveObservable } from "@ogw_front/utils/live_observable"; import { useDataCollections } from "./data_helpers/collections.js"; import { useViewerStore } from "@ogw_front/stores/viewer"; @@ -137,16 +137,11 @@ export const useDataStore = defineStore("data", () => { return items; } function refItem(id: string): Ref { - // Dexie's liveQuery() returns Dexie's own minimal Observable shape, not an - // Actual rxjs Observable instance (useObservable's declared parameter type); - // The two are structurally close enough at runtime (vueuse only calls - // `.subscribe`) but not identical, hence the cast. return useObservable( - liveQuery(async () => { + liveObservable(async () => { const data_item = await data_db.get(id); return data_item; - // oxlint-disable-next-line no-unsafe-type-assertion -- trusted vueuse/Dexie Observable boundary; see comment above. - }) as unknown as Observable, + }), { // oxlint-disable-next-line no-unsafe-type-assertion -- placeholder until the live query resolves; consumers must treat this as possibly incomplete. initialValue: {} as DataItem, @@ -155,11 +150,10 @@ export const useDataStore = defineStore("data", () => { } function refAllItems(): Ref { return useObservable( - liveQuery(async () => { + liveObservable(async () => { const items = await data_db.toArray(); return items; - // oxlint-disable-next-line no-unsafe-type-assertion -- trusted vueuse/Dexie Observable boundary; see comment above. - }) as unknown as Observable, + }), { initialValue: [] as DataItem[], }, diff --git a/app/stores/data_helpers/collections.ts b/app/stores/data_helpers/collections.ts index a40d2270..847d36f9 100644 --- a/app/stores/data_helpers/collections.ts +++ b/app/stores/data_helpers/collections.ts @@ -1,9 +1,8 @@ import { type FormattedComponent, type ModelComponentRecord, useDataMesh } from "./mesh.js"; import { MESH_COMPONENT_TYPES } from "@ogw_front/utils/default_styles"; -import type { Observable } from "rxjs"; import type { Ref } from "vue"; import { getTable } from "@ogw_internal/database/database.js"; -import { liveQuery } from "dexie"; +import { liveObservable } from "@ogw_front/utils/live_observable"; import { useObservable } from "@vueuse/rxjs"; interface ModelComponentRelationRecord { @@ -115,12 +114,10 @@ export function useDataCollections(): { modelId: string, ): Ref { return useObservable( - // Dexie's liveQuery returns its own Observable-like type, not rxjs's Observable, so bridging needs a cast. - liveQuery(async () => { + liveObservable(async () => { const groups = await formatedCollectionComponents(modelId); return groups; - // oxlint-disable-next-line no-unsafe-type-assertion - }) as unknown as Observable, + }), { initialValue: undefined, }, diff --git a/app/stores/data_helpers/mesh.ts b/app/stores/data_helpers/mesh.ts index 07f07e0b..fbcd228f 100644 --- a/app/stores/data_helpers/mesh.ts +++ b/app/stores/data_helpers/mesh.ts @@ -1,6 +1,6 @@ -import { type Table, liveQuery } from "dexie"; -import type { Observable } from "rxjs"; +import type { Table } from "dexie"; import { database } from "@ogw_internal/database/database.js"; +import { liveObservable } from "@ogw_front/utils/live_observable"; import { useObservable } from "@vueuse/rxjs"; interface ModelComponentRecord { @@ -87,11 +87,10 @@ export function useDataMesh(): { function refFormatedMeshComponents(modelId: string): Ref { return useObservable( - liveQuery(async () => { + liveObservable(async () => { const components = await formatedMeshComponents(modelId); return components; - // oxlint-disable-next-line no-unsafe-type-assertion - }) as unknown as Observable, + }), { initialValue: undefined, }, diff --git a/app/stores/viewer.ts b/app/stores/viewer.ts index c5265e4a..c81a34dd 100644 --- a/app/stores/viewer.ts +++ b/app/stores/viewer.ts @@ -142,12 +142,13 @@ export const useViewerStore = defineStore( } try { status.value = Status.CONNECTING; - client.value = (await initWebSocketClient(base_url.value, client.value, { + const wsClient = await initWebSocketClient(base_url.value, client.value, { onConnectionClose: () => { status.value = Status.NOT_CONNECTED; }, - // oxlint-disable-next-line no-unsafe-type-assertion -- initWebSocketClient's return is not typed as RpcClient. - })) as unknown as RpcClient; + }); + // oxlint-disable-next-line no-unsafe-type-assertion -- initWebSocketClient's return is not typed as RpcClient. + client.value = wsClient as unknown as RpcClient; connectImageStream(client.value.getConnection().getSession()); // oxlint-disable-next-line no-unsafe-type-assertion -- endBusy is not part of the RpcClient type. (client.value as unknown as { endBusy: () => void }).endBusy(); diff --git a/app/utils/live_observable.ts b/app/utils/live_observable.ts new file mode 100644 index 00000000..5872702f --- /dev/null +++ b/app/utils/live_observable.ts @@ -0,0 +1,14 @@ +// Third party imports +import type { Observable } from "rxjs"; +import { liveQuery } from "dexie"; + +// Dexie's liveQuery() returns Dexie's own minimal Observable shape, not an +// Actual rxjs Observable instance (useObservable's declared parameter type); +// The two are structurally close enough at runtime (vueuse only calls +// `.subscribe`) but not identical, hence the cast. +function liveObservable(query: () => Promise): Observable { + // oxlint-disable-next-line no-unsafe-type-assertion -- trusted vueuse/Dexie Observable boundary; see comment above. + return liveQuery(query) as unknown as Observable; +} + +export { liveObservable }; diff --git a/package.json b/package.json index ddea41cf..51f65e0f 100644 --- a/package.json +++ b/package.json @@ -69,6 +69,7 @@ "@geode/opengeodeweb-viewer": "0.0.0", "@kitware/vtk.js": "33.3.0", "@mdi/font": "7.4.47", + "@msgpack/msgpack": "2.8.0", "@pinia/nuxt": "1.0.1", "@types/busboy": "1.5.4", "@types/node": "22.20.3", @@ -101,7 +102,7 @@ "vuetify": "4.2.0", "vuetify-nuxt-module": "1.0.0-rc.5", "ws": "8.21.2", - "wslink": "1.12.4", + "wslink": "2.5.0", "zod": "4.4.3" }, "devDependencies": { diff --git a/server/utils/cleanup.ts b/server/utils/cleanup.ts index 5472ac63..9786196d 100644 --- a/server/utils/cleanup.ts +++ b/server/utils/cleanup.ts @@ -4,11 +4,13 @@ import path from "node:path"; import { setTimeout } from "node:timers/promises"; // Third party imports -import { WebSocket } from "ws"; import { consola } from "consola"; import pTimeout from "p-timeout"; import { rimraf } from "rimraf"; +// Local imports +import { createServerWsRpcClient } from "./ws_client.js"; + interface Microservice { type: "back" | "viewer"; name: string; @@ -88,49 +90,21 @@ async function killWebsocketMicroservice(microservice: Microservice): Promise((resolve) => { - const socket = new WebSocket(socketUrl); - socket.on("open", () => { - consola.info("Connected to WebSocket server"); - socket.send( - JSON.stringify({ - id: "system:hello", - method: "wslink.hello", - args: [ - { - secret: "wslink-secret", - }, - ], - }), - ); - }); - socket.on("message", (data: Buffer | string) => { - const message = data.toString(); - consola.debug("Received from server:", message); - if (message.includes("hello")) { - socket.send( - JSON.stringify({ - id: "application.exit", - method: "application.exit", - }), - ); - consola.info(successMessage); - socket.close(); - resolve(); - } - }); - socket.on("close", () => { - consola.info(successMessage); - resolve(); - }); - socket.on("error", (error: Error) => { - consola.error("WebSocket error:", error); - socket.close(); - resolve(); - }); + const closed = new Promise((resolve) => { + client.onConnectionClose(resolve); }); + try { + await client.notify("application.exit"); + // Wslink handles messages asynchronously: closing right away would drop the exit request + await closed; + consola.info(successMessage); + } catch (error) { + consola.error("WebSocket error:", error); + } finally { + client.close(); + } } await pTimeout(do_kill(), { milliseconds: 5000, diff --git a/server/utils/ws_client.ts b/server/utils/ws_client.ts index 5772c694..d953dd05 100644 --- a/server/utils/ws_client.ts +++ b/server/utils/ws_client.ts @@ -1,4 +1,6 @@ // Third party imports +import { Decoder, encode } from "@msgpack/msgpack"; +import { UnChunker, generateChunks } from "wslink/src/WebsocketConnection/chunking.js"; import { WebSocket } from "ws"; import { v4 as uuidv4 } from "uuid"; @@ -6,6 +8,8 @@ import { v4 as uuidv4 } from "uuid"; const HELLO_ID = "system:hello"; const HELLO_SECRET = "wslink-secret"; +// Wslink default (512 KiB), until the hello response gives the server's limit +const DEFAULT_MAX_MSG_SIZE = 524_288; interface PendingCall { resolve: (value: unknown) => void; @@ -15,7 +19,7 @@ interface PendingCall { interface WsRpcMessage { id?: string; error?: { message?: string }; - result?: unknown; + result?: { maxMsgSize?: number } & Record; } function isWsRpcMessage(value: unknown): value is WsRpcMessage { @@ -23,14 +27,11 @@ function isWsRpcMessage(value: unknown): value is WsRpcMessage { } // oxlint-disable-next-line prefer-readonly-parameter-types -function rawDataToString(raw: WebSocket.RawData): string { - if (Buffer.isBuffer(raw)) { - return raw.toString(); +function rawDataToBlob(raw: WebSocket.RawData): Blob { + if (raw instanceof ArrayBuffer) { + return new Blob([raw]); } - if (Array.isArray(raw)) { - return Buffer.concat(raw).toString(); - } - return Buffer.from(raw).toString(); + return new Blob([Uint8Array.from(Array.isArray(raw) ? Buffer.concat(raw) : raw)]); } interface ServerWsRpcClient { @@ -42,6 +43,7 @@ interface ServerWsRpcClient { }; }; isOpen: () => boolean; + notify: (method: string) => Promise; onConnectionClose: (callback: () => void) => void; onConnectionError: (callback: (error: unknown) => void) => void; ready: Promise; @@ -51,35 +53,43 @@ interface ServerWsRpcClient { function createServerWsRpcClient(baseUrl: string): ServerWsRpcClient { const socket = new WebSocket(baseUrl); const pending = new Map(); + // Wslink 2 messages are msgpack-encoded and split into chunks + const unChunker = new UnChunker(); + let maxMsgSize = DEFAULT_MAX_MSG_SIZE; let onCloseCallback: (() => void) | undefined = undefined; let onErrorCallback: ((error: unknown) => void) | undefined = undefined; + function send(message: object): void { + for (const chunk of generateChunks(encode(message), maxMsgSize)) { + socket.send(chunk); + } + } + //oxlint-disable-next-line promise/avoid-new max-lines-per-function const ready = new Promise((resolve, reject) => { socket.on("open", () => { - socket.send( - JSON.stringify({ - id: HELLO_ID, - method: "wslink.hello", - args: [{ secret: HELLO_SECRET }], - }), - ); + send({ + id: HELLO_ID, + method: "wslink.hello", + args: [{ secret: HELLO_SECRET }], + }); }); // oxlint-disable-next-line prefer-readonly-parameter-types - socket.on("message", (raw: WebSocket.RawData) => { + async function handleMessage(raw: WebSocket.RawData): Promise { let message: WsRpcMessage | undefined = undefined; try { - const parsed: unknown = JSON.parse(rawDataToString(raw)); - if (!isWsRpcMessage(parsed)) { + const decoded = await unChunker.processChunk(rawDataToBlob(raw), () => new Decoder()); + if (!isWsRpcMessage(decoded)) { return; } - message = parsed; + message = decoded; } catch { return; } if (message.id === HELLO_ID) { + maxMsgSize = message.result?.maxMsgSize ?? maxMsgSize; resolve(); return; } @@ -98,6 +108,11 @@ function createServerWsRpcClient(baseUrl: string): ServerWsRpcClient { } else { entry.resolve(message.result); } + } + + // oxlint-disable-next-line prefer-readonly-parameter-types + socket.on("message", (raw: WebSocket.RawData) => { + void handleMessage(raw); }); socket.on("error", (error: Error) => { @@ -106,6 +121,7 @@ function createServerWsRpcClient(baseUrl: string): ServerWsRpcClient { }); socket.on("close", () => { + reject(new Error("WebSocket closed")); onCloseCallback?.(); for (const { reject: rejectPending } of pending.values()) { rejectPending(new Error("WebSocket closed")); @@ -120,18 +136,21 @@ function createServerWsRpcClient(baseUrl: string): ServerWsRpcClient { //oxlint-disable-next-line promise/avoid-new return new Promise((resolve, reject) => { pending.set(id, { resolve, reject }); - socket.send( - JSON.stringify({ - wslink: "1.0", - id, - method: rpc, - args: [params], - kwargs: { stream: true }, - }), - ); + send({ + id, + method: rpc, + args: [params], + kwargs: { stream: true }, + }); }); } + // Call an RPC without arguments nor waiting for its response (e.g. "application.exit") + async function notify(method: string): Promise { + await ready; + send({ id: method, method }); + } + function close(): void { socket.close(); } @@ -161,7 +180,16 @@ function createServerWsRpcClient(baseUrl: string): ServerWsRpcClient { onErrorCallback = callback; } - return { call, close, getConnection, isOpen, onConnectionClose, onConnectionError, ready }; + return { + call, + close, + getConnection, + isOpen, + notify, + onConnectionClose, + onConnectionError, + ready, + }; } export { createServerWsRpcClient }; diff --git a/server/utils/wslink_chunking.d.ts b/server/utils/wslink_chunking.d.ts new file mode 100644 index 00000000..6aa501a3 --- /dev/null +++ b/server/utils/wslink_chunking.d.ts @@ -0,0 +1,12 @@ +// Wslink ships no types for its chunking module; this describes what ws_client.ts uses. +declare module "wslink/src/WebsocketConnection/chunking.js" { + interface Decoder { + decode: (buffer: Uint8Array) => unknown; + } + function generateChunks(message: Uint8Array, maxSize: number): Generator; + class UnChunker { + public processChunk(chunk: Blob, decoderFactory: () => Decoder): Promise; + public releasePendingMessages(): void; + } + export { UnChunker, generateChunks }; +} diff --git a/shared/utils/call_raw.ts b/shared/utils/call_raw.ts index 915021c4..ecaccc1c 100644 --- a/shared/utils/call_raw.ts +++ b/shared/utils/call_raw.ts @@ -26,12 +26,35 @@ interface CallRawOptions extends CallClientOptions { timeout?: number; } +function isPlainRecord(value: unknown): value is Record { + return _.isPlainObject(value); +} + +// Wslink encodes messages with msgpack, which sends undefined values as null where JSON omitted them +function omitUndefinedFields(record: object): Record { + function omitUndefined(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map((item: unknown) => omitUndefined(item)); + } + if (isPlainRecord(value)) { + return omitUndefinedFields(value); + } + return value; + } + return Object.fromEntries( + Object.entries(record) + .filter(([, value]) => value !== undefined) + .map(([key, value]) => [key, omitUndefined(value)]), + ); +} + async function callClient({ rpc, params = {}, client }: CallClientOptions): Promise { + const definedParams = omitUndefinedFields(params); if (globalThis.window !== undefined) { - const response = await client.getConnection().getSession().call(rpc, [params]); + const response = await client.getConnection().getSession().call(rpc, [definedParams]); return response; } - const response = await client.call(rpc, params); + const response = await client.call(rpc, definedParams); return response; } diff --git a/tests/integration/setup.ts b/tests/integration/setup.ts index 28da9529..0f70bb63 100644 --- a/tests/integration/setup.ts +++ b/tests/integration/setup.ts @@ -13,6 +13,7 @@ import { addMicroserviceMetadatas, runBack, runViewer } from "@ogw_server/utils/ import { createPath, generateProjectFolderPath } from "@ogw_server/utils/path"; import { Status } from "@ogw_front/utils/status"; import { appMode } from "@ogw_shared/app_mode"; +import { cleanupBackend } from "@ogw_server/utils/cleanup"; import { importFile } from "@ogw_front/utils/import_workflow"; import { setupActivePinia } from "@ogw_tests/utils"; import { useBackStore } from "@ogw_front/stores/back"; @@ -22,6 +23,8 @@ import { useViewerStore } from "@ogw_front/stores/viewer"; // Local constants const beforeAllTimeout = 40_000; const data_folder = path.join("tests", "integration", "data", "uploads"); +// Cleaned up after each test file, even when its setup failed before returning the folder +const startedProjectFolderPaths: string[] = []; async function runMicroservices(): Promise<{ projectFolderPath: string }> { const backStore = useBackStore(); @@ -31,33 +34,27 @@ async function runMicroservices(): Promise<{ projectFolderPath: string }> { const { COMMAND_BACK, PROJECT, COMMAND_VIEWER, NUXT_ROOT_PATH } = useRuntimeConfig().public; const projectFolderPath = generateProjectFolderPath(PROJECT); createPath(projectFolderPath); + startedProjectFolderPaths.push(projectFolderPath); - const [back_port, viewer_port] = await Promise.all([ - runBack(COMMAND_BACK, NUXT_ROOT_PATH, { + // Each microservice is registered as soon as it runs, so it is killed even if the other fails + async function startBack(): Promise { + const port = await runBack(COMMAND_BACK, NUXT_ROOT_PATH, { projectFolderPath, uploadFolderPath: data_folder, - }), - runViewer(COMMAND_VIEWER, NUXT_ROOT_PATH, { projectFolderPath }), - ]); + }); + addMicroserviceMetadatas(projectFolderPath, { type: "back", name: COMMAND_BACK, port }); + return port; + } + async function startViewer(): Promise { + const port = await runViewer(COMMAND_VIEWER, NUXT_ROOT_PATH, { projectFolderPath }); + addMicroserviceMetadatas(projectFolderPath, { type: "viewer", name: COMMAND_VIEWER, port }); + return port; + } + const [back_port, viewer_port] = await Promise.all([startBack(), startViewer()]); consola.debug("back_port", back_port); consola.debug("viewer_port", viewer_port); - if (back_port === undefined || viewer_port === undefined) { - throw new Error("Failed to start microservices: back_port or viewer_port is undefined"); - } - - addMicroserviceMetadatas(projectFolderPath, { - type: "back", - name: COMMAND_BACK, - port: back_port, - }); - addMicroserviceMetadatas(projectFolderPath, { - type: "viewer", - name: COMMAND_VIEWER, - port: viewer_port, - }); - backStore.default_local_port = String(back_port); viewerStore.default_local_port = String(viewer_port); @@ -98,8 +95,13 @@ beforeAll(() => { globalThis.WebSocket = WebSocket as unknown as typeof globalThis.WebSocket; }); -afterAll(() => { +afterAll(async () => { Reflect.deleteProperty(globalThis, "WebSocket"); + await Promise.all( + startedProjectFolderPaths.splice(0).map(async (folderPath) => { + await cleanupBackend(folderPath); + }), + ); }); export { beforeAllTimeout, runMicroservices, setupIntegrationTests };