refactor(websocket): implement a class, commands and DI

package.json

@@ -112,6 +112,7 @@
     "sade": "^1.8.1",
     "semver": "^7.6.3",
     "sirv": "^3.0.0",
+    "ts-pattern": "^5.6.2",
     "ws": "^8.18.0",
     "zup": "0.0.2"
   }

src/http-server/index.js

@@ -6,7 +6,6 @@ import kleur from "kleur";
 import polka from "polka";
 import open from "open";
 import * as i18n from "@nodesecure/i18n";
-import { WebSocketServer } from "ws";
 
 // Import Internal Dependencies
 import * as root from "./endpoints/root.js";
@@ -20,9 +19,8 @@ import * as scorecard from "./endpoints/ossf-scorecard.js";
 import * as locali18n from "./endpoints/i18n.js";
 import * as report from "./endpoints/report.js";
 import * as middlewares from "./middlewares/index.js";
-import * as wsHandlers from "./websocket/index.js";
-import { logger } from "../logger.js";
 import { appCache } from "../cache.js";
+import { WebSocketServerInstanciator } from "./websocket/index.js";
 
 export function buildServer(dataFilePath, options = {}) {
   const httpConfigPort = typeof options.port === "number" ? options.port : 0;
@@ -33,15 +31,17 @@ export function buildServer(dataFilePath, options = {}) {
 
   const httpServer = polka();
 
+  const asyncStoreProperties = {};
   if (runFromPayload) {
     fs.accessSync(dataFilePath, fs.constants.R_OK | fs.constants.W_OK);
-    httpServer.use(
-      middlewares.buildContextMiddleware(dataFilePath, hotReload)
-    );
+    asyncStoreProperties.dataFilePath = dataFilePath;
   }
   else {
     appCache.startFromZero = true;
   }
+  httpServer.use(
+    middlewares.buildContextMiddleware(hotReload, asyncStoreProperties)
+  );
 
   httpServer.use(middlewares.addStaticFiles);
   httpServer.get("/", root.get);
@@ -72,24 +72,7 @@ export function buildServer(dataFilePath, options = {}) {
     }
   });
 
-  if (enableWS) {
-    const websocket = new WebSocketServer({ port: 1338 });
-    websocket.on("connection", async(socket) => {
-      socket.on("message", async(rawMessage) => {
-        const message = JSON.parse(rawMessage);
-        logger.info(`[ws](message: ${JSON.stringify(message)})`);
-
-        if (message.action === "SEARCH") {
-          wsHandlers.search(socket, message.pkg);
-        }
-        else if (message.action === "REMOVE") {
-          wsHandlers.remove(socket, message.pkg);
-        }
-      });
-
-      wsHandlers.init(socket);
-    });
-  }
+  enableWS && new WebSocketServerInstanciator();
 
   return httpServer;
 }

src/http-server/middlewares/context.js

@@ -3,15 +3,15 @@ import { context } from "../../ALS.js";
 import { ViewBuilder } from "../ViewBuilder.class.js";
 
 export function buildContextMiddleware(
-  dataFilePath,
-  autoReload = false
+  autoReload = false,
+  storeProperties = {}
 ) {
   const viewBuilder = new ViewBuilder({
     autoReload
   });
 
   return function addContext(_req, _res, next) {
-    const store = { dataFilePath, viewBuilder };
+    const store = { ...storeProperties, viewBuilder };
     context.run(store, next);
   };
 }

src/http-server/websocket/remove.js → src/http-server/websocket/commands/remove.js

