Merge pull request #9 from drift-labs/master

WS Server on Mainnet

.gitignore

@@ -140,3 +140,6 @@ src/**.js.map
 
 .env
 lib
+src/playground.ts
+
+*dump.rdb

example/wsClient.ts

@@ -0,0 +1,32 @@
+import WebSocket from 'ws';
+const ws = new WebSocket('wss://master.dlob.drift.trade/ws');
+import { sleep } from '../src/utils/utils';
+
+ws.on('open', async () => {
+  console.log('Connected to the server');
+  ws.send(JSON.stringify({ type: 'subscribe', channel: 'SOL-PERP' }));
+  ws.send(JSON.stringify({ type: 'subscribe', channel: 'LINK-PERP' }));
+  ws.send(JSON.stringify({ type: 'subscribe', channel: 'INJ-PERP' }));
+  await sleep(5000);
+  
+  ws.send(JSON.stringify({ type: 'unsubscribe', channel: 'SOL-PERP' }));
+  console.log("####################");
+});
+
+ws.on('message', (data: WebSocket.Data) => {
+  try {
+    const message = JSON.parse(data.toString());
+    console.log(`Received data from market ${message.channel}`);
+    // book data is in message.data
+  } catch (e) {
+    console.error('Invalid message:', data);
+  }
+});
+
+ws.on('close', () => {
+  console.log('Disconnected from the server');
+});
+
+ws.on('error', (error: Error) => {
+  console.error('WebSocket error:', error);
+});

package.json

@@ -13,6 +13,8 @@
     "@project-serum/anchor": "^0.19.1-beta.1",
     "@project-serum/serum": "^0.13.65",
     "@solana/web3.js": "^1.22.0",
+    "@types/redis": "^4.0.11",
+    "@types/ws": "^8.5.8",
     "async-mutex": "^0.4.0",
     "commander": "^9.4.0",
     "compression": "^1.7.4",
@@ -20,10 +22,14 @@
     "dotenv": "^10.0.0",
     "express": "^4.18.2",
     "express-rate-limit": "^6.7.0",
+    "ioredis": "^5.3.2",
     "morgan": "^1.10.0",
+    "redis": "^4.6.10",
     "response-time": "^2.3.2",
+    "socket.io-redis": "^6.1.1",
     "typescript": "4.5.4",
-    "winston": "^3.8.1"
+    "winston": "^3.8.1",
+    "ws": "^8.14.2"
   },
   "devDependencies": {
     "@types/k6": "^0.45.0",
@@ -43,6 +49,8 @@
     "clean": "rm -rf lib",
     "start": "node lib/index.js",
     "dev": "ts-node src/index.ts",
+    "ws-publish": "ts-node src/wsPublish.ts",
+    "ws-manager": "ts-node src/wsConnectionManager.ts",
     "dev:inspect": "yarn build && node --inspect ./lib/index.js",
     "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js",
     "example": "ts-node example/client.ts",
@@ -50,6 +58,7 @@
     "prettify": "prettier --check './src/**/*.ts'",
     "prettify:fix": "prettier --write './src/**/*.ts'",
     "lint": "eslint . --ext ts --quiet",
-    "lint:fix": "eslint . --ext ts --fix"
+    "lint:fix": "eslint . --ext ts --fix",
+    "playground": "ts-node src/playground.ts"
   }
 }

src/core/metrics.ts

@@ -0,0 +1,168 @@
+import { ObservableResult } from '@opentelemetry/api';
+import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';
+import { logger } from '../utils/logger';
+import {
+	ExplicitBucketHistogramAggregation,
+	InstrumentType,
+	MeterProvider,
+	View,
+} from '@opentelemetry/sdk-metrics-base';
+import {
+	commitHash,
+	driftEnv,
+	endpoint,
+	getSlotHealthCheckInfo,
+	wsEndpoint,
+} from '..';
+
+/**
+ * Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size".
+ * @param start
+ * @param increment
+ * @param count
+ * @returns
+ */
+const createHistogramBuckets = (
+	start: number,
+	increment: number,
+	count: number
+) => {
+	return new ExplicitBucketHistogramAggregation(
+		Array.from(new Array(count), (_, i) => start + i * increment)
+	);
+};
+
+enum METRIC_TYPES {
+	runtime_specs = 'runtime_specs',
+	endpoint_response_times_histogram = 'endpoint_response_times_histogram',
+	endpoint_response_status = 'endpoint_response_status',
+	health_status = 'health_status',
+}
+
+export enum HEALTH_STATUS {
+	Ok = 0,
+	StaleBulkAccountLoader,
+	UnhealthySlotSubscriber,
+	LivenessTesting,
+}
+
+const metricsPort =
+	parseInt(process.env.METRICS_PORT) || PrometheusExporter.DEFAULT_OPTIONS.port;
+const { endpoint: defaultEndpoint } = PrometheusExporter.DEFAULT_OPTIONS;
+const exporter = new PrometheusExporter(
+	{
+		port: metricsPort,
+		endpoint: defaultEndpoint,
+	},
+	() => {
+		logger.info(
+			`prometheus scrape endpoint started: http://localhost:${metricsPort}${defaultEndpoint}`
+		);
+	}
+);
+const meterName = 'dlob-meter';
+const meterProvider = new MeterProvider({
+	views: [
+		new View({
+			instrumentName: METRIC_TYPES.endpoint_response_times_histogram,
+			instrumentType: InstrumentType.HISTOGRAM,
+			meterName,
+			aggregation: createHistogramBuckets(0, 20, 30),
+		}),
+	],
+});
+meterProvider.addMetricReader(exporter);
+const meter = meterProvider.getMeter(meterName);
+
+const runtimeSpecsGauge = meter.createObservableGauge(
+	METRIC_TYPES.runtime_specs,
+	{
+		description: 'Runtime sepcification of this program',
+	}
+);
+const bootTimeMs = Date.now();
+runtimeSpecsGauge.addCallback((obs) => {
+	obs.observe(bootTimeMs, {
+		commit: commitHash,
+		driftEnv,
+		rpcEndpoint: endpoint,
+		wsEndpoint: wsEndpoint,
+	});
+});
+
+let healthStatus: HEALTH_STATUS = HEALTH_STATUS.Ok;
+const healthStatusGauge = meter.createObservableGauge(
+	METRIC_TYPES.health_status,
+	{
+		description: 'Health status of this program',
+	}
+);
+healthStatusGauge.addCallback((obs: ObservableResult) => {
+	obs.observe(healthStatus, {});
+});
+
+const endpointResponseTimeHistogram = meter.createHistogram(
+	METRIC_TYPES.endpoint_response_times_histogram,
+	{
+		description: 'Duration of endpoint responses',
+		unit: 'ms',
+	}
+);
+
+const responseStatusCounter = meter.createCounter(
+	METRIC_TYPES.endpoint_response_status,
+	{
+		description: 'Count of endpoint responses by status code',
+	}
+);
+
+const healthCheckInterval = 2000;
+let lastHealthCheckSlot = -1;
+let lastHealthCheckSlotUpdated = Date.now();
+const handleHealthCheck = async (req, res, next) => {
+	const { lastSlotReceived, lastSlotReceivedMutex } = getSlotHealthCheckInfo();
+
+	try {
+		if (req.url === '/health' || req.url === '/') {
+			// check if a slot was received recently
+			let healthySlotSubscriber = false;
+			await lastSlotReceivedMutex.runExclusive(async () => {
+				const slotChanged = lastSlotReceived > lastHealthCheckSlot;
+				const slotChangedRecently =
+					Date.now() - lastHealthCheckSlotUpdated < healthCheckInterval;
+				healthySlotSubscriber = slotChanged || slotChangedRecently;
+				logger.debug(
+					`Slotsubscriber health check: lastSlotReceived: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, slotChanged: ${slotChanged}, slotChangedRecently: ${slotChangedRecently}`
+				);
+				if (slotChanged) {
+					lastHealthCheckSlot = lastSlotReceived;
+					lastHealthCheckSlotUpdated = Date.now();
+				}
+			});
+			if (!healthySlotSubscriber) {
+				healthStatus = HEALTH_STATUS.UnhealthySlotSubscriber;
+				logger.error(`SlotSubscriber is not healthy`);
+
+				res.writeHead(500);
+				res.end(`SlotSubscriber is not healthy`);
+				return;
+			}
+
+			// liveness check passed
+			healthStatus = HEALTH_STATUS.Ok;
+			res.writeHead(200);
+			res.end('OK');
+		} else {
+			res.writeHead(404);
+			res.end('Not found');
+		}
+	} catch (e) {
+		next(e);
+	}
+};
+
+export {
+	endpointResponseTimeHistogram,
+	responseStatusCounter,
+	handleHealthCheck,
+};

src/core/middleware.ts

