chore(snapshot): sync workspace state

Dockerfile

@@ -1,17 +1,21 @@
 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:24-alpine
@@ -19,7 +23,7 @@ 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

README.md

@@ -82,12 +82,6 @@ To properly configure the DLOB server, set the following environment variables i
 | `ELASTICACHE_HOST`            | (for websocket server) Redis host endpoint.                     | `localhost`                         |
 | `ELASTICACHE_PORT`            | (for websocket server) Redis port.                              | `6379`                              |
 | `REDIS_CLIENT`                | (for websocket server) Redis client type (DLOB/DLOB_HELIUS).    | `DLOB`                              |
-| `ENABLE_PERSISTENT_STORE`     | Persist canonical latest DLOB snapshots to Postgres.            | `true`                              |
-| `POSTGRES_URL`                | Connection string for canonical DLOB store.                     | `postgres://user:pass@host:5432/db` |
-| `DLOB_SOURCE`                 | Source key stored with canonical DLOB rows.                     | `mevnode`                           |
-| `PRICE_PRECISION`             | Price scale used by projection views.                           | `1000000`                           |
-| `BASE_PRECISION`              | Base size scale used by projection views.                       | `1000000000`                        |
-| `PERSISTENT_STATS_DEPTH`      | Depth used by SQL projection views for stats.                   | `10`                                |
 | `WS_PORT`                     | (for websocket server) The port to run the websocket server on. | `3000`                              |
 
 Note: multiple Redis hosts can be provided by providing a comma separated string.
@@ -137,30 +131,6 @@ When you're done, stop the redis cluster:
 bash redisCluster.sh stop
 ```
 
-## Persistent canonical store
-
-The publisher can now persist canonical latest DLOB snapshots directly to Postgres while still using Redis as the hot path.
-
-- Redis remains the runtime cache and pub/sub layer.
-- Postgres stores canonical latest snapshots in:
-  - `dlob_orderbook_l2_latest`
-  - `dlob_orderbook_l3_latest`
-  - `dlob_best_makers_latest`
-- Postgres exposes projection views for Hasura and frontend adaptation:
-  - `dlob_l2_latest_projection`
-  - `dlob_l3_latest_projection`
-  - `dlob_best_makers_latest_projection`
-  - `dlob_stats_latest_projection`
-- This aligns the DLOB stack with a `Redis hot state + Postgres durable canonical store + projection views` model.
-
-To initialize the persistent tables without connecting to chain inputs yet:
-
-```bash
-ENABLE_PERSISTENT_STORE=true \
-POSTGRES_URL=postgres://user:pass@host:5432/db \
-yarn dlob-storage:init
-```
-
 # Run the example client
 
 Documentation for connecting to the dlob server are available [here](https://drift-labs.github.io/v2-teacher/?python#orderbook-trades-dlob-server)

package.json

@@ -32,7 +32,7 @@
 		"express": "^4.18.2",
 		"ioredis": "^5.4.1",
 		"morgan": "^1.10.0",
-		"pg": "^8.13.1",
+		"pg": "^8.12.0",
 		"redis": "^4.6.10",
 		"response-time": "^2.3.2",
 		"rxjs": "^7.8.1",
@@ -71,7 +71,6 @@
 		"dev": "ts-node src/index.ts",
 		"server-lite": "ts-node src/serverLite.ts",
 		"dlob-publish": "ts-node src/publishers/dlobPublisher.ts",
-		"dlob-storage:init": "ts-node src/scripts/initPersistentStore.ts",
 		"trades-publish": "ts-node src/publishers/tradesPublisher.ts",
 		"fees-publish": "ts-node src/publishers/priorityFeesPublisher.ts",
 		"pnl-publish": "ts-node src/scripts/publishUnsettledPnlUsers.ts",

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/DLOBSubscriberIO.ts

@@ -32,7 +32,6 @@ import { OffloadQueue } from '../utils/offload';
 import { setHealthStatus, HEALTH_STATUS } from '../core/healthCheck';
 import { CounterValue } from '../core/metricsV2';
 import { COMMON_MATH } from '@drift/common';
-import { DlobSnapshotStore } from '../storage/dlobSnapshotStore';
 
 export type wsMarketArgs = {
 	marketIndex: number;
@@ -71,7 +70,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
 	redisClient: RedisClient;
 	indicativeQuotesRedisClient?: RedisClient;
-	persistentStore?: DlobSnapshotStore;
 	public killSwitchSlotDiffThreshold: number;
 	public offloadQueue?: ReturnType<typeof OffloadQueue>;
 	public enableOffload: boolean;
@@ -81,7 +79,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			env: DriftEnv;
 			redisClient: RedisClient;
 			indicativeQuotesRedisClient?: RedisClient;
-			persistentStore?: DlobSnapshotStore;
 			enableOffloadQueue?: boolean;
 			offloadQueueCounter?: CounterValue;
 			perpMarketInfos: wsMarketInfo[];
@@ -93,7 +90,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		super(config);
 		this.redisClient = config.redisClient;
 		this.indicativeQuotesRedisClient = config.indicativeQuotesRedisClient;
-		this.persistentStore = config.persistentStore;
 
 		this.killSwitchSlotDiffThreshold =
 			config.killSwitchSlotDiffThreshold || 200;
@@ -474,19 +470,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			l2Formatted_depth100
 		);
 
-		if (this.persistentStore) {
-			void this.persistentStore
-				.writeL2Snapshot(l2Formatted, {
-					indicative: Boolean(this.indicativeQuotesRedisClient),
-				})
-				.catch((error) => {
-				logger.error(
-					`Failed to persist canonical L2 snapshot for ${marketName}`,
-					error
-				);
-				});
-		}
-
 		publishGroupings(
 			l2Formatted,
 			marketArgs,
@@ -525,23 +508,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				`last_update_orderbook_best_makers_${marketType}_${marketArgs.marketIndex}`,
 				{ bids, asks, slot }
 			);
-			if (this.persistentStore) {
-				void this.persistentStore
-					.writeBestMakersSnapshot({
-						marketName: marketName?.toUpperCase() ?? String(marketArgs.marketIndex),
-						marketType: marketType?.toLowerCase(),
-						marketIndex: marketArgs.marketIndex,
-						slot: slot ?? null,
-						bids,
-						asks,
-					})
-					.catch((error) => {
-						logger.error(
-							`Failed to persist canonical best makers for ${marketName}`,
-							error
-						);
-					});
-			}
 		}
 	}
 
@@ -605,18 +571,5 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			}`,
 			l3
 		);
