Fork sse pubsub, typescript, logging (growthbook#15)

* fork sse-pubsub, typescript, logging, add license

* increase version

* cleanup

* cleanup

Author: bryce-fitzsimons

LICENSE

@@ -0,0 +1,22 @@
+MIT License
+
+Copyright (c) 2021 GrowthBook, Inc.
+Copyright (c) 2017 Andrew Betts
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

package.json

@@ -4,7 +4,7 @@
     "node": ">=16"
   },
   "description": "GrowthBook proxy server for caching, realtime updates, telemetry, etc",
-  "version": "1.0.7",
+  "version": "1.0.8",
   "main": "dist/app.js",
   "license": "MIT",
   "repository": {
@@ -30,7 +30,6 @@
     "pino-http": "^8.3.1",
     "redis": "^4.5.1",
     "spdy": "^4.0.2",
-    "sse-pubsub": "^1.4.1",
     "uuid": "^9.0.0"
   },
   "devDependencies": {

src/app.ts

@@ -8,7 +8,10 @@ import { featuresRouter } from "./controllers/featuresController";
 import proxyMiddleware from "./middleware/proxyMiddleware";
 import { featuresCache, initializeCache } from "./services/cache";
 import { initializeRegistrar, registrar } from "./services/registrar";
-import { eventStreamManager } from "./services/eventStreamManager";
+import {
+  eventStreamManager,
+  initializeEventStreamManager,
+} from "./services/eventStreamManager";
 import { Context, GrowthBookProxy } from "./types";
 import logger, { initializeLogger } from "./services/logger";
 
@@ -70,6 +73,7 @@ export const growthBookProxy = async (
   initializeLogger(ctx);
   await initializeRegistrar(ctx);
   ctx.enableCache && (await initializeCache(ctx));
+  ctx.enableEventStream && initializeEventStreamManager(ctx);
 
   // set up handlers
   ctx.enableCors && app.use(cors());

src/controllers/eventStreamController.ts

@@ -4,7 +4,9 @@ import { apiKeyMiddleware } from "../middleware/apiKeyMiddleware";
 import { validateEventStreamChannelMiddleware } from "../middleware/eventStream/validateEventStreamChannelMiddleware";
 
 const getSubscribeToEventStream = async (req: Request, res: Response) => {
-  eventStreamManager.subscribe(req, res);
+  if (eventStreamManager) {
+    eventStreamManager.subscribe(req, res);
+  }
 };
 
 export const eventStreamRouter = express.Router();

src/middleware/cache/readThroughCacheMiddleware.ts

@@ -24,7 +24,7 @@ const interceptor = (proxyTarget: string) =>
       errorCounts[proxyTarget] = 0;
 
       const response = responseBuffer.toString("utf-8");
-      if (!featuresCache) {
+      if (!featuresCache || !eventStreamManager) {
         return response;
       }
 

src/middleware/cache/refreshStaleCacheMiddleware.ts

@@ -3,11 +3,14 @@ import { Request, Response } from "express";
 import { featuresCache } from "../../services/cache";
 import logger from "../../services/logger";
 import { eventStreamManager } from "../../services/eventStreamManager";
+import { Context } from "../../types";
 
 const activeFetchUrls = new Set<string>();
 
 export default ({ proxyTarget }: { proxyTarget: string }) =>
   async (req: Request, res: Response) => {
+    const ctx = req.app.locals?.ctx as Context;
+
     if (!featuresCache) {
       return;
     }
@@ -32,7 +35,14 @@ export default ({ proxyTarget }: { proxyTarget: string }) =>
       const oldEntry = await featuresCache.get(apiKey);
       await featuresCache.set(apiKey, entry);
 
-      eventStreamManager.publish(apiKey, "features", entry, oldEntry?.payload);
+      if (ctx.enableEventStream && eventStreamManager) {
+        eventStreamManager.publish(
+          apiKey,
+          "features",
+          entry,
+          oldEntry?.payload
+        );
+      }
     } else {
       logger.error("Unable to parse response");
     }

src/middleware/eventStream/broadcastEventStreamMiddleware.ts

@@ -12,20 +12,14 @@ export const broadcastEventStreamMiddleware = async (
   const ctx = req.app.locals?.ctx as Context;
   ctx?.verboseDebugging && logger.info("broadcastEventStreamMiddleware");
 
-  if (ctx?.enableEventStream) {
+  if (ctx?.enableEventStream && eventStreamManager) {
     const apiKey = res.locals.apiKey;
 
     const oldEntry = featuresCache
       ? await featuresCache.get(apiKey)
       : undefined;
 
-    eventStreamManager.publish(
-      apiKey,
-      "features",
-      req.body,
-      oldEntry?.payload,
-      ctx
-    );
+    eventStreamManager.publish(apiKey, "features", req.body, oldEntry?.payload);
   }
   next();
 };

src/services/cache/RedisCache.ts

@@ -169,7 +169,7 @@ export class RedisCache {
           if (uuid === this.clientUUID) return;
 
           // 1. emit SSE to SDK clients (if new payload !== old payload)
-          if (this.appContext?.enableEventStream) {
+          if (this.appContext?.enableEventStream && eventStreamManager) {
             const oldEntry = await this.get(key);
             eventStreamManager.publish(
               key,

src/services/eventStreamManager/index.ts

@@ -1,38 +1,9 @@
 import { Request, Response } from "express";
 import logger from "../logger";
 import { Context } from "../../types";
-const SSEChannel = require("sse-pubsub");
+import { SSEChannel, Options } from "./ssePubsub";
 
-// START hacky TS binding for sse-pubsub
-interface SSEChannel {
-  /* eslint-disable @typescript-eslint/no-explicit-any */
-  constructor: (options: SSEChannelOptions) => void;
-  /* eslint-disable @typescript-eslint/no-explicit-any */
-  publish: (data: any, eventName: string) => number;
-  subscribe: (
-    req: Request,
-    res: Response,
-    /* eslint-disable @typescript-eslint/no-explicit-any */
-    events?: any[]
-    /* eslint-disable @typescript-eslint/no-explicit-any */
-  ) => { req: Request; res: Response; events?: any[] };
-  /* eslint-disable @typescript-eslint/no-explicit-any */
-  unsubscribe: (c: { req: Request; res: Response; events?: any[] }) => void;
-  close: () => void;
-  listClients: () => { [ip: string]: number };
-  getSubscriberCount: () => number;
-}
-interface SSEChannelOptions {
-  pingInterval?: number; // default 3000
-  maxStreamDuration?: number; // default 30000
-  clientRetryInterval?: number; // default 1000
-  startId?: number; // default 1
-  historySize?: number; // default 100
-  rewind?: number; // default 0
-}
-// END hacky TS binding for sse-pubsub
-
-const defaultOptions: SSEChannelOptions = {
+const defaultOptions: Partial<Options> = {
   historySize: 1,
 };
 
@@ -41,9 +12,15 @@ interface ScopedChannel {
   channel: SSEChannel;
 }
 
-export class EventStreamManager {
+export class SSEManager {
   private scopedChannels = new Map<string, ScopedChannel>();
 
+  private appContext: Context;
+
+  constructor(appContext: Context) {
+    this.appContext = appContext;
+  }
+
   public subscribe(req: Request, res: Response) {
     const apiKey = res.locals.apiKey;
     if (apiKey) {
@@ -92,49 +69,55 @@ export class EventStreamManager {
     /* eslint-disable @typescript-eslint/no-explicit-any */
     payload: any,
     /* eslint-disable @typescript-eslint/no-explicit-any */
-    oldPayload?: any,
-    ctx?: Context
+    oldPayload?: any
   ) {
-    ctx?.verboseDebugging &&
+    this.appContext?.verboseDebugging &&
       logger.info(
         { apiKey, event, payload, oldPayload },
         "EventStreamManager.publish"
       );
     const scopedChannel = this.getScopedChannel(apiKey);
     if (scopedChannel) {
       if (oldPayload === undefined) {
-        ctx?.verboseDebugging &&
+        this.appContext?.verboseDebugging &&
           logger.info({ payload, event }, "publishing SSE");
         scopedChannel.channel.publish(payload, event);
       } else {
         const hasChanges =
           JSON.stringify(payload) !== JSON.stringify(oldPayload);
         if (hasChanges) {
-          ctx?.verboseDebugging &&
+          this.appContext?.verboseDebugging &&
             logger.info({ payload, event }, "publishing SSE");
           scopedChannel.channel.publish(payload, event);
           return;
         }
-        ctx?.verboseDebugging &&
+        this.appContext?.verboseDebugging &&
           logger.info({ payload, event }, "skipping SSE publish, no changes");
       }
       return;
     }
-    ctx?.verboseDebugging && logger.info("No scoped channel found");
+    this.appContext?.verboseDebugging && logger.info("No scoped channel found");
   }
 
   private getScopedChannel(apiKey: string): ScopedChannel | undefined {
     let scopedChannel = this.scopedChannels.get(apiKey);
     if (!scopedChannel) {
       this.scopedChannels.set(apiKey, {
         apiKey,
-        channel: new SSEChannel(defaultOptions),
+        channel: new SSEChannel(defaultOptions, this.appContext),
       });
       scopedChannel = this.scopedChannels.get(apiKey);
     }
     return scopedChannel;
   }
 }
 
-export const eventStreamManager = new EventStreamManager();
-Object.freeze(eventStreamManager);
+export type EventStreamManager = SSEManager | null;
+export let eventStreamManager: SSEManager | null = null;
+
+export const initializeEventStreamManager = (appContext: Context) => {
+  if (appContext.enableEventStream) {
+    eventStreamManager = new SSEManager(appContext);
+  }
+  Object.freeze(eventStreamManager);
+};