feat: add event-query-tx-for cmd to subscribe and wait for transaction (backport #17274) (#17435)

Co-authored-by: mmsqe <[email protected]>
Co-authored-by: Julien Robert <[email protected]>
Co-authored-by: marbar3778 <[email protected]>

Author: 4 people

CHANGELOG.md

@@ -39,6 +39,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ### Features
 
+* (client/rpc) [#17274](https://github.com/cosmos/cosmos-sdk/pull/17274) Add `QueryEventForTxCmd` cmd to subscribe and wait event for transaction by hash.
 * (keyring) [#17424](https://github.com/cosmos/cosmos-sdk/pull/17424) Allows to import private keys encoded in hex. 
 
 ### Improvements

client/rpc/tx.go

@@ -0,0 +1,140 @@
+package rpc
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+	"strings"
+	"time"
+
+	rpchttp "github.com/cometbft/cometbft/rpc/client/http"
+	coretypes "github.com/cometbft/cometbft/rpc/core/types"
+	tmtypes "github.com/cometbft/cometbft/types"
+	"github.com/spf13/cobra"
+
+	"github.com/cosmos/cosmos-sdk/client"
+	"github.com/cosmos/cosmos-sdk/client/flags"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/cosmos/cosmos-sdk/types/errors"
+)
+
+func newTxResponseCheckTx(res *coretypes.ResultBroadcastTxCommit) *sdk.TxResponse {
+	if res == nil {
+		return nil
+	}
+
+	var txHash string
+	if res.Hash != nil {
+		txHash = res.Hash.String()
+	}
+
+	parsedLogs, _ := sdk.ParseABCILogs(res.CheckTx.Log)
+
+	return &sdk.TxResponse{
+		Height:    res.Height,
+		TxHash:    txHash,
+		Codespace: res.CheckTx.Codespace,
+		Code:      res.CheckTx.Code,
+		Data:      strings.ToUpper(hex.EncodeToString(res.CheckTx.Data)),
+		RawLog:    res.CheckTx.Log,
+		Logs:      parsedLogs,
+		Info:      res.CheckTx.Info,
+		GasWanted: res.CheckTx.GasWanted,
+		GasUsed:   res.CheckTx.GasUsed,
+		Events:    res.CheckTx.Events,
+	}
+}
+
+func newTxResponseDeliverTx(res *coretypes.ResultBroadcastTxCommit) *sdk.TxResponse {
+	if res == nil {
+		return nil
+	}
+
+	var txHash string
+	if res.Hash != nil {
+		txHash = res.Hash.String()
+	}
+
+	parsedLogs, _ := sdk.ParseABCILogs(res.DeliverTx.Log)
+
+	return &sdk.TxResponse{
+		Height:    res.Height,
+		TxHash:    txHash,
+		Codespace: res.DeliverTx.Codespace,
+		Code:      res.DeliverTx.Code,
+		Data:      strings.ToUpper(hex.EncodeToString(res.DeliverTx.Data)),
+		RawLog:    res.DeliverTx.Log,
+		Logs:      parsedLogs,
+		Info:      res.DeliverTx.Info,
+		GasWanted: res.DeliverTx.GasWanted,
+		GasUsed:   res.DeliverTx.GasUsed,
+		Events:    res.DeliverTx.Events,
+	}
+}
+
+func newResponseFormatBroadcastTxCommit(res *coretypes.ResultBroadcastTxCommit) *sdk.TxResponse {
+	if res == nil {
+		return nil
+	}
+
+	if !res.CheckTx.IsOK() {
+		return newTxResponseCheckTx(res)
+	}
+
+	return newTxResponseDeliverTx(res)
+}
+
+// QueryEventForTxCmd returns a CLI command that subscribes to a WebSocket connection and waits for a transaction event with the given hash.
+func QueryEventForTxCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "event-query-tx-for [hash]",
+		Short: "Query for a transaction by hash",
+		Long:  `Subscribes to a CometBFT WebSocket connection and waits for a transaction event with the given hash.`,
+		Args:  cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			clientCtx, err := client.GetClientTxContext(cmd)
+			if err != nil {
+				return err
+			}
+			c, err := rpchttp.New(clientCtx.NodeURI, "/websocket")
+			if err != nil {
+				return err
+			}
+			if err := c.Start(); err != nil {
+				return err
+			}
+			defer c.Stop() //nolint:errcheck // ignore stop error
+
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
+			defer cancel()
+
+			hash := args[0]
+			query := fmt.Sprintf("%s='%s' AND %s='%s'", tmtypes.EventTypeKey, tmtypes.EventTx, tmtypes.TxHashKey, hash)
+			const subscriber = "subscriber"
+			eventCh, err := c.Subscribe(ctx, subscriber, query)
+			if err != nil {
+				return fmt.Errorf("failed to subscribe to tx: %w", err)
+			}
+			defer c.UnsubscribeAll(context.Background(), subscriber) //nolint:errcheck // ignore unsubscribe error
+
+			select {
+			case evt := <-eventCh:
+				if txe, ok := evt.Data.(tmtypes.EventDataTx); ok {
+					res := &coretypes.ResultBroadcastTxCommit{
+						DeliverTx: txe.Result,
+						Hash:      tmtypes.Tx(txe.Tx).Hash(),
+						Height:    txe.Height,
+					}
+					return clientCtx.PrintProto(newResponseFormatBroadcastTxCommit(res))
+				}
+			case <-ctx.Done():
+				return errors.ErrLogic.Wrapf("timed out waiting for event, the transaction could have already been included or wasn't yet included")
+			}
+			return nil
+		},
+	}
+
+	flags.AddTxFlagsToCmd(cmd)
+
+	return cmd
+}

simapp/simd/cmd/root.go

@@ -214,6 +214,7 @@ func queryCommand() *cobra.Command {
 	}
 
 	cmd.AddCommand(
+		rpc.QueryEventForTxCmd(),
 		authcmd.GetAccountCmd(),
 		rpc.ValidatorCommand(),
 		rpc.BlockCommand(),