Merge pull request #28 from drift-labs/master

publisher fix

example/clientWithSlot.ts

@@ -55,8 +55,8 @@ const driftClientPublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
 let lastSeenSlot = 0;
 
 /********** initializing driftClient as usual **********/
-var accountSubscription: DriftClientSubscriptionConfig;
-var logProviderConfig: LogProviderConfig;
+let accountSubscription: DriftClientSubscriptionConfig;
+let logProviderConfig: LogProviderConfig;
 if (useWebsocket) {
     accountSubscription = {
         type: 'websocket',

package.json

@@ -5,7 +5,7 @@
   "main": "lib/index.js",
   "license": "Apache-2.0",
   "dependencies": {
-    "@drift-labs/sdk": "2.48.0-beta.0",
+    "@drift-labs/sdk": "2.48.0-beta.9",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/auto-instrumentations-node": "^0.31.1",
     "@opentelemetry/exporter-prometheus": "^0.31.0",

src/index.ts

@@ -16,7 +16,10 @@ import {
 	DLOBOrdersCoder,
 	DLOBSubscriber,
 	DriftClient,
+	DriftClientSubscriptionConfig,
 	DriftEnv,
+	SlotSource,
+	SlotSubscriber,
 	UserMap,
 	UserStatsMap,
 	Wallet,
@@ -53,6 +56,7 @@ const sdkConfig = initialize({ env: process.env.ENV });
 const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
 export const ORDERBOOK_UPDATE_INTERVAL = 1000;
+const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
 
 const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
 	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
@@ -117,6 +121,7 @@ const endpoint = process.env.ENDPOINT;
 const wsEndpoint = process.env.WS_ENDPOINT;
 logger.info(`RPC endpoint: ${endpoint}`);
 logger.info(`WS endpoint:  ${wsEndpoint}`);
+logger.info(`useWebsocket: ${useWebsocket}`);
 logger.info(`DriftEnv:     ${driftEnv}`);
 logger.info(`Commit:       ${commitHash}`);
 
@@ -171,33 +176,51 @@ const main = async () => {
 		commitment: stateCommitment,
 	});
 
-	const bulkAccountLoader = new BulkAccountLoader(
-		connection,
-		stateCommitment,
-		ORDERBOOK_UPDATE_INTERVAL
-	);
+	// only set when polling
+	let bulkAccountLoader: BulkAccountLoader | undefined;
+
+	// only set when using websockets
+	let slotSubscriber: SlotSubscriber | undefined;
+
+	let accountSubscription: DriftClientSubscriptionConfig;
+	let slotSource: SlotSource;
+
+	if (!useWebsocket) {
+		bulkAccountLoader = new BulkAccountLoader(
+			connection,
+			stateCommitment,
+			ORDERBOOK_UPDATE_INTERVAL
+		);
+
+		accountSubscription = {
+			type: 'polling',
+			accountLoader: bulkAccountLoader,
+		};
+		slotSource = {
+			getSlot: () => bulkAccountLoader!.getSlot(),
+		};
+	} else {
+		accountSubscription = {
+			type: 'websocket',
+			commitment: stateCommitment,
+		};
+		slotSubscriber = new SlotSubscriber(connection);
+		await slotSubscriber.subscribe();
+		slotSource = {
+			getSlot: () => slotSubscriber!.getSlot(),
+		};
+	}
 
 	driftClient = new DriftClient({
 		connection,
 		wallet,
 		programID: clearingHousePublicKey,
-		accountSubscription: {
-			type: 'polling',
-			accountLoader: bulkAccountLoader,
-		},
+		accountSubscription,
 		env: driftEnv,
-		userStats: true,
 	});
 
 	const dlobCoder = DLOBOrdersCoder.create();
 
-	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');
@@ -205,28 +228,37 @@ const main = async () => {
 	});
 
 	setInterval(async () => {
-		lastSlotReceived = bulkAccountLoader.getSlot();
+		lastSlotReceived = slotSource.getSlot();
 	}, ORDERBOOK_UPDATE_INTERVAL);
 
+	const userStatsMap = new UserStatsMap(driftClient);
+
+	logger.info(`Initializing userMap...`);
+	const initUserMapStart = Date.now();
 	const userMap = new UserMap(
 		driftClient,
 		driftClient.userAccountSubscriptionConfig,
-		false
+		false,
+		async (authorities) => {
+			await userStatsMap.sync(authorities);
+		},
+		{ hasOpenOrders: true }
 	);
 	await userMap.subscribe();
-	const userStatsMap = new UserStatsMap(driftClient, {
-		type: 'polling',
-		accountLoader: new BulkAccountLoader(connection, stateCommitment, 0),
-	});
-	await userStatsMap.subscribe();
+	logger.info(`userMap initialized in ${Date.now() - initUserMapStart} ms`);
 
+	logger.info(`Initializing DLOBSubscriber...`);
+	const initDlobSubscriberStart = Date.now();
 	const dlobSubscriber = new DLOBSubscriber({
 		driftClient,
 		dlobSource: userMap,
-		slotSource: bulkAccountLoader,
+		slotSource,
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
 	});
 	await dlobSubscriber.subscribe();
+	logger.info(
+		`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms`
+	);
 
 	MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
 
@@ -253,7 +285,7 @@ const main = async () => {
 			// object with userAccount key and orders object serialized
 			const orders: Array<any> = [];
 			const oracles: Array<any> = [];
-			const slot = bulkAccountLoader.getSlot();
+			const slot = slotSource.getSlot();
 
 			for (const market of driftClient.getPerpMarketAccounts()) {
 				const oracle = driftClient.getOracleDataForPerpMarket(
@@ -297,7 +329,7 @@ const main = async () => {
 	app.get('/orders/json', async (_req, res, next) => {
 		try {
 			// object with userAccount key and orders object serialized
-			const slot = bulkAccountLoader.getSlot();
+			const slot = slotSource.getSlot();
 			const orders: Array<any> = [];
 			const oracles: Array<any> = [];
 			for (const market of driftClient.getPerpMarketAccounts()) {
@@ -460,7 +492,7 @@ const main = async () => {
 
 			res.end(
 				JSON.stringify({
-					slot: bulkAccountLoader.getSlot(),
+					slot: slotSource.getSlot(),
 					data: dlobCoder.encode(dlobOrders).toString('base64'),
 				})
 			);
@@ -549,7 +581,7 @@ const main = async () => {
 						.getDLOB()
 						.getRestingLimitBids(
 							normedMarketIndex,
-							bulkAccountLoader.getSlot(),
+							slotSource.getSlot(),
 							normedMarketType,
 							oracle
 						)
@@ -560,7 +592,7 @@ const main = async () => {
 						.getDLOB()
 						.getRestingLimitAsks(
 							normedMarketIndex,
-							bulkAccountLoader.getSlot(),
+							slotSource.getSlot(),
 							normedMarketType,
 							oracle
 						)

src/publishers/dlobPublisher.ts

@@ -66,7 +66,6 @@ const main = async () => {
 			accountLoader: bulkAccountLoader,
 		},
 		env: driftEnv,
-		userStats: true,
 	});
 
 	const slotSubscriber = new SlotSubscriber(connection, {});

src/publishers/tradesPublisher.ts

@@ -18,6 +18,7 @@ import {
 	PRICE_PRECISION,
 	getVariant,
 	ZERO,
+	BN,
 } from '@drift-labs/sdk';
 
 import { logger, setLogLevel } from '../utils/logger';
@@ -74,7 +75,6 @@ const main = async () => {
 			accountLoader: bulkAccountLoader,
 		},
 		env: driftEnv,
-		userStats: true,
 	});
 
 	const slotSubscriber = new SlotSubscriber(connection, {});
@@ -147,7 +147,9 @@ const main = async () => {
 					),
 					taker: fill.taker?.toBase58(),
 					takerOrderId: fill.takerOrderId,
-					takerOrderDirection: getVariant(fill.takerOrderDirection),
+					takerOrderDirection: fill.takerOrderDirection
+						? getVariant(fill.takerOrderDirection)
+						: undefined,
 					takerOrderBaseAssetAmount: convertToNumber(
 						fill.takerOrderBaseAssetAmount,
 						BASE_PRECISION
@@ -183,7 +185,7 @@ const main = async () => {
 					action: 'fill',
 					actionExplanation: getVariant(fill.actionExplanation),
 					referrerReward: convertToNumber(
-						fill.referrerReward ?? ZERO,
+						new BN(fill.referrerReward ?? ZERO),
 						QUOTE_PRECISION
 					),
 				};

yarn.lock

@@ -80,10 +80,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.48.0-beta.0":
-  version "2.48.0-beta.0"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.48.0-beta.0.tgz#dcda51d878bf458cc5e070528f8b0714f9c53589"
-  integrity sha512-okRb3ErrApZvks24AFMwLrmsdrcntLAUTRNcSfutg/Ri10hjMAIaiRdsYSbMZcDGWgjRq/+sozKIqiqfd7xfpg==
+"@drift-labs/sdk@2.48.0-beta.9":
+  version "2.48.0-beta.9"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.48.0-beta.9.tgz#2d0c99f58387ff3c5e45859ae4fe088339c8dd55"
+  integrity sha512-y0wrq680nv7Uwvcg9jOrjZPXxoae9yRg8M9B4bQgDFeNJmZNmcEmnLhWQ+CbPi8u/O0glhOdtKUk4UF9eSPXAg==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"