diff --git a/package.json b/package.json index 291117e..4dc2e0f 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "license": "Apache-2.0", "dependencies": { "@coral-xyz/anchor": "^0.29.0", - "@drift-labs/sdk": "2.76.0-beta.6", + "@drift-labs/sdk": "2.76.0-beta.7", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", @@ -56,6 +56,7 @@ "server-lite": "ts-node src/serverLite.ts", "dlob-publish": "ts-node src/publishers/dlobPublisher.ts", "trades-publish": "ts-node src/publishers/tradesPublisher.ts", + "fees-publish": "ts-node src/publishers/priorityFeesPublisher.ts", "ws-manager": "ts-node src/wsConnectionManager.ts", "ws-manager:inspect": "yarn build && node --inspect ./lib/wsConnectionManager.js", "dev:inspect": "yarn build && node --inspect ./lib/index.js", diff --git a/src/index.ts b/src/index.ts index 489fa49..a1bbb41 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,9 +9,6 @@ import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js'; import { BulkAccountLoader, DLOBNode, - DLOBOrder, - DLOBOrders, - DLOBOrdersCoder, DLOBSubscriber, DriftClient, DriftClientSubscriptionConfig, @@ -299,8 +296,6 @@ const main = async (): Promise => { const dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber); - const dlobCoder = DLOBOrdersCoder.create(); - await driftClient.subscribe(); driftClient.eventEmitter.on('error', (e) => { logger.info('clearing house error'); @@ -461,237 +456,24 @@ const main = async (): Promise => { app.get('/startup', handleStartup); app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)); - if (FEATURE_FLAGS.ENABLE_ORDERS_ENDPOINTS) { - app.get('/orders/json/raw', async (_req, res, next) => { - try { - // object with userAccount key and orders object serialized - const orders: Array = []; - const oracles: Array = []; - const slot = dlobProvider.getSlot(); + app.get('/priorityFees', async (req, res, next) => { + try { + const { marketIndex, marketType } = req.query; - for (const market of driftClient.getPerpMarketAccounts()) { - const oracle = driftClient.getOracleDataForPerpMarket( - market.marketIndex - ); - oracles.push({ - marketIndex: market.marketIndex, - ...oracle, - }); - } - - for (const { - userAccount, - publicKey, - } of dlobProvider.getUserAccounts()) { - for (const order of userAccount.orders) { - if (isVariant(order.status, 'init')) { - continue; - } - - orders.push({ - user: publicKey.toBase58(), - order: order, - }); - } - } - - // respond with orders - res.writeHead(200); - res.end( - JSON.stringify({ - slot, - oracles, - orders, - }) - ); - } catch (e) { - next(e); + const fees = await redisClients[ + parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0 + ].client.get(`priorityFees_${marketType}_${marketIndex}`); + if (fees) { + res.status(200).json(JSON.parse(fees)); + return; + } else { + res.writeHead(404); + res.end('Not found'); } - }); - - app.get('/orders/json', async (_req, res, next) => { - try { - // object with userAccount key and orders object serialized - const slot = dlobProvider.getSlot(); - 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, - maxPrice: oracle.maxPrice, - }; - if (oracle.twap) { - oracleHuman['twap'] = oracle.twap.toString(); - } - if (oracle.twapConfidence) { - oracleHuman['twapConfidence'] = oracle.twapConfidence.toString(); - } - oracles.push(oracleHuman); - } - for (const { - userAccount, - publicKey, - } of dlobProvider.getUserAccounts()) { - 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.quoteAssetAmountFilled) { - orderHuman['quoteAssetAmount'] = - order.quoteAssetAmountFilled.toString(); - orderHuman['quoteAssetAmountFilled'] = - order.quoteAssetAmountFilled.toString(); - } - - orders.push({ - user: publicKey.toBase58(), - order: orderHuman, - }); - } - } - - // respond with orders - res.writeHead(200); - res.end( - JSON.stringify({ - slot, - oracles, - orders, - }) - ); - } catch (err) { - next(err); - } - }); - - app.get('/orders/idl', async (_req, res, next) => { - try { - const dlobOrders: DLOBOrders = []; - - for (const { - userAccount, - publicKey, - } of dlobProvider.getUserAccounts()) { - for (const order of userAccount.orders) { - if (isVariant(order.status, 'init')) { - continue; - } - - dlobOrders.push({ - user: publicKey, - order, - } as DLOBOrder); - } - } - - res.writeHead(200); - res.end(dlobCoder.encode(dlobOrders)); - } catch (err) { - next(err); - } - }); - - app.get('/orders/idlWithSlot', async (req, res, next) => { - try { - const { marketName, marketIndex, marketType } = req.query; - const { normedMarketType, normedMarketIndex, error } = - validateDlobQuery( - driftClient, - driftEnv, - 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; - } - } - - const dlobOrders: DLOBOrders = []; - - for (const { - userAccount, - publicKey, - } of dlobProvider.getUserAccounts()) { - for (const order of userAccount.orders) { - if (isVariant(order.status, 'init')) { - continue; - } - - if (useFilter) { - if ( - getVariant(order.marketType) !== getVariant(normedMarketType) || - order.marketIndex !== normedMarketIndex - ) { - continue; - } - } - - dlobOrders.push({ - user: publicKey, - order, - } as DLOBOrder); - } - } - - res.end( - JSON.stringify({ - slot: dlobProvider.getSlot(), - data: dlobCoder.encode(dlobOrders).toString('base64'), - }) - ); - } catch (err) { - next(err); - } - }); - } + } catch (err) { + next(err); + } + }); app.get('/topMakers', async (req, res, next) => { try { diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts new file mode 100644 index 0000000..2ebd136 --- /dev/null +++ b/src/publishers/priorityFeesPublisher.ts @@ -0,0 +1,247 @@ +import { program } from 'commander'; + +import { Connection, Commitment, Keypair } from '@solana/web3.js'; + +import { + DriftClient, + initialize, + DriftEnv, + Wallet, + BulkAccountLoader, + getMarketsAndOraclesForSubscription, +} from '@drift-labs/sdk'; + +import { logger, setLogLevel } from '../utils/logger'; +import { sleep } from '../utils/utils'; +import express from 'express'; +// import { handleHealthCheck } from '../core/metrics'; +import { RedisClient } from '../utils/redisClient'; + +require('dotenv').config(); +const stateCommitment: Commitment = 'confirmed'; +const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; +const commitHash = process.env.COMMIT; +const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; +const REDIS_PORT = process.env.REDIS_PORT || '6379'; +const REDIS_PASSWORD = process.env.REDIS_PASSWORD; + +// Set up express for health checks +const app = express(); + +//@ts-ignore +const sdkConfig = initialize({ env: process.env.ENV }); +let driftClient: DriftClient; + +const opts = program.opts(); +setLogLevel(opts.debug ? 'debug' : 'info'); + +const token = process.env.TOKEN; +const endpoint = token + ? process.env.ENDPOINT + `/${token}` + : process.env.ENDPOINT; +const wsEndpoint = process.env.WS_ENDPOINT; +const FEE_POLLING_FREQUENCY = + parseInt(process.env.FEE_POLLING_FREQUENCY) || 5000; + +if (!endpoint.includes('helius')) { + throw new Error('We use helius for fee publisher fellas'); +} + +logger.info(`RPC endpoint: ${endpoint}`); +logger.info(`WS endpoint: ${wsEndpoint}`); +logger.info(`DriftEnv: ${driftEnv}`); +logger.info(`Commit: ${commitHash}`); + +class PriorityFeeSubscriber { + endpoint: string; + perpMarketPubkeys: { marketIndex: number; pubkey: string }[]; + spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[]; + redisClient: RedisClient; + frequencyMs: number; + + constructor(config: { + endpoint: string; + redisClient: RedisClient; + perpMarketPubkeys: { marketIndex: number; pubkey: string }[]; + spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[]; + frequencyMs?: number; + }) { + this.endpoint = config.endpoint; + this.perpMarketPubkeys = config.perpMarketPubkeys; + this.spotMarketPubkeys = config.spotMarketPubkeys; + this.redisClient = config.redisClient; + this.frequencyMs = config.frequencyMs ?? FEE_POLLING_FREQUENCY; + } + + async subscribe() { + await this.fetchAndPushPriorityFees(); + setInterval(async () => { + await this.fetchAndPushPriorityFees(); + }, this.frequencyMs); + } + + async fetchAndPushPriorityFees() { + const [resultPerp, resultSpot] = await Promise.all([ + fetch(this.endpoint, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify( + this.perpMarketPubkeys.map((xx) => { + return { + jsonrpc: '2.0', + id: xx.marketIndex.toString(), + method: 'getPriorityFeeEstimate', + params: [ + { + accountKeys: [xx.pubkey], + options: { + includeAllPriorityFeeLevels: true, + }, + }, + ], + }; + }) + ), + }), + fetch(this.endpoint, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify( + this.spotMarketPubkeys.map((xx) => { + return { + jsonrpc: '2.0', + id: (100 + xx.marketIndex).toString(), + method: 'getPriorityFeeEstimate', + params: [ + { + accountKeys: xx.pubkeys, + options: { + includeAllPriorityFeeLevels: true, + }, + }, + ], + }; + }) + ), + }), + ]); + + const [dataPerp, dataSpot] = await Promise.all([ + resultPerp.json(), + resultSpot.json(), + ]); + + dataPerp.forEach((result: any) => { + const marketIndex = parseInt(result['id']); + this.redisClient.client.publish( + `priorityFees_perp_${marketIndex}`, + JSON.stringify(result.result['priorityFeeLevels']) + ); + this.redisClient.client.set( + `priorityFees_perp_${marketIndex}`, + JSON.stringify(result.result['priorityFeeLevels']) + ); + }); + + dataSpot.forEach((result: any) => { + const marketIndex = parseInt(result['id']) - 100; + this.redisClient.client.publish( + `priorityFees_spot_${marketIndex}`, + JSON.stringify(result.result['priorityFeeLevels']) + ); + this.redisClient.client.set( + `priorityFees_spot_${marketIndex}`, + JSON.stringify(result.result['priorityFeeLevels']) + ); + }); + } +} + +const main = async () => { + const connection = new Connection(endpoint, { + wsEndpoint: wsEndpoint, + commitment: stateCommitment, + }); + + const { perpMarketIndexes, spotMarketIndexes, oracleInfos } = + getMarketsAndOraclesForSubscription(sdkConfig.ENV); + + const driftClient = new DriftClient({ + connection, + wallet: new Wallet(new Keypair()), + perpMarketIndexes, + spotMarketIndexes, + oracleInfos, + accountSubscription: { + type: 'polling', + accountLoader: new BulkAccountLoader(connection, stateCommitment, 0), + }, + }); + await driftClient.subscribe(); + + const perpMarketPubkeys = driftClient.getPerpMarketAccounts().map((acct) => { + return { marketIndex: acct.marketIndex, pubkey: acct.pubkey.toString() }; + }); + + const usdcMarket = driftClient.getSpotMarketAccount(0).pubkey.toString(); + const spotMarketPubkeys: { marketIndex: number; pubkeys: string[] }[] = []; + for (const market of sdkConfig.SPOT_MARKETS) { + const pubkeysForMarket = [usdcMarket]; + + const driftMarket = driftClient.getSpotMarketAccount(market.marketIndex); + pubkeysForMarket.push(driftMarket.pubkey.toString()); + + if (market.serumMarket) { + pubkeysForMarket.push(market.serumMarket.toString()); + } + + if (market.phoenixMarket) { + pubkeysForMarket.push(market.phoenixMarket.toString()); + } + + spotMarketPubkeys.push({ + marketIndex: market.marketIndex, + pubkeys: pubkeysForMarket, + }); + } + + const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); + await redisClient.connect(); + + const priorityFeeSubscriber = new PriorityFeeSubscriber({ + endpoint, + perpMarketPubkeys, + spotMarketPubkeys, + redisClient, + }); + + await priorityFeeSubscriber.subscribe(); + const server = app.listen(8080); + + // Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients + // sometimes get 502s. + // https://shuheikagawa.com/blog/2019/04/25/keep-alive-timeout/ + // https://stackoverflow.com/a/68922692 + server.keepAliveTimeout = 61 * 1000; + server.headersTimeout = 65 * 1000; + + console.log('Priority fee publisher Publishing Messages'); +}; + +async function recursiveTryCatch(f: () => void) { + try { + await f(); + } catch (e) { + console.error(e); + await sleep(15000); + await recursiveTryCatch(f); + } +} + +recursiveTryCatch(() => main()); + +export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient }; diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 2f56720..3f37f7c 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -65,11 +65,13 @@ const getRedisChannelFromMessage = (message: any): string => { throw new Error('Bad market specified'); } - switch (channel) { + switch (channel.toLowerCase()) { case 'trades': return `trades_${marketType}_${marketIndex}`; case 'orderbook': return `orderbook_${marketType}_${marketIndex}`; + case 'priorityfees': + return `priorityFees_${marketType}_${marketIndex}`; case undefined: default: throw new Error('Bad channel specified'); @@ -145,17 +147,14 @@ async function main() { JSON.stringify({ channel: requestChannel, error: - 'Error subscribing to channel with data: ' + - JSON.stringify(parsedMessage), + 'Error subscribing to channel' }) ); } else { ws.close( 1003, JSON.stringify({ - error: - 'Error subscribing to channel with data: ' + - JSON.stringify(parsedMessage), + error: 'Error subscribing to channel' }) ); } @@ -221,8 +220,7 @@ async function main() { JSON.stringify({ channel: requestChannel, error: - 'Error unsubscribing from channel with data: ' + - JSON.stringify(parsedMessage), + 'Error unsubscribing from channel' }) ); } else { @@ -230,8 +228,7 @@ async function main() { 1003, JSON.stringify({ error: - 'Error unsubscribing from channel with data: ' + - JSON.stringify(parsedMessage), + 'Error unsubscribing from channel' }) ); } diff --git a/yarn.lock b/yarn.lock index 8af98f0..e12f1f2 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.76.0-beta.6": - version "2.76.0-beta.6" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.6.tgz#c1f882f52cef4ef33f1ab9c5e346067cd3a325d0" - integrity sha512-0H5MD88lrCL2r4l6dBe8a9kaFrDeiHZfuxGq2Nf0O99V5UU57SJ236uM1OIhumQua4kptL6Mv+snKIZji6NgXQ== +"@drift-labs/sdk@2.76.0-beta.7": + version "2.76.0-beta.7" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.7.tgz#aa910d2db9446ed825636da1c9984441eb5daa3c" + integrity sha512-nO8eVXQ3hWXJsKm++MRcyUZEXbwUjLLTVpEHvSCOE5Kaqc+z21y+Aqq+44PRMEljqNAE1ouf0Qe6hTsoaqhkZw== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2"