feat(indexer): add to modules and implement proto fields

baseapp/abci.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"sort"
 	"strconv"
 	"strings"
@@ -905,6 +906,9 @@ func (app *BaseApp) FinalizeBlock(req *abci.FinalizeBlockRequest) (res *abci.Fin
 		for _, streamingListener := range app.streamingManager.ABCIListeners {
 			if err := streamingListener.ListenFinalizeBlock(app.finalizeBlockState.Context(), *req, *res); err != nil {
 				app.logger.Error("ListenFinalizeBlock listening hook failed", "height", req.Height, "err", err)
+				if app.streamingManager.StopNodeOnErr {
+					os.Exit(1)
+				}
 			}
 		}
 	}()
@@ -976,8 +980,6 @@ func (app *BaseApp) Commit() (*abci.CommitResponse, error) {
 		rms.SetCommitHeader(header)
 	}
 
-	app.cms.Commit()
-
 	resp := &abci.CommitResponse{
 		RetainHeight: retainHeight,
 	}
@@ -991,10 +993,17 @@ func (app *BaseApp) Commit() (*abci.CommitResponse, error) {
 		for _, abciListener := range abciListeners {
 			if err := abciListener.ListenCommit(ctx, *resp, changeSet); err != nil {
 				app.logger.Error("Commit listening hook failed", "height", blockHeight, "err", err)
+				if app.streamingManager.StopNodeOnErr {
+					os.Exit(1)
+				}
 			}
 		}
 	}
 
+	// Commit after all listeners have been called, in case they error and we
+	// need to stop before committing.
+	app.cms.Commit()
+
 	// Reset the CheckTx state to the latest committed.
 	//
 	// NOTE: This is safe because CometBFT holds a lock on the mempool for

codec/collections.go

@@ -1,16 +1,23 @@
 package codec
 
 import (
+	"encoding/json"
 	"fmt"
 	"reflect"
 
 	"github.com/cosmos/gogoproto/proto"
 	gogotypes "github.com/cosmos/gogoproto/types"
+	gogoprotoany "github.com/cosmos/gogoproto/types/any"
 	"google.golang.org/protobuf/encoding/protojson"
 	protov2 "google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/types/dynamicpb"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"cosmossdk.io/collections"
 	collcodec "cosmossdk.io/collections/codec"
+	"cosmossdk.io/schema"
 )
 
 // BoolValue implements a ValueCodec that saves the bool value
@@ -51,12 +58,17 @@ type protoMessage[T any] interface {
 	proto.Message
 }
 
+type protoCollValueCodec[T any] interface {
+	collcodec.HasSchemaCodec[T]
+	collcodec.ValueCodec[T]
+}
+
 // CollValue inits a collections.ValueCodec for a generic gogo protobuf message.
 func CollValue[T any, PT protoMessage[T]](cdc interface {
 	Marshal(proto.Message) ([]byte, error)
 	Unmarshal([]byte, proto.Message) error
 },
-) collcodec.ValueCodec[T] {
+) protoCollValueCodec[T] {
 	return &collValue[T, PT]{cdc.(Codec), proto.MessageName(PT(new(T)))}
 }
 
@@ -91,6 +103,150 @@ func (c collValue[T, PT]) ValueType() string {
 	return "github.com/cosmos/gogoproto/" + c.messageName
 }
 
+type hasUnpackInterfaces interface {
+	UnpackInterfaces(unpacker gogoprotoany.AnyUnpacker) error // Replace `AnyUnpacker` with the actual type from gogoprotoany
+}
+
+func (c collValue[T, PT]) SchemaCodec() (collcodec.SchemaCodec[T], error) {
+	var (
+		t  T
+		pt PT
+	)
+	msgName := proto.MessageName(pt)
+	desc, err := proto.HybridResolver.FindDescriptorByName(protoreflect.FullName(msgName))
+	if err != nil {
+		return collcodec.SchemaCodec[T]{}, fmt.Errorf("could not find descriptor for %s: %w", msgName, err)
+	}
+	schemaFields := protoCols(desc.(protoreflect.MessageDescriptor))
+
+	kind := schema.KindForGoValue(t)
+	if err := kind.Validate(); err == nil {
+		return collcodec.SchemaCodec[T]{
+			Fields: []schema.Field{{
+				// we don't set any name so that this can be set to a good default by the caller
+				Name: "",
+				Kind: kind,
+			}},
+			// these can be nil because T maps directly to a schema value for this kind
+			ToSchemaType:   nil,
+			FromSchemaType: nil,
+		}, nil
+	} else {
+		// we default to encoding everything to JSON String
+		return collcodec.SchemaCodec[T]{
+			Fields: schemaFields,
+			ToSchemaType: func(t T) (any, error) {
+				values := []interface{}{}
+				msgDesc, ok := desc.(protoreflect.MessageDescriptor)
+				if !ok {
+					return nil, fmt.Errorf("expected message descriptor, got %T", desc)
+				}
+
+				nm := dynamicpb.NewMessage(msgDesc)
+				bz, err := c.cdc.Marshal(any(&t).(PT))
+				if err != nil {
+					return nil, err
+				}
+
+				err = c.cdc.Unmarshal(bz, nm)
+				if err != nil {
+					return nil, err
+				}
+
+				for _, field := range schemaFields {
+					// Find the field descriptor by the Protobuf field name
+					fieldDesc := msgDesc.Fields().ByName(protoreflect.Name(field.Name))
+					if fieldDesc == nil {
+						return nil, fmt.Errorf("field %q not found in message %s", field.Name, desc.FullName())
+					}
+
+					val := nm.ProtoReflect().Get(fieldDesc)
+
+					// if the field is a map or list, we need to convert it to a slice of values
+					if fieldDesc.IsList() {
+						repeatedVals := []interface{}{}
+						list := val.List()
+						for i := 0; i < list.Len(); i++ {
+							repeatedVals = append(repeatedVals, list.Get(i).Interface())
+						}
+						values = append(values, repeatedVals)
+						continue
+					}
+
+					switch fieldDesc.Kind() {
+					case protoreflect.BoolKind:
+						values = append(values, val.Bool())
+					case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind,
+						protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
+						values = append(values, val.Int())
+					case protoreflect.Uint32Kind, protoreflect.Fixed32Kind, protoreflect.Uint64Kind,
+						protoreflect.Fixed64Kind:
+						values = append(values, val.Uint())
+					case protoreflect.FloatKind, protoreflect.DoubleKind:
+						values = append(values, val.Float())
+					case protoreflect.StringKind:
+						values = append(values, val.String())
+					case protoreflect.BytesKind:
+						values = append(values, val.Bytes())
+					case protoreflect.EnumKind:
+						// TODO: support enums better, with actual enums
+						values = append(values, string(fieldDesc.Enum().Values().ByNumber(val.Enum()).Name()))
+					case protoreflect.MessageKind:
+						msg := val.Interface().(*dynamicpb.Message)
+						msgbz, err := c.cdc.Marshal(msg)
+						if err != nil {
+							return nil, err
+						}
+
+						if field.Kind == schema.TimeKind {
+							// make it a time.Time
+							ts := &timestamppb.Timestamp{}
+							err = c.cdc.Unmarshal(msgbz, ts)
+							if err != nil {
+								return nil, fmt.Errorf("error unmarshalling timestamp: %w %x %s", err, msgbz, fieldDesc.FullName())
+							}
+							values = append(values, ts.AsTime())
+						} else if field.Kind == schema.DurationKind {
+							// make it a time.Duration
+							dur := &durationpb.Duration{}
+							err = c.cdc.Unmarshal(msgbz, dur)
+							if err != nil {
+								return nil, fmt.Errorf("error unmarshalling duration: %w", err)
+							}
+							values = append(values, dur.AsDuration())
+						} else {
+							// if not a time or duration, just keep it as a JSON object
+							// we might want to change this to include the entire object as separate fields
+							bz, err := c.cdc.MarshalJSON(msg)
+							if err != nil {
+								return nil, fmt.Errorf("error marshalling message: %w", err)
+							}
+
+							values = append(values, json.RawMessage(bz))
+						}
+					}
+
+				}
+
+				// if there's only one value, return it directly
+				if len(values) == 1 {
+					return values[0], nil
+				}
+				return values, nil
+			},
+			FromSchemaType: func(a any) (T, error) {
+				var t T
+				sz, ok := a.(string)
+				if !ok {
+					return t, fmt.Errorf("expected string, got %T", a)
+				}
+				err := json.Unmarshal([]byte(sz), &t)
+				return t, err
+			},
+		}, nil
+	}
+}
+
 type protoMessageV2[T any] interface {
 	*T
 	protov2.Message
@@ -179,3 +335,69 @@ func (c collInterfaceValue[T]) ValueType() string {
 	var t T
 	return fmt.Sprintf("%T", t)
 }
+
+func protoCols(desc protoreflect.MessageDescriptor) []schema.Field {
+	nFields := desc.Fields()
+	cols := make([]schema.Field, 0, nFields.Len())
+	for i := 0; i < nFields.Len(); i++ {
+		f := nFields.Get(i)
+		cols = append(cols, protoCol(f))
+	}
+	return cols
+}
+
+func protoCol(f protoreflect.FieldDescriptor) schema.Field {
+	col := schema.Field{Name: string(f.Name())}
+	if f.IsMap() || f.IsList() {
+		col.Kind = schema.JSONKind
+		col.Nullable = true
+	} else {
+		switch f.Kind() {
+		case protoreflect.BoolKind:
+			col.Kind = schema.BoolKind
+		case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind:
+			col.Kind = schema.Int32Kind
+		case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
+			col.Kind = schema.Int64Kind
+		case protoreflect.Uint32Kind, protoreflect.Fixed32Kind:
+			col.Kind = schema.Int64Kind
+		case protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
+			col.Kind = schema.Uint64Kind
+		case protoreflect.FloatKind:
+			col.Kind = schema.Float32Kind
+		case protoreflect.DoubleKind:
+			col.Kind = schema.Float64Kind
+		case protoreflect.StringKind:
+			col.Kind = schema.StringKind
+		case protoreflect.BytesKind:
+			col.Kind = schema.BytesKind
+		case protoreflect.EnumKind:
+			// for now we'll use a string for enums.
+			col.Kind = schema.StringKind
+			// TODO: support enums
+			// col.Kind = schema.EnumKind
+			// enumDesc := f.Enum()
+			// var vals []string
+			// n := enumDesc.Values().Len()
+			// for i := 0; i < n; i++ {
+			// 	vals = append(vals, string(enumDesc.Values().Get(i).Name()))
+			// }
+			// col.ReferencedType = string(enumDesc.Name())
+		case protoreflect.MessageKind:
+			col.Nullable = true
+			fullName := f.Message().FullName()
+			if fullName == "google.protobuf.Timestamp" {
+				col.Kind = schema.TimeKind
+			} else if fullName == "google.protobuf.Duration" {
+				col.Kind = schema.DurationKind
+			} else {
+				col.Kind = schema.JSONKind
+			}
+		}
+		if f.HasPresence() {
+			col.Nullable = true
+		}
+	}
+
+	return col
+}

collections/codec/indexing.go

@@ -77,7 +77,9 @@ func FallbackSchemaCodec[T any]() SchemaCodec[T] {
 				Kind: kind,
 			}},
 			// these can be nil because T maps directly to a schema value for this kind
-			ToSchemaType:   nil,
+			ToSchemaType: func(t T) (any, error) {
+				return t, nil
+			},
 			FromSchemaType: nil,
 		}
 	} else {

indexer/postgres/create_table.go

@@ -17,7 +17,7 @@ func (tm *objectIndexer) createTable(ctx context.Context, conn dbConn) error {
 
 	sqlStr := buf.String()
 	if tm.options.logger != nil {
-		tm.options.logger.Debug("Creating table %s", "table", tm.tableName(), "sql", sqlStr)
+		tm.options.logger.Debug("Creating table", "table", tm.tableName(), "sql", sqlStr)
 	}
 	_, err = conn.ExecContext(ctx, sqlStr)
 	return err

indexer/postgres/listener.go

@@ -22,7 +22,26 @@ func (i *indexerImpl) listener() appdata.Listener {
 			return mm.initializeSchema(i.ctx, i.tx)
 		},
 		StartBlock: func(data appdata.StartBlockData) error {
-			_, err := i.tx.Exec("INSERT INTO block (number) VALUES ($1)", data.Height)
+			var (
+				headerBz []byte
+				err      error
+			)
+
+			if data.HeaderJSON != nil {
+				headerBz, err = data.HeaderJSON()
+				if err != nil {
+					return err
+				}
+			} else if data.HeaderBytes != nil {
+				headerBz, err = data.HeaderBytes()
+				if err != nil {
+					return err
+				}
+			}
+
+			// TODO: verify the format of headerBz, otherwise we'll get `ERROR: invalid input syntax for type json (SQLSTATE 22P02)`
+			_, err = i.tx.Exec("INSERT INTO block (number, header) VALUES ($1, $2)", data.Height, headerBz)
+
 			return err
 		},
 		OnObjectUpdate: func(data appdata.ObjectUpdateData) error {

schema/appdata/async.go

@@ -46,6 +46,13 @@ func AsyncListener(opts AsyncListenerOptions, listener Listener) Listener {
 	done := ctx.Done()
 
 	go func() {
+		defer func() {
+			close(packetChan)
+			if opts.DoneWaitGroup != nil {
+				opts.DoneWaitGroup.Done()
+			}
+		}()
+
 		if opts.DoneWaitGroup != nil {
 			opts.DoneWaitGroup.Add(1)
 		}
@@ -74,10 +81,6 @@ func AsyncListener(opts AsyncListenerOptions, listener Listener) Listener {
 				}
 
 			case <-done:
-				close(packetChan)
-				if opts.DoneWaitGroup != nil {
-					opts.DoneWaitGroup.Done()
-				}
 				return
 			}
 		}

schema/field.go

@@ -35,6 +35,10 @@ type Field struct {
 // Validate validates the field.
 func (c Field) Validate(typeSet TypeSet) error {
 	// valid name
+	if c.Name == "" {
+		return fmt.Errorf("field name cannot be empty, might be missing the named key codec")
+	}
+
 	if !ValidateName(c.Name) {
 		return fmt.Errorf("invalid field name %q", c.Name)
 	}