diff --git a/.env.example b/.env.example index 57469aa..f09e1ca 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,3 @@ -ANCHOR_PRIVATE_KEY=246,79,83,235,227,63,148,45,236,118,164,3,0,99,197,152,7,161,4,247,132,15,56,14,71,41,175,39,108,68,32,37,233,229,35,89,133,166,36,228,162,196,142,255,237,118,168,210,61,163,132,32,11,89,22,89,116,119,126,116,203,65,29,77 ENDPOINT=https://api.devnet.solana.com ENV=devnet PORT=6969 \ No newline at end of file diff --git a/.prettierrc.js b/.prettierrc.js new file mode 100644 index 0000000..a205cfe --- /dev/null +++ b/.prettierrc.js @@ -0,0 +1,8 @@ +module.exports = { + semi: true, + trailingComma: 'es5', + singleQuote: true, + printWidth: 80, + tabWidth: 2, + useTabs: true, +}; diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000..3cce174 --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,16 @@ +{ + // Use IntelliSense to learn about possible attributes. + // Hover to view descriptions of existing attributes. + // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 + "version": "0.2.0", + "configurations": [ + { + "type": "node", + "request": "attach", + "name": "Debug", + "port": 2230, + "timeout": 3000, + "restart": true + } + ] +} diff --git a/package.json b/package.json index 46282fa..ba77875 100644 --- a/package.json +++ b/package.json @@ -1,52 +1,53 @@ { - "name": "@drift-labs/dlob-server", - "version": "0.1.0", - "author": "wphan", - "main": "lib/index.js", - "license": "Apache-2.0", - "dependencies": { - "@drift-labs/sdk": "2.31.0-beta.1", - "@opentelemetry/api": "^1.1.0", - "@opentelemetry/auto-instrumentations-node": "^0.31.1", - "@opentelemetry/exporter-prometheus": "^0.31.0", - "@opentelemetry/sdk-node": "^0.31.0", - "@project-serum/anchor": "^0.19.1-beta.1", - "@project-serum/serum": "^0.13.65", - "@solana/web3.js": "^1.22.0", - "async-mutex": "^0.4.0", - "commander": "^9.4.0", - "compression": "^1.7.4", - "cors": "^2.8.5", - "dotenv": "^10.0.0", - "express": "^4.18.2", - "express-rate-limit": "^6.7.0", - "morgan": "^1.10.0", - "response-time": "^2.3.2", - "typescript": "4.5.4", - "winston": "^3.8.1" - }, - "devDependencies": { - "@typescript-eslint/eslint-plugin": "^4.28.0", - "@typescript-eslint/parser": "^4.28.0", - "eslint": "^7.29.0", - "eslint-config-prettier": "^8.3.0", - "eslint-plugin-prettier": "^3.4.0", - "husky": "^7.0.4", - "prettier": "^2.4.1", - "ts-node": "^10.9.1" - }, - "scripts": { - "prepare": "husky install", - "build": "yarn clean && tsc", - "clean": "rm -rf lib", - "start": "node lib/index.js", - "dev": "ts-node src/index.ts", - "dev:inspect": "yarn build && node --inspect ./lib/index.js", - "example": "ts-node example/client.ts", - "exampleWithSlot": "ts-node example/clientWithSlot.ts", - "prettify": "prettier --check './src/**/*.ts'", - "prettify:fix": "prettier --write './src/**/*.ts'", - "lint": "eslint . --ext ts --quiet", - "lint:fix": "eslint . --ext ts --fix" - } + "name": "@drift-labs/dlob-server", + "version": "0.1.0", + "author": "wphan", + "main": "lib/index.js", + "license": "Apache-2.0", + "dependencies": { + "@drift-labs/sdk": "2.31.0-beta.7", + "@opentelemetry/api": "^1.1.0", + "@opentelemetry/auto-instrumentations-node": "^0.31.1", + "@opentelemetry/exporter-prometheus": "^0.31.0", + "@opentelemetry/sdk-node": "^0.31.0", + "@project-serum/anchor": "^0.19.1-beta.1", + "@project-serum/serum": "^0.13.65", + "@solana/web3.js": "^1.22.0", + "async-mutex": "^0.4.0", + "commander": "^9.4.0", + "compression": "^1.7.4", + "cors": "^2.8.5", + "dotenv": "^10.0.0", + "express": "^4.18.2", + "express-rate-limit": "^6.7.0", + "morgan": "^1.10.0", + "response-time": "^2.3.2", + "typescript": "4.5.4", + "winston": "^3.8.1" + }, + "devDependencies": { + "@typescript-eslint/eslint-plugin": "^4.28.0", + "@typescript-eslint/parser": "^4.28.0", + "eslint": "^7.29.0", + "eslint-config-prettier": "^8.3.0", + "eslint-plugin-prettier": "^3.4.0", + "husky": "^7.0.4", + "prettier": "^2.4.1", + "ts-node": "^10.9.1" + }, + "scripts": { + "prepare": "husky install", + "build": "yarn clean && tsc", + "clean": "rm -rf lib", + "start": "node lib/index.js", + "dev": "ts-node src/index.ts", + "dev:inspect": "yarn build && node --inspect ./lib/index.js", + "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js", + "example": "ts-node example/client.ts", + "exampleWithSlot": "ts-node example/clientWithSlot.ts", + "prettify": "prettier --check './src/**/*.ts'", + "prettify:fix": "prettier --write './src/**/*.ts'", + "lint": "eslint . --ext ts --quiet", + "lint:fix": "eslint . --ext ts --fix" + } } diff --git a/src/index.ts b/src/index.ts index cebcf32..a2a8adc 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,116 +1,111 @@ -import { program, Option } from "commander"; +import { program } from 'commander'; -import responseTime = require("response-time"); -import express from "express"; -import rateLimit from "express-rate-limit"; -import compression from "compression"; -import morgan from "morgan"; -import cors from "cors"; +import responseTime = require('response-time'); +import express from 'express'; +import rateLimit from 'express-rate-limit'; +import compression from 'compression'; +import morgan from 'morgan'; +import cors from 'cors'; -import { Connection, Commitment, PublicKey } from "@solana/web3.js"; +import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js'; import { - getVariant, - BulkAccountLoader, - DriftClient, - initialize, - DriftEnv, - SlotSubscriber, - UserMap, - DLOBOrder, - DLOBOrders, - DLOBOrdersCoder, - SpotMarkets, - PerpMarkets, - DLOBSubscriber, - MarketType, - isVariant, -} from "@drift-labs/sdk"; + getVariant, + BulkAccountLoader, + DriftClient, + initialize, + DriftEnv, + SlotSubscriber, + UserMap, + DLOBOrder, + DLOBOrders, + DLOBOrdersCoder, + SpotMarkets, + PerpMarkets, + DLOBSubscriber, + MarketType, + SpotMarketConfig, + PhoenixSubscriber, + SerumSubscriber, + DLOBNode, + isVariant, + BN, + groupL2, + L2OrderBook, + Wallet, +} from '@drift-labs/sdk'; -import { Mutex } from "async-mutex"; +import { Mutex } from 'async-mutex'; -import { getWallet } from "./utils"; -import { logger, setLogLevel } from "./logger"; +import { logger, setLogLevel } from './logger'; -import { PrometheusExporter } from "@opentelemetry/exporter-prometheus"; +import { PrometheusExporter } from '@opentelemetry/exporter-prometheus'; import { - ExplicitBucketHistogramAggregation, - InstrumentType, - MeterProvider, - View, -} from "@opentelemetry/sdk-metrics-base"; -import { ObservableResult } from "@opentelemetry/api"; + ExplicitBucketHistogramAggregation, + InstrumentType, + MeterProvider, + View, +} from '@opentelemetry/sdk-metrics-base'; +import { ObservableResult } from '@opentelemetry/api'; -require("dotenv").config(); -const driftEnv = (process.env.ENV || "devnet") as DriftEnv; +require('dotenv').config(); +const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; //@ts-ignore const sdkConfig = initialize({ env: process.env.ENV }); -const stateCommitment: Commitment = "confirmed"; +const stateCommitment: Commitment = 'confirmed'; const serverPort = process.env.PORT || 6969; const bulkAccountLoaderPollingInterval = process.env - .BULK_ACCOUNT_LOADER_POLLING_INTERVAL - ? parseInt(process.env.BULK_ACCOUNT_LOADER_POLLING_INTERVAL) - : 5000; + .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) - : 10; + ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) + : 10; const logFormat = - ':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]'; + ':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]'; const logHttp = morgan(logFormat, { - skip: (_req, res) => res.statusCode < 400, + skip: (_req, res) => res.statusCode < 400, }); function errorHandler(err, _req, res, _next) { - logger.error(err.stack); - res.status(500).send("Internal error"); + logger.error(err.stack); + res.status(500).send('Internal error'); } const app = express(); -app.use(cors({ origin: "*" })); +app.use(cors({ origin: '*' })); app.use(compression()); -app.set("trust proxy", 1); +app.set('trust proxy', 1); app.use(logHttp); app.use( - rateLimit({ - windowMs: 1000, // 1 second - max: rateLimitCallsPerSecond, - standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers - legacyHeaders: false, // Disable the `X-RateLimit-*` headers - }) + rateLimit({ + windowMs: 1000, // 1 second + max: rateLimitCallsPerSecond, + standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers + legacyHeaders: false, // Disable the `X-RateLimit-*` headers + }) ); // strip off /dlob, if the request comes from exchange history server LB app.use((req, _res, next) => { - if (req.url.startsWith("/dlob")) { - req.url = req.url.replace("/dlob", ""); - if (req.url === "") { - req.url = "/"; - } - } - next(); + if (req.url.startsWith('/dlob')) { + req.url = req.url.replace('/dlob', ''); + if (req.url === '') { + req.url = '/'; + } + } + next(); }); -program - .option("-d, --dry-run", "Dry run, do not send transactions on chain") - .option("--test-liveness", "Purposefully fail liveness test after 1 minute") - .addOption( - new Option( - "-p, --private-key ", - "private key, supports path to id.json, or list of comma separate numbers" - ).env("ANCHOR_PRIVATE_KEY") - ) - .option("--debug", "Enable debug logging") - .parse(); - const opts = program.opts(); -setLogLevel(opts.debug ? "debug" : "info"); +setLogLevel(opts.debug ? 'debug' : 'info'); const endpoint = process.env.ENDPOINT; const wsEndpoint = process.env.WS_ENDPOINT; @@ -120,7 +115,7 @@ logger.info(`DriftEnv: ${driftEnv}`); logger.info(`Commit: ${commitHash}`); function sleep(ms) { - return new Promise((resolve) => setTimeout(resolve, ms)); + return new Promise((resolve) => setTimeout(resolve, ms)); } /** @@ -131,656 +126,1016 @@ function sleep(ms) { * @returns */ const createHistogramBuckets = ( - start: number, - increment: number, - count: number + start: number, + increment: number, + count: number ) => { - return new ExplicitBucketHistogramAggregation( - Array.from(new Array(count), (_, i) => start + i * increment) - ); + 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", - health_status = "health_status", + runtime_specs = 'runtime_specs', + endpoint_response_times_histogram = 'endpoint_response_times_histogram', + health_status = 'health_status', } export enum HEALTH_STATUS { - Ok = 0, - StaleBulkAccountLoader, - UnhealthySlotSubscriber, - LivenessTesting, + Ok = 0, + StaleBulkAccountLoader, + UnhealthySlotSubscriber, + LivenessTesting, } const metricsPort = - parseInt(process.env.METRICS_PORT) || PrometheusExporter.DEFAULT_OPTIONS.port; + 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}` - ); - } + { + port: metricsPort, + endpoint: defaultEndpoint, + }, + () => { + logger.info( + `prometheus scrape endpoint started: http://localhost:${metricsPort}${defaultEndpoint}` + ); + } ); -const meterName = "dlob-meter"; +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, 50, 30), - }), - ], + views: [ + new View({ + instrumentName: METRIC_TYPES.endpoint_response_times_histogram, + instrumentType: InstrumentType.HISTOGRAM, + meterName, + aggregation: createHistogramBuckets(0, 50, 30), + }), + ], }); meterProvider.addMetricReader(exporter); const meter = meterProvider.getMeter(meterName); const runtimeSpecsGauge = meter.createObservableGauge( - METRIC_TYPES.runtime_specs, - { - description: "Runtime sepcification of this program", - } + 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, - }); + 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", - } + METRIC_TYPES.health_status, + { + description: 'Health status of this program', + } ); healthStatusGauge.addCallback((obs: ObservableResult) => { - obs.observe(healthStatus, {}); + obs.observe(healthStatus, {}); }); const endpointResponseTimeHistogram = meter.createHistogram( - METRIC_TYPES.endpoint_response_times_histogram, - { - description: "Duration of endpoint responses", - unit: "ms", - } + METRIC_TYPES.endpoint_response_times_histogram, + { + description: 'Duration of endpoint responses', + unit: 'ms', + } ); +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 markets: SubscriberLookup = {}; + + for (const market of sdkConfig.SPOT_MARKETS) { + markets[market.marketIndex] = { + phoenix: undefined, + serum: undefined, + }; + + if (market.phoenixMarket) { + const phoenixSubscriber = getPhoenixSubscriber( + driftClient, + market, + bulkAccountLoader + ); + await phoenixSubscriber.subscribe(); + markets[market.marketIndex].phoenix = phoenixSubscriber; + } + + if (market.serumMarket) { + const serumSubscriber = getSerumSubscriber( + driftClient, + market, + bulkAccountLoader + ); + await serumSubscriber.subscribe(); + markets[market.marketIndex].serum = serumSubscriber; + } + } + + return markets; +}; + const main = async () => { - const wallet = getWallet(); - const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID); + const wallet = new Wallet(new Keypair()); + const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID); - const connection = new Connection(endpoint, { - wsEndpoint: wsEndpoint, - commitment: stateCommitment, - }); + const connection = new Connection(endpoint, { + wsEndpoint: wsEndpoint, + 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({ - connection, - wallet, - programID: clearingHousePublicKey, - perpMarketIndexes: PerpMarkets[driftEnv].map((mkt) => mkt.marketIndex), - spotMarketIndexes: SpotMarkets[driftEnv].map((mkt) => mkt.marketIndex), - oracleInfos: PerpMarkets[driftEnv].map((mkt) => { - return { publicKey: mkt.oracle, source: mkt.oracleSource }; - }), - accountSubscription: { - type: "polling", - accountLoader: bulkAccountLoader, - }, - env: driftEnv, - userStats: true, - }); + const bulkAccountLoader = new BulkAccountLoader( + 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', + accountLoader: bulkAccountLoader, + }, + env: driftEnv, + userStats: true, + }); - 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 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( - `DriftClient ProgramId: ${driftClient.program.programId.toBase58()}` - ); - logger.info(`Wallet pubkey: ${wallet.publicKey.toBase58()}`); - logger.info(` . SOL balance: ${lamportsBalance / 10 ** 9}`); + const lamportsBalance = await connection.getBalance(wallet.publicKey); + logger.info( + `DriftClient ProgramId: ${driftClient.program.programId.toBase58()}` + ); + logger.info(`Wallet pubkey: ${wallet.publicKey.toBase58()}`); + logger.info(` . SOL balance: ${lamportsBalance / 10 ** 9}`); - await driftClient.subscribe(); - driftClient.eventEmitter.on("error", (e) => { - logger.info("clearing house error"); - logger.error(e); - }); + await driftClient.subscribe(); + driftClient.eventEmitter.on('error', (e) => { + logger.info('clearing house error'); + logger.error(e); + }); - await slotSubscriber.subscribe(); - slotSubscriber.eventEmitter.on("newSlot", async (slot: number) => { - await lastSlotReceivedMutex.runExclusive(async () => { - lastSlotReceived = slot; - }); - }); + await slotSubscriber.subscribe(); + slotSubscriber.eventEmitter.on('newSlot', async (slot: number) => { + await lastSlotReceivedMutex.runExclusive(async () => { + lastSlotReceived = slot; + }); + }); - if (!(await driftClient.getUser().exists())) { - logger.error(`User for ${wallet.publicKey} does not exist`); - if (opts.initUser) { - logger.info(`Creating User for ${wallet.publicKey}`); - const [txSig] = await driftClient.initializeUserAccount(); - logger.info(`Initialized user account in transaction: ${txSig}`); - } else { - throw new Error("Run with '--init-user' flag to initialize a User"); - } - } + const userMap = new UserMap( + driftClient, + driftClient.userAccountSubscriptionConfig, + false + ); + await userMap.subscribe(); - const userMap = new UserMap( - driftClient, - driftClient.userAccountSubscriptionConfig, - false - ); - await userMap.subscribe(); + const dlobSubscriber = new DLOBSubscriber({ + driftClient, + dlobSource: userMap, + slotSource: slotSubscriber, + updateFrequency: bulkAccountLoaderPollingInterval, + }); + await dlobSubscriber.subscribe(); - const dlobSubscriber = new DLOBSubscriber({ - driftClient, - dlobSource: userMap, - slotSource: slotSubscriber, - updateFrequency: bulkAccountLoaderPollingInterval, - }); - await dlobSubscriber.subscribe(); + const handleResponseTime = responseTime((req: Request, _res, time) => { + const endpoint = req.url; - const handleResponseTime = responseTime((req: Request, _res, time) => { - const endpoint = req.url; + const responseTimeMs = time; + endpointResponseTimeHistogram.record(responseTimeMs, { + endpoint, + }); + }); - const responseTimeMs = time; - endpointResponseTimeHistogram.record(responseTimeMs, { - endpoint, - }); - }); + MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers( + driftClient, + bulkAccountLoader + ); - // start http server listening to /health endpoint using http package - app.get("/health", handleResponseTime, async (req, res, next) => { - try { - if (req.url === "/health") { - if (opts.testLiveness) { - if (Date.now() > startupTime + 60 * 1000) { - healthStatus = HEALTH_STATUS.LivenessTesting; + // start http server listening to /health endpoint using http package + app.get('/health', handleResponseTime, async (req, res, next) => { + try { + if (req.url === '/health') { + 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('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; - } + 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` - ); + 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; - } - } + 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); - } - }); + // 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); + } + }); - app.get("/orders/json/raw", handleResponseTime, async (_req, res, next) => { - try { - // object with userAccount key and orders object serialized - const orders: Array = []; - const oracles: Array = []; - const slot = bulkAccountLoader.mostRecentSlot; + app.get('/orders/json/raw', handleResponseTime, async (_req, res, next) => { + try { + // object with userAccount key and orders object serialized + const orders: Array = []; + const oracles: Array = []; + const slot = bulkAccountLoader.mostRecentSlot; - for (const market of driftClient.getPerpMarketAccounts()) { - const oracle = driftClient.getOracleDataForPerpMarket( - market.marketIndex - ); - oracles.push({ - marketIndex: market.marketIndex, - ...oracle, - }); - } + for (const market of driftClient.getPerpMarketAccounts()) { + const oracle = driftClient.getOracleDataForPerpMarket( + market.marketIndex + ); + oracles.push({ + marketIndex: market.marketIndex, + ...oracle, + }); + } - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); + for (const user of userMap.values()) { + const userAccount = user.getUserAccount(); - for (const order of userAccount.orders) { - if (getVariant(order.status) === "init") { - continue; - } + for (const order of userAccount.orders) { + if (isVariant(order.status, 'init')) { + continue; + } - orders.push({ - user: user.getUserAccountPublicKey().toBase58(), - order: order, - }); - } - } + orders.push({ + user: user.getUserAccountPublicKey().toBase58(), + order: order, + }); + } + } - // respond with orders - res.writeHead(200); - res.end( - JSON.stringify({ - slot, - oracles, - orders, - }) - ); - } catch (e) { - next(e); - } - }); + // respond with orders + res.writeHead(200); + res.end( + JSON.stringify({ + slot, + oracles, + orders, + }) + ); + } catch (e) { + next(e); + } + }); - app.get("/orders/json", handleResponseTime, async (_req, res, next) => { - try { - // object with userAccount key and orders object serialized - const slot = bulkAccountLoader.mostRecentSlot; - const orders: Array = []; - const oracles: Array = []; - for (const market of driftClient.getPerpMarketAccounts()) { - const oracle = driftClient.getOracleDataForPerpMarket( - market.marketIndex - ); - const oracleHuman = { - marketIndex: market.marketIndex, - price: oracle.price.toString(), - slot: oracle.slot.toString(), - confidence: oracle.confidence.toString(), - hasSufficientNumberOfDataPoints: - oracle.hasSufficientNumberOfDataPoints, - }; - if (oracle.twap) { - oracleHuman["twap"] = oracle.twap.toString(); - } - if (oracle.twapConfidence) { - oracleHuman["twapConfidence"] = oracle.twapConfidence.toString(); - } - oracles.push(oracleHuman); - } - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); + app.get('/orders/json', handleResponseTime, async (_req, res, next) => { + try { + // object with userAccount key and orders object serialized + const slot = bulkAccountLoader.mostRecentSlot; + const orders: Array = []; + const oracles: Array = []; + for (const market of driftClient.getPerpMarketAccounts()) { + const oracle = driftClient.getOracleDataForPerpMarket( + market.marketIndex + ); + const oracleHuman = { + marketIndex: market.marketIndex, + price: oracle.price.toString(), + slot: oracle.slot.toString(), + confidence: oracle.confidence.toString(), + hasSufficientNumberOfDataPoints: + oracle.hasSufficientNumberOfDataPoints, + }; + if (oracle.twap) { + oracleHuman['twap'] = oracle.twap.toString(); + } + if (oracle.twapConfidence) { + oracleHuman['twapConfidence'] = oracle.twapConfidence.toString(); + } + oracles.push(oracleHuman); + } + for (const user of userMap.values()) { + const userAccount = user.getUserAccount(); - for (const order of userAccount.orders) { - if (getVariant(order.status) === "init") { - continue; - } + for (const order of userAccount.orders) { + if (isVariant(order.status, 'init')) { + continue; + } - const orderHuman = { - status: getVariant(order.status), - orderType: getVariant(order.orderType), - marketType: getVariant(order.marketType), - slot: order.slot.toString(), - orderId: order.orderId, - userOrderId: order.userOrderId, - marketIndex: order.marketIndex, - price: order.price.toString(), - baseAssetAmount: order.baseAssetAmount.toString(), - baseAssetAmountFilled: order.baseAssetAmountFilled.toString(), - quoteAssetAmountFilled: order.quoteAssetAmountFilled.toString(), - direction: getVariant(order.direction), - reduceOnly: order.reduceOnly, - triggerPrice: order.triggerPrice.toString(), - triggerCondition: getVariant(order.triggerCondition), - existingPositionDirection: getVariant( - order.existingPositionDirection - ), - postOnly: order.postOnly, - immediateOrCancel: order.immediateOrCancel, - oraclePriceOffset: order.oraclePriceOffset, - auctionDuration: order.auctionDuration, - auctionStartPrice: order.auctionStartPrice.toString(), - auctionEndPrice: order.auctionEndPrice.toString(), - maxTs: order.maxTs.toString(), - }; - if (order.quoteAssetAmount) { - orderHuman["quoteAssetAmount"] = order.quoteAssetAmount.toString(); - } + const orderHuman = { + status: getVariant(order.status), + orderType: getVariant(order.orderType), + marketType: getVariant(order.marketType), + slot: order.slot.toString(), + orderId: order.orderId, + userOrderId: order.userOrderId, + marketIndex: order.marketIndex, + price: order.price.toString(), + baseAssetAmount: order.baseAssetAmount.toString(), + baseAssetAmountFilled: order.baseAssetAmountFilled.toString(), + quoteAssetAmountFilled: order.quoteAssetAmountFilled.toString(), + direction: getVariant(order.direction), + reduceOnly: order.reduceOnly, + triggerPrice: order.triggerPrice.toString(), + triggerCondition: getVariant(order.triggerCondition), + existingPositionDirection: getVariant( + order.existingPositionDirection + ), + postOnly: order.postOnly, + immediateOrCancel: order.immediateOrCancel, + oraclePriceOffset: order.oraclePriceOffset, + auctionDuration: order.auctionDuration, + auctionStartPrice: order.auctionStartPrice.toString(), + auctionEndPrice: order.auctionEndPrice.toString(), + maxTs: order.maxTs.toString(), + }; + if (order.quoteAssetAmount) { + orderHuman['quoteAssetAmount'] = order.quoteAssetAmount.toString(); + } - orders.push({ - user: user.getUserAccountPublicKey().toBase58(), - order: orderHuman, - }); - } - } + orders.push({ + user: user.getUserAccountPublicKey().toBase58(), + order: orderHuman, + }); + } + } - // respond with orders - res.writeHead(200); - res.end( - JSON.stringify({ - slot, - oracles, - orders, - }) - ); - } catch (err) { - next(err); - } - }); + // respond with orders + res.writeHead(200); + res.end( + JSON.stringify({ + slot, + oracles, + orders, + }) + ); + } catch (err) { + next(err); + } + }); - app.get("/orders/idl", handleResponseTime, async (_req, res, next) => { - try { - const dlobOrders: DLOBOrders = []; + app.get('/orders/idl', handleResponseTime, async (_req, res, next) => { + try { + const dlobOrders: DLOBOrders = []; - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); + for (const user of userMap.values()) { + const userAccount = user.getUserAccount(); - for (const order of userAccount.orders) { - if (getVariant(order.status) === "init") { - continue; - } + for (const order of userAccount.orders) { + if (isVariant(order.status, 'init')) { + continue; + } - dlobOrders.push({ - user: user.getUserAccountPublicKey(), - order, - } as DLOBOrder); - } - } + dlobOrders.push({ + user: user.getUserAccountPublicKey(), + order, + } as DLOBOrder); + } + } - res.writeHead(200); - res.end(dlobCoder.encode(dlobOrders)); - } catch (err) { - next(err); - } - }); + res.writeHead(200); + res.end(dlobCoder.encode(dlobOrders)); + } catch (err) { + next(err); + } + }); - app.get("/orders/idlWithSlot", handleResponseTime, async (req, res, next) => { - try { - const { marketName, marketIndex, marketType } = req.query; - const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( - marketType as string, - marketIndex as string, - marketName as string - ); - const useFilter = - marketName !== undefined || - marketIndex !== undefined || - marketType !== undefined; + app.get('/orders/idlWithSlot', handleResponseTime, async (req, res, next) => { + try { + const { marketName, marketIndex, marketType } = req.query; + const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( + marketType as string, + marketIndex as string, + marketName as string + ); + const useFilter = + marketName !== undefined || + marketIndex !== undefined || + marketType !== undefined; - if (useFilter) { - if ( - error || - normedMarketType === undefined || - normedMarketIndex === undefined - ) { - res.status(400).send(error); - return; - } - } + if (useFilter) { + if ( + error || + normedMarketType === undefined || + normedMarketIndex === undefined + ) { + res.status(400).send(error); + return; + } + } - const dlobOrders: DLOBOrders = []; + const dlobOrders: DLOBOrders = []; - for (const user of userMap.values()) { - const userAccount = user.getUserAccount(); + for (const user of userMap.values()) { + const userAccount = user.getUserAccount(); - for (const order of userAccount.orders) { - if (getVariant(order.status) === "init") { - continue; - } + for (const order of userAccount.orders) { + if (isVariant(order.status, 'init')) { + continue; + } - if (useFilter) { - if ( - getVariant(order.marketType) !== getVariant(normedMarketType) || - order.marketIndex !== normedMarketIndex - ) { - continue; - } - } + if (useFilter) { + if ( + getVariant(order.marketType) !== getVariant(normedMarketType) || + order.marketIndex !== normedMarketIndex + ) { + continue; + } + } - dlobOrders.push({ - user: user.getUserAccountPublicKey(), - order, - } as DLOBOrder); - } - } + dlobOrders.push({ + user: user.getUserAccountPublicKey(), + order, + } as DLOBOrder); + } + } - res.end( - JSON.stringify({ - slot: bulkAccountLoader.mostRecentSlot, - data: dlobCoder.encode(dlobOrders).toString("base64"), - }) - ); - } catch (err) { - next(err); - } - }); + res.end( + JSON.stringify({ + slot: bulkAccountLoader.mostRecentSlot, + data: dlobCoder.encode(dlobOrders).toString('base64'), + }) + ); + } catch (err) { + next(err); + } + }); - 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", - }; - } + 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; - } + // 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, - }; - }; + return { + normedMarketType, + normedMarketIndex, + }; + }; - app.get("/l2", handleResponseTime, async (req, res, next) => { - try { - const { marketName, marketIndex, marketType, depth, includeVamm } = - req.query; + app.get('/topMakers', handleResponseTime, async (req, res, next) => { + try { + const { + marketName, + marketIndex, + marketType, + side, // bid or ask + limit, // number of unique makers to return, if undefined will return all + } = req.query; - const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( - marketType as string, - marketIndex as string, - marketName as string - ); - if (error) { - res.status(400).send(error); - return; - } + const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( + marketType as string, + marketIndex as string, + marketName as string + ); + if (error) { + res.status(400).send(error); + return; + } - const l2 = dlobSubscriber.getL2({ - marketIndex: normedMarketIndex, - marketType: normedMarketType, - depth: depth ? parseInt(depth as string) : 10, - includeVamm: includeVamm - ? (includeVamm as string).toLowerCase() === "true" - : false, - }); + if (side !== 'bid' && side !== 'ask') { + res.status(400).send('Bad Request: side must be either bid or ask'); + return; + } + const normedSide = (side as string).toLowerCase(); + const oracle = driftClient.getOracleDataForPerpMarket(normedMarketIndex); - 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, - }; - } - } + let normedLimit = undefined; + if (limit) { + if (isNaN(parseInt(limit as string))) { + res + .status(400) + .send('Bad Request: limit must be a number if supplied'); + return; + } + normedLimit = parseInt(limit as string); + } - res.writeHead(200); - res.end(JSON.stringify(l2)); - } catch (err) { - next(err); - } - }); + const topMakers: Set = new Set(); + let foundMakers = 0; + const findMakers = (sideGenerator: Generator) => { + for (const side of sideGenerator) { + if (limit && foundMakers >= normedLimit) { + break; + } + if (side.userAccount) { + const maker = side.userAccount.toBase58(); + if (topMakers.has(maker)) { + continue; + } else { + topMakers.add(side.userAccount.toBase58()); + foundMakers++; + } + } else { + continue; + } + } + }; - app.get("/l3", handleResponseTime, async (req, res, next) => { - try { - const { marketName, marketIndex, marketType } = req.query; + if (normedSide === 'bid') { + findMakers( + dlobSubscriber + .getDLOB() + .getRestingLimitBids( + normedMarketIndex, + slotSubscriber.getSlot(), + normedMarketType, + oracle + ) + ); + } else { + findMakers( + dlobSubscriber + .getDLOB() + .getRestingLimitAsks( + normedMarketIndex, + slotSubscriber.getSlot(), + normedMarketType, + oracle + ) + ); + } - const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( - marketType as string, - marketIndex as string, - marketName as string - ); - if (error) { - res.status(400).send(error); - return; - } + res.writeHead(200); + res.end(JSON.stringify([...topMakers])); + } catch (err) { + next(err); + } + }); - const l3 = dlobSubscriber.getL3({ - marketIndex: normedMarketIndex, - marketType: normedMarketType, - }); + 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; + }; - for (const key of Object.keys(l3)) { - for (const idx in l3[key]) { - const level = l3[key][idx]; - l3[key][idx] = { - ...level, - price: level.price.toString(), - size: level.size.toString(), - }; - } - } + app.get('/l2', handleResponseTime, async (req, res, next) => { + try { + const { + marketName, + marketIndex, + marketType, + depth, + includeVamm, + includePhoenix, + includeSerum, + grouping, // undefined or PRICE_PRECISION + } = req.query; - res.writeHead(200); - res.end(JSON.stringify(l3)); - } catch (err) { - next(err); - } - }); + const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( + marketType as string, + marketIndex as string, + marketName as string + ); + if (error) { + res.status(400).send(error); + return; + } - app.use(errorHandler); - app.listen(serverPort, () => { - logger.info(`DLOB server listening on port http://localhost:${serverPort}`); - }); + const isSpot = isVariant(normedMarketType, 'spot'); + + let adjustedDepth = depth ?? '10'; + if (grouping !== undefined) { + // If grouping is also supplied, we want the entire book depth. + // we will apply depth after grouping + adjustedDepth = '-1'; + } + + const l2 = dlobSubscriber.getL2({ + marketIndex: normedMarketIndex, + marketType: normedMarketType, + depth: parseInt(adjustedDepth as string), + includeVamm: `${includeVamm}`.toLowerCase() === 'true', + fallbackL2Generators: isSpot + ? [ + `${includePhoenix}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, + `${includeSerum}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].serum, + ].filter((a) => !!a) + : [], + }); + + if (grouping) { + const finalDepth = depth ? parseInt(depth as string) : 10; + if (isNaN(parseInt(grouping as string))) { + res + .status(400) + .send('Bad Request: grouping must be a number if supplied'); + return; + } + const groupingBN = new BN(parseInt(grouping as string)); + res.writeHead(200); + res.end( + JSON.stringify(l2WithBNToStrings(groupL2(l2, groupingBN, finalDepth))) + ); + } else { + // make the BNs into strings + res.writeHead(200); + res.end(JSON.stringify(l2WithBNToStrings(l2))); + } + } catch (err) { + next(err); + } + }); + + /** + * 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', handleResponseTime, async (req, res, next) => { + try { + const { + marketName, + marketIndex, + marketType, + depth, + includeVamm, + includePhoenix, + includeSerum, + grouping, // undefined or PRICE_PRECISION + } = req.query; + + const normedParams = normalizeBatchQueryParams({ + marketName: marketName as string | undefined, + marketIndex: marketIndex as string | undefined, + marketType: marketType as string | undefined, + depth: depth as string | undefined, + includeVamm: includeVamm as string | undefined, + includePhoenix: includePhoenix as string | undefined, + includeSerum: includeSerum as string | undefined, + grouping: grouping as string | undefined, + }); + + if (normedParams === undefined) { + res + .status(400) + .send( + 'Bad Request: all params for batch request must be the same length' + ); + return; + } + + const l2s = normedParams.map((normedParam) => { + const { normedMarketType, normedMarketIndex, error } = + validateDlobQuery( + normedParam['marketType'] as string, + normedParam['marketIndex'] as string, + normedParam['marketName'] as string + ); + if (error) { + res.status(400).send(`Bad Request: ${error}`); + return; + } + + const isSpot = isVariant(normedMarketType, 'spot'); + + let adjustedDepth = normedParam['depth'] ?? '10'; + if (normedParam['grouping'] !== undefined) { + // If grouping is also supplied, we want the entire book depth. + // we will apply depth after grouping + adjustedDepth = '-1'; + } + + const l2 = dlobSubscriber.getL2({ + marketIndex: normedMarketIndex, + marketType: normedMarketType, + depth: parseInt(adjustedDepth as string), + includeVamm: `${normedParam['includeVamm']}`.toLowerCase() === 'true', + fallbackL2Generators: isSpot + ? [ + `${normedParam['includePhoenix']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].phoenix, + `${normedParam['includeSerum']}`.toLowerCase() === 'true' && + MARKET_SUBSCRIBERS[normedMarketIndex].serum, + ].filter((a) => !!a) + : [], + }); + + if (normedParam['grouping']) { + const finalDepth = normedParam['depth'] + ? parseInt(normedParam['depth'] as string) + : 10; + if (isNaN(parseInt(normedParam['grouping'] as string))) { + res + .status(400) + .send('Bad Request: grouping must be a number if supplied'); + return; + } + const groupingBN = new BN( + parseInt(normedParam['grouping'] as string) + ); + + return l2WithBNToStrings(groupL2(l2, groupingBN, finalDepth)); + } else { + // make the BNs into strings + return l2WithBNToStrings(l2); + } + }); + + res.writeHead(200); + res.end(JSON.stringify({ l2s })); + } catch (err) { + next(err); + } + }); + + app.get('/l3', handleResponseTime, async (req, res, next) => { + try { + const { marketName, marketIndex, marketType } = req.query; + + const { normedMarketType, normedMarketIndex, error } = validateDlobQuery( + marketType as string, + marketIndex as string, + marketName as string + ); + if (error) { + res.status(400).send(error); + return; + } + + const l3 = dlobSubscriber.getL3({ + marketIndex: normedMarketIndex, + marketType: normedMarketType, + }); + + for (const key of Object.keys(l3)) { + for (const idx in l3[key]) { + const level = l3[key][idx]; + l3[key][idx] = { + ...level, + price: level.price.toString(), + size: level.size.toString(), + }; + } + } + + res.writeHead(200); + res.end(JSON.stringify(l3)); + } catch (err) { + next(err); + } + }); + + app.use(errorHandler); + app.listen(serverPort, () => { + logger.info(`DLOB server listening on port http://localhost:${serverPort}`); + }); }; async function recursiveTryCatch(f: () => void) { - try { - await f(); - } catch (e) { - console.error(e); - await sleep(15000); - await recursiveTryCatch(f); - } + try { + await f(); + } catch (e) { + console.error(e); + await sleep(15000); + await recursiveTryCatch(f); + } } recursiveTryCatch(() => main()); diff --git a/src/logger.ts b/src/logger.ts index aab6e20..32667d1 100644 --- a/src/logger.ts +++ b/src/logger.ts @@ -1,16 +1,16 @@ -import { createLogger, transports, format } from "winston"; +import { createLogger, transports, format } from 'winston'; export const logger = createLogger({ - transports: [new transports.Console()], - format: format.combine( - format.colorize(), - format.timestamp(), - format.printf(({ timestamp, level, message }) => { - return `[${timestamp}] ${level}: ${message}`; - }) - ), + transports: [new transports.Console()], + format: format.combine( + format.colorize(), + format.timestamp(), + format.printf(({ timestamp, level, message }) => { + return `[${timestamp}] ${level}: ${message}`; + }) + ), }); export const setLogLevel = (logLevel: string) => { - logger.level = logLevel; + logger.level = logLevel; }; diff --git a/src/utils.ts b/src/utils.ts deleted file mode 100644 index 9e16a04..0000000 --- a/src/utils.ts +++ /dev/null @@ -1,36 +0,0 @@ -import fs from "fs"; -import { Keypair } from "@solana/web3.js"; -import { bs58 } from "@project-serum/anchor/dist/cjs/utils/bytes"; -import { Wallet } from "@drift-labs/sdk"; - -import { logger } from "./logger"; - -export function getWallet(): Wallet { - const privateKey = process.env.ANCHOR_PRIVATE_KEY; - if (!privateKey) { - throw new Error( - "Must set environment variable ANCHOR_PRIVATE_KEY with the path to a id.json or a list of commma separated numbers" - ); - } - // try to load privateKey as a filepath - let loadedKey: Uint8Array; - if (fs.existsSync(privateKey)) { - logger.info(`loading private key from ${privateKey}`); - loadedKey = new Uint8Array( - JSON.parse(fs.readFileSync(privateKey).toString()) - ); - } else { - if (privateKey.includes(",")) { - logger.info(`Trying to load private key as comma separated numbers`); - loadedKey = Uint8Array.from( - privateKey.split(",").map((val) => Number(val)) - ); - } else { - logger.info(`Trying to load private key as base58 string`); - loadedKey = new Uint8Array(bs58.decode(privateKey)); - } - } - - const keypair = Keypair.fromSecretKey(Uint8Array.from(loadedKey)); - return new Wallet(keypair); -} diff --git a/yarn.lock b/yarn.lock index 895ced6..029c07b 100644 --- a/yarn.lock +++ b/yarn.lock @@ -80,10 +80,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.31.0-beta.1": - version "2.31.0-beta.1" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.31.0-beta.1.tgz#60e14806430fdcd427e9949ab511ab659c8a83fc" - integrity sha512-GQENFRU+qNIeFlC0k/JaaluTtgAQ4t6a9EHWfa4sIC6T+2KhBu2e6zZ0QiCSDYFm5cJleqE9z3EHEW3T2o5Exg== +"@drift-labs/sdk@2.31.0-beta.7": + version "2.31.0-beta.7" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.31.0-beta.7.tgz#d7337d80d7ed9bd8a660cfd5280d660987e42f37" + integrity sha512-+SpE4gv8BXvr/y/Gdjk7VS96Qg7u8SCMM8cB+WyI8KrJJsz569ORSEUu4FE2PVcu5HeypzUosiphLfDmZuXDIA== dependencies: "@coral-xyz/anchor" "0.26.0" "@ellipsis-labs/phoenix-sdk" "^1.4.2"