fix: adjusted the new blocks from depth subscription mechanism

Author: mschead

indexer/src/kadena-server/repository/application/block-repository.ts

@@ -28,6 +28,7 @@ export interface GetLatestBlocksParams {
 
 export type BlockOutput = Omit<Block, 'parent' | 'events' | 'minerAccount' | 'transactions'> & {
   parentHash: string;
+  blockId: number;
 };
 
 export type FungibleChainAccountOutput = Omit<
@@ -65,6 +66,13 @@ export default interface BlockRepository {
     transactions: TransactionOutput[],
   ): Promise<TransactionOutput[]>;
 
+  getLastBlocksWithDepth(
+    chainIds: string[],
+    minimumDepth: number,
+    startingTimestamp: number,
+    id?: string,
+  ): Promise<BlockOutput[]>;
+
   // dataloader
   getBlocksByEventIds(eventIds: string[]): Promise<BlockOutput[]>;
   getBlocksByTransactionIds(transactionIds: string[]): Promise<BlockOutput[]>;

indexer/src/kadena-server/repository/infra/repository/block-db-repository.ts

@@ -648,4 +648,126 @@ export default class BlockDbRepository implements BlockRepository {
 
     return output;
   }
+
+  async getLastBlocksWithDepth(
+    chainIdsParam: string[],
+    minimumDepth: number,
+    startingTimestamp: number,
+    id?: string,
+  ): Promise<BlockOutput[]> {
+    const chainIds = chainIdsParam.length ? chainIdsParam.map(Number) : await this.getChainIds();
+    const maxHeights = await this.getMaxHeights();
+
+    const queryParams: any[] = [];
+    const conditions: string[] = [];
+
+    chainIds.forEach(chainId => {
+      const maxHeight = maxHeights[chainId];
+      if (maxHeight !== undefined) {
+        conditions.push(
+          `("chainId" = $${queryParams.length + 1} AND height <= $${queryParams.length + 2})`,
+        );
+        queryParams.push(chainId, maxHeight - minimumDepth);
+      }
+    });
+
+    queryParams.push(startingTimestamp.toFixed());
+
+    let query = `
+      SELECT *
+      FROM "Blocks"
+      WHERE ${conditions.join(' OR ')}
+      AND "creationTime" > $${queryParams.length}
+    `;
+
+    if (id) {
+      queryParams.push(id);
+      query += `
+        AND id > $${queryParams.length}
+        ORDER BY id DESC
+        LIMIT 100
+      `;
+    } else {
+      query += `
+        ORDER BY id DESC
+        LIMIT 5
+      `;
+    }
+
+    const { rows: blockRows } = await rootPgPool.query(query, queryParams);
+
+    const blocksToReturn = blockRows.map(row => blockValidator.validate(row));
+    const blockHashToDepth = await this.createBlockDepthMap(blocksToReturn, 'hash', minimumDepth);
+
+    return blocksToReturn.filter(block => blockHashToDepth[block.hash] >= minimumDepth);
+  }
+
+  async getConfirmationDepth(blockHash: string, minimumDepth: number): Promise<number> {
+    const query = `
+      WITH RECURSIVE BlockDescendants AS (
+        SELECT hash, parent, 0 AS depth, height, "chainId"
+        FROM "Blocks"
+        WHERE hash = $1
+        UNION ALL
+        SELECT b.hash, b.parent, d.depth + 1 AS depth, b.height, b."chainId"
+        FROM BlockDescendants d
+        JOIN "Blocks" b ON d.hash = b.parent AND b.height = d.height + 1 AND b."chainId" = d."chainId"
+        WHERE d.depth <= $2
+      )
+      SELECT MAX(depth) AS depth
+      FROM BlockDescendants;
+    `;
+
+    const { rows } = await rootPgPool.query(query, [blockHash, minimumDepth]);
+
+    if (rows.length && rows[0].depth) {
+      return Number(rows[0].depth);
+    } else {
+      return 0;
+    }
+  }
+
+  /**
+   *
+   * @param items - all the items to create a block depth map for
+   * @param hashProp - the property of the item that contains the block hash
+   * @returns a map of block hashes to their confirmation depths
+   */
+  async createBlockDepthMap<T>(
+    items: T[],
+    hashProp: keyof T,
+    minimumDepth: number,
+  ): Promise<Record<string, number>> {
+    const uniqueBlockHashes = [...new Set(items.map(item => item[hashProp]))];
+
+    const confirmationDepths = await Promise.all(
+      uniqueBlockHashes.map(blockHash =>
+        this.getConfirmationDepth(blockHash as string, minimumDepth),
+      ),
+    );
+
+    return uniqueBlockHashes.reduce((map: Record<string, number>, blockHash, index) => {
+      map[blockHash as string] = confirmationDepths[index];
+      return map;
+    }, {});
+  }
+
+  async getMaxHeights(): Promise<Record<string, number>> {
+    const chainIds = await this.getChainIds();
+
+    const maxHeightsByChainIdPromises = chainIds.map(async chainId => {
+      const query = `
+        SELECT MAX(height) as max_height
+        FROM "Blocks"
+        WHERE "chainId" = $1
+      `;
+
+      const { rows } = await rootPgPool.query(query, [chainId]);
+      return { [chainId.toString()]: parseInt(rows[0].max_height, 10) };
+    });
+
+    const maxHeightsArray = await Promise.all(maxHeightsByChainIdPromises);
+
+    return Object.assign({}, ...maxHeightsArray);
+  }
 }

