Improve block iteration, fix missing non-message events, add event type

packages/node/CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Changed
+- Update block type to allow access to decoded transactions, messages and events (#305)
 
 ## [4.3.0] - 2024-12-17
 ### Changed

packages/node/src/indexer/indexer.manager.ts

@@ -115,18 +115,39 @@ export class IndexerManager extends BaseIndexerManager<
       await this.indexEvent(evt, dataSources, getVM);
     }
 
+    // Group messages so we only iterate once.
+    const groupedMessages = blockContent.messages.reduce((acc, msg) => {
+      acc[msg.tx.hash] ??= [];
+      acc[msg.tx.hash].push(msg);
+      return acc;
+    }, {} as Record<string, CosmosMessage[]>);
+
+    // Group events so we only iterate once.
+    const groupedEvents = blockContent.events.reduce((acc, evt) => {
+      acc[evt.tx.hash] ??= {};
+      // -1 for events that arent associated with a message
+      const idxKey = evt.msg?.idx ?? -1;
+      acc[evt.tx.hash][idxKey] ??= [];
+      acc[evt.tx.hash][idxKey].push(evt);
+      return acc;
+    }, {} as Record<string /* TX hash*/, Record<number /* MSG index or -1 for tx event */, CosmosEvent[]>>);
+
     for (const tx of blockContent.transactions) {
       await this.indexTransaction(tx, dataSources, getVM);
-      const msgs = blockContent.messages.filter(
-        (msg) => msg.tx.hash === tx.hash,
-      );
-      for (const msg of msgs) {
+
+      const txEvents = groupedEvents[tx.hash];
+
+      for (const msg of groupedMessages[tx.hash]) {
         await this.indexMessage(msg, dataSources, getVM);
-        const events = blockContent.events.filter(
-          (event) => event.tx.hash === tx.hash && event.msg?.idx === msg.idx,
-        );
+        if (txEvents?.[msg.idx]) {
+          for (const evt of txEvents[msg.idx]) {
+            await this.indexEvent(evt, dataSources, getVM);
+          }
+        }
+      }
 
-        for (const evt of events) {
+      if (txEvents?.[-1]) {
+        for (const evt of txEvents[-1]) {
           await this.indexEvent(evt, dataSources, getVM);
         }
       }

packages/node/src/utils/cosmos.spec.ts

@@ -121,11 +121,11 @@ describe('CosmosUtils', () => {
       expect(event.idx).toEqual(17);
 
       expect(event.msg).toBeDefined();
-      expect(event.msg.msg.typeUrl).toEqual(
+      expect(event.msg?.msg.typeUrl).toEqual(
         '/cosmwasm.wasm.v1.MsgExecuteContract',
       );
 
-      expect(event.msg.tx.hash).toEqual(event.tx.hash);
+      expect(event.msg?.tx.hash).toEqual(event.tx.hash);
 
       expect(event.event).toBeDefined();
       expect(event.event.type).toEqual(
@@ -369,7 +369,7 @@ describe('Cosmos 0.50 support', () => {
     const registry = new Registry([...defaultRegistryTypes, ...wasmTypes]);
     api = new CosmosClient(client, registry);
 
-    const [firstBlock] = await fetchBlocksBatches(api, [12_495_419]); // https://www.mintscan.io/neutron/block/12495419
+    const [firstBlock] = await fetchBlocksBatches(api, [19_091_812]); // https://www.mintscan.io/neutron/block/12495419
     block = firstBlock.block;
   });
 
@@ -393,11 +393,11 @@ describe('Cosmos 0.50 support', () => {
     expect(event.idx).toEqual(0);
 
     expect(event.msg).toBeDefined();
-    expect(event.msg.msg.typeUrl).toEqual(
+    expect(event.msg?.msg.typeUrl).toEqual(
       '/ibc.core.client.v1.MsgUpdateClient',
     );
 
-    expect(event.msg.tx.hash).toEqual(event.tx.hash);
+    expect(event.msg?.tx.hash).toEqual(event.tx.hash);
 
     expect(event.event).toBeDefined();
     expect(event.event.type).toEqual('message');
@@ -406,6 +406,23 @@ describe('Cosmos 0.50 support', () => {
     expect(event.log.events.length).toEqual(0);
   });
 
+  it('Correctly wraps events not associated to a message', async () => {
+    const [{ block }] = await fetchBlocksBatches(api, [19_091_812]);
+
+    expect(block.events.length).toBe(279);
+
+    expect(block.transactions[0].tx.events.length).toBe(21);
+
+    const txEvents = block.events.filter(
+      (evt) =>
+        evt.tx.hash ===
+        '5F5DC2EECF1D8EDDE07BC0AD4F91A48BEB35E2A0D813BD2D21EA90B85F0BAB95',
+    );
+    expect(txEvents.length).toBe(21);
+    const nonMessageTxs = txEvents.filter((evt) => evt.msg === undefined);
+    expect(nonMessageTxs.length).toBe(16);
+  });
+
   // block.tx when block.block.tx cannot be decoded
   // {
   //    code: 2,

packages/node/src/utils/cosmos.ts

@@ -29,12 +29,13 @@ import {
   CosmosMessageFilter,
   CosmosBlock,
   CosmosEvent,
+  CosmosEventKind,
   CosmosTransaction,
   CosmosMessage,
   CosmosBlockFilter,
   CosmosTxFilter,
 } from '@subql/types-cosmos';
-import { isObjectLike } from 'lodash';
+import { isObjectLike, omit } from 'lodash';
 import { isLong } from 'long';
 import { SubqlProjectBlockFilter } from '../configure/SubqueryProject';
 import { CosmosClient } from '../indexer/api.service';
@@ -264,15 +265,6 @@ export async function fetchCosmosBlocksArray(
   );
 }
 
-export function wrapBlock(block: BlockResponse, txs: TxData[]): CosmosBlock {
-  return {
-    blockId: block.blockId,
-    block: { id: toHex(block.blockId.hash).toUpperCase(), ...block.block },
-    header: block.block.header,
-    txs: txs,
-  };
-}
-
 export function wrapTx(
   block: CosmosBlock,
   txResults: TxData[],
@@ -342,6 +334,7 @@ export function wrapBlockBeginAndEndEvents(
   block: CosmosBlock,
   events: TxEvent[],
   idxOffset: number,
+  kind: CosmosEventKind,
 ): CosmosEvent[] {
   return events.map(
     (event) =>
@@ -352,6 +345,7 @@ export function wrapBlockBeginAndEndEvents(
         msg: null,
         tx: null,
         log: null,
+        kind,
       } as unknown as CosmosEvent),
   );
 }
@@ -371,14 +365,20 @@ export function wrapEvent(
 ): CosmosEvent[] {
   const events: CosmosEvent[] = [];
   for (const tx of txs) {
-    const appendEvent = (msg: CosmosMessage, event: TxEvent, log: Log) => {
+    const appendEvent = (
+      msg: CosmosMessage | undefined,
+      event: TxEvent,
+      log: Log,
+      kind: CosmosEventKind,
+    ) => {
       events.push({
         idx: idxOffset++,
         block,
         tx,
         msg,
         event,
         log,
+        kind,
       });
     };
 
@@ -410,32 +410,30 @@ export function wrapEvent(
           continue;
         }
         for (let i = 0; i < log.events.length; i++) {
-          appendEvent(msg, log.events[i], log);
+          appendEvent(msg, log.events[i], log, CosmosEventKind.Message);
         }
       }
     } else if (tx.tx?.events) {
       // Comet38
       for (const txEvent of tx.tx.events) {
-        let msg: CosmosMessage;
-        try {
-          const eventMsgIndex = txEvent.attributes.find(
-            (attr) => attrToString(attr.key) === 'msg_index',
-          )?.value;
-
-          // Event doesn't have a message
-          if (eventMsgIndex === undefined) {
-            continue;
-          }
+        let msg: CosmosMessage | undefined;
+        const eventMsgIndex = txEvent.attributes.find(
+          (attr) => attrToString(attr.key) === 'msg_index',
+        )?.value;
 
+        // Event doesn't have a message
+        if (eventMsgIndex !== undefined) {
           const msgNumber = parseInt(attrToString(eventMsgIndex), 10);
           msg = wrapCosmosMsg(block, tx, msgNumber, registry);
-        } catch (e) {
-          logger.warn(`Unable to find message for event. tx=${tx.hash}`);
-          continue;
         }
 
         // TODO does a log still exist in Comet38?
-        appendEvent(msg, txEvent, { events: [], log: '', msg_index: -1 });
+        appendEvent(
+          msg,
+          txEvent,
+          { events: [], log: '', msg_index: -1 },
+          msg ? CosmosEventKind.Message : CosmosEventKind.Transaction,
+        );
       }
     } else {
       // For some tests that have invalid data
@@ -511,9 +509,34 @@ export class LazyBlockContent implements BlockContent {
 
   get block(): CosmosBlock {
     if (!this._wrappedBlock) {
-      this._wrappedBlock = wrapBlock(this._blockInfo, [
-        ...this._results.results,
-      ]);
+      // Need to keep reference to LazyBlockContent for the getter methods
+      // eslint-disable-next-line @typescript-eslint/no-this-alias
+      const self = this;
+
+      this._wrappedBlock = {
+        blockId: this._blockInfo.blockId,
+        block: {
+          id: toHex(this._blockInfo.blockId.hash).toUpperCase(),
+          ...this._blockInfo.block,
+        },
+        header: this._blockInfo.block.header,
+        txs: [...this._results.results],
+
+        get transactions() {
+          return self.transactions;
+        },
+        get messages() {
+          return self.messages;
+        },
+        get events() {
+          return [
+            ...self.beginBlockEvents,
+            ...self.events,
+            ...self.endBlockEvents,
+            ...self.finalizeBlockEvents,
+          ];
+        },
+      } as CosmosBlock;
     }
     return this._wrappedBlock;
   }
@@ -562,6 +585,7 @@ export class LazyBlockContent implements BlockContent {
         this.block,
         [...results.beginBlockEvents],
         this._eventIdx,
+        CosmosEventKind.BeginBlock,
       );
       this._eventIdx += this._wrappedBeginBlockEvents.length;
     }
@@ -582,6 +606,7 @@ export class LazyBlockContent implements BlockContent {
         this.block,
         [...results.endBlockEvents],
         this._eventIdx,
+        CosmosEventKind.EndBlock,
       );
       this._eventIdx += this._wrappedEndBlockEvents.length;
     }
@@ -600,6 +625,7 @@ export class LazyBlockContent implements BlockContent {
         this.block,
         [...results.finalizeBlockEvents],
         this._eventIdx,
+        CosmosEventKind.FinalizeBlock,
       );
       this._eventIdx += this._wrappedFinalizedBlockEvents.length;
     }

packages/types/CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Changed
+- Update block type to allow access to decoded transactions, messages and events (#305)
 
 ## [4.0.0] - 2024-10-23
 ### Changed

packages/types/src/interfaces.ts

@@ -22,7 +22,22 @@ export interface CosmosBlock {
   blockId: BlockId;
   block: {id: string} & Block;
   header: Header; // Full header
+  /* The raw transaction data */
   txs: TxData[];
+
+  /**
+   * Decoded transactions, this is the same data as passed to a transaction handler
+   * */
+  transactions: CosmosTransaction[];
+  /**
+   * Decoded messages, this is the same data as passed to a message handler
+   * */
+  messages: CosmosMessage[];
+  /**
+   * Decoded events, this is the same data as passed to a event handler.
+   * This is all events including, beginBlockEvents, endBlockEvents and finalizedBlockEvents
+   * */
+  events: CosmosEvent[];
 }
 
 export interface CosmosTransaction {
@@ -46,13 +61,22 @@ export interface CosmosMessage<T = any> {
   };
 }
 
+export enum CosmosEventKind {
+  BeginBlock = 'begin_block',
+  EndBlock = 'end_block',
+  FinalizeBlock = 'finalize_block',
+  Message = 'message',
+  Transaction = 'transaction',
+}
+
 export interface CosmosEvent {
   idx: number;
   block: CosmosBlock;
   tx: CosmosTransaction;
-  msg: CosmosMessage;
+  msg?: CosmosMessage;
   log: Log;
   event: TxEvent;
+  kind: CosmosEventKind;
 }
 
 export type DynamicDatasourceCreator = (name: string, args: Record<string, unknown>) => Promise<void>;