Nv/optimize slinky adapter

indexer/packages/kafka/__tests__/batch-kafka-producer.test.ts

@@ -1,16 +1,18 @@
 import { KafkaTopics } from '../src';
 import { BatchKafkaProducer, ProducerMessage } from '../src/batch-kafka-producer';
 import { producer } from '../src/producer';
+import { IHeaders } from 'kafkajs';
 import _ from 'lodash';
 
 interface TestMessage {
   key?: string,
   value: string,
+  headers?: IHeaders,
 }
 
 function testMessage2ProducerMessage(data: TestMessage): ProducerMessage {
   const key: Buffer | undefined = data.key === undefined ? undefined : Buffer.from(data.key);
-  return { key, value: Buffer.from(data.value) };
+  return { key, value: Buffer.from(data.value), headers: data.headers };
 }
 
 function testMessage2ProducerMessages(data: TestMessage[]): ProducerMessage[] {
@@ -35,9 +37,9 @@ describe('batch-kafka-producer', () => {
     [
       'will send key if key is not undefined',
       5,
-      [{ key: '1', value: 'a' }, { key: '2', value: 'b' }, { key: '3', value: 'c' }],
+      [{ key: '1', value: 'a' }, { key: '2', value: 'b' }, { key: '3', value: 'c', headers: { timestamp: 'value' } }],
       [[{ key: '1', value: 'a' }, { key: '2', value: 'b' }]],
-      [{ key: '3', value: 'c' }],
+      [{ key: '3', value: 'c', headers: { timestamp: 'value' } }],
     ],
     [
       'will not send message until the batch size is reached',
@@ -104,7 +106,9 @@ describe('batch-kafka-producer', () => {
 
     for (const msg of messages) {
       const key: Buffer | undefined = msg.key === undefined ? undefined : Buffer.from(msg.key);
-      batchProducer.addMessageAndMaybeFlush({ value: Buffer.from(msg.value), key });
+      batchProducer.addMessageAndMaybeFlush(
+        { value: Buffer.from(msg.value), key, headers: msg.headers },
+      );
     }
 
     expect(producerSendMock.mock.calls).toHaveLength(expectedMessagesPerCall.length);

indexer/packages/kafka/src/batch-kafka-producer.ts

@@ -1,5 +1,5 @@
 import { logger } from '@dydxprotocol-indexer/base';
-import { Producer, RecordMetadata } from 'kafkajs';
+import { IHeaders, Producer, RecordMetadata } from 'kafkajs';
 import _ from 'lodash';
 
 import { KafkaTopics } from './types';
@@ -10,6 +10,7 @@ import { KafkaTopics } from './types';
 export type ProducerMessage = {
   key?: Buffer,
   value: Buffer,
+  headers?: IHeaders,
 };
 
 /**
@@ -52,7 +53,7 @@ export class BatchKafkaProducer {
     if (this.currentSize + msgBuffer.byteLength + keyByteLength > this.maxBatchSizeBytes) {
       this.sendBatch();
     }
-    this.producerMessages.push({ key: message.key, value: msgBuffer });
+    this.producerMessages.push({ key: message.key, value: msgBuffer, headers: message.headers });
     this.currentSize += msgBuffer.byteLength;
     this.currentSize += keyByteLength;
   }

indexer/packages/postgres/__tests__/helpers/constants.ts

@@ -87,6 +87,20 @@ export const defaultSubaccount3: SubaccountCreateObject = {
   updatedAtHeight: createdHeight,
 };
 
+export const isolatedSubaccount: SubaccountCreateObject = {
+  address: defaultAddress,
+  subaccountNumber: 128,
+  updatedAt: createdDateTime.toISO(),
+  updatedAtHeight: createdHeight,
+};
+
+export const isolatedSubaccount2: SubaccountCreateObject = {
+  address: defaultAddress,
+  subaccountNumber: 256,
+  updatedAt: createdDateTime.toISO(),
+  updatedAtHeight: createdHeight,
+};
+
 export const defaultWalletAddress: string = 'defaultWalletAddress';
 
 export const defaultSubaccountId: string = SubaccountTable.uuid(
@@ -101,6 +115,14 @@ export const defaultSubaccountId3: string = SubaccountTable.uuid(
   defaultAddress,
   defaultSubaccount3.subaccountNumber,
 );
+export const isolatedSubaccountId: string = SubaccountTable.uuid(
+  defaultAddress,
+  isolatedSubaccount.subaccountNumber,
+);
+export const isolatedSubaccountId2: string = SubaccountTable.uuid(
+  defaultAddress,
+  isolatedSubaccount2.subaccountNumber,
+);
 
 // ============== Wallets ==============
 export const defaultWallet: WalletCreateObject = {
@@ -212,6 +234,42 @@ export const defaultPerpetualMarket3: PerpetualMarketCreateObject = {
   liquidityTierId: 0,
 };
 
+export const isolatedPerpetualMarket: PerpetualMarketCreateObject = {
+  id: '3',
+  clobPairId: '4',
+  ticker: 'ISO-USD',
+  marketId: 3,
+  status: PerpetualMarketStatus.ACTIVE,
+  priceChange24H: '0.000000001',
+  volume24H: '10000000',
+  trades24H: 200,
+  nextFundingRate: '1.2',
+  openInterest: '40000',
+  quantumConversionExponent: -16,
+  atomicResolution: -2,
+  subticksPerTick: 10,
+  stepBaseQuantums: 1,
+  liquidityTierId: 0,
+};
+
+export const isolatedPerpetualMarket2: PerpetualMarketCreateObject = {
+  id: '4',
+  clobPairId: '5',
+  ticker: 'ISO2-USD',
+  marketId: 4,
+  status: PerpetualMarketStatus.ACTIVE,
+  priceChange24H: '0.000000001',
+  volume24H: '10000000',
+  trades24H: 200,
+  nextFundingRate: '1.2',
+  openInterest: '40000',
+  quantumConversionExponent: -16,
+  atomicResolution: -2,
+  subticksPerTick: 10,
+  stepBaseQuantums: 1,
+  liquidityTierId: 0,
+};
+
 // ============== Orders ==============
 
 export const defaultOrder: OrderCreateObject = {
@@ -233,6 +291,25 @@ export const defaultOrder: OrderCreateObject = {
   updatedAtHeight: '1',
 };
 
+export const isolatedMarketOrder: OrderCreateObject = {
+  subaccountId: isolatedSubaccountId,
+  clientId: '1',
+  clobPairId: '4',
+  side: OrderSide.BUY,
+  size: '25',
+  totalFilled: '0',
+  price: '20000',
+  type: OrderType.LIMIT,
+  status: OrderStatus.OPEN,
+  timeInForce: TimeInForce.FOK,
+  reduceOnly: false,
+  goodTilBlock: '100',
+  orderFlags: ORDER_FLAG_SHORT_TERM.toString(),
+  clientMetadata: '0',
+  updatedAt: '2023-01-22T00:00:00.000Z',
+  updatedAtHeight: '1',
+};
+
 export const defaultOrderGoodTilBlockTime: OrderCreateObject = {
   ...defaultOrder,
   clientId: '2',
@@ -257,6 +334,13 @@ export const defaultOrderId: string = OrderTable.uuid(
   defaultOrder.orderFlags,
 );
 
+export const isolatedMarketOrderId: string = OrderTable.uuid(
+  isolatedMarketOrder.subaccountId,
+  isolatedMarketOrder.clientId,
+  isolatedMarketOrder.clobPairId,
+  isolatedMarketOrder.orderFlags,
+);
+
 export const defaultOrderGoodTilBlockTimeId: string = OrderTable.uuid(
   defaultOrderGoodTilBlockTime.subaccountId,
   defaultOrderGoodTilBlockTime.clientId,
@@ -381,6 +465,42 @@ export const defaultFill: FillCreateObject = {
   fee: '1.1',
 };
 
+export const isolatedMarketFill: FillCreateObject = {
+  subaccountId: isolatedSubaccountId,
+  side: OrderSide.BUY,
+  liquidity: Liquidity.TAKER,
+  type: FillType.LIMIT,
+  clobPairId: '4',
+  orderId: isolatedMarketOrderId,
+  size: '10',
+  price: '20000',
+  quoteAmount: '200000',
+  eventId: defaultTendermintEventId2,
+  transactionHash: '', // TODO: Add a real transaction Hash
+  createdAt: createdDateTime.toISO(),
+  createdAtHeight: createdHeight,
+  clientMetadata: '0',
+  fee: '1.1',
+};
+
+export const isolatedMarketFill2: FillCreateObject = {
+  subaccountId: isolatedSubaccountId2,
+  side: OrderSide.BUY,
+  liquidity: Liquidity.TAKER,
+  type: FillType.LIMIT,
+  clobPairId: '4',
+  orderId: isolatedMarketOrderId,
+  size: '10',
+  price: '20000',
+  quoteAmount: '200000',
+  eventId: defaultTendermintEventId3,
+  transactionHash: '', // TODO: Add a real transaction Hash
+  createdAt: createdDateTime.toISO(),
+  createdAtHeight: createdHeight,
+  clientMetadata: '0',
+  fee: '1.1',
+};
+
 // ============== Transfers ==============
 
 export const defaultTransfer: TransferCreateObject = {
@@ -480,6 +600,22 @@ export const defaultMarket3: MarketCreateObject = {
   oraclePrice: '0.000000065',
 };
 
+export const isolatedMarket: MarketCreateObject = {
+  id: 3,
+  pair: 'ISO-USD',
+  exponent: -12,
+  minPriceChangePpm: 50,
+  oraclePrice: '0.000000075',
+};
+
+export const isolatedMarket2: MarketCreateObject = {
+  id: 4,
+  pair: 'ISO2-USD',
+  exponent: -12,
+  minPriceChangePpm: 50,
+  oraclePrice: '0.000000085',
+};
+
 // ============== LiquidityTiers ==============
 
 export const defaultLiquidityTier: LiquidityTiersCreateObject = {

indexer/packages/postgres/__tests__/helpers/mock-generators.ts

@@ -27,17 +27,27 @@ import {
   defaultTendermintEvent3,
   defaultTendermintEvent4,
   defaultWallet,
+  isolatedMarket,
+  isolatedMarket2,
+  isolatedPerpetualMarket,
+  isolatedPerpetualMarket2,
+  isolatedSubaccount,
+  isolatedSubaccount2,
 } from './constants';
 
 export async function seedData() {
   await Promise.all([
     SubaccountTable.create(defaultSubaccount),
     SubaccountTable.create(defaultSubaccount2),
+    SubaccountTable.create(isolatedSubaccount),
+    SubaccountTable.create(isolatedSubaccount2),
   ]);
   await Promise.all([
     MarketTable.create(defaultMarket),
     MarketTable.create(defaultMarket2),
     MarketTable.create(defaultMarket3),
+    MarketTable.create(isolatedMarket),
+    MarketTable.create(isolatedMarket2),
   ]);
   await Promise.all([
     LiquidityTiersTable.create(defaultLiquidityTier),
@@ -47,6 +57,8 @@ export async function seedData() {
     PerpetualMarketTable.create(defaultPerpetualMarket),
     PerpetualMarketTable.create(defaultPerpetualMarket2),
     PerpetualMarketTable.create(defaultPerpetualMarket3),
+    PerpetualMarketTable.create(isolatedPerpetualMarket),
+    PerpetualMarketTable.create(isolatedPerpetualMarket2),
   ]);
   await Promise.all([
     BlockTable.create(defaultBlock),

indexer/packages/postgres/__tests__/stores/funding-index-updates-table.test.ts

@@ -8,7 +8,6 @@ import {
   defaultFundingIndexUpdateId,
   defaultPerpetualMarket,
   defaultPerpetualMarket2,
-  defaultPerpetualMarket3,
   defaultTendermintEventId2,
   defaultTendermintEventId3,
 } from '../helpers/constants';
@@ -212,11 +211,10 @@ describe('funding index update store', () => {
         '3',
       );
 
-    expect(fundingIndexMap).toEqual({
-      [defaultFundingIndexUpdate.perpetualId]: Big(defaultFundingIndexUpdate.fundingIndex),
-      [fundingIndexUpdates3.perpetualId]: Big(fundingIndexUpdates3.fundingIndex),
-      [defaultPerpetualMarket3.id]: Big(0),
-    });
+    expect(fundingIndexMap[defaultFundingIndexUpdate.perpetualId])
+      .toEqual(Big(defaultFundingIndexUpdate.fundingIndex));
+    expect(fundingIndexMap[fundingIndexUpdates3.perpetualId])
+      .toEqual(Big(fundingIndexUpdates3.fundingIndex));
   });
 
   it('Gets default funding index of 0 in funding index map if no funding indexes', async () => {
@@ -225,11 +223,8 @@ describe('funding index update store', () => {
         '3',
       );
 
-    expect(fundingIndexMap).toEqual({
-      [defaultPerpetualMarket.id]: Big(0),
-      [defaultPerpetualMarket2.id]: Big(0),
-      [defaultPerpetualMarket3.id]: Big(0),
-    });
+    expect(fundingIndexMap[defaultPerpetualMarket.id]).toEqual(Big(0));
+    expect(fundingIndexMap[defaultPerpetualMarket2.id]).toEqual(Big(0));
   });
 
   it(
@@ -242,11 +237,9 @@ describe('funding index update store', () => {
           '3',
         );
 
-      expect(fundingIndexMap).toEqual({
-        [defaultPerpetualMarket.id]: Big(defaultFundingIndexUpdate.fundingIndex),
-        [defaultPerpetualMarket2.id]: Big(0),
-        [defaultPerpetualMarket3.id]: Big(0),
-      });
+      expect(fundingIndexMap[defaultPerpetualMarket.id])
+        .toEqual(Big(defaultFundingIndexUpdate.fundingIndex));
+      expect(fundingIndexMap[defaultPerpetualMarket2.id]).toEqual(Big(0));
     },
   );
 });