indexer/src/kadena-server/repository/infra/schema-validator/block-schema-validator.ts

@@ -93,6 +93,7 @@ const validate = (row: any): BlockOutput => {
       chainId,
       hash,
     })),
+    blockId: res.id,
   };
 };
 
@@ -126,6 +127,7 @@ const mapFromSequelize = (blockModel: BlockAttributes): BlockOutput => {
       chainId,
       hash,
     })),
+    blockId: blockModel.id,
   };
 };
 

indexer/src/kadena-server/resolvers/subscription/new-blocks-from-depth-subscription-resolver.ts

@@ -5,33 +5,47 @@
  * which allows clients to receive real-time updates when blocks reach a specified minimum
  * confirmation depth. This is particularly useful for applications that need to wait for
  * a certain number of confirmations before considering a transaction finalized.
- *
- * Unlike the standard newBlocks subscription, this subscription uses a PubSub event
- * filtering mechanism to only notify clients when blocks meet their criteria.
  */
 
-import { withFilter } from 'graphql-subscriptions';
 import { ResolverContext } from '../../config/apollo-server-config';
-import {
-  SubscriptionNewBlocksFromDepthArgs,
-  SubscriptionResolvers,
-} from '../../config/graphql-types';
-import { NEW_BLOCKS_FROM_DEPTH_EVENT } from './consts';
-import zod from 'zod';
-import { blockQueryResolver } from '../query/block-query-resolver';
+import { SubscriptionResolvers } from '../../config/graphql-types';
+import { BlockOutput } from '../../repository/application/block-repository';
 
-/**
- * Zod schema for validating incoming block notification events
- *
- * This schema ensures that the payload from the PubSub system contains
- * the required properties (chainId, height, hash) and that they are of
- * the correct types before processing the event.
- */
-const newBlocksFromDepthSubscriptionSchema = zod.object({
-  chainId: zod.string(),
-  height: zod.number(),
-  hash: zod.string(),
-});
+async function* iteratorFn(
+  chainIds: string[],
+  minimumDepth: number,
+  context: ResolverContext,
+): AsyncGenerator<BlockOutput[], void, unknown> {
+  const startingTimestamp = new Date().getTime() / 1000000;
+  const blockResult = await context.blockRepository.getLastBlocksWithDepth(
+    chainIds,
+    minimumDepth,
+    startingTimestamp,
+  );
+
+  let lastBlockId: string | undefined;
+
+  if (blockResult.length > 0) {
+    lastBlockId = blockResult[0].blockId.toString();
+    yield [];
+  }
+
+  while (context.signal) {
+    const newBlocks = await context.blockRepository.getLastBlocksWithDepth(
+      chainIds,
+      minimumDepth,
+      startingTimestamp,
+      lastBlockId,
+    );
+
+    if (newBlocks.length > 0) {
+      lastBlockId = newBlocks[0].blockId.toString();
+      yield newBlocks;
+    }
+
+    await new Promise(resolve => setTimeout(resolve, 3000));
+  }
+}
 
 /**
  * GraphQL subscription resolver for the 'newBlocksFromDepth' subscription
@@ -47,34 +61,11 @@ const newBlocksFromDepthSubscriptionSchema = zod.object({
  * - Blocks on chains specified by the chainIds argument (or all chains if not specified)
  * - Blocks with a depth (confirmations) greater than or equal to minimumDepth
  *
- * This approach allows for a single PubSub event stream to serve multiple clients with
- * different filtering requirements.
  */
 export const newBlocksFromDepthSubscriptionResolver: SubscriptionResolvers<ResolverContext>['newBlocksFromDepth'] =
   {
-    resolve: async (payload: any, _args: any, context: ResolverContext) => {
-      const res = await (blockQueryResolver as any)({}, { hash: payload.hash }, context);
-      return [res];
-    },
-    subscribe: (_parent, args: SubscriptionNewBlocksFromDepthArgs, context) => {
-      return {
-        [Symbol.asyncIterator]: withFilter(
-          () => context.pubSub.asyncIterator(NEW_BLOCKS_FROM_DEPTH_EVENT),
-          payload => {
-            const res = newBlocksFromDepthSubscriptionSchema.safeParse(payload);
-            if (!res.success) {
-              console.error(
-                '[ERROR][API][BIZ_FLOW] Invalid payload on newBlocksFromDepthSubscription',
-                payload,
-              );
-              return false;
-            }
-            const { chainId, height } = res.data;
-            return (
-              (!args.chainIds || args.chainIds.includes(chainId)) && height >= args.minimumDepth
-            );
-          },
-        ),
-      };
+    resolve: (payload: any) => payload,
+    subscribe: (_root, args, context) => {
+      return iteratorFn(args.chainIds ?? [], args.minimumDepth, context);
     },
   };