Use in-process event emitter instead of Redis pub/sub

Author: goto-bus-stop

package.json

@@ -38,6 +38,7 @@
     "cookie": "^1.0.1",
     "cookie-parser": "^1.4.4",
     "cors": "^2.8.5",
+    "emittery": "^1.2.0",
     "escape-string-regexp": "^5.0.0",
     "explain-error": "^1.0.4",
     "express": "^5.0.0",

src/SocketServer.js

@@ -1,6 +1,5 @@
 import { promisify } from 'node:util';
 import lodash from 'lodash';
-import sjson from 'secure-json-parse';
 import { WebSocketServer } from 'ws';
 import Ajv from 'ajv';
 import { stdSerializers } from 'pino';
@@ -96,8 +95,6 @@ class SocketServer {
 
   #logger;
 
-  #redisSubscription;
-
   #wss;
 
   #closing = false;
@@ -134,6 +131,8 @@ class SocketServer {
    */
   #serverActions;
 
+  #unsubscribe;
+
   /**
    * Create a socket server.
    *
@@ -157,7 +156,6 @@ class SocketServer {
         req: stdSerializers.req,
       },
     });
-    this.#redisSubscription = uw.redis.duplicate();
 
     this.options = {
       /** @type {(_socket: import('ws').WebSocket | undefined, err: Error) => void} */
@@ -176,11 +174,8 @@ class SocketServer {
       port: options.server ? undefined : options.port,
     });
 
-    uw.use(() => this.#redisSubscription.subscribe('uwave'));
-    this.#redisSubscription.on('message', (channel, command) => {
-      // this returns a promise, but we don't handle the error case:
-      // there is not much we can do, so just let node.js crash w/ an unhandled rejection
-      this.onServerMessage(channel, command);
+    this.#unsubscribe = uw.events.onAny((command, data) => {
+      this.#onServerMessage(command, data);
     });
 
     this.#wss.on('error', (error) => {
@@ -678,33 +673,20 @@ class SocketServer {
   }
 
   /**
-   * Handle command messages coming in from Redis.
+   * Handle command messages coming in from elsewhere in the app.
    * Some commands are intended to broadcast immediately to all connected
    * clients, but others require special action.
    *
-   * @param {string} channel
-   * @param {string} rawCommand
-   * @returns {Promise<void>}
-   * @private
+   * @template {keyof import('./redisMessages.js').ServerActionParameters} K
+   * @param {K} command
+   * @param {import('./redisMessages.js').ServerActionParameters[K]} data
    */
