chore(snapshot): sync workspace state

Dockerfile

@@ -1,25 +1,29 @@
-FROM public.ecr.aws/docker/library/node:22 AS builder
+FROM public.ecr.aws/docker/library/node:24 AS builder
 RUN npm install -g typescript@5.4.5
+RUN yarn config set registry https://registry.npmjs.org
 
 WORKDIR /app
 
 COPY . .
 
-WORKDIR /app/drift-common/protocol/sdk
+RUN rm -rf /app/dlob-server/drift-common && \
+    cp -a /app/_tmp/trade-drift-dlob-main-20260313/drift-common /app/dlob-server/drift-common
+
+WORKDIR /app/dlob-server/drift-common/protocol/sdk
 RUN yarn && yarn build
 
-WORKDIR /app/drift-common/common-ts
+WORKDIR /app/dlob-server/drift-common/common-ts
 RUN yarn && yarn build
 
-WORKDIR /app
+WORKDIR /app/dlob-server
 RUN yarn && yarn build
 
-FROM public.ecr.aws/docker/library/node:22-alpine
+FROM public.ecr.aws/docker/library/node:24-alpine
 RUN apk add python3 make g++ --virtual .build &&\
     npm install -C /lib bigint-buffer @triton-one/yellowstone-grpc@1.3.0 helius-laserstream@0.1.8 rpc-websockets@7.5.1 &&\
     apk del .build
     
-COPY --from=builder /app/lib/ ./lib/
+COPY --from=builder /app/dlob-server/lib/ ./lib/
 
 ENV NODE_ENV=production
 EXPOSE 9464

package.json

@@ -32,6 +32,7 @@
 		"express": "^4.18.2",
 		"ioredis": "^5.4.1",
 		"morgan": "^1.10.0",
+		"pg": "^8.12.0",
 		"redis": "^4.6.10",
 		"response-time": "^2.3.2",
 		"rxjs": "^7.8.1",

src/athena/testRedisPolling.ts

@@ -1,4 +1,4 @@
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
+import { RedisClient, HOT_REDIS_KEY_PREFIX } from '../utils/redisClients';
 import { logger, setLogLevel } from '../utils/logger';
 
 setLogLevel('info');
