Adam/ct 948 add functionality to auxo to create new kafka topics

.github/workflows/indexer-build-and-push-dev-staging.yml

@@ -3,56 +3,16 @@ name: Indexer Build & Push Images to AWS ECR for Dev / Staging branches
 on: # yamllint disable-line rule:truthy
   push:
     branches:
+      - adam/ct-948-add-functionality-to-auxo-to-create-new-kafka-topics
       - main
       - 'release/indexer/v[0-9]+.[0-9]+.x'  # e.g. release/indexer/v0.1.x
       - 'release/indexer/v[0-9]+.x'  # e.g. release/indexer/v1.x
     # TODO(DEC-837): Customize github build and push to ECR by service with paths
 
 jobs:
-  # Build and push to dev
-  call-build-and-push-ecs-services-dev:
-    name: (Dev) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev
-    secrets: inherit
-
-  # Build and push to dev2
-  call-build-and-push-ecs-services-dev2:
-    name: (Dev2) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev2
-    secrets: inherit
-
-  # Build and push to dev3
-  call-build-and-push-ecs-services-dev3:
-    name: (Dev3) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev3
-    secrets: inherit
-
-  # Build and push to dev4
-  call-build-and-push-ecs-services-dev4:
-    name: (Dev4) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: dev4
-    secrets: inherit
-
-  # Build and push to dev5
   call-build-and-push-ecs-services-dev5:
     name: (Dev5) Build and Push ECS Services
     uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
     with:
       ENVIRONMENT: dev5
-    secrets: inherit
-
-  # Build and push to staging
-  call-build-and-push-ecs-services-staging:
-    name: (Staging) Build and Push ECS Services
-    uses: ./.github/workflows/indexer-build-and-push-all-ecr-images.yml
-    with:
-      ENVIRONMENT: staging
-    secrets: inherit
+    secrets: inherit

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/postgres/src/db/migrations/migration_files/20240613174832_add_gb_to_compliance_reasons.ts

@@ -0,0 +1,19 @@
+import * as Knex from 'knex';
+
+import { formatAlterTableEnumSql } from '../helpers';
+
+export async function up(knex: Knex): Promise<void> {
+  return knex.raw(formatAlterTableEnumSql(
+    'compliance_status',
+    'reason',
+    ['MANUAL', 'US_GEO', 'CA_GEO', 'GB_GEO', 'SANCTIONED_GEO', 'COMPLIANCE_PROVIDER'],
+  ));
+}
+
+export async function down(knex: Knex): Promise<void> {
+  return knex.raw(formatAlterTableEnumSql(
+    'compliance_status',
+    'reason',
+    ['MANUAL', 'US_GEO', 'CA_GEO', 'SANCTIONED_GEO', 'COMPLIANCE_PROVIDER'],
+  ));
+}

indexer/packages/postgres/src/types/compliance-status-types.ts

@@ -6,6 +6,7 @@ export enum ComplianceReason {
   MANUAL = 'MANUAL',
   US_GEO = 'US_GEO',
   CA_GEO = 'CA_GEO',
+  GB_GEO = 'GB_GEO',
   SANCTIONED_GEO = 'SANCTIONED_GEO',
   COMPLIANCE_PROVIDER = 'COMPLIANCE_PROVIDER',
 }

indexer/packages/redis/__tests__/caches/orderbook-levels-cache.test.ts

@@ -15,7 +15,7 @@ import {
 } from '../../src/caches/orderbook-levels-cache';
 import { OrderSide } from '@dydxprotocol-indexer/postgres';
 import { OrderbookLevels, PriceLevel } from '../../src/types';
