address cmts

indexer/services/comlink/public/websocket-documentation.md

@@ -260,6 +260,12 @@ export enum FillType {
   LIQUIDATED = 'LIQUIDATED',
   // LIQUIDATION is for the maker side of the fill, never used for orders
   LIQUIDATION = 'LIQUIDATION',
+  // DELEVERAGED is for the subaccount that was deleveraged in a deleveraging event.
+  // The fill type will be set to taker.
+  DELEVERAGED = 'DELEVERAGED',
+  // OFFSETTING is for the offsetting subaccount in a deleveraging event.
+  // The fill type will be set to maker.
+  OFFSETTING = 'OFFSETTING',
 }
 
 export interface TransferSubaccountMessageContents {

indexer/services/ender/__tests__/handlers/deleveraging-handler.test.ts → indexer/services/ender/__tests__/handlers/order-fills/deleveraging-handler.test.ts

@@ -19,11 +19,11 @@ import {
   testConstants,
   testMocks,
 } from '@dydxprotocol-indexer/postgres';
-import { updateBlockCache } from '../../src/caches/block-cache';
-import { defaultDeleveragingEvent, defaultPreviousHeight } from '../helpers/constants';
-import { clearCandlesMap } from '../../src/caches/candle-cache';
-import { createPostgresFunctions } from '../../src/helpers/postgres/postgres-functions';
-import { redisClient } from '../../src/helpers/redis/redis-controller';
+import { updateBlockCache } from '../../../src/caches/block-cache';
+import { defaultDeleveragingEvent, defaultPreviousHeight } from '../../helpers/constants';
+import { clearCandlesMap } from '../../../src/caches/candle-cache';
+import { createPostgresFunctions } from '../../../src/helpers/postgres/postgres-functions';
+import { redisClient } from '../../../src/helpers/redis/redis-controller';
 import {
   DeleveragingEventV1,
   IndexerSubaccountId,
@@ -38,19 +38,19 @@ import {
   expectDefaultTradeKafkaMessageFromTakerFillId,
   expectFillInDatabase,
   expectFillSubaccountKafkaMessageFromLiquidationEvent,
-} from '../helpers/indexer-proto-helpers';
-import { DydxIndexerSubtypes } from '../../src/lib/types';
+} from '../../helpers/indexer-proto-helpers';
+import { DydxIndexerSubtypes } from '../../../src/lib/types';
 import {
   DELEVERAGING_EVENT_TYPE,
   MILLIS_IN_NANOS,
   SECONDS_IN_MILLIS,
   SUBACCOUNT_ORDER_FILL_EVENT_TYPE,
-} from '../../src/constants';
+} from '../../../src/constants';
 import { DateTime } from 'luxon';
 import Long from 'long';
-import { DeleveragingHandler } from '../../src/handlers/deleveraging-handler';
+import { DeleveragingHandler } from '../../../src/handlers/order-fills/deleveraging-handler';
 import { KafkaMessage } from 'kafkajs';
-import { onMessage } from '../../src/lib/on-message';
+import { onMessage } from '../../../src/lib/on-message';
 import { producer } from '@dydxprotocol-indexer/kafka';
 import { createdDateTime, createdHeight } from '@dydxprotocol-indexer/postgres/build/__tests__/helpers/constants';
 

indexer/services/ender/__tests__/lib/on-message.test.ts

@@ -51,12 +51,12 @@ import { updateBlockCache } from '../../src/caches/block-cache';
 import { MarketModifyHandler } from '../../src/handlers/markets/market-modify-handler';
 import Long from 'long';
 import { createPostgresFunctions } from '../../src/helpers/postgres/postgres-functions';
-import { DeleveragingHandler } from '../../src/handlers/deleveraging-handler';
+import { DeleveragingHandler } from '../../src/handlers/order-fills/deleveraging-handler';
 
 jest.mock('../../src/handlers/subaccount-update-handler');
 jest.mock('../../src/handlers/transfer-handler');
 jest.mock('../../src/handlers/funding-handler');
