feat(schema): indexing API

baseapp/streaming.go

@@ -1,17 +1,25 @@
 package baseapp
 
 import (
+	"context"
 	"fmt"
 	"sort"
 	"strings"
 
 	"github.com/spf13/cast"
 
+	"cosmossdk.io/schema/indexer"
+
+	"cosmossdk.io/schema"
+	"cosmossdk.io/schema/appdata"
+	"cosmossdk.io/schema/decoding"
 	"cosmossdk.io/store/streaming"
 	storetypes "cosmossdk.io/store/types"
 
 	"github.com/cosmos/cosmos-sdk/client/flags"
 	servertypes "github.com/cosmos/cosmos-sdk/server/types"
+
+	abci "github.com/cometbft/cometbft/api/cometbft/abci/v1"
 )
 
 const (
@@ -22,6 +30,31 @@
 	StreamingABCIStopNodeOnErrTomlKey = "stop-node-on-err"
 )
 
+// EnableIndexer enables the built-in indexer with the provided options (usually from the app.toml indexer key),
+// kv-store keys, and app modules. Using the built-in indexer framework is mutually exclusive from using other
+// types of streaming listeners.
+func (app *BaseApp) EnableIndexer(indexerOpts interface{}, keys map[string]*storetypes.KVStoreKey, appModules map[string]any) error {
+	listener, err := indexer.StartManager(indexer.ManagerOptions{
+		Config:     indexerOpts,
+		Resolver:   decoding.ModuleSetDecoderResolver(appModules),
+		SyncSource: nil,
+		Logger:     app.logger.With("module", "indexer"),
+	})
+	if err != nil {
+		return err
+	}
+
+	exposedKeys := exposeStoreKeysSorted([]string{"*"}, keys)
+	app.cms.AddListeners(exposedKeys)
+
+	app.streamingManager = storetypes.StreamingManager{
+		ABCIListeners: []storetypes.ABCIListener{listenerWrapper{listener}},
+		StopNodeOnErr: true,
+	}
+
+	return nil
+}
+
 // RegisterStreamingServices registers streaming services with the BaseApp.
 func (app *BaseApp) RegisterStreamingServices(appOpts servertypes.AppOptions, keys map[string]*storetypes.KVStoreKey) error {
 	// register streaming services
@@ -110,3 +143,51 @@
 
 	return exposeStoreKeys
 }
+
+type listenerWrapper struct {
+	listener appdata.Listener
+}
+
+func (p listenerWrapper) ListenFinalizeBlock(_ context.Context, req abci.FinalizeBlockRequest, res abci.FinalizeBlockResponse) error {
+	if p.listener.StartBlock != nil {
+		err := p.listener.StartBlock(appdata.StartBlockData{
+			Height: uint64(req.Height),
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	//// TODO txs, events
+
+	return nil
+}
+
+func (p listenerWrapper) ListenCommit(ctx context.Context, res abci.CommitResponse, changeSet []*storetypes.StoreKVPair) error {
+	if cb := p.listener.OnKVPair; cb != nil {
+		updates := make([]appdata.ModuleKVPairUpdate, len(changeSet))
+		for i, pair := range changeSet {
+			updates[i] = appdata.ModuleKVPairUpdate{
+				ModuleName: pair.StoreKey,
+				Update: schema.KVPairUpdate{
+					Key:    pair.Key,
+					Value:  pair.Value,
+					Delete: pair.Delete,
+				},
+			}
+		}
+		err := cb(appdata.KVPairData{Updates: updates})
+		if err != nil {
+			return err
+		}
+	}
+
+	if p.listener.Commit != nil {
+		err := p.listener.Commit(appdata.CommitData{})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

go.mod

@@ -11,6 +11,7 @@ require (
 	cosmossdk.io/errors v1.0.1
 	cosmossdk.io/log v1.3.1
 	cosmossdk.io/math v1.3.0
+	cosmossdk.io/schema v0.0.0
 	cosmossdk.io/store v1.1.1-0.20240418092142-896cdf1971bc
 	cosmossdk.io/x/auth v0.0.0-00010101000000-000000000000
 	cosmossdk.io/x/bank v0.0.0-20240226161501-23359a0b6d91
@@ -190,6 +191,7 @@ replace (
 	cosmossdk.io/core/testing => ./core/testing
 	cosmossdk.io/depinject => ./depinject
 	cosmossdk.io/log => ./log
+	cosmossdk.io/schema => ./schema
 	cosmossdk.io/store => ./store
 	cosmossdk.io/x/accounts => ./x/accounts
 	cosmossdk.io/x/auth => ./x/auth

schema/decoding/resolver.go

@@ -0,0 +1,66 @@
+package decoding
+
+import (
+	"sort"
+
+	"cosmossdk.io/schema"
+)
+
+// DecoderResolver is an interface that allows indexers to discover and use module decoders.
+type DecoderResolver interface {
+	// IterateAll iterates over all available module decoders.
+	IterateAll(func(moduleName string, cdc schema.ModuleCodec) error) error
+
+	// LookupDecoder looks up a specific module decoder.
+	LookupDecoder(moduleName string) (decoder schema.ModuleCodec, found bool, err error)
+}
+
+// ModuleSetDecoderResolver returns DecoderResolver that will discover modules implementing
+// DecodeableModule in the provided module set.
+func ModuleSetDecoderResolver(moduleSet map[string]interface{}) DecoderResolver {
+	return &moduleSetDecoderResolver{
+		moduleSet: moduleSet,
+	}
+}
+
+type moduleSetDecoderResolver struct {
+	moduleSet map[string]interface{}
+}
+
+func (a moduleSetDecoderResolver) IterateAll(f func(string, schema.ModuleCodec) error) error {
+	keys := make([]string, 0, len(a.moduleSet))
+	for k := range a.moduleSet {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		module := a.moduleSet[k]
+		dm, ok := module.(schema.HasModuleCodec)
+		if ok {
+			decoder, err := dm.ModuleCodec()
+			if err != nil {
+				return err
+			}
+			err = f(k, decoder)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (a moduleSetDecoderResolver) LookupDecoder(moduleName string) (schema.ModuleCodec, bool, error) {
+	mod, ok := a.moduleSet[moduleName]
+	if !ok {
+		return schema.ModuleCodec{}, false, nil
+	}
+
+	dm, ok := mod.(schema.HasModuleCodec)
+	if !ok {
+		return schema.ModuleCodec{}, false, nil
+	}
+
+	decoder, err := dm.ModuleCodec()
+	return decoder, true, err
+}

schema/decoding/resolver_test.go

@@ -0,0 +1,125 @@
+package decoding
+
+import (
+	"fmt"
+	"testing"
+
+	"cosmossdk.io/schema"
+)
+
+type modA struct{}
+
+func (m modA) ModuleCodec() (schema.ModuleCodec, error) {
+	return schema.ModuleCodec{
+		Schema: schema.ModuleSchema{ObjectTypes: []schema.ObjectType{{Name: "A"}}},
+	}, nil
+}
+
+type modB struct{}
+
+func (m modB) ModuleCodec() (schema.ModuleCodec, error) {
+	return schema.ModuleCodec{
+		Schema: schema.ModuleSchema{ObjectTypes: []schema.ObjectType{{Name: "B"}}},
+	}, nil
+}
+
+type modC struct{}
+
+var moduleSet = map[string]interface{}{
+	"modA": modA{},
+	"modB": modB{},
+	"modC": modC{},
+}
+
+var resolver = ModuleSetDecoderResolver(moduleSet)
+
+func TestModuleSetDecoderResolver_IterateAll(t *testing.T) {
+	objectTypes := map[string]bool{}
+	err := resolver.IterateAll(func(moduleName string, cdc schema.ModuleCodec) error {
+		objectTypes[cdc.Schema.ObjectTypes[0].Name] = true
+		return nil
+	})
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if len(objectTypes) != 2 {
+		t.Fatalf("expected 2 object types, got %d", len(objectTypes))
+	}
+
+	if !objectTypes["A"] {
+		t.Fatalf("expected object type A")
+	}
+
+	if !objectTypes["B"] {
+		t.Fatalf("expected object type B")
+	}
+}
+
+func TestModuleSetDecoderResolver_LookupDecoder(t *testing.T) {
+	decoder, found, err := resolver.LookupDecoder("modA")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !found {
+		t.Fatalf("expected to find decoder for modA")
+	}
+
+	if decoder.Schema.ObjectTypes[0].Name != "A" {
+		t.Fatalf("expected object type A, got %s", decoder.Schema.ObjectTypes[0].Name)
+	}
+
+	decoder, found, err = resolver.LookupDecoder("modB")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if !found {
+		t.Fatalf("expected to find decoder for modB")
+	}
+
+	if decoder.Schema.ObjectTypes[0].Name != "B" {
+		t.Fatalf("expected object type B, got %s", decoder.Schema.ObjectTypes[0].Name)
+	}
+
+	decoder, found, err = resolver.LookupDecoder("modC")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if found {
+		t.Fatalf("expected not to find decoder")
+	}
+
+	decoder, found, err = resolver.LookupDecoder("modD")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if found {
+		t.Fatalf("expected not to find decoder")
+	}
+}
+
+type modD struct{}
+
+func (m modD) ModuleCodec() (schema.ModuleCodec, error) {
+	return schema.ModuleCodec{}, fmt.Errorf("an error")
+}
+
+func TestModuleSetDecoderResolver_IterateAll_Error(t *testing.T) {
+	resolver := ModuleSetDecoderResolver(map[string]interface{}{
+		"modD": modD{},
+	})
+	err := resolver.IterateAll(func(moduleName string, cdc schema.ModuleCodec) error {
+		if moduleName == "modD" {
+			t.Fatalf("expected error")
+		}
+		return nil
+	})
+	if err == nil {
+		t.Fatalf("expected error")
+	}
+}

schema/decoding/sync.go

@@ -0,0 +1,9 @@
+package decoding
+
+// SyncSource is an interface that allows indexers to start indexing modules with pre-existing state.
+// It should generally be a wrapper around the key-value store.
+type SyncSource interface {
+
+	// IterateAllKVPairs iterates over all key-value pairs for a given module.
+	IterateAllKVPairs(moduleName string, fn func(key, value []byte) error) error
+}

schema/indexer/README.md

@@ -0,0 +1,15 @@
+# Indexer Framework
+
+# Defining an Indexer
+
+Indexer implementations should be registered with the `indexer.Register` function with a unique type name. Indexers take the configuration options defined by `indexer.Config` which defines a common set of configuration options as well as indexer-specific options under the `config` sub-key. Indexers do not need to manage the common filtering options specified in `Config` - the indexer manager will manage these for the indexer. Indexer implementations just need to return a correct `InitResult` response.
+
+# Integrating the Indexer Manager
+
+The indexer manager should be used for managing all indexers and should be integrated directly with applications wishing to support indexing. The `StartManager` function is used to start the manager. The configuration options for the manager and all indexer targets should be passed as the ManagerOptions.Config field and should match the json structure of ManagerConfig. An example configuration section in `app.toml` might look like this:
+
+```toml
+[indexer.target.postgres]
+type = "postgres"
+config.database_url = "postgres://user:password@localhost:5432/dbname"
+```