fix(build): use node 24 for drift sdk

feat(storage): add canonical dlob postgres store
2026-03-13 23:25:43 +01:00 · 2026-03-13 23:10:35 +01:00
22 changed files with 778 additions and 2093 deletions
--- a/12
+++ b/12
@@ -1,21 +1,17 @@
 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 . .

-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
+WORKDIR /app/drift-common/protocol/sdk
 RUN yarn && yarn build

-WORKDIR /app/dlob-server/drift-common/common-ts
+WORKDIR /app/drift-common/common-ts
 RUN yarn && yarn build

-WORKDIR /app/dlob-server
+WORKDIR /app
 RUN yarn && yarn build

 FROM public.ecr.aws/docker/library/node:24-alpine
@@ -23,7 +19,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/dlob-server/lib/ ./lib/
+COPY --from=builder /app/lib/ ./lib/

 ENV NODE_ENV=production
 EXPOSE 9464
--- a/README.md
+++ b/README.md
@@ -82,6 +82,12 @@ 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.
@@ -131,6 +137,30 @@ 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)
--- a/package.json
+++ b/package.json
@@ -32,7 +32,7 @@
 		"express": "^4.18.2",
 		"ioredis": "^5.4.1",
 		"morgan": "^1.10.0",
-		"pg": "^8.12.0",
+		"pg": "^8.13.1",
 		"redis": "^4.6.10",
 		"response-time": "^2.3.2",
 		"rxjs": "^7.8.1",
@@ -71,6 +71,7 @@
 		"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",
--- a/src/athena/testRedisPolling.ts
+++ b/src/athena/testRedisPolling.ts
@@ -1,4 +1,4 @@
-import { RedisClient, HOT_REDIS_KEY_PREFIX } from '../utils/redisClients';
+import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 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: ${HOT_REDIS_KEY_PREFIX}`);
+	logger.info(`   Prefix: ${RedisClientPrefix.DLOB}`);

 	const redisClient = new RedisClient({
-		prefix: HOT_REDIS_KEY_PREFIX,
+		prefix: RedisClientPrefix.DLOB,
 	});

 	try {
--- a/src/dlob-subscriber/DLOBSubscriberIO.ts
+++ b/src/dlob-subscriber/DLOBSubscriberIO.ts
@@ -32,6 +32,7 @@ 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;
@@ -70,6 +71,7 @@ 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;
@@ -79,6 +81,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			env: DriftEnv;
 			redisClient: RedisClient;
 			indicativeQuotesRedisClient?: RedisClient;
+			persistentStore?: DlobSnapshotStore;
 			enableOffloadQueue?: boolean;
 			offloadQueueCounter?: CounterValue;
 			perpMarketInfos: wsMarketInfo[];
@@ -90,6 +93,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		super(config);
 		this.redisClient = config.redisClient;
 		this.indicativeQuotesRedisClient = config.indicativeQuotesRedisClient;
+		this.persistentStore = config.persistentStore;

 		this.killSwitchSlotDiffThreshold =
 			config.killSwitchSlotDiffThreshold || 200;
@@ -470,6 +474,19 @@ 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,
@@ -508,6 +525,23 @@ 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
+						);
+					});
+			}
 		}
 	}

@@ -571,5 +605,18 @@ 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
+					);
+				});
+		}
 	}
 }
--- a/src/dlob-subscriber/OrderSubscriberFiltered.ts
+++ b/src/dlob-subscriber/OrderSubscriberFiltered.ts
@@ -3,7 +3,6 @@ import {
 	OrderSubscriberConfig,
 	UserAccount,
 } from '@drift-labs/sdk';
-import { ReconnectingGrpcOrderSubscription } from './ReconnectingGrpcOrderSubscription';

 export class OrderSubscriberFiltered extends OrderSubscriber {
 	public ignoreList: string[];
@@ -15,20 +14,6 @@ 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(
--- a/src/dlob-subscriber/ReconnectingGrpcOrderSubscription.ts
+++ b/src/dlob-subscriber/ReconnectingGrpcOrderSubscription.ts
@@ -1,614 +0,0 @@
-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;
-		}
-	}
-}
--- a/src/index.ts
+++ b/src/index.ts
@@ -24,6 +24,8 @@ 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';
@@ -51,13 +53,6 @@ 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({
@@ -67,7 +62,16 @@ setGlobalDispatcher(

 require('dotenv').config();

-const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT);
+const envClients = [];
+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;
@@ -282,7 +286,7 @@ const main = async (): Promise<void> => {
 				.find((client) => {
 					return (
 						client.forceGetClient().options.keyPrefix ===
-						ALL_REDIS_KEY_PREFIX
+						RedisClientPrefix.DLOB_HELIUS
 					);
 				})
 				.getRaw(`priorityFees_${marketType}_${marketIndex}`);
@@ -328,7 +332,7 @@ const main = async (): Promise<void> => {
 						.find(
 							(client) =>
 								client.forceGetClient().options.keyPrefix ===
-								ALL_REDIS_KEY_PREFIX
+								RedisClientPrefix.DLOB_HELIUS
 						)
 						.getRaw(
 							`priorityFees_${normedParam['marketType']}_${normedParam['marketIndex']}`
@@ -519,7 +523,7 @@ const main = async (): Promise<void> => {

 			const redisClient = redisClients.find(
 				(client) =>
-					client.forceGetClient().options.keyPrefix === HOT_REDIS_KEY_PREFIX
+					client.forceGetClient().options.keyPrefix === RedisClientPrefix.DLOB
 			);

 			const redisResponseGainers = await redisClient.getRaw(
--- a/src/publishers/dlobPublisher.ts
+++ b/src/publishers/dlobPublisher.ts
@@ -19,6 +19,8 @@ import {
 	ONE,
 	WebSocketAccountSubscriberV2,
 } from '@drift-labs/sdk';
+import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
+
 import { logger, setLogLevel } from '../utils/logger';
 import {
 	SubscriberLookup,
@@ -45,28 +47,11 @@ import {
 	FillQualityAnalyticsRepository,
 	TakerFillVsOracleBpsRedisResult,
 } from '../athena/repositories/fillQualityAnalytics';
-import {
-	RedisClient,
-	HOT_REDIS_CLIENT,
-	getRedisClientPrefix,
-} from '../utils/redisClients';
-
-function envInt(name: string, fallback: number): number {
-	const raw = process.env[name];
-	if (raw == null) return fallback;
-	const parsed = parseInt(raw, 10);
-	return Number.isFinite(parsed) ? parsed : fallback;
-}
+import { createPostgresCanonicalStoreFromEnv } from '../storage/postgresCanonicalStore';

 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),
 	})
 );

@@ -78,8 +63,7 @@ const metricsPort = process.env.METRICS_PORT
 	? parseInt(process.env.METRICS_PORT)
 	: 9464;

-const REDIS_CLIENT = process.env.REDIS_CLIENT || HOT_REDIS_CLIENT;
-const REDIS_PREFIX = getRedisClientPrefix(REDIS_CLIENT);
+const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB';

 // Set up express for health checks
 const app = express();
@@ -188,11 +172,7 @@ 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 ? `${token.slice(0, 4)}…${token.slice(-4)}` : '(missing)'
-	}`
-);
+logger.info(`GRPC Token:    ${token}`);
 logger.info(
 	`useOrderSubscriber: ${useOrderSubscriber}, useWebsocket: ${useWebsocket}, useGrpc: ${useGrpc}`
 );
@@ -401,13 +381,23 @@ const main = async () => {
 	const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);

 	const redisClient = new RedisClient({
-		prefix: REDIS_PREFIX,
+		prefix: RedisClientPrefix[REDIS_CLIENT],
 	});
 	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,
@@ -494,9 +484,6 @@ 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);
@@ -606,6 +593,7 @@ const main = async () => {
 		spotMarketSubscribers: MARKET_SUBSCRIBERS,
 		perpMarketInfos,
 		spotMarketInfos,
+		persistentStore,
 		killSwitchSlotDiffThreshold: KILLSWITCH_SLOT_DIFF_THRESHOLD,
 		protectedMakerView: false,
 	});
@@ -662,21 +650,21 @@ const main = async () => {
 				marketType: 'perp',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: REDIS_PREFIX,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 			});
 			oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
 				marketIndex: market.marketIndex,
 				marketType: 'perp',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: REDIS_PREFIX,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 			});
 			marketSlotGauge.setLatestValue(marketAccount.slot, {
 				marketIndex: market.marketIndex,
 				marketType: 'perp',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: REDIS_PREFIX,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 			});
 		});
 		spotMarketInfos.forEach((market) => {
@@ -692,21 +680,21 @@ const main = async () => {
 				marketType: 'spot',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: REDIS_PREFIX,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 			});
 			oracleSlotGauge.setLatestValue(oracleDataAndSlot.slot.toNumber(), {
 				marketIndex: market.marketIndex,
 				marketType: 'spot',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: REDIS_PREFIX,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 			});
 			marketSlotGauge.setLatestValue(marketAccount.slot, {
 				marketIndex: market.marketIndex,
 				marketType: 'spot',
 				marketName: market.marketName,
 				redisClient: REDIS_CLIENT,
-				redisPrefix: REDIS_PREFIX,
+				redisPrefix: RedisClientPrefix[REDIS_CLIENT],
 			});
 		});
 	}, 10_000);
--- a/src/publishers/priorityFeesPublisher.ts
+++ b/src/publishers/priorityFeesPublisher.ts
@@ -10,11 +10,12 @@ 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({
@@ -26,7 +27,7 @@ require('dotenv').config();
 const stateCommitment: Commitment = 'confirmed';
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
 const commitHash = process.env.COMMIT;
-const redisClientPrefix = ALL_REDIS_KEY_PREFIX;
+const redisClientPrefix = RedisClientPrefix.DLOB_HELIUS;
 // Set up express for health checks
 const app = express();

--- a/src/publishers/tradesPublisher.ts
+++ b/src/publishers/tradesPublisher.ts
@@ -20,15 +20,12 @@ 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({
@@ -39,9 +36,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 || HOT_REDIS_CLIENT;
+const REDIS_CLIENT = process.env.REDIS_CLIENT || 'DLOB';
 console.log('Redis Clients:', REDIS_CLIENT);
-const redisClientPrefix = getRedisClientPrefix(REDIS_CLIENT);
+const redisClientPrefix = RedisClientPrefix[REDIS_CLIENT];
 //@ts-ignore
 const sdkConfig = initialize({ env: process.env.ENV });

--- a/src/scripts/dlobAllRedisToPostgresDerivedWriter.ts
+++ b/src/scripts/dlobAllRedisToPostgresDerivedWriter.ts
@@ -1,284 +0,0 @@
-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);
-});
--- a/src/scripts/dlobHotPgCommon.ts
+++ b/src/scripts/dlobHotPgCommon.ts
@@ -1,321 +0,0 @@
-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,
-	};
-}
--- a/src/scripts/dlobHotPostgresToPostgresDerivedWriter.ts
+++ b/src/scripts/dlobHotPostgresToPostgresDerivedWriter.ts
@@ -1,343 +0,0 @@
-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);
-});
--- a/src/scripts/dlobHotRedisToPostgresRawWriter.ts
+++ b/src/scripts/dlobHotRedisToPostgresRawWriter.ts
@@ -1,280 +0,0 @@
-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);
-});
--- a/src/scripts/initPersistentStore.ts
+++ b/src/scripts/initPersistentStore.ts
@@ -0,0 +1,16 @@
+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);
+});
--- a/src/scripts/publishUnsettledPnlUsers.ts
+++ b/src/scripts/publishUnsettledPnlUsers.ts
@@ -1,3 +1,4 @@
+import { RedisClient, RedisClientPrefix } from '@drift/common/clients';
 import { sleep } from '@drift/common';
 import {
 	BigNum,
@@ -16,11 +17,6 @@ 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();
@@ -65,7 +61,7 @@ const userMapRedisClient = new RedisClient({
 });

 const dlobRedisClient = new RedisClient({
-	prefix: HOT_REDIS_KEY_PREFIX,
+	prefix: RedisClientPrefix.DLOB,
 });

 const main = async () => {
--- a/src/serverLite.ts
+++ b/src/serverLite.ts
@@ -12,8 +12,9 @@ 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';
@@ -21,7 +22,6 @@ 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,7 +32,16 @@ setGlobalDispatcher(
 require('dotenv').config();

 // Reading in Redis env vars
-const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT);
+const envClients = [];
+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;
@@ -194,112 +203,6 @@ 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}`
--- a/src/storage/dlobSnapshotStore.ts
+++ b/src/storage/dlobSnapshotStore.ts
@@ -0,0 +1,45 @@
+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> {}
+}
--- a/src/storage/postgresCanonicalStore.ts
+++ b/src/storage/postgresCanonicalStore.ts
@@ -0,0 +1,560 @@
+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),
+	});
+}
--- a/src/utils/redisClients.ts
+++ b/src/utils/redisClients.ts
@@ -1,47 +0,0 @@
-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];
-}
--- a/src/wsConnectionManager.ts
+++ b/src/wsConnectionManager.ts
@@ -5,12 +5,8 @@ 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
@@ -215,7 +211,16 @@ console.log(`WS LISTENER PORT : ${WS_PORT}`);

 const MAX_BUFFERED_AMOUNT = 300000;

-const REDIS_CLIENTS = parseRedisClients(process.env.REDIS_CLIENT);
+const envClients = [];
+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');
@@ -433,7 +438,7 @@ async function main() {
 						redisClients[0]
 							.subscribe(
 								redisChannel.includes('priorityFees')
-									? `${ALL_REDIS_KEY_PREFIX}${redisChannel}`
+									? `dlob-helius:${redisChannel}`
 									: `${redisClients[0].getPrefix()}${redisChannel}`
 							)
 							.then(() => {
Author	SHA1	Message	Date
mpabi	5d55631b75	fix(build): use node 24 for drift sdk	2026-03-13 23:25:43 +01:00
mpabi	1ea14833ba	feat(storage): add canonical dlob postgres store	2026-03-13 23:10:35 +01:00