-import { InvalidOptionsError, InvalidPriceLevelUpdateError } from '../../src/errors';
+import { InvalidOptionsError } from '../../src/errors';
 import { logger } from '@dydxprotocol-indexer/base';
 
 describe('orderbookLevelsCache', () => {
@@ -176,11 +176,10 @@ describe('orderbookLevelsCache', () => {
       expect(orderbookLevels.bids).toEqual([]);
     });
 
-    it('throws error if update will cause quantums to be negative', async () => {
+    it('sets price level to 0 if update will cause quantums to be negative', async () => {
       const humanPrice: string = '50000';
       const quantums: string = '1000';
       const invalidDelta: string = '-2000';
-      const resultingQuantums: string = '-1000';
       // Set existing quantums for the level
       await updatePriceLevel({
         ticker,
@@ -190,31 +189,26 @@ describe('orderbookLevelsCache', () => {
         client,
       });
 
-      // Test that an error is thrown if the update results in a negative quantums for the price
-      // level
-      await expect(updatePriceLevel({
+      await updatePriceLevel({
         ticker,
         side: OrderSide.BUY,
         humanPrice,
         sizeDeltaInQuantums: invalidDelta,
         client,
-      })).rejects.toBeInstanceOf(InvalidPriceLevelUpdateError);
+      });
       expect(logger.crit).toHaveBeenCalledTimes(1);
-      await expect(updatePriceLevel({
+
+      // Expect that the value in the orderbook is set to 0
+      const orderbookLevels: OrderbookLevels = await getOrderBookLevels(
         ticker,
-        side: OrderSide.BUY,
-        humanPrice,
-        sizeDeltaInQuantums: invalidDelta,
         client,
-      })).rejects.toEqual(expect.objectContaining({
-        message: expect.stringContaining(resultingQuantums),
-      }));
-
-      // Expect that the value in the orderbook is unchanged
-      const orderbookLevels: OrderbookLevels = await getOrderBookLevels(ticker, client);
+        {
+          removeZeros: false,
+        },
+      );
       expect(orderbookLevels.bids).toMatchObject([{
         humanPrice,
-        quantums,
+        quantums: '0',
       }]);
     });
   });

indexer/packages/redis/src/caches/orderbook-levels-cache.ts

@@ -4,7 +4,7 @@ import Big from 'big.js';
 import _ from 'lodash';
 import { Callback, RedisClient } from 'redis';
 
-import { InvalidOptionsError, InvalidPriceLevelUpdateError } from '../errors';
+import { InvalidOptionsError } from '../errors';
 import { hGetAsync } from '../helpers/redis';
 import { OrderbookLevels, PriceLevel } from '../types';
 import {
@@ -58,32 +58,29 @@ export async function updatePriceLevel({
   // NOTE: If this happens from a single price level update, it's possible for multiple subsequent
   // price level updates to fail with the same error due to interleaved price level updates.
   if (updatedQuantums < 0) {
-    // Undo the update. This can't be done in a Lua script as Redis runs Lua 5.1, which only
-    // uses doubles which support up to 53-bit integers. Race-condition where it's possible for a
-    // price-level to have negative quantums handled in `getOrderbookLevels` where price-levels with
-    // negative quantums are filtered out. Note: even though we are reverting this information, each
-    // call to incrementOrderbookLevel updates the lastUpdated key in the cache.
+    // Set the price level to 0.
+    // Race-condition where it's possible for a price-level to have negative quantums handled in
+    // `getOrderbookLevels` where price-levels with negative quantums are filtered out. Note: even
+    // though we are reverting this information, each call to incrementOrderbookLevel updates the
+    // lastUpdated key in the cache.
     await incrementOrderbookLevel(
       ticker,
       side,
       humanPrice,
       // Needs to be an integer
-      Big(sizeDeltaInQuantums).mul(-1).toFixed(0),
+      Big(updatedQuantums).mul(-1).toFixed(0),
       client,
     );
     logger.crit({
       at: 'orderbookLevelsCache#updatePriceLevel',
-      message: 'Price level updated to negative quantums',
+      message: 'Price level updated to negative quantums, set to zero',
       ticker,
       side,
       humanPrice,
       updatedQuantums,
       sizeDeltaInQuantums,
     });
-    throw new InvalidPriceLevelUpdateError(
-      '#updatePriceLevel: Resulting price level has negative quantums, quantums = ' +
-      `${updatedQuantums}`,
-    );
+    return 0;
   }
 
   return updatedQuantums;

indexer/packages/redis/src/errors.ts

@@ -21,11 +21,3 @@ export class InvalidOptionsError extends Error {
     Error.captureStackTrace(this, this.constructor);
   }
 }
-
-export class InvalidPriceLevelUpdateError extends Error {
-  constructor(message: string) {
-    super(`Invalid price level update: ${message}`);
-    this.name = this.constructor.name;
-    Error.captureStackTrace(this, this.constructor);
-  }
-}