remove grpc impl, use sdk grpc OrderSubscriber

src/publishers/dlobPublisher.ts

@@ -20,6 +20,7 @@ import {
 	PhoenixSubscriber,
 	MarketType,
 	OraclePriceData,
+	OrderSubscriberConfig,
 } from '@drift-labs/sdk';
 import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 
@@ -37,12 +38,10 @@ import {
 } from '../dlob-subscriber/DLOBSubscriberIO';
 import {
 	DLOBProvider,
-	getDLOBProviderFromGrpcOrderSubscriber,
 	getDLOBProviderFromOrderSubscriber,
 	getDLOBProviderFromUserMap,
 } from '../dlobProvider';
 import FEATURE_FLAGS from '../utils/featureFlags';
-import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC';
 import express, { Response, Request } from 'express';
 import { handleHealthCheck } from '../core/metrics';
 import { setGlobalDispatcher, Agent } from 'undici';
@@ -76,6 +75,17 @@ const dlobSlotGauge = new Gauge({
 		'redisClient',
 	],
 });
+const oracleSlotGauge = new Gauge({
+	name: 'oracle_slot',
+	help: 'Last updated slot of oracle',
+	labelNames: [
+		'marketIndex',
+		'marketType',
+		'marketName',
+		'redisPrefix',
+		'redisClient',
+	],
+});
 
 //@ts-ignore
 const sdkConfig = initialize({ env: process.env.ENV });
@@ -116,6 +126,10 @@ const SPOT_MARKETS_TO_LOAD =
 
 logger.info(`RPC endpoint: ${endpoint}`);
 logger.info(`WS endpoint:  ${wsEndpoint}`);
+logger.info(`Token:        ${token}`);
+logger.info(
+	`useOrderSubscriber: ${useOrderSubscriber}, useWebsocket: ${useWebsocket}, useGrpc: ${useGrpc}`
+);
 logger.info(`DriftEnv:     ${driftEnv}`);
 logger.info(`Commit:       ${commitHash}`);
 
@@ -377,17 +391,35 @@ const main = async () => {
 
 	let dlobProvider: DLOBProvider;
 	if (useOrderSubscriber) {
-		let subscriptionConfig;
+		let subscriptionConfig: any = {
+			type: 'polling',
+			commitment: stateCommitment,
+			frequency: ORDERBOOK_UPDATE_INTERVAL,
+		};
+
 		if (useWebsocket) {
 			subscriptionConfig = {
 				type: 'websocket',
 				commitment: stateCommitment,
 			};
-		} else {
+		}
+
+		// USE_GRPC=true will override websocket
+		if (useGrpc) {
+			if (!token) {
+				throw new Error('TOKEN is required for grpc');
+			}
+			if (!endpoint) {
+				throw new Error('ENDPOINT is required for grpc');
+			}
 			subscriptionConfig = {
-				type: 'polling',
+				type: 'grpc',
+				grpcConfigs: {
+					endpoint: endpoint,
+					token: token,
+					commitmentLevel: stateCommitment,
+				},
 				commitment: stateCommitment,
-				frequency: ORDERBOOK_UPDATE_INTERVAL,
 			};
 		}
 
@@ -401,17 +433,6 @@ const main = async () => {
 		slotSource = {
 			getSlot: () => orderSubscriber.getSlot(),
 		};
-	} else if (useGrpc) {
-		const grpcOrderSubscriber = new GeyserOrderSubscriber(driftClient, {
-			endpoint: endpoint,
-			token: token,
-		});
-
-		dlobProvider = getDLOBProviderFromGrpcOrderSubscriber(grpcOrderSubscriber);
-
-		slotSource = {
-			getSlot: () => grpcOrderSubscriber.getSlot(),
-		};
 	} else {
 		const userMap = new UserMap({
 			driftClient,
@@ -463,6 +484,9 @@ const main = async () => {
 	setInterval(() => {
 		const slot = slotSource.getSlot();
 		perpMarketInfos.forEach((market) => {
+			const oracleDataAndSlot = driftClient.getOracleDataForPerpMarket(
+				market.marketIndex
+			);
 			dlobSlotGauge.set(
 				{
 					marketIndex: market.marketIndex,
@@ -473,8 +497,21 @@ const main = async () => {
 				},
 				slot
 			);
+			oracleSlotGauge.set(
+				{
+					marketIndex: market.marketIndex,
+					marketType: 'perp',
+					marketName: market.marketName,
+					redisClient: REDIS_CLIENT,
+					redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				},
+				oracleDataAndSlot.slot.toNumber()
+			);
 		});
 		spotMarketInfos.forEach((market) => {
+			const oracleDataAndSlot = driftClient.getOracleDataForSpotMarket(
+				market.marketIndex
+			);
 			dlobSlotGauge.set(
 				{
 					marketIndex: market.marketIndex,
@@ -483,7 +520,7 @@ const main = async () => {
 					redisClient: REDIS_CLIENT,
 					redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 				},
-				slot
+				oracleDataAndSlot.slot.toNumber()
 			);
 		});
 	}, 10_000);