[IND-387] add script to parse block and print json

indexer/services/scripts/.env

@@ -1 +1,3 @@
 SERVICE_NAME=scripts
+
+KAFKA_ENABLE_UNIQUE_CONSUMER_GROUP_IDS=true

indexer/services/scripts/package.json

@@ -9,6 +9,7 @@
     "build:prod": "pnpm run build",
     "build:watch": "pnpm run build -- --watch",
     "validate-pnl": "ts-node src/validate-pnl.ts",
+    "print-block": "ts-node src/print-block.ts",
     "coverage": "pnpm test -- --coverage",
     "lint": "eslint --ext .ts,.js .",
     "lint:fix": "eslint --ext .ts,.js . --fix",
@@ -19,11 +20,15 @@
   "dependencies": {
     "@dydxprotocol-indexer/base": "workspace:^0.0.1",
     "@dydxprotocol-indexer/postgres": "workspace:^0.0.1",
+    "@dydxprotocol-indexer/kafka": "workspace:^0.0.1",
+    "@dydxprotocol-indexer/v4-proto-parser": "workspace:^0.0.1",
+    "@dydxprotocol-indexer/v4-protos": "workspace:^0.0.1",
     "dotenv-flow": "^3.2.0",
     "long": "^5.2.1",
     "big.js": "^6.0.2",
     "lodash": "^4.17.21",
-    "yargs": "^13.3.0"
+    "yargs": "^13.3.0",
+    "kafkajs": "^2.1.0"
   },
   "devDependencies": {
     "@dydxprotocol-indexer/dev": "workspace:^0.0.1",

indexer/services/scripts/src/config.ts

@@ -0,0 +1,11 @@
+import { baseConfigSchema, parseSchema } from '@dydxprotocol-indexer/base';
+import { kafkaConfigSchema } from '@dydxprotocol-indexer/kafka';
+import { postgresConfigSchema } from '@dydxprotocol-indexer/postgres';
+
+export const configSchema = {
+  ...baseConfigSchema,
+  ...postgresConfigSchema,
+  ...kafkaConfigSchema,
+};
+
+export default parseSchema(configSchema);

indexer/services/scripts/src/helpers/block-helpers.ts

@@ -0,0 +1,110 @@
+import { logger } from '@dydxprotocol-indexer/base';
+import {
+  AssetCreateEventV1,
+  FundingEventV1,
+  IndexerTendermintEvent,
+  LiquidityTierUpsertEventV1,
+  MarketEventV1,
+  OrderFillEventV1,
+  PerpetualMarketCreateEventV1,
+  StatefulOrderEventV1,
+  SubaccountUpdateEventV1,
+  TransferEventV1,
+  UpdateClobPairEventV1,
+  UpdatePerpetualEventV1,
+} from '@dydxprotocol-indexer/v4-protos';
+
+import { AnnotatedIndexerTendermintEvent, DydxIndexerSubtypes } from './types';
+
+export function annotateIndexerTendermintEvent(
+  event: IndexerTendermintEvent,
+): AnnotatedIndexerTendermintEvent | undefined {
+  const eventDataBinary: Uint8Array = event.dataBytes;
+  switch (event.subtype) {
+    case (DydxIndexerSubtypes.ORDER_FILL.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(OrderFillEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.SUBACCOUNT_UPDATE.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(SubaccountUpdateEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.TRANSFER.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(TransferEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.MARKET.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(MarketEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.STATEFUL_ORDER.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(StatefulOrderEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.FUNDING.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(FundingEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.ASSET.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(AssetCreateEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.PERPETUAL_MARKET.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(PerpetualMarketCreateEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.LIQUIDITY_TIER.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(LiquidityTierUpsertEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.UPDATE_PERPETUAL.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(UpdatePerpetualEventV1.decode(eventDataBinary)),
+      };
+    }
+    case (DydxIndexerSubtypes.UPDATE_CLOB_PAIR.toString()): {
+      return {
+        ...event,
+        dataBytes: new Uint8Array(),
+        data: JSON.stringify(UpdateClobPairEventV1.decode(eventDataBinary)),
+      };
+    }
+    default: {
+      const message: string = `Unable to parse event subtype: ${event.subtype}`;
+      logger.error({
+        at: 'block-helpers#annotateIndexerTendermintEvent',
+        message,
+      });
+      return undefined;
+    }
+  }
+}

indexer/services/scripts/src/helpers/types.ts

@@ -0,0 +1,23 @@
+import { IndexerTendermintBlock, IndexerTendermintEvent } from '@dydxprotocol-indexer/v4-protos';
+
+export interface AnnotatedIndexerTendermintEvent extends IndexerTendermintEvent {
+  data: string;
+}
+
+export interface AnnotatedIndexerTendermintBlock extends IndexerTendermintBlock {
+  annotatedEvents: AnnotatedIndexerTendermintEvent[];
+}
+
+export enum DydxIndexerSubtypes {
+  ORDER_FILL = 'order_fill',
+  SUBACCOUNT_UPDATE = 'subaccount_update',
+  TRANSFER = 'transfer',
+  MARKET = 'market',
+  STATEFUL_ORDER = 'stateful_order',
+  FUNDING = 'funding_values',
+  ASSET = 'asset',
+  PERPETUAL_MARKET = 'perpetual_market',
+  LIQUIDITY_TIER = 'liquidity_tier',
+  UPDATE_PERPETUAL = 'update_perpetual',
+  UPDATE_CLOB_PAIR = 'update_clob_pair',
+}

indexer/services/scripts/src/print-block.ts

@@ -0,0 +1,169 @@
+import { logger, wrapBackgroundTask } from '@dydxprotocol-indexer/base';
+import {
+  addOnMessageFunction,
+  consumer,
+  producer,
+  startConsumer,
+  stopConsumer,
+  TO_ENDER_TOPIC,
+} from '@dydxprotocol-indexer/kafka';
+import { IndexerTendermintBlock, IndexerTendermintEvent } from '@dydxprotocol-indexer/v4-protos';
+import { KafkaMessage } from 'kafkajs';
+import _ from 'lodash';
+import yargs from 'yargs';
+
+import config from './config';
+import { annotateIndexerTendermintEvent } from './helpers/block-helpers';
+import { AnnotatedIndexerTendermintBlock, AnnotatedIndexerTendermintEvent } from './helpers/types';
+
+/**
+ * Creates an IndexerTendermintBlock from a KafkaMessage.
+ * Throws an error if there's an issue.
+ */
+function getIndexerTendermintBlock(
+  message: KafkaMessage,
+): IndexerTendermintBlock | undefined {
+  if (!message || !message.value || !message.timestamp) {
+    throw Error('Empty message');
+  }
+  const messageValueBinary: Uint8Array = new Uint8Array(message.value);
+
+  const block: IndexerTendermintBlock = IndexerTendermintBlock.decode(
+    messageValueBinary,
+  );
+
+  return block;
+}
+
+export function seek(offset: bigint): void {
+  logger.info({
+    at: 'consumer#seek',
+    message: 'Seeking...',
+    offset: offset.toString(),
+  });
+
+  consumer.seek({
+    topic: TO_ENDER_TOPIC,
+    partition: 0,
+    offset: offset.toString(),
+  });
+
+  logger.info({
+    at: 'consumer#seek',
+    message: 'Seeked.',
+    offset: offset.toString(),
+  });
+}
+
+export async function connect(height: number): Promise<void> {
+  await Promise.all([
+    consumer.connect(),
+    producer.connect(),
+  ]);
+
+  await consumer.subscribe({
+    topic: TO_ENDER_TOPIC,
+    fromBeginning: true,
+  });
+
+  addOnMessageFunction((_topic: string, message: KafkaMessage): Promise<void> => {
+    return printMessageAtHeight(message, height);
+  });
+
+  logger.info({
+    at: 'consumers#connect',
+    message: 'Connected to Kafka',
+  });
+}
+
+export async function printMessageAtHeight(
+  currentMessage: KafkaMessage,
+  targetHeight: number,
+): Promise<void> {
+  const indexerTendermintBlock: IndexerTendermintBlock | undefined = getIndexerTendermintBlock(
+    currentMessage,
+  );
+  if (indexerTendermintBlock === undefined) {
+    return;
+  }
+
+  const currentBlockHeight: number = parseInt(indexerTendermintBlock.height.toString(), 10);
+  if (currentBlockHeight < targetHeight) {
+    const offsetToSeek: number = targetHeight - currentBlockHeight + Number(currentMessage.offset);
+    await seek(BigInt(offsetToSeek));
+  } else if (currentBlockHeight === targetHeight) {
+    const annotatedEvents: AnnotatedIndexerTendermintEvent[] = [];
+    _.forEach(indexerTendermintBlock.events, (event: IndexerTendermintEvent) => {
+      const annotatedEvent:
+      AnnotatedIndexerTendermintEvent | undefined = annotateIndexerTendermintEvent(
+        event,
+      );
+      if (annotatedEvent === undefined) {
+        logger.error({
+          at: 'printMessageAtHeight',
+          message: 'Failed to parse event',
+          event,
+        });
+        throw Error('Failed to parse event');
+      }
+      annotatedEvents.push(annotatedEvent);
+    });
+    const annotatedBlock: AnnotatedIndexerTendermintBlock = {
+      ...indexerTendermintBlock,
+      events: [],
+      annotatedEvents,
+    };
+    logger.info({
+      at: 'printMessageAtHeight',
+      message: 'Printing block',
+      block: annotatedBlock,
+    });
+    await stopConsumer();
+  }
+}
+
+async function startKafka(height: number): Promise<void> {
+  logger.info({
+    at: 'index#start',
+    message: `Starting in env ${config.NODE_ENV}`,
+  });
+
+  await connect(height);
+  await startConsumer();
+
+  logger.info({
+    at: 'index#start',
+    message: 'Successfully started',
+  });
+}
+
+process.on('SIGTERM', async () => {
+  logger.info({
+    at: 'index#SIGTERM',
+    message: 'Received SIGTERM, shutting down',
+  });
+  await stopConsumer();
+});
+
+async function start(height: number): Promise<void> {
+  logger.info({
+    at: 'index#start',
+    message: `Connecting to kafka brokers: ${config.KAFKA_BROKER_URLS}`,
+  });
+  await startKafka(height);
+  logger.info({
+    at: 'index#start',
+    message: `Successfully connected to kafka brokers: ${config.KAFKA_BROKER_URLS}`,
+  });
+}
+
+const args = yargs.options({
+  height: {
+    type: 'number',
+    alias: 'h',
+    description: 'Height to print block at',
+    required: true,
+  },
+}).argv;
+
+wrapBackgroundTask(start(args.height), false, 'main');