JS-730 Websockets implementation (#5398)

Co-authored-by: Victor Diez <victor.diez@sonarsource.com>

Author: zglicz

its/plugin/tests/src/test/java/com/sonar/javascript/it/plugin/SonarJsIntegrationTest.java

@@ -166,7 +166,7 @@ class Bridge {
     private Process process;
 
     Bridge() {
-      this.client = HttpClient.newHttpClient();
+      this.client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build();
     }
 
     void start(Path dest) throws IOException {
@@ -184,6 +184,7 @@ void start(Path dest) throws IOException {
 
     String request(String json, String endpoint) throws IOException, InterruptedException {
       var request = HttpRequest.newBuilder(url(endpoint))
+        .version(HttpClient.Version.HTTP_1_1)
         .header("Content-Type", "application/json")
         .POST(HttpRequest.BodyPublishers.ofString(json))
         .build();

package-lock.json

@@ -73,6 +73,7 @@
     "@types/node": "22.15.24",
     "@types/semver": "7.7.0",
     "@types/tmp": "0.2.6",
+    "@types/ws": "8.18.1",
     "cpy-cli": "5.0.0",
     "cross-env": "7.0.3",
     "dir-compare": "5.0.0",
@@ -121,7 +122,7 @@
     "functional-red-black-tree": "1.0.1",
     "globals": "16.2.0",
     "htmlparser2": "10.0.0",
-    "http-status-codes": "^2.3.0",
+    "http-status-codes": "2.3.0",
     "jsx-ast-utils": "3.3.5",
     "lodash.merge": "4.6.2",
     "minimatch": "9.0.5",
@@ -141,6 +142,7 @@
     "tmp": "0.2.3",
     "typescript": "5.8.3",
     "vue-eslint-parser": "10.1.3",
+    "ws": "8.18.2",
     "yaml": "2.8.0"
   },
   "prettier": {

package.json

@@ -16,19 +16,24 @@
  */
 import express from 'express';
 import { Worker } from 'node:worker_threads';
-import { JsTsAnalysisOutputWithAst } from '../../jsts/src/analysis/analysis.js';
 import { handleRequest } from './handle-request.js';
-import { AnalysisOutput } from '../../shared/src/types/analysis.js';
-import { RequestResult, RequestType } from './request.js';
+import { RequestResult, RequestType, WsIncrementalResult } from './request.js';
 import { WorkerData } from '../../shared/src/helpers/worker.js';
+import { info, debug, error } from '../../shared/src/helpers/logging.js';
+import type { RawData, WebSocket } from 'ws';
+import { WorkerMessageListeners } from './router.js';
 
 /**
  * Returns a delegate function to handle an HTTP request
  */
-export function createDelegator(worker: Worker | undefined, workerData: WorkerData) {
-  return function (type: RequestType) {
-    return worker ? createWorkerHandler(worker, type) : createHandler(type, workerData);
-  };
+export function createDelegator(
+  worker: Worker | undefined,
+  workerData: WorkerData,
+  listeners: WorkerMessageListeners,
+) {
+  return worker
+    ? (type: RequestType) => createWorkerHandler(worker, type, listeners)
+    : (type: RequestType) => createHandler(type, workerData);
 }
 
 /**
@@ -46,14 +51,16 @@ function createHandler(type: RequestType, workerData: WorkerData) {
   };
 }
 
-function createWorkerHandler(worker: Worker, type: RequestType) {
+function createWorkerHandler(worker: Worker, type: RequestType, listeners: WorkerMessageListeners) {
   return async (
     request: express.Request,
     response: express.Response,
     next: express.NextFunction,
   ) => {
-    worker.once('message', message => {
-      handleResult(message, response, next);
+    listeners.oneTimers.push(message => {
+      if (!message.ws) {
+        handleResult(message, response, next);
+      }
     });
     worker.postMessage({ type, data: request.body });
   };
@@ -75,6 +82,53 @@ function handleResult(
   }
 }
 
-export function outputContainsAst(result: AnalysisOutput): result is JsTsAnalysisOutputWithAst {
-  return 'astFilePath' in result;
+/**
+ * Returns a delegate function to handle a web socket message
+ */
+export function createWsDelegator(
+  worker: Worker | undefined,
+  workerData: WorkerData,
+  listeners: WorkerMessageListeners,
+) {
+  return (ws: WebSocket) => {
+    info('WebSocket client connected on /ws');
+    if (worker) {
+      listeners.permanent.push(message => {
+        if (message.ws) {
+          handleWsResult(ws, message.results);
+        }
+      });
+    }
+
+    ws.on('message', async message => {
+      const data = { type: 'on-analyze-project' as const, data: decodeMessage(message), ws: true };
+      if (worker) {
+        worker.postMessage(data);
+      } else {
+        await handleRequest(data, workerData, message => handleWsResult(ws, message));
+      }
+    });
+
+    ws.on('close', (code, reason) => {
+      debug(`WebSocket client disconnected: ${reason} with code ${code}`);
+    });
+
+    ws.on('error', err => {
+      error(`WebSocket client error: ${err}`);
+    });
+  };
+}
+
+function handleWsResult(ws: WebSocket, message: WsIncrementalResult) {
+  ws.send(JSON.stringify(message));
+}
+
+function decodeMessage(message: RawData) {
+  let jsonString = '';
+  if (Buffer.isBuffer(message)) {
+    jsonString = message.toString('utf8');
+  } else if (Array.isArray(message)) {
+    jsonString = Buffer.concat(message).toString('utf8');
+  }
+  return JSON.parse(jsonString);
 }

packages/bridge/src/delegate.ts

@@ -28,12 +28,13 @@ import {
 } from '../../jsts/src/program/program.js';
 import { Linter } from '../../jsts/src/linter/linter.js';
 import { clearTypeScriptESLintParserCaches } from '../../jsts/src/parsers/eslint.js';
-import { BridgeRequest, RequestResult, serializeError } from './request.js';
+import { BridgeRequest, RequestResult, serializeError, WsIncrementalResult } from './request.js';
 import { WorkerData } from '../../shared/src/helpers/worker.js';
 
 export async function handleRequest(
   request: BridgeRequest,
   workerData: WorkerData,
+  incrementalResultsChannel?: (result: WsIncrementalResult) => void,
 ): Promise<RequestResult> {
   try {
     switch (request.type) {
@@ -101,7 +102,7 @@ export async function handleRequest(
       }
       case 'on-analyze-project': {
         logHeapStatistics(workerData?.debugMemory);
-        const output = await analyzeProject(request.data);
+        const output = await analyzeProject(request.data, incrementalResultsChannel);
         logHeapStatistics(workerData?.debugMemory);
         return { type: 'success', result: output };
       }

packages/bridge/src/handle-request.ts

@@ -15,7 +15,11 @@
  * along with this program; if not, see https://sonarsource.com/license/ssal/
  */
 import { AnalysisOutput } from '../../shared/src/types/analysis.js';
-import { ProjectAnalysisInput } from '../../jsts/src/analysis/projectAnalysis/projectAnalysis.js';
+import {
+  type FileResult,
+  ProjectAnalysisInput,
+  ProjectAnalysisMeta,
+} from '../../jsts/src/analysis/projectAnalysis/projectAnalysis.js';
 import { TsConfigJson } from 'type-fest';
 import { RuleConfig } from '../../jsts/src/linter/config/rule-config.js';
 import { APIError, ErrorCode } from '../../shared/src/errors/error.js';
@@ -34,6 +38,10 @@ export type RequestResult =
       error: ReturnType<typeof serializeError>;
     };
 
+export type WsMetaResult = { messageType: 'meta' } & ProjectAnalysisMeta;
+export type WsFileResult = { filename: string; messageType: 'fileResult' } & FileResult;
+export type WsIncrementalResult = WsFileResult | WsMetaResult;
+
 export type Telemetry = {
   dependencies: NamedDependency[];
 };

packages/bridge/src/request.ts

@@ -16,14 +16,33 @@
  */
 import * as express from 'express';
 import { Worker } from 'worker_threads';
-import { createDelegator } from './delegate.js';
+import { createDelegator, createWsDelegator } from './delegate.js';
 import { WorkerData } from '../../shared/src/helpers/worker.js';
 import { StatusCodes } from 'http-status-codes';
+import { WebSocketServer } from 'ws';
 
-export default function (worker: Worker | undefined, workerData: WorkerData): express.Router {
-  const router = express.Router();
-  const delegate = createDelegator(worker, workerData);
+export type WorkerMessageListeners = {
+  permanent: ((message: any) => void)[];
+  oneTimers: ((message: any) => void)[];
+};
+
+export default function (
+  worker: Worker | undefined,
+  workerData: WorkerData,
+  wss: WebSocketServer,
+): express.Router {
+  const workerMessageListeners: WorkerMessageListeners = { permanent: [], oneTimers: [] };
+  if (worker) {
+    worker.on('message', message => {
+      workerMessageListeners.permanent.forEach(listener => listener(message));
+      workerMessageListeners.oneTimers.forEach(listener => listener(message));
+      workerMessageListeners.oneTimers = [];
+    });
+  }
 
+  const router = express.Router();
+  const delegate = createDelegator(worker, workerData, workerMessageListeners);
+  const wsDelegate = createWsDelegator(worker, workerData, workerMessageListeners);
   /** Endpoints running on the worker thread */
   router.post('/analyze-project', delegate('on-analyze-project'));
   router.post('/analyze-css', delegate('on-analyze-css'));
@@ -38,6 +57,8 @@ export default function (worker: Worker | undefined, workerData: WorkerData): ex
   router.post('/tsconfig-files', delegate('on-tsconfig-files'));
   router.get('/get-telemetry', delegate('on-get-telemetry'));
 
+  wss.on('connection', wsDelegate);
+
   /** Endpoints running on the main thread */
   router.get('/status', (_, response) => {
     response.sendStatus(StatusCodes.OK);

packages/bridge/src/router.ts

@@ -31,6 +31,7 @@ import {
   logMemoryConfiguration,
   logMemoryError,
 } from './memory.js';
+import { WebSocketServer } from 'ws';
 
 /**
  * The maximum request body size
@@ -99,6 +100,18 @@ export function start(
 
     const app = express();
     const server = http.createServer(app);
+    const wss = new WebSocketServer({ noServer: true });
+
+    server.on('upgrade', (request, socket, head) => {
+      // Only handle upgrade requests for /ws
+      if (request.headers.upgrade?.toLowerCase() === 'websocket' && request.url === '/ws') {
+        wss.handleUpgrade(request, socket, head, ws => {
+          wss.emit('connection', ws, request);
+        });
+      } else {
+        socket.destroy();
+      }
+    });
 
     /**
      * Builds a timeout middleware to shut down the server
@@ -112,7 +125,7 @@ export function start(
      */
     app.use(express.json({ limit: MAX_REQUEST_SIZE }));
     app.use(orphanTimeout.middleware);
-    app.use(router(worker, { debugMemory }));
+    app.use(router(worker, { debugMemory }, wss));
     app.use(errorMiddleware);
 
     app.post('/close', (_: express.Request, response: express.Response) => {
@@ -157,20 +170,27 @@ export function start(
      * Shutdown the server and the worker thread
      */
     function closeServer() {
-      unregisterGarbageCollectionObserver();
-      if (server.listening) {
-        while (pendingCloseRequests.length) {
-          pendingCloseRequests.pop()?.end();
+      debug('Closing server');
+      wss.clients.forEach(client => {
+        client.terminate(); // Immediately destroys the connection
+      });
+      wss.close(() => {
+        debug('Closed WebSocket connection');
+        unregisterGarbageCollectionObserver();
+        if (server.listening) {
+          while (pendingCloseRequests.length) {
+            pendingCloseRequests.pop()?.end();
+          }
+          /**
+           * At this point, the worker thread can no longer respond to any request from the plugin.
+           * If we reached this due to worker failure, existing requests are stalled until they time out.
+           * Since the bridge server is about to be shut down in an unexpected manner anyway, we can
+           * close all connections and avoid waiting unnecessarily for them to eventually close.
+           */
+          server.closeAllConnections();
+          server.close();
         }
-        /**
-         * At this point, the worker thread can no longer respond to any request from the plugin.
-         * If we reached this due to worker failure, existing requests are stalled until they time out.
-         * Since the bridge server is about to be shut down in an unexpected manner anyway, we can
-         * close all connections and avoid waiting unnecessarily for them to eventually close.
-         */
-        server.closeAllConnections();
-        server.close();
-      }
+      });
     }
   });
 }

packages/bridge/src/server.ts

@@ -16,19 +16,26 @@
  */
 import { parentPort, workerData } from 'worker_threads';
 import { handleRequest } from './handle-request.js';
-import { BridgeRequest } from './request.js';
+import { BridgeRequest, WsIncrementalResult } from './request.js';
 
 /**
  * Code executed by the worker thread
  */
 if (parentPort) {
   const parentThread = parentPort;
-  parentThread.on('message', async (message: BridgeRequest | { type: 'close' }) => {
-    const { type } = message;
-    if (type === 'close') {
-      parentThread.close();
-    } else {
-      parentThread.postMessage(await handleRequest(message, workerData));
-    }
-  });
+  parentThread.on(
+    'message',
+    async (message: (BridgeRequest | { type: 'close' }) & { ws?: boolean }) => {
+      const { type, ws } = message;
+      if (type === 'close') {
+        parentThread.close();
+      } else if (ws) {
+        await handleRequest(message, workerData, (results: WsIncrementalResult) =>
+          parentThread.postMessage({ ws: true, results }),
+        );
+      } else {
+        parentThread.postMessage(await handleRequest(message, workerData));
+      }
+    },
+  );
 }

packages/bridge/src/worker.ts

@@ -41,7 +41,7 @@ describe('router', () => {
   let server: http.Server;
 
   before(async () => {
-    const worker = createWorker(workerPath);
+    const worker = await createWorker(workerPath);
     const { server: serverInstance, serverClosed } = await start(port, '127.0.0.1', worker);
     server = serverInstance;
     closePromise = serverClosed;