@@ -194,10 +194,10 @@ const main = async () => {
 	logger.info(`📡 Connecting to Redis:`);
 	logger.info(`   Host: ${redisHost}`);
 	logger.info(`   Port: ${redisPort}`);
-	logger.info(`   Prefix: ${RedisClientPrefix.DLOB}`);
+	logger.info(`   Prefix: ${HOT_REDIS_KEY_PREFIX}`);
 
 	const redisClient = new RedisClient({
-		prefix: RedisClientPrefix.DLOB,
+		prefix: HOT_REDIS_KEY_PREFIX,
 	});
 
 	try {

src/dlob-subscriber/OrderSubscriberFiltered.ts

@@ -3,6 +3,7 @@ import {
 	OrderSubscriberConfig,
 	UserAccount,
 } from '@drift-labs/sdk';
+import { ReconnectingGrpcOrderSubscription } from './ReconnectingGrpcOrderSubscription';
 
 export class OrderSubscriberFiltered extends OrderSubscriber {
 	public ignoreList: string[];
@@ -14,6 +15,20 @@ export class OrderSubscriberFiltered extends OrderSubscriber {
 	) {
 		super(config);
 		this.ignoreList = config.ignoreList ?? [];
+
+		if (config.subscriptionConfig.type === 'grpc') {
+			(this as any).subscription = new ReconnectingGrpcOrderSubscription({
+				orderSubscriber: this,
+				grpcConfigs: config.subscriptionConfig.grpcConfigs,
+				skipInitialLoad: config.subscriptionConfig.skipInitialLoad,
+				resubOpts: {
+					resubTimeoutMs: config.subscriptionConfig.resubTimeoutMs,
+					logResubMessages: config.subscriptionConfig.logResubMessages,
+				},
+				resyncIntervalMs: config.subscriptionConfig.resyncIntervalMs,
+				decoded: config.decodeData,
+			});
+		}
 	}
 
 	override async tryUpdateUserAccount(

src/dlob-subscriber/ReconnectingGrpcOrderSubscription.ts

@@ -0,0 +1,614 @@
+import type { OrderSubscriber, UserAccount } from '@drift-labs/sdk';
+import { Context, MemcmpFilter, PublicKey } from '@solana/web3.js';
+import { Buffer } from 'buffer';
+import {
+	CommitmentLevel,
+	createClient,
+	type Client,
+	type ClientDuplexStream,
+	type SubscribeRequest,
+	type SubscribeUpdate,
+} from '@drift-labs/sdk/lib/node/isomorphic/grpc';
+import type {
+	GrpcConfigs,
+	ResubOpts,
+} from '@drift-labs/sdk/lib/node/accounts/types';
+import { WebSocketProgramAccountSubscriber } from '@drift-labs/sdk/lib/node/accounts/webSocketProgramAccountSubscriber';
+import { LaserstreamProgramAccountSubscriber } from '@drift-labs/sdk/lib/node/accounts/laserProgramAccountSubscriber';
+import { getNonIdleUserFilter, getUserFilter } from '@drift-labs/sdk';
+import bs58 from 'bs58';
+import { logger } from '../utils/logger';
+
+type StreamFailureEvent = 'error' | 'end' | 'close';
+
+function sleep(ms: number): Promise<void> {
+	return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function getReconnectDelayMs(): number {
+	const raw = process.env.GRPC_RECONNECT_DELAY_MS;
+	const parsed = raw ? parseInt(raw, 10) : 1000;
+	return Number.isFinite(parsed) && parsed >= 0 ? parsed : 1000;
+}
+
+function formatError(error: unknown): string {
+	if (error instanceof Error) {
+		return `${error.name}: ${error.message}`;
+	}
+	return String(error);
+}
+
+function isExpectedGrpcTeardownError(error: unknown): boolean {
+	if (!error || typeof error !== 'object') {
+		return false;
+	}
+
+	const maybeGrpcError = error as {
+		code?: unknown;
+		details?: unknown;
+		message?: unknown;
+	};
+
+	const details =
+		typeof maybeGrpcError.details === 'string' ? maybeGrpcError.details : '';
+	const message =
+		typeof maybeGrpcError.message === 'string' ? maybeGrpcError.message : '';
+
+	return (
+		maybeGrpcError.code === 1 ||
+		details.includes('Cancelled on client') ||
+		message.includes('Cancelled on client')
+	);
+}
+
+class ReconnectingGrpcProgramAccountSubscriber<
+	T,
+> extends WebSocketProgramAccountSubscriber<T> {
+	private client: Client;
+	private stream?: ClientDuplexStream<SubscribeRequest, SubscribeUpdate>;
+	private commitmentLevel: CommitmentLevel;
+	private streamFailureHandled = false;
+	private readonly onStreamFailure?: (
+		event: StreamFailureEvent,
+		error?: unknown
+	) => void;
+
+	private dataHandler?: (chunk: SubscribeUpdate) => void;
+	private errorHandler?: (error: Error) => void;
+	private endHandler?: () => void;
+	private closeHandler?: () => void;
+
+	private constructor(
+		client: Client,
+		commitmentLevel: CommitmentLevel,
+		subscriptionName: string,
+		accountDiscriminator: string,
+		program: any,
+		decodeBufferFn: (accountName: string, ix: Buffer) => T,
+		options: { filters: MemcmpFilter[] } = {
+			filters: [],
+		},
+		resubOpts?: ResubOpts,
+		onStreamFailure?: (event: StreamFailureEvent, error?: unknown) => void
+	) {
+		super(
+			subscriptionName,
+			accountDiscriminator,
+			program,
+			decodeBufferFn,
+			options,
+			resubOpts
+		);
+		this.client = client;
+		this.commitmentLevel = commitmentLevel;
+		this.onStreamFailure = onStreamFailure;
+	}
+
+	public static async create<U>(
+		grpcConfigs: GrpcConfigs,
+		subscriptionName: string,
+		accountDiscriminator: string,
+		program: any,
+		decodeBufferFn: (accountName: string, ix: Buffer) => U,
+		options: { filters: MemcmpFilter[] } = {
+			filters: [],
+		},
+		resubOpts?: ResubOpts,
+		onStreamFailure?: (event: StreamFailureEvent, error?: unknown) => void
+	): Promise<ReconnectingGrpcProgramAccountSubscriber<U>> {
+		const client = await createClient(
+			grpcConfigs.endpoint,
+			grpcConfigs.token,
+			grpcConfigs.channelOptions ?? {}
+		);
+		const commitmentLevel =
+			// @ts-ignore runtime enum value is valid, TS types are incomplete here
+			grpcConfigs.commitmentLevel ?? CommitmentLevel.CONFIRMED;
+
+		return new ReconnectingGrpcProgramAccountSubscriber<U>(
+			client,
+			commitmentLevel,
+			subscriptionName,
+			accountDiscriminator,
+			program,
+			decodeBufferFn,
+			options,
+			resubOpts,
+			onStreamFailure
+		);
+	}
+
+	override async subscribe(
+		onChange: (
+			accountId: PublicKey,
+			data: T,
+			context: Context,
+			buffer: Buffer
+		) => void
+	): Promise<void> {
+		if (this.listenerId != null || this.isUnsubscribing) {
+			return;
+		}
+
+		this.onChange = onChange;
+		this.streamFailureHandled = false;
+		this.stream =
+			(await this.client.subscribe()) as unknown as typeof this.stream;
+
+		const filters = this.options.filters.map((filter) => {
+			return {
+				memcmp: {
+					offset: filter.memcmp.offset.toString(),
+					bytes: bs58.decode(filter.memcmp.bytes),
+				},
+			};
+		});
+
+		const request: SubscribeRequest = {
+			slots: {},
+			accounts: {
+				drift: {
+					account: [],
+					owner: [this.program.programId.toBase58()],
+					filters,
+				},
+			},
+			transactions: {},
+			blocks: {},
+			blocksMeta: {},
+			accountsDataSlice: [],
+			commitment: this.commitmentLevel,
+			entry: {},
+			transactionsStatus: {},
+		};
+
+		this.dataHandler = (chunk: SubscribeUpdate) => {
+			if (!chunk.account) {
+				return;
+			}
+
+			const slot = Number(chunk.account.slot);
+			const accountInfo = {
+				owner: new PublicKey(chunk.account.account.owner),
+				lamports: Number(chunk.account.account.lamports),
+				data: Buffer.from(chunk.account.account.data),
+				executable: chunk.account.account.executable,
+				rentEpoch: Number(chunk.account.account.rentEpoch),
+			};
+
+			if (this.resubOpts?.resubTimeoutMs) {
+				this.receivingData = true;
+				clearTimeout(this.timeoutId);
+				this.handleRpcResponse(
+					{ slot },
+					{
+						accountId: new PublicKey(chunk.account.account.pubkey),
+						accountInfo,
+					}
+				);
+				this.setTimeout();
+			} else {
+				this.handleRpcResponse(
+					{ slot },
+					{
+						accountId: new PublicKey(chunk.account.account.pubkey),
+						accountInfo,
+					}
+				);
+			}
+		};
+		this.errorHandler = (error: Error) => {
+			this.handleStreamFailure('error', error);
+		};
+		this.endHandler = () => {
+			this.handleStreamFailure('end');
+		};
+		this.closeHandler = () => {
+			this.handleStreamFailure('close');
+		};
+
+		this.stream.on('data', this.dataHandler);
+		this.stream.on('error', this.errorHandler);
+		this.stream.on('end', this.endHandler);
+		this.stream.on('close', this.closeHandler);
+
+		try {
+			await new Promise<void>((resolve, reject) => {
+				this.stream!.write(request, (err) => {
+					if (err === null || err === undefined) {
+						this.listenerId = 1;
+						if (this.resubOpts?.resubTimeoutMs) {
+							this.receivingData = true;
+							this.setTimeout();
+						}
+						resolve();
+					} else {
+						reject(err);
+					}
+				});
+			});
+		} catch (reason) {
+			this.cleanupStream();
+			console.error(reason);
+			throw reason;
+		}
+	}
+
+	override async unsubscribe(onResub = false): Promise<void> {
+		if (!onResub && this.resubOpts) {
+			this.resubOpts.resubTimeoutMs = undefined;
+		}
+		this.isUnsubscribing = true;
+		clearTimeout(this.timeoutId);
+		this.timeoutId = undefined;
+
+		try {
+			if (this.listenerId != null && this.stream) {
+				const request: SubscribeRequest = {
+					slots: {},
+					accounts: {},
+					transactions: {},
+					blocks: {},
+					blocksMeta: {},
+					accountsDataSlice: [],
+					entry: {},
+					transactionsStatus: {},
+				};
+
+				await new Promise<void>((resolve, reject) => {
+					this.stream!.write(request, (err) => {
+						if (err === null || err === undefined) {
+							resolve();
+						} else {
+							reject(err);
+						}
+					});
+				});
+			}
+		} catch (reason) {
+			console.error(reason);
+		} finally {
+			this.listenerId = undefined;
+			this.receivingData = false;
+			this.cleanupStream();
+			this.isUnsubscribing = false;
+		}
+	}
+
+	private handleStreamFailure(
+		event: StreamFailureEvent,
+		error?: unknown
+	): void {
+		if (this.isUnsubscribing || this.streamFailureHandled) {
+			return;
+		}
+
+		this.streamFailureHandled = true;
+		this.listenerId = undefined;
+		this.receivingData = false;
+		clearTimeout(this.timeoutId);
+		this.timeoutId = undefined;
+		this.cleanupStream();
+		this.onStreamFailure?.(event, error);
+	}
+
+	private cleanupStream(): void {
+		const stream = this.stream;
+		this.stream = undefined;
+
+		if (!stream) {
+			return;
+		}
+
+		if (this.dataHandler) {
+			stream.removeListener('data', this.dataHandler);
+			this.dataHandler = undefined;
+		}
+		if (this.errorHandler) {
+			stream.removeListener('error', this.errorHandler);
+			this.errorHandler = undefined;
+		}
+		if (this.endHandler) {
+			stream.removeListener('end', this.endHandler);
+			this.endHandler = undefined;
+		}
+		if (this.closeHandler) {
+			stream.removeListener('close', this.closeHandler);
+			this.closeHandler = undefined;
+		}
+
+		stream.once('error', (error) => {
+			if (isExpectedGrpcTeardownError(error)) {
+				logger.debug(
+					`Ignoring expected gRPC stream error during teardown: ${formatError(
+						error
+					)}`
+				);
+				return;
+			}
+
+			logger.warn(
+				`Ignoring unexpected gRPC stream error during teardown: ${formatError(
+					error
+				)}`
+			);
+		});
+
+		try {
+			if (typeof (stream as any).cancel === 'function') {
+				(stream as any).cancel();
+			}
+		} catch {
+			// Ignore stream teardown errors while recovering.
+		}
+	}
+}
+
+export class ReconnectingGrpcOrderSubscription {
+	private orderSubscriber: OrderSubscriber;
+	private skipInitialLoad: boolean;
+	private resubOpts?: ResubOpts;
+	private resyncIntervalMs?: number;
+	private subscriber?:
+		| ReconnectingGrpcProgramAccountSubscriber<UserAccount>
+		| LaserstreamProgramAccountSubscriber<UserAccount>;
+	private resyncTimeoutId?: ReturnType<typeof setTimeout>;
+	private decoded?: boolean;
+	private grpcConfigs: GrpcConfigs;
+	private reconnectPromise?: Promise<void>;
+	private stopped = false;
+	private reconnectDelayMs = getReconnectDelayMs();
+
+	constructor({
+		grpcConfigs,
+		orderSubscriber,
+		skipInitialLoad = false,
+		resubOpts,
+		resyncIntervalMs,
+		decoded = true,
+	}: {
+		grpcConfigs: GrpcConfigs;
+		orderSubscriber: OrderSubscriber;
+		skipInitialLoad?: boolean;
+		resubOpts?: ResubOpts;
+		resyncIntervalMs?: number;
+		decoded?: boolean;
+	}) {
+		this.orderSubscriber = orderSubscriber;
+		this.skipInitialLoad = skipInitialLoad;
+		this.resubOpts = resubOpts;
+		this.resyncIntervalMs = resyncIntervalMs;
+		this.decoded = decoded;
+		this.grpcConfigs = grpcConfigs;
+	}
+
+	public async subscribe(): Promise<void> {
+		return this.subscribeInternal(true);
+	}
+
+	private async subscribeInternal(resetStopped: boolean): Promise<void> {
+		if (resetStopped) {
+			this.stopped = false;
+		}
+		if (this.subscriber) {
+			return;
+		}
+
+		const nextSubscriber = await this.createSubscriber();
+		this.subscriber = nextSubscriber;
+
+		try {
+			await nextSubscriber.subscribe(
+				(
+					accountId: PublicKey,
+					account: UserAccount,
+					context: Context,
+					buffer: Buffer
+				) => {
+					const userKey = accountId.toBase58();
+					if (this.decoded ?? true) {
+						this.orderSubscriber.tryUpdateUserAccount(
+							userKey,
+							'decoded',
+							account,
+							context.slot
+						);
+					} else {
+						this.orderSubscriber.tryUpdateUserAccount(
+							userKey,
+							'buffer',
+							buffer,
+							context.slot
+						);
+					}
+				}
+			);
+
+			if (!this.skipInitialLoad) {
+				await this.orderSubscriber.fetch();
+			}
+
+			this.startResyncLoop();
+		} catch (error) {
+			this.subscriber = undefined;
+			try {
+				await nextSubscriber.unsubscribe(true);
+			} catch {
+				// Best effort cleanup before retry.
+			}
+			throw error;
+		}
+	}
+
+	public async unsubscribe(): Promise<void> {
+		this.stopped = true;
+		this.clearResyncLoop();
+
+		const currentSubscriber = this.subscriber;
+		this.subscriber = undefined;
+		if (!currentSubscriber) {
+			return;
+		}
+
+		try {
+			await currentSubscriber.unsubscribe();
+		} catch (error) {
+			logger.warn(
+				`OrderSubscriber gRPC unsubscribe failed during shutdown: ${formatError(
+					error
+				)}`
+			);
+		}
+	}
+
+	private async createSubscriber() {
+		if (this.grpcConfigs.client === 'laser') {
+			return LaserstreamProgramAccountSubscriber.create<UserAccount>(
+				this.grpcConfigs,
+				'OrderSubscriber',
+				'User',
+				this.orderSubscriber.driftClient.program,
+				this.orderSubscriber.decodeFn,
+				{
+					filters: [getUserFilter(), getNonIdleUserFilter()],
+				},
+				this.resubOpts
+			);
+		}
+
+		return ReconnectingGrpcProgramAccountSubscriber.create<UserAccount>(
+			this.grpcConfigs,
+			'OrderSubscriber',
+			'User',
+			this.orderSubscriber.driftClient.program,
+			this.orderSubscriber.decodeFn,
+			{
+				filters: [getUserFilter(), getNonIdleUserFilter()],
+			},
+			this.resubOpts,
+			(event, error) => {
+				void this.handleStreamFailure(event, error);
+			}
+		);
+	}
+
+	private async handleStreamFailure(
+		event: StreamFailureEvent,
+		error?: unknown
+	): Promise<void> {
+		if (this.stopped) {
+			return;
+		}
+		if (this.reconnectPromise) {
+			return this.reconnectPromise;
+		}
+
+		const errorSuffix = error
+			? `: ${formatError(error)}`
+			: '';
+		logger.warn(
+			`OrderSubscriber gRPC stream ${event}${errorSuffix}; reconnecting`
+		);
+
+		this.reconnectPromise = this.reconnectLoop().finally(() => {
+			this.reconnectPromise = undefined;
+		});
+		return this.reconnectPromise;
+	}
+
+	private async reconnectLoop(): Promise<void> {
+		const currentSubscriber = this.subscriber;
+		this.subscriber = undefined;
+		this.clearResyncLoop();
+
+		if (currentSubscriber) {
+			try {
+				await currentSubscriber.unsubscribe(true);
+			} catch (error) {
+				logger.warn(
+					`OrderSubscriber gRPC cleanup failed before reconnect: ${formatError(
+						error
+					)}`
+				);
+			}
+		}
+
+		while (!this.stopped) {
+				if (this.reconnectDelayMs > 0) {
+					await sleep(this.reconnectDelayMs);
+				}
+				if (this.stopped) {
+					return;
+				}
+
+				try {
+					await this.subscribeInternal(false);
+					logger.info('OrderSubscriber gRPC reconnect succeeded');
+					return;
+				} catch (error) {
+				logger.error(
+					`OrderSubscriber gRPC reconnect failed: ${formatError(error)}`
+				);
+			}
+		}
+	}
+
+	private startResyncLoop(): void {
+		if (!this.resyncIntervalMs) {
+			return;
+		}
+
+		this.clearResyncLoop();
+
+		const recursiveResync = () => {
+			this.resyncTimeoutId = setTimeout(() => {
+				if (this.stopped) {
+					return;
+				}
+
+				this.orderSubscriber
+					.fetch()
+					.catch((error) => {
+						logger.error(
+							`Failed to resync OrderSubscriber: ${formatError(error)}`
+						);
+					})
+					.finally(() => {
+						if (!this.resyncTimeoutId || this.stopped) {
+							return;
+						}
+						recursiveResync();
+					});
+			}, this.resyncIntervalMs);
+		};
+
+		recursiveResync();
+	}
+
+	private clearResyncLoop(): void {
+		if (this.resyncTimeoutId !== undefined) {
+			clearTimeout(this.resyncTimeoutId);
+			this.resyncTimeoutId = undefined;
+		}
+	}
+}

src/index.ts

@@ -24,8 +24,6 @@ import {
 	AssetType,
 	MarketType,
 } from '@drift-labs/sdk';
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
-
 import { logger, setLogLevel } from './utils/logger';
 
 import * as http from 'http';
@@ -53,6 +51,13 @@ import { HermesClient } from '@pythnetwork/hermes-client';
 import { COMMON_UI_UTILS, ENUM_UTILS } from '@drift/common';
 import { AuctionParamArgs } from './utils/types';
 import { TakerFillVsOracleBpsRedisResult } from './athena/repositories/fillQualityAnalytics';
+import {
+	RedisClient,
+	RedisClientPrefix,
+	HOT_REDIS_KEY_PREFIX,
+	ALL_REDIS_KEY_PREFIX,
+	parseRedisClients,
+} from './utils/redisClients';
 
 setGlobalDispatcher(
 	new Agent({
@@ -62,16 +67,7 @@ setGlobalDispatcher(
 
 require('dotenv').config();
 
-const envClients = [];
+const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT);
-const clients = process.env.REDIS_CLIENT?.trim()
-	.replace(/^\[|\]$/g, '')
-	.split(/\s*,\s*/);
-
-clients?.forEach((client) => envClients.push(RedisClientPrefix[client]));
-
-const REDIS_CLIENTS = envClients.length
-	? envClients
-	: [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
 console.log('Redis Clients:', REDIS_CLIENTS);
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -286,7 +282,7 @@ const main = async (): Promise<void> => {
 				.find((client) => {
 					return (
 						client.forceGetClient().options.keyPrefix ===
-						RedisClientPrefix.DLOB_HELIUS
+						ALL_REDIS_KEY_PREFIX
 					);
 				})
 				.getRaw(`priorityFees_${marketType}_${marketIndex}`);
@@ -332,7 +328,7 @@ const main = async (): Promise<void> => {
 						.find(
 							(client) =>
 								client.forceGetClient().options.keyPrefix ===
-								RedisClientPrefix.DLOB_HELIUS
+								ALL_REDIS_KEY_PREFIX
 						)
 						.getRaw(
 							`priorityFees_${normedParam['marketType']}_${normedParam['marketIndex']}`
@@ -523,7 +519,7 @@ const main = async (): Promise<void> => {
 
 			const redisClient = redisClients.find(
 				(client) =>
-					client.forceGetClient().options.keyPrefix === RedisClientPrefix.DLOB
+					client.forceGetClient().options.keyPrefix === HOT_REDIS_KEY_PREFIX
 			);
 
 			const redisResponseGainers = await redisClient.getRaw(

src/publishers/dlobPublisher.ts

@@ -19,8 +19,6 @@ import {
 	ONE,
 	WebSocketAccountSubscriberV2,
 } from '@drift-labs/sdk';
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
-
 import { logger, setLogLevel } from '../utils/logger';
 import {
 	SubscriberLookup,
@@ -47,10 +45,28 @@ import {
 	FillQualityAnalyticsRepository,
 	TakerFillVsOracleBpsRedisResult,
 } from '../athena/repositories/fillQualityAnalytics';
+import {
+	RedisClient,
+	HOT_REDIS_CLIENT,
+	getRedisClientPrefix,
+} from '../utils/redisClients';
+
+function envInt(name: string, fallback: number): number {
+	const raw = process.env[name];
+	if (raw == null) return fallback;
+	const parsed = parseInt(raw, 10);
+	return Number.isFinite(parsed) ? parsed : fallback;
+}
 
 setGlobalDispatcher(
 	new Agent({
 		connections: 200,
+		// Solana RPC calls like getProgramAccounts can legitimately take >30s during
+		// startup (esp. on self-hosted nodes). Default undici timeouts are too tight
+		// and cause startup loops (UND_ERR_HEADERS_TIMEOUT).
+		connectTimeout: envInt('FETCH_CONNECT_TIMEOUT_MS', 10_000),
+		headersTimeout: envInt('FETCH_HEADERS_TIMEOUT_MS', 180_000),
+		bodyTimeout: envInt('FETCH_BODY_TIMEOUT_MS', 180_000),
 	})
 );
 
@@ -62,7 +78,8 @@ const metricsPort = process.env.METRICS_PORT
 	? parseInt(process.env.METRICS_PORT)
 	: 9464;
 
-const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB';
+const REDIS_CLIENT = process.env.REDIS_CLIENT || HOT_REDIS_CLIENT;
+const REDIS_PREFIX = getRedisClientPrefix(REDIS_CLIENT);
 
 // Set up express for health checks
 const app = express();
@@ -171,7 +188,11 @@ const ignoreList = process.env.IGNORE_LIST?.split(',') || [
 logger.info(`RPC endpoint:  ${endpoint}`);
 logger.info(`WS endpoint:   ${wsEndpoint}`);
 logger.info(`GRPC endpoint: ${grpcEndpoint}`);
-logger.info(`GRPC Token:    ${token}`);
+logger.info(
+	`GRPC Token:    ${
+		token ? `${token.slice(0, 4)}…${token.slice(-4)}` : '(missing)'
+	}`
+);
 logger.info(
 	`useOrderSubscriber: ${useOrderSubscriber}, useWebsocket: ${useWebsocket}, useGrpc: ${useGrpc}`
 );
@@ -380,7 +401,7 @@ const main = async () => {
 	const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
 
 	const redisClient = new RedisClient({
-		prefix: RedisClientPrefix[REDIS_CLIENT],
+		prefix: REDIS_PREFIX,
 	});
 	await redisClient.connect();
 
@@ -473,6 +494,9 @@ const main = async () => {
 		perpMarketIndexes: perpMarketInfos.map((m) => m.marketIndex),
 		spotMarketIndexes: spotMarketInfos.map((m) => m.marketIndex),
 		oracleInfos,
+		// Publisher doesn't need to load/subscribe to any user accounts.
+		// Avoids expensive RPC lookups that can time out on self-hosted nodes.
+		skipLoadUsers: true,
 	});
 
 	const lamportsBalance = await connection.getBalance(wallet.publicKey);
@@ -638,21 +662,21 @@ const main = async () => {
 				marketType: 'perp',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				redisPrefix: REDIS_PREFIX,
 			});
 			oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
 				marketIndex: market.marketIndex,
 				marketType: 'perp',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				redisPrefix: REDIS_PREFIX,
 			});
 			marketSlotGauge.setLatestValue(marketAccount.slot, {
 				marketIndex: market.marketIndex,
 				marketType: 'perp',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				redisPrefix: REDIS_PREFIX,
 			});
 		});
 		spotMarketInfos.forEach((market) => {
@@ -668,21 +692,21 @@ const main = async () => {
 				marketType: 'spot',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				redisPrefix: REDIS_PREFIX,
 			});
 			oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
 				marketIndex: market.marketIndex,
 				marketType: 'spot',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				redisPrefix: REDIS_PREFIX,
 			});
 			marketSlotGauge.setLatestValue(marketAccount.slot, {
 				marketIndex: market.marketIndex,
 				marketType: 'spot',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
+				redisPrefix: REDIS_PREFIX,
 			});
 		});
 	}, 10_000);

src/publishers/priorityFeesPublisher.ts

@@ -10,12 +10,11 @@ import {
 	BulkAccountLoader,
 	getMarketsAndOraclesForSubscription,
 } from '@drift-labs/sdk';
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
-
 import { logger, setLogLevel } from '../utils/logger';
 import { sleep } from '../utils/utils';
 import express from 'express';
 import { setGlobalDispatcher, Agent } from 'undici';
+import { RedisClient, ALL_REDIS_KEY_PREFIX } from '../utils/redisClients';
 
 setGlobalDispatcher(
 	new Agent({
@@ -27,7 +26,7 @@ require('dotenv').config();
 const stateCommitment: Commitment = 'confirmed';
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
-const redisClientPrefix = RedisClientPrefix.DLOB_HELIUS;
+const redisClientPrefix = ALL_REDIS_KEY_PREFIX;
 // Set up express for health checks
 const app = express();
 

src/publishers/tradesPublisher.ts

@@ -20,12 +20,15 @@ import {
 	OrderActionRecord,
 	Event,
 } from '@drift-labs/sdk';
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
-
 import { logger, setLogLevel } from '../utils/logger';
 import { sleep } from '../utils/utils';
 import { fromEvent, filter, map } from 'rxjs';
 import { setGlobalDispatcher, Agent } from 'undici';
+import {
+	RedisClient,
+	HOT_REDIS_CLIENT,
+	getRedisClientPrefix,
+} from '../utils/redisClients';
 
 setGlobalDispatcher(
 	new Agent({
@@ -36,9 +39,9 @@ setGlobalDispatcher(
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
-const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB';
+const REDIS_CLIENT = process.env.REDIS_CLIENT || HOT_REDIS_CLIENT;
 console.log('Redis Clients:', REDIS_CLIENT);
-const redisClientPrefix = RedisClientPrefix[REDIS_CLIENT];
+const redisClientPrefix = getRedisClientPrefix(REDIS_CLIENT);
 //@ts-ignore
 const sdkConfig = initialize({ env: process.env.ENV });
 

src/scripts/dlobAllRedisToPostgresDerivedWriter.ts

@@ -0,0 +1,284 @@
+import Redis from 'ioredis';
+import { Pool } from 'pg';
+import process from 'node:process';
+
+import {
+	deriveNormalizedRow,
+	deriveRawSnapshotRow,
+	envInt,
+	sleep,
+} from './dlobHotPgCommon';
+
+const cfg = {
+	source: String(process.env.DLOB_SOURCE || 'mevnode_bot_all_derived'),
+	redisHost: String(process.env.REDIS_HOST || process.env.ELASTICACHE_HOST || 'dlob-redis'),
+	redisPort: envInt('REDIS_PORT', envInt('ELASTICACHE_PORT', 6379)),
+	redisKeyPrefix: String(process.env.REDIS_KEY_PREFIX || 'dlob-all:'),
+	pollMs: envInt('DLOB_POLL_MS', 1000),
+	depthLevels: envInt('NORMALIZED_DEPTH', 10),
+	pricePrecision: envInt('PRICE_PRECISION', 1_000_000),
+	basePrecision: envInt('BASE_PRECISION', 1_000_000_000),
+	pgHost: String(process.env.PGHOST || 'postgres'),
+	pgPort: envInt('PGPORT', 5432),
+	pgUser: String(process.env.PGUSER || 'admin'),
+	pgPassword: String(process.env.PGPASSWORD || ''),
+	pgDatabase: String(process.env.PGDATABASE || 'crypto'),
+};
+
+const redis = new Redis({
+	host: cfg.redisHost,
+	port: cfg.redisPort,
+	maxRetriesPerRequest: null,
+	enableReadyCheck: false,
+});
+
+const pool = new Pool({
+	host: cfg.pgHost,
+	port: cfg.pgPort,
+	user: cfg.pgUser,
+	password: cfg.pgPassword,
+	database: cfg.pgDatabase,
+	max: 4,
+});
+
+const seen = new Map<string, string>();
+
+const insertDerivedTsSql = `
+INSERT INTO public.dlob_all_derived_ts (
+	event_ts,
+	source,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	mark_price,
+	oracle_price,
+	best_bid_price,
+	best_ask_price,
+	mid_price,
+	spread_quote,
+	spread_bps,
+	depth_levels,
+	bid_levels,
+	ask_levels,
+	top_bid_size,
+	top_ask_size,
+	top_bid_notional,
+	top_ask_notional,
+	depth_bid_base,
+	depth_ask_base,
+	depth_bid_quote,
+	depth_ask_quote,
+	imbalance,
+	bids_norm,
+	asks_norm,
+	raw_payload_hash
+) VALUES (
+	to_timestamp($1::double precision / 1000.0),
+	$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31
+)
+ON CONFLICT DO NOTHING
+`;
+
+const upsertDerivedLatestSql = `
+INSERT INTO public.dlob_all_derived_latest (
+	source,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	mark_price,
+	oracle_price,
+	best_bid_price,
+	best_ask_price,
+	mid_price,
+	spread_quote,
+	spread_bps,
+	depth_levels,
+	bid_levels,
+	ask_levels,
+	top_bid_size,
+	top_ask_size,
+	top_bid_notional,
+	top_ask_notional,
+	depth_bid_base,
+	depth_ask_base,
+	depth_bid_quote,
+	depth_ask_quote,
+	imbalance,
+	bids_norm,
+	asks_norm,
+	raw_payload_hash,
+	updated_at
+) VALUES (
+	$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,now()
+)
+ON CONFLICT (source, market_type, market_index, is_indicative)
+DO UPDATE SET
+	market_name = EXCLUDED.market_name,
+	ts_ms = EXCLUDED.ts_ms,
+	slot = EXCLUDED.slot,
+	market_slot = EXCLUDED.market_slot,
+	mark_price = EXCLUDED.mark_price,
+	oracle_price = EXCLUDED.oracle_price,
+	best_bid_price = EXCLUDED.best_bid_price,
+	best_ask_price = EXCLUDED.best_ask_price,
+	mid_price = EXCLUDED.mid_price,
+	spread_quote = EXCLUDED.spread_quote,
+	spread_bps = EXCLUDED.spread_bps,
+	depth_levels = EXCLUDED.depth_levels,
+	bid_levels = EXCLUDED.bid_levels,
+	ask_levels = EXCLUDED.ask_levels,
+	top_bid_size = EXCLUDED.top_bid_size,
+	top_ask_size = EXCLUDED.top_ask_size,
+	top_bid_notional = EXCLUDED.top_bid_notional,
+	top_ask_notional = EXCLUDED.top_ask_notional,
+	depth_bid_base = EXCLUDED.depth_bid_base,
+	depth_ask_base = EXCLUDED.depth_ask_base,
+	depth_bid_quote = EXCLUDED.depth_bid_quote,
+	depth_ask_quote = EXCLUDED.depth_ask_quote,
+	imbalance = EXCLUDED.imbalance,
+	bids_norm = EXCLUDED.bids_norm,
+	asks_norm = EXCLUDED.asks_norm,
+	raw_payload_hash = EXCLUDED.raw_payload_hash,
+	updated_at = EXCLUDED.updated_at
+`;
+
+async function syncOnce(): Promise<void> {
+	const keys = (await redis.keys(`${cfg.redisKeyPrefix}*`)).sort();
+	if (keys.length === 0) return;
+
+	const client = await pool.connect();
+	let writes = 0;
+
+	try {
+		await client.query('BEGIN');
+		for (const key of keys) {
+			const payloadText = await redis.get(key);
+			if (!payloadText) continue;
+
+			const raw = deriveRawSnapshotRow({
+				source: cfg.source,
+				redisKey: key,
+				payloadText,
+			});
+			if (raw == null) continue;
+
+			const normalized = deriveNormalizedRow(raw, {
+				source: cfg.source,
+				pricePrecision: cfg.pricePrecision,
+				basePrecision: cfg.basePrecision,
+				depthLevels: cfg.depthLevels,
+			});
+			if (normalized == null) continue;
+
+			const token = `${normalized.tsMs}:${normalized.rawPayloadHash}`;
+			if (seen.get(key) === token) continue;
+
+			await client.query(insertDerivedTsSql, [
+				normalized.tsMs,
+				normalized.source,
+				normalized.marketType,
+				normalized.marketIndex,
+				normalized.marketName,
+				normalized.isIndicative,
+				normalized.tsMs,
+				normalized.slot,
+				normalized.marketSlot,
+				normalized.markPrice,
+				normalized.oraclePrice,
+				normalized.bestBidPrice,
+				normalized.bestAskPrice,
+				normalized.midPrice,
+				normalized.spreadQuote,
+				normalized.spreadBps,
+				normalized.depthLevels,
+				normalized.bidLevels,
+				normalized.askLevels,
+				normalized.topBidSize,
+				normalized.topAskSize,
+				normalized.topBidNotional,
+				normalized.topAskNotional,
+				normalized.depthBidBase,
+				normalized.depthAskBase,
+				normalized.depthBidQuote,
+				normalized.depthAskQuote,
+				normalized.imbalance,
+				JSON.stringify(normalized.bidsNorm),
+				JSON.stringify(normalized.asksNorm),
+				normalized.rawPayloadHash,
+			]);
+
+			await client.query(upsertDerivedLatestSql, [
+				normalized.source,
+				normalized.marketType,
+				normalized.marketIndex,
+				normalized.marketName,
+				normalized.isIndicative,
+				normalized.tsMs,
+				normalized.slot,
+				normalized.marketSlot,
+				normalized.markPrice,
+				normalized.oraclePrice,
+				normalized.bestBidPrice,
+				normalized.bestAskPrice,
+				normalized.midPrice,
+				normalized.spreadQuote,
+				normalized.spreadBps,
+				normalized.depthLevels,
+				normalized.bidLevels,
+				normalized.askLevels,
+				normalized.topBidSize,
+				normalized.topAskSize,
+				normalized.topBidNotional,
+				normalized.topAskNotional,
+				normalized.depthBidBase,
+				normalized.depthAskBase,
+				normalized.depthBidQuote,
+				normalized.depthAskQuote,
+				normalized.imbalance,
+				JSON.stringify(normalized.bidsNorm),
+				JSON.stringify(normalized.asksNorm),
+				normalized.rawPayloadHash,
+			]);
+
+			seen.set(key, token);
+			writes += 1;
+		}
+		await client.query('COMMIT');
+		if (writes > 0) {
+			console.log(
+				`[dlob-all-redis-to-postgres-derived-writer] wrote ${writes} normalized snapshot(s) from prefix ${cfg.redisKeyPrefix}`
+			);
+		}
+	} catch (error) {
+		await client.query('ROLLBACK');
+		console.error('[dlob-all-redis-to-postgres-derived-writer] sync failed', error);
+	} finally {
+		client.release();
+	}
+}
+
+async function main(): Promise<void> {
+	console.log(
+		`[dlob-all-redis-to-postgres-derived-writer] source=${cfg.source} redis=${cfg.redisHost}:${cfg.redisPort} prefix=${cfg.redisKeyPrefix} pg=${cfg.pgHost}:${cfg.pgPort}/${cfg.pgDatabase}`
+	);
+	await redis.ping();
+	await pool.query('select 1');
+
+	for (;;) {
+		await syncOnce();
+		await sleep(cfg.pollMs);
+	}
+}
+
+main().catch((error) => {
+	console.error('[dlob-all-redis-to-postgres-derived-writer] fatal', error);
+	process.exit(1);
+});

src/scripts/dlobHotPgCommon.ts

@@ -0,0 +1,321 @@
+import crypto from 'node:crypto';
+
+export type SnapshotKind = 'orderbook_l2' | 'orderbook_l3' | 'best_makers';
+
+export type RawSnapshotRow = {
+	source: string;
+	redisKey: string;
+	snapshotKind: SnapshotKind;
+	marketType: string;
+	marketIndex: number | null;
+	marketName: string;
+	isIndicative: boolean;
+	tsMs: number;
+	slot: number | null;
+	marketSlot: number | null;
+	payloadHash: string;
+	markPriceRaw: string | null;
+	oraclePriceRaw: string | null;
+	bestBidPriceRaw: string | null;
+	bestAskPriceRaw: string | null;
+	spreadPctRaw: string | null;
+	spreadQuoteRaw: string | null;
+	oracleData: unknown | null;
+	bids: unknown[] | null;
+	asks: unknown[] | null;
+	payload: Record<string, unknown>;
+};
+
+export type DerivedSnapshotRow = {
+	source: string;
+	marketType: string;
+	marketIndex: number | null;
+	marketName: string;
+	isIndicative: boolean;
+	tsMs: number;
+	slot: number | null;
+	marketSlot: number | null;
+	markPrice: number | null;
+	oraclePrice: number | null;
+	bestBidPrice: number | null;
+	bestAskPrice: number | null;
+	midPrice: number | null;
+	spreadQuote: number | null;
+	spreadBps: number | null;
+	depthLevels: number;
+	bidLevels: number;
+	askLevels: number;
+	topBidSize: number | null;
+	topAskSize: number | null;
+	topBidNotional: number | null;
+	topAskNotional: number | null;
+	depthBidBase: number;
+	depthAskBase: number;
+	depthBidQuote: number;
+	depthAskQuote: number;
+	imbalance: number | null;
+	bidsNorm: Array<Record<string, unknown>>;
+	asksNorm: Array<Record<string, unknown>>;
+	rawPayloadHash: string;
+};
+
+export function envInt(name: string, fallback: number): number {
+	const raw = process.env[name];
+	if (raw == null || raw === '') return fallback;
+	const parsed = Number.parseInt(raw, 10);
+	return Number.isFinite(parsed) ? parsed : fallback;
+}
+
+export function sleep(ms: number): Promise<void> {
+	return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+export function toOptionalInt(value: unknown): number | null {
+	if (value == null) return null;
+	const parsed = Number.parseInt(String(value), 10);
+	return Number.isFinite(parsed) ? parsed : null;
+}
+
+export function toOptionalString(value: unknown): string | null {
+	if (value == null) return null;
+	const text = String(value).trim();
+	return text.length > 0 ? text : null;
+}
+
+export function toScaledNumber(value: unknown, scale: number): number | null {
+	const text = toOptionalString(value);
+	if (text == null) return null;
+	const parsed = Number(text);
+	if (!Number.isFinite(parsed)) return null;
+	return parsed / scale;
+}
+
+export function parseRedisKey(redisKey: string): {
+	snapshotKind: SnapshotKind;
+	marketType: string;
+	marketIndex: number | null;
+	isIndicative: boolean;
+} | null {
+	const key = redisKey.includes(':') ? redisKey.slice(redisKey.indexOf(':') + 1) : redisKey;
+	const isIndicative = key.endsWith('_indicative');
+	const baseKey = isIndicative ? key.slice(0, -'_indicative'.length) : key;
+
+	let snapshotKind: SnapshotKind;
+	if (baseKey.includes('best_makers')) {
+		snapshotKind = 'best_makers';
+	} else if (baseKey.includes('orderbook_l3_')) {
+		snapshotKind = 'orderbook_l3';
+	} else if (baseKey.includes('orderbook_')) {
+		snapshotKind = 'orderbook_l2';
+	} else {
+		return null;
+	}
+
+	const match = baseKey.match(/_(perp|spot)_(\d+)$/);
+	if (!match) return null;
+
+	return {
+		snapshotKind,
+		marketType: match[1],
+		marketIndex: Number.parseInt(match[2], 10),
+		isIndicative,
+	};
+}
+
+export function hashPayload(payloadText: string): string {
+	return crypto.createHash('sha1').update(payloadText).digest('hex');
+}
+
+export function pickTsMs(payload: Record<string, unknown>): number {
+	const direct = toOptionalInt(payload.ts);
+	if (direct != null) return direct;
+	const updated = toOptionalInt(payload.updatedAtTs);
+	if (updated != null) return updated;
+	return Date.now();
+}
+
+export function deriveRawSnapshotRow(input: {
+	source: string;
+	redisKey: string;
+	payloadText: string;
+}): RawSnapshotRow | null {
+	const meta = parseRedisKey(input.redisKey);
+	if (meta == null) return null;
+
+	let payload: Record<string, unknown>;
+	try {
+		payload = JSON.parse(input.payloadText);
+	} catch {
+		return null;
+	}
+
+	const bids = Array.isArray(payload.bids) ? payload.bids : [];
+	const asks = Array.isArray(payload.asks) ? payload.asks : [];
+	const marketName =
+		toOptionalString(payload.marketName) ??
+		`${meta.marketType.toUpperCase()}-${meta.marketIndex ?? 'unknown'}`;
+
+	return {
+		source: input.source,
+		redisKey: input.redisKey,
+		snapshotKind: meta.snapshotKind,
+		marketType: toOptionalString(payload.marketType) ?? meta.marketType,
+		marketIndex:
+			typeof payload.marketIndex === 'number'
+				? payload.marketIndex
+				: meta.marketIndex,
+		marketName,
+		isIndicative: meta.isIndicative,
+		tsMs: pickTsMs(payload),
+		slot: toOptionalInt(payload.slot),
+		marketSlot: toOptionalInt(payload.marketSlot),
+		payloadHash: hashPayload(input.payloadText),
+		markPriceRaw: toOptionalString(payload.markPrice),
+		oraclePriceRaw: toOptionalString(
+			(payload.oracleData as Record<string, unknown> | undefined)?.price ?? payload.oracle
+		),
+		bestBidPriceRaw: toOptionalString(payload.bestBidPrice),
+		bestAskPriceRaw: toOptionalString(payload.bestAskPrice),
+		spreadPctRaw: toOptionalString(payload.spreadPct),
+		spreadQuoteRaw: toOptionalString(payload.spreadQuote),
+		oracleData:
+			payload.oracleData && typeof payload.oracleData === 'object'
+				? payload.oracleData
+				: null,
+		bids,
+		asks,
+		payload,
+	};
+}
+
+function normalizeLevels(
+	levels: unknown[],
+	limit: number,
+	pricePrecision: number,
+	basePrecision: number
+): Array<Record<string, unknown>> {
+	return levels
+		.slice(0, limit)
+		.map((level) => {
+			const item = level as Record<string, unknown>;
+			const price = toScaledNumber(item.price, pricePrecision);
+			const size = toScaledNumber(item.size, basePrecision);
+			if (price == null || size == null) return null;
+			return {
+				price,
+				sizeBase: size,
+				notional: price * size,
+				sources:
+					item.sources && typeof item.sources === 'object' ? item.sources : null,
+			};
+		})
+		.filter((item): item is Record<string, unknown> => item != null);
+}
+
+export function deriveNormalizedRow(
+	raw: RawSnapshotRow,
+	config: {
+		source: string;
+		pricePrecision: number;
+		basePrecision: number;
+		depthLevels: number;
+	}
+): DerivedSnapshotRow | null {
+	if (raw.snapshotKind !== 'orderbook_l2') return null;
+
+	const bidsNorm = normalizeLevels(
+		raw.bids ?? [],
+		config.depthLevels,
+		config.pricePrecision,
+		config.basePrecision
+	);
+	const asksNorm = normalizeLevels(
+		raw.asks ?? [],
+		config.depthLevels,
+		config.pricePrecision,
+		config.basePrecision
+	);
+
+	const markPrice = toScaledNumber(raw.markPriceRaw, config.pricePrecision);
+	const oraclePrice = toScaledNumber(raw.oraclePriceRaw, config.pricePrecision);
+	const bestBidPrice = toScaledNumber(raw.bestBidPriceRaw, config.pricePrecision);
+	const bestAskPrice = toScaledNumber(raw.bestAskPriceRaw, config.pricePrecision);
+
+	const midPrice =
+		bestBidPrice != null && bestAskPrice != null
+			? (bestBidPrice + bestAskPrice) / 2
+			: null;
+	const spreadQuote =
+		bestBidPrice != null && bestAskPrice != null
+			? bestAskPrice - bestBidPrice
+			: null;
+	const spreadBps =
+		spreadQuote != null && midPrice != null && midPrice > 0
+			? (spreadQuote / midPrice) * 10_000
+			: null;
+
+	const bidLevels = bidsNorm.length;
+	const askLevels = asksNorm.length;
+
+	const topBidSize =
+		bidLevels > 0 ? Number((bidsNorm[0].sizeBase as number | null) ?? null) : null;
+	const topAskSize =
+		askLevels > 0 ? Number((asksNorm[0].sizeBase as number | null) ?? null) : null;
+	const topBidNotional =
+		bidLevels > 0 ? Number((bidsNorm[0].notional as number | null) ?? null) : null;
+	const topAskNotional =
+		askLevels > 0 ? Number((asksNorm[0].notional as number | null) ?? null) : null;
+
+	const depthBidBase = bidsNorm.reduce(
+		(sum, level) => sum + Number(level.sizeBase ?? 0),
+		0
+	);
+	const depthAskBase = asksNorm.reduce(
+		(sum, level) => sum + Number(level.sizeBase ?? 0),
+		0
+	);
+	const depthBidQuote = bidsNorm.reduce(
+		(sum, level) => sum + Number(level.notional ?? 0),
+		0
+	);
+	const depthAskQuote = asksNorm.reduce(
+		(sum, level) => sum + Number(level.notional ?? 0),
+		0
+	);
+	const denom = depthBidQuote + depthAskQuote;
+	const imbalance =
+		denom > 0 ? (depthBidQuote - depthAskQuote) / denom : null;
+
+	return {
+		source: config.source,
+		marketType: raw.marketType,
+		marketIndex: raw.marketIndex,
+		marketName: raw.marketName,
+		isIndicative: raw.isIndicative,
+		tsMs: raw.tsMs,
+		slot: raw.slot,
+		marketSlot: raw.marketSlot,
+		markPrice,
+		oraclePrice,
+		bestBidPrice,
+		bestAskPrice,
+		midPrice,
+		spreadQuote,
+		spreadBps,
+		depthLevels: config.depthLevels,
+		bidLevels,
+		askLevels,
+		topBidSize,
+		topAskSize,
+		topBidNotional,
+		topAskNotional,
+		depthBidBase,
+		depthAskBase,
+		depthBidQuote,
+		depthAskQuote,
+		imbalance,
+		bidsNorm,
+		asksNorm,
+		rawPayloadHash: raw.payloadHash,
+	};
+}

src/scripts/dlobHotPostgresToPostgresDerivedWriter.ts

@@ -0,0 +1,343 @@
+import { Pool } from 'pg';
+import process from 'node:process';
+
+import {
+	deriveNormalizedRow,
+	envInt,
+	sleep,
+	type RawSnapshotRow,
+} from './dlobHotPgCommon';
+
+const cfg = {
+	source: String(process.env.DLOB_SOURCE || 'mevnode_bot_hot_derived'),
+	rawSource: String(process.env.RAW_SOURCE || 'mevnode_bot_hot_raw'),
+	pollMs: envInt('DLOB_POLL_MS', 5000),
+	depthLevels: envInt('NORMALIZED_DEPTH', 10),
+	pricePrecision: envInt('PRICE_PRECISION', 1_000_000),
+	basePrecision: envInt('BASE_PRECISION', 1_000_000_000),
+	pgHost: String(process.env.PGHOST || 'postgres'),
+	pgPort: envInt('PGPORT', 5432),
+	pgUser: String(process.env.PGUSER || 'admin'),
+	pgPassword: String(process.env.PGPASSWORD || ''),
+	pgDatabase: String(process.env.PGDATABASE || 'crypto'),
+};
+
+const pool = new Pool({
+	host: cfg.pgHost,
+	port: cfg.pgPort,
+	user: cfg.pgUser,
+	password: cfg.pgPassword,
+	database: cfg.pgDatabase,
+	max: 4,
+});
+
+const seen = new Map<string, string>();
+let syncQueue: Promise<void> = Promise.resolve();
+
+const selectRawLatestSql = `
+SELECT
+	source,
+	redis_key,
+	snapshot_kind,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	payload_hash,
+	mark_price_raw,
+	oracle_price_raw,
+	best_bid_price_raw,
+	best_ask_price_raw,
+	spread_pct_raw,
+	spread_quote_raw,
+	oracle_data,
+	bids,
+	asks,
+	payload
+FROM public.dlob_hot_snapshot_latest
+WHERE source = $1
+  AND snapshot_kind = 'orderbook_l2'
+ORDER BY market_index NULLS LAST, is_indicative ASC
+`;
+
+const insertDerivedTsSql = `
+INSERT INTO public.dlob_hot_derived_ts (
+	event_ts,
+	source,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	mark_price,
+	oracle_price,
+	best_bid_price,
+	best_ask_price,
+	mid_price,
+	spread_quote,
+	spread_bps,
+	depth_levels,
+	bid_levels,
+	ask_levels,
+	top_bid_size,
+	top_ask_size,
+	top_bid_notional,
+	top_ask_notional,
+	depth_bid_base,
+	depth_ask_base,
+	depth_bid_quote,
+	depth_ask_quote,
+	imbalance,
+	bids_norm,
+	asks_norm,
+	raw_payload_hash
+) VALUES (
+	to_timestamp($1::double precision / 1000.0),
+	$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31
+)
+ON CONFLICT DO NOTHING
+`;
+
+const upsertDerivedLatestSql = `
+INSERT INTO public.dlob_hot_derived_latest (
+	source,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	mark_price,
+	oracle_price,
+	best_bid_price,
+	best_ask_price,
+	mid_price,
+	spread_quote,
+	spread_bps,
+	depth_levels,
+	bid_levels,
+	ask_levels,
+	top_bid_size,
+	top_ask_size,
+	top_bid_notional,
+	top_ask_notional,
+	depth_bid_base,
+	depth_ask_base,
+	depth_bid_quote,
+	depth_ask_quote,
+	imbalance,
+	bids_norm,
+	asks_norm,
+	raw_payload_hash,
+	updated_at
+) VALUES (
+	$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,now()
+)
+ON CONFLICT (source, market_type, market_index, is_indicative)
+DO UPDATE SET
+	market_name = EXCLUDED.market_name,
+	ts_ms = EXCLUDED.ts_ms,
+	slot = EXCLUDED.slot,
+	market_slot = EXCLUDED.market_slot,
+	mark_price = EXCLUDED.mark_price,
+	oracle_price = EXCLUDED.oracle_price,
+	best_bid_price = EXCLUDED.best_bid_price,
+	best_ask_price = EXCLUDED.best_ask_price,
+	mid_price = EXCLUDED.mid_price,
+	spread_quote = EXCLUDED.spread_quote,
+	spread_bps = EXCLUDED.spread_bps,
+	depth_levels = EXCLUDED.depth_levels,
+	bid_levels = EXCLUDED.bid_levels,
+	ask_levels = EXCLUDED.ask_levels,
+	top_bid_size = EXCLUDED.top_bid_size,
+	top_ask_size = EXCLUDED.top_ask_size,
+	top_bid_notional = EXCLUDED.top_bid_notional,
+	top_ask_notional = EXCLUDED.top_ask_notional,
+	depth_bid_base = EXCLUDED.depth_bid_base,
+	depth_ask_base = EXCLUDED.depth_ask_base,
+	depth_bid_quote = EXCLUDED.depth_bid_quote,
+	depth_ask_quote = EXCLUDED.depth_ask_quote,
+	imbalance = EXCLUDED.imbalance,
+	bids_norm = EXCLUDED.bids_norm,
+	asks_norm = EXCLUDED.asks_norm,
+	raw_payload_hash = EXCLUDED.raw_payload_hash,
+	updated_at = EXCLUDED.updated_at
+`;
+
+function rowToRawSnapshot(row: Record<string, unknown>): RawSnapshotRow {
+	return {
+		source: String(row.source),
+		redisKey: String(row.redis_key),
+		snapshotKind: 'orderbook_l2',
+		marketType: String(row.market_type),
+		marketIndex:
+			typeof row.market_index === 'number' ? row.market_index : null,
+		marketName: String(row.market_name),
+		isIndicative: Boolean(row.is_indicative),
+		tsMs: Number(row.ts_ms),
+		slot: typeof row.slot === 'number' ? row.slot : null,
+		marketSlot: typeof row.market_slot === 'number' ? row.market_slot : null,
+		payloadHash: String(row.payload_hash),
+		markPriceRaw: row.mark_price_raw ? String(row.mark_price_raw) : null,
+		oraclePriceRaw: row.oracle_price_raw ? String(row.oracle_price_raw) : null,
+		bestBidPriceRaw: row.best_bid_price_raw ? String(row.best_bid_price_raw) : null,
+		bestAskPriceRaw: row.best_ask_price_raw ? String(row.best_ask_price_raw) : null,
+		spreadPctRaw: row.spread_pct_raw ? String(row.spread_pct_raw) : null,
+		spreadQuoteRaw: row.spread_quote_raw ? String(row.spread_quote_raw) : null,
+		oracleData: row.oracle_data ?? null,
+		bids: Array.isArray(row.bids) ? row.bids : null,
+		asks: Array.isArray(row.asks) ? row.asks : null,
+		payload: (row.payload as Record<string, unknown>) ?? {},
+	};
+}
+
+async function syncOnce(): Promise<void> {
+	const client = await pool.connect();
+	let writes = 0;
+
+	try {
+		const { rows } = await client.query(selectRawLatestSql, [cfg.rawSource]);
+		await client.query('BEGIN');
+
+		for (const dbRow of rows as Array<Record<string, unknown>>) {
+			const raw = rowToRawSnapshot(dbRow);
+			const normalized = deriveNormalizedRow(raw, {
+				source: cfg.source,
+				pricePrecision: cfg.pricePrecision,
+				basePrecision: cfg.basePrecision,
+				depthLevels: cfg.depthLevels,
+			});
+			if (normalized == null) continue;
+
+			const token = `${normalized.tsMs}:${normalized.rawPayloadHash}`;
+			const dedupeKey = `${normalized.marketType}:${normalized.marketIndex}:${normalized.isIndicative}`;
+			if (seen.get(dedupeKey) === token) continue;
+
+			await client.query(insertDerivedTsSql, [
+				normalized.tsMs,
+				normalized.source,
+				normalized.marketType,
+				normalized.marketIndex,
+				normalized.marketName,
+				normalized.isIndicative,
+				normalized.tsMs,
+				normalized.slot,
+				normalized.marketSlot,
+				normalized.markPrice,
+				normalized.oraclePrice,
+				normalized.bestBidPrice,
+				normalized.bestAskPrice,
+				normalized.midPrice,
+				normalized.spreadQuote,
+				normalized.spreadBps,
+				normalized.depthLevels,
+				normalized.bidLevels,
+				normalized.askLevels,
+				normalized.topBidSize,
+				normalized.topAskSize,
+				normalized.topBidNotional,
+				normalized.topAskNotional,
+				normalized.depthBidBase,
+				normalized.depthAskBase,
+				normalized.depthBidQuote,
+				normalized.depthAskQuote,
+				normalized.imbalance,
+				JSON.stringify(normalized.bidsNorm),
+				JSON.stringify(normalized.asksNorm),
+				normalized.rawPayloadHash,
+			]);
+
+			await client.query(upsertDerivedLatestSql, [
+				normalized.source,
+				normalized.marketType,
+				normalized.marketIndex,
+				normalized.marketName,
+				normalized.isIndicative,
+				normalized.tsMs,
+				normalized.slot,
+				normalized.marketSlot,
+				normalized.markPrice,
+				normalized.oraclePrice,
+				normalized.bestBidPrice,
+				normalized.bestAskPrice,
+				normalized.midPrice,
+				normalized.spreadQuote,
+				normalized.spreadBps,
+				normalized.depthLevels,
+				normalized.bidLevels,
+				normalized.askLevels,
+				normalized.topBidSize,
+				normalized.topAskSize,
+				normalized.topBidNotional,
+				normalized.topAskNotional,
+				normalized.depthBidBase,
+				normalized.depthAskBase,
+				normalized.depthBidQuote,
+				normalized.depthAskQuote,
+				normalized.imbalance,
+				JSON.stringify(normalized.bidsNorm),
+				JSON.stringify(normalized.asksNorm),
+				normalized.rawPayloadHash,
+			]);
+
+			seen.set(dedupeKey, token);
+			writes += 1;
+		}
+
+		await client.query('COMMIT');
+		if (writes > 0) {
+			console.log(
+				`[dlob-hot-postgres-to-postgres-derived-writer] wrote ${writes} normalized snapshot(s) from raw source ${cfg.rawSource}`
+			);
+		}
+	} catch (error) {
+		await client.query('ROLLBACK');
+		console.error('[dlob-hot-postgres-to-postgres-derived-writer] sync failed', error);
+	} finally {
+		client.release();
+	}
+}
+
+function enqueueSync(reason: string): void {
+	syncQueue = syncQueue
+		.then(async () => {
+			await syncOnce();
+		})
+		.catch((error) => {
+			console.error(
+				`[dlob-hot-postgres-to-postgres-derived-writer] queued sync failed (${reason})`,
+				error
+			);
+		});
+}
+
+async function main(): Promise<void> {
+	console.log(
+		`[dlob-hot-postgres-to-postgres-derived-writer] source=${cfg.source} rawSource=${cfg.rawSource} pg=${cfg.pgHost}:${cfg.pgPort}/${cfg.pgDatabase}`
+	);
+	await pool.query('select 1');
+	const listener = await pool.connect();
+	await listener.query('LISTEN dlob_hot_raw_updates');
+	listener.on('notification', () => {
+		enqueueSync('notify:dlob_hot_raw_updates');
+	});
+
+	await syncOnce();
+
+	for (;;) {
+		await sleep(cfg.pollMs);
+		enqueueSync('fallback-poll');
+		await syncQueue;
+	}
+}
+
+main().catch((error) => {
+	console.error('[dlob-hot-postgres-to-postgres-derived-writer] fatal', error);
+	process.exit(1);
+});

src/scripts/dlobHotRedisToPostgresRawWriter.ts

@@ -0,0 +1,280 @@
+import Redis from 'ioredis';
+import { Pool } from 'pg';
+import process from 'node:process';
+
+import {
+	deriveRawSnapshotRow,
+	envInt,
+	sleep,
+} from './dlobHotPgCommon';
+
+const cfg = {
+	source: String(process.env.DLOB_SOURCE || 'mevnode_bot_hot_raw'),
+	redisHost: String(process.env.REDIS_HOST || process.env.ELASTICACHE_HOST || 'dlob-redis'),
+	redisPort: envInt('REDIS_PORT', envInt('ELASTICACHE_PORT', 6379)),
+	redisKeyPrefix: String(process.env.REDIS_KEY_PREFIX || 'dlob-hot:'),
+	pollMs: envInt('DLOB_POLL_MS', 5000),
+	pgHost: String(process.env.PGHOST || 'postgres'),
+	pgPort: envInt('PGPORT', 5432),
+	pgUser: String(process.env.PGUSER || 'admin'),
+	pgPassword: String(process.env.PGPASSWORD || ''),
+	pgDatabase: String(process.env.PGDATABASE || 'crypto'),
+};
+
+const redis = new Redis({
+	host: cfg.redisHost,
+	port: cfg.redisPort,
+	maxRetriesPerRequest: null,
+	enableReadyCheck: false,
+});
+
+const pool = new Pool({
+	host: cfg.pgHost,
+	port: cfg.pgPort,
+	user: cfg.pgUser,
+	password: cfg.pgPassword,
+	database: cfg.pgDatabase,
+	max: 4,
+});
+const subscriber = new Redis({
+	host: cfg.redisHost,
+	port: cfg.redisPort,
+	maxRetriesPerRequest: null,
+	enableReadyCheck: false,
+});
+
+const seen = new Map<string, string>();
+let writeQueue: Promise<void> = Promise.resolve();
+
+const insertRawTsSql = `
+INSERT INTO public.dlob_hot_snapshot_ts (
+	event_ts,
+	source,
+	redis_key,
+	snapshot_kind,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	payload_hash,
+	mark_price_raw,
+	oracle_price_raw,
+	best_bid_price_raw,
+	best_ask_price_raw,
+	spread_pct_raw,
+	spread_quote_raw,
+	oracle_data,
+	bids,
+	asks,
+	payload
+) VALUES (
+	to_timestamp($1::double precision / 1000.0),
+	$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22
+)
+ON CONFLICT DO NOTHING
+`;
+
+const upsertRawLatestSql = `
+INSERT INTO public.dlob_hot_snapshot_latest (
+	source,
+	redis_key,
+	snapshot_kind,
+	market_type,
+	market_index,
+	market_name,
+	is_indicative,
+	ts_ms,
+	slot,
+	market_slot,
+	payload_hash,
+	mark_price_raw,
+	oracle_price_raw,
+	best_bid_price_raw,
+	best_ask_price_raw,
+	spread_pct_raw,
+	spread_quote_raw,
+	oracle_data,
+	bids,
+	asks,
+	payload,
+	updated_at
+) VALUES (
+	$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,now()
+)
+ON CONFLICT (source, market_type, market_index, snapshot_kind, is_indicative)
+DO UPDATE SET
+	redis_key = EXCLUDED.redis_key,
+	market_name = EXCLUDED.market_name,
+	ts_ms = EXCLUDED.ts_ms,
+	slot = EXCLUDED.slot,
+	market_slot = EXCLUDED.market_slot,
+	payload_hash = EXCLUDED.payload_hash,
+	mark_price_raw = EXCLUDED.mark_price_raw,
+	oracle_price_raw = EXCLUDED.oracle_price_raw,
+	best_bid_price_raw = EXCLUDED.best_bid_price_raw,
+	best_ask_price_raw = EXCLUDED.best_ask_price_raw,
+	spread_pct_raw = EXCLUDED.spread_pct_raw,
+	spread_quote_raw = EXCLUDED.spread_quote_raw,
+	oracle_data = EXCLUDED.oracle_data,
+	bids = EXCLUDED.bids,
+	asks = EXCLUDED.asks,
+	payload = EXCLUDED.payload,
+	updated_at = EXCLUDED.updated_at
+`;
+
+function canonicalRedisKey(redisKey: string): string {
+	if (!redisKey.startsWith(cfg.redisKeyPrefix)) return redisKey;
+	const suffix = redisKey.slice(cfg.redisKeyPrefix.length);
+	if (suffix.startsWith('last_update_')) return redisKey;
+	return `${cfg.redisKeyPrefix}last_update_${suffix}`;
+}
+
+async function persistRows(rows: Array<ReturnType<typeof deriveRawSnapshotRow>>, reason: string): Promise<void> {
+	const filteredRows = rows.filter((row): row is NonNullable<typeof row> => row != null);
+	if (filteredRows.length === 0) return;
+	const client = await pool.connect();
+	let writes = 0;
+
+	try {
+		await client.query('BEGIN');
+		for (const row of filteredRows) {
+			const token = `${row.tsMs}:${row.payloadHash}`;
+			if (seen.get(row.redisKey) === token) continue;
+
+			await client.query(insertRawTsSql, [
+				row.tsMs,
+				row.source,
+				row.redisKey,
+				row.snapshotKind,
+				row.marketType,
+				row.marketIndex,
+				row.marketName,
+				row.isIndicative,
+				row.tsMs,
+				row.slot,
+				row.marketSlot,
+				row.payloadHash,
+				row.markPriceRaw,
+				row.oraclePriceRaw,
+				row.bestBidPriceRaw,
+				row.bestAskPriceRaw,
+				row.spreadPctRaw,
+				row.spreadQuoteRaw,
+				row.oracleData ? JSON.stringify(row.oracleData) : null,
+				row.bids ? JSON.stringify(row.bids) : null,
+				row.asks ? JSON.stringify(row.asks) : null,
+				JSON.stringify(row.payload),
+			]);
+
+			await client.query(upsertRawLatestSql, [
+				row.source,
+				row.redisKey,
+				row.snapshotKind,
+				row.marketType,
+				row.marketIndex,
+				row.marketName,
+				row.isIndicative,
+				row.tsMs,
+				row.slot,
+				row.marketSlot,
+				row.payloadHash,
+				row.markPriceRaw,
+				row.oraclePriceRaw,
+				row.bestBidPriceRaw,
+				row.bestAskPriceRaw,
+				row.spreadPctRaw,
+				row.spreadQuoteRaw,
+				row.oracleData ? JSON.stringify(row.oracleData) : null,
+				row.bids ? JSON.stringify(row.bids) : null,
+				row.asks ? JSON.stringify(row.asks) : null,
+				JSON.stringify(row.payload),
+			]);
+
+			seen.set(row.redisKey, token);
+			writes += 1;
+		}
+		if (writes > 0) {
+			await client.query(`select pg_notify('dlob_hot_raw_updates', $1)`, [
+				String(Date.now()),
+			]);
+		}
+		await client.query('COMMIT');
+		if (writes > 0) {
+			console.log(
+				`[dlob-hot-redis-to-postgres-raw-writer] wrote ${writes} snapshot(s) via ${reason}`
+			);
+		}
+	} catch (error) {
+		await client.query('ROLLBACK');
+		console.error('[dlob-hot-redis-to-postgres-raw-writer] sync failed', error);
+	} finally {
+		client.release();
+	}
+}
+
+async function syncOnce(): Promise<void> {
+	const keys = (await redis.keys(`${cfg.redisKeyPrefix}*`)).sort();
+	if (keys.length === 0) return;
+
+	const rows: Array<ReturnType<typeof deriveRawSnapshotRow>> = [];
+	for (const key of keys) {
+		const payloadText = await redis.get(key);
+		if (!payloadText) continue;
+		rows.push(
+			deriveRawSnapshotRow({
+				source: cfg.source,
+				redisKey: key,
+				payloadText,
+			})
+		);
+	}
+
+	await persistRows(rows, `full-sync:${cfg.redisKeyPrefix}`);
+}
+
+function enqueueWrite(task: () => Promise<void>): void {
+	writeQueue = writeQueue
+		.then(task)
+		.catch((error) => {
+			console.error('[dlob-hot-redis-to-postgres-raw-writer] queued task failed', error);
+		});
+}
+
+function handlePublishedSnapshot(channel: string, payloadText: string): void {
+	enqueueWrite(async () => {
+		const row = deriveRawSnapshotRow({
+			source: cfg.source,
+			redisKey: canonicalRedisKey(channel),
+			payloadText,
+		});
+		await persistRows([row], `redis-publish:${channel}`);
+	});
+}
+
+async function main(): Promise<void> {
+	console.log(
+		`[dlob-hot-redis-to-postgres-raw-writer] source=${cfg.source} redis=${cfg.redisHost}:${cfg.redisPort} prefix=${cfg.redisKeyPrefix} pg=${cfg.pgHost}:${cfg.pgPort}/${cfg.pgDatabase}`
+	);
+	await redis.ping();
+	await pool.query('select 1');
+	await subscriber.psubscribe(`${cfg.redisKeyPrefix}orderbook_*`);
+	subscriber.on('pmessage', (_pattern, channel, message) => {
+		handlePublishedSnapshot(channel, message);
+	});
+
+	await syncOnce();
+
+	for (;;) {
+		await sleep(cfg.pollMs);
+		enqueueWrite(syncOnce);
+		await writeQueue;
+	}
+}
+
+main().catch((error) => {
+	console.error('[dlob-hot-redis-to-postgres-raw-writer] fatal', error);
+	process.exit(1);
+});

src/scripts/publishUnsettledPnlUsers.ts

@@ -1,4 +1,3 @@
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 import { sleep } from '@drift/common';
 import {
 	BigNum,
@@ -17,6 +16,11 @@ import {
 import { Connection, Keypair } from '@solana/web3.js';
 import { logger } from '../utils/logger';
 import Bottleneck from 'bottleneck';
+import {
+	RedisClient,
+	RedisClientPrefix,
+	HOT_REDIS_KEY_PREFIX,
+} from '../utils/redisClients';
 
 const dotenv = require('dotenv');
 dotenv.config();
@@ -61,7 +65,7 @@ const userMapRedisClient = new RedisClient({
 });
 
 const dlobRedisClient = new RedisClient({
-	prefix: RedisClientPrefix.DLOB,
+	prefix: HOT_REDIS_KEY_PREFIX,
 });
 
 const main = async () => {

src/serverLite.ts

@@ -12,9 +12,8 @@ import {
 	initialize,
 	MarketType,
 	getVariant,
+	isVariant,
 } from '@drift-labs/sdk';
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
-
 import { logger, setLogLevel } from './utils/logger';
 
 import * as http from 'http';
@@ -22,6 +21,7 @@ import { handleHealthCheck } from './core/middleware';
 import { errorHandler, selectMostRecentBySlot, sleep } from './utils/utils';
 import { setGlobalDispatcher, Agent } from 'undici';
 import { Metrics } from './core/metricsV2';
+import { RedisClient, parseRedisClients } from './utils/redisClients';
 
 setGlobalDispatcher(
 	new Agent({
@@ -32,16 +32,7 @@ setGlobalDispatcher(
 require('dotenv').config();
 
 // Reading in Redis env vars
-const envClients = [];
+const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT);
-const clients = process.env.REDIS_CLIENT?.trim()
-	.replace(/^\[|\]$/g, '')
-	.split(/\s*,\s*/);
-
-clients?.forEach((client) => envClients.push(RedisClientPrefix[client]));
-
-const REDIS_CLIENTS = envClients.length
-	? envClients
-	: [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
 console.log('Redis Clients:', REDIS_CLIENTS);
 
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -203,6 +194,112 @@ const main = async (): Promise<void> => {
 		}
 	});
 
+	app.get('/l2', async (req, res, next) => {
+		try {
+			const { marketName, marketIndex, marketType, depth, includeIndicative } =
+				req.query;
+
+			let normedMarketType: MarketType | undefined;
+			let normedMarketIndex: number | undefined;
+
+			if (marketName != null && String(marketName).trim() !== '') {
+				const name = String(marketName).toUpperCase().trim();
+				const perp = sdkConfig?.PERP_MARKETS?.find((m) => m?.symbol === name);
+				if (perp) {
+					normedMarketType = MarketType.PERP;
+					normedMarketIndex = perp.marketIndex;
+				} else {
+					const spot = sdkConfig?.SPOT_MARKETS?.find((m) => m?.symbol === name);
+					if (spot) {
+						normedMarketType = MarketType.SPOT;
+						normedMarketIndex = spot.marketIndex;
+					}
+				}
+
+				if (normedMarketType == null || normedMarketIndex == null) {
+					res.status(400).send('Bad Request: unrecognized marketName');
+					return;
+				}
+			} else {
+				if (marketIndex == null || marketType == null) {
+					res
+						.status(400)
+						.send(
+							'Bad Request: (marketName) or (marketIndex and marketType) must be supplied'
+						);
+					return;
+				}
+
+				const mt = String(marketType).toLowerCase().trim();
+				const idx = parseInt(String(marketIndex), 10);
+				if (!Number.isFinite(idx) || idx < 0) {
+					res.status(400).send('Bad Request: invalid marketIndex');
+					return;
+				}
+
+				if (mt === 'perp') {
+					if (!sdkConfig?.PERP_MARKETS?.[idx]) {
+						res.status(400).send('Bad Request: invalid marketIndex');
+						return;
+					}
+					normedMarketType = MarketType.PERP;
+					normedMarketIndex = idx;
+				} else if (mt === 'spot') {
+					if (!sdkConfig?.SPOT_MARKETS?.[idx]) {
+						res.status(400).send('Bad Request: invalid marketIndex');
+						return;
+					}
+					normedMarketType = MarketType.SPOT;
+					normedMarketIndex = idx;
+				} else {
+					res
+						.status(400)
+						.send('Bad Request: marketType must be either "spot" or "perp"');
+					return;
+				}
+			}
+
+			const includeIndicativeStr =
+				(includeIndicative as string)?.toLowerCase() === 'true';
+			const depthToUse = Math.min(parseInt(String(depth ?? '100'), 10) || 1, 100);
+			const indicativeSuffix = includeIndicativeStr ? '_indicative' : '';
+
+			const redisL2 = await fetchFromRedis(
+				`last_update_orderbook_${getVariant(normedMarketType)}_${normedMarketIndex}${indicativeSuffix}`,
+				selectMostRecentBySlot
+			);
+
+			if (
+				redisL2 &&
+				slotSubscriber.getSlot() - redisL2['slot'] < SLOT_STALENESS_TOLERANCE
+			) {
+				redisL2['bids'] = redisL2['bids']?.slice(0, depthToUse);
+				redisL2['asks'] = redisL2['asks']?.slice(0, depthToUse);
+				cacheHitCounter.add(1, {
+					miss: false,
+					path: req.baseUrl + req.path,
+					marketIndex: normedMarketIndex,
+					marketType: isVariant(normedMarketType, 'spot') ? 'spot' : 'perp',
+				});
+				res.writeHead(200);
+				res.end(JSON.stringify(redisL2));
+				return;
+			} else {
+				cacheHitCounter.add(1, {
+					miss: true,
+					path: req.baseUrl + req.path,
+					marketIndex: normedMarketIndex,
+					marketType: isVariant(normedMarketType, 'spot') ? 'spot' : 'perp',
+				});
+				res.writeHead(500);
+				res.end(JSON.stringify({ error: 'No cached L2 found' }));
+				return;
+			}
+		} catch (err) {
+			next(err);
+		}
+	});
+
 	server.listen(serverPort, () => {
 		logger.info(
 			`DLOB server lite listening on port http://localhost:${serverPort}`

src/utils/redisClients.ts

@@ -0,0 +1,47 @@
+import {
+	RedisClient,
+	RedisClientPrefix as DriftRedisClientPrefix,
+} from '@drift/common/clients';
+
+export { RedisClient };
+
+export const HOT_REDIS_CLIENT = 'DLOB_HOT';
+export const ALL_REDIS_CLIENT = 'DLOB_ALL';
+
+export const HOT_REDIS_KEY_PREFIX = 'dlob-hot:';
+export const ALL_REDIS_KEY_PREFIX = 'dlob-all:';
+
+export const RedisClientPrefix = {
+	USER_MAP: DriftRedisClientPrefix.USER_MAP,
+	DLOB_HOT: HOT_REDIS_KEY_PREFIX,
+	DLOB_ALL: ALL_REDIS_KEY_PREFIX,
+} as const;
+
+export function getRedisClientPrefix(clientName?: string): string {
+	switch ((clientName || HOT_REDIS_CLIENT).trim()) {
+		case 'DLOB':
+		case HOT_REDIS_CLIENT:
+			return HOT_REDIS_KEY_PREFIX;
+		case 'DLOB_HELIUS':
+		case ALL_REDIS_CLIENT:
+			return ALL_REDIS_KEY_PREFIX;
+		case 'USER_MAP':
+			return RedisClientPrefix.USER_MAP;
+		default:
+			throw new Error(`Unknown REDIS_CLIENT: ${clientName}`);
+	}
+}
+
+export function parseRedisClients(clientNames?: string): string[] {
+	const prefixes =
+		clientNames
+			?.trim()
+			.replace(/^\[|\]$/g, '')
+			.split(/\s*,\s*/)
+			.filter(Boolean)
+			.map((client) => getRedisClientPrefix(client)) ?? [];
+
+	return prefixes.length
+		? prefixes
+		: [HOT_REDIS_KEY_PREFIX, ALL_REDIS_KEY_PREFIX];
+}

src/wsConnectionManager.ts

@@ -5,8 +5,12 @@ import compression from 'compression';
 import { WebSocket, WebSocketServer } from 'ws';
 import { sleep, selectMostRecentBySlot, GROUPING_OPTIONS } from './utils/utils';
 import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk';
-import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 import { Metrics, GaugeValue } from './core/metricsV2';
+import {
+	RedisClient,
+	ALL_REDIS_KEY_PREFIX,
+	parseRedisClients,
+} from './utils/redisClients';
 
 // Stream selector with hysteresis to prevent flip-flopping between sources
 const STREAM_SWITCH_THRESHOLD_MS = 10_000; // 10 seconds
@@ -211,16 +215,7 @@ console.log(`WS LISTENER PORT : ${WS_PORT}`);
 
 const MAX_BUFFERED_AMOUNT = 300000;
 
-const envClients = [];
+const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT);
-const clients = process.env.REDIS_CLIENT?.trim()
-	.replace(/^\[|\]$/g, '')
-	.split(/\s*,\s*/);
-
-clients?.forEach((client) => envClients.push(RedisClientPrefix[client]));
-
-const REDIS_CLIENTS = envClients.length
-	? envClients
-	: [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS];
 console.log('Redis Clients:', REDIS_CLIENTS);
 
 const regexp = new RegExp(REDIS_CLIENTS.join('|'), 'g');
@@ -438,7 +433,7 @@ async function main() {
 						redisClients[0]
 							.subscribe(
 								redisChannel.includes('priorityFees')
-									? `dlob-helius:${redisChannel}`
+									? `${ALL_REDIS_KEY_PREFIX}${redisChannel}`
 									: `${redisClients[0].getPrefix()}${redisChannel}`
 							)
 							.then(() => {