import { AccountSubscriber, DataAndSlot, DelistedMarketSetting, DriftClientAccountEvents, DriftClientAccountSubscriber, NotSubscribedError, ResubOpts, } from './types'; import { PerpMarketAccount, SpotMarketAccount, StateAccount } from '../types'; import { Program } from '@coral-xyz/anchor'; import StrictEventEmitter from 'strict-event-emitter-types'; import { EventEmitter } from 'events'; import { getDriftStateAccountPublicKey, getPerpMarketPublicKey, getPerpMarketPublicKeySync, getSpotMarketPublicKey, getSpotMarketPublicKeySync, } from '../addresses/pda'; import { WebSocketAccountSubscriber } from './webSocketAccountSubscriber'; import { Commitment, PublicKey } from '@solana/web3.js'; import { OracleInfo, OraclePriceData } from '../oracles/types'; import { OracleClientCache } from '../oracles/oracleClientCache'; import * as Buffer from 'buffer'; import { QUOTE_ORACLE_PRICE_DATA } from '../oracles/quoteAssetOracleClient'; import { findAllMarketAndOracles } from '../config'; import { findDelistedPerpMarketsAndOracles } from './utils'; import RemoteUtil, { MarketInfo, PublicAccountInfo } from './remoteUtil'; const ORACLE_DEFAULT_KEY = PublicKey.default.toBase58(); export class WebSocketDriftClientAccountSubscriber implements DriftClientAccountSubscriber { isSubscribed: boolean; program: Program; commitment?: Commitment; perpMarketIndexes: number[]; spotMarketIndexes: number[]; oracleInfos: OracleInfo[]; oracleClientCache = new OracleClientCache(); resubOpts?: ResubOpts; shouldFindAllMarketsAndOracles: boolean; eventEmitter: StrictEventEmitter; stateAccountSubscriber?: AccountSubscriber; perpMarketAccountSubscribers = new Map< number, AccountSubscriber >(); perpOracleMap = new Map(); perpOracleStringMap = new Map(); spotMarketAccountSubscribers = new Map< number, AccountSubscriber >(); spotOracleMap = new Map(); spotOracleStringMap = new Map(); oracleSubscribers = new Map>(); delistedMarketSetting: DelistedMarketSetting; initialPerpMarketAccountData: Map; initialSpotMarketAccountData: Map; initialOraclePriceData: Map; remoteUtil: RemoteUtil; publicAccountInfo: PublicAccountInfo; protected isSubscribing = false; protected subscriptionPromise: Promise; protected subscriptionPromiseResolver: (val: boolean) => void; public constructor( program: Program, perpMarketIndexes: number[], spotMarketIndexes: number[], oracleInfos: OracleInfo[], shouldFindAllMarketsAndOracles: boolean, delistedMarketSetting: DelistedMarketSetting, resubOpts?: ResubOpts, commitment?: Commitment ) { this.isSubscribed = false; this.program = program; this.eventEmitter = new EventEmitter(); this.perpMarketIndexes = perpMarketIndexes; this.spotMarketIndexes = spotMarketIndexes; this.oracleInfos = oracleInfos; this.shouldFindAllMarketsAndOracles = shouldFindAllMarketsAndOracles; this.delistedMarketSetting = delistedMarketSetting; this.resubOpts = resubOpts; this.commitment = commitment; // added by 0x_will this.remoteUtil = new RemoteUtil( this.eventEmitter, this.perpMarketAccountSubscribers, this.spotMarketAccountSubscribers, this.oracleSubscribers ); } public async subscribe(): Promise { if (this.isSubscribed) { return true; } if (this.isSubscribing) { return await this.subscriptionPromise; } this.isSubscribing = true; this.subscriptionPromise = new Promise((res) => { this.subscriptionPromiseResolver = res; }); this.publicAccountInfo = await this.remoteUtil.getPublicAccountInfo(); // if (this.shouldFindAllMarketsAndOracles) { // const { // perpMarketIndexes, // perpMarketAccounts, // spotMarketIndexes, // spotMarketAccounts, // oracleInfos, // } = await findAllMarketAndOracles(this.program); // this.perpMarketIndexes = perpMarketIndexes; // this.spotMarketIndexes = spotMarketIndexes; // this.oracleInfos = oracleInfos; // // front run and set the initial data here to save extra gma call in set initial data // this.initialPerpMarketAccountData = new Map( // perpMarketAccounts.map((market) => [market.marketIndex, market]) // ); // this.initialSpotMarketAccountData = new Map( // spotMarketAccounts.map((market) => [market.marketIndex, market]) // ); // } // const statePublicKey = await getDriftStateAccountPublicKey( // this.program.programId // ); // create and activate main state account subscription this.stateAccountSubscriber = new WebSocketAccountSubscriber( 'state', this.program, // statePublicKey, this.publicAccountInfo.statePublicKey, undefined, undefined, this.commitment ); // await this.stateAccountSubscriber.subscribe((data: StateAccount) => { // this.eventEmitter.emit('stateAccountUpdate', data); // this.eventEmitter.emit('update'); // }); // set initial data to avoid spamming getAccountInfo calls in webSocketAccountSubscriber // await this.setInitialData(); await Promise.all([ // subscribe to market accounts this.subscribeToPerpMarketAccounts(), // subscribe to spot market accounts this.subscribeToSpotMarketAccounts(), // subscribe to oracles this.subscribeToOracles(), ]); this.remoteUtil.setStateAccountSubscriber(this.stateAccountSubscriber); await this.remoteUtil.connect(); this.eventEmitter.emit('update'); // await this.handleDelistedMarkets(); await Promise.all([this.setPerpOracleMap(), this.setSpotOracleMap()]); this.isSubscribing = false; this.isSubscribed = true; this.subscriptionPromiseResolver(true); // delete initial data // this.removeInitialData(); return true; } async setInitialData(): Promise { const connection = this.program.provider.connection; if (!this.initialPerpMarketAccountData) { const perpMarketPublicKeys = this.perpMarketIndexes.map((marketIndex) => getPerpMarketPublicKeySync(this.program.programId, marketIndex) ); const perpMarketAccountInfos = await connection.getMultipleAccountsInfo( perpMarketPublicKeys ); this.initialPerpMarketAccountData = new Map( perpMarketAccountInfos .filter((accountInfo) => !!accountInfo) .map((accountInfo) => { const perpMarket = this.program.coder.accounts.decode( 'PerpMarket', accountInfo.data ); return [perpMarket.marketIndex, perpMarket]; }) ); } if (!this.initialSpotMarketAccountData) { const spotMarketPublicKeys = this.spotMarketIndexes.map((marketIndex) => getSpotMarketPublicKeySync(this.program.programId, marketIndex) ); const spotMarketAccountInfos = await connection.getMultipleAccountsInfo( spotMarketPublicKeys ); this.initialSpotMarketAccountData = new Map( spotMarketAccountInfos .filter((accountInfo) => !!accountInfo) .map((accountInfo) => { const spotMarket = this.program.coder.accounts.decode( 'SpotMarket', accountInfo.data ); return [spotMarket.marketIndex, spotMarket]; }) ); } const oracleAccountInfos = await connection.getMultipleAccountsInfo( this.oracleInfos.map((oracleInfo) => oracleInfo.publicKey) ); this.initialOraclePriceData = new Map( this.oracleInfos.reduce((result, oracleInfo, i) => { if (!oracleAccountInfos[i]) { return result; } const oracleClient = this.oracleClientCache.get( oracleInfo.source, connection, this.program ); const oraclePriceData = oracleClient.getOraclePriceDataFromBuffer( oracleAccountInfos[i].data ); result.push([oracleInfo.publicKey.toString(), oraclePriceData]); return result; }, []) ); } removeInitialData() { this.initialPerpMarketAccountData = new Map(); this.initialSpotMarketAccountData = new Map(); this.initialOraclePriceData = new Map(); } async subscribeToPerpMarketAccounts(): Promise { await Promise.all( // this.perpMarketIndexes.map((marketIndex) => // this.subscribeToPerpMarketAccount(marketIndex) // ) this.publicAccountInfo.perpMarketInfos.map((marketInfo) => this.subscribeToPerpMarketAccount(marketInfo.marketIndex, marketInfo.publicKey) ) ); return true; } async subscribeToPerpMarketAccount(marketIndex: number, publicKey?: PublicKey): Promise { // const perpMarketPublicKey = await getPerpMarketPublicKey( // this.program.programId, // marketIndex // ); let perpMarketPublicKey: PublicKey; if (publicKey) { perpMarketPublicKey = publicKey; } else { perpMarketPublicKey = await getPerpMarketPublicKey( this.program.programId, marketIndex ); } const accountSubscriber = new WebSocketAccountSubscriber( 'perpMarket', this.program, perpMarketPublicKey, undefined, this.resubOpts, this.commitment ); // accountSubscriber.setData( // this.initialPerpMarketAccountData.get(marketIndex) // ); // await accountSubscriber.subscribe((data: PerpMarketAccount) => { // this.eventEmitter.emit('perpMarketAccountUpdate', data); // this.eventEmitter.emit('update'); // }); this.perpMarketAccountSubscribers.set(marketIndex, accountSubscriber); return true; } async subscribeToSpotMarketAccounts(): Promise { await Promise.all( // this.spotMarketIndexes.map((marketIndex) => // this.subscribeToSpotMarketAccount(marketIndex) // ) this.publicAccountInfo.spotMarketInfos.map((marketInfo) => this.subscribeToSpotMarketAccount(marketInfo.marketIndex, marketInfo.publicKey) ) ); return true; } async subscribeToSpotMarketAccount(marketIndex: number, publicKey?: PublicKey): Promise { // const marketPublicKey = await getSpotMarketPublicKey( // this.program.programId, // marketIndex // ); let marketPublicKey: PublicKey; if (publicKey) { marketPublicKey = publicKey; } else { marketPublicKey = await getSpotMarketPublicKey( this.program.programId, marketIndex ); } const accountSubscriber = new WebSocketAccountSubscriber( 'spotMarket', this.program, marketPublicKey, undefined, this.resubOpts, this.commitment ); // accountSubscriber.setData( // this.initialSpotMarketAccountData.get(marketIndex) // ); // await accountSubscriber.subscribe((data: SpotMarketAccount) => { // this.eventEmitter.emit('spotMarketAccountUpdate', data); // this.eventEmitter.emit('update'); // }); this.spotMarketAccountSubscribers.set(marketIndex, accountSubscriber); return true; } async subscribeToOracles(): Promise { await Promise.all( // this.oracleInfos // .filter((oracleInfo) => !oracleInfo.publicKey.equals(PublicKey.default)) // .map((oracleInfo) => this.subscribeToOracle(oracleInfo)) this.publicAccountInfo.oracleInfos .filter((oracleInfo) => !oracleInfo.publicKey.equals(PublicKey.default)) .map((oracleInfo) => this.subscribeToOracle(oracleInfo)) ); return true; } async subscribeToOracle(oracleInfo: OracleInfo): Promise { const oracleString = oracleInfo.publicKey.toString(); const client = this.oracleClientCache.get( oracleInfo.source, this.program.provider.connection, this.program ); const accountSubscriber = new WebSocketAccountSubscriber( 'oracle', this.program, oracleInfo.publicKey, (buffer: Buffer) => { return client.getOraclePriceDataFromBuffer(buffer); }, this.resubOpts, this.commitment ); // const initialOraclePriceData = // this.initialOraclePriceData.get(oracleString); // if (initialOraclePriceData) { // accountSubscriber.setData(initialOraclePriceData); // } // await accountSubscriber.subscribe((data: OraclePriceData) => { // this.eventEmitter.emit('oraclePriceUpdate', oracleInfo.publicKey, data); // this.eventEmitter.emit('update'); // }); this.oracleSubscribers.set(oracleString, accountSubscriber); return true; } async unsubscribeFromMarketAccounts(): Promise { await Promise.all( Array.from(this.perpMarketAccountSubscribers.values()).map( (accountSubscriber) => accountSubscriber.unsubscribe() ) ); } async unsubscribeFromSpotMarketAccounts(): Promise { await Promise.all( Array.from(this.spotMarketAccountSubscribers.values()).map( (accountSubscriber) => accountSubscriber.unsubscribe() ) ); } async unsubscribeFromOracles(): Promise { await Promise.all( Array.from(this.oracleSubscribers.values()).map((accountSubscriber) => accountSubscriber.unsubscribe() ) ); } public async fetch(): Promise { if (!this.isSubscribed) { return; } const promises = [this.stateAccountSubscriber.fetch()] .concat( Array.from(this.perpMarketAccountSubscribers.values()).map( (subscriber) => subscriber.fetch() ) ) .concat( Array.from(this.spotMarketAccountSubscribers.values()).map( (subscriber) => subscriber.fetch() ) ); await Promise.all(promises); } public async unsubscribe(): Promise { if (!this.isSubscribed) { return; } await this.stateAccountSubscriber.unsubscribe(); await this.unsubscribeFromMarketAccounts(); await this.unsubscribeFromSpotMarketAccounts(); await this.unsubscribeFromOracles(); this.isSubscribed = false; } async addSpotMarket(marketIndex: number): Promise { if (this.spotMarketAccountSubscribers.has(marketIndex)) { return true; } const subscriptionSuccess = this.subscribeToSpotMarketAccount(marketIndex); await this.setSpotOracleMap(); return subscriptionSuccess; } async addPerpMarket(marketIndex: number): Promise { if (this.perpMarketAccountSubscribers.has(marketIndex)) { return true; } const subscriptionSuccess = this.subscribeToPerpMarketAccount(marketIndex); await this.setPerpOracleMap(); return subscriptionSuccess; } async addOracle(oracleInfo: OracleInfo): Promise { if (this.oracleSubscribers.has(oracleInfo.publicKey.toString())) { return true; } if (oracleInfo.publicKey.equals(PublicKey.default)) { return true; } return this.subscribeToOracle(oracleInfo); } async setPerpOracleMap() { const perpMarkets = this.getMarketAccountsAndSlots(); const addOraclePromises = []; for (const perpMarket of perpMarkets) { if (!perpMarket || !perpMarket.data) { continue; } const perpMarketAccount = perpMarket.data; const perpMarketIndex = perpMarketAccount.marketIndex; const oracle = perpMarketAccount.amm.oracle; if (!this.oracleSubscribers.has(oracle.toBase58())) { addOraclePromises.push( this.addOracle({ publicKey: oracle, source: perpMarket.data.amm.oracleSource, }) ); } this.perpOracleMap.set(perpMarketIndex, oracle); this.perpOracleStringMap.set(perpMarketIndex, oracle.toBase58()); } await Promise.all(addOraclePromises); } async setSpotOracleMap() { const spotMarkets = this.getSpotMarketAccountsAndSlots(); const addOraclePromises = []; for (const spotMarket of spotMarkets) { if (!spotMarket || !spotMarket.data) { continue; } const spotMarketAccount = spotMarket.data; const spotMarketIndex = spotMarketAccount.marketIndex; const oracle = spotMarketAccount.oracle; if (!this.oracleSubscribers.has(oracle.toBase58())) { addOraclePromises.push( this.addOracle({ publicKey: oracle, source: spotMarketAccount.oracleSource, }) ); } this.spotOracleMap.set(spotMarketIndex, oracle); this.spotOracleStringMap.set(spotMarketIndex, oracle.toBase58()); } await Promise.all(addOraclePromises); } async handleDelistedMarkets(): Promise { if (this.delistedMarketSetting === DelistedMarketSetting.Subscribe) { return; } const { perpMarketIndexes, oracles } = findDelistedPerpMarketsAndOracles( this.getMarketAccountsAndSlots(), this.getSpotMarketAccountsAndSlots() ); for (const perpMarketIndex of perpMarketIndexes) { await this.perpMarketAccountSubscribers .get(perpMarketIndex) .unsubscribe(); if (this.delistedMarketSetting === DelistedMarketSetting.Discard) { this.perpMarketAccountSubscribers.delete(perpMarketIndex); } } for (const oracle of oracles) { await this.oracleSubscribers.get(oracle.toBase58()).unsubscribe(); if (this.delistedMarketSetting === DelistedMarketSetting.Discard) { this.oracleSubscribers.delete(oracle.toBase58()); } } } assertIsSubscribed(): void { if (!this.isSubscribed) { throw new NotSubscribedError( 'You must call `subscribe` before using this function' ); } } public getStateAccountAndSlot(): DataAndSlot { this.assertIsSubscribed(); return this.stateAccountSubscriber.dataAndSlot; } public getMarketAccountAndSlot( marketIndex: number ): DataAndSlot | undefined { this.assertIsSubscribed(); return this.perpMarketAccountSubscribers.get(marketIndex).dataAndSlot; } public getMarketAccountsAndSlots(): DataAndSlot[] { return Array.from(this.perpMarketAccountSubscribers.values()).map( (subscriber) => subscriber.dataAndSlot ); } public getSpotMarketAccountAndSlot( marketIndex: number ): DataAndSlot | undefined { this.assertIsSubscribed(); return this.spotMarketAccountSubscribers.get(marketIndex).dataAndSlot; } public getSpotMarketAccountsAndSlots(): DataAndSlot[] { return Array.from(this.spotMarketAccountSubscribers.values()).map( (subscriber) => subscriber.dataAndSlot ); } public getOraclePriceDataAndSlot( oraclePublicKey: PublicKey | string ): DataAndSlot | undefined { this.assertIsSubscribed(); const oracleString = typeof oraclePublicKey === 'string' ? oraclePublicKey : oraclePublicKey.toBase58(); if (oracleString === ORACLE_DEFAULT_KEY) { return { data: QUOTE_ORACLE_PRICE_DATA, slot: 0, }; } return this.oracleSubscribers.get(oracleString).dataAndSlot; } public getOraclePriceDataAndSlotForPerpMarket( marketIndex: number ): DataAndSlot | undefined { const perpMarketAccount = this.getMarketAccountAndSlot(marketIndex); const oracle = this.perpOracleMap.get(marketIndex); const oracleString = this.perpOracleStringMap.get(marketIndex); if (!perpMarketAccount || !oracle) { return undefined; } if (!perpMarketAccount.data.amm.oracle.equals(oracle)) { // If the oracle has changed, we need to update the oracle map in background this.setPerpOracleMap(); } return this.getOraclePriceDataAndSlot(oracleString); } public getOraclePriceDataAndSlotForSpotMarket( marketIndex: number ): DataAndSlot | undefined { const spotMarketAccount = this.getSpotMarketAccountAndSlot(marketIndex); const oracle = this.spotOracleMap.get(marketIndex); const oracleString = this.spotOracleStringMap.get(marketIndex); if (!spotMarketAccount || !oracle) { return undefined; } if (!spotMarketAccount.data.oracle.equals(oracle)) { // If the oracle has changed, we need to update the oracle map in background this.setSpotOracleMap(); } return this.getOraclePriceDataAndSlot(oracleString); } }