-  async onServerMessage(channel, rawCommand) {
-    /**
-     * @type {{ command: string, data: import('type-fest').JsonValue }|undefined}
-     */
-    const json = sjson.safeParse(rawCommand);
-    if (!json) {
-      return;
-    }
-    const { command, data } = json;
+  #onServerMessage(command, data) {
+    this.#logger.trace({ channel: command, command, data }, 'server message');
 
-    this.#logger.trace({ channel, command, data }, 'server message');
-
-    if (has(this.#serverActions, command)) {
-      const action = this.#serverActions[command];
-      if (action !== undefined) { // the types for `ServerActions` allow undefined, so...
-        // @ts-expect-error TS2345 `data` is validated
-        action(data);
-      }
+    const action = this.#serverActions[command];
+    if (action !== undefined) {
+      action(data);
     }
   }
 
@@ -714,9 +696,10 @@ class SocketServer {
    * @returns {Promise<void>}
    */
   async destroy() {
-    clearInterval(this.#pinger);
-
     this.#closing = true;
+
+    this.#unsubscribe();
+    clearInterval(this.#pinger);
     clearInterval(this.#guestCountInterval);
 
     for (const connection of this.#connections) {
@@ -725,7 +708,6 @@ class SocketServer {
 
     const closeWsServer = promisify(this.#wss.close.bind(this.#wss));
     await closeWsServer();
-    await this.#redisSubscription.quit();
   }
 
   /**

src/Uwave.js

@@ -21,6 +21,7 @@ import waitlist from './plugins/waitlist.js';
 import passport from './plugins/passport.js';
 import migrations from './plugins/migrations.js';
 import { SqliteDateColumnsPlugin, connect as connectSqlite } from './utils/sqlite.js';
+import Emittery from 'emittery';
 
 const DEFAULT_SQLITE_PATH = './uwave.sqlite';
 const DEFAULT_REDIS_URL = 'redis://localhost:6379';
@@ -125,6 +126,11 @@ class UwaveServer extends EventEmitter {
   // @ts-expect-error TS2564 Definitely assigned in a plugin
   socketServer;
 
+  /** @type {Emittery<import('./redisMessages.js').ServerActionParameters>} */
+  events = new Emittery({
+    debug: { name: 'u-wave-core' },
+  });
+
   /**
    * @type {Map<string, Source>}
    */
@@ -294,7 +300,7 @@ class UwaveServer extends EventEmitter {
    * @param {import('./redisMessages.js').ServerActionParameters[CommandName]} data
    */
   publish(command, data) {
-    return this.redis.publish('uwave', JSON.stringify({ command, data }));
+    return this.events.emit(command, data);
   }
 
   async listen() {

src/plugins/configStore.js

@@ -4,7 +4,6 @@ import EventEmitter from 'node:events';
 import Ajv from 'ajv/dist/2019.js';
 import formats from 'ajv-formats';
 import jsonMergePatch from 'json-merge-patch';
-import sjson from 'secure-json-parse';
 import ValidationError from '../errors/ValidationError.js';
 import { sql } from 'kysely';
 import { fromJson, json, jsonb } from '../utils/sqlite.js';
@@ -29,12 +28,12 @@ class ConfigStore {
 
   #logger;
 
-  #subscriber;
-
   #ajv;
 
   #emitter = new EventEmitter();
 
+  #unsubscribe;
+
   /** @type {Map<string, import('ajv').ValidateFunction<unknown>>} */
   #validators = new Map();
 
@@ -44,7 +43,6 @@ class ConfigStore {
   constructor(uw) {
     this.#uw = uw;
     this.#logger = uw.logger.child({ ns: 'uwave:config' });
-    this.#subscriber = uw.redis.duplicate();
     this.#ajv = new Ajv({
       useDefaults: true,
       // Allow unknown keywords (`uw:xyz`)
@@ -59,40 +57,16 @@ class ConfigStore {
       fs.readFileSync(new URL('../schemas/definitions.json', import.meta.url), 'utf8'),
     ));
 
-    this.#subscriber.on('message', (_channel, command) => {
-      this.#onServerMessage(command);
-    });
-
-    uw.use(async () => this.#subscriber.subscribe('uwave'));
-  }
-
-  /**
-   * @param {string} rawCommand
-   */
-  async #onServerMessage(rawCommand) {
-    /**
-     * @type {undefined|{
-     *   command: string,
-     *   data: import('../redisMessages.js').ServerActionParameters['configStore:update'],
-     * }}
-     */
-    const json = sjson.safeParse(rawCommand);
-    if (!json) {
-      return;
-    }
-    const { command, data } = json;
-    if (command !== CONFIG_UPDATE_MESSAGE) {
-      return;
-    }
-
-    this.#logger.trace({ command, data }, 'handle config update');
+    this.#unsubscribe = uw.events.on(CONFIG_UPDATE_MESSAGE, async (data) => {
+      this.#logger.trace({ data }, 'handle config update');
 
-    try {
-      const updatedSettings = await this.get(data.key);
-      this.#emitter.emit(data.key, updatedSettings, data.user, data.patch);
-    } catch (error) {
-      this.#logger.error({ err: error }, 'could not retrieve settings after update');
-    }
+      try {
+        const updatedSettings = await this.get(data.key);
+        this.#emitter.emit(data.key, updatedSettings, data.user, data.patch);
+      } catch (error) {
+        this.#logger.error({ err: error }, 'could not retrieve settings after update');
+      }
+    });
   }
 
   /**
@@ -258,8 +232,8 @@ class ConfigStore {
     };
   }
 
-  async destroy() {
-    await this.#subscriber.quit();
+  destroy() {
+    this.#unsubscribe();
   }
 }
 
@@ -268,7 +242,7 @@ class ConfigStore {
  */
 async function configStorePlugin(uw) {
   uw.config = new ConfigStore(uw);
-  uw.onClose(() => uw.config.destroy());
+  uw.onClose(async () => uw.config.destroy());
 }
 
 export default configStorePlugin;