@@ -1,13 +1,14 @@
-// Import Internal Dependencies
-import { appCache } from "../../cache.js";
-import { logger } from "../../logger.js";
+export async function* remove(
+  pkg,
+  context
+) {
+  const { cache, logger } = context;
 
-export async function remove(ws, pkg) {
   const formattedPkg = pkg.replace("/", "-");
   logger.info(`[ws|remove](pkg: ${pkg}|formatted: ${formattedPkg})`);
 
   try {
-    const { lru, older, current, lastUsed, root } = await appCache.payloadsList();
+    const { lru, older, current, lastUsed, root } = await cache.payloadsList();
     logger.debug(`[ws|remove](lru: ${lru}|current: ${current})`);
 
     if (lru.length === 1 && older.length === 0) {
@@ -46,12 +47,12 @@ export async function remove(ws, pkg) {
         current: current === pkg ? updatedLru[0] : current,
         root
       };
-      await appCache.updatePayloadsList(updatedList);
+      await cache.updatePayloadsList(updatedList);
 
-      ws.send(JSON.stringify({
+      yield {
         status: "RELOAD",
         ...updatedList
-      }));
+      };
     }
     else {
       logger.info(`[ws|remove](remove from older)`);
@@ -66,15 +67,15 @@ export async function remove(ws, pkg) {
         current,
         root
       };
-      await appCache.updatePayloadsList(updatedList);
+      await cache.updatePayloadsList(updatedList);
 
-      ws.send(JSON.stringify({
+      yield {
         status: "RELOAD",
         ...updatedList
-      }));
+      };
     }
 
-    appCache.removePayload(formattedPkg);
+    cache.removePayload(formattedPkg);
   }
   catch (error) {
     logger.error(`[ws|remove](error: ${error.message})`);

src/http-server/websocket/search.js → src/http-server/websocket/commands/search.js

@@ -1,62 +1,62 @@
 // Import Third-party Dependencies
 import * as Scanner from "@nodesecure/scanner";
 
-// Import Internal Dependencies
-import { logger } from "../../logger.js";
-import { appCache } from "../../cache.js";
-
-export async function search(ws, pkg) {
+export async function* search(
+  pkg,
+  context
+) {
+  const { logger, cache } = context;
   logger.info(`[ws|search](pkg: ${pkg})`);
 
-  const cache = await appCache.getPayloadOrNull(pkg);
-  if (cache) {
+  const cachedPayload = await cache.getPayloadOrNull(pkg);
+  if (cachedPayload) {
     logger.info(`[ws|search](payload: ${pkg} found in cache)`);
-    const cacheList = await appCache.payloadsList();
+    const cacheList = await cache.payloadsList();
     if (cacheList.lru.includes(pkg)) {
       logger.info(`[ws|search](payload: ${pkg} is already in the LRU)`);
       const updatedList = {
         ...cacheList,
         current: pkg,
         lastUsed: { ...cacheList.lastUsed, [pkg]: Date.now() }
       };
-      await appCache.updatePayloadsList(updatedList);
-      ws.send(JSON.stringify(cache));
+      await cache.updatePayloadsList(updatedList);
+      yield cachedPayload;
 
-      if (appCache.startFromZero) {
-        ws.send(JSON.stringify({
+      if (cache.startFromZero) {
+        yield {
           status: "RELOAD",
           ...updatedList
-        }));
-        appCache.startFromZero = false;
+        };
+        cache.startFromZero = false;
       }
 
       return;
     }
 
-    const { lru, older, lastUsed, ...updatedCache } = await appCache.removeLastLRU();
+    const { lru, older, lastUsed, ...updatedCache } = await cache.removeLastLRU();
     const updatedList = {
       ...updatedCache,
       lru: [...new Set([...lru, pkg])],
       current: pkg,
       older: older.filter((pckg) => pckg !== pkg),
       lastUsed: { ...lastUsed, [pkg]: Date.now() }
     };
-    await appCache.updatePayloadsList(updatedList);
+    await cache.updatePayloadsList(updatedList);
 
-    ws.send(JSON.stringify(cache));
-    ws.send(JSON.stringify({
+    yield cachedPayload;
+    yield {
       status: "RELOAD",
       ...updatedList
-    }));
+    };
 
-    appCache.startFromZero = false;
+    cache.startFromZero = false;
 
     return;
   }
 
   // at this point we don't have the payload in cache so we have to scan it.
   logger.info(`[ws|search](scan ${pkg} in progress)`);
-  ws.send(JSON.stringify({ status: "SCAN", pkg }));
+  yield { status: "SCAN", pkg };
 
   const payload = await Scanner.from(pkg, { maxDepth: 4 });
   const name = payload.rootDependencyName;
@@ -68,25 +68,25 @@ export async function search(ws, pkg) {
     logger.info(`[ws|search](scan ${pkg} done|cache: updated)`);
 
     // update the payloads list
-    const { lru, older, lastUsed, ...cache } = await appCache.removeLastLRU();
+    const { lru, older, lastUsed, ...LRUCache } = await cache.removeLastLRU();
     lru.push(pkg);
-    appCache.updatePayload(pkg, payload);
+    cache.updatePayload(pkg, payload);
     const updatedList = {
-      ...cache,
+      ...LRUCache,
       lru: [...new Set(lru)],
       older,
       lastUsed: { ...lastUsed, [pkg]: Date.now() },
       current: pkg
     };
-    await appCache.updatePayloadsList(updatedList);
+    await cache.updatePayloadsList(updatedList);
 
-    ws.send(JSON.stringify(payload));
-    ws.send(JSON.stringify({
+    yield payload;
+    yield {
       status: "RELOAD",
       ...updatedList
-    }));
+    };
 
-    appCache.startFromZero = false;
+    cache.startFromZero = false;
 
     logger.info(`[ws|search](data sent to client|cache: updated)`);
   }

src/http-server/websocket/index.js

@@ -1,3 +1,86 @@
-export * from "./search.js";
-export * from "./remove.js";
-export * from "./init.js";
+// Import Third-party Dependencies
+import { WebSocketServer } from "ws";
+import { match } from "ts-pattern";
+
+// Import Internal Dependencies
+import { appCache } from "../../cache.js";
+import { logger } from "../../logger.js";
+import { search } from "./commands/search.js";
+import { remove } from "./commands/remove.js";
+
+export class WebSocketServerInstanciator {
+  constructor() {
+    const websocket = new WebSocketServer({
+      port: 1338
+    });
+    websocket.on("connection", this.onConnectionHandler.bind(this));
+  }
+
+  async onConnectionHandler(socket) {
+    socket.on("message", (rawData) => {
+      logger.info(`[ws](message: ${rawData})`);
+
+      this.onMessageHandler(socket, JSON.parse(rawData))
+        .catch(console.error);
+    });
+
+    const data = await this.initializeServer();
+    sendSocketResponse(socket, data);
+  }
+
+  async onMessageHandler(
+    socket,
+    message
+  ) {
+    const ctx = { socket, cache: appCache, logger };
+
+    const socketMessages = await match(message.action)
+      .with("SEARCH", () => search(message.pkg, ctx))
+      .with("REMOVE", () => remove(message.pkg, ctx))
+      .exhaustive();
+
+    for await (const message of socketMessages) {
+      sendSocketResponse(socket, message);
+    }
+  }
+
+  async initializeServer(
+    stopInitializationOnError = false
+  ) {
+    try {
+      const { current, lru, older, root } = await appCache.payloadsList();
+      logger.info(`[ws|init](lru: ${lru}|older: ${older}|current: ${current}|root: ${root})`);
+
+      if (lru === void 0 || current === void 0) {
+        throw new Error("Payloads list not found in cache.");
+      }
+
+      return {
+        status: "INIT",
+        current,
+        lru,
+        older,
+        root
+      };
+    }
+    catch {
+      if (stopInitializationOnError) {
+        return null;
+      }
+
+      logger.error(`[ws|init](No cache yet. Creating one...)`);
+      await appCache.initPayloadsList();
+
+      return this.initializeServer(true);
+    }
+  }
+}
+
+function sendSocketResponse(
+  socket,
+  message
+) {
+  if (message !== null) {
+    socket.send(JSON.stringify(message));
+  }
+}