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

(cherry picked from commit bd76b47)

# Conflicts:
#	collections/CHANGELOG.md
#	collections/go.mod
#	collections/go.sum
#	collections/indexing.go
#	collections/protocodec/collections.go
#	collections/protocodec/collections_test.go
#	indexer/postgres/create_table.go
#	indexer/postgres/enum.go
#	indexer/postgres/listener.go
#	indexer/postgres/tests/testdata/init_schema.txt
#	indexer/postgres/tests/testdata/init_schema_no_retain_delete.txt
#	math/legacy_dec_test.go
#	schema/appdata/async.go
#	schema/field.go
#	schema/field_test.go
#	schema/state_object.go
#	schema/state_object_test.go
#	server/v2/cometbft/go.mod
#	server/v2/cometbft/go.sum
#	simapp/go.mod
#	simapp/go.sum
#	store/cachekv/branch_bench_test.go
#	tests/integration/v2/auth/app_test.go
#	tests/integration/v2/services.go
#	x/bank/keeper/view.go
#	x/circuit/go.mod
#	x/circuit/go.sum
#	x/consensus/go.mod
#	x/consensus/go.sum
#	x/distribution/go.mod
#	x/epochs/go.mod
#	x/epochs/go.sum
#	x/evidence/go.mod
#	x/evidence/go.sum
#	x/mint/go.mod
#	x/mint/go.sum
#	x/nft/go.mod
#	x/nft/go.sum
#	x/params/go.mod
#	x/params/go.sum
#	x/protocolpool/go.mod
#	x/protocolpool/go.sum
#	x/slashing/go.mod
#	x/slashing/go.sum
#	x/staking/go.mod
#	x/upgrade/go.mod
#	x/upgrade/go.sum

baseapp/abci.go

@@ -903,8 +903,12 @@ func (app *BaseApp) FinalizeBlock(req *abci.FinalizeBlockRequest) (res *abci.Fin
 	defer func() {
 		// call the streaming service hooks with the FinalizeBlock messages
 		for _, streamingListener := range app.streamingManager.ABCIListeners {
-			if err := streamingListener.ListenFinalizeBlock(app.finalizeBlockState.Context(), *req, *res); err != nil {
+			if streamErr := streamingListener.ListenFinalizeBlock(app.finalizeBlockState.Context(), *req, *res); streamErr != nil {
 				app.logger.Error("ListenFinalizeBlock listening hook failed", "height", req.Height, "err", err)
+				if app.streamingManager.StopNodeOnErr {
+					// if StopNodeOnErr is set, we should return the streamErr in order to stop the node
+					err = streamErr
+				}
 			}
 		}
 	}()
@@ -976,12 +980,12 @@ func (app *BaseApp) Commit() (*abci.CommitResponse, error) {
 		rms.SetCommitHeader(header)
 	}
 
-	app.cms.Commit()
-
 	resp := &abci.CommitResponse{
 		RetainHeight: retainHeight,
 	}
 
+	app.cms.Commit()
+
 	abciListeners := app.streamingManager.ABCIListeners
 	if len(abciListeners) > 0 {
 		ctx := app.finalizeBlockState.Context()
@@ -991,6 +995,14 @@ 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 {
+					err = fmt.Errorf("Commit listening hook failed: %w", err)
+					rollbackErr := app.cms.RollbackToVersion(blockHeight - 1)
+					if rollbackErr != nil {
+						return nil, errors.Join(err, rollbackErr)
+					}
+					return nil, err
+				}
 			}
 		}
 	}

client/v2/go.mod

@@ -1,6 +1,6 @@
 module cosmossdk.io/client/v2
 
-go 1.23.1
+go 1.23.3
 
 require (
 	cosmossdk.io/api v0.8.0

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,139 @@ 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 {
+		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 marshaling 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) {
+				panic("not implemented")
+			},
+		}, nil
+	}
+}
+
 type protoMessageV2[T any] interface {
 	*T
 	protov2.Message
@@ -179,3 +324,101 @@ func (c collInterfaceValue[T]) ValueType() string {
 	var t T
 	return fmt.Sprintf("%T", t)
 }
+
+// SchemaCodec returns a schema codec, which will always have a single JSON field
+// as there is no way to know in advance the necessary fields for an interface.
+func (c collInterfaceValue[T]) SchemaCodec() (collcodec.SchemaCodec[T], error) {
+	var pt T
+
+	kind := schema.KindForGoValue(pt)
+	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 {
+		return collcodec.SchemaCodec[T]{
+			Fields: []schema.Field{{
+				Name: "value",
+				Kind: schema.JSONKind,
+			}},
+			ToSchemaType: func(t T) (any, error) {
+				bz, err := c.codec.MarshalInterfaceJSON(t)
+				if err != nil {
+					return nil, err
+				}
+
+				return json.RawMessage(bz), nil
+			},
+			FromSchemaType: func(a any) (T, error) {
+				panic("not implemented")
+			},
+		}, nil
+	}
+}
+
+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
+}