Added the handling of begin and end block events

.changeset/long-hats-jam.md

@@ -0,0 +1,5 @@
+---
+'@confio/relayer': patch
+---
+
+Added the handling of begin and end block events

README.md

@@ -200,9 +200,9 @@ The `GET /metrics` endpoint will be exposed by default on port `8080`, which you
 
 ## Chain Requirements
 
-The blockchain must be based on Cosmos SDK `v0.41.1+`. In particular it must have
-[PR 8458](https://github.com/cosmos/cosmos-sdk/pull/8458) merged (if you are using a fork)
-in order for the relayer to work properly. `ibc-setup` should work on `v0.40.0+`
+The blockchain must be based on Cosmos SDK `v0.42.4+`. In particular it must have
+[PR 8458](https://github.com/cosmos/cosmos-sdk/pull/8458) and [PR 9081](https://github.com/cosmos/cosmos-sdk/pull/9081) 
+merged (if you are using a fork) in order for the relayer to work properly. `ibc-setup` should work on `v0.40.0+`
 
 The chain must have a large value for `staking.params.historical_entries` (often set in genesis).
 The default is "10000" and this should work with "1000", but no relayer will work if it is set to 0.

package.json

@@ -35,7 +35,7 @@
     "@cosmjs/proto-signing": "0.25.3",
     "@cosmjs/stargate": "0.25.3",
     "@cosmjs/stream": "0.25.3",
-    "@cosmjs/tendermint-rpc": "0.25.3",
+    "@cosmjs/tendermint-rpc": "https://gitpkg.now.sh/desmos-labs/cosmjs/packages/tendermint-rpc?riccardo/add-block-search",
     "@cosmjs/utils": "0.25.3",
     "ajv": "7.1.1",
     "axios": "0.21.1",

src/lib/endpoint.ts

@@ -5,7 +5,12 @@ import { CommitResponse } from '@cosmjs/tendermint-rpc';
 import { Packet } from '../codec/ibc/core/channel/v1/channel';
 
 import { IbcClient } from './ibcclient';
-import { Ack, parseAcksFromLogs, parsePacketsFromLogs } from './utils';
+import {
+  Ack,
+  parseAcksFromLogs,
+  parsePacketsFromBlockResult,
+  parsePacketsFromLogs,
+} from './utils';
 
 export interface PacketWithMetadata {
   packet: Packet;
@@ -54,8 +59,35 @@ export class Endpoint {
     return this.client.getCommit();
   }
 
-  // returns all packets (auto-paginates, so be careful about not setting a minHeight)
-  public async querySentPackets({
+  private async getPacketsFromBlockEvents({
+    minHeight,
+    maxHeight,
+  }: QueryOpts = {}): Promise<PacketWithMetadata[]> {
+    let query = `send_packet.packet_connection='${this.connectionID}'`;
+    if (minHeight) {
+      query = `${query} AND block.height>=${minHeight}`;
+    }
+    if (maxHeight) {
+      query = `${query} AND block.height<=${maxHeight}`;
+    }
+
+    const search = await this.client.tm.blockSearchAll({ query });
+    const resultsNested = await Promise.all(
+      search.blocks.map(async ({ block }) => {
+        const height = block.header.height;
+        const result = await this.client.tm.blockResults(height);
+        return parsePacketsFromBlockResult(result).map((packet) => ({
+          packet,
+          height,
+          sender: '',
+        }));
+      })
+    );
+
+    return ([] as PacketWithMetadata[]).concat(...resultsNested);
+  }
+
+  private async getPacketsFromTxs({
     minHeight,
     maxHeight,
   }: QueryOpts = {}): Promise<PacketWithMetadata[]> {
@@ -92,6 +124,21 @@ export class Endpoint {
     return ([] as PacketWithMetadata[]).concat(...resultsNested);
   }
 
+  // returns all packets (auto-paginates, so be careful about not setting a minHeight)
+  public async querySentPackets({
+    minHeight,
+    maxHeight,
+  }: QueryOpts = {}): Promise<PacketWithMetadata[]> {
+    const txsPackets = await this.getPacketsFromTxs({ minHeight, maxHeight });
+    const eventsPackets = await this.getPacketsFromBlockEvents({
+      minHeight,
+      maxHeight,
+    });
+    return ([] as PacketWithMetadata[])
+      .concat(...txsPackets)
+      .concat(...eventsPackets);
+  }
+
   // returns all acks (auto-paginates, so be careful about not setting a minHeight)
   public async queryWrittenAcks({
     minHeight,

src/lib/utils.ts

@@ -1,6 +1,8 @@
 import { fromUtf8, toHex, toUtf8 } from '@cosmjs/encoding';
 import { BroadcastTxFailure, Coin, logs, StdFee } from '@cosmjs/stargate';
+const { parseEvent } = logs;
 import {
+  BlockResultsResponse,
   ReadonlyDateWithNanoseconds,
   Header as RpcHeader,
   ValidatorPubkey as RpcPubKey,
@@ -193,6 +195,28 @@ interface ParsedEvent {
   readonly attributes: readonly ParsedAttribute[];
 }
 
+function decodeBase64(value: Uint8Array): string {
+  return Buffer.from(value).toString('binary');
+}
+
+export function parsePacketsFromBlockResult(
+  result: BlockResultsResponse
+): Packet[] {
+  const allEvents: ParsedEvent[] = result.beginBlockEvents
+    .concat(...result.endBlockEvents)
+    .filter(({ type }) => type === 'send_packet')
+    .map(({ type, attributes }) => ({
+      type,
+      attributes: attributes.map(({ key, value }) => ({
+        key: decodeBase64(key),
+        value: decodeBase64(value),
+      })),
+    }));
+
+  const flatEvents = ([] as ParsedEvent[]).concat(allEvents.map(parseEvent));
+  return flatEvents.map(parsePacket);
+}
+
 export function parsePacketsFromLogs(logs: readonly logs.Log[]): Packet[] {
   // grab all send_packet events from the logs
   const allEvents: ParsedEvent[][] = logs.map((log) =>

yarn.lock

@@ -511,6 +511,20 @@
     readonly-date "^1.0.0"
     xstream "^11.14.0"
 
+"@cosmjs/tendermint-rpc@https://gitpkg.now.sh/desmos-labs/cosmjs/packages/tendermint-rpc?riccardo/add-block-search":
+  version "0.25.3"
+  resolved "https://gitpkg.now.sh/desmos-labs/cosmjs/packages/tendermint-rpc?riccardo/add-block-search#03c68a35639b17d50646cc62c2cf9644cfdfd46d"
+  dependencies:
+    "@cosmjs/crypto" "^0.25.3"
+    "@cosmjs/encoding" "^0.25.3"
+    "@cosmjs/json-rpc" "^0.25.3"
+    "@cosmjs/math" "^0.25.3"
+    "@cosmjs/socket" "^0.25.3"
+    "@cosmjs/stream" "^0.25.3"
+    axios "^0.21.1"
+    readonly-date "^1.0.0"
+    xstream "^11.14.0"
+
 "@cosmjs/utils@0.25.3", "@cosmjs/utils@^0.25.3":
   version "0.25.3"
   resolved "https://registry.yarnpkg.com/@cosmjs/utils/-/utils-0.25.3.tgz#e74a85d0be74ef6a9cb3efa45fd219e459cdaa03"