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

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

@@ -1,8 +1,5 @@
 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';
@@ -12,9 +9,16 @@ import { ConsolidatedKafkaEvent, 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,
+  createConsolidatedKafkaEventFromTrade,
+} from '../helpers/kafka-publisher-helpers';
 
 describe('kafka-publisher', () => {
   let producerSendMock: jest.SpyInstance;
@@ -139,7 +143,7 @@ describe('kafka-publisher', () => {
         consolidatedBeforeTrade,
       ]);
 
-      publisher.sortTradeEvents();
+      publisher.sortEvents(KafkaTopics.TO_WEBSOCKETS_TRADES);
       expect(publisher.tradeMessages).toEqual([beforeTrade, trade, afterTrade]);
     });
   });

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();

indexer/services/ender/src/lib/kafka-publisher.ts

@@ -10,19 +10,22 @@ import { TradeMessageContents } from '@dydxprotocol-indexer/postgres';
 import {
   CandleMessage, MarketMessage, OffChainUpdateV1, SubaccountMessage, TradeMessage,
 } from '@dydxprotocol-indexer/v4-protos';
-import Big from 'big.js';
 import _ from 'lodash';
 
 import config from '../config';
-import { ConsolidatedKafkaEvent, SingleTradeMessage, VulcanMessage } from './types';
+import {
+  AnnotatedSubaccountMessage, ConsolidatedKafkaEvent, SingleTradeMessage, VulcanMessage,
+} from './types';
 
 type TopicKafkaMessages = {
   topic: KafkaTopics;
   messages: ProducerMessage[];
 };
 