@@ -0,0 +1,26 @@
+import { Request, Response } from 'express';
+import responseTime = require('response-time');
+import {
+	endpointResponseTimeHistogram,
+	responseStatusCounter,
+} from './metrics';
+
+export const handleResponseTime = responseTime(
+	(req: Request, res: Response, time: number) => {
+		const endpoint = req.path;
+
+		if (endpoint === '/health' || req.url === '/') {
+			return;
+		}
+
+		responseStatusCounter.add(1, {
+			endpoint,
+			status: res.statusCode,
+		});
+
+		const responseTimeMs = time;
+		endpointResponseTimeHistogram.record(responseTimeMs, {
+			endpoint,
+		});
+	}
+);

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -0,0 +1,114 @@
+import {
+	BN,
+	DLOBSubscriber,
+	DLOBSubscriptionConfig,
+	DevnetPerpMarkets,
+	DevnetSpotMarkets,
+	L2OrderBookGenerator,
+	MainnetPerpMarkets,
+	MainnetSpotMarkets,
+	MarketType,
+	groupL2,
+} from '@drift-labs/sdk';
+import { getOracleForMarket, l2WithBNToStrings } from '../utils/utils';
+import { RedisClient } from '../utils/redisClient';
+import { driftEnv } from '../wsPublish';
+
+type wsMarketL2Args = {
+	marketIndex: number;
+	marketType: MarketType;
+	marketName: string;
+	depth: number;
+	includeVamm: boolean;
+	numVammOrders?: number;
+	grouping?: number;
+	fallbackL2Generators?: L2OrderBookGenerator[];
+	updateOnChange?: boolean;
+};
+
+export class DLOBSubscriberIO extends DLOBSubscriber {
+	public marketL2Args: wsMarketL2Args[] = [];
+	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
+	redisClient: RedisClient;
+
+	constructor(config: DLOBSubscriptionConfig & { redisClient: RedisClient }) {
+		super(config);
+		this.redisClient = config.redisClient;
+
+		// Set up appropriate maps
+		this.lastSeenL2Formatted = new Map();
+		this.lastSeenL2Formatted.set(MarketType.SPOT, new Map());
+		this.lastSeenL2Formatted.set(MarketType.PERP, new Map());
+
+		// Add all active markets to the market L2Args
+		const perpMarkets =
+			driftEnv === 'devnet' ? DevnetPerpMarkets : MainnetPerpMarkets;
+		const spotMarkets =
+			driftEnv === 'devnet' ? DevnetSpotMarkets : MainnetSpotMarkets;
+
+		for (const market of perpMarkets) {
+			this.marketL2Args.push({
+				marketIndex: market.marketIndex,
+				marketType: MarketType.PERP,
+				marketName: market.symbol,
+				depth: -1,
+				includeVamm: true,
+				numVammOrders: 100,
+				updateOnChange: true,
+				fallbackL2Generators: [],
+			});
+		}
+		for (const market of spotMarkets) {
+			this.marketL2Args.push({
+				marketIndex: market.marketIndex,
+				marketType: MarketType.SPOT,
+				marketName: market.symbol,
+				depth: -1,
+				includeVamm: false,
+				updateOnChange: true,
+				fallbackL2Generators: [],
+			});
+		}
+	}
+
+	override async updateDLOB(): Promise<void> {
+		await super.updateDLOB();
+		for (const l2Args of this.marketL2Args) {
+			this.getL2AndSendMsg(l2Args);
+		}
+	}
+
+	getL2AndSendMsg(l2Args: wsMarketL2Args): void {
+		const grouping = l2Args.grouping;
+		const { marketName, ...l2FuncArgs } = l2Args;
+		const l2 = this.getL2(l2FuncArgs);
+		let l2Formatted: any;
+		if (grouping) {
+			const groupingBN = new BN(grouping);
+			l2Formatted = l2WithBNToStrings(groupL2(l2, groupingBN, l2Args.depth));
+		} else {
+			l2Formatted = l2WithBNToStrings(l2);
+		}
+
+		if (l2Args.updateOnChange) {
+			if (
+				this.lastSeenL2Formatted
+					.get(l2Args.marketType)
+					?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted)
+			)
+				return;
+		}
+		this.lastSeenL2Formatted
+			.get(l2Args.marketType)
+			?.set(l2Args.marketIndex, JSON.stringify(l2Formatted));
+		l2Formatted['marketName'] = marketName;
+		l2Formatted['marketType'] = l2Args.marketType;
+		l2Formatted['marketIndex'] = l2Args.marketIndex;
+		l2Formatted['oracle'] = getOracleForMarket(
+			this.driftClient,
+			l2Args.marketType,
+			l2Args.marketIndex
+		);
+		this.redisClient.client.publish(marketName, JSON.stringify(l2Formatted));
+	}
+}

src/index.ts

@@ -1,7 +1,5 @@
 import { program } from 'commander';
-
+import express from 'express';
-import responseTime = require('response-time');
-import express, { Request, Response } from 'express';
 import rateLimit from 'express-rate-limit';
 import compression from 'compression';
 import morgan from 'morgan';
@@ -11,7 +9,6 @@ import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js';
 
 import {
 	getVariant,
-	BulkAccountLoader,
 	DriftClient,
 	initialize,
 	DriftEnv,
@@ -20,34 +17,32 @@ import {
 	DLOBOrder,
 	DLOBOrders,
 	DLOBOrdersCoder,
-	SpotMarkets,
-	PerpMarkets,
-	DLOBSubscriber,
-	MarketType,
-	SpotMarketConfig,
-	PhoenixSubscriber,
-	SerumSubscriber,
 	DLOBNode,
 	isVariant,
 	BN,
 	groupL2,
-	L2OrderBook,
 	Wallet,
 	UserStatsMap,
+	DLOBSubscriber,
 } from '@drift-labs/sdk';
 
-import { Mutex } from 'async-mutex';
+import { logger, setLogLevel } from './utils/logger';
 
-import { logger, setLogLevel } from './logger';
+import * as http from 'http';
-
-import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';
 import {
-	ExplicitBucketHistogramAggregation,
+	l2WithBNToStrings,
-	InstrumentType,
+	sleep,
-	MeterProvider,
+	getOracleForMarket,
-	View,
+	normalizeBatchQueryParams,
-} from '@opentelemetry/sdk-metrics-base';
+	SubscriberLookup,
-import { ObservableResult } from '@opentelemetry/api';
+	errorHandler,
+	getPhoenixSubscriber,
+	getSerumSubscriber,
+	validateDlobQuery,
+} from './utils/utils';
+import { handleResponseTime } from './core/middleware';
+import { handleHealthCheck } from './core/metrics';
+import { Mutex } from 'async-mutex';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -57,17 +52,11 @@ const sdkConfig = initialize({ env: process.env.ENV });
 
 const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
-
+const ORDERBOOK_UPDATE_INTERVAL = 1000;
-const bulkAccountLoaderPollingInterval = process.env
-	.BULK_ACCOUNT_LOADER_POLLING_INTERVAL
-	? parseInt(process.env.BULK_ACCOUNT_LOADER_POLLING_INTERVAL)
-	: 5000;
-const healthCheckInterval = bulkAccountLoaderPollingInterval * 2;
 
 const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
 	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
 	: 1;
-
 const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.toLowerCase() === 'true';
 
 const logFormat =
@@ -76,40 +65,14 @@ const logHttp = morgan(logFormat, {
 	skip: (_req, res) => res.statusCode < 400,
 });
 
-function errorHandler(err, _req, res, _next) {
+let driftClient: DriftClient;
-	logger.error(`errorHandler, message: ${err.message}, stack: ${err.stack}`);
-	if (!res.headersSent) {
-		res.status(500).send('Internal error');
-	}
-}
 
 const app = express();
 app.use(cors({ origin: '*' }));
 app.use(compression());
 app.set('trust proxy', 1);
 app.use(logHttp);
-
-const handleResponseTime = responseTime(
-	(req: Request, res: Response, time: number) => {
-		const endpoint = req.path;
-
-		if (endpoint === '/health' || req.url === '/') {
-			return;
-		}
-
-		responseStatusCounter.add(1, {
-			endpoint,
-			status: res.statusCode,
-		});
-
-		const responseTimeMs = time;
-		endpointResponseTimeHistogram.record(responseTimeMs, {
-			endpoint,
-		});
-	}
-);
 app.use(handleResponseTime);
-
 app.use(
 	rateLimit({
 		windowMs: 1000, // 1 second
@@ -137,6 +100,9 @@ app.use((req, _res, next) => {
 	next();
 });
 
+app.use(errorHandler);
+const server = http.createServer(app);
+
 const opts = program.opts();
 setLogLevel(opts.debug ? 'debug' : 'info');
 
@@ -147,156 +113,16 @@ logger.info(`WS endpoint:  ${wsEndpoint}`);
 logger.info(`DriftEnv:     ${driftEnv}`);
 logger.info(`Commit:       ${commitHash}`);
 
-function sleep(ms) {
-	return new Promise((resolve) => setTimeout(resolve, ms));
-}
-
-/**
- * Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size".
- * @param start
- * @param increment
- * @param count
- * @returns
- */
-const createHistogramBuckets = (
-	start: number,
-	increment: number,
-	count: number
-) => {
-	return new ExplicitBucketHistogramAggregation(
-		Array.from(new Array(count), (_, i) => start + i * increment)
-	);
-};
-
-enum METRIC_TYPES {
-	runtime_specs = 'runtime_specs',
-	endpoint_response_times_histogram = 'endpoint_response_times_histogram',
-	endpoint_response_status = 'endpoint_response_status',
-	health_status = 'health_status',
-}
-
-export enum HEALTH_STATUS {
-	Ok = 0,
-	StaleBulkAccountLoader,
-	UnhealthySlotSubscriber,
-	LivenessTesting,
-}
-
-const metricsPort =
-	parseInt(process.env.METRICS_PORT) || PrometheusExporter.DEFAULT_OPTIONS.port;
-const { endpoint: defaultEndpoint } = PrometheusExporter.DEFAULT_OPTIONS;
-const exporter = new PrometheusExporter(
-	{
-		port: metricsPort,
-		endpoint: defaultEndpoint,
-	},
-	() => {
-		logger.info(
-			`prometheus scrape endpoint started: http://localhost:${metricsPort}${defaultEndpoint}`
-		);
-	}
-);
-const meterName = 'dlob-meter';
-const meterProvider = new MeterProvider({
-	views: [
-		new View({
-			instrumentName: METRIC_TYPES.endpoint_response_times_histogram,
-			instrumentType: InstrumentType.HISTOGRAM,
-			meterName,
-			aggregation: createHistogramBuckets(0, 20, 30),
-		}),
-	],
-});
-meterProvider.addMetricReader(exporter);
-const meter = meterProvider.getMeter(meterName);
-
-const runtimeSpecsGauge = meter.createObservableGauge(
-	METRIC_TYPES.runtime_specs,
-	{
-		description: 'Runtime sepcification of this program',
-	}
-);
-const bootTimeMs = Date.now();
-runtimeSpecsGauge.addCallback((obs) => {
-	obs.observe(bootTimeMs, {
-		commit: commitHash,
-		driftEnv,
-		rpcEndpoint: endpoint,
-		wsEndpoint: wsEndpoint,
-	});
-});
-
-let healthStatus: HEALTH_STATUS = HEALTH_STATUS.Ok;
-const healthStatusGauge = meter.createObservableGauge(
-	METRIC_TYPES.health_status,
-	{
-		description: 'Health status of this program',
-	}
-);
-healthStatusGauge.addCallback((obs: ObservableResult) => {
-	obs.observe(healthStatus, {});
-});
-
-const endpointResponseTimeHistogram = meter.createHistogram(
-	METRIC_TYPES.endpoint_response_times_histogram,
-	{
-		description: 'Duration of endpoint responses',
-		unit: 'ms',
-	}
-);
-
-const responseStatusCounter = meter.createCounter(
-	METRIC_TYPES.endpoint_response_status,
-	{
-		description: 'Count of endpoint responses by status code',
-	}
-);
-
-const getPhoenixSubscriber = (
-	driftClient: DriftClient,
-	marketConfig: SpotMarketConfig,
-	accountLoader: BulkAccountLoader
-) => {
-	return new PhoenixSubscriber({
-		connection: driftClient.connection,
-		programId: new PublicKey(sdkConfig.PHOENIX),
-		marketAddress: marketConfig.phoenixMarket,
-		accountSubscription: {
-			type: 'polling',
-			accountLoader,
-		},
-	});
-};
-
-const getSerumSubscriber = (
-	driftClient: DriftClient,
-	marketConfig: SpotMarketConfig,
-	accountLoader: BulkAccountLoader
-) => {
-	return new SerumSubscriber({
-		connection: driftClient.connection,
-		programId: new PublicKey(sdkConfig.SERUM_V3),
-		marketAddress: marketConfig.serumMarket,
-		accountSubscription: {
-			type: 'polling',
-			accountLoader,
-		},
-	});
-};
-
-type SubscriberLookup = {
-	[marketIndex: number]: {
-		phoenix?: PhoenixSubscriber;
-		serum?: SerumSubscriber;
-	};
-};
-
 let MARKET_SUBSCRIBERS: SubscriberLookup = {};
 
-const initializeAllMarketSubscribers = async (
+const lastSlotReceivedMutex = new Mutex();
-	driftClient: DriftClient,
+let lastSlotReceived: number;
-	bulkAccountLoader: BulkAccountLoader
+
-) => {
+export const getSlotHealthCheckInfo = () => {
+	return { lastSlotReceived, lastSlotReceivedMutex };
+};
+
+const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
 	const markets: SubscriberLookup = {};
 
 	for (const market of sdkConfig.SPOT_MARKETS) {
@@ -309,7 +135,7 @@ const initializeAllMarketSubscribers = async (
 			const phoenixSubscriber = getPhoenixSubscriber(
 				driftClient,
 				market,
-				bulkAccountLoader
+				sdkConfig
 			);
 			await phoenixSubscriber.subscribe();
 			markets[market.marketIndex].phoenix = phoenixSubscriber;
@@ -319,7 +145,7 @@ const initializeAllMarketSubscribers = async (
 			const serumSubscriber = getSerumSubscriber(
 				driftClient,
 				market,
-				bulkAccountLoader
+				sdkConfig
 			);
 			await serumSubscriber.subscribe();
 			markets[market.marketIndex].serum = serumSubscriber;
@@ -338,21 +164,13 @@ const main = async () => {
 		commitment: stateCommitment,
 	});
 
-	const bulkAccountLoader = new BulkAccountLoader(
+	driftClient = new DriftClient({
-		connection,
-		stateCommitment,
-		bulkAccountLoaderPollingInterval
-	);
-	const lastBulkAccountLoaderSlotMutex = new Mutex();
-	let lastBulkAccountLoaderSlot = bulkAccountLoader.mostRecentSlot;
-	let lastBulkAccountLoaderSlotUpdated = Date.now();
-	const driftClient = new DriftClient({
 		connection,
 		wallet,
 		programID: clearingHousePublicKey,
 		accountSubscription: {
-			type: 'polling',
+			type: 'websocket',
-			accountLoader: bulkAccountLoader,
+			resubTimeoutMs: 60000,
 		},
 		env: driftEnv,
 		userStats: true,
@@ -360,11 +178,6 @@ const main = async () => {
 
 	const dlobCoder = DLOBOrdersCoder.create();
 	const slotSubscriber = new SlotSubscriber(connection, {});
-	const lastSlotReceivedMutex = new Mutex();
-	let lastSlotReceived: number;
-	let lastHealthCheckSlot = -1;
-	let lastHealthCheckSlotUpdated = Date.now();
-	const startupTime = Date.now();
 
 	const lamportsBalance = await connection.getBalance(wallet.publicKey);
 	logger.info(
@@ -402,92 +215,11 @@ const main = async () => {
 		driftClient,
 		dlobSource: userMap,
 		slotSource: slotSubscriber,
-		updateFrequency: bulkAccountLoaderPollingInterval,
+		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
 	});
 	await dlobSubscriber.subscribe();
 
-	MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(
+	MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
-		driftClient,
-		bulkAccountLoader
-	);
-
-	const handleHealthCheck = async (req, res, next) => {
-		try {
-			if (req.url === '/health' || req.url === '/') {
-				if (opts.testLiveness) {
-					if (Date.now() > startupTime + 60 * 1000) {
-						healthStatus = HEALTH_STATUS.LivenessTesting;
-
-						res.writeHead(500);
-						res.end('Testing liveness test fail');
-						return;
-					}
-				}
-				// check if a slot was received recently
-				let healthySlotSubscriber = false;
-				await lastSlotReceivedMutex.runExclusive(async () => {
-					const slotChanged = lastSlotReceived > lastHealthCheckSlot;
-					const slotChangedRecently =
-						Date.now() - lastHealthCheckSlotUpdated < healthCheckInterval;
-					healthySlotSubscriber = slotChanged || slotChangedRecently;
-					logger.debug(
-						`Slotsubscriber health check: lastSlotReceived: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, slotChanged: ${slotChanged}, slotChangedRecently: ${slotChangedRecently}`
-					);
-					if (slotChanged) {
-						lastHealthCheckSlot = lastSlotReceived;
-						lastHealthCheckSlotUpdated = Date.now();
-					}
-				});
-				if (!healthySlotSubscriber) {
-					healthStatus = HEALTH_STATUS.UnhealthySlotSubscriber;
-					logger.error(`SlotSubscriber is not healthy`);
-
-					res.writeHead(500);
-					res.end(`SlotSubscriber is not healthy`);
-					return;
-				}
-
-				if (bulkAccountLoader) {
-					let healthyBulkAccountLoader = false;
-					await lastBulkAccountLoaderSlotMutex.runExclusive(async () => {
-						const slotChanged =
-							bulkAccountLoader.mostRecentSlot > lastBulkAccountLoaderSlot;
-						const slotChangedRecently =
-							Date.now() - lastBulkAccountLoaderSlotUpdated <
-							healthCheckInterval;
-						healthyBulkAccountLoader = slotChanged || slotChangedRecently;
-						logger.debug(
-							`BulkAccountLoader health check: bulkAccountLoader.mostRecentSlot: ${bulkAccountLoader.mostRecentSlot}, lastBulkAccountLoaderSlot: ${lastBulkAccountLoaderSlot}, slotChanged: ${slotChanged}, slotChangedRecently: ${slotChangedRecently}`
-						);
-						if (slotChanged) {
-							lastBulkAccountLoaderSlot = bulkAccountLoader.mostRecentSlot;
-							lastBulkAccountLoaderSlotUpdated = Date.now();
-						}
-					});
-					if (!healthyBulkAccountLoader) {
-						healthStatus = HEALTH_STATUS.StaleBulkAccountLoader;
-						logger.error(
-							`Health check failed due to stale bulkAccountLoader.mostRecentSlot`
-						);
-
-						res.writeHead(501);
-						res.end(`bulkAccountLoader.mostRecentSlot is not healthy`);
-						return;
-					}
-				}
-
-				// liveness check passed
-				healthStatus = HEALTH_STATUS.Ok;
-				res.writeHead(200);
-				res.end('OK');
-			} else {
-				res.writeHead(404);
-				res.end('Not found');
-			}
-		} catch (e) {
-			next(e);
-		}
-	};
 
 	const handleStartup = async (_req, res, _next) => {
 		if (
@@ -503,7 +235,6 @@ const main = async () => {
 		}
 	};
 
-	// start http server listening to /health endpoint using http package
 	app.get('/health', handleHealthCheck);
 	app.get('/startup', handleStartup);
 	app.get('/', handleHealthCheck);
@@ -513,7 +244,7 @@ const main = async () => {
 			// object with userAccount key and orders object serialized
 			const orders: Array<any> = [];
 			const oracles: Array<any> = [];
-			const slot = bulkAccountLoader.mostRecentSlot;
+			const slot = slotSubscriber.currentSlot;
 
 			for (const market of driftClient.getPerpMarketAccounts()) {
 				const oracle = driftClient.getOracleDataForPerpMarket(
@@ -557,7 +288,7 @@ const main = async () => {
 	app.get('/orders/json', async (_req, res, next) => {
 		try {
 			// object with userAccount key and orders object serialized
-			const slot = bulkAccountLoader.mostRecentSlot;
+			const slot = slotSubscriber.currentSlot;
 			const orders: Array<any> = [];
 			const oracles: Array<any> = [];
 			for (const market of driftClient.getPerpMarketAccounts()) {
@@ -670,6 +401,8 @@ const main = async () => {
 		try {
 			const { marketName, marketIndex, marketType } = req.query;
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -718,7 +451,7 @@ const main = async () => {
 
 			res.end(
 				JSON.stringify({
-					slot: bulkAccountLoader.mostRecentSlot,
+					slot: slotSubscriber.currentSlot,
 					data: dlobCoder.encode(dlobOrders).toString('base64'),
 				})
 			);
@@ -727,79 +460,6 @@ const main = async () => {
 		}
 	});
 
-	const validateDlobQuery = (
-		marketType?: string,
-		marketIndex?: string,
-		marketName?: string
-	): {
-		normedMarketType?: MarketType;
-		normedMarketIndex?: number;
-		error?: string;
-	} => {
-		let normedMarketType: MarketType = undefined;
-		let normedMarketIndex: number = undefined;
-		let normedMarketName: string = undefined;
-		if (marketName === undefined) {
-			if (marketIndex === undefined || marketType === undefined) {
-				return {
-					error:
-						'Bad Request: (marketName) or (marketIndex and marketType) must be supplied',
-				};
-			}
-
-			// validate marketType
-			switch ((marketType as string).toLowerCase()) {
-				case 'spot': {
-					normedMarketType = MarketType.SPOT;
-					normedMarketIndex = parseInt(marketIndex as string);
-					const spotMarketIndicies = SpotMarkets[driftEnv].map(
-						(mkt) => mkt.marketIndex
-					);
-					if (!spotMarketIndicies.includes(normedMarketIndex)) {
-						return {
-							error: 'Bad Request: invalid marketIndex',
-						};
-					}
-					break;
-				}
-				case 'perp': {
-					normedMarketType = MarketType.PERP;
-					normedMarketIndex = parseInt(marketIndex as string);
-					const perpMarketIndicies = PerpMarkets[driftEnv].map(
-						(mkt) => mkt.marketIndex
-					);
-					if (!perpMarketIndicies.includes(normedMarketIndex)) {
-						return {
-							error: 'Bad Request: invalid marketIndex',
-						};
-					}
-					break;
-				}
-				default:
-					return {
-						error: 'Bad Request: marketType must be either "spot" or "perp"',
-					};
-			}
-		} else {
-			// validate marketName
-			normedMarketName = (marketName as string).toUpperCase();
-			const derivedMarketInfo =
-				driftClient.getMarketIndexAndType(normedMarketName);
-			if (!derivedMarketInfo) {
-				return {
-					error: 'Bad Request: unrecognized marketName',
-				};
-			}
-			normedMarketType = derivedMarketInfo.marketType;
-			normedMarketIndex = derivedMarketInfo.marketIndex;
-		}
-
-		return {
-			normedMarketType,
-			normedMarketIndex,
-		};
-	};
-
 	app.get('/topMakers', async (req, res, next) => {
 		try {
 			const {
@@ -812,6 +472,8 @@ const main = async () => {
 			} = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -903,39 +565,6 @@ const main = async () => {
 		}
 	});
 
-	const l2WithBNToStrings = (l2: L2OrderBook): any => {
-		for (const key of Object.keys(l2)) {
-			for (const idx in l2[key]) {
-				const level = l2[key][idx];
-				const sources = level['sources'];
-				for (const sourceKey of Object.keys(sources)) {
-					sources[sourceKey] = sources[sourceKey].toString();
-				}
-				l2[key][idx] = {
-					price: level.price.toString(),
-					size: level.size.toString(),
-					sources,
-				};
-			}
-		}
-		return l2;
-	};
-
-	const getOracleForMarket = (
-		marketType: MarketType,
-		marketIndex: number
-	): number => {
-		if (isVariant(marketType, 'spot')) {
-			return driftClient
-				.getOracleDataForSpotMarket(marketIndex)
-				.price.toNumber();
-		} else if (isVariant(marketType, 'perp')) {
-			return driftClient
-				.getOracleDataForPerpMarket(marketIndex)
-				.price.toNumber();
-		}
-	};
-
 	app.get('/l2', async (req, res, next) => {
 		try {
 			const {
@@ -952,6 +581,8 @@ const main = async () => {
 			} = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -1000,6 +631,7 @@ const main = async () => {
 				);
 				if (`${includeOracle}`.toLowerCase() === 'true') {
 					l2Formatted['oracle'] = getOracleForMarket(
+						driftClient,
 						normedMarketType,
 						normedMarketIndex
 					);
@@ -1012,6 +644,7 @@ const main = async () => {
 				const l2Formatted = l2WithBNToStrings(l2);
 				if (`${includeOracle}`.toLowerCase() === 'true') {
 					l2Formatted['oracle'] = getOracleForMarket(
+						driftClient,
 						normedMarketType,
 						normedMarketIndex
 					);
@@ -1025,71 +658,6 @@ const main = async () => {
 		}
 	});
 
-	/**
-	 * Takes in a req.query like: `{
-	 * 		marketName: 'SOL-PERP,BTC-PERP,ETH-PERP',
-	 * 		marketType: undefined,
-	 * 		marketIndices: undefined,
-	 * 		...
-	 * 	}` and returns a normalized object like:
-	 *
-	 * `[
-	 * 		{marketName: 'SOL-PERP', marketType: undefined, marketIndex: undefined,...},
-	 * 		{marketName: 'BTC-PERP', marketType: undefined, marketIndex: undefined,...},
-	 * 		{marketName: 'ETH-PERP', marketType: undefined, marketIndex: undefined,...}
-	 * ]`
-	 *
-	 * @param rawParams req.query object
-	 * @returns normalized query params for batch requests, or undefined if there is a mismatched length
-	 */
-	const normalizeBatchQueryParams = (rawParams: {
-		[key: string]: string | undefined;
-	}): Array<{ [key: string]: string | undefined }> => {
-		const normedParams: Array<{ [key: string]: string | undefined }> = [];
-		const parsedParams = {};
-
-		// parse the query string into arrays
-		for (const key of Object.keys(rawParams)) {
-			const rawParam = rawParams[key];
-			if (rawParam === undefined) {
-				parsedParams[key] = [];
-			} else {
-				parsedParams[key] = rawParam.split(',') || [rawParam];
-			}
-		}
-
-		// of all parsedParams, find the max length
-		const maxLength = Math.max(
-			...Object.values(parsedParams).map(
-				(param: Array<unknown>) => param.length
-			)
-		);
-
-		// all params have to be either 0 length, or maxLength to be valid
-		const values = Object.values(parsedParams);
-		const validParams = values.every(
-			(value: Array<unknown>) =>
-				value.length === 0 || value.length === maxLength
-		);
-		if (!validParams) {
-			return undefined;
-		}
-
-		// merge all params into an array of objects
-		// normalize all params to the same length, filling in undefineds
-		for (let i = 0; i < maxLength; i++) {
-			const newParam = {};
-			for (const key of Object.keys(parsedParams)) {
-				const parsedParam = parsedParams[key];
-				newParam[key] =
-					parsedParam.length === maxLength ? parsedParam[i] : undefined;
-			}
-			normedParams.push(newParam);
-		}
-
-		return normedParams;
-	};
-
 	app.get('/batchL2', async (req, res, next) => {
 		try {
 			const {
@@ -1128,6 +696,8 @@ const main = async () => {
 			const l2s = normedParams.map((normedParam) => {
 				const { normedMarketType, normedMarketIndex, error } =
 					validateDlobQuery(
+						driftClient,
+						driftEnv,
 						normedParam['marketType'] as string,
 						normedParam['marketIndex'] as string,
 						normedParam['marketName'] as string
@@ -1182,6 +752,7 @@ const main = async () => {
 					);
 					if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
 						l2Formatted['oracle'] = getOracleForMarket(
+							driftClient,
 							normedMarketType,
 							normedMarketIndex
 						);
@@ -1192,6 +763,7 @@ const main = async () => {
 					const l2Formatted = l2WithBNToStrings(l2);
 					if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
 						l2Formatted['oracle'] = getOracleForMarket(
+							driftClient,
 							normedMarketType,
 							normedMarketIndex
 						);
@@ -1212,6 +784,8 @@ const main = async () => {
 			const { marketName, marketIndex, marketType, includeOracle } = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -1238,7 +812,11 @@ const main = async () => {
 			}
 
 			if (`${includeOracle}`.toLowerCase() === 'true') {
-				l3['oracle'] = getOracleForMarket(normedMarketType, normedMarketIndex);
+				l3['oracle'] = getOracleForMarket(
+					driftClient,
+					normedMarketType,
+					normedMarketIndex
+				);
 			}
 
 			res.writeHead(200);
@@ -1248,8 +826,7 @@ const main = async () => {
 		}
 	});
 
-	app.use(errorHandler);
+	server.listen(serverPort, () => {
-	app.listen(serverPort, () => {
 		logger.info(`DLOB server listening on port http://localhost:${serverPort}`);
 	});
 };
@@ -1265,3 +842,5 @@ async function recursiveTryCatch(f: () => void) {
 }
 
 recursiveTryCatch(() => main());
+
+export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };

src/utils/redisClient.ts

@@ -0,0 +1,124 @@
+import Redis, { RedisOptions } from 'ioredis';
+import { sleep } from './utils';
+
+export const getRedisClient = (
+	host?: string,
+	port?: string,
+	password?: string,
+	db?: number,
+	opts?: RedisOptions
+): Redis => {
+	if (host && port) {
+		console.log(`Connecting to configured redis:: ${host}:${port}`);
+
+		const redisClient = new Redis({
+			host: host,
+			port: parseInt(port, 10),
+			password: password,
+			...(opts ?? {}),
+			retryStrategy: (times) => {
+				const delay = Math.min(times * 1000, 10000);
+				console.log(
+					`Reconnecting to Redis in ${delay}ms... (retries: ${times})`
+				);
+				return delay;
+			},
+			reconnectOnError: (err) => {
+				const targetError = 'ECONNREFUSED';
+				if (err.message.includes(targetError)) {
+					console.log(
+						`Redis error: ${targetError}. Attempting to reconnect...`
+					);
+					return true;
+				}
+				return false;
+			},
+			maxRetriesPerRequest: null, // unlimited retries
+		});
+
+		redisClient.on('connect', () => {
+			console.log('Connected to Redis.');
+		});
+
+		redisClient.on('error', (err) => {
+			console.error('Redis error:', err);
+		});
+
+		redisClient.on('reconnecting', () => {
+			console.log('Reconnecting to Redis...');
+		});
+
+		return redisClient;
+	}
+
+	console.log(`Using default redis`);
+	return new Redis({ db });
+};
+
+/**
+ * Wrapper around the redis client.
+ *
+ * You can hover over the underlying redis client methods for explanations of the methods, but will also include links to DOCS for some important concepts below:
+ *
+ * zRange, zRangeByScore etc.:
+ * - All of the "z" methods are methods that use sorted sets.
+ * - Sorted sets are explained here : https://redis.io/docs/data-types/sorted-sets/
+ */
+export class RedisClient {
+	public client: Redis;
+
+	connectionPromise: Promise<void>;
+
+	constructor(
+		host?: string,
+		port?: string,
+		password?: string,
+		db?: number,
+		opts?: RedisOptions
+	) {
+		this.client = getRedisClient(host, port, password, db, opts);
+	}
+
+	/**
+	 * Should avoid using this unless necessary.
+	 * @returns
+	 */
+	public forceGetClient() {
+		return this.client;
+	}
+
+	public get connected() {
+		return this?.client?.status === 'ready';
+	}
+
+	async connect() {
+		if (this.client.status === 'ready' || this.client.status === 'connect') {
+			return;
+		}
+
+		if (this.client.status === 'connecting') {
+			await sleep(100);
+			return this.connect(); // recursive call to check again
+		}
+
+		try {
+			await this.client.connect();
+		} catch (e) {
+			console.error(e);
+		}
+	}
+
+	disconnect() {
+		this.client.disconnect();
+	}
+
+	private assertConnected() {
+		if (!this.connected) {
+			throw 'Redis client not connected';
+		}
+	}
+
+	private getPrefixForMetrics(key: string) {
+		return key.split(':')?.[0];
+	}
+}

src/utils/utils.ts

@@ -0,0 +1,287 @@
+import {
+	DriftClient,
+	DriftEnv,
+	L2OrderBook,
+	MarketType,
+	PerpMarkets,
+	PhoenixSubscriber,
+	PublicKey,
+	SerumSubscriber,
+	SpotMarketConfig,
+	SpotMarkets,
+	isVariant,
+} from '@drift-labs/sdk';
+import { logger } from './logger';
+import { NextFunction, Request, Response } from 'express';
+
+export const l2WithBNToStrings = (l2: L2OrderBook): any => {
+	for (const key of Object.keys(l2)) {
+		for (const idx in l2[key]) {
+			const level = l2[key][idx];
+			const sources = level['sources'];
+			for (const sourceKey of Object.keys(sources)) {
+				sources[sourceKey] = sources[sourceKey].toString();
+			}
+			l2[key][idx] = {
+				price: level.price.toString(),
+				size: level.size.toString(),
+				sources,
+			};
+		}
+	}
+	return l2;
+};
+
+export function sleep(ms: number): Promise<void> {
+	return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+export const getOracleForMarket = (
+	driftClient: DriftClient,
+	marketType: MarketType,
+	marketIndex: number
+): number => {
+	if (isVariant(marketType, 'spot')) {
+		return driftClient.getOracleDataForSpotMarket(marketIndex).price.toNumber();
+	} else if (isVariant(marketType, 'perp')) {
+		return driftClient.getOracleDataForPerpMarket(marketIndex).price.toNumber();
+	}
+};
+
+/**
+ * Takes in a req.query like: `{
+ * 		marketName: 'SOL-PERP,BTC-PERP,ETH-PERP',
+ * 		marketType: undefined,
+ * 		marketIndices: undefined,
+ * 		...
+ * 	}` and returns a normalized object like:
+ *
+ * `[
+ * 		{marketName: 'SOL-PERP', marketType: undefined, marketIndex: undefined,...},
+ * 		{marketName: 'BTC-PERP', marketType: undefined, marketIndex: undefined,...},
+ * 		{marketName: 'ETH-PERP', marketType: undefined, marketIndex: undefined,...}
+ * ]`
+ *
+ * @param rawParams req.query object
+ * @returns normalized query params for batch requests, or undefined if there is a mismatched length
+ */
+export const normalizeBatchQueryParams = (rawParams: {
+	[key: string]: string | undefined;
+}): Array<{ [key: string]: string | undefined }> => {
+	const normedParams: Array<{ [key: string]: string | undefined }> = [];
+	const parsedParams = {};
+
+	// parse the query string into arrays
+	for (const key of Object.keys(rawParams)) {
+		const rawParam = rawParams[key];
+		if (rawParam === undefined) {
+			parsedParams[key] = [];
+		} else {
+			parsedParams[key] = rawParam.split(',') || [rawParam];
+		}
+	}
+
+	// of all parsedParams, find the max length
+	const maxLength = Math.max(
+		...Object.values(parsedParams).map((param: Array<unknown>) => param.length)
+	);
+
+	// all params have to be either 0 length, or maxLength to be valid
+	const values = Object.values(parsedParams);
+	const validParams = values.every(
+		(value: Array<unknown>) => value.length === 0 || value.length === maxLength
+	);
+	if (!validParams) {
+		return undefined;
+	}
+
+	// merge all params into an array of objects
+	// normalize all params to the same length, filling in undefineds
+	for (let i = 0; i < maxLength; i++) {
+		const newParam = {};
+		for (const key of Object.keys(parsedParams)) {
+			const parsedParam = parsedParams[key];
+			newParam[key] =
+				parsedParam.length === maxLength ? parsedParam[i] : undefined;
+		}
+		normedParams.push(newParam);
+	}
+
+	return normedParams;
+};
+
+export const validateWsSubscribeMsg = (
+	msg: any,
+	sdkConfig: any
+): { valid: boolean; msg?: string } => {
+	const maxPerpMarketIndex = Math.max(
+		...sdkConfig.PERP_MARKETS.map((m) => m.marketIndex)
+	);
+	const maxSpotMarketIndex = Math.max(
+		...sdkConfig.SPOT_MARKETS.map((m) => m.marketIndex)
+	);
+
+	if (msg['marketIndex'] < 0) {
+		return { valid: false, msg: `Invalid marketIndex, must be >= 0` };
+	}
+
+	if (
+		msg['marketType'].toLowerCase() == 'spot' &&
+		parseInt(msg['marketIndex']) > maxSpotMarketIndex
+	) {
+		return {
+			valid: false,
+			msg: `Invalid marketIndex for marketType: ${msg['marketType']}`,
+		};
+	}
+
+	if (
+		msg['marketType'].toLowerCase() == 'perp' &&
+		parseInt(msg['marketIndex']) > maxPerpMarketIndex
+	) {
+		return {
+			valid: false,
+			msg: `Invalid marketIndex for marketType: ${msg['marketType']}`,
+		};
+	}
+
+	if (
+		msg['marketType'].toLowerCase() != 'perp' &&
+		msg['marketType'] != 'spot'
+	) {
+		return {
+			valid: false,
+			msg: `Invalid marketType: ${msg['marketType']}`,
+		};
+	}
+
+	return { valid: true };
+};
+
+export const validateDlobQuery = (
+	driftClient: DriftClient,
+	driftEnv: DriftEnv,
+	marketType?: string,
+	marketIndex?: string,
+	marketName?: string
+): {
+	normedMarketType?: MarketType;
+	normedMarketIndex?: number;
+	error?: string;
+} => {
+	let normedMarketType: MarketType = undefined;
+	let normedMarketIndex: number = undefined;
+	let normedMarketName: string = undefined;
+	if (marketName === undefined) {
+		if (marketIndex === undefined || marketType === undefined) {
+			return {
+				error:
+					'Bad Request: (marketName) or (marketIndex and marketType) must be supplied',
+			};
+		}
+
+		// validate marketType
+		switch ((marketType as string).toLowerCase()) {
+			case 'spot': {
+				normedMarketType = MarketType.SPOT;
+				normedMarketIndex = parseInt(marketIndex as string);
+				const spotMarketIndicies = SpotMarkets[driftEnv].map(
+					(mkt) => mkt.marketIndex
+				);
+				if (!spotMarketIndicies.includes(normedMarketIndex)) {
+					return {
+						error: 'Bad Request: invalid marketIndex',
+					};
+				}
+				break;
+			}
+			case 'perp': {
+				normedMarketType = MarketType.PERP;
+				normedMarketIndex = parseInt(marketIndex as string);
+				const perpMarketIndicies = PerpMarkets[driftEnv].map(
+					(mkt) => mkt.marketIndex
+				);
+				if (!perpMarketIndicies.includes(normedMarketIndex)) {
+					return {
+						error: 'Bad Request: invalid marketIndex',
+					};
+				}
+				break;
+			}
+			default:
+				return {
+					error: 'Bad Request: marketType must be either "spot" or "perp"',
+				};
+		}
+	} else {
+		// validate marketName
+		normedMarketName = (marketName as string).toUpperCase();
+		const derivedMarketInfo =
+			driftClient.getMarketIndexAndType(normedMarketName);
+		if (!derivedMarketInfo) {
+			return {
+				error: 'Bad Request: unrecognized marketName',
+			};
+		}
+		normedMarketType = derivedMarketInfo.marketType;
+		normedMarketIndex = derivedMarketInfo.marketIndex;
+	}
+
+	return {
+		normedMarketType,
+		normedMarketIndex,
+	};
+};
+
+export function errorHandler(
+	err: Error,
+	_req: Request,
+	res: Response,
+	_next: NextFunction
+): void {
+	logger.error(`errorHandler, message: ${err.message}, stack: ${err.stack}`);
+	if (!res.headersSent) {
+		res.status(500).send('Internal error');
+	}
+}
+
+/**
+ * Spot market utils
+ */
+
+export const getPhoenixSubscriber = (
+	driftClient: DriftClient,
+	marketConfig: SpotMarketConfig,
+	sdkConfig
+): PhoenixSubscriber => {
+	return new PhoenixSubscriber({
+		connection: driftClient.connection,
+		programId: new PublicKey(sdkConfig.PHOENIX),
+		marketAddress: marketConfig.phoenixMarket,
+		accountSubscription: {
+			type: 'websocket',
+		},
+	});
+};
+
+export const getSerumSubscriber = (
+	driftClient: DriftClient,
+	marketConfig: SpotMarketConfig,
+	sdkConfig
+): SerumSubscriber => {
+	return new SerumSubscriber({
+		connection: driftClient.connection,
+		programId: new PublicKey(sdkConfig.SERUM_V3),
+		marketAddress: marketConfig.serumMarket,
+		accountSubscription: {
+			type: 'websocket',
+		},
+	});
+};
+
+export type SubscriberLookup = {
+	[marketIndex: number]: {
+		phoenix?: PhoenixSubscriber;
+		serum?: SerumSubscriber;
+	};
+};

src/wsConnectionManager.ts

@@ -0,0 +1,138 @@
+import cors from 'cors';
+import express from 'express';
+import * as http from 'http';
+import compression from 'compression';
+import { WebSocket, WebSocketServer } from 'ws';
+import { sleep } from './utils/utils';
+import { RedisClient } from './utils/redisClient';
+
+// Set up env constants
+require('dotenv').config();
+
+const app = express();
+app.use(cors({ origin: '*' }));
+app.use(compression());
+app.set('trust proxy', 1);
+
+const server = http.createServer(app);
+const wss = new WebSocketServer({ server, path: '/ws' });
+
+const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const WS_PORT = process.env.WS_PORT || '3000';
+const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
+
+async function main() {
+	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+	await redisClient.connect();
+
+	const channelSubscribers = new Map<string, Set<WebSocket>>();
+	const subscribedChannels = new Set<string>();
+
+	redisClient.client.on('message', (subscribedChannel, message) => {
+		const subscribers = channelSubscribers.get(subscribedChannel);
+		subscribers.forEach((ws) => {
+			ws.send(JSON.stringify({ channel: subscribedChannel, data: message }));
+		});
+	});
+
+	wss.on('connection', (ws: WebSocket) => {
+		console.log('Client connected');
+
+		ws.on('message', async (msg) => {
+			const parsedMessage = JSON.parse(msg.toString());
+
+			switch (parsedMessage.type.toLowerCase()) {
+				case 'subscribe': {
+					const channel = parsedMessage.channel;
+					if (!subscribedChannels.has(channel)) {
+						console.log('Subscribing to channel', channel);
+						redisClient.client
+							.subscribe(channel)
+							.then(() => {
+								subscribedChannels.add(channel);
+							})
+							.catch(() => {
+								ws.send(
+									JSON.stringify({
+										channel,
+										error: `Invalid channel: ${channel}`,
+									})
+								);
+								return;
+							});
+					}
+
+					if (!channelSubscribers.get(channel)) {
+						const subscribers = new Set<WebSocket>();
+						channelSubscribers.set(channel, subscribers);
+					}
+					channelSubscribers.get(channel).add(ws);
+					break;
+				}
+				case 'unsubscribe': {
+					const channel = parsedMessage.channel;
+					const subscribers = channelSubscribers.get(channel);
+					if (subscribers) {
+						channelSubscribers.get(channel).delete(ws);
+					}
+					break;
+				}
+				default:
+					break;
+			}
+		});
+
+		// Ping/pong connection timeout
+		let pongTimeoutId;
+		let isAlive = true;
+		const pingIntervalId = setInterval(() => {
+			isAlive = false;
+			pongTimeoutId = setTimeout(() => {
+				if (!isAlive) {
+					console.log('Disconnecting because of ping/pong timeout');
+					ws.terminate();
+				}
+			}, 5000); // 5 seconds to wait for a pong
+			ws.ping();
+		}, 30000);
+
+		// Listen for pong messages
+		ws.on('pong', () => {
+			console.log('poooooooong');
+			isAlive = true;
+			clearTimeout(pongTimeoutId);
+		});
+
+		// Handle disconnection
+		ws.on('close', () => {
+			// Clear any existing intervals and timeouts
+			clearInterval(pingIntervalId);
+			clearTimeout(pongTimeoutId);
+		});
+
+		ws.on('disconnect', () => {
+			console.log('Client disconnected');
+		});
+
+		ws.on('error', (error) => {
+			console.error('Socket error:', error);
+		});
+	});
+
+	server.listen(WS_PORT, () => {
+		console.log(`connection manager running on ${WS_PORT}`);
+	});
+}
+
+async function recursiveTryCatch(f: () => void) {
+	try {
+		await f();
+	} catch (e) {
+		console.error(e);
+		await sleep(15000);
+		await recursiveTryCatch(f);
+	}
+}
+
+recursiveTryCatch(() => main());

src/wsPublish.ts

@@ -0,0 +1,115 @@
+import { program } from 'commander';
+
+import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js';
+
+import {
+	DriftClient,
+	initialize,
+	DriftEnv,
+	SlotSubscriber,
+	UserMap,
+	Wallet,
+} from '@drift-labs/sdk';
+
+import { logger, setLogLevel } from './utils/logger';
+import { sleep } from './utils/utils';
+import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO';
+import { RedisClient } from './utils/redisClient';
+
+require('dotenv').config();
+const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
+const commitHash = process.env.COMMIT;
+const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
+
+//@ts-ignore
+const sdkConfig = initialize({ env: process.env.ENV });
+
+const stateCommitment: Commitment = 'processed';
+const ORDERBOOK_UPDATE_INTERVAL = 1000;
+
+let driftClient: DriftClient;
+
+const opts = program.opts();
+setLogLevel(opts.debug ? 'debug' : 'info');
+
+const endpoint = process.env.ENDPOINT;
+const wsEndpoint = process.env.WS_ENDPOINT;
+logger.info(`RPC endpoint: ${endpoint}`);
+logger.info(`WS endpoint:  ${wsEndpoint}`);
+logger.info(`DriftEnv:     ${driftEnv}`);
+logger.info(`Commit:       ${commitHash}`);
+
+const main = async () => {
+	const wallet = new Wallet(new Keypair());
+	const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
+
+	const connection = new Connection(endpoint, {
+		wsEndpoint: wsEndpoint,
+		commitment: stateCommitment,
+	});
+
+	driftClient = new DriftClient({
+		connection,
+		wallet,
+		programID: clearingHousePublicKey,
+		accountSubscription: {
+			type: 'websocket',
+		},
+		env: driftEnv,
+		userStats: true,
+	});
+
+	const slotSubscriber = new SlotSubscriber(connection, {});
+
+	const lamportsBalance = await connection.getBalance(wallet.publicKey);
+	logger.info(
+		`DriftClient ProgramId: ${driftClient.program.programId.toBase58()}`
+	);
+	logger.info(`Wallet pubkey: ${wallet.publicKey.toBase58()}`);
+	logger.info(` . SOL balance: ${lamportsBalance / 10 ** 9}`);
+
+	await driftClient.subscribe();
+	driftClient.eventEmitter.on('error', (e) => {
+		logger.info('clearing house error');
+		logger.error(e);
+	});
+
+	await slotSubscriber.subscribe();
+
+	const userMap = new UserMap(
+		driftClient,
+		driftClient.userAccountSubscriptionConfig,
+		false
+	);
+	await userMap.subscribe();
+
+	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+	await redisClient.connect();
+
+	const dlobSubscriber = new DLOBSubscriberIO({
+		driftClient,
+		dlobSource: userMap,
+		slotSource: slotSubscriber,
+		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
+		redisClient,
+	});
+	await dlobSubscriber.subscribe();
+
+	console.log('DLOBSubscriber Publishing Messages');
+};
+
+async function recursiveTryCatch(f: () => void) {
+	try {
+		await f();
+	} catch (e) {
+		console.error(e);
+		await sleep(15000);
+		await recursiveTryCatch(f);
+	}
+}
+
+recursiveTryCatch(() => main());
+
+export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };

yarn.lock

@@ -220,6 +220,11 @@
   resolved "https://registry.yarnpkg.com/@humanwhocodes/object-schema/-/object-schema-1.2.1.tgz#b520529ec21d8e5945a1851dfd1c32e94e39ff45"
   integrity sha512-ZnQMnLV4e7hDlUvw8H+U8ASL02SS2Gn6+9Ac3wGGLIe7+je2AeAOxPY+izIPJDfFDb7eDjev0Us8MO1iFRN8hA==
 
+"@ioredis/commands@^1.1.1":
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/@ioredis/commands/-/commands-1.2.0.tgz#6d61b3097470af1fdbbe622795b8921d42018e11"
+  integrity sha512-Sx1pU8EM64o2BrqNpEO1CNLtKQwyhuXuqyfH7oGKCk+1a33d2r5saW8zNwm3j6BTExtjrv2BxTgzzkMwts6vGg==
+
 "@jridgewell/resolve-uri@^3.0.3":
   version "3.1.0"
   resolved "https://registry.yarnpkg.com/@jridgewell/resolve-uri/-/resolve-uri-3.1.0.tgz#2203b118c157721addfe69d47b70465463066d78"
@@ -884,6 +889,40 @@
     assert "^2.0.0"
     buffer "^6.0.1"
 
+"@redis/bloom@1.2.0":
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.2.0.tgz#d3fd6d3c0af3ef92f26767b56414a370c7b63b71"
+  integrity sha512-HG2DFjYKbpNmVXsa0keLHp/3leGJz1mjh09f2RLGGLQZzSHpkmZWuwJbAvo3QcRY8p80m5+ZdXZdYOSBLlp7Cg==
+
+"@redis/client@1.5.11":
+  version "1.5.11"
+  resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.5.11.tgz#5ee8620fea56c67cb427228c35d8403518efe622"
+  integrity sha512-cV7yHcOAtNQ5x/yQl7Yw1xf53kO0FNDTdDU6bFIMbW6ljB7U7ns0YRM+QIkpoqTAt6zK5k9Fq0QWlUbLcq9AvA==
+  dependencies:
+    cluster-key-slot "1.1.2"
+    generic-pool "3.9.0"
+    yallist "4.0.0"
+
+"@redis/graph@1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.1.0.tgz#cc2b82e5141a29ada2cce7d267a6b74baa6dd519"
+  integrity sha512-16yZWngxyXPd+MJxeSr0dqh2AIOi8j9yXKcKCwVaKDbH3HTuETpDVPcLujhFYVPtYrngSco31BUcSa9TH31Gqg==
+
+"@redis/json@1.0.6":
+  version "1.0.6"
+  resolved "https://registry.yarnpkg.com/@redis/json/-/json-1.0.6.tgz#b7a7725bbb907765d84c99d55eac3fcf772e180e"
+  integrity sha512-rcZO3bfQbm2zPRpqo82XbW8zg4G/w4W3tI7X8Mqleq9goQjAGLL7q/1n1ZX4dXEAmORVZ4s1+uKLaUOg7LrUhw==
+
+"@redis/search@1.1.5":
+  version "1.1.5"
+  resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.1.5.tgz#682b68114049ff28fdf2d82c580044dfb74199fe"
+  integrity sha512-hPP8w7GfGsbtYEJdn4n7nXa6xt6hVZnnDktKW4ArMaFQ/m/aR7eFvsLQmG/mn1Upq99btPJk+F27IQ2dYpCoUg==
+
+"@redis/time-series@1.0.5":
+  version "1.0.5"
+  resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.0.5.tgz#a6d70ef7a0e71e083ea09b967df0a0ed742bc6ad"
+  integrity sha512-IFjIgTusQym2B5IZJG3XKr5llka7ey84fw/NOYqESP5WUfQs9zz1ww/9+qoz4ka/S6KcGBodzlCeZ5UImKbscg==
+
 "@sideway/address@^4.1.3":
   version "4.1.4"
   resolved "https://registry.yarnpkg.com/@sideway/address/-/address-4.1.4.tgz#03dccebc6ea47fdc226f7d3d1ad512955d4783f0"
@@ -1332,6 +1371,13 @@
   dependencies:
     "@types/node" "*"
 
+"@types/redis@^4.0.11":
+  version "4.0.11"
+  resolved "https://registry.yarnpkg.com/@types/redis/-/redis-4.0.11.tgz#0bb4c11ac9900a21ad40d2a6768ec6aaf651c0e1"
+  integrity sha512-bI+gth8La8Wg/QCR1+V1fhrL9+LZUSWfcqpOj2Kc80ZQ4ffbdL173vQd5wovmoV9i071FU9oP2g6etLuEwb6Rg==
+  dependencies:
+    redis "*"
+
 "@types/restify@4.3.8":
   version "4.3.8"
   resolved "https://registry.yarnpkg.com/@types/restify/-/restify-4.3.8.tgz#1e504995ef770a7149271e2cd639767cc7925245"
@@ -1355,6 +1401,13 @@
   dependencies:
     "@types/node" "*"
 
+"@types/ws@^8.5.8":
+  version "8.5.8"
+  resolved "https://registry.yarnpkg.com/@types/ws/-/ws-8.5.8.tgz#13efec7bd439d0bdf2af93030804a94f163b1430"
+  integrity sha512-flUksGIQCnJd6sZ1l5dqCEG/ksaoAg/eUwiLAGTJQcfgvZJKF++Ta4bJA6A5aPSJmsr+xlseHn4KLgVlNnvPTg==
+  dependencies:
+    "@types/node" "*"
+
 "@typescript-eslint/eslint-plugin@^4.28.0":
   version "4.33.0"
   resolved "https://registry.yarnpkg.com/@typescript-eslint/eslint-plugin/-/eslint-plugin-4.33.0.tgz#c24dc7c8069c7706bc40d99f6fa87edcb2005276"
@@ -1800,6 +1853,11 @@ chalk@^4.0.0:
     ansi-styles "^4.1.0"
     supports-color "^7.1.0"
 
+cluster-key-slot@1.1.2, cluster-key-slot@^1.1.0:
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz#88ddaa46906e303b5de30d3153b7d9fe0a0c19ac"
+  integrity sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==
+
 color-convert@^1.9.0, color-convert@^1.9.3:
   version "1.9.3"
   resolved "https://registry.yarnpkg.com/color-convert/-/color-convert-1.9.3.tgz#bb71850690e1f136567de629d2d5471deda4c1e8"
@@ -1951,7 +2009,7 @@ debug@2.6.9:
   dependencies:
     ms "2.0.0"
 
-debug@^4.0.0, debug@^4.0.1, debug@^4.1.0, debug@^4.1.1, debug@^4.3.1, debug@^4.3.3, debug@^4.3.4:
+debug@^4.0.0, debug@^4.0.1, debug@^4.1.0, debug@^4.1.1, debug@^4.3.1, debug@^4.3.3, debug@^4.3.4, debug@~4.3.1:
   version "4.3.4"
   resolved "https://registry.yarnpkg.com/debug/-/debug-4.3.4.tgz#1319f6579357f2338d3337d2cdd4914bb5dcc865"
   integrity sha512-PRWFHuSU3eDtQJPvnNY7Jcket1j0t5OuOsFzPPzsekD52Zl8qUfFIPEiswXqIvHWGVHOgX+7G/vCNNhehwxfkQ==
@@ -1981,6 +2039,16 @@ delay@^5.0.0:
   resolved "https://registry.yarnpkg.com/delay/-/delay-5.0.0.tgz#137045ef1b96e5071060dd5be60bf9334436bd1d"
   integrity sha512-ReEBKkIfe4ya47wlPYf/gu5ib6yUG0/Aez0JQZQz94kiWtRQvZIQbTiehsnwHvLSWJnQdhVeqYue7Id1dKr0qw==
 
+denque@^1.5.0:
+  version "1.5.1"
+  resolved "https://registry.yarnpkg.com/denque/-/denque-1.5.1.tgz#07f670e29c9a78f8faecb2566a1e2c11929c5cbf"
+  integrity sha512-XwE+iZ4D6ZUB7mfYRMb5wByE8L74HCn30FBN7sWnXksWc1LO1bPDl67pBR9o/kC4z/xSNAwkMYcGgqDV3BE3Hw==
+
+denque@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/denque/-/denque-2.1.0.tgz#e93e1a6569fb5e66f16a3c2a2964617d349d6ab1"
+  integrity sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==
+
 depd@2.0.0, depd@~2.0.0:
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/depd/-/depd-2.0.0.tgz#b696163cc757560d09cf22cc8fad1571b79e76df"
@@ -2502,6 +2570,11 @@ functional-red-black-tree@^1.0.1:
   resolved "https://registry.yarnpkg.com/functional-red-black-tree/-/functional-red-black-tree-1.0.1.tgz#1b0ab3bd553b2a0d6399d29c0e3ea0b252078327"
   integrity sha512-dsKNQNdj6xA3T+QlADDA7mOSlX0qiMINjn0cgr+eGHGsbSHzTabcIogz2+p/iqP1Xs6EP/sS2SbqH+brGTbq0g==
 
+generic-pool@3.9.0:
+  version "3.9.0"
+  resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.9.0.tgz#36f4a678e963f4fdb8707eab050823abc4e8f5e4"
+  integrity sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==
+
 get-intrinsic@^1.0.2, get-intrinsic@^1.1.1, get-intrinsic@^1.1.3:
   version "1.1.3"
   resolved "https://registry.yarnpkg.com/get-intrinsic/-/get-intrinsic-1.1.3.tgz#063c84329ad93e83893c7f4f243ef63ffa351385"
@@ -2668,6 +2741,21 @@ inherits@2, inherits@2.0.4, inherits@^2.0.3:
   resolved "https://registry.yarnpkg.com/inherits/-/inherits-2.0.4.tgz#0fa2c64f932917c3433a0ded55363aae37416b7c"
   integrity sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==
 
+ioredis@^5.3.2:
+  version "5.3.2"
+  resolved "https://registry.yarnpkg.com/ioredis/-/ioredis-5.3.2.tgz#9139f596f62fc9c72d873353ac5395bcf05709f7"
+  integrity sha512-1DKMMzlIHM02eBBVOFQ1+AolGjs6+xEcM4PDL7NqOS6szq7H9jSaEkIUH6/a5Hl241LzW6JLSiAbNvTQjUupUA==
+  dependencies:
+    "@ioredis/commands" "^1.1.1"
+    cluster-key-slot "^1.1.0"
+    debug "^4.3.4"
+    denque "^2.1.0"
+    lodash.defaults "^4.2.0"
+    lodash.isarguments "^3.1.0"
+    redis-errors "^1.2.0"
+    redis-parser "^3.0.0"
+    standard-as-callback "^2.1.0"
+
 ipaddr.js@1.9.1:
   version "1.9.1"
   resolved "https://registry.yarnpkg.com/ipaddr.js/-/ipaddr.js-1.9.1.tgz#bff38543eeb8984825079ff3a2a8e6cbd46781b3"
@@ -2862,6 +2950,16 @@ light-my-request@^4.2.0:
     process-warning "^1.0.0"
     set-cookie-parser "^2.4.1"
 
+lodash.defaults@^4.2.0:
+  version "4.2.0"
+  resolved "https://registry.yarnpkg.com/lodash.defaults/-/lodash.defaults-4.2.0.tgz#d09178716ffea4dde9e5fb7b37f6f0802274580c"
+  integrity sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==
+
+lodash.isarguments@^3.1.0:
+  version "3.1.0"
+  resolved "https://registry.yarnpkg.com/lodash.isarguments/-/lodash.isarguments-3.1.0.tgz#2f573d85c6a24289ff00663b491c1d338ff3458a"
+  integrity sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==
+
 lodash.merge@4.6.2, lodash.merge@^4.6.2:
   version "4.6.2"
   resolved "https://registry.yarnpkg.com/lodash.merge/-/lodash.merge-4.6.2.tgz#558aa53b43b661e1925a0afdfa36a9a1085fe57a"
@@ -3027,6 +3125,11 @@ node-gyp-build@^4.3.0:
   resolved "https://registry.yarnpkg.com/node-gyp-build/-/node-gyp-build-4.5.0.tgz#7a64eefa0b21112f89f58379da128ac177f20e40"
   integrity sha512-2iGbaQBV+ITgCz76ZEjmhUKAKVf7xfY1sRl4UiKQspfZMH2h06SyhNsnSVy50cwkFQDGLyif6m/6uFXHkOZ6rg==
 
+notepack.io@~2.2.0:
+  version "2.2.0"
+  resolved "https://registry.yarnpkg.com/notepack.io/-/notepack.io-2.2.0.tgz#d7ea71d1cb90094f88c6f3c8d84277c2d0cd101c"
+  integrity sha512-9b5w3t5VSH6ZPosoYnyDONnUTF8o0UkBw7JLA6eBlYJWyGT1Q3vQa8Hmuj1/X6RYvHjjygBDgw6fJhe0JEojfw==
+
 object-assign@^4:
   version "4.1.1"
   resolved "https://registry.yarnpkg.com/object-assign/-/object-assign-4.1.1.tgz#2109adc7965887cfc05cbbd442cac8bfbb360863"
@@ -3329,6 +3432,45 @@ real-require@^0.1.0:
   resolved "https://registry.yarnpkg.com/real-require/-/real-require-0.1.0.tgz#736ac214caa20632847b7ca8c1056a0767df9381"
   integrity sha512-r/H9MzAWtrv8aSVjPCMFpDMl5q66GqtmmRkRjpHTsp4zBAa+snZyiQNlMONiUmEJcsnaw0wCauJ2GWODr/aFkg==
 
+redis-commands@^1.7.0:
+  version "1.7.0"
+  resolved "https://registry.yarnpkg.com/redis-commands/-/redis-commands-1.7.0.tgz#15a6fea2d58281e27b1cd1acfb4b293e278c3a89"
+  integrity sha512-nJWqw3bTFy21hX/CPKHth6sfhZbdiHP6bTawSgQBlKOVRG7EZkfHbbHwQJnrE4vsQf0CMNE+3gJ4Fmm16vdVlQ==
+
+redis-errors@^1.0.0, redis-errors@^1.2.0:
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/redis-errors/-/redis-errors-1.2.0.tgz#eb62d2adb15e4eaf4610c04afe1529384250abad"
+  integrity sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==
+
+redis-parser@^3.0.0:
+  version "3.0.0"
+  resolved "https://registry.yarnpkg.com/redis-parser/-/redis-parser-3.0.0.tgz#b66d828cdcafe6b4b8a428a7def4c6bcac31c8b4"
+  integrity sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==
+  dependencies:
+    redis-errors "^1.0.0"
+
+redis@*, redis@^4.6.10:
+  version "4.6.10"
+  resolved "https://registry.yarnpkg.com/redis/-/redis-4.6.10.tgz#07f6ea2b2c5455b098e76d1e8c9b3376114e9458"
+  integrity sha512-mmbyhuKgDiJ5TWUhiKhBssz+mjsuSI/lSZNPI9QvZOYzWvYGejtb+W3RlDDf8LD6Bdl5/mZeG8O1feUGhXTxEg==
+  dependencies:
+    "@redis/bloom" "1.2.0"
+    "@redis/client" "1.5.11"
+    "@redis/graph" "1.1.0"
+    "@redis/json" "1.0.6"
+    "@redis/search" "1.1.5"
+    "@redis/time-series" "1.0.5"
+
+redis@^3.0.0:
+  version "3.1.2"
+  resolved "https://registry.yarnpkg.com/redis/-/redis-3.1.2.tgz#766851117e80653d23e0ed536254677ab647638c"
+  integrity sha512-grn5KoZLr/qrRQVwoSkmzdbw6pwF+/rwODtrOr6vuBRiR/f3rjSTGupbF90Zpqm2oenix8Do6RV7pYEkGwlKkw==
+  dependencies:
+    denque "^1.5.0"
+    redis-commands "^1.7.0"
+    redis-errors "^1.2.0"
+    redis-parser "^3.0.0"
+
 regenerator-runtime@^0.13.11:
   version "0.13.11"
   resolved "https://registry.yarnpkg.com/regenerator-runtime/-/regenerator-runtime-0.13.11.tgz#f6dca3e7ceec20590d07ada785636a90cdca17f9"
@@ -3575,6 +3717,22 @@ snake-case@^3.0.4:
     dot-case "^3.0.4"
     tslib "^2.0.3"
 
+socket.io-adapter@~2.2.0:
+  version "2.2.0"
+  resolved "https://registry.yarnpkg.com/socket.io-adapter/-/socket.io-adapter-2.2.0.tgz#43af9157c4609e74b8addc6867873ac7eb48fda2"
+  integrity sha512-rG49L+FwaVEwuAdeBRq49M97YI3ElVabJPzvHT9S6a2CWhDKnjSFasvwAwSYPRhQzfn4NtDIbCaGYgOCOU/rlg==
+
+socket.io-redis@^6.1.1:
+  version "6.1.1"
+  resolved "https://registry.yarnpkg.com/socket.io-redis/-/socket.io-redis-6.1.1.tgz#2361029a6c0b25c602d1422e1beb41907fd0e8bf"
+  integrity sha512-jeaXe3TGKC20GMSlPHEdwTUIWUpay/L7m5+S9TQcOf22p9Llx44/RkpJV08+buXTZ8E+aivOotj2RdeFJJWJJQ==
+  dependencies:
+    debug "~4.3.1"
+    notepack.io "~2.2.0"
+    redis "^3.0.0"
+    socket.io-adapter "~2.2.0"
+    uid2 "0.0.3"
+
 sonic-boom@^1.0.2:
   version "1.4.1"
   resolved "https://registry.yarnpkg.com/sonic-boom/-/sonic-boom-1.4.1.tgz#d35d6a74076624f12e6f917ade7b9d75e918f53e"
@@ -3613,6 +3771,11 @@ stack-trace@0.0.x:
   resolved "https://registry.yarnpkg.com/stack-trace/-/stack-trace-0.0.10.tgz#547c70b347e8d32b4e108ea1a2a159e5fdde19c0"
   integrity sha512-KGzahc7puUKkzyMt+IqAep+TVNbKP+k2Lmwhub39m1AsTSkaDutx56aDCo+HLDzf/D26BIHTJWNiTG1KAJiQCg==
 
+standard-as-callback@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/standard-as-callback/-/standard-as-callback-2.1.0.tgz#8953fc05359868a77b5b9739a665c5977bb7df45"
+  integrity sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==
+
 statuses@2.0.1:
   version "2.0.1"
   resolved "https://registry.yarnpkg.com/statuses/-/statuses-2.0.1.tgz#55cb000ccf1d48728bd23c685a063998cf1a1b63"
@@ -3826,6 +3989,11 @@ typescript@4.5.4:
   resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.5.4.tgz#a17d3a0263bf5c8723b9c52f43c5084edf13c2e8"
   integrity sha512-VgYs2A2QIRuGphtzFV7aQJduJ2gyfTljngLzjpfW9FoYZF6xuw1W0vW9ghCKLfcWrCFxK81CSGRAvS1pn4fIUg==
 
+uid2@0.0.3:
+  version "0.0.3"
+  resolved "https://registry.yarnpkg.com/uid2/-/uid2-0.0.3.tgz#483126e11774df2f71b8b639dcd799c376162b82"
+  integrity sha512-5gSP1liv10Gjp8cMEnFd6shzkL/D6W1uhXSFNCxDC+YI8+L8wkCYCbJ7n77Ezb4wE/xzMogecE+DtamEe9PZjg==
+
 unpipe@1.0.0, unpipe@~1.0.0:
   version "1.0.0"
   resolved "https://registry.yarnpkg.com/unpipe/-/unpipe-1.0.0.tgz#b2bf4ee8514aae6165b4817829d21b2ef49904ec"
@@ -3959,6 +4127,11 @@ ws@^7.4.5:
   resolved "https://registry.yarnpkg.com/ws/-/ws-7.5.9.tgz#54fa7db29f4c7cec68b1ddd3a89de099942bb591"
   integrity sha512-F+P9Jil7UiSKSkppIiD94dN07AwvFixvLIj1Og1Rl9GGMuNipJnV9JzjD6XuqmAeiswGvUmNLjr5cFuXwNS77Q==
 
+ws@^8.14.2:
+  version "8.14.2"
+  resolved "https://registry.yarnpkg.com/ws/-/ws-8.14.2.tgz#6c249a806eb2db7a20d26d51e7709eab7b2e6c7f"
+  integrity sha512-wEBG1ftX4jcglPxgFCMJmZ2PLtSbJ2Peg6TmpJFTbe9GZYOQCDPdMYu/Tm0/bGZkw8paZnJY45J4K2PZrLYq8g==
+
 ws@^8.5.0:
   version "8.11.0"
   resolved "https://registry.yarnpkg.com/ws/-/ws-8.11.0.tgz#6a0d36b8edfd9f96d8b25683db2f8d7de6e8e143"
@@ -3969,7 +4142,7 @@ xtend@^4.0.0:
   resolved "https://registry.yarnpkg.com/xtend/-/xtend-4.0.2.tgz#bb72779f5fa465186b1f438f674fa347fdb5db54"
   integrity sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==
 
-yallist@^4.0.0:
+yallist@4.0.0, yallist@^4.0.0:
   version "4.0.0"
   resolved "https://registry.yarnpkg.com/yallist/-/yallist-4.0.0.tgz#9bb92790d9c0effec63be73519e11a35019a3a72"
   integrity sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==