feat(indexer/postgres)!: add basic support for header, txs and events…

… (backport #22695) (#22808) Co-authored-by: Facundo Medica <[email protected]> Co-authored-by: Julien Robert <[email protected]>
cosmos · Dec 9, 2024 · 9ec294a · 9ec294a
1 parent 10efb9c
commit 9ec294a
Show file tree

Hide file tree

Showing 62 changed files with 244 additions and 182 deletions.
diff --git a/baseapp/abci.go b/baseapp/abci.go
@@ -999,6 +999,10 @@ func (app *BaseApp) Commit() (*abci.CommitResponse, error) {
 				app.logger.Error("Commit listening hook failed", "height", blockHeight, "err", err)
 				if app.streamingManager.StopNodeOnErr {
 					err = fmt.Errorf("Commit listening hook failed: %w", err)
+					if blockHeight == 1 {
+						// can't rollback to height 0, so just return the error
+						return nil, fmt.Errorf("failed to commit block 1, can't automatically rollback: %w", err)
+					}
 					rollbackErr := app.cms.RollbackToVersion(blockHeight - 1)
 					if rollbackErr != nil {
 						return nil, errors.Join(err, rollbackErr)

diff --git a/baseapp/streaming.go b/baseapp/streaming.go
@@ -2,6 +2,7 @@ package baseapp
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sort"
 	"strconv"
@@ -20,6 +21,7 @@ import (
 
 	"github.com/cosmos/cosmos-sdk/client/flags"
 	servertypes "github.com/cosmos/cosmos-sdk/server/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
 const (
@@ -48,7 +50,7 @@ func (app *BaseApp) EnableIndexer(indexerOpts interface{}, keys map[string]*stor
 	app.cms.AddListeners(exposedKeys)
 
 	app.streamingManager = storetypes.StreamingManager{
-		ABCIListeners: []storetypes.ABCIListener{listenerWrapper{listener.Listener}},
+		ABCIListeners: []storetypes.ABCIListener{listenerWrapper{listener.Listener, app.txDecoder}},
 		StopNodeOnErr: true,
 	}
 
@@ -144,9 +146,10 @@ func exposeStoreKeysSorted(keysStr []string, keys map[string]*storetypes.KVStore
 	return exposeStoreKeys
 }
 
-func eventToAppDataEvent(event abci.Event) (appdata.Event, error) {
+func eventToAppDataEvent(event abci.Event, height int64) (appdata.Event, error) {
 	appdataEvent := appdata.Event{
-		Type: event.Type,
+		BlockNumber: uint64(height),
+		Type:        event.Type,
 		Attributes: func() ([]appdata.EventAttribute, error) {
 			attrs := make([]appdata.EventAttribute, len(event.Attributes))
 			for j, attr := range event.Attributes {
@@ -197,7 +200,8 @@ func eventToAppDataEvent(event abci.Event) (appdata.Event, error) {
 }
 
 type listenerWrapper struct {
-	listener appdata.Listener
+	listener  appdata.Listener
+	txDecoder sdk.TxDecoder
 }
 
 // NewListenerWrapper creates a new listenerWrapper.
@@ -208,20 +212,35 @@ func NewListenerWrapper(listener appdata.Listener) listenerWrapper {
 
 func (p listenerWrapper) ListenFinalizeBlock(_ context.Context, req abci.FinalizeBlockRequest, res abci.FinalizeBlockResponse) error {
 	if p.listener.StartBlock != nil {
+		// clean up redundant data
+		reqWithoutTxs := req
+		reqWithoutTxs.Txs = nil
+
 		if err := p.listener.StartBlock(appdata.StartBlockData{
 			Height:      uint64(req.Height),
 			HeaderBytes: nil, // TODO: https://github.com/cosmos/cosmos-sdk/issues/22009
-			HeaderJSON:  nil, // TODO: https://github.com/cosmos/cosmos-sdk/issues/22009
+			HeaderJSON: func() (json.RawMessage, error) {
+				return json.Marshal(reqWithoutTxs)
+			},
 		}); err != nil {
 			return err
 		}
 	}
 	if p.listener.OnTx != nil {
 		for i, tx := range req.Txs {
 			if err := p.listener.OnTx(appdata.TxData{
-				TxIndex: int32(i),
-				Bytes:   func() ([]byte, error) { return tx, nil },
-				JSON:    nil, // TODO: https://github.com/cosmos/cosmos-sdk/issues/22009
+				BlockNumber: uint64(req.Height),
+				TxIndex:     int32(i),
+				Bytes:       func() ([]byte, error) { return tx, nil },
+				JSON: func() (json.RawMessage, error) {
+					sdkTx, err := p.txDecoder(tx)
+					if err != nil {
+						// if the transaction cannot be decoded, return the error as JSON
+						// as there are some txs that might not be decodeable by the txDecoder
+						return json.Marshal(err)
+					}
+					return json.Marshal(sdkTx)
+				},
 			}); err != nil {
 				return err
 			}
@@ -231,14 +250,14 @@ func (p listenerWrapper) ListenFinalizeBlock(_ context.Context, req abci.Finaliz
 		events := make([]appdata.Event, len(res.Events))
 		var err error
 		for i, event := range res.Events {
-			events[i], err = eventToAppDataEvent(event)
+			events[i], err = eventToAppDataEvent(event, req.Height)
 			if err != nil {
 				return err
 			}
 		}
 		for _, txResult := range res.TxResults {
 			for _, event := range txResult.Events {
-				appdataEvent, err := eventToAppDataEvent(event)
+				appdataEvent, err := eventToAppDataEvent(event, req.Height)
 				if err != nil {
 					return err
 				}

diff --git a/client/v2/go.mod b/client/v2/go.mod
@@ -33,12 +33,12 @@ require (
 require (
 	buf.build/gen/go/cometbft/cometbft/protocolbuffers/go v1.35.2-20241120201313-68e42a58b301.1 // indirect
 	buf.build/gen/go/cosmos/gogo-proto/protocolbuffers/go v1.35.2-20240130113600-88ef6483f90f.1 // indirect
-	cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b // indirect
+	cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b // indirect
 	cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e // indirect
 	cosmossdk.io/errors v1.0.1 // indirect
 	cosmossdk.io/log v1.5.0
 	cosmossdk.io/math v1.4.0
-	cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b // indirect
+	cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b // indirect
 	cosmossdk.io/store v1.1.1-0.20240909133312-50288938d1b6 // indirect
 	cosmossdk.io/x/staking v0.0.0-00010101000000-000000000000 // indirect
 	filippo.io/edwards25519 v1.1.0 // indirect

diff --git a/client/v2/go.sum b/client/v2/go.sum
@@ -6,8 +6,8 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cosmossdk.io/api v0.7.3-0.20240924065902-eb7653cfecdf h1:CttA/mEIxGm4E7vwrjUpju7/Iespns08d9bOza70cIc=
 cosmossdk.io/api v0.7.3-0.20240924065902-eb7653cfecdf/go.mod h1:YMfx2ATpgITsoydD3hIBa8IkDHtyXp/14rmG0d3sEew=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b h1:MgU4EDOo/pXgepHCUFQFnIfUCxk/JO0AJGDTUQhhEhg=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b h1:smupoVhpdK+5pztIylyIGkCc+0QaAaGLEvnM7Wnrq18=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
 cosmossdk.io/core v1.0.0-alpha.6 h1:5ukC4JcQKmemLQXcAgu/QoOvJI50hpBkIIg4ZT2EN8E=
 cosmossdk.io/core v1.0.0-alpha.6/go.mod h1:3u9cWq1FAVtiiCrDPpo4LhR+9V6k/ycSG4/Y/tREWCY=
 cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e h1:F+ScucYxwrrDJU8guJXQXpGhdpziYSbxW6HMP2wCNxs=
@@ -20,8 +20,8 @@ cosmossdk.io/log v1.5.0 h1:dVdzPJW9kMrnAYyMf1duqacoidB9uZIl+7c6z0mnq0g=
 cosmossdk.io/log v1.5.0/go.mod h1:Tr46PUJjiUthlwQ+hxYtUtPn4D/oCZXAkYevBeh5+FI=
 cosmossdk.io/math v1.4.0 h1:XbgExXFnXmF/CccPPEto40gOO7FpWu9yWNAZPN3nkNQ=
 cosmossdk.io/math v1.4.0/go.mod h1:O5PkD4apz2jZs4zqFdTr16e1dcaQCc5z6lkEnrrppuk=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b h1:svpFdulZRrYz+RTHu2u9CeKkMKrIHx5354vjiHerovo=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b h1:ymo4PrM7Q4tPpEtce6IdH/kurphOAE6DmzHux1tniNY=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
 cosmossdk.io/store v1.0.0-rc.0.0.20241204123127-eb3bf8b0469d h1:KQM4Q6kjwlM4HuDZRV8/ZDXX3whjfStndYNTsRrbboQ=
 cosmossdk.io/store v1.0.0-rc.0.0.20241204123127-eb3bf8b0469d/go.mod h1:oZBBY4BrkYnghr6MFL0MP5mGqpkPedHcWkXwXddd6tU=
 cosmossdk.io/x/protocolpool v0.0.0-20230925135524-a1bc045b3190 h1:XQJj9Dv9Gtze0l2TF79BU5lkP6MkUveTUuKICmxoz+o=

diff --git a/codec/collections.go b/codec/collections.go
@@ -398,6 +398,7 @@ func protoCol(f protoreflect.FieldDescriptor) schema.Field {
 			col.Kind = schema.StringKind
 		case protoreflect.BytesKind:
 			col.Kind = schema.BytesKind
+			col.Nullable = true
 		case protoreflect.EnumKind:
 			// TODO: support enums
 			col.Kind = schema.EnumKind

diff --git a/go.mod b/go.mod
@@ -5,14 +5,14 @@ module github.com/cosmos/cosmos-sdk
 require (
 	buf.build/gen/go/cometbft/cometbft/protocolbuffers/go v1.35.2-20241120201313-68e42a58b301.1 // indirect
 	cosmossdk.io/api v0.8.0 // main
-	cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b // main
+	cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b // main
 	cosmossdk.io/core v1.0.0-alpha.6
 	cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e // main
 	cosmossdk.io/depinject v1.1.0
 	cosmossdk.io/errors v1.0.1
 	cosmossdk.io/log v1.5.0
 	cosmossdk.io/math v1.4.0
-	cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b
+	cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b
 	cosmossdk.io/store v1.1.1-0.20240909133312-50288938d1b6
 	cosmossdk.io/x/bank v0.0.0-20240226161501-23359a0b6d91
 	cosmossdk.io/x/staking v0.0.0-00010101000000-000000000000

diff --git a/go.sum b/go.sum
@@ -6,8 +6,8 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cosmossdk.io/api v0.7.3-0.20241127063259-f296a5005ce8 h1:KCi0Wq5M0JST0I01HebEDHt7//tZMx2bW4tmlc4IRnI=
 cosmossdk.io/api v0.7.3-0.20241127063259-f296a5005ce8/go.mod h1:vZy0Ev95gwANXt5ssiDui4L5nlMYO5bzqR77hCjIz9s=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b h1:MgU4EDOo/pXgepHCUFQFnIfUCxk/JO0AJGDTUQhhEhg=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b h1:smupoVhpdK+5pztIylyIGkCc+0QaAaGLEvnM7Wnrq18=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
 cosmossdk.io/core v1.0.0-alpha.6 h1:5ukC4JcQKmemLQXcAgu/QoOvJI50hpBkIIg4ZT2EN8E=
 cosmossdk.io/core v1.0.0-alpha.6/go.mod h1:3u9cWq1FAVtiiCrDPpo4LhR+9V6k/ycSG4/Y/tREWCY=
 cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e h1:F+ScucYxwrrDJU8guJXQXpGhdpziYSbxW6HMP2wCNxs=
@@ -20,8 +20,8 @@ cosmossdk.io/log v1.5.0 h1:dVdzPJW9kMrnAYyMf1duqacoidB9uZIl+7c6z0mnq0g=
 cosmossdk.io/log v1.5.0/go.mod h1:Tr46PUJjiUthlwQ+hxYtUtPn4D/oCZXAkYevBeh5+FI=
 cosmossdk.io/math v1.4.0 h1:XbgExXFnXmF/CccPPEto40gOO7FpWu9yWNAZPN3nkNQ=
 cosmossdk.io/math v1.4.0/go.mod h1:O5PkD4apz2jZs4zqFdTr16e1dcaQCc5z6lkEnrrppuk=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b h1:svpFdulZRrYz+RTHu2u9CeKkMKrIHx5354vjiHerovo=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b h1:ymo4PrM7Q4tPpEtce6IdH/kurphOAE6DmzHux1tniNY=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
 cosmossdk.io/store v1.0.0-rc.0.0.20241204123127-eb3bf8b0469d h1:KQM4Q6kjwlM4HuDZRV8/ZDXX3whjfStndYNTsRrbboQ=
 cosmossdk.io/store v1.0.0-rc.0.0.20241204123127-eb3bf8b0469d/go.mod h1:oZBBY4BrkYnghr6MFL0MP5mGqpkPedHcWkXwXddd6tU=
 cosmossdk.io/x/tx v1.0.0-alpha.2 h1:UW80FMm7B0fiAMsrfe5+HabSJ3XBg+tQa6/GK9prqWk=

diff --git a/server/v2/cometbft/abci.go b/server/v2/cometbft/abci.go
@@ -547,7 +547,7 @@ func (c *consensus[T]) FinalizeBlock(
 	events = append(events, resp.EndBlockEvents...)
 
 	// listen to state streaming changes in accordance with the block
-	err = c.streamDeliverBlockChanges(ctx, req.Height, req.Txs, resp.TxResults, events, stateChanges)
+	err = c.streamDeliverBlockChanges(ctx, req.Height, req.Txs, decodedTxs, resp.TxResults, events, stateChanges)
 	if err != nil {
 		return nil, err
 	}
@@ -590,7 +590,7 @@ func (c *consensus[T]) internalFinalizeBlock(
 	// TODO(tip): can we expect some txs to not decode? if so, what we do in this case? this does not seem to be the case,
 	// considering that prepare and process always decode txs, assuming they're the ones providing txs we should never
 	// have a tx that fails decoding.
-	decodedTxs, err := decodeTxs(req.Txs, c.txCodec)
+	decodedTxs, err := decodeTxs(c.logger, req.Txs, c.txCodec)
 	if err != nil {
 		return nil, nil, nil, err
 	}
@@ -723,12 +723,13 @@ func (c *consensus[T]) ExtendVote(ctx context.Context, req *abciproto.ExtendVote
 	return resp, err
 }
 
-func decodeTxs[T transaction.Tx](rawTxs [][]byte, codec transaction.Codec[T]) ([]T, error) {
+func decodeTxs[T transaction.Tx](logger log.Logger, rawTxs [][]byte, codec transaction.Codec[T]) ([]T, error) {
 	txs := make([]T, len(rawTxs))
 	for i, rawTx := range rawTxs {
 		tx, err := codec.Decode(rawTx)
 		if err != nil {
-			return nil, fmt.Errorf("unable to decode tx: %d: %w", i, err)
+			// do not return an error here, as we want to deliver the block even if some txs are invalid
+			logger.Debug("failed to decode tx", "err", err)
 		}
 		txs[i] = tx
 	}

diff --git a/server/v2/cometbft/go.mod b/server/v2/cometbft/go.mod
@@ -15,12 +15,12 @@ replace (
 
 require (
 	cosmossdk.io/api v0.8.0
-	cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b // main
+	cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b // main
 	cosmossdk.io/core v1.0.0-alpha.6
 	cosmossdk.io/errors v1.0.1
 	cosmossdk.io/errors/v2 v2.0.0-20240731132947-df72853b3ca5
 	cosmossdk.io/log v1.5.0
-	cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b //main
+	cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b //main
 	cosmossdk.io/server/v2 v2.0.0-20241209145349-34f407d6367a // main
 	cosmossdk.io/server/v2/appmanager v0.0.0-20241203212527-7d117425d880 // main
 	cosmossdk.io/server/v2/stf v0.0.0-20241204101618-7fa2356c07aa // main

diff --git a/server/v2/cometbft/go.sum b/server/v2/cometbft/go.sum
@@ -6,8 +6,8 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cosmossdk.io/api v0.7.3-0.20240924065902-eb7653cfecdf h1:CttA/mEIxGm4E7vwrjUpju7/Iespns08d9bOza70cIc=
 cosmossdk.io/api v0.7.3-0.20240924065902-eb7653cfecdf/go.mod h1:YMfx2ATpgITsoydD3hIBa8IkDHtyXp/14rmG0d3sEew=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b h1:MgU4EDOo/pXgepHCUFQFnIfUCxk/JO0AJGDTUQhhEhg=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b h1:smupoVhpdK+5pztIylyIGkCc+0QaAaGLEvnM7Wnrq18=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
 cosmossdk.io/core v1.0.0-alpha.6 h1:5ukC4JcQKmemLQXcAgu/QoOvJI50hpBkIIg4ZT2EN8E=
 cosmossdk.io/core v1.0.0-alpha.6/go.mod h1:3u9cWq1FAVtiiCrDPpo4LhR+9V6k/ycSG4/Y/tREWCY=
 cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e h1:F+ScucYxwrrDJU8guJXQXpGhdpziYSbxW6HMP2wCNxs=
@@ -22,8 +22,8 @@ cosmossdk.io/log v1.5.0 h1:dVdzPJW9kMrnAYyMf1duqacoidB9uZIl+7c6z0mnq0g=
 cosmossdk.io/log v1.5.0/go.mod h1:Tr46PUJjiUthlwQ+hxYtUtPn4D/oCZXAkYevBeh5+FI=
 cosmossdk.io/math v1.4.0 h1:XbgExXFnXmF/CccPPEto40gOO7FpWu9yWNAZPN3nkNQ=
 cosmossdk.io/math v1.4.0/go.mod h1:O5PkD4apz2jZs4zqFdTr16e1dcaQCc5z6lkEnrrppuk=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b h1:svpFdulZRrYz+RTHu2u9CeKkMKrIHx5354vjiHerovo=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b h1:ymo4PrM7Q4tPpEtce6IdH/kurphOAE6DmzHux1tniNY=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
 cosmossdk.io/server/v2 v2.0.0-20241209145349-34f407d6367a h1:qkO+rB9yD6+bTGgQpaf+oyvgEdkPs5TUaFK3OEYh3AI=
 cosmossdk.io/server/v2 v2.0.0-20241209145349-34f407d6367a/go.mod h1:sb6WEIMHAT+8z7iM6IbBeSf+62wSkss2q+coDxmOi/o=
 cosmossdk.io/server/v2/appmanager v0.0.0-20241203212527-7d117425d880 h1:0mtB8fSvDjD835WwWF4rGk9qy5TjVjk2jsW14L37v0E=

diff --git a/server/v2/cometbft/streaming.go b/server/v2/cometbft/streaming.go
@@ -2,6 +2,7 @@ package cometbft
 
 import (
 	"context"
+	"encoding/json"
 
 	"cosmossdk.io/core/event"
 	"cosmossdk.io/core/server"
@@ -16,6 +17,7 @@ func (c *consensus[T]) streamDeliverBlockChanges(
 	ctx context.Context,
 	height int64,
 	txs [][]byte,
+	decodedTxs []T,
 	txResults []server.TxResult,
 	events []event.Event,
 	stateChanges []store.StateChanges,
@@ -76,9 +78,12 @@ func (c *consensus[T]) streamDeliverBlockChanges(
 	if c.listener.OnTx != nil {
 		for i, tx := range txs {
 			if err := c.listener.OnTx(appdata.TxData{
-				TxIndex: int32(i),
-				Bytes:   func() ([]byte, error) { return tx, nil },
-				JSON:    nil, // TODO: https://github.com/cosmos/cosmos-sdk/issues/22009
+				BlockNumber: uint64(height),
+				TxIndex:     int32(i),
+				Bytes:       func() ([]byte, error) { return tx, nil },
+				JSON: func() (json.RawMessage, error) {
+					return json.Marshal(decodedTxs[i])
+				},
 			}); err != nil {
 				return err
 			}

diff --git a/simapp/go.mod b/simapp/go.mod
@@ -5,7 +5,7 @@ go 1.23.3
 require (
 	cosmossdk.io/api v0.8.0 // main
 	cosmossdk.io/client/v2 v2.0.0-20230630094428-02b760776860
-	cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b // main
+	cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b // main
 	cosmossdk.io/core v1.0.0-alpha.6 // main
 	cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e // main
 	cosmossdk.io/depinject v1.1.0
@@ -61,7 +61,7 @@ require (
 	cloud.google.com/go/iam v1.1.8 // indirect
 	cloud.google.com/go/storage v1.42.0 // indirect
 	cosmossdk.io/errors v1.0.1 // indirect
-	cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b // indirect
+	cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b // indirect
 	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect
 	github.com/99designs/keyring v1.2.2 // indirect

diff --git a/simapp/go.sum b/simapp/go.sum
@@ -194,8 +194,8 @@ cloud.google.com/go/workflows v1.6.0/go.mod h1:6t9F5h/unJz41YqfBmqSASJSXccBLtD1V
 cloud.google.com/go/workflows v1.7.0/go.mod h1:JhSrZuVZWuiDfKEFxU0/F1PQjmpnpcoISEXH2bcHC3M=
 cosmossdk.io/api v0.7.3-0.20241127063259-f296a5005ce8 h1:KCi0Wq5M0JST0I01HebEDHt7//tZMx2bW4tmlc4IRnI=
 cosmossdk.io/api v0.7.3-0.20241127063259-f296a5005ce8/go.mod h1:vZy0Ev95gwANXt5ssiDui4L5nlMYO5bzqR77hCjIz9s=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b h1:MgU4EDOo/pXgepHCUFQFnIfUCxk/JO0AJGDTUQhhEhg=
-cosmossdk.io/collections v0.4.1-0.20241128094659-bd76b47e1d8b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b h1:smupoVhpdK+5pztIylyIGkCc+0QaAaGLEvnM7Wnrq18=
+cosmossdk.io/collections v0.4.1-0.20241209183624-332d0b106d1b/go.mod h1:uf12i1yKvzEIHt2ok7poNqFDQTb71O00RQLitSynmrg=
 cosmossdk.io/core v1.0.0-alpha.6 h1:5ukC4JcQKmemLQXcAgu/QoOvJI50hpBkIIg4ZT2EN8E=
 cosmossdk.io/core v1.0.0-alpha.6/go.mod h1:3u9cWq1FAVtiiCrDPpo4LhR+9V6k/ycSG4/Y/tREWCY=
 cosmossdk.io/core/testing v0.0.0-20241108153815-606544c7be7e h1:F+ScucYxwrrDJU8guJXQXpGhdpziYSbxW6HMP2wCNxs=
@@ -211,8 +211,8 @@ cosmossdk.io/log v1.5.0/go.mod h1:Tr46PUJjiUthlwQ+hxYtUtPn4D/oCZXAkYevBeh5+FI=
 cosmossdk.io/math v1.4.0 h1:XbgExXFnXmF/CccPPEto40gOO7FpWu9yWNAZPN3nkNQ=
 cosmossdk.io/math v1.4.0/go.mod h1:O5PkD4apz2jZs4zqFdTr16e1dcaQCc5z6lkEnrrppuk=
 cosmossdk.io/schema v0.3.0/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b h1:svpFdulZRrYz+RTHu2u9CeKkMKrIHx5354vjiHerovo=
-cosmossdk.io/schema v0.3.1-0.20241128094659-bd76b47e1d8b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b h1:ymo4PrM7Q4tPpEtce6IdH/kurphOAE6DmzHux1tniNY=
+cosmossdk.io/schema v0.3.1-0.20241209183624-332d0b106d1b/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
 cosmossdk.io/store v1.0.0-rc.0.0.20241204123127-eb3bf8b0469d h1:KQM4Q6kjwlM4HuDZRV8/ZDXX3whjfStndYNTsRrbboQ=
 cosmossdk.io/store v1.0.0-rc.0.0.20241204123127-eb3bf8b0469d/go.mod h1:oZBBY4BrkYnghr6MFL0MP5mGqpkPedHcWkXwXddd6tU=
 cosmossdk.io/x/tx v1.0.0-alpha.2 h1:UW80FMm7B0fiAMsrfe5+HabSJ3XBg+tQa6/GK9prqWk=

diff --git a/simapp/v2/app_di.go b/simapp/v2/app_di.go
@@ -4,10 +4,13 @@ import (
 	_ "embed"
 	"fmt"
 
+	_ "github.com/jackc/pgx/v5/stdlib" // Import and register pgx driver
+
 	"cosmossdk.io/core/registry"
 	"cosmossdk.io/core/server"
 	"cosmossdk.io/core/transaction"
 	"cosmossdk.io/depinject"
+	_ "cosmossdk.io/indexer/postgres" // register the postgres indexer
 	"cosmossdk.io/log"
 	"cosmossdk.io/runtime/v2"
 	serverstore "cosmossdk.io/server/v2/store"