feat: Add x/marketmap module

indexer/README.md

@@ -230,4 +230,5 @@ Other example subscription events:
 { "type": "subscribe", "channel": "v4_markets" }
 { "type": "subscribe", "channel": "v4_orderbook", "id": "BTC-USD" }
 { "type": "subscribe", "channel": "v4_subaccounts", "id": "address/0" }
+{ "type": "subscribe", "channel": "v4_block_height" }
 ```

indexer/docker-compose-local-deployment.yml

@@ -13,7 +13,8 @@ services:
  to-websockets-subaccounts:1:1,\
  to-websockets-trades:1:1,\
  to-websockets-markets:1:1,\
- to-websockets-candles:1:1"
+ to-websockets-candles:1:1,\
+ to-websockets-block-height:1:1"
  KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL_SAME_HOST://:29092
  KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL_SAME_HOST://localhost:29092
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL_SAME_HOST:PLAINTEXT

indexer/docker-compose.yml

@@ -7,14 +7,15 @@ services:
  environment:
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  KAFKA_ADVERTISED_HOST_NAME: localhost
- KAFKA_CREATE_TOPICS: 
+ KAFKA_CREATE_TOPICS:
  "to-ender:1:1,\
  to-vulcan:1:1,\
  to-websockets-orderbooks:1:1,\
  to-websockets-subaccounts:1:1,\
  to-websockets-trades:1:1,\
  to-websockets-markets:1:1,\
- to-websockets-candles:1:1"
+ to-websockets-candles:1:1,\
+ to-websockets-block-height:1:1"
  postgres-test:
  build:
  context: .

indexer/packages/kafka/src/constants.ts

@@ -5,3 +5,4 @@ export const SUBACCOUNTS_WEBSOCKET_MESSAGE_VERSION: string = '3.0.0';
 export const TRADES_WEBSOCKET_MESSAGE_VERSION: string = '2.1.0';
 export const MARKETS_WEBSOCKET_MESSAGE_VERSION: string = '1.0.0';
 export const CANDLES_WEBSOCKET_MESSAGE_VERSION: string = '1.0.0';
+export const BLOCK_HEIGHT_WEBSOCKET_MESSAGE_VERSION: string = '1.0.0';

indexer/packages/kafka/src/types.ts

@@ -4,6 +4,7 @@ export enum WebsocketTopics {
  TO_WEBSOCKETS_TRADES = 'to-websockets-trades',
  TO_WEBSOCKETS_MARKETS = 'to-websockets-markets',
  TO_WEBSOCKETS_CANDLES = 'to-websockets-candles',
+ TO_WEBSOCKETS_BLOCK_HEIGHT = 'to-websockets-block-height',
 }
 
 export enum KafkaTopics {
@@ -14,4 +15,5 @@ export enum KafkaTopics {
  TO_WEBSOCKETS_TRADES = 'to-websockets-trades',
  TO_WEBSOCKETS_MARKETS = 'to-websockets-markets',
  TO_WEBSOCKETS_CANDLES = 'to-websockets-candles',
+ TO_WEBSOCKETS_BLOCK_HEIGHT = 'to-websockets-block-height',
 }

indexer/packages/v4-protos/src/codegen/dydxprotocol/indexer/socks/messages.ts

@@ -265,6 +265,30 @@ export interface CandleMessageSDKType {
 
  version: string;
 }
+/** Message to be sent through the 'to-websockets-block-height` kafka topic. */
+
+export interface BlockHeightMessage {
+ /** Block height where the contents occur. */
+ blockHeight: string;
+ /** ISO formatted time of the block height. */
+
+ time: string;
+ /** Version of the websocket message. */
+
+ version: string;
+}
+/** Message to be sent through the 'to-websockets-block-height` kafka topic. */
+
+export interface BlockHeightMessageSDKType {
+ /** Block height where the contents occur. */
+ block_height: string;
+ /** ISO formatted time of the block height. */
+
+ time: string;
+ /** Version of the websocket message. */
+
+ version: string;
+}
 
 function createBaseOrderbookMessage(): OrderbookMessage {
  return {
@@ -629,4 +653,69 @@ export const CandleMessage = {
  return message;
  }
 
+};
+
+function createBaseBlockHeightMessage(): BlockHeightMessage {
+ return {
+ blockHeight: "",
+ time: "",
+ version: ""
+ };
+}
+
+export const BlockHeightMessage = {
+ encode(message: BlockHeightMessage, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+ if (message.blockHeight !== "") {
+ writer.uint32(10).string(message.blockHeight);
+ }
+
+ if (message.time !== "") {
+ writer.uint32(18).string(message.time);
+ }
+
+ if (message.version !== "") {
+ writer.uint32(26).string(message.version);
+ }
+
+ return writer;
+ },
+
+ decode(input: _m0.Reader | Uint8Array, length?: number): BlockHeightMessage {
+ const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
+ let end = length === undefined ? reader.len : reader.pos + length;
+ const message = createBaseBlockHeightMessage();
+
+ while (reader.pos < end) {
+ const tag = reader.uint32();
+
+ switch (tag >>> 3) {
+ case 1:
+ message.blockHeight = reader.string();
+ break;
+
+ case 2:
+ message.time = reader.string();
+ break;
+
+ case 3:
+ message.version = reader.string();
+ break;
+
+ default:
+ reader.skipType(tag & 7);
+ break;
+ }
+ }
+
+ return message;
+ },
+
+ fromPartial(object: DeepPartial<BlockHeightMessage>): BlockHeightMessage {
+ const message = createBaseBlockHeightMessage();
+ message.blockHeight = object.blockHeight ?? "";
+ message.time = object.time ?? "";
+ message.version = object.version ?? "";
+ return message;
+ }
+
 };

indexer/packages/v4-protos/src/codegen/dydxprotocol/vault/genesis.ts

@@ -1,22 +1,55 @@
 import { Params, ParamsSDKType } from "./params";
+import { VaultId, VaultIdSDKType, NumShares, NumSharesSDKType } from "./vault";
+import { OwnerShare, OwnerShareSDKType } from "./query";
 import * as _m0 from "protobufjs/minimal";
 import { DeepPartial } from "../../helpers";
 /** GenesisState defines `x/vault`'s genesis state. */
 
 export interface GenesisState {
  /** The parameters of the module. */
  params?: Params;
+ /** The vaults. */
+
+ vaults: Vault[];
 }
 /** GenesisState defines `x/vault`'s genesis state. */
 
 export interface GenesisStateSDKType {
  /** The parameters of the module. */
  params?: ParamsSDKType;
+ /** The vaults. */
+
+ vaults: VaultSDKType[];
+}
+/** Vault defines the total shares and owner shares of a vault. */
+
+export interface Vault {
+ /** The ID of the vault. */
+ vaultId?: VaultId;
+ /** The total number of shares in the vault. */
+
+ totalShares?: NumShares;
+ /** The shares of each owner in the vault. */
+
+ ownerShares: OwnerShare[];
+}
+/** Vault defines the total shares and owner shares of a vault. */
+
+export interface VaultSDKType {
+ /** The ID of the vault. */
+ vault_id?: VaultIdSDKType;
+ /** The total number of shares in the vault. */
+
+ total_shares?: NumSharesSDKType;
+ /** The shares of each owner in the vault. */
+
+ owner_shares: OwnerShareSDKType[];
 }
 
 function createBaseGenesisState(): GenesisState {
  return {
- params: undefined
+ params: undefined,
+ vaults: []
  };
 }
 
@@ -26,6 +59,10 @@ export const GenesisState = {
  Params.encode(message.params, writer.uint32(10).fork()).ldelim();
  }
 
+ for (const v of message.vaults) {
+ Vault.encode(v!, writer.uint32(18).fork()).ldelim();
+ }
+
  return writer;
  },
 