-jest.mock('../../src/handlers/deleveraging-handler');
+jest.mock('../../src/handlers/order-fills/deleveraging-handler');
 jest.mock('../../src/handlers/markets/market-modify-handler');
 
 describe('on-message', () => {

indexer/services/ender/src/handlers/deleveraging-handler.ts → indexer/services/ender/src/handlers/order-fills/deleveraging-handler.ts

@@ -8,21 +8,17 @@ import {
   PerpetualPositionFromDatabase,
   PerpetualPositionModel,
   storeHelpers,
-  SubaccountMessageContents,
   SubaccountTable,
-  TradeMessageContents,
-  UpdatedPerpetualPositionSubaccountKafkaObject,
 } from '@dydxprotocol-indexer/postgres';
-import { DeleveragingEventV1, IndexerSubaccountId } from '@dydxprotocol-indexer/v4-protos';
+import { DeleveragingEventV1 } from '@dydxprotocol-indexer/v4-protos';
 import * as pg from 'pg';
 
-import { DELEVERAGING_EVENT_TYPE, SUBACCOUNT_ORDER_FILL_EVENT_TYPE } from '../constants';
-import { generateFillSubaccountMessage, generatePerpetualPositionsContents } from '../helpers/kafka-helper';
-import { indexerTendermintEventToTransactionIndex } from '../lib/helper';
-import { ConsolidatedKafkaEvent } from '../lib/types';
-import { Handler } from './handler';
+import { DELEVERAGING_EVENT_TYPE, SUBACCOUNT_ORDER_FILL_EVENT_TYPE } from '../../constants';
+import { indexerTendermintEventToTransactionIndex } from '../../lib/helper';
+import { ConsolidatedKafkaEvent } from '../../lib/types';
+import { AbstractOrderFillHandler } from './abstract-order-fill-handler';
 
-export class DeleveragingHandler extends Handler<DeleveragingEventV1> {
+export class DeleveragingHandler extends AbstractOrderFillHandler<DeleveragingEventV1> {
   eventType: string = 'DeleveragingEvent';
 
   public getParallelizationIds(): string[] {
@@ -55,53 +51,6 @@ export class DeleveragingHandler extends Handler<DeleveragingEventV1> {
     ];
   }
 
-  protected generateConsolidatedKafkaEvent(
-    subaccountIdProto: IndexerSubaccountId,
-    position: UpdatedPerpetualPositionSubaccountKafkaObject | undefined,
-    fill: FillFromDatabase,
-    perpetualMarket: PerpetualMarketFromDatabase,
-  ): ConsolidatedKafkaEvent {
-    const message: SubaccountMessageContents = {
-      fills: [
-        generateFillSubaccountMessage(fill, perpetualMarket.ticker),
-      ],
-      perpetualPositions: position === undefined ? undefined : generatePerpetualPositionsContents(
-        subaccountIdProto,
-        [position],
-        perpetualMarketRefresher.getPerpetualMarketsMap(),
-      ),
-    };
-    return this.generateConsolidatedSubaccountKafkaEvent(
-      JSON.stringify(message),
-      subaccountIdProto,
-      undefined,
-      true,
-      message,
-    );
-  }
-
-  protected generateTradeKafkaEventFromDeleveraging(
-    fill: FillFromDatabase,
-  ): ConsolidatedKafkaEvent {
-    const tradeContents: TradeMessageContents = {
-      trades: [
-        {
-          id: fill.eventId.toString('hex'),
-          size: fill.size,
-          price: fill.price,
-          side: fill.side.toString(),
-          createdAt: fill.createdAt,
-          liquidation: false,
-          deleveraging: true,
-        },
-      ],
-    };
-    return this.generateConsolidatedTradeKafkaEvent(
-      JSON.stringify(tradeContents),
-      fill.clobPairId,
-    );
-  }
-
   // eslint-disable-next-line @typescript-eslint/require-await
   public async internalHandle(): Promise<ConsolidatedKafkaEvent[]> {
     const eventDataBinary: Uint8Array = this.indexerTendermintEvent.dataBytes;
@@ -141,17 +90,19 @@ export class DeleveragingHandler extends Handler<DeleveragingEventV1> {
     const kafkaEvents: ConsolidatedKafkaEvent[] = [
       this.generateConsolidatedKafkaEvent(
         this.event.liquidated!,
+        undefined,
         liquidatedPerpetualPosition,
         liquidatedFill,
         perpetualMarket,
       ),
       this.generateConsolidatedKafkaEvent(
         this.event.offsetting!,
+        undefined,
         offsettingPerpetualPosition,
         offsettingFill,
         perpetualMarket,
       ),
-      this.generateTradeKafkaEventFromDeleveraging(
+      this.generateTradeKafkaEventFromTakerOrderFill(
         liquidatedFill,
       ),
     ];

indexer/services/ender/src/helpers/postgres/postgres-functions.ts

@@ -58,7 +58,7 @@ const scripts: string[] = [
   'dydx_trim_scale.sql',
   'dydx_update_clob_pair_handler.sql',
   'dydx_update_perpetual_handler.sql',
-  'dydx_update_perpetual_position.sql',
+  'dydx_update_perpetual_position_aggregate_fields.sql',
   'dydx_uuid.sql',
   'dydx_uuid_from_asset_position_parts.sql',
   'dydx_uuid_from_fill_event_parts.sql',

indexer/services/ender/src/scripts/dydx_deleveraging_handler.sql

@@ -104,13 +104,13 @@ BEGIN
     RETURNING * INTO offsetting_fill_record;
 
     /* Upsert the perpetual_position records for this deleveraging event. */
-    liquidated_perpetual_position_record = dydx_update_perpetual_position(
+    liquidated_perpetual_position_record = dydx_update_perpetual_position_aggregate_fields(
         liquidated_subaccount_uuid,
         perpetual_id,
         liquidated_side,
         size,
         price);
-    offsetting_perpetual_position_record = dydx_update_perpetual_position(
+    offsetting_perpetual_position_record = dydx_update_perpetual_position_aggregate_fields(
         offsetting_subaccount_uuid,
         perpetual_id,
         offsetting_side,

indexer/services/ender/src/scripts/dydx_order_fill_handler_per_order.sql

@@ -165,7 +165,7 @@ BEGIN
     RETURNING * INTO fill_record;
 
     /* Upsert the perpetual_position record for this order_fill event. */
-    perpetual_position_record = dydx_update_perpetual_position(
+    perpetual_position_record = dydx_update_perpetual_position_aggregate_fields(
             subaccount_uuid,
             perpetual_market_record."id",
             order_side,

indexer/services/ender/src/scripts/dydx_update_perpetual_position.sql → indexer/services/ender/src/scripts/dydx_update_perpetual_position_aggregate_fields.sql

@@ -7,7 +7,7 @@
     - price: The price of the fill.
   Returns: the updated perpetual position.
 */
-CREATE OR REPLACE FUNCTION dydx_update_perpetual_position(
+CREATE OR REPLACE FUNCTION dydx_update_perpetual_position_aggregate_fields(
     subaccount_uuid uuid,
     perpetual_id bigint,
     side text,
@@ -21,7 +21,8 @@ DECLARE
     sum_close numeric;
     exit_price numeric;
 BEGIN
-    -- Retrieve the latest perpetual position record
+    -- Retrieve the latest perpetual position record.
+    -- TODO(IND-485): Order by openEventId instead of createdAtHeight.
     SELECT * INTO perpetual_position_record
     FROM perpetual_positions
     WHERE "subaccountId" = subaccount_uuid

indexer/services/ender/src/validators/deleveraging-validator.ts

@@ -1,7 +1,7 @@
 import { IndexerTendermintEvent, DeleveragingEventV1 } from '@dydxprotocol-indexer/v4-protos';
 
-import { DeleveragingHandler } from '../handlers/deleveraging-handler';
 import { Handler } from '../handlers/handler';
+import { DeleveragingHandler } from '../handlers/order-fills/deleveraging-handler';
 import { Validator } from './validator';
 
 export class DeleveragingValidator extends Validator<DeleveragingEventV1> {