feat(schema/indexer)!: implement start indexing

indexer/postgres/indexer.go

@@ -3,8 +3,8 @@ package postgres
 import (
 	"context"
 	"database/sql"
-	"encoding/json"
 	"errors"
+	"fmt"
 
 	"cosmossdk.io/schema/indexer"
 	"cosmossdk.io/schema/logutil"
@@ -21,8 +21,6 @@ type Config struct {
 	DisableRetainDeletions bool `json:"disable_retain_deletions"`
 }
 
-type SqlLogger = func(msg, sql string, params ...interface{})
-
 type indexerImpl struct {
 	ctx     context.Context
 	db      *sql.DB
@@ -32,10 +30,17 @@ type indexerImpl struct {
 	logger  logutil.Logger
 }
 
-func StartIndexer(params indexer.InitParams) (indexer.InitResult, error) {
-	config, err := decodeConfig(params.Config.Config)
-	if err != nil {
-		return indexer.InitResult{}, err
+func init() {
+	indexer.Register("postgres", indexer.Initializer{
+		InitFunc:   startIndexer,
+		ConfigType: Config{},
+	})
+}
+
+func startIndexer(params indexer.InitParams) (indexer.InitResult, error) {
+	config, ok := params.Config.Config.(Config)
+	if !ok {
+		return indexer.InitResult{}, fmt.Errorf("invalid config type, expected %T got %T", Config{}, params.Config.Config)
 	}
 
 	ctx := params.Context
@@ -89,18 +94,3 @@ func StartIndexer(params indexer.InitParams) (indexer.InitResult, error) {
 		View:     idx,
 	}, nil
 }
-
-func decodeConfig(rawConfig map[string]interface{}) (*Config, error) {
-	bz, err := json.Marshal(rawConfig)
-	if err != nil {
-		return nil, err
-	}
-
-	var config Config
-	err = json.Unmarshal(bz, &config)
-	if err != nil {
-		return nil, err
-	}
-
-	return &config, nil
-}

indexer/postgres/tests/init_schema_test.go

@@ -33,17 +33,20 @@ func testInitSchema(t *testing.T, disableRetainDeletions bool, goldenFileName st
 	connectionUrl := createTestDB(t)
 
 	buf := &strings.Builder{}
-
-	cfg, err := postgresConfigToIndexerConfig(postgres.Config{
-		DatabaseURL:            connectionUrl,
-		DisableRetainDeletions: disableRetainDeletions,
-	})
-	require.NoError(t, err)
-
-	res, err := postgres.StartIndexer(indexer.InitParams{
-		Config:  cfg,
+	res, err := indexer.StartIndexing(indexer.IndexingOptions{
+		Config: indexer.IndexingConfig{
+			Target: map[string]indexer.Config{
+				"postgres": {
+					Type: "postgres",
+					Config: postgres.Config{
+						DatabaseURL:            connectionUrl,
+						DisableRetainDeletions: disableRetainDeletions,
+					},
+				},
+			},
+		},
 		Context: context.Background(),
-		Logger:  &prettyLogger{buf},
+		Logger:  prettyLogger{buf},
 	})
 	require.NoError(t, err)
 	listener := res.Listener

indexer/postgres/tests/postgres_test.go

@@ -52,31 +52,39 @@ func testPostgresIndexer(t *testing.T, retainDeletions bool) {
 		require.NoError(t, err)
 	})
 
-	cfg, err := postgresConfigToIndexerConfig(postgres.Config{
-		DatabaseURL:            dbUrl,
-		DisableRetainDeletions: !retainDeletions,
-	})
-	require.NoError(t, err)
-
 	debugLog := &strings.Builder{}
 
-	pgIndexer, err := postgres.StartIndexer(indexer.InitParams{
-		Config:       cfg,
+	res, err := indexer.StartIndexing(indexer.IndexingOptions{
+		Config: indexer.IndexingConfig{
+			Target: map[string]indexer.Config{
+				"postgres": {
+					Type: "postgres",
+					Config: postgres.Config{
+						DatabaseURL:            dbUrl,
+						DisableRetainDeletions: !retainDeletions,
+					},
+				},
+			},
+		},
 		Context:      ctx,
 		Logger:       &prettyLogger{debugLog},
 		AddressCodec: addressutil.HexAddressCodec{},
 	})
 	require.NoError(t, err)
+	require.NoError(t, err)
 
 	sim, err := appdatasim.NewSimulator(appdatasim.Options{
-		Listener:  pgIndexer.Listener,
+		Listener:  res.Listener,
 		AppSchema: indexertesting.ExampleAppSchema,
 		StateSimOptions: statesim.Options{
 			CanRetainDeletions: retainDeletions,
 		},
 	})
 	require.NoError(t, err)
 
+	pgIndexerView := res.IndexerInfos["postgres"].View
+	require.NotNil(t, pgIndexerView)
+
 	blockDataGen := sim.BlockDataGenN(10, 100)
 	numBlocks := 200
 	if testing.Short() {
@@ -93,7 +101,7 @@ func testPostgresIndexer(t *testing.T, retainDeletions bool) {
 		require.NoError(t, sim.ProcessBlockData(blockData), debugLog.String())
 
 		// compare the expected state in the simulator to the actual state in the indexer and expect the diff to be empty
-		require.Empty(t, appdatasim.DiffAppData(sim, pgIndexer.View), debugLog.String())
+		require.Empty(t, appdatasim.DiffAppData(sim, pgIndexerView), debugLog.String())
 
 		// reset the debug log after each successful block so that it doesn't get too long when debugging
 		debugLog.Reset()

schema/decoding/resolver.go

@@ -15,8 +15,8 @@ type DecoderResolver interface {
 	// EncodeModuleName encodes a module name into a byte slice that can be used as the actor in a KVPairUpdate.
 	EncodeModuleName(string) ([]byte, error)
 
-	// IterateAll iterates over all available module decoders.
-	IterateAll(func(moduleName string, cdc schema.ModuleCodec) error) error
+	// AllDecoders iterates over all available module decoders.
+	AllDecoders(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)
@@ -48,7 +48,7 @@ func (a moduleSetDecoderResolver) EncodeModuleName(s string) ([]byte, error) {
 	return nil, fmt.Errorf("module %s not found", s)
 }
 
-func (a moduleSetDecoderResolver) IterateAll(f func(string, schema.ModuleCodec) error) error {
+func (a moduleSetDecoderResolver) AllDecoders(f func(string, schema.ModuleCodec) error) error {
 	keys := make([]string, 0, len(a.moduleSet))
 	for k := range a.moduleSet {
 		keys = append(keys, k)

schema/decoding/resolver_test.go

@@ -43,7 +43,7 @@ var testResolver = ModuleSetDecoderResolver(moduleSet)
 
 func TestModuleSetDecoderResolver_IterateAll(t *testing.T) {
 	objectTypes := map[string]bool{}
-	err := testResolver.IterateAll(func(moduleName string, cdc schema.ModuleCodec) error {
+	err := testResolver.AllDecoders(func(moduleName string, cdc schema.ModuleCodec) error {
 		cdc.Schema.AllTypes(func(t schema.Type) bool {
 			objTyp, ok := t.(schema.StateObjectType)
 			if ok {
@@ -128,7 +128,7 @@ func TestModuleSetDecoderResolver_IterateAll_Error(t *testing.T) {
 	resolver := ModuleSetDecoderResolver(map[string]interface{}{
 		"modD": modD{},
 	})
-	err := resolver.IterateAll(func(moduleName string, cdc schema.ModuleCodec) error {
+	err := resolver.AllDecoders(func(moduleName string, cdc schema.ModuleCodec) error {
 		if moduleName == "modD" {
 			t.Fatalf("expected error")
 		}

schema/decoding/sync.go

@@ -27,7 +27,7 @@ func Sync(listener appdata.Listener, source SyncSource, resolver DecoderResolver
 		return nil
 	}
 
-	return resolver.IterateAll(func(moduleName string, cdc schema.ModuleCodec) error {
+	return resolver.AllDecoders(func(moduleName string, cdc schema.ModuleCodec) error {
 		if opts.ModuleFilter != nil && !opts.ModuleFilter(moduleName) {
 			// ignore this module
 			return nil

schema/indexer/config.go

@@ -0,0 +1,52 @@
+package indexer
+
+// Config species the configuration passed to an indexer initialization function.
+// It includes both common configuration options related to include or excluding
+// parts of the data stream as well as indexer specific options under the config
+// subsection.
+//
+// NOTE: it is an error for an indexer to change its common options, such as adding
+// or removing indexed modules, after the indexer has been initialized because this
+// could result in an inconsistent state.
+type Config struct {
+	// Type is the name of the indexer type as registered with Register.
+	Type string `mapstructure:"type" toml:"type" json:"type" comment:"The name of the registered indexer type."`
+
+	// Config are the indexer specific config options specified by the user.
+	Config interface{} `mapstructure:"config" toml:"config" json:"config,omitempty" comment:"Indexer specific configuration options."`
+
+	// Filter is the filter configuration for the indexer.
+	Filter *FilterConfig `mapstructure:"filter" toml:"filter" json:"filter,omitempty" comment:"Filter configuration for the indexer. Currently UNSUPPORTED!"`
+}
+
+// FilterConfig specifies the configuration for filtering the data stream
+type FilterConfig struct {
+	// ExcludeState specifies that the indexer will not receive state updates.
+	ExcludeState bool `mapstructure:"exclude_state" toml:"exclude_state" json:"exclude_state" comment:"Exclude all state updates."`
+
+	// ExcludeEvents specifies that the indexer will not receive events.
+	ExcludeEvents bool `mapstructure:"exclude_events" toml:"exclude_events" json:"exclude_events" comment:"Exclude all events."`
+
+	// ExcludeTxs specifies that the indexer will not receive transaction's.
+	ExcludeTxs bool `mapstructure:"exclude_txs" toml:"exclude_txs" json:"exclude_txs" comment:"Exclude all transactions."`
+
+	// ExcludeBlockHeaders specifies that the indexer will not receive block headers,
+	// although it will still receive StartBlock and Commit callbacks, just without
+	// the header data.
+	ExcludeBlockHeaders bool `mapstructure:"exclude_block_headers" toml:"exclude_block_headers" json:"exclude_block_headers" comment:"Exclude all block headers."`
+
+	Modules *ModuleFilterConfig `mapstructure:"modules" toml:"modules" json:"modules,omitempty" comment:"Module filter configuration."`
+}
+
+// ModuleFilterConfig specifies the configuration for filtering modules.
+type ModuleFilterConfig struct {
+	// Include specifies a list of modules whose state the indexer will
+	// receive state updates for.
+	// Only one of include or exclude modules should be specified.
+	Include []string `mapstructure:"include" toml:"include" json:"include" comment:"List of modules to include. Only one of include or exclude should be specified."`
+
+	// Exclude specifies a list of modules whose state the indexer will not
+	// receive state updates for.
+	// Only one of include or exclude modules should be specified.
+	Exclude []string `mapstructure:"exclude" toml:"exclude" json:"exclude" comment:"List of modules to exclude. Only one of include or exclude should be specified."`
+}

schema/indexer/indexer.go

@@ -9,44 +9,13 @@ import (
 	"cosmossdk.io/schema/view"
 )
 
-// Config species the configuration passed to an indexer initialization function.
-// It includes both common configuration options related to include or excluding
-// parts of the data stream as well as indexer specific options under the config
-// subsection.
-//
-// NOTE: it is an error for an indexer to change its common options, such as adding
-// or removing indexed modules, after the indexer has been initialized because this
-// could result in an inconsistent state.
-type Config struct {
-	// Type is the name of the indexer type as registered with Register.
-	Type string `json:"type"`
+// Initializer describes an indexer initialization function and other metadata.
+type Initializer struct {
+	// InitFunc is the function that initializes the indexer.
+	InitFunc InitFunc
 
-	// Config are the indexer specific config options specified by the user.
-	Config map[string]interface{} `json:"config"`
-
-	// ExcludeState specifies that the indexer will not receive state updates.
-	ExcludeState bool `json:"exclude_state"`
-
-	// ExcludeEvents specifies that the indexer will not receive events.
-	ExcludeEvents bool `json:"exclude_events"`
-
-	// ExcludeTxs specifies that the indexer will not receive transaction's.
-	ExcludeTxs bool `json:"exclude_txs"`
-
-	// ExcludeBlockHeaders specifies that the indexer will not receive block headers,
-	// although it will still receive StartBlock and Commit callbacks, just without
-	// the header data.
-	ExcludeBlockHeaders bool `json:"exclude_block_headers"`
-
-	// IncludeModules specifies a list of modules whose state the indexer will
-	// receive state updates for.
-	// Only one of include or exclude modules should be specified.
-	IncludeModules []string `json:"include_modules"`
-
-	// ExcludeModules specifies a list of modules whose state the indexer will not
-	// receive state updates for.
-	// Only one of include or exclude modules should be specified.
-	ExcludeModules []string `json:"exclude_modules"`
+	// ConfigType is the type of the configuration object that the indexer expects.
+	ConfigType interface{}
 }
 
 type InitFunc = func(InitParams) (InitResult, error)

schema/indexer/registry.go

@@ -3,12 +3,16 @@ package indexer
 import "fmt"
 
 // Register registers an indexer type with the given initialization function.
-func Register(indexerType string, initFunc InitFunc) {
+func Register(indexerType string, descriptor Initializer) {
 	if _, ok := indexerRegistry[indexerType]; ok {
 		panic(fmt.Sprintf("indexer %s already registered", indexerType))
 	}
 
-	indexerRegistry[indexerType] = initFunc
+	if descriptor.InitFunc == nil {
+		panic(fmt.Sprintf("indexer %s has no initialization function", indexerType))
+	}
+
+	indexerRegistry[indexerType] = descriptor
 }
 
-var indexerRegistry = map[string]InitFunc{}
+var indexerRegistry = map[string]Initializer{}