-
-		if (this.persistentStore) {
-			void this.persistentStore
-				.writeL3Snapshot(l3, {
-					indicative: Boolean(this.indicativeQuotesRedisClient),
-				})
-				.catch((error) => {
-					logger.error(
-						`Failed to persist canonical L3 snapshot for ${marketName}`,
-						error
-					);
-				});
-		}
 	}
 }

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,11 +45,28 @@ import {
 	FillQualityAnalyticsRepository,
 	TakerFillVsOracleBpsRedisResult,
 } from '../athena/repositories/fillQualityAnalytics';
-import { createPostgresCanonicalStoreFromEnv } from '../storage/postgresCanonicalStore';
+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),
 	})
 );
 
@@ -63,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();
@@ -172,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}`
 );
@@ -381,23 +401,13 @@ 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();
-	const persistentStore = createPostgresCanonicalStoreFromEnv();
-	await persistentStore.init();
 
 	const indicativeRedisClient = new RedisClient({});
 	await indicativeRedisClient.connect();
 
-	logger.info(
-		`Persistent store: ${
-			process.env.ENABLE_PERSISTENT_STORE?.toLowerCase() === 'true'
-				? 'postgres canonical + projections'
-				: 'disabled'
-		}`
-	);
-
 	const connection = new Connection(endpoint, {
 		wsEndpoint: wsEndpoint,
 		commitment: stateCommitment,
@@ -484,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);
@@ -593,7 +606,6 @@ const main = async () => {
 		spotMarketSubscribers: MARKET_SUBSCRIBERS,
 		perpMarketInfos,
 		spotMarketInfos,
-		persistentStore,
 		killSwitchSlotDiffThreshold: KILLSWITCH_SLOT_DIFF_THRESHOLD,
 		protectedMakerView: false,
 	});
@@ -650,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) => {
@@ -680,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/initPersistentStore.ts

@@ -1,16 +0,0 @@
-import { createPostgresCanonicalStoreFromEnv } from '../storage/postgresCanonicalStore';
-import { logger } from '../utils/logger';
-
-require('dotenv').config();
-
-async function main() {
-	const store = createPostgresCanonicalStoreFromEnv();
-	await store.init();
-	logger.info('Persistent DLOB store initialized');
-	await store.close();
-}
-
-main().catch((error) => {
-	logger.error('Failed to initialize persistent DLOB store', 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/storage/dlobSnapshotStore.ts

@@ -1,45 +0,0 @@
-export interface DlobSnapshotStore {
-	init(): Promise<void>;
-	writeL2Snapshot(
-		l2Snapshot: Record<string, any>,
-		options: { indicative: boolean }
-	): Promise<void>;
-	writeL3Snapshot(
-		l3Snapshot: Record<string, any>,
-		options: { indicative: boolean }
-	): Promise<void>;
-	writeBestMakersSnapshot(snapshot: {
-		marketName: string;
-		marketType: string;
-		marketIndex: number;
-		slot: number | null;
-		bids: string[];
-		asks: string[];
-	}): Promise<void>;
-	close(): Promise<void>;
-}
-
-export class NoopDlobSnapshotStore implements DlobSnapshotStore {
-	async init(): Promise<void> {}
-
-	async writeL2Snapshot(
-		_l2Snapshot: Record<string, any>,
-		_options: { indicative: boolean }
-	): Promise<void> {}
-
-	async writeL3Snapshot(
-		_l3Snapshot: Record<string, any>,
-		_options: { indicative: boolean }
-	): Promise<void> {}
-
-	async writeBestMakersSnapshot(_snapshot: {
-		marketName: string;
-		marketType: string;
-		marketIndex: number;
-		slot: number | null;
-		bids: string[];
-		asks: string[];
-	}): Promise<void> {}
-
-	async close(): Promise<void> {}
-}

src/storage/postgresCanonicalStore.ts

@@ -1,560 +0,0 @@
-import { Pool } from 'pg';
-
-import { logger } from '../utils/logger';
-import {
-	DlobSnapshotStore,
-	NoopDlobSnapshotStore,
-} from './dlobSnapshotStore';
-
-type PostgresCanonicalStoreConfig = {
-	connectionString?: string;
-	host?: string;
-	port?: number;
-	user?: string;
-	password?: string;
-	database?: string;
-	source: string;
-	pricePrecision: number;
-	basePrecision: number;
-	statsDepth: number;
-};
-
-function envBool(name: string, fallback = false): boolean {
-	const raw = process.env[name];
-	if (raw == null) return fallback;
-	const value = String(raw).trim().toLowerCase();
-	if (['1', 'true', 'yes', 'y', 'on'].includes(value)) return true;
-	if (['0', 'false', 'no', 'n', 'off'].includes(value)) return false;
-	return fallback;
-}
-
-function envInt(name: string, fallback: number): number {
-	const raw = process.env[name];
-	if (raw == null) return fallback;
-	const parsed = Number.parseInt(String(raw), 10);
-	return Number.isFinite(parsed) ? parsed : fallback;
-}
-
-function toIntOrNull(value: unknown): number | null {
-	if (value == null) return null;
-	const parsed =
-		typeof value === 'number' ? value : Number.parseInt(String(value), 10);
-	return Number.isFinite(parsed) ? Math.trunc(parsed) : null;
-}
-
-function toJson(value: unknown): string {
-	return JSON.stringify(value ?? null);
-}
-
-export class PostgresCanonicalStore implements DlobSnapshotStore {
-	private readonly pool: Pool;
-	private writeChain: Promise<void> = Promise.resolve();
-
-	constructor(private readonly config: PostgresCanonicalStoreConfig) {
-		this.pool = new Pool(
-			config.connectionString
-				? { connectionString: config.connectionString }
-				: {
-						host: config.host,
-						port: config.port,
-						user: config.user,
-						password: config.password,
-						database: config.database,
-				  }
-		);
-	}
-
-	async init(): Promise<void> {
-		await this.pool.query(`
-			CREATE TABLE IF NOT EXISTS public.dlob_orderbook_l2_latest (
-				source TEXT NOT NULL,
-				market_name TEXT NOT NULL,
-				market_type TEXT NOT NULL,
-				market_index INTEGER,
-				is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
-				ts BIGINT,
-				slot BIGINT,
-				mark_price_raw TEXT,
-				oracle_price_raw TEXT,
-				best_bid_price_raw TEXT,
-				best_ask_price_raw TEXT,
-				spread_pct_raw TEXT,
-				spread_quote_raw TEXT,
-				oracle_data JSONB,
-				bids JSONB,
-				asks JSONB,
-				payload JSONB NOT NULL,
-				updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-				PRIMARY KEY (source, market_name, is_indicative)
-			);
-		`);
-
-		await this.pool.query(`
-			CREATE TABLE IF NOT EXISTS public.dlob_orderbook_l3_latest (
-				source TEXT NOT NULL,
-				market_name TEXT NOT NULL,
-				market_type TEXT NOT NULL,
-				market_index INTEGER,
-				is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
-				ts BIGINT,
-				slot BIGINT,
-				oracle_data JSONB,
-				bids JSONB,
-				asks JSONB,
-				payload JSONB NOT NULL,
-				updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-				PRIMARY KEY (source, market_name, is_indicative)
-			);
-		`);
-
-		await this.pool.query(`
-			CREATE TABLE IF NOT EXISTS public.dlob_best_makers_latest (
-				source TEXT NOT NULL,
-				market_name TEXT NOT NULL,
-				market_type TEXT NOT NULL,
-				market_index INTEGER,
-				slot BIGINT,
-				bids JSONB,
-				asks JSONB,
-				payload JSONB NOT NULL,
-				updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-				PRIMARY KEY (source, market_name)
-			);
-		`);
-
-		await this.pool.query(`
-			CREATE INDEX IF NOT EXISTS dlob_orderbook_l2_latest_updated_at_idx
-			ON public.dlob_orderbook_l2_latest (updated_at DESC);
-		`);
-		await this.pool.query(`
-			CREATE INDEX IF NOT EXISTS dlob_orderbook_l3_latest_updated_at_idx
-			ON public.dlob_orderbook_l3_latest (updated_at DESC);
-		`);
-		await this.pool.query(`
-			CREATE INDEX IF NOT EXISTS dlob_best_makers_latest_updated_at_idx
-			ON public.dlob_best_makers_latest (updated_at DESC);
-		`);
-
-		await this.pool.query(`
-			CREATE OR REPLACE VIEW public.dlob_l2_latest_projection AS
-			SELECT
-				source,
-				market_name,
-				market_type,
-				market_index,
-				ts,
-				slot,
-				CASE
-					WHEN NULLIF(mark_price_raw, '') IS NULL THEN NULL
-					ELSE NULLIF(mark_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-				END AS mark_price,
-				CASE
-					WHEN NULLIF(oracle_price_raw, '') IS NULL THEN NULL
-					ELSE NULLIF(oracle_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-				END AS oracle_price,
-				CASE
-					WHEN NULLIF(best_bid_price_raw, '') IS NULL THEN NULL
-					ELSE NULLIF(best_bid_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-				END AS best_bid_price,
-				CASE
-					WHEN NULLIF(best_ask_price_raw, '') IS NULL THEN NULL
-					ELSE NULLIF(best_ask_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-				END AS best_ask_price,
-				bids,
-				asks,
-				payload AS raw,
-				updated_at
-			FROM public.dlob_orderbook_l2_latest
-			WHERE is_indicative = FALSE;
-		`);
-
-		await this.pool.query(`
-			CREATE OR REPLACE VIEW public.dlob_l3_latest_projection AS
-			SELECT
-				source,
-				market_name,
-				market_type,
-				market_index,
-				ts,
-				slot,
-				bids,
-				asks,
-				payload AS raw,
-				updated_at
-			FROM public.dlob_orderbook_l3_latest
-			WHERE is_indicative = FALSE;
-		`);
-
-		await this.pool.query(`
-			CREATE OR REPLACE VIEW public.dlob_best_makers_latest_projection AS
-			SELECT
-				source,
-				market_name,
-				market_type,
-				market_index,
-				slot,
-				bids,
-				asks,
-				payload AS raw,
-				updated_at
-			FROM public.dlob_best_makers_latest;
-		`);
-
-		await this.pool.query(`
-			CREATE OR REPLACE VIEW public.dlob_stats_latest_projection AS
-			WITH base AS (
-				SELECT
-					source,
-					market_name,
-					market_type,
-					market_index,
-					ts,
-					slot,
-					CASE
-						WHEN NULLIF(mark_price_raw, '') IS NULL THEN NULL
-						ELSE NULLIF(mark_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-					END AS mark_price,
-					CASE
-						WHEN NULLIF(oracle_price_raw, '') IS NULL THEN NULL
-						ELSE NULLIF(oracle_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-					END AS oracle_price,
-					CASE
-						WHEN NULLIF(best_bid_price_raw, '') IS NULL THEN NULL
-						ELSE NULLIF(best_bid_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-					END AS best_bid_price,
-					CASE
-						WHEN NULLIF(best_ask_price_raw, '') IS NULL THEN NULL
-						ELSE NULLIF(best_ask_price_raw, '')::numeric / ${this.config.pricePrecision}::numeric
-					END AS best_ask_price,
-					bids,
-					asks,
-					payload,
-					updated_at
-				FROM public.dlob_orderbook_l2_latest
-				WHERE is_indicative = FALSE
-			),
-			bid_levels AS (
-				SELECT
-					b.source,
-					b.market_name,
-					COALESCE(
-						SUM((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric),
-						0::numeric
-					) AS depth_bid_base,
-					COALESCE(
-						SUM(
-							((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric) *
-							((NULLIF(level->>'price', '')::numeric) / ${this.config.pricePrecision}::numeric)
-						),
-						0::numeric
-					) AS depth_bid_usd
-				FROM base b
-				LEFT JOIN LATERAL (
-					SELECT level
-					FROM jsonb_array_elements(COALESCE(b.bids, '[]'::jsonb)) WITH ORDINALITY AS e(level, ord)
-					WHERE ord <= ${this.config.statsDepth}
-				) levels ON TRUE
-				GROUP BY b.source, b.market_name
-			),
-			ask_levels AS (
-				SELECT
-					b.source,
-					b.market_name,
-					COALESCE(
-						SUM((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric),
-						0::numeric
-					) AS depth_ask_base,
-					COALESCE(
-						SUM(
-							((NULLIF(level->>'size', '')::numeric) / ${this.config.basePrecision}::numeric) *
-							((NULLIF(level->>'price', '')::numeric) / ${this.config.pricePrecision}::numeric)
-						),
-						0::numeric
-					) AS depth_ask_usd
-				FROM base b
-				LEFT JOIN LATERAL (
-					SELECT level
-					FROM jsonb_array_elements(COALESCE(b.asks, '[]'::jsonb)) WITH ORDINALITY AS e(level, ord)
-					WHERE ord <= ${this.config.statsDepth}
-				) levels ON TRUE
-				GROUP BY b.source, b.market_name
-			)
-			SELECT
-				b.source,
-				b.market_name,
-				b.market_type,
-				b.market_index,
-				b.ts,
-				b.slot,
-				b.mark_price,
-				b.oracle_price,
-				b.best_bid_price,
-				b.best_ask_price,
-				CASE
-					WHEN b.best_bid_price IS NOT NULL AND b.best_ask_price IS NOT NULL
-					THEN (b.best_bid_price + b.best_ask_price) / 2
-					ELSE NULL
-				END AS mid_price,
-				CASE
-					WHEN b.best_bid_price IS NOT NULL AND b.best_ask_price IS NOT NULL
-					THEN b.best_ask_price - b.best_bid_price
-					ELSE NULL
-				END AS spread_abs,
-				CASE
-					WHEN b.best_bid_price IS NOT NULL
-						AND b.best_ask_price IS NOT NULL
-						AND (b.best_bid_price + b.best_ask_price) > 0
-					THEN
-						((b.best_ask_price - b.best_bid_price) / ((b.best_bid_price + b.best_ask_price) / 2)) * 10000
-					ELSE NULL
-				END AS spread_bps,
-				${this.config.statsDepth}::integer AS depth_levels,
-				bids.depth_bid_base,
-				asks.depth_ask_base,
-				bids.depth_bid_usd,
-				asks.depth_ask_usd,
-				CASE
-					WHEN (bids.depth_bid_usd + asks.depth_ask_usd) > 0
-					THEN (bids.depth_bid_usd - asks.depth_ask_usd) / (bids.depth_bid_usd + asks.depth_ask_usd)
-					ELSE NULL
-				END AS imbalance,
-				jsonb_build_object(
-					'canonical_table', 'dlob_orderbook_l2_latest',
-					'projection', 'dlob_stats_latest_projection'
-				) AS raw,
-				b.updated_at
-			FROM base b
-			LEFT JOIN bid_levels bids USING (source, market_name)
-			LEFT JOIN ask_levels asks USING (source, market_name);
-		`);
-	}
-
-	async writeL2Snapshot(
-		l2Snapshot: Record<string, any>,
-		options: { indicative: boolean }
-	): Promise<void> {
-		this.writeChain = this.writeChain
-			.then(() =>
-				this.pool.query(
-					`
-						INSERT INTO public.dlob_orderbook_l2_latest (
-							source,
-							market_name,
-							market_type,
-							market_index,
-							is_indicative,
-							ts,
-							slot,
-							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::jsonb, $15::jsonb, $16::jsonb, $17::jsonb, now()
-						)
-						ON CONFLICT (source, market_name, is_indicative) DO UPDATE SET
-							market_type = EXCLUDED.market_type,
-							market_index = EXCLUDED.market_index,
-							ts = EXCLUDED.ts,
-							slot = EXCLUDED.slot,
-							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 = now()
-					`,
-					[
-						this.config.source,
-						String(l2Snapshot.marketName),
-						String(l2Snapshot.marketType || 'perp'),
-						toIntOrNull(l2Snapshot.marketIndex),
-						options.indicative,
-						toIntOrNull(l2Snapshot.ts),
-						toIntOrNull(l2Snapshot.slot),
-						l2Snapshot.markPrice ?? null,
-						l2Snapshot.oracleData?.price ?? l2Snapshot.oracle ?? null,
-						l2Snapshot.bestBidPrice ?? null,
-						l2Snapshot.bestAskPrice ?? null,
-						l2Snapshot.spreadPct ?? null,
-						l2Snapshot.spreadQuote ?? null,
-						toJson(l2Snapshot.oracleData ?? null),
-						toJson(l2Snapshot.bids ?? null),
-						toJson(l2Snapshot.asks ?? null),
-						toJson(l2Snapshot),
-					]
-				)
-			)
-			.catch((error) => {
-				logger.error('Failed to persist canonical DLOB L2 snapshot', error);
-			});
-
-		return this.writeChain;
-	}
-
-	async writeL3Snapshot(
-		l3Snapshot: Record<string, any>,
-		options: { indicative: boolean }
-	): Promise<void> {
-		this.writeChain = this.writeChain
-			.then(() =>
-				this.pool.query(
-					`
-						INSERT INTO public.dlob_orderbook_l3_latest (
-							source,
-							market_name,
-							market_type,
-							market_index,
-							is_indicative,
-							ts,
-							slot,
-							oracle_data,
-							bids,
-							asks,
-							payload,
-							updated_at
-						)
-						VALUES (
-							$1, $2, $3, $4, $5, $6, $7,
-							$8::jsonb, $9::jsonb, $10::jsonb, $11::jsonb, now()
-						)
-						ON CONFLICT (source, market_name, is_indicative) DO UPDATE SET
-							market_type = EXCLUDED.market_type,
-							market_index = EXCLUDED.market_index,
-							ts = EXCLUDED.ts,
-							slot = EXCLUDED.slot,
-							oracle_data = EXCLUDED.oracle_data,
-							bids = EXCLUDED.bids,
-							asks = EXCLUDED.asks,
-							payload = EXCLUDED.payload,
-							updated_at = now()
-					`,
-					[
-						this.config.source,
-						String(l3Snapshot.marketName),
-						String(l3Snapshot.marketType || 'perp'),
-						toIntOrNull(l3Snapshot.marketIndex),
-						options.indicative,
-						toIntOrNull(l3Snapshot.ts),
-						toIntOrNull(l3Snapshot.slot),
-						toJson(l3Snapshot.oracleData ?? null),
-						toJson(l3Snapshot.bids ?? null),
-						toJson(l3Snapshot.asks ?? null),
-						toJson(l3Snapshot),
-					]
-				)
-			)
-			.catch((error) => {
-				logger.error('Failed to persist canonical DLOB L3 snapshot', error);
-			});
-
-		return this.writeChain;
-	}
-
-	async writeBestMakersSnapshot(snapshot: {
-		marketName: string;
-		marketType: string;
-		marketIndex: number;
-		slot: number | null;
-		bids: string[];
-		asks: string[];
-	}): Promise<void> {
-		this.writeChain = this.writeChain
-			.then(() =>
-				this.pool.query(
-					`
-						INSERT INTO public.dlob_best_makers_latest (
-							source,
-							market_name,
-							market_type,
-							market_index,
-							slot,
-							bids,
-							asks,
-							payload,
-							updated_at
-						)
-						VALUES (
-							$1, $2, $3, $4, $5, $6::jsonb, $7::jsonb, $8::jsonb, now()
-						)
-						ON CONFLICT (source, market_name) DO UPDATE SET
-							market_type = EXCLUDED.market_type,
-							market_index = EXCLUDED.market_index,
-							slot = EXCLUDED.slot,
-							bids = EXCLUDED.bids,
-							asks = EXCLUDED.asks,
-							payload = EXCLUDED.payload,
-							updated_at = now()
-					`,
-					[
-						this.config.source,
-						snapshot.marketName,
-						snapshot.marketType,
-						snapshot.marketIndex,
-						snapshot.slot,
-						toJson(snapshot.bids),
-						toJson(snapshot.asks),
-						toJson(snapshot),
-					]
-				)
-			)
-			.catch((error) => {
-				logger.error('Failed to persist canonical DLOB best makers', error);
-			});
-
-		return this.writeChain;
-	}
-
-	async close(): Promise<void> {
-		await this.writeChain.catch(() => undefined);
-		await this.pool.end();
-	}
-}
-
-export function createPostgresCanonicalStoreFromEnv(): DlobSnapshotStore {
-	if (!envBool('ENABLE_PERSISTENT_STORE', false)) {
-		return new NoopDlobSnapshotStore();
-	}
-
-	const connectionString =
-		process.env.POSTGRES_URL || process.env.DATABASE_URL || undefined;
-
-	if (
-		!connectionString &&
-		(!process.env.PGHOST ||
-			!process.env.PGUSER ||
-			!process.env.PGDATABASE)
-	) {
-		throw new Error(
-			'ENABLE_PERSISTENT_STORE=true requires POSTGRES_URL or PGHOST/PGUSER/PGDATABASE'
-		);
-	}
-
-	return new PostgresCanonicalStore({
-		connectionString,
-		host: process.env.PGHOST,
-		port: envInt('PGPORT', 5432),
-		user: process.env.PGUSER,
-		password: process.env.PGPASSWORD,
-		database: process.env.PGDATABASE,
-		source: process.env.DLOB_SOURCE || 'mevnode',
-		pricePrecision: envInt('PRICE_PRECISION', 1_000_000),
-		basePrecision: envInt('BASE_PRECISION', 1_000_000_000),
-		statsDepth: envInt('PERSISTENT_STATS_DEPTH', 10),
-	});
-}

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(() => {