rename folder and actually sanity check

src/core/metrics.ts

@@ -0,0 +1,142 @@
+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, 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 startupTime = Date.now();
+const handleHealthCheck = async (req, res, next) => {
+	try {
+		if (req.url === '/health' || req.url === '/') {
+			if (Date.now() > startupTime + 60 * 1000) {
+				healthStatus = HEALTH_STATUS.LivenessTesting;
+
+				res.writeHead(500);
+				res.end('Testing liveness test fail');
+				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

@@ -6,7 +6,7 @@ import {
 	MarketType,
 	groupL2,
 } from '@drift-labs/sdk';
-import { l2WithBNToStrings } from '../utils';
+import { getOracleForMarket, l2WithBNToStrings } from '../utils/utils';
 import { Server } from 'socket.io';
 import { getIOServer } from '..';
 
@@ -37,7 +37,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	}
 
 	getL2AndSendMsg(l2Args: wsMarketL2Args): void {
-		console.time('getL2AndSendMsg');
 		const grouping = l2Args.grouping;
 		const l2 = this.getL2(l2Args);
 		let l2Formatted: any;
@@ -50,7 +49,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		l2Formatted['marketName'] = l2Args.marketName;
 		l2Formatted['marketType'] = l2Args.marketType;
 		l2Formatted['marketIndex'] = l2Args.marketIndex;
+		l2Formatted['oracle'] = getOracleForMarket(
+			this.driftClient,
+			l2Args.marketType,
+			l2Args.marketIndex
+		);
 		this.io.emit('l2Update', l2Formatted);
-		console.timeEnd('getL2AndSendMsg');
 	}
 }

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,12 +17,7 @@ import {
 	DLOBOrder,
 	DLOBOrders,
 	DLOBOrdersCoder,
-	SpotMarkets,
-	PerpMarkets,
 	MarketType,
-	SpotMarketConfig,
-	PhoenixSubscriber,
-	SerumSubscriber,
 	DLOBNode,
 	isVariant,
 	BN,
@@ -34,22 +26,25 @@ import {
 	UserStatsMap,
 } from '@drift-labs/sdk';
 
-import { Mutex } from 'async-mutex';
+import { logger, setLogLevel } from './utils/logger';
 
-import { logger, setLogLevel } from './logger';
-
-import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';
-import {
-	ExplicitBucketHistogramAggregation,
-	InstrumentType,
-	MeterProvider,
-	View,
-} from '@opentelemetry/sdk-metrics-base';
-import { ObservableResult } from '@opentelemetry/api';
 import * as http from 'http';
 import { Server } from 'socket.io';
-import { l2WithBNToStrings, sleep } from './utils';
+import {
+	l2WithBNToStrings,
+	sleep,
+	getOracleForMarket,
+	normalizeBatchQueryParams,
+	SubscriberLookup,
+	errorHandler,
+	getPhoenixSubscriber,
+	getSerumSubscriber,
+	validateWsSubscribeMsg,
+	validateDlobQuery,
+} from './utils/utils';
 import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO';
+import { handleResponseTime } from './core/middleware';
+import { handleHealthCheck } from './core/metrics';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -59,23 +54,11 @@ const sdkConfig = initialize({ env: process.env.ENV });
 
 const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
-const maxPerpMarketIndex = Math.max(
+const ORDERBOOK_UPDATE_INTERVAL = 100;
-	...sdkConfig.PERP_MARKETS.map((m) => m.marketIndex)
-);
-const maxSpotMarketIndex = Math.max(
-	...sdkConfig.SPOT_MARKETS.map((m) => m.marketIndex)
-);
-
-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 =
@@ -84,47 +67,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 server = http.createServer(app);
-const io = new Server(server);
-
-export function getIOServer(): Server {
-	return io;
-}
-
-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
@@ -152,6 +102,14 @@ app.use((req, _res, next) => {
 	next();
 });
 
+app.use(errorHandler);
+const server = http.createServer(app);
+const io = new Server(server);
+
+export function getIOServer(): Server {
+	return io;
+}
+
 const opts = program.opts();
 setLogLevel(opts.debug ? 'debug' : 'info');
 
@@ -162,152 +120,9 @@ logger.info(`WS endpoint:  ${wsEndpoint}`);
 logger.info(`DriftEnv:     ${driftEnv}`);
 logger.info(`Commit:       ${commitHash}`);
 
-/**
- * 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 initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
-	driftClient: DriftClient,
-	bulkAccountLoader: BulkAccountLoader
-) => {
 	const markets: SubscriberLookup = {};
 
 	for (const market of sdkConfig.SPOT_MARKETS) {
@@ -317,21 +132,13 @@ const initializeAllMarketSubscribers = async (
 		};
 
 		if (market.phoenixMarket) {
-			const phoenixSubscriber = getPhoenixSubscriber(
+			const phoenixSubscriber = getPhoenixSubscriber(driftClient, market);
-				driftClient,
-				market,
-				bulkAccountLoader
-			);
 			await phoenixSubscriber.subscribe();
 			markets[market.marketIndex].phoenix = phoenixSubscriber;
 		}
 
 		if (market.serumMarket) {
-			const serumSubscriber = getSerumSubscriber(
+			const serumSubscriber = getSerumSubscriber(driftClient, market);
-				driftClient,
-				market,
-				bulkAccountLoader
-			);
 			await serumSubscriber.subscribe();
 			markets[market.marketIndex].serum = serumSubscriber;
 		}
@@ -349,21 +156,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,
@@ -371,11 +170,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(
@@ -391,11 +185,6 @@ const main = async () => {
 	});
 
 	await slotSubscriber.subscribe();
-	slotSubscriber.eventEmitter.on('newSlot', async (slot: number) => {
-		await lastSlotReceivedMutex.runExclusive(async () => {
-			lastSlotReceived = slot;
-		});
-	});
 
 	const userMap = new UserMap(
 		driftClient,
@@ -413,92 +202,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 (
@@ -514,7 +222,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);
@@ -524,7 +231,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(
@@ -568,7 +275,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()) {
@@ -729,7 +436,7 @@ const main = async () => {
 
 			res.end(
 				JSON.stringify({
-					slot: bulkAccountLoader.mostRecentSlot,
+					slot: slotSubscriber.currentSlot,
 					data: dlobCoder.encode(dlobOrders).toString('base64'),
 				})
 			);
@@ -738,79 +445,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 {
@@ -914,21 +548,6 @@ const main = async () => {
 		}
 	});
 
-	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 {
@@ -993,6 +612,7 @@ const main = async () => {
 				);
 				if (`${includeOracle}`.toLowerCase() === 'true') {
 					l2Formatted['oracle'] = getOracleForMarket(
+						driftClient,
 						normedMarketType,
 						normedMarketIndex
 					);
@@ -1005,6 +625,7 @@ const main = async () => {
 				const l2Formatted = l2WithBNToStrings(l2);
 				if (`${includeOracle}`.toLowerCase() === 'true') {
 					l2Formatted['oracle'] = getOracleForMarket(
+						driftClient,
 						normedMarketType,
 						normedMarketIndex
 					);
@@ -1018,71 +639,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 {
@@ -1175,6 +731,7 @@ const main = async () => {
 					);
 					if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
 						l2Formatted['oracle'] = getOracleForMarket(
+							driftClient,
 							normedMarketType,
 							normedMarketIndex
 						);
@@ -1185,6 +742,7 @@ const main = async () => {
 					const l2Formatted = l2WithBNToStrings(l2);
 					if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
 						l2Formatted['oracle'] = getOracleForMarket(
+							driftClient,
 							normedMarketType,
 							normedMarketIndex
 						);
@@ -1231,7 +789,11 @@ const main = async () => {
 			}
 
 			if (`${includeOracle}`.toLowerCase() === 'true') {
-				l3['oracle'] = getOracleForMarket(normedMarketType, normedMarketIndex);
+				l3['oracle'] = getOracleForMarket(
+					driftClient,
+					normedMarketType,
+					normedMarketIndex
+				);
 			}
 
 			res.writeHead(200);
@@ -1241,15 +803,10 @@ const main = async () => {
 		}
 	});
 
-	app.use(errorHandler);
 	server.listen(serverPort, () => {
 		logger.info(`DLOB server listening on port http://localhost:${serverPort}`);
 	});
 
-	/**
-	 * Handle websocket logic here
-	 */
-
 	io.on('connection', (socket) => {
 		console.log('a user connected');
 
@@ -1260,7 +817,15 @@ const main = async () => {
 		socket.on('message', (msg: any) => {
 			const parsed = JSON.parse(msg);
 			if (parsed['type'] == 'subscribe') {
-				console.log('here');
+				const valid = validateWsSubscribeMsg(parsed);
+				if (!valid.valid) {
+					socket.emit('message', {
+						type: 'error',
+						message: `Bad Request: Invalid subscribe message: ${valid.msg}`,
+					});
+					return;
+				}
+
 				dlobSubscriber.marketL2Args.push({
 					marketIndex: parseInt(parsed['marketIndex']),
 					marketType: isVariant(parsed['marketType'].toLowerCase(), 'spot')
@@ -1297,3 +862,5 @@ async function recursiveTryCatch(f: () => void) {
 }
 
 recursiveTryCatch(() => main());
+
+export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };

src/utils.ts

@@ -1,23 +0,0 @@
-import { L2OrderBook } from '@drift-labs/sdk';
-
-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));
-}

src/utils/utils.ts

@@ -0,0 +1,282 @@
+import {
+	DriftClient,
+	L2OrderBook,
+	MarketType,
+	PerpMarkets,
+	PhoenixSubscriber,
+	PublicKey,
+	SerumSubscriber,
+	SpotMarketConfig,
+	SpotMarkets,
+	isVariant,
+} from '@drift-labs/sdk';
+import { logger } from './logger';
+import { NextFunction, Request, Response } from 'express';
+import { driftClient, driftEnv, sdkConfig } from '..';
+
+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
+): { 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 = (
+	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
+): 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
+): 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;
+	};
+};