@@ -42,6 +79,10 @@ export const GenesisState = {
  message.params = Params.decode(reader, reader.uint32());
  break;
 
+ case 2:
+ message.vaults.push(Vault.decode(reader, reader.uint32()));
+ break;
+
  default:
  reader.skipType(tag & 7);
  break;
@@ -54,6 +95,72 @@ export const GenesisState = {
  fromPartial(object: DeepPartial<GenesisState>): GenesisState {
  const message = createBaseGenesisState();
  message.params = object.params !== undefined && object.params !== null ? Params.fromPartial(object.params) : undefined;
+ message.vaults = object.vaults?.map(e => Vault.fromPartial(e)) || [];
+ return message;
+ }
+
+};
+
+function createBaseVault(): Vault {
+ return {
+ vaultId: undefined,
+ totalShares: undefined,
+ ownerShares: []
+ };
+}
+
+export const Vault = {
+ encode(message: Vault, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+ if (message.vaultId !== undefined) {
+ VaultId.encode(message.vaultId, writer.uint32(10).fork()).ldelim();
+ }
+
+ if (message.totalShares !== undefined) {
+ NumShares.encode(message.totalShares, writer.uint32(18).fork()).ldelim();
+ }
+
+ for (const v of message.ownerShares) {
+ OwnerShare.encode(v!, writer.uint32(26).fork()).ldelim();
+ }
+
+ return writer;
+ },
+
+ decode(input: _m0.Reader | Uint8Array, length?: number): Vault {
+ const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
+ let end = length === undefined ? reader.len : reader.pos + length;
+ const message = createBaseVault();
+
+ while (reader.pos < end) {
+ const tag = reader.uint32();
+
+ switch (tag >>> 3) {
+ case 1:
+ message.vaultId = VaultId.decode(reader, reader.uint32());
+ break;
+
+ case 2:
+ message.totalShares = NumShares.decode(reader, reader.uint32());
+ break;
+
+ case 3:
+ message.ownerShares.push(OwnerShare.decode(reader, reader.uint32()));
+ break;
+
+ default:
+ reader.skipType(tag & 7);
+ break;
+ }
+ }
+
+ return message;
+ },
+
+ fromPartial(object: DeepPartial<Vault>): Vault {
+ const message = createBaseVault();
+ message.vaultId = object.vaultId !== undefined && object.vaultId !== null ? VaultId.fromPartial(object.vaultId) : undefined;
+ message.totalShares = object.totalShares !== undefined && object.totalShares !== null ? NumShares.fromPartial(object.totalShares) : undefined;
+ message.ownerShares = object.ownerShares?.map(e => OwnerShare.fromPartial(e)) || [];
  return message;
  }
 

indexer/services/auxo/src/constants.ts

@@ -10,6 +10,13 @@ export const BAZOOKA_DB_MIGRATION_PAYLOAD: Uint8Array = new TextEncoder().encode
  }),
 );
 
