Merge pull request #169 from drift-labs/master

Elasticache migration

.github/workflows/on-sdk-update.yml

@@ -16,12 +16,12 @@ jobs:
         node-version: '18.x'
         registry-url: 'https://registry.npmjs.org'
 
+    - name: Pull git modules
+      run: git submodule update --init --recursive
+
     - name: Install dependencies
       run: yarn install
 
-    - name: Add specific version of sdk
-      run: yarn add @drift-labs/sdk@${{ github.event.client_payload.version }}
-
     - name: Build after new dependency
       run: yarn run build
 

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "drift-common"]
+	path = drift-common
+	url = https://github.com/drift-labs/drift-common

Dockerfile

@@ -5,7 +5,15 @@ RUN npm install -g yarn
 RUN npm install -g typescript
 
 WORKDIR /app
+COPY drift-common /app/drift-common
 COPY . .
+WORKDIR /app/drift-common/protocol/sdk
+RUN yarn
+RUN yarn build
+WORKDIR /app/drift-common/common-ts
+RUN yarn
+RUN yarn build
+WORKDIR /app
 RUN yarn
 RUN yarn build
 

package.json

@@ -6,7 +6,8 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@coral-xyz/anchor": "^0.29.0",
-    "@drift-labs/sdk": "2.83.0-beta.2",
+    "@drift/common": "file:./drift-common/common-ts",
+		"@drift-labs/sdk": "file:./drift-common/protocol/sdk",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/auto-instrumentations-node": "^0.31.1",
     "@opentelemetry/exporter-prometheus": "^0.31.0",

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -9,7 +9,8 @@ import {
 	groupL2,
 	isVariant,
 } from '@drift-labs/sdk';
-import { RedisClient } from '../utils/redisClient';
+import { RedisClient } from '@drift/common';
+
 import {
 	SubscriberLookup,
 	addMarketSlotToResponse,
@@ -136,6 +137,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	}
 
 	getL2AndSendMsg(marketArgs: wsMarketArgs): void {
+		const clientPrefix = this.redisClient.forceGetClient().options.keyPrefix;
 		const grouping = marketArgs.grouping;
 		const { marketName, ...l2FuncArgs } = marketArgs;
 		const l2 = this.getL2(l2FuncArgs);
@@ -261,25 +263,25 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			asks: l2Formatted.asks.slice(0, 5),
 		});
 
-		this.redisClient.client.publish(
+		this.redisClient.publish(
-			`orderbook_${marketType}_${marketArgs.marketIndex}`,
+			`${clientPrefix}orderbook_${marketType}_${marketArgs.marketIndex}`,
-			JSON.stringify(l2Formatted)
+			l2Formatted
 		);
-		this.redisClient.client.set(
+		this.redisClient.set(
 			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}`,
-			JSON.stringify(l2Formatted_depth100)
+			l2Formatted_depth100
 		);
-		this.redisClient.client.set(
+		this.redisClient.set(
 			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_100`,
-			JSON.stringify(l2Formatted_depth100)
+			l2Formatted_depth100
 		);
-		this.redisClient.client.set(
+		this.redisClient.set(
 			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_20`,
-			JSON.stringify(l2Formatted_depth20)
+			l2Formatted_depth20
 		);
-		this.redisClient.client.set(
+		this.redisClient.set(
 			`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_5`,
-			JSON.stringify(l2Formatted_depth5)
+			l2Formatted_depth5
 		);
 
 		const oraclePriceData =
@@ -306,9 +308,9 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				numMakers: 4,
 			})
 			.map((x) => x.toString());
-		this.redisClient.client.set(
+		this.redisClient.set(
 			`last_update_orderbook_best_makers_${marketType}_${marketArgs.marketIndex}`,
-			JSON.stringify({ bids, asks, slot })
+			{ bids, asks, slot }
 		);
 	}
 
@@ -361,9 +363,9 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			marketArgs.marketIndex
 		);
 
-		this.redisClient.client.set(
+		this.redisClient.set(
 			`last_update_orderbook_l3_${marketType}_${marketArgs.marketIndex}`,
-			JSON.stringify(l3)
+			l3
 		);
 	}
 }

src/index.ts

@@ -21,6 +21,7 @@ import {
 	PhoenixSubscriber,
 	BulkAccountLoader,
 } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 import { logger, setLogLevel } from './utils/logger';
 
@@ -43,10 +44,11 @@ import {
 	normalizeBatchQueryParams,
 	sleep,
 	validateDlobQuery,
+	getAccountFromId,
+	getRawAccountFromId,
 } from './utils/utils';
 import FEATURE_FLAGS from './utils/featureFlags';
 import { getDLOBProviderFromOrderSubscriber } from './dlobProvider';
-import { RedisClient } from './utils/redisClient';
 import { setGlobalDispatcher, Agent } from 'undici';
 
 setGlobalDispatcher(
@@ -58,35 +60,8 @@ setGlobalDispatcher(
 require('dotenv').config();
 
 // Reading in Redis env vars
-const REDIS_HOSTS = process.env.REDIS_HOSTS?.replace(/^\[|\]$/g, '')
+const REDIS_CLIENTS = [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
-	.split(',')
+console.log('Redis Clients:', REDIS_CLIENTS);
-	.map((host) => host.trim()) || ['localhost'];
-const REDIS_PORTS = process.env.REDIS_PORTS?.replace(/^\[|\]$/g, '')
-	.split(',')
-	.map((port) => parseInt(port.trim(), 10)) || [6379];
-const REDIS_PASSWORDS_ENV = process.env.REDIS_PASSWORDS || "['']";
-
-let REDIS_PASSWORDS;
-
-if (REDIS_PASSWORDS_ENV.trim() === "['']") {
-	REDIS_PASSWORDS = [undefined];
-} else {
-	REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.replace(/^\[|\]$/g, '')
-		.split(/\s*,\s*/)
-		.map((pwd) => pwd.replace(/(^'|'$)/g, '').trim())
-		.map((pwd) => (pwd === '' ? undefined : pwd));
-}
-
-console.log('Redis Hosts:', REDIS_HOSTS);
-console.log('Redis Ports:', REDIS_PORTS);
-console.log('Redis Passwords:', REDIS_PASSWORDS);
-
-if (
-	REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
-	REDIS_PORTS.length !== REDIS_HOSTS.length
-) {
-	throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length';
-}
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
@@ -180,7 +155,7 @@ const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
 				const bulkAccountLoader = new BulkAccountLoader(
 					driftClient.connection,
 					stateCommitment,
-					2_000
+					5_000
 				);
 				const phoenixSubscriber = new PhoenixSubscriber({
 					connection: driftClient.connection,
@@ -333,16 +308,10 @@ const main = async (): Promise<void> => {
 	> = new Map();
 	if (useRedis) {
 		logger.info('Connecting to redis');
-		for (let i = 0; i < REDIS_HOSTS.length; i++) {
+		for (let i = 0; i < REDIS_CLIENTS.length; i++) {
-			redisClients.push(
+			redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] }));
-				new RedisClient(
-					REDIS_HOSTS[i],
-					REDIS_PORTS[i].toString(),
-					REDIS_PASSWORDS[i] || undefined
-				)
-			);
-			await redisClients[i].connect();
 		}
+
 		for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
 			spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, {
 				client: redisClients[0],
@@ -361,6 +330,8 @@ const main = async (): Promise<void> => {
 		}
 	}
 
+	const userMapClient = new RedisClient({ prefix: RedisClientPrefix.USER_MAP });
+
 	function canRotate(marketType: MarketType, marketIndex: number) {
 		if (isVariant(marketType, 'spot')) {
 			const state = spotMarketRedisMap.get(marketIndex);
@@ -444,15 +415,22 @@ const main = async (): Promise<void> => {
 		try {
 			const { marketIndex, marketType } = req.query;
 
-			const fees = await redisClients[
+			const fees = await redisClients
-				parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0
+				.find((client) => {
-			].client.get(`priorityFees_${marketType}_${marketIndex}`);
+					return (
+						client.forceGetClient().options.keyPrefix ===
+						RedisClientPrefix.DLOB_HELIUS
+					);
+				})
+				.getRaw(`priorityFees_${marketType}_${marketIndex}`);
+
 			if (fees) {
 				res.status(200).json({
 					...JSON.parse(fees),
 					marketType,
 					marketIndex,
 				});
+
 				return;
 			} else {
 				res.writeHead(404);
@@ -483,11 +461,16 @@ const main = async (): Promise<void> => {
 
 			const fees = await Promise.all(
 				normedParams.map(async (normedParam) => {
-					const fees = await redisClients[
+					const fees = await redisClients
-						parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0
+						.find(
-					].client.get(
+							(client) =>
+								client.forceGetClient().options.keyPrefix ===
+								RedisClientPrefix.DLOB_HELIUS
+						)
+						.getRaw(
 							`priorityFees_${normedParam['marketType']}_${normedParam['marketIndex']}`
 						);
+
 					return {
 						...JSON.parse(fees),
 						marketType: normedParam['marketType'],
@@ -516,6 +499,7 @@ const main = async (): Promise<void> => {
 				marketType,
 				side, // bid or ask
 				limit,
+				includeAccounts,
 			} = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
@@ -550,12 +534,17 @@ const main = async (): Promise<void> => {
 				normedLimit = 4;
 			}
 
+			let accountFlag = false;
+			if (includeAccounts) {
+				accountFlag = (includeAccounts as string).toLowerCase() === 'true';
+			}
+
 			let topMakers: string[];
 			if (useRedis) {
 				const redisClient = isVariant(normedMarketType, 'perp')
 					? perpMarketRedisMap.get(normedMarketIndex).client
 					: spotMarketRedisMap.get(normedMarketIndex).client;
-				const redisResponse = await redisClient.client.get(
+				const redisResponse = await redisClient.getRaw(
 					`last_update_orderbook_best_makers_${getVariant(
 						normedMarketType
 					)}_${marketIndex}`
@@ -582,6 +571,16 @@ const main = async (): Promise<void> => {
 					path: req.baseUrl + req.path,
 				});
 				res.writeHead(200);
+
+				if (accountFlag) {
+					const topAccounts = await getRawAccountFromId(
+						userMapClient,
+						topMakers
+					);
+					res.end(JSON.stringify(topAccounts));
+					return;
+				}
+
 				res.end(JSON.stringify(topMakers));
 				return;
 			}
@@ -636,6 +635,13 @@ const main = async (): Promise<void> => {
 				path: req.baseUrl + req.path,
 			});
 			res.writeHead(200);
+
+			if (accountFlag) {
+				const topAccounts = await getAccountFromId(userMapClient, topMakers);
+				res.end(JSON.stringify(topAccounts));
+				return;
+			}
+
 			res.end(JSON.stringify(topMakers));
 		} catch (err) {
 			next(err);
@@ -678,15 +684,15 @@ const main = async (): Promise<void> => {
 					let redisL2: string;
 					const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
 					if (parseInt(adjustedDepth as string) === 5) {
-						redisL2 = await redisClient.client.get(
+						redisL2 = await redisClient.getRaw(
 							`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
 						);
 					} else if (parseInt(adjustedDepth as string) === 20) {
-						redisL2 = await redisClient.client.get(
+						redisL2 = await redisClient.getRaw(
 							`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
 						);
 					} else if (parseInt(adjustedDepth as string) === 100) {
-						redisL2 = await redisClient.client.get(
+						redisL2 = await redisClient.getRaw(
 							`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
 						);
 					}
@@ -711,15 +717,15 @@ const main = async (): Promise<void> => {
 					let redisL2: string;
 					const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
 					if (parseInt(adjustedDepth as string) === 5) {
-						redisL2 = await redisClient.client.get(
+						redisL2 = await redisClient.getRaw(
 							`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
 						);
 					} else if (parseInt(adjustedDepth as string) === 20) {
-						redisL2 = await redisClient.client.get(
+						redisL2 = await redisClient.getRaw(
 							`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
 						);
 					} else if (parseInt(adjustedDepth as string) === 100) {
-						redisL2 = await redisClient.client.get(
+						redisL2 = await redisClient.getRaw(
 							`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
 						);
 					}
@@ -851,15 +857,15 @@ const main = async (): Promise<void> => {
 							const redisClient =
 								perpMarketRedisMap.get(normedMarketIndex).client;
 							if (parseInt(adjustedDepth as string) === 5) {
-								redisL2 = await redisClient.client.get(
+								redisL2 = await redisClient.getRaw(
 									`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
 								);
 							} else if (parseInt(adjustedDepth as string) === 20) {
-								redisL2 = await redisClient.client.get(
+								redisL2 = await redisClient.getRaw(
 									`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
 								);
 							} else if (parseInt(adjustedDepth as string) === 100) {
-								redisL2 = await redisClient.client.get(
+								redisL2 = await redisClient.getRaw(
 									`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
 								);
 							}
@@ -887,15 +893,15 @@ const main = async (): Promise<void> => {
 							const redisClient =
 								spotMarketRedisMap.get(normedMarketIndex).client;
 							if (parseInt(adjustedDepth as string) === 5) {
-								redisL2 = await redisClient.client.get(
+								redisL2 = await redisClient.getRaw(
 									`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
 								);
 							} else if (parseInt(adjustedDepth as string) === 20) {
-								redisL2 = await redisClient.client.get(
+								redisL2 = await redisClient.getRaw(
 									`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
 								);
 							} else if (parseInt(adjustedDepth as string) === 100) {
-								redisL2 = await redisClient.client.get(
+								redisL2 = await redisClient.getRaw(
 									`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
 								);
 							}
@@ -993,7 +999,7 @@ const main = async (): Promise<void> => {
 				const redisClient = (
 					marketTypeStr === 'spot' ? spotMarketRedisMap : perpMarketRedisMap
 				).get(normedMarketIndex).client;
-				const redisL3 = await redisClient.client.get(
+				const redisL3 = await redisClient.getRaw(
 					`last_update_orderbook_l3_${marketTypeStr}_${normedMarketIndex}`
 				);
 				if (

src/publishers/dlobPublisher.ts

@@ -19,6 +19,7 @@ import {
 	SpotMarketConfig,
 	PhoenixSubscriber,
 } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 import { logger, setLogLevel } from '../utils/logger';
 import {
@@ -31,7 +32,6 @@ import {
 	DLOBSubscriberIO,
 	wsMarketInfo,
 } from '../dlob-subscriber/DLOBSubscriberIO';
-import { RedisClient } from '../utils/redisClient';
 import {
 	DLOBProvider,
 	getDLOBProviderFromGrpcOrderSubscriber,
@@ -54,9 +54,8 @@ require('dotenv').config();
 const stateCommitment: Commitment = 'confirmed';
 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_CLIENT = process.env.REDIS_CLIENT || 'DLOB';
-const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
 
 // Set up express for health checks
 const app = express();
@@ -275,6 +274,11 @@ const main = async () => {
 	const wallet = new Wallet(new Keypair());
 	const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
 
+	const redisClient = new RedisClient({
+		prefix: RedisClientPrefix[REDIS_CLIENT],
+	});
+	await redisClient.connect();
+
 	const connection = new Connection(endpoint, {
 		wsEndpoint: wsEndpoint,
 		commitment: stateCommitment,
@@ -407,9 +411,6 @@ const main = async () => {
 
 	await dlobProvider.subscribe();
 
-	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
-	await redisClient.connect();
-
 	const dlobSubscriber = new DLOBSubscriberIO({
 		driftClient,
 		env: driftEnv,

src/publishers/priorityFeesPublisher.ts

@@ -10,12 +10,12 @@ import {
 	BulkAccountLoader,
 	getMarketsAndOraclesForSubscription,
 } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 import { logger, setLogLevel } from '../utils/logger';
 import { sleep } from '../utils/utils';
 import express from 'express';
 // import { handleHealthCheck } from '../core/metrics';
-import { RedisClient } from '../utils/redisClient';
 import { setGlobalDispatcher, Agent } from 'undici';
 
 setGlobalDispatcher(
@@ -28,10 +28,7 @@ require('dotenv').config();
 const stateCommitment: Commitment = 'confirmed';
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
-const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const redisClientPrefix = RedisClientPrefix.DLOB_HELIUS;
-const REDIS_PORT = process.env.REDIS_PORT || '6379';
-const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
-
 // Set up express for health checks
 const app = express();
 
@@ -144,25 +141,25 @@ class PriorityFeeSubscriber {
 
 		dataPerp.forEach((result: any) => {
 			const marketIndex = parseInt(result['id']);
-			this.redisClient.client.publish(
+			this.redisClient.publish(
-				`priorityFees_perp_${marketIndex}`,
+				`${redisClientPrefix}priorityFees_perp_${marketIndex}`,
-				JSON.stringify(result.result['priorityFeeLevels'])
+				result.result['priorityFeeLevels']
 			);
-			this.redisClient.client.set(
+			this.redisClient.set(
 				`priorityFees_perp_${marketIndex}`,
-				JSON.stringify(result.result['priorityFeeLevels'])
+				result.result['priorityFeeLevels']
 			);
 		});
 
 		dataSpot.forEach((result: any) => {
 			const marketIndex = parseInt(result['id']) - 100;
-			this.redisClient.client.publish(
+			this.redisClient.publish(
-				`priorityFees_spot_${marketIndex}`,
+				`${redisClientPrefix}priorityFees_spot_${marketIndex}`,
-				JSON.stringify(result.result['priorityFeeLevels'])
+				result.result['priorityFeeLevels']
 			);
-			this.redisClient.client.set(
+			this.redisClient.set(
 				`priorityFees_spot_${marketIndex}`,
-				JSON.stringify(result.result['priorityFeeLevels'])
+				result.result['priorityFeeLevels']
 			);
 		});
 	}
@@ -174,6 +171,11 @@ const main = async () => {
 		commitment: stateCommitment,
 	});
 
+	const redisClient = new RedisClient({
+		prefix: redisClientPrefix,
+	});
+	await redisClient.connect();
+
 	const { perpMarketIndexes, spotMarketIndexes, oracleInfos } =
 		getMarketsAndOraclesForSubscription(sdkConfig.ENV);
 
@@ -216,9 +218,6 @@ const main = async () => {
 		});
 	}
 
-	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
-	await redisClient.connect();
-
 	const priorityFeeSubscriber = new PriorityFeeSubscriber({
 		endpoint,
 		perpMarketPubkeys,

src/publishers/tradesPublisher.ts

@@ -20,10 +20,10 @@ import {
 	OrderActionRecord,
 	Event,
 } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 import { logger, setLogLevel } from '../utils/logger';
 import { sleep } from '../utils/utils';
-import { RedisClient } from '../utils/redisClient';
 import { fromEvent, filter, map } from 'rxjs';
 import { setGlobalDispatcher, Agent } from 'undici';
 
@@ -36,10 +36,7 @@ setGlobalDispatcher(
 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 redisClientPrefix = RedisClientPrefix.DLOB;
-const REDIS_PORT = process.env.REDIS_PORT || '6379';
-const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
-
 //@ts-ignore
 const sdkConfig = initialize({ env: process.env.ENV });
 
@@ -77,6 +74,9 @@ const main = async () => {
 		env: driftEnv,
 	});
 
+	const redisClient = new RedisClient({ prefix: redisClientPrefix });
+	await redisClient.connect();
+
 	const slotSubscriber = new SlotSubscriber(connection, {});
 
 	const lamportsBalance = await connection.getBalance(wallet.publicKey);
@@ -93,9 +93,6 @@ const main = async () => {
 
 	await slotSubscriber.subscribe();
 
-	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
-	await redisClient.connect();
-
 	const eventSubscriber = new EventSubscriber(connection, driftClient.program, {
 		maxTx: 8192,
 		maxEventsPerType: 4096,
@@ -189,9 +186,9 @@ const main = async () => {
 			})
 		)
 		.subscribe((fillEvent) => {
-			redisClient.client.publish(
+			redisClient.publish(
-				`trades_${fillEvent.marketType}_${fillEvent.marketIndex}`,
+				`${redisClientPrefix}trades_${fillEvent.marketType}_${fillEvent.marketIndex}`,
-				JSON.stringify(fillEvent)
+				fillEvent
 			);
 		});
 

src/serverLite.ts

@@ -16,7 +16,6 @@ import {
 } from './core/metrics';
 import { handleResponseTime } from './core/middleware';
 import { errorHandler, normalizeBatchQueryParams, sleep } from './utils/utils';
-import { RedisClient } from './utils/redisClient';
 import {
 	DriftEnv,
 	MarketType,
@@ -26,36 +25,13 @@ import {
 	isVariant,
 	initialize,
 } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 require('dotenv').config();
 
 // Reading in Redis env vars
-const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost';
+const REDIS_CLIENTS = [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
-const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',')
+console.log('Redis Clients:', REDIS_CLIENTS);
-	? REDIS_HOSTS_ENV.trim()
-			.replace(/^\[|\]$/g, '')
-			.split(/\s*,\s*/)
-	: [REDIS_HOSTS_ENV];
-
-const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || '';
-const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',')
-	? REDIS_PASSWORDS_ENV.trim()
-			.replace(/^\[|\]$/g, '')
-			.split(/\s*,\s*/)
-	: [REDIS_PASSWORDS_ENV];
-
-const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379';
-const REDIS_PORTS = REDIS_PORTS_ENV.includes(',')
-	? REDIS_PORTS_ENV.trim()
-			.replace(/^\[|\]$/g, '')
-			.split(/\s*,\s*/)
-	: [REDIS_PORTS_ENV];
-if (
-	REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
-	REDIS_PORTS.length !== REDIS_HOSTS.length
-) {
-	throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length';
-}
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
@@ -145,14 +121,9 @@ const main = async () => {
 		number,
 		{ client: RedisClient; clientIndex: number }
 	> = new Map();
-	for (let i = 0; i < REDIS_HOSTS.length; i++) {
+	for (let i = 0; i < REDIS_CLIENTS.length; i++) {
-		redisClients.push(
+		const prefix = RedisClientPrefix[REDIS_CLIENTS[i]];
-			new RedisClient(
+		redisClients.push(new RedisClient({ prefix }));
-				REDIS_HOSTS[i],
-				REDIS_PORTS[i],
-				REDIS_PASSWORDS[i] || undefined
-			)
-		);
 		await redisClients[i].connect();
 	}
 	for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
@@ -245,15 +216,15 @@ const main = async () => {
 				let redisL2: string;
 				const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
 				if (parseInt(adjustedDepth as string) === 5) {
-					redisL2 = await redisClient.client.get(
+					redisL2 = await redisClient.getRaw(
 						`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
 					);
 				} else if (parseInt(adjustedDepth as string) === 20) {
-					redisL2 = await redisClient.client.get(
+					redisL2 = await redisClient.getRaw(
 						`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
 					);
 				} else if (parseInt(adjustedDepth as string) === 100) {
-					redisL2 = await redisClient.client.get(
+					redisL2 = await redisClient.getRaw(
 						`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
 					);
 				}
@@ -287,15 +258,15 @@ const main = async () => {
 				let redisL2: string;
 				const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
 				if (parseInt(adjustedDepth as string) === 5) {
-					redisL2 = await redisClient.client.get(
+					redisL2 = await redisClient.getRaw(
 						`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
 					);
 				} else if (parseInt(adjustedDepth as string) === 20) {
-					redisL2 = await redisClient.client.get(
+					redisL2 = await redisClient.getRaw(
 						`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
 					);
 				} else if (parseInt(adjustedDepth as string) === 100) {
-					redisL2 = await redisClient.client.get(
+					redisL2 = await redisClient.getRaw(
 						`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
 					);
 				}
@@ -410,15 +381,15 @@ const main = async () => {
 						const redisClient =
 							perpMarketRedisMap.get(normedMarketIndex).client;
 						if (parseInt(adjustedDepth as string) === 5) {
-							redisL2 = await redisClient.client.get(
+							redisL2 = await redisClient.getRaw(
 								`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
 							);
 						} else if (parseInt(adjustedDepth as string) === 20) {
-							redisL2 = await redisClient.client.get(
+							redisL2 = await redisClient.getRaw(
 								`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
 							);
 						} else if (parseInt(adjustedDepth as string) === 100) {
-							redisL2 = await redisClient.client.get(
+							redisL2 = await redisClient.getRaw(
 								`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
 							);
 						}
@@ -455,15 +426,15 @@ const main = async () => {
 						const redisClient =
 							spotMarketRedisMap.get(normedMarketIndex).client;
 						if (parseInt(adjustedDepth as string) === 5) {
-							redisL2 = await redisClient.client.get(
+							redisL2 = await redisClient.getRaw(
 								`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
 							);
 						} else if (parseInt(adjustedDepth as string) === 20) {
-							redisL2 = await redisClient.client.get(
+							redisL2 = await redisClient.getRaw(
 								`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
 							);
 						} else if (parseInt(adjustedDepth as string) === 100) {
-							redisL2 = await redisClient.client.get(
+							redisL2 = await redisClient.getRaw(
 								`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
 							);
 						}

src/utils/redisClient.ts

@@ -1,124 +0,0 @@
-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

@@ -11,8 +11,11 @@ import {
 	SerumSubscriber,
 	SpotMarketConfig,
 	SpotMarkets,
+	decodeUser,
 	isVariant,
 } from '@drift-labs/sdk';
+import { RedisClient } from '@drift/common';
+
 import { logger } from './logger';
 import { NextFunction, Request, Response } from 'express';
 import FEATURE_FLAGS from './featureFlags';
@@ -338,6 +341,55 @@ export const validateDlobQuery = (
 	};
 };
 
+export const getAccountFromId = async (
+	userMapClient: RedisClient,
+	topMakers: string[]
+) => {
+	return Promise.all(
+		topMakers.map(async (userAccountPubKey) => {
+			const userAccountEncoded = await userMapClient.getRaw(userAccountPubKey);
+			if (userAccountEncoded) {
+				return {
+					userAccountPubKey,
+					account: decodeUser(
+						Buffer.from(userAccountEncoded.split('::')[1], 'base64')
+					),
+				};
+			}
+			return {
+				userAccountPubKey,
+				account: null,
+			};
+		})
+	).then((results) => results.filter((user) => !!user));
+};
+
+export const getRawAccountFromId = async (
+	userMapClient: RedisClient,
+	topMakers: string[]
+): Promise<
+	{
+		userAccountPubKey: string;
+		accountBase64: string;
+	}[]
+> => {
+	return Promise.all(
+		topMakers.map(async (userAccountPubKey) => {
+			const userAccountEncoded = await userMapClient.getRaw(userAccountPubKey);
+			if (userAccountEncoded) {
+				return {
+					userAccountPubKey,
+					accountBase64: userAccountEncoded.split('::')[1],
+				};
+			}
+			return {
+				userAccountPubKey,
+				accountBase64: null,
+			};
+		})
+	).then((results) => results.filter((user) => !!user));
+};
+
 export function errorHandler(
 	err: Error,
 	_req: Request,

src/wsConnectionManager.ts

@@ -4,9 +4,9 @@ import * as http from 'http';
 import compression from 'compression';
 import { WebSocket, WebSocketServer } from 'ws';
 import { sleep } from './utils/utils';
-import { RedisClient } from './utils/redisClient';
 import { register, Gauge } from 'prom-client';
 import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 // Set up env constants
 require('dotenv').config();
@@ -29,17 +29,18 @@ const wss = new WebSocketServer({
 	perMessageDeflate: true,
 });
 
-const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
-const REDIS_PORT = process.env.REDIS_PORT || '6379';
 const WS_PORT = process.env.WS_PORT || '3000';
 
 console.log(`WS LISTENER PORT : ${WS_PORT}`);
 
-const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
 const MAX_BUFFERED_AMOUNT = 300000;
+const CHANNEL_PREFIX = RedisClientPrefix.DLOB;
+const CHANNEL_PREFIX_HELIUS = RedisClientPrefix.DLOB_HELIUS;
 
-const safeGetRawChannelFromMessage = (message: any): string => {
+const sanitiseChannelForClient = (channel: string): string => {
-	return message?.channel;
+	return channel
+		.replace(CHANNEL_PREFIX, '')
+		.replace(CHANNEL_PREFIX_HELIUS, '');
 };
 
 const getRedisChannelFromMessage = (message: any): string => {
@@ -67,11 +68,11 @@ const getRedisChannelFromMessage = (message: any): string => {
 
 	switch (channel.toLowerCase()) {
 		case 'trades':
-			return `trades_${marketType}_${marketIndex}`;
+			return `${CHANNEL_PREFIX}trades_${marketType}_${marketIndex}`;
 		case 'orderbook':
-			return `orderbook_${marketType}_${marketIndex}`;
+			return `${CHANNEL_PREFIX}orderbook_${marketType}_${marketIndex}`;
 		case 'priorityfees':
-			return `priorityFees_${marketType}_${marketIndex}`;
+			return `${CHANNEL_PREFIX_HELIUS}priorityFees_${marketType}_${marketIndex}`;
 		case undefined:
 		default:
 			throw new Error('Bad channel specified');
@@ -79,12 +80,8 @@ const getRedisChannelFromMessage = (message: any): string => {
 };
 
 async function main() {
-	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+	const redisClient = new RedisClient({});
-	const lastMessageRetriever = new RedisClient(
+	const lastMessageRetriever = new RedisClient({ prefix: CHANNEL_PREFIX });
-		REDIS_HOST,
-		REDIS_PORT,
-		REDIS_PASSWORD
-	);
 
 	await redisClient.connect();
 	await lastMessageRetriever.connect();
@@ -92,17 +89,17 @@ async function main() {
 	const channelSubscribers = new Map<string, Set<WebSocket>>();
 	const subscribedChannels = new Set<string>();
 
-	redisClient.client.on('connect', () => {
+	redisClient.forceGetClient().on('connect', () => {
 		subscribedChannels.forEach(async (channel) => {
 			try {
-				await redisClient.client.subscribe(channel);
+				await redisClient.subscribe(channel);
 			} catch (error) {
 				console.error(`Error subscribing to ${channel}:`, error);
 			}
 		});
 	});
 
-	redisClient.client.on('message', (subscribedChannel, message) => {
+	redisClient.forceGetClient().on('message', (subscribedChannel, message) => {
 		const subscribers = channelSubscribers.get(subscribedChannel);
 		if (subscribers) {
 			subscribers.forEach((ws) => {
@@ -111,13 +108,16 @@ async function main() {
 					ws.bufferedAmount < MAX_BUFFERED_AMOUNT
 				)
 					ws.send(
-						JSON.stringify({ channel: subscribedChannel, data: message })
+						JSON.stringify({
+							channel: sanitiseChannelForClient(subscribedChannel),
+							data: message,
+						})
 					);
 			});
 		}
 	});
 
-	redisClient.client.on('error', (error) => {
+	redisClient.forceGetClient().on('error', (error) => {
 		console.error('Redis client error:', error);
 	});
 
@@ -141,7 +141,7 @@ async function main() {
 					try {
 						redisChannel = getRedisChannelFromMessage(parsedMessage);
 					} catch (error) {
-						const requestChannel = safeGetRawChannelFromMessage(parsedMessage);
+						const requestChannel = sanitiseChannelForClient(parsedMessage?.channel);
 						if (requestChannel) {
 							ws.send(
 								JSON.stringify({
@@ -162,7 +162,7 @@ async function main() {
 
 					if (!subscribedChannels.has(redisChannel)) {
 						console.log('Trying to subscribe to channel', redisChannel);
-						redisClient.client
+						redisClient
 							.subscribe(redisChannel)
 							.then(() => {
 								subscribedChannels.add(redisChannel);
@@ -190,11 +190,13 @@ async function main() {
 					);
 					// Fetch and send last message
 					if (redisChannel.includes('orderbook')) {
-						const lastUpdateChannel = `last_update_${redisChannel}`;
+						const lastUpdateChannel = `last_update_${redisChannel}`.replace(
-						const lastMessage = await lastMessageRetriever.client.get(
+							CHANNEL_PREFIX,
+							''
+						);
+						const lastMessage = await lastMessageRetriever.getRaw(
 							lastUpdateChannel
 						);
-
 						if (lastMessage) {
 							ws.send(
 								JSON.stringify({
@@ -212,7 +214,7 @@ async function main() {
 					try {
 						redisChannel = getRedisChannelFromMessage(parsedMessage);
 					} catch (error) {
-						const requestChannel = safeGetRawChannelFromMessage(parsedMessage);
+						const requestChannel = sanitiseChannelForClient(parsedMessage?.channel);
 						if (requestChannel) {
 							console.log('Error unsubscribing from channel:', error.message);
 							ws.send(
@@ -270,7 +272,7 @@ async function main() {
 			clearInterval(bufferInterval);
 			channelSubscribers.forEach((subscribers, channel) => {
 				if (subscribers.delete(ws) && subscribers.size === 0) {
-					redisClient.client.unsubscribe(channel);
+					redisClient.unsubscribe(channel);
 					channelSubscribers.delete(channel);
 					subscribedChannels.delete(channel);
 				}