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

baseapp/abci.go

@@ -997,6 +997,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)

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
 				}

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

go.mod

@@ -186,6 +186,7 @@ require (
 // TODO remove after all modules have their own go.mods
 replace (
 	cosmossdk.io/api => ./api
+	cosmossdk.io/schema => ./schema
 	cosmossdk.io/store => ./store
 	cosmossdk.io/x/bank => ./x/bank
 	cosmossdk.io/x/staking => ./x/staking

go.sum

@@ -18,8 +18,6 @@ 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=
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
 github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 h1:/vQbFIOMbk2FiG/kXiLl8BRyzTWDw7gX/Hz7Dd5eDMs=

indexer/postgres/base_sql.go

@@ -17,17 +17,19 @@ CREATE TABLE IF NOT EXISTS tx
     id             BIGSERIAL PRIMARY KEY,
     block_number   BIGINT NOT NULL REFERENCES block (number),
     index_in_block BIGINT NOT NULL,
-    data           JSONB  NOT NULL
+    data           JSONB NULL,
+    bytes          BYTEA NULL 
 );
 
 CREATE TABLE IF NOT EXISTS event
 (
     id           BIGSERIAL PRIMARY KEY,
     block_number BIGINT NOT NULL REFERENCES block (number),
-    tx_id        BIGINT NULL REFERENCES tx (id),
-    msg_index    BIGINT NULL,
-    event_index  BIGINT NULL,
-    type         TEXT   NOT NULL,
-    data         JSONB  NOT NULL
+    block_stage  INTEGER NOT NULL,
+    tx_index     BIGINT NOT NULL,
+    msg_index    BIGINT NOT NULL,
+    event_index  BIGINT NOT NULL,
+    type         TEXT   NULL,
+    data         JSONB  NULL
 );
 `

indexer/postgres/listener.go

@@ -1,6 +1,7 @@
 package postgres
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"cosmossdk.io/schema/appdata"
@@ -81,5 +82,72 @@ func (i *indexerImpl) listener() appdata.Listener {
 			i.tx, err = i.db.BeginTx(i.ctx, nil)
 			return nil, err
 		},
+		OnTx:    txListener(i),
+		OnEvent: eventListener(i),
+	}
+}
+
+func txListener(i *indexerImpl) func(data appdata.TxData) error {
+	return func(td appdata.TxData) error {
+		var bz []byte
+		if td.Bytes != nil {
+			var err error
+			bz, err = td.Bytes()
+			if err != nil {
+				return err
+			}
+		}
+
+		var jsonData json.RawMessage
+		if td.JSON != nil {
+			var err error
+			jsonData, err = td.JSON()
+			if err != nil {
+				return err
+			}
+		}
+
+		_, err := i.tx.Exec("INSERT INTO tx (block_number, index_in_block, data, bytes) VALUES ($1, $2, $3, $4)",
+			td.BlockNumber, td.TxIndex, jsonData, bz)
+
+		return err
+	}
+}
+
+func eventListener(i *indexerImpl) func(data appdata.EventData) error {
+	return func(data appdata.EventData) error {
+		for _, e := range data.Events {
+			var jsonData json.RawMessage
+
+			if e.Data != nil {
+				var err error
+				jsonData, err = e.Data()
+				if err != nil {
+					return fmt.Errorf("failed to get event data: %w", err)
+				}
+			} else if e.Attributes != nil {
+				attrs, err := e.Attributes()
+				if err != nil {
+					return fmt.Errorf("failed to get event attributes: %w", err)
+				}
+
+				attrsMap := map[string]interface{}{}
+				for _, attr := range attrs {
+					attrsMap[attr.Key] = attr.Value
+				}
+
+				jsonData, err = json.Marshal(attrsMap)
+				if err != nil {
+					return fmt.Errorf("failed to marshal event attributes: %w", err)
+				}
+			}
+
+			_, err := i.tx.Exec("INSERT INTO event (block_number, block_stage, tx_index, msg_index, event_index, type, data) VALUES ($1, $2, $3, $4, $5, $6, $7)",
+				e.BlockNumber, e.BlockStage, e.TxIndex, e.MsgIndex, e.EventIndex, e.Type, jsonData)
+			if err != nil {
+				return fmt.Errorf("failed to index event: %w", err)
+			}
+		}
+		return nil
 	}
 }

runtime/v2/app.go

@@ -94,6 +94,11 @@ func (a *App[T]) SchemaDecoderResolver() decoding.DecoderResolver {
 	for moduleName, module := range a.moduleManager.Modules() {
 		moduleSet[moduleName] = module
 	}
+
+	for _, overrideKey := range a.config.OverrideStoreKeys {
+		moduleSet[overrideKey.KvStoreKey] = moduleSet[overrideKey.ModuleName]
+	}
+
 	return decoding.ModuleSetDecoderResolver(moduleSet)
 }
 

runtime/v2/go.mod

@@ -5,6 +5,7 @@ go 1.23
 // server v2 integration
 replace (
 	cosmossdk.io/api => ../../api
+	cosmossdk.io/schema => ../../schema
 	cosmossdk.io/core/testing => ../../core/testing
 	cosmossdk.io/server/v2/appmanager => ../../server/v2/appmanager
 	cosmossdk.io/server/v2/stf => ../../server/v2/stf

schema/appdata/data.go

@@ -31,6 +31,9 @@ type StartBlockData struct {
 
 // TxData represents the raw transaction data that is passed to a listener.
 type TxData struct {
+	// BlockNumber is the block number to which this event is associated.
+	BlockNumber uint64
+
 	// TxIndex is the index of the transaction in the block.
 	TxIndex int32
 
@@ -53,6 +56,9 @@ type Event struct {
 	// If the block stage is unknown, it should be set to UnknownBlockStage.
 	BlockStage BlockStage
 
+	// BlockNumber is the block number to which this event is associated.
+	BlockNumber uint64
+
 	// TxIndex is the 1-based index of the transaction in the block to which this event is associated.
 	// If TxIndex is zero, it means that we do not know the transaction index.
 	// Otherwise, the index should start with 1.

schema/decoding/middleware.go

@@ -40,7 +40,8 @@ func Middleware(target appdata.Listener, resolver DecoderResolver, opts Middlewa
 				var err error
 				moduleName, err = resolver.DecodeModuleName(kvUpdate.Actor)
 				if err != nil {
-					return err
+					// we don't have a codec for this module so continue
+					continue
 				}
 
 				moduleNames[string(kvUpdate.Actor)] = moduleName

server/v2/cometbft/abci.go

@@ -538,7 +538,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
 	}
@@ -575,7 +575,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
 	}
@@ -708,12 +708,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
 	}

server/v2/cometbft/abci_test.go

@@ -2,32 +2,32 @@ package cometbft
 
 import (
 	"context"
-	"cosmossdk.io/core/server"
 	"crypto/sha256"
 	"encoding/json"
 	"errors"
-	abci "github.com/cometbft/cometbft/abci/types"
 	"io"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
-	"cosmossdk.io/server/v2/cometbft/oe"
+	abci "github.com/cometbft/cometbft/abci/types"
 	abciproto "github.com/cometbft/cometbft/api/cometbft/abci/v1"
 	v1 "github.com/cometbft/cometbft/api/cometbft/types/v1"
 	"github.com/cosmos/gogoproto/proto"
 	gogotypes "github.com/cosmos/gogoproto/types"
 	"github.com/stretchr/testify/require"
 
 	appmodulev2 "cosmossdk.io/core/appmodule/v2"
+	"cosmossdk.io/core/server"
 	"cosmossdk.io/core/store"
 	"cosmossdk.io/core/transaction"
 	"cosmossdk.io/log"
 	"cosmossdk.io/server/v2/appmanager"
 	"cosmossdk.io/server/v2/cometbft/handlers"
 	cometmock "cosmossdk.io/server/v2/cometbft/internal/mock"
 	"cosmossdk.io/server/v2/cometbft/mempool"
+	"cosmossdk.io/server/v2/cometbft/oe"
 	"cosmossdk.io/server/v2/cometbft/types"
 	"cosmossdk.io/server/v2/stf"
 	"cosmossdk.io/server/v2/stf/branch"
@@ -60,10 +60,10 @@ func getQueryRouterBuilder[T any, PT interface {
 	*T
 	proto.Message
 },
-U any, UT interface {
-	*U
-	proto.Message
-}](
+	U any, UT interface {
+		*U
+		proto.Message
+	}](
 	t *testing.T,
 	handler func(ctx context.Context, msg PT) (UT, error),
 ) *stf.MsgRouterBuilder {
@@ -90,10 +90,10 @@ func getMsgRouterBuilder[T any, PT interface {
 	*T
 	transaction.Msg
 },
-U any, UT interface {
-	*U
-	transaction.Msg
-}](
+	U any, UT interface {
+		*U
+		transaction.Msg
+	}](
 	t *testing.T,
 	handler func(ctx context.Context, msg PT) (UT, error),
 ) *stf.MsgRouterBuilder {