+export const BAZOOKA_DB_MIGRATION_AND_CREATE_KAFKA_PAYLOAD: Uint8Array = new TextEncoder().encode(
+ JSON.stringify({
+ migrate: true,
+ create_kafka_topics: true,
+ }),
+);
+
 export const ECS_SERVICE_NAMES: EcsServiceNames[] = [
  EcsServiceNames.COMLINK,
  EcsServiceNames.ENDER,

indexer/services/auxo/src/index.ts

@@ -30,6 +30,7 @@ import _ from 'lodash';
 
 import config from './config';
 import {
+ BAZOOKA_DB_MIGRATION_AND_CREATE_KAFKA_PAYLOAD,
  BAZOOKA_DB_MIGRATION_PAYLOAD,
  BAZOOKA_LAMBDA_FUNCTION_NAME,
  ECS_SERVICE_NAMES,
@@ -40,7 +41,7 @@ import { AuxoEventJson, EcsServiceNames, TaskDefinitionArnMap } from './types';
 /**
  * Upgrades all services and run migrations
  * 1. Upgrade Bazooka
- * 2. Run db migration in Bazooka
+ * 2. Run db migration in Bazooka, and update kafka topics
  * 3. Create new ECS Task Definition for ECS Services with new image
  * 4. Upgrade all ECS Services (comlink, ender, roundtable, socks, vulcan)
  */
@@ -66,8 +67,9 @@ export async function handler(
  // 1. Upgrade Bazooka
  await upgradeBazooka(lambda, ecr, event);
 
- // 2. Run db migration in Bazooka
- await runDbMigration(lambda);
+ // 2. Run db migration in Bazooka,
+ // boolean flag used to determine if new kafka topics should be created
+ await runDbAndKafkaMigration(event.addNewKafkaTopics, lambda);
 
  // 3. Create new ECS Task Definition for ECS Services with new image
  const taskDefinitionArnMap: TaskDefinitionArnMap = await createNewEcsTaskDefinitions(
@@ -192,16 +194,20 @@ async function getImageDetail(
 
 }
 
-async function runDbMigration(
+async function runDbAndKafkaMigration(
+ createNewKafkaTopics: boolean,
  lambda: ECRClient,
 ): Promise<void> {
  logger.info({
  at: 'index#runDbMigration',
  message: 'Running db migration',
  });
+ const payload = createNewKafkaTopics
+ ? BAZOOKA_DB_MIGRATION_AND_CREATE_KAFKA_PAYLOAD
+ : BAZOOKA_DB_MIGRATION_PAYLOAD;
  const response: InvokeCommandOutput = await lambda.send(new InvokeCommand({
  FunctionName: BAZOOKA_LAMBDA_FUNCTION_NAME,
- Payload: BAZOOKA_DB_MIGRATION_PAYLOAD,
+ Payload: payload,
  // RequestResponse means that the lambda is synchronously invoked
  InvocationType: 'RequestResponse',
  }));

indexer/services/auxo/src/types.ts

@@ -13,6 +13,7 @@ export interface AuxoEventJson {
  region: string;
  // In our naming we often times use the appreviated region name
  regionAbbrev: string;
+ addNewKafkaTopics: boolean;
 }
 
 // EcsServiceName to task definition arn mapping