feat(schema): indexing API

indexer/base/README.md

@@ -11,9 +11,9 @@ The basic types for specifying index sources, targets and decoders are provided
 ```mermaid
 sequenceDiagram
     actor Source
-    participant Indexer    
-    Source ->> Indexer: Initialize
+    participant Indexer
     Source -->> Indexer: InitializeModuleSchema
+    Source ->> Indexer: Initialize
     loop Block
         Source ->> Indexer: StartBlock
         Source ->> Indexer: OnBlockHeader
@@ -25,7 +25,7 @@ sequenceDiagram
     end
 ```
 
-`Initialize` must be called before any other method and should only be invoked once. `InitializeModuleSchema` should be called at most once for every module with logical data.
+`InitializeModuleSchema` should be called at most once for every module with logical data and all calls to should happen even before `Initialize` is called. After that `Initialize` MUST be called before any other method and should only be invoked once. 
 
 Sources will generally only call `InitializeModuleSchema` and `OnObjectUpdate` if they have native logical decoding capabilities. Usually, the indexer framework will provide this functionality based on `OnKVPair` data and `IndexableModule` implementations.
 

indexer/base/decoder.go

@@ -1,9 +1,11 @@
 package indexerbase
 
+import "sort"
+
 // DecodableModule is an interface that modules can implement to provide a ModuleDecoder.
 // Usually these modules would also implement appmodule.AppModule, but that is not included
-// to keep this package free of any dependencies.
 type DecodableModule interface {
+
 	// ModuleDecoder returns a ModuleDecoder for the module.
 	ModuleDecoder() (ModuleDecoder, error)
 }
@@ -23,4 +25,68 @@ type ModuleDecoder struct {
 // If the KV-pair doesn't represent an object update, the function should return false
 // as the second return value. Error should only be non-nil when the decoder expected
 // to parse a valid update and was unable to.
-type KVDecoder = func(key, value []byte) (ObjectUpdate, bool, error)
+type KVDecoder = func(key, value []byte, deleted bool) (ObjectUpdate, bool, error)
+
+type DecoderResolver interface {
+	// Iterate iterates over all module decoders which should be initialized at startup.
+	Iterate(func(string, ModuleDecoder) error) error
+
+	// LookupDecoder allows for resolving decoders dynamically. For instance, some module-like
+	// things may come into existence dynamically (like x/accounts or EVM or WASM contracts).
+	// The first time the manager sees one of these appearing in KV-store writes, it will
+	// lookup a decoder for it and cache it for future use. The manager will also perform
+	// a catch-up sync before passing any new writes to ensure that all historical state has
+	// been synced if there is any This check will only happen the first time a module is seen
+	// by the manager in a given process (a process restart will cause this check to happen again).
+	LookupDecoder(moduleName string) (decoder ModuleDecoder, found bool, err error)
+}
+
+type moduleSetDecoderResolver struct {
+	moduleSet map[string]interface{}
+}
+
+// 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,
+	}
+}
+
+func (a moduleSetDecoderResolver) Iterate(f func(string, ModuleDecoder) 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 := any(module).(DecodableModule)
+		if ok {
+			decoder, err := dm.ModuleDecoder()
+			if err != nil {
+				return err
+			}
+			err = f(k, decoder)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (a moduleSetDecoderResolver) LookupDecoder(moduleName string) (ModuleDecoder, bool, error) {
+	mod, ok := a.moduleSet[moduleName]
+	if !ok {
+		return ModuleDecoder{}, false, nil
+	}
+
+	dm, ok := any(mod).(DecodableModule)
+	if !ok {
+		return ModuleDecoder{}, false, nil
+	}
+
+	decoder, err := dm.ModuleDecoder()
+	return decoder, true, err
+}

indexer/base/listener.go

@@ -15,7 +15,8 @@ type Listener struct {
 	// The lastBlockPersisted return value should be the last block height the listener persisted if it is
 	// persisting block data, 0 if it is not interested in persisting block data, or -1 if it is
 	// persisting block data but has not persisted any data yet. This check allows the indexer
-	// framework to ensure that the listener has not missed blocks.
+	// framework to ensure that the listener has not missed blocks. Data sources MUST call
+	// initialize before any other method is called, otherwise, no data will be processed.
 	Initialize func(InitializationData) (lastBlockPersisted int64, err error)
 
 	// StartBlock is called at the beginning of processing a block.
@@ -31,11 +32,12 @@ type Listener struct {
 	OnEvent func(EventData) error
 
 	// OnKVPair is called when a key-value has been written to the store for a given module.
-	OnKVPair func(moduleName string, key, value []byte, delete bool) error
+	OnKVPair func(KVPairData) error
 
 	// Commit is called when state is committed, usually at the end of a block. Any
 	// indexers should commit their data when this is called and return an error if
-	// they are unable to commit.
+	// they are unable to commit. Data sources MUST call Commit when data is committed,
+	// otherwise it should be assumed that indexers have not persisted their state.
 	Commit func() error
 
 	// InitializeModuleSchema should be called whenever the blockchain process starts OR whenever
@@ -49,7 +51,7 @@ type Listener struct {
 	// OnObjectUpdate is called whenever an object is updated in a module's state. This is only called
 	// when logical data is available. It should be assumed that the same data in raw form
 	// is also passed to OnKVPair.
-	OnObjectUpdate func(module string, update ObjectUpdate) error
+	OnObjectUpdate func(ObjectUpdateData) error
 }
 
 // InitializationData represents initialization data that is passed to a listener.
@@ -117,3 +119,24 @@ type ToBytes = func() ([]byte, error)
 
 // ToJSON is a function that lazily returns the JSON representation of data.
 type ToJSON = func() (json.RawMessage, error)
+
+// KVPairData represents key-value pair data that is passed to a listener.
+type KVPairData struct {
+	// ModuleName is the name of the module that the key-value pair belongs to.
+	ModuleName string
+
+	// Key is the key of the key-value pair.
+	Key []byte
+
+	// Value is the value of the key-value pair. It should be ignored when Delete is true.
+	Value []byte
+
+	// Delete is a flag that indicates that the key-value pair was deleted. If it is false,
+	// then it is assumed that this has been a set operation.
+	Delete bool
+}
+
+type ObjectUpdateData struct {
+	ModuleName string
+	Update     ObjectUpdate
+}

indexer/base/log.go

@@ -0,0 +1,20 @@
+package indexerbase
+
+// Logger is the expected interface of loggers used in the indexer.
+type Logger interface {
+	// Info takes a message and a set of key/value pairs and logs with level INFO.
+	// The key of the tuple must be a string.
+	Info(msg string, keyVals ...any)
+
+	// Warn takes a message and a set of key/value pairs and logs with level WARN.
+	// The key of the tuple must be a string.
+	Warn(msg string, keyVals ...any)
+
+	// Error takes a message and a set of key/value pairs and logs with level ERR.
+	// The key of the tuple must be a string.
+	Error(msg string, keyVals ...any)
+
+	// Debug takes a message and a set of key/value pairs and logs with level DEBUG.
+	// The key of the tuple must be a string.
+	Debug(msg string, keyVals ...any)
+}