trade/trade-drift-dlob
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: update to common redis + top markers account flag

Browse Source
This commit is contained in:
Jack Waller
2024-05-16 12:39:44 +10:00
parent cd26f39c72
commit 956a031fec
13 changed files with 1699 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitmodules vendored Normal file
View File

@@ -0,0 +1,3 @@
[submodule "drift-common"]
path = drift-common
url = https://github.com/drift-labs/drift-common

8
Dockerfile
View File

@@ -5,7 +5,15 @@ RUN npm install -g yarn
RUN npm install -g typescript
WORKDIR /app
COPY drift-common /app/drift-common
COPY . .
WORKDIR /app/drift-common/protocol/sdk
RUN yarn
RUN yarn build
WORKDIR /app/drift-common/common-ts
RUN yarn
RUN yarn build
WORKDIR /app
RUN yarn
RUN yarn build

3
package.json
View File

@@ -6,7 +6,8 @@
"license": "Apache-2.0",
"dependencies": {
"@coral-xyz/anchor": "^0.29.0",
"@drift-labs/sdk": "2.82.0-beta.16",
"@drift/common": "file:./drift-common/common-ts",
"@drift-labs/sdk": "file:./drift-common/protocol/sdk",
"@opentelemetry/api": "^1.1.0",
"@opentelemetry/auto-instrumentations-node": "^0.31.1",
"@opentelemetry/exporter-prometheus": "^0.31.0",

31
src/dlob-subscriber/DLOBSubscriberIO.ts
View File

@@ -9,7 +9,8 @@ import {
groupL2,
isVariant,
} from '@drift-labs/sdk';
import { RedisClient } from '../utils/redisClient';
import { RedisClient } from '@drift/common';
import {
SubscriberLookup,
addMarketSlotToResponse,
@@ -261,25 +262,25 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
asks: l2Formatted.asks.slice(0, 5),
});
this.redisClient.client.publish(
this.redisClient.publish(
`orderbook_${marketType}_${marketArgs.marketIndex}`,
JSON.stringify(l2Formatted)
l2Formatted
);
this.redisClient.client.set(
this.redisClient.set(
`last_update_orderbook_${marketType}_${marketArgs.marketIndex}`,
JSON.stringify(l2Formatted_depth100)
l2Formatted_depth100
);
this.redisClient.client.set(
this.redisClient.set(
`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_100`,
JSON.stringify(l2Formatted_depth100)
l2Formatted_depth100
);
this.redisClient.client.set(
this.redisClient.set(
`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_20`,
JSON.stringify(l2Formatted_depth20)
l2Formatted_depth20
);
this.redisClient.client.set(
this.redisClient.set(
`last_update_orderbook_${marketType}_${marketArgs.marketIndex}_depth_5`,
JSON.stringify(l2Formatted_depth5)
l2Formatted_depth5
);
const oraclePriceData =
@@ -306,9 +307,9 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
numMakers: 4,
})
.map((x) => x.toString());
this.redisClient.client.set(
this.redisClient.set(
`last_update_orderbook_best_makers_${marketType}_${marketArgs.marketIndex}`,
JSON.stringify({ bids, asks, slot })
{ bids, asks, slot }
);
}
@@ -361,9 +362,9 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
marketArgs.marketIndex
);
this.redisClient.client.set(
this.redisClient.set(
`last_update_orderbook_l3_${marketType}_${marketArgs.marketIndex}`,
JSON.stringify(l3)
l3
);
}
}

123
src/index.ts
View File

@@ -19,6 +19,7 @@ import {
OrderSubscriber,
MarketType,
} from '@drift-labs/sdk';
import { RedisClient, RedisClientPrefix } from '@drift/common';
import { logger, setLogLevel } from './utils/logger';
@@ -42,43 +43,19 @@ import {
normalizeBatchQueryParams,
sleep,
validateDlobQuery,
getAccountFromId,
} from './utils/utils';
import FEATURE_FLAGS from './utils/featureFlags';
import { getDLOBProviderFromOrderSubscriber } from './dlobProvider';
import { RedisClient } from './utils/redisClient';
require('dotenv').config();
// Reading in Redis env vars
const REDIS_HOSTS = process.env.REDIS_HOSTS?.replace(/^\[|\]$/g, '')
const REDIS_CLIENTS = process.env.REDIS_CLIENTS?.replace(/^\[|\]$/g, '')
.split(',')
.map((host) => host.trim()) || ['localhost'];
const REDIS_PORTS = process.env.REDIS_PORTS?.replace(/^\[|\]$/g, '')
.split(',')
.map((port) => parseInt(port.trim(), 10)) || [6379];
const REDIS_PASSWORDS_ENV = process.env.REDIS_PASSWORDS || "['']";
.map((clients) => clients.trim()) || ['DLOB'];
let REDIS_PASSWORDS;
if (REDIS_PASSWORDS_ENV.trim() === "['']") {
REDIS_PASSWORDS = [undefined];
} else {
REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.replace(/^\[|\]$/g, '')
.split(/\s*,\s*/)
.map((pwd) => pwd.replace(/(^'|'$)/g, '').trim())
.map((pwd) => (pwd === '' ? undefined : pwd));
}
console.log('Redis Hosts:', REDIS_HOSTS);
console.log('Redis Ports:', REDIS_PORTS);
console.log('Redis Passwords:', REDIS_PASSWORDS);
if (
REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
REDIS_PORTS.length !== REDIS_HOSTS.length
) {
throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length';
}
console.log('Redis Clients:', REDIS_CLIENTS);
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
@@ -316,16 +293,11 @@ const main = async (): Promise<void> => {
> = new Map();
if (useRedis) {
logger.info('Connecting to redis');
for (let i = 0; i < REDIS_HOSTS.length; i++) {
redisClients.push(
new RedisClient(
REDIS_HOSTS[i],
REDIS_PORTS[i].toString(),
REDIS_PASSWORDS[i] || undefined
)
);
await redisClients[i].connect();
for (let i = 0; i < REDIS_CLIENTS.length; i++) {
const prefix = RedisClientPrefix[REDIS_CLIENTS[i]];
redisClients.push(new RedisClient({ prefix }));
}
for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
spotMarketRedisMap.set(sdkConfig.SPOT_MARKETS[i].marketIndex, {
client: redisClients[0],
@@ -344,6 +316,8 @@ const main = async (): Promise<void> => {
}
}
const userMapClient = new RedisClient({ prefix: RedisClientPrefix.USER_MAP });
function canRotate(marketType: MarketType, marketIndex: number) {
if (isVariant(marketType, 'spot')) {
const state = spotMarketRedisMap.get(marketIndex);
@@ -427,15 +401,21 @@ const main = async (): Promise<void> => {
try {
const { marketIndex, marketType } = req.query;
const fees = await redisClients[
parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0
].client.get(`priorityFees_${marketType}_${marketIndex}`);
const fees = await redisClients
.find(
(client) =>
client.forceGetClient().options.keyPrefix ===
RedisClientPrefix.DLOB_HELIUS
)
.getRaw(`priorityFees_${marketType}_${marketIndex}`);
if (fees) {
res.status(200).json({
...JSON.parse(fees),
marketType,
marketIndex,
});
return;
} else {
res.writeHead(404);
@@ -466,11 +446,16 @@ const main = async (): Promise<void> => {
const fees = await Promise.all(
normedParams.map(async (normedParam) => {
const fees = await redisClients[
parseInt(process.env.HELIUS_REDIS_HOST_INDEX) ?? 0
].client.get(
`priorityFees_${normedParam['marketType']}_${normedParam['marketIndex']}`
);
const fees = await redisClients
.find(
(client) =>
client.forceGetClient().options.keyPrefix ===
RedisClientPrefix.DLOB_HELIUS
)
.getRaw(
`priorityFees_${normedParam['marketType']}_${normedParam['marketIndex']}`
);
return {
...JSON.parse(fees),
marketType: normedParam['marketType'],
@@ -499,6 +484,7 @@ const main = async (): Promise<void> => {
marketType,
side, // bid or ask
limit,
includeAccounts,
} = req.query;
const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
@@ -533,12 +519,17 @@ const main = async (): Promise<void> => {
normedLimit = 4;
}
let accountFlag = false;
if (includeAccounts) {
accountFlag = includeAccounts === 'true';
}
let topMakers: string[];
if (useRedis) {
const redisClient = isVariant(normedMarketType, 'perp')
? perpMarketRedisMap.get(normedMarketIndex).client
: spotMarketRedisMap.get(normedMarketIndex).client;
const redisResponse = await redisClient.client.get(
const redisResponse = await redisClient.getRaw(
`last_update_orderbook_best_makers_${getVariant(
normedMarketType
)}_${marketIndex}`
@@ -565,6 +556,13 @@ const main = async (): Promise<void> => {
path: req.baseUrl + req.path,
});
res.writeHead(200);
if (accountFlag) {
const topAccounts = await getAccountFromId(userMapClient, topMakers);
res.end(JSON.stringify(topAccounts));
return;
}
res.end(JSON.stringify(topMakers));
return;
}
@@ -619,6 +617,13 @@ const main = async (): Promise<void> => {
path: req.baseUrl + req.path,
});
res.writeHead(200);
if (accountFlag) {
const topAccounts = await getAccountFromId(userMapClient, topMakers);
res.end(JSON.stringify(topAccounts));
return;
}
res.end(JSON.stringify(topMakers));
} catch (err) {
next(err);
@@ -661,15 +666,15 @@ const main = async (): Promise<void> => {
let redisL2: string;
const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
);
}
@@ -694,15 +699,15 @@ const main = async (): Promise<void> => {
let redisL2: string;
const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
);
}
@@ -834,15 +839,15 @@ const main = async (): Promise<void> => {
const redisClient =
perpMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
);
}
@@ -870,15 +875,15 @@ const main = async (): Promise<void> => {
const redisClient =
spotMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
);
}
@@ -976,7 +981,7 @@ const main = async (): Promise<void> => {
const redisClient = (
marketTypeStr === 'spot' ? spotMarketRedisMap : perpMarketRedisMap
).get(normedMarketIndex).client;
const redisL3 = await redisClient.client.get(
const redisL3 = await redisClient.getRaw(
`last_update_orderbook_l3_${marketTypeStr}_${normedMarketIndex}`
);
if (

15
src/publishers/dlobPublisher.ts
View File

@@ -18,6 +18,7 @@ import {
PerpMarketConfig,
SpotMarketConfig,
} from '@drift-labs/sdk';
import { RedisClient, RedisClientPrefix } from '@drift/common';
import { logger, setLogLevel } from '../utils/logger';
import {
@@ -31,7 +32,6 @@ import {
DLOBSubscriberIO,
wsMarketInfo,
} from '../dlob-subscriber/DLOBSubscriberIO';
import { RedisClient } from '../utils/redisClient';
import {
DLOBProvider,
getDLOBProviderFromGrpcOrderSubscriber,
@@ -47,9 +47,8 @@ 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;
const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB';
// Set up express for health checks
const app = express();
@@ -259,6 +258,11 @@ const main = async () => {
const wallet = new Wallet(new Keypair());
const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
const redisClient = new RedisClient({
prefix: RedisClientPrefix[REDIS_CLIENT],
});
await redisClient.connect();
const connection = new Connection(endpoint, {
wsEndpoint: wsEndpoint,
commitment: stateCommitment,
@@ -391,9 +395,6 @@ const main = async () => {
await dlobProvider.subscribe();
const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
await redisClient.connect();
const dlobSubscriber = new DLOBSubscriberIO({
driftClient,
env: driftEnv,

29
src/publishers/priorityFeesPublisher.ts
View File

@@ -10,20 +10,17 @@ import {
BulkAccountLoader,
getMarketsAndOraclesForSubscription,
} from '@drift-labs/sdk';
import { RedisClient, RedisClientPrefix } from '@drift/common';
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();
@@ -137,25 +134,25 @@ class PriorityFeeSubscriber {
dataPerp.forEach((result: any) => {
const marketIndex = parseInt(result['id']);
this.redisClient.client.publish(
this.redisClient.publish(
`priorityFees_perp_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
result.result['priorityFeeLevels']
);
this.redisClient.client.set(
this.redisClient.set(
`priorityFees_perp_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
result.result['priorityFeeLevels']
);
});
dataSpot.forEach((result: any) => {
const marketIndex = parseInt(result['id']) - 100;
this.redisClient.client.publish(
this.redisClient.publish(
`priorityFees_spot_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
result.result['priorityFeeLevels']
);
this.redisClient.client.set(
this.redisClient.set(
`priorityFees_spot_${marketIndex}`,
JSON.stringify(result.result['priorityFeeLevels'])
result.result['priorityFeeLevels']
);
});
}
@@ -167,6 +164,11 @@ const main = async () => {
commitment: stateCommitment,
});
const redisClient = new RedisClient({
prefix: RedisClientPrefix.DLOB_HELIUS,
});
await redisClient.connect();
const { perpMarketIndexes, spotMarketIndexes, oracleInfos } =
getMarketsAndOraclesForSubscription(sdkConfig.ENV);
@@ -209,9 +211,6 @@ const main = async () => {
});
}
const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
await redisClient.connect();
const priorityFeeSubscriber = new PriorityFeeSubscriber({
endpoint,
perpMarketPubkeys,

15
src/publishers/tradesPublisher.ts
View File

@@ -20,18 +20,15 @@ import {
OrderActionRecord,
Event,
} from '@drift-labs/sdk';
import { RedisClient, RedisClientPrefix } from '@drift/common';
import { logger, setLogLevel } from '../utils/logger';
import { sleep } from '../utils/utils';
import { RedisClient } from '../utils/redisClient';
import { fromEvent, filter, map } from 'rxjs';
require('dotenv').config();
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;
//@ts-ignore
const sdkConfig = initialize({ env: process.env.ENV });
@@ -70,6 +67,9 @@ const main = async () => {
env: driftEnv,
});
const redisClient = new RedisClient({ prefix: RedisClientPrefix.DLOB });
await redisClient.connect();
const slotSubscriber = new SlotSubscriber(connection, {});
const lamportsBalance = await connection.getBalance(wallet.publicKey);
@@ -86,9 +86,6 @@ const main = async () => {
await slotSubscriber.subscribe();
const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
await redisClient.connect();
const eventSubscriber = new EventSubscriber(connection, driftClient.program, {
maxTx: 8192,
maxEventsPerType: 4096,
@@ -182,9 +179,9 @@ const main = async () => {
})
)
.subscribe((fillEvent) => {
redisClient.client.publish(
redisClient.publish(
`trades_${fillEvent.marketType}_${fillEvent.marketIndex}`,
JSON.stringify(fillEvent)
fillEvent
);
});

66
src/serverLite.ts
View File

@@ -16,7 +16,6 @@ import {
} from './core/metrics';
import { handleResponseTime } from './core/middleware';
import { errorHandler, normalizeBatchQueryParams, sleep } from './utils/utils';
import { RedisClient } from './utils/redisClient';
import {
DriftEnv,
MarketType,
@@ -26,36 +25,16 @@ import {
isVariant,
initialize,
} from '@drift-labs/sdk';
import { RedisClient, RedisClientPrefix } from '@drift/common';
require('dotenv').config();
// Reading in Redis env vars
const REDIS_HOSTS_ENV = (process.env.REDIS_HOSTS as string) || 'localhost';
const REDIS_HOSTS = REDIS_HOSTS_ENV.includes(',')
? REDIS_HOSTS_ENV.trim()
.replace(/^\[|\]$/g, '')
.split(/\s*,\s*/)
: [REDIS_HOSTS_ENV];
const REDIS_CLIENTS = process.env.REDIS_CLIENTS?.replace(/^\[|\]$/g, '')
.split(',')
.map((clients) => clients.trim()) || ['DLOB'];
const REDIS_PASSWORDS_ENV = (process.env.REDIS_PASSWORDS as string) || '';
const REDIS_PASSWORDS = REDIS_PASSWORDS_ENV.includes(',')
? REDIS_PASSWORDS_ENV.trim()
.replace(/^\[|\]$/g, '')
.split(/\s*,\s*/)
: [REDIS_PASSWORDS_ENV];
const REDIS_PORTS_ENV = (process.env.REDIS_PORTS as string) || '6379';
const REDIS_PORTS = REDIS_PORTS_ENV.includes(',')
? REDIS_PORTS_ENV.trim()
.replace(/^\[|\]$/g, '')
.split(/\s*,\s*/)
: [REDIS_PORTS_ENV];
if (
REDIS_PORTS.length !== REDIS_PASSWORDS.length ||
REDIS_PORTS.length !== REDIS_HOSTS.length
) {
throw 'REDIS_HOSTS and REDIS_PASSWORDS and REDIS_PORTS must be the same length';
}
console.log('Redis Clients:', REDIS_CLIENTS);
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
@@ -145,14 +124,9 @@ const main = async () => {
number,
{ client: RedisClient; clientIndex: number }
> = new Map();
for (let i = 0; i < REDIS_HOSTS.length; i++) {
redisClients.push(
new RedisClient(
REDIS_HOSTS[i],
REDIS_PORTS[i],
REDIS_PASSWORDS[i] || undefined
)
);
for (let i = 0; i < REDIS_CLIENTS.length; i++) {
const prefix = RedisClientPrefix[REDIS_CLIENTS[i]];
redisClients.push(new RedisClient({ prefix }));
await redisClients[i].connect();
}
for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) {
@@ -245,15 +219,15 @@ const main = async () => {
let redisL2: string;
const redisClient = perpMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
);
}
@@ -287,15 +261,15 @@ const main = async () => {
let redisL2: string;
const redisClient = spotMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
);
}
@@ -410,15 +384,15 @@ const main = async () => {
const redisClient =
perpMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
);
}
@@ -455,15 +429,15 @@ const main = async () => {
const redisClient =
spotMarketRedisMap.get(normedMarketIndex).client;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
redisL2 = await redisClient.getRaw(
`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
);
}

124
src/utils/redisClient.ts
View File

@@ -1,124 +0,0 @@
import Redis, { RedisOptions } from 'ioredis';
import { sleep } from './utils';
export const getRedisClient = (
host?: string,
port?: string,
password?: string,
db?: number,
opts?: RedisOptions
): Redis => {
if (host && port) {
console.log(`Connecting to configured redis:: ${host}:${port}`);
const redisClient = new Redis({
host: host,
port: parseInt(port, 10),
password: password,
...(opts ?? {}),
retryStrategy: (times) => {
const delay = Math.min(times * 1000, 10000);
console.log(
`Reconnecting to Redis in ${delay}ms... (retries: ${times})`
);
return delay;
},
reconnectOnError: (err) => {
const targetError = 'ECONNREFUSED';
if (err.message.includes(targetError)) {
console.log(
`Redis error: ${targetError}. Attempting to reconnect...`
);
return true;
}
return false;
},
maxRetriesPerRequest: null, // unlimited retries
});
redisClient.on('connect', () => {
console.log('Connected to Redis.');
});
redisClient.on('error', (err) => {
console.error('Redis error:', err);
});
redisClient.on('reconnecting', () => {
console.log('Reconnecting to Redis...');
});
return redisClient;
}
console.log(`Using default redis`);
return new Redis({ db });
};
/**
* Wrapper around the redis client.
*
* You can hover over the underlying redis client methods for explanations of the methods, but will also include links to DOCS for some important concepts below:
*
* zRange, zRangeByScore etc.:
* - All of the "z" methods are methods that use sorted sets.
* - Sorted sets are explained here : https://redis.io/docs/data-types/sorted-sets/
*/
export class RedisClient {
public client: Redis;
connectionPromise: Promise<void>;
constructor(
host?: string,
port?: string,
password?: string,
db?: number,
opts?: RedisOptions
) {
this.client = getRedisClient(host, port, password, db, opts);
}
/**
* Should avoid using this unless necessary.
* @returns
*/
public forceGetClient() {
return this.client;
}
public get connected() {
return this?.client?.status === 'ready';
}
async connect() {
if (this.client.status === 'ready' || this.client.status === 'connect') {
return;
}
if (this.client.status === 'connecting') {
await sleep(100);
return this.connect(); // recursive call to check again
}
try {
await this.client.connect();
} catch (e) {
console.error(e);
}
}
disconnect() {
this.client.disconnect();
}
private assertConnected() {
if (!this.connected) {
throw 'Redis client not connected';
}
}
private getPrefixForMetrics(key: string) {
return key.split(':')?.[0];
}
}

24
src/utils/utils.ts
View File

@@ -11,8 +11,11 @@ import {
SerumSubscriber,
SpotMarketConfig,
SpotMarkets,
decodeUser,
isVariant,
} from '@drift-labs/sdk';
import { RedisClient } from '@drift/common';
import { logger } from './logger';
import { NextFunction, Request, Response } from 'express';
import FEATURE_FLAGS from './featureFlags';
@@ -338,6 +341,27 @@ export const validateDlobQuery = (
};
};
export const getAccountFromId = async (
userMapClient: RedisClient,
topMakers: string[]
) => {
return Promise.all(
topMakers.map(async (accountId) => {
const userMap = await userMapClient.getRaw(accountId);
if (userMap) {
return {
accountId,
account: decodeUser(Buffer.from(userMap.split('::')[1], 'base64')),
};
}
return {
accountId,
account: null,
};
})
).then((results) => results.filter((user) => !!user));
};
export function errorHandler(
err: Error,
_req: Request,

27
src/wsConnectionManager.ts
View File

@@ -4,9 +4,9 @@ import * as http from 'http';
import compression from 'compression';
import { WebSocket, WebSocketServer } from 'ws';
import { sleep } from './utils/utils';
import { RedisClient } from './utils/redisClient';
import { register, Gauge } from 'prom-client';
import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk';
import { RedisClient } from '@drift/common';
// Set up env constants
require('dotenv').config();
@@ -29,13 +29,10 @@ const wss = new WebSocketServer({
perMessageDeflate: true,
});
const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
const REDIS_PORT = process.env.REDIS_PORT || '6379';
const WS_PORT = process.env.WS_PORT || '3000';
console.log(`WS LISTENER PORT : ${WS_PORT}`);
const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
const MAX_BUFFERED_AMOUNT = 300000;
const safeGetRawChannelFromMessage = (message: any): string => {
@@ -79,12 +76,8 @@ const getRedisChannelFromMessage = (message: any): string => {
};
async function main() {
const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
const lastMessageRetriever = new RedisClient(
REDIS_HOST,
REDIS_PORT,
REDIS_PASSWORD
);
const redisClient = new RedisClient({});
const lastMessageRetriever = new RedisClient({});
await redisClient.connect();
await lastMessageRetriever.connect();
@@ -92,17 +85,17 @@ async function main() {
const channelSubscribers = new Map<string, Set<WebSocket>>();
const subscribedChannels = new Set<string>();
redisClient.client.on('connect', () => {
redisClient.forceGetClient().on('connect', () => {
subscribedChannels.forEach(async (channel) => {
try {
await redisClient.client.subscribe(channel);
await redisClient.subscribe(channel);
} catch (error) {
console.error(`Error subscribing to ${channel}:`, error);
}
});
});
redisClient.client.on('message', (subscribedChannel, message) => {
redisClient.forceGetClient().on('message', (subscribedChannel, message) => {
const subscribers = channelSubscribers.get(subscribedChannel);
if (subscribers) {
subscribers.forEach((ws) => {
@@ -117,7 +110,7 @@ async function main() {
}
});
redisClient.client.on('error', (error) => {
redisClient.forceGetClient().on('error', (error) => {
console.error('Redis client error:', error);
});
@@ -162,7 +155,7 @@ async function main() {
if (!subscribedChannels.has(redisChannel)) {
console.log('Trying to subscribe to channel', redisChannel);
redisClient.client
redisClient
.subscribe(redisChannel)
.then(() => {
subscribedChannels.add(redisChannel);
@@ -191,7 +184,7 @@ async function main() {
// Fetch and send last message
if (redisChannel.includes('orderbook')) {
const lastUpdateChannel = `last_update_${redisChannel}`;
const lastMessage = await lastMessageRetriever.client.get(
const lastMessage = await lastMessageRetriever.get(
lastUpdateChannel
);
@@ -270,7 +263,7 @@ async function main() {
clearInterval(bufferInterval);
channelSubscribers.forEach((subscribers, channel) => {
if (subscribers.delete(ws) && subscribers.size === 0) {
redisClient.client.unsubscribe(channel);
redisClient.unsubscribe(channel);
channelSubscribers.delete(channel);
subscribedChannels.delete(channel);
}

1547
yarn.lock
View File

File diff suppressed because it is too large Load Diff