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"
+	"strings"
 
 	"github.com/cosmos/gogoproto/proto"
 	gogotypes "github.com/cosmos/gogoproto/types"
 	"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,146 @@ func (c collValue[T, PT]) ValueType() string {
 	return "github.com/cosmos/gogoproto/" + c.messageName
 }
 
+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: postgres uses the enum name, not the number
+						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 +331,63 @@ 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:
+			// TODO: support enums
+			col.Kind = schema.EnumKind
+			// use the full name to avoid collissions
+			col.ReferencedType = string(f.Enum().FullName())
+			col.ReferencedType = strings.ReplaceAll(col.ReferencedType, ".", "_")
+		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 {

collections/go.mod

@@ -1,18 +1,25 @@
 module cosmossdk.io/collections
 
-go 1.23
+go 1.23.2
+
+toolchain go1.23.3
 
 require (
 	cosmossdk.io/core v1.0.0-alpha.6
 	cosmossdk.io/core/testing v0.0.0-20240923163230-04da382a9f29
 	cosmossdk.io/schema v0.3.0
+	github.com/cosmos/gogoproto v1.7.0
 	github.com/stretchr/testify v1.9.0
 	github.com/tidwall/btree v1.7.0
 	pgregory.net/rapid v1.1.0
 )
 
 require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	google.golang.org/protobuf v1.35.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+

collections/go.sum

@@ -4,14 +4,22 @@ cosmossdk.io/core/testing v0.0.0-20240923163230-04da382a9f29 h1:NxxUo0GMJUbIuVg0
 cosmossdk.io/core/testing v0.0.0-20240923163230-04da382a9f29/go.mod h1:8s2tPeJtSiQuoyPmr2Ag7meikonISO4Fv4MoO8+ORrs=
 cosmossdk.io/schema v0.3.0 h1:01lcaM4trhzZ1HQTfTV8z6Ma1GziOZ/YmdzBN3F720c=
 cosmossdk.io/schema v0.3.0/go.mod h1:RDAhxIeNB4bYqAlF4NBJwRrgtnciMcyyg0DOKnhNZQQ=
+github.com/cosmos/gogoproto v1.7.0 h1:79USr0oyXAbxg3rspGh/m4SWNyoz/GLaAh0QlCe2fro=
+github.com/cosmos/gogoproto v1.7.0/go.mod h1:yWChEv5IUEYURQasfyBW5ffkMHR/90hiHgbNgrtp4j0=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/tidwall/btree v1.7.0 h1:L1fkJH/AuEh5zBnnBbmTwQ5Lt+bRJ5A8EWecslvo9iI=
 github.com/tidwall/btree v1.7.0/go.mod h1:twD9XRA5jj9VUQGELzDO4HPQTNJsoWWfYEL+EUQ2cKY=
+google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
+google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=