+type OrderedMessage = AnnotatedSubaccountMessage | SingleTradeMessage;
+
 export class KafkaPublisher {
-  subaccountMessages: SubaccountMessage[];
+  subaccountMessages: AnnotatedSubaccountMessage[];
   tradeMessages: SingleTradeMessage[];
   marketMessages: MarketMessage[];
   candleMessages: CandleMessage[];
@@ -43,37 +46,94 @@ export class KafkaPublisher {
   }
 
   public addEvent(event: ConsolidatedKafkaEvent) {
-    switch (event.topic) {
+    this.getMessages(event.topic)!.push(event.message);
+  }
+
+  /**
+   * Helper function to get messages for a given topic.
+   *
+   * @param kafkaTopic
+   * @private
+   */
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  private getMessages(kafkaTopic: KafkaTopics): any[] | undefined {
+    switch (kafkaTopic) {
       case KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS:
-        this.subaccountMessages.push(event.message);
-        break;
+        return this.subaccountMessages;
       case KafkaTopics.TO_WEBSOCKETS_TRADES:
-        this.tradeMessages.push(event.message);
-        break;
+        return this.tradeMessages;
       case KafkaTopics.TO_WEBSOCKETS_MARKETS:
-        this.marketMessages.push(event.message);
-        break;
+        return this.marketMessages;
       case KafkaTopics.TO_WEBSOCKETS_CANDLES:
-        this.candleMessages.push(event.message);
-        break;
+        return this.candleMessages;
       case KafkaTopics.TO_VULCAN:
-        this.vulcanMessages.push(event.message);
-        break;
+        return this.vulcanMessages;
       default:
         throw new Error('Invalid Topic');
     }
   }
 
   /**
-   * Sort trade events by block height, transaction index, and event index in ascending order,
-   * where the first trade event should be the earliest trade in the block.
+   * Sort subaccountMessages that represent fills by block height, transaction index,
+   * and event index in ascending order per order id. Only keep the subaccount message if
+   * it represents the last fill event per order id.
+   */
+  public retainLastFillEventsForSubaccountMessages() {
+    // Create a map to store the last fill event per order ID
+    const lastFillEvents: Record<string, AnnotatedSubaccountMessage> = {};
+    const nonFillEvents: AnnotatedSubaccountMessage[] = [];
+
+    this.subaccountMessages.forEach((message) => {
+      if (message.isFill && message.orderId) {
+        const orderId = message.orderId;
+        // If we haven't seen this order ID before or if the current message
+        // has a higher block height, update the lastFillEvents for this order ID
+        if (
+          !lastFillEvents[orderId] ||
+          this.compareMessages(message, lastFillEvents[orderId]) > 0
+        ) {
+          lastFillEvents[orderId] = message;
+        }
+      } else {
+        nonFillEvents.push(message);
+      }
+    });
+
+    this.subaccountMessages = Object.values(lastFillEvents).concat(nonFillEvents);
+    this.sortEvents(KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS);
+  }
+
+  /** Helper function to compare two AnnotatedSubaccountMessages based on block height,
+   * transaction index, and event index.
+   */
+  private compareMessages(a: AnnotatedSubaccountMessage, b: AnnotatedSubaccountMessage) {
+    if (a.blockHeight === b.blockHeight) {
+      if (a.transactionIndex === b.transactionIndex) {
+        return a.eventIndex - b.eventIndex;
+      }
+      return a.transactionIndex - b.transactionIndex;
+    }
+    return Number(a.blockHeight) - Number(b.blockHeight);
+  }
+
+  /**
+   * Sort events by block height, transaction index, and event index in ascending order,
+   * where the first event should be the earliest event in the block.
    */
-  public sortTradeEvents() {
-    this.tradeMessages = this.tradeMessages.sort(
-      (a: SingleTradeMessage, b: SingleTradeMessage) => {
-        if (Big(a.blockHeight).lt(b.blockHeight)) {
+  public sortEvents(kafkaTopic: KafkaTopics) {
+    if (![
+      KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS,
+      KafkaTopics.TO_WEBSOCKETS_TRADES,
+    ].includes(kafkaTopic)) {
+      throw new Error('Sorting events is only supported for subaccount and trade kafka websocket topics');
+    }
+    const msgs: OrderedMessage[] = this.getMessages(kafkaTopic) as OrderedMessage[];
+
+    if (msgs) {
+      msgs.sort((a: OrderedMessage, b: OrderedMessage) => {
+        if (a.blockHeight < b.blockHeight) {
           return -1;
-        } else if (Big(a.blockHeight).gt(b.blockHeight)) {
+        } else if (a.blockHeight > b.blockHeight) {
           return 1;
         }
 
@@ -83,9 +143,9 @@ export class KafkaPublisher {
           return 1;
         }
 
-        return a.eventIndex < b.eventIndex ? -1 : 1;
-      },
-    );
+        return a.eventIndex - b.eventIndex;
+      });
+    }
   }
 
   public async publish() {
@@ -112,6 +172,7 @@ export class KafkaPublisher {
   private generateAllTopicKafkaMessages(): TopicKafkaMessages[] {
     const allTopicKafkaMessages: TopicKafkaMessages[] = [];
     if (this.subaccountMessages.length > 0) {
+      this.retainLastFillEventsForSubaccountMessages();
       allTopicKafkaMessages.push({
         topic: KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS,
         messages: _.map(this.subaccountMessages, (message: SubaccountMessage) => {

indexer/services/ender/src/lib/types.ts

@@ -188,14 +188,19 @@ export interface SingleTradeMessage extends TradeMessage {
   eventIndex: number,
 }
 
+export interface AnnotatedSubaccountMessage extends SubaccountMessage {
+  orderId?: string,
+  isFill?: boolean,
+}
+
 export interface VulcanMessage {
   key: Buffer,
   value: OffChainUpdateV1,
 }
 
 export type ConsolidatedKafkaEvent = {
   topic: KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS,
-  message: SubaccountMessage,
+  message: AnnotatedSubaccountMessage,
 } | {
   topic: KafkaTopics.TO_WEBSOCKETS_TRADES,
   message: SingleTradeMessage,