diff --git a/README.md b/README.md index 4066f58..e27bfb3 100644 --- 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) diff --git a/package.json b/package.json index 440a8af..30a25a7 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "express": "^4.18.2", "ioredis": "^5.4.1", "morgan": "^1.10.0", + "pg": "^8.13.1", "redis": "^4.6.10", "response-time": "^2.3.2", "rxjs": "^7.8.1", @@ -70,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", diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index 31cc782..7b3042c 100644 --- 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>; redisClient: RedisClient; indicativeQuotesRedisClient?: RedisClient; + persistentStore?: DlobSnapshotStore; public killSwitchSlotDiffThreshold: number; public offloadQueue?: ReturnType; 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 + ); + }); + } } } diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index cf91ae0..88e6a90 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -47,6 +47,7 @@ import { FillQualityAnalyticsRepository, TakerFillVsOracleBpsRedisResult, } from '../athena/repositories/fillQualityAnalytics'; +import { createPostgresCanonicalStoreFromEnv } from '../storage/postgresCanonicalStore'; setGlobalDispatcher( new Agent({ @@ -383,10 +384,20 @@ const main = async () => { 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, @@ -582,6 +593,7 @@ const main = async () => { spotMarketSubscribers: MARKET_SUBSCRIBERS, perpMarketInfos, spotMarketInfos, + persistentStore, killSwitchSlotDiffThreshold: KILLSWITCH_SLOT_DIFF_THRESHOLD, protectedMakerView: false, }); diff --git a/src/scripts/initPersistentStore.ts b/src/scripts/initPersistentStore.ts new file mode 100644 index 0000000..ca4456b --- /dev/null +++ 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); +}); diff --git a/src/storage/dlobSnapshotStore.ts b/src/storage/dlobSnapshotStore.ts new file mode 100644 index 0000000..3bc3bb7 --- /dev/null +++ b/src/storage/dlobSnapshotStore.ts @@ -0,0 +1,45 @@ +export interface DlobSnapshotStore { + init(): Promise; + writeL2Snapshot( + l2Snapshot: Record, + options: { indicative: boolean } + ): Promise; + writeL3Snapshot( + l3Snapshot: Record, + options: { indicative: boolean } + ): Promise; + writeBestMakersSnapshot(snapshot: { + marketName: string; + marketType: string; + marketIndex: number; + slot: number | null; + bids: string[]; + asks: string[]; + }): Promise; + close(): Promise; +} + +export class NoopDlobSnapshotStore implements DlobSnapshotStore { + async init(): Promise {} + + async writeL2Snapshot( + _l2Snapshot: Record, + _options: { indicative: boolean } + ): Promise {} + + async writeL3Snapshot( + _l3Snapshot: Record, + _options: { indicative: boolean } + ): Promise {} + + async writeBestMakersSnapshot(_snapshot: { + marketName: string; + marketType: string; + marketIndex: number; + slot: number | null; + bids: string[]; + asks: string[]; + }): Promise {} + + async close(): Promise {} +} diff --git a/src/storage/postgresCanonicalStore.ts b/src/storage/postgresCanonicalStore.ts new file mode 100644 index 0000000..a2a7c40 --- /dev/null +++ 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 = 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 { + 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, + options: { indicative: boolean } + ): Promise { + 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, + options: { indicative: boolean } + ): Promise { + 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 { + 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 { + 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), + }); +}