[IND-288] send final message per order id for order fills, with aggregated fill sfrom the entire block

indexer/services/ender/__tests__/helpers/kafka-publisher-helpers.ts

@@ -2,7 +2,7 @@ import { TRADES_WEBSOCKET_MESSAGE_VERSION, KafkaTopics } from '@dydxprotocol-ind
 import { testConstants, TradeContent, TradeMessageContents } from '@dydxprotocol-indexer/postgres';
 import { TradeMessage } from '@dydxprotocol-indexer/v4-protos';
 
-import { ConsolidatedKafkaEvent, SingleTradeMessage } from '../../src/lib/types';
+import { AnnotatedSubaccountMessage, ConsolidatedKafkaEvent, SingleTradeMessage } from '../../src/lib/types';
 
 export function contentToTradeMessage(
   tradeContent: TradeContent,
@@ -40,3 +40,12 @@ export function createConsolidatedKafkaEventFromTrade(
     message: trade,
   };
 }
+
+export function createConsolidatedKafkaEventFromSubaccount(
+  subaccount: AnnotatedSubaccountMessage,
+): ConsolidatedKafkaEvent {
+  return {
+    topic: KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS,
+    message: subaccount,
+  };
+}

indexer/services/ender/__tests__/lib/kafka-publisher.test.ts

@@ -1,20 +1,30 @@
 import {
-  producer,
-  TRADES_WEBSOCKET_MESSAGE_VERSION,
-  KafkaTopics,
-  ProducerMessage,
+  KafkaTopics, producer, ProducerMessage, TRADES_WEBSOCKET_MESSAGE_VERSION,
 } from '@dydxprotocol-indexer/kafka';
 import { testConstants, TradeContent, TradeMessageContents } from '@dydxprotocol-indexer/postgres';
 import { SubaccountMessage, TradeMessage } from '@dydxprotocol-indexer/v4-protos';
 import Big from 'big.js';
 import _ from 'lodash';
-import { ConsolidatedKafkaEvent, SingleTradeMessage } from '../../src/lib/types';
+import {
+  AnnotatedSubaccountMessage,
+  ConsolidatedKafkaEvent,
+  convertToSubaccountMessage,
+  SingleTradeMessage,
+} from '../../src/lib/types';
 
 import { KafkaPublisher } from '../../src/lib/kafka-publisher';
 import {
-  defaultSubaccountMessage, defaultTradeContent, defaultTradeMessage, defaultTradeKafkaEvent,
+  defaultSubaccountMessage,
+  defaultTradeContent,
+  defaultTradeKafkaEvent,
+  defaultTradeMessage,
 } from '../helpers/constants';
-import { contentToSingleTradeMessage, contentToTradeMessage, createConsolidatedKafkaEventFromTrade } from '../helpers/kafka-publisher-helpers';
+import {
+  contentToSingleTradeMessage,
+  contentToTradeMessage,
+  createConsolidatedKafkaEventFromSubaccount,
+  createConsolidatedKafkaEventFromTrade,
+} from '../helpers/kafka-publisher-helpers';
 
 describe('kafka-publisher', () => {
   let producerSendMock: jest.SpyInstance;
@@ -139,11 +149,164 @@ describe('kafka-publisher', () => {
         consolidatedBeforeTrade,
       ]);
 
-      publisher.sortTradeEvents();
+      publisher.sortEvents(KafkaTopics.TO_WEBSOCKETS_TRADES);
       expect(publisher.tradeMessages).toEqual([beforeTrade, trade, afterTrade]);
     });
   });
 
+  describe('sortSubaccountEvents', () => {
+    const subaccount: AnnotatedSubaccountMessage = defaultSubaccountMessage;
+    const consolidatedSubaccount:
+    ConsolidatedKafkaEvent = createConsolidatedKafkaEventFromSubaccount(subaccount);
+    it.each([
+      [
+        'blockHeight',
+        {
+          ...subaccount,
+          blockHeight: Big(subaccount.blockHeight).minus(1).toString(),
+        },
+        {
+          ...subaccount,
+          blockHeight: Big(subaccount.blockHeight).plus(1).toString(),
+        },
+      ],
+      [
+        'transactionIndex',
+        {
+          ...subaccount,
+          transactionIndex: subaccount.transactionIndex - 1,
+        },
+        {
+          ...subaccount,
+          transactionIndex: subaccount.transactionIndex + 1,
+        },
+      ],
+      [
+        'eventIndex',
+        {
+          ...subaccount,
+          eventIndex: subaccount.eventIndex - 1,
+        },
+        {
+          ...subaccount,
+          eventIndex: subaccount.eventIndex + 1,
+        },
+      ],
+    ])('successfully subaccounts events by %s', (
+      _field: string,
+      beforeSubaccount: AnnotatedSubaccountMessage,
+      afterSubaccount: AnnotatedSubaccountMessage,
+    ) => {
+      const publisher: KafkaPublisher = new KafkaPublisher();
+      const consolidatedBeforeSubaccount:
+      ConsolidatedKafkaEvent = createConsolidatedKafkaEventFromSubaccount(
+        beforeSubaccount,
+      );
+      const consolidatedAfterSubaccount:
+      ConsolidatedKafkaEvent = createConsolidatedKafkaEventFromSubaccount(
+        afterSubaccount,
+      );
+
+      publisher.addEvents([
+        consolidatedAfterSubaccount,
+        consolidatedSubaccount,
+        consolidatedBeforeSubaccount,
+      ]);
+
+      publisher.sortEvents(KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS);
+      expect(publisher.subaccountMessages).toEqual([beforeSubaccount, subaccount, afterSubaccount]);
+    });
+  });
+
+  describe('retainLastFillEventsForSubaccountMessages', () => {
+    it('successfully retains the last fill event per order id and sorts messages', async () => {
+      const publisher: KafkaPublisher = new KafkaPublisher();
+
+      // over-written by message 3.
+      const message1: AnnotatedSubaccountMessage = {
+        blockHeight: '1',
+        transactionIndex: 1,
+        eventIndex: 1,
+        contents: 'Message 1',
+        subaccountId: {
+          owner: 'owner1',
+          number: 0,
+        },
+        version: '1',
+        orderId: 'order1',
+        isFill: true,
+      };
+
+      const message2: AnnotatedSubaccountMessage = {
+        blockHeight: '1',
+        transactionIndex: 2,
+        eventIndex: 2,
+        contents: 'Message 2',
+        subaccountId: {
+          owner: 'owner2',
+          number: 0,
+        },
+        version: '1',
+        orderId: 'order2',
+        isFill: true,
+      };
+
+      const message3: AnnotatedSubaccountMessage = {
+        blockHeight: '1',
+        transactionIndex: 3,
+        eventIndex: 3,
+        contents: 'Message 3',
+        subaccountId: {
+          owner: 'owner3',
+          number: 0,
+        },
+        version: '1',
+        orderId: 'order1',
+        isFill: true,
+      };
+
+      // non-fill subaccount message.
+      const message4: AnnotatedSubaccountMessage = {
+        blockHeight: '1',
+        transactionIndex: 3,
+        eventIndex: 4,
+        contents: 'Message 3',
+        subaccountId: {
+          owner: 'owner3',
+          number: 0,
+        },
+        version: '4',
+      };
+
+      publisher.addEvents([
+        createConsolidatedKafkaEventFromSubaccount(message1),
+        createConsolidatedKafkaEventFromSubaccount(message2),
+        createConsolidatedKafkaEventFromSubaccount(message3),
+        createConsolidatedKafkaEventFromSubaccount(message4),
+      ]);
+
+      publisher.retainLastFillEventsForSubaccountMessages();
+      const expectedMsgs: SubaccountMessage[] = [
+        convertToSubaccountMessage(message2),
+        convertToSubaccountMessage(message3),
+        convertToSubaccountMessage(message4),
+      ];
+      expect(publisher.subaccountMessages).toEqual(expectedMsgs);
+
+      await publisher.publish();
+
+      expect(producerSendMock).toHaveBeenCalledTimes(1);
+      expect(producerSendMock).toHaveBeenCalledWith({
+        topic: KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS,
+        messages: _.map(expectedMsgs, (message: SubaccountMessage) => {
+          return {
+            value: Buffer.from(Uint8Array.from(SubaccountMessage.encode(message).finish())),
+          };
+        }),
+      });
+    });
+  });
+
   describe('groupKafkaTradesByClobPairId', () => {
     it('successfully groups kafka trade messages', () => {
       const kafkaPublisher: KafkaPublisher = new KafkaPublisher();

indexer/services/ender/src/handlers/handler.ts

@@ -16,13 +16,14 @@ import {
   MarketMessage,
   OffChainUpdateV1,
   SubaccountId,
-  SubaccountMessage,
 } from '@dydxprotocol-indexer/v4-protos';
 import { DateTime } from 'luxon';
 
 import config from '../config';
 import { indexerTendermintEventToTransactionIndex } from '../lib/helper';
-import { ConsolidatedKafkaEvent, EventMessage, SingleTradeMessage } from '../lib/types';
+import {
+  AnnotatedSubaccountMessage, ConsolidatedKafkaEvent, EventMessage, SingleTradeMessage,
+} from '../lib/types';
 
 export type HandlerInitializer = new (
   block: IndexerTendermintBlock,
@@ -103,9 +104,11 @@ export abstract class Handler<T> {
   protected generateConsolidatedSubaccountKafkaEvent(
     contents: string,
     subaccountId: SubaccountId,
+    orderId?: string,
+    isFill?: boolean,
   ): ConsolidatedKafkaEvent {
     stats.increment(`${config.SERVICE_NAME}.create_subaccount_kafka_event`, 1);
-    const subaccountMessage: SubaccountMessage = {
+    const subaccountMessage: AnnotatedSubaccountMessage = {
       blockHeight: this.block.height.toString(),
       transactionIndex: indexerTendermintEventToTransactionIndex(
         this.indexerTendermintEvent,
@@ -114,6 +117,8 @@ export abstract class Handler<T> {
       contents,
       subaccountId,
       version: SUBACCOUNTS_WEBSOCKET_MESSAGE_VERSION,
+      orderId,
+      isFill,
     };
 
     return {

indexer/services/ender/src/handlers/order-fills/abstract-order-fill-handler.ts

@@ -407,6 +407,8 @@ export abstract class AbstractOrderFillHandler<T> extends Handler<T> {
     return this.generateConsolidatedSubaccountKafkaEvent(
       JSON.stringify(message),
       subaccountIdProto,
+      order?.id,
+      true,
     );
   }
 

indexer/services/ender/src/lib/candles-generator.ts

@@ -1,23 +1,23 @@
 import { stats } from '@dydxprotocol-indexer/base';
 import { CANDLES_WEBSOCKET_MESSAGE_VERSION, KafkaTopics } from '@dydxprotocol-indexer/kafka';
 import {
+  CANDLE_RESOLUTION_TO_PROTO,
+  CandleColumns,
   CandleCreateObject,
   CandleFromDatabase,
+  CandleMessageContents,
   CandleResolution,
   CandleTable,
   CandleUpdateObject,
-  CANDLE_RESOLUTION_TO_PROTO,
+  MarketOpenInterest,
+  NUM_SECONDS_IN_CANDLE_RESOLUTIONS,
+  Options,
+  PerpetualMarketColumns,
+  PerpetualMarketFromDatabase,
   perpetualMarketRefresher,
+  PerpetualPositionTable,
   TradeContent,
   TradeMessageContents,
-  CandleMessageContents,
-  CandleColumns,
-  PerpetualPositionTable,
-  PerpetualMarketFromDatabase,
-  MarketOpenInterest,
-  PerpetualMarketColumns,
-  Options,
-  NUM_SECONDS_IN_CANDLE_RESOLUTIONS,
 } from '@dydxprotocol-indexer/postgres';
 import { CandleMessage } from '@dydxprotocol-indexer/v4-protos';
 import Big from 'big.js';
@@ -69,7 +69,7 @@ export class CandlesGenerator {
   public async updateCandles(): Promise<CandleFromDatabase[]> {
     const start: number = Date.now();
     // 1. Sort trade messages by order in the block
-    this.kafkaPublisher.sortTradeEvents();
+    this.kafkaPublisher.sortEvents(KafkaTopics.TO_WEBSOCKETS_TRADES);
 
     // 2. Generate BlockCandleUpdatesMap
     const blockCandleUpdatesMap: BlockCandleUpdatesMap = this.generateBlockCandleUpdatesMap();