From 255ef1f7eb4be18c87640c5fe15816a2c0b580a8 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Tue, 24 Oct 2023 17:12:16 -0700 Subject: [PATCH] rename folder and actually sanity check --- src/core/metrics.ts | 142 ++++++ src/core/middleware.ts | 26 ++ src/dlob-subscriber/DLOBSubscriberIO.ts | 9 +- src/index.ts | 547 +++--------------------- src/utils.ts | 23 - src/{ => utils}/logger.ts | 0 src/utils/utils.ts | 282 ++++++++++++ 7 files changed, 513 insertions(+), 516 deletions(-) create mode 100644 src/core/metrics.ts create mode 100644 src/core/middleware.ts delete mode 100644 src/utils.ts rename src/{ => utils}/logger.ts (100%) create mode 100644 src/utils/utils.ts diff --git a/src/core/metrics.ts b/src/core/metrics.ts new file mode 100644 index 0000000..4eaabca --- /dev/null +++ b/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, +}; diff --git a/src/core/middleware.ts b/src/core/middleware.ts new file mode 100644 index 0000000..7705d3c --- /dev/null +++ b/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, + }); + } +); diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index c87cbfa..6f1b586 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/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'); } } diff --git a/src/index.ts b/src/index.ts index ef2d26f..f4343a2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,5 @@ import { program } from 'commander'; - -import responseTime = require('response-time'); -import express, { Request, Response } from 'express'; +import express 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( - ...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 ORDERBOOK_UPDATE_INTERVAL = 100; 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) { - logger.error(`errorHandler, message: ${err.message}, stack: ${err.stack}`); - if (!res.headersSent) { - res.status(500).send('Internal error'); - } -} +let driftClient: DriftClient; 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 ( - driftClient: DriftClient, - bulkAccountLoader: BulkAccountLoader -) => { +const initializeAllMarketSubscribers = async (driftClient: DriftClient) => { const markets: SubscriberLookup = {}; for (const market of sdkConfig.SPOT_MARKETS) { @@ -317,21 +132,13 @@ const initializeAllMarketSubscribers = async ( }; if (market.phoenixMarket) { - const phoenixSubscriber = getPhoenixSubscriber( - driftClient, - market, - bulkAccountLoader - ); + const phoenixSubscriber = getPhoenixSubscriber(driftClient, market); await phoenixSubscriber.subscribe(); markets[market.marketIndex].phoenix = phoenixSubscriber; } if (market.serumMarket) { - const serumSubscriber = getSerumSubscriber( - driftClient, - market, - bulkAccountLoader - ); + const serumSubscriber = getSerumSubscriber(driftClient, market); await serumSubscriber.subscribe(); markets[market.marketIndex].serum = serumSubscriber; } @@ -349,21 +156,13 @@ const main = async () => { commitment: stateCommitment, }); - const bulkAccountLoader = new BulkAccountLoader( - connection, - stateCommitment, - bulkAccountLoaderPollingInterval - ); - const lastBulkAccountLoaderSlotMutex = new Mutex(); - let lastBulkAccountLoaderSlot = bulkAccountLoader.mostRecentSlot; - let lastBulkAccountLoaderSlotUpdated = Date.now(); - const driftClient = new DriftClient({ + driftClient = new DriftClient({ connection, wallet, programID: clearingHousePublicKey, accountSubscription: { - type: 'polling', - accountLoader: bulkAccountLoader, + type: 'websocket', + 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( - 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); - } - }; + MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); 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 = []; const oracles: Array = []; - 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 = []; const oracles: Array = []; 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) => 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) => - 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 }; diff --git a/src/utils.ts b/src/utils.ts deleted file mode 100644 index 73d8fa1..0000000 --- a/src/utils.ts +++ /dev/null @@ -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 { - return new Promise((resolve) => setTimeout(resolve, ms)); -} diff --git a/src/logger.ts b/src/utils/logger.ts similarity index 100% rename from src/logger.ts rename to src/utils/logger.ts diff --git a/src/utils/utils.ts b/src/utils/utils.ts new file mode 100644 index 0000000..37e7936 --- /dev/null +++ b/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 { + 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) => 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) => 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; + }; +};