chore: update channel prefixes for pubsub

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -137,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);
@@ -263,7 +264,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		});
 
 		this.redisClient.publish(
-			`orderbook_${marketType}_${marketArgs.marketIndex}`,
+			`${clientPrefix}orderbook_${marketType}_${marketArgs.marketIndex}`,
 			l2Formatted
 		);
 		this.redisClient.set(

src/index.ts

@@ -51,10 +51,7 @@ import { getDLOBProviderFromOrderSubscriber } from './dlobProvider';
 require('dotenv').config();
 
 // Reading in Redis env vars
-const REDIS_CLIENTS = process.env.REDIS_CLIENTS?.replace(/^\[|\]$/g, '')
+const REDIS_CLIENTS = [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
-	.split(',')
-	.map((clients) => clients.trim()) || ['DLOB'];
-
 console.log('Redis Clients:', REDIS_CLIENTS);
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -294,8 +291,7 @@ const main = async (): Promise<void> => {
 	if (useRedis) {
 		logger.info('Connecting to redis');
 		for (let i = 0; i < REDIS_CLIENTS.length; i++) {
-			const prefix = RedisClientPrefix[REDIS_CLIENTS[i]];
+			redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] }));
-			redisClients.push(new RedisClient({ prefix }));
 		}
 
 		for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
@@ -402,11 +398,12 @@ const main = async (): Promise<void> => {
 			const { marketIndex, marketType } = req.query;
 
 			const fees = await redisClients
-				.find(
+				.find((client) => {
-					(client) =>
+					return (
 						client.forceGetClient().options.keyPrefix ===
 						RedisClientPrefix.DLOB_HELIUS
-				)
+					);
+				})
 				.getRaw(`priorityFees_${marketType}_${marketIndex}`);
 
 			if (fees) {
@@ -521,7 +518,7 @@ const main = async (): Promise<void> => {
 
 			let accountFlag = false;
 			if (includeAccounts) {
-				accountFlag = includeAccounts === 'true';
+				accountFlag = (includeAccounts as string).toLowerCase() === 'true';
 			}
 
 			let topMakers: string[];

src/publishers/priorityFeesPublisher.ts

@@ -21,7 +21,7 @@ require('dotenv').config();
 const stateCommitment: Commitment = 'confirmed';
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
-
+const redisClientPrefix = RedisClientPrefix.DLOB_HELIUS;
 // Set up express for health checks
 const app = express();
 
@@ -135,7 +135,7 @@ class PriorityFeeSubscriber {
 		dataPerp.forEach((result: any) => {
 			const marketIndex = parseInt(result['id']);
 			this.redisClient.publish(
-				`priorityFees_perp_${marketIndex}`,
+				`${redisClientPrefix}priorityFees_perp_${marketIndex}`,
 				result.result['priorityFeeLevels']
 			);
 			this.redisClient.set(
@@ -147,7 +147,7 @@ class PriorityFeeSubscriber {
 		dataSpot.forEach((result: any) => {
 			const marketIndex = parseInt(result['id']) - 100;
 			this.redisClient.publish(
-				`priorityFees_spot_${marketIndex}`,
+				`${redisClientPrefix}priorityFees_spot_${marketIndex}`,
 				result.result['priorityFeeLevels']
 			);
 			this.redisClient.set(
@@ -165,7 +165,7 @@ const main = async () => {
 	});
 
 	const redisClient = new RedisClient({
-		prefix: RedisClientPrefix.DLOB_HELIUS,
+		prefix: redisClientPrefix,
 	});
 	await redisClient.connect();
 

src/publishers/tradesPublisher.ts

@@ -29,7 +29,7 @@ import { fromEvent, filter, map } from 'rxjs';
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
-
+const redisClientPrefix = RedisClientPrefix.DLOB;
 //@ts-ignore
 const sdkConfig = initialize({ env: process.env.ENV });
 
@@ -67,7 +67,7 @@ const main = async () => {
 		env: driftEnv,
 	});
 
-	const redisClient = new RedisClient({ prefix: RedisClientPrefix.DLOB });
+	const redisClient = new RedisClient({ prefix: redisClientPrefix });
 	await redisClient.connect();
 
 	const slotSubscriber = new SlotSubscriber(connection, {});
@@ -180,7 +180,7 @@ const main = async () => {
 		)
 		.subscribe((fillEvent) => {
 			redisClient.publish(
-				`trades_${fillEvent.marketType}_${fillEvent.marketIndex}`,
+				`${redisClientPrefix}trades_${fillEvent.marketType}_${fillEvent.marketIndex}`,
 				fillEvent
 			);
 		});

src/serverLite.ts

@@ -30,10 +30,7 @@ import { RedisClient, RedisClientPrefix } from '@drift/common';
 require('dotenv').config();
 
 // Reading in Redis env vars
-const REDIS_CLIENTS = process.env.REDIS_CLIENTS?.replace(/^\[|\]$/g, '')
+const REDIS_CLIENTS = [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
-	.split(',')
-	.map((clients) => clients.trim()) || ['DLOB'];
-
 console.log('Redis Clients:', REDIS_CLIENTS);
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;

src/utils/utils.ts

@@ -347,11 +347,13 @@ export const getAccountFromId = async (
 ) => {
 	return Promise.all(
 		topMakers.map(async (accountId) => {
-			const userMap = await userMapClient.getRaw(accountId);
+			const userAccountEncoded = await userMapClient.getRaw(accountId);
-			if (userMap) {
+			if (userAccountEncoded) {
 				return {
 					accountId,
-					account: decodeUser(Buffer.from(userMap.split('::')[1], 'base64')),
+					account: decodeUser(
+						Buffer.from(userAccountEncoded.split('::')[1], 'base64')
+					),
 				};
 			}
 			return {

src/wsConnectionManager.ts

@@ -6,7 +6,7 @@ import { WebSocket, WebSocketServer } from 'ws';
 import { sleep } from './utils/utils';
 import { register, Gauge } from 'prom-client';
 import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk';
-import { RedisClient } from '@drift/common';
+import { RedisClient, RedisClientPrefix } from '@drift/common';
 
 // Set up env constants
 require('dotenv').config();
@@ -34,6 +34,7 @@ const WS_PORT = process.env.WS_PORT || '3000';
 console.log(`WS LISTENER PORT : ${WS_PORT}`);
 
 const MAX_BUFFERED_AMOUNT = 300000;
+const CHANNEL_PREFIX = RedisClientPrefix.DLOB;
 
 const safeGetRawChannelFromMessage = (message: any): string => {
 	return message?.channel;
@@ -64,11 +65,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}priorityFees_${marketType}_${marketIndex}`;
 		case undefined:
 		default:
 			throw new Error('Bad channel specified');
@@ -77,7 +78,7 @@ const getRedisChannelFromMessage = (message: any): string => {
 
 async function main() {
 	const redisClient = new RedisClient({});
-	const lastMessageRetriever = new RedisClient({});
+	const lastMessageRetriever = new RedisClient({ prefix: CHANNEL_PREFIX });
 
 	await redisClient.connect();
 	await lastMessageRetriever.connect();
@@ -183,11 +184,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.get(
+							CHANNEL_PREFIX,
+							''
+						);
+						const lastMessage = await lastMessageRetriever.getRaw(
 							lastUpdateChannel
 						);
-
 						if (lastMessage) {
 							ws.send(
 								JSON.stringify({