feat: ePBS SSE topic dispatch + typed handlers

api/eventsopts.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ethpandaops/go-eth2-client/spec/altair"
 	"github.com/ethpandaops/go-eth2-client/spec/capella"
 	"github.com/ethpandaops/go-eth2-client/spec/electra"
+	"github.com/ethpandaops/go-eth2-client/spec/gloas"
 	"github.com/ethpandaops/go-eth2-client/spec/phase0"
 )
 
@@ -53,12 +54,26 @@ type EventsOpts struct {
 	ContributionAndProofHandler ContributionAndProofEventHandlerFunc
 	// DataColumnSidecarHandler is a handler for the data_column_sidecar event.
 	DataColumnSidecarHandler DataColumnSidecarEventHandlerFunc
+	// ExecutionPayloadHandler is a handler for the execution_payload event
+	// (SignedExecutionPayloadEnvelope successfully imported on the fork-choice).
+	ExecutionPayloadHandler ExecutionPayloadEventHandlerFunc
+	// ExecutionPayloadAvailableHandler is a handler for the execution_payload_available event.
+	ExecutionPayloadAvailableHandler ExecutionPayloadAvailableEventHandlerFunc
+	// ExecutionPayloadBidHandler is a handler for the execution_payload_bid event.
+	ExecutionPayloadBidHandler ExecutionPayloadBidEventHandlerFunc
+	// ExecutionPayloadGossipHandler is a handler for the execution_payload_gossip event
+	// (SignedExecutionPayloadEnvelope passed gossip validation).
+	ExecutionPayloadGossipHandler ExecutionPayloadGossipEventHandlerFunc
 	// FinalizedCheckpointHandler is a handler for the finalized_checkpoint event.
 	FinalizedCheckpointHandler FinalizedCheckpointEventHandlerFunc
 	// HeadHandler is a handler for the head event.
 	HeadHandler HeadEventHandlerFunc
+	// PayloadAttestationMessageHandler is a handler for the payload_attestation_message event.
+	PayloadAttestationMessageHandler PayloadAttestationMessageEventHandlerFunc
 	// PayloadAttributesHandler is a handler for the payload_attributes event.
 	PayloadAttributesHandler PayloadAttributesEventHandlerFunc
+	// ProposerPreferencesHandler is a handler for the proposer_preferences event.
+	ProposerPreferencesHandler ProposerPreferencesEventHandlerFunc
 	// ProposerSlashingHandler is a handler for the proposer_slashing event.
 	ProposerSlashingHandler ProposerSlashingEventHandlerFunc
 	// SingleAttestationHandler is a handler for the single_attestation event.
@@ -114,3 +129,21 @@ type VoluntaryExitEventHandlerFunc func(context.Context, *phase0.SignedVoluntary
 
 // DataColumnSidecarEventHandlerFunc is the handler for data_column_sidecar events.
 type DataColumnSidecarEventHandlerFunc func(context.Context, *apiv1.DataColumnSidecarEvent)
+
+// ExecutionPayloadEventHandlerFunc is the handler for execution_payload events.
+type ExecutionPayloadEventHandlerFunc func(context.Context, *gloas.SignedExecutionPayloadEnvelope)
+
+// ExecutionPayloadAvailableEventHandlerFunc is the handler for execution_payload_available events.
+type ExecutionPayloadAvailableEventHandlerFunc func(context.Context, *apiv1.ExecutionPayloadAvailableEvent)
+
+// ExecutionPayloadBidEventHandlerFunc is the handler for execution_payload_bid events.
+type ExecutionPayloadBidEventHandlerFunc func(context.Context, *gloas.SignedExecutionPayloadBid)
+
+// ExecutionPayloadGossipEventHandlerFunc is the handler for execution_payload_gossip events.
+type ExecutionPayloadGossipEventHandlerFunc func(context.Context, *gloas.SignedExecutionPayloadEnvelope)
+
+// PayloadAttestationMessageEventHandlerFunc is the handler for payload_attestation_message events.
+type PayloadAttestationMessageEventHandlerFunc func(context.Context, *gloas.PayloadAttestationMessage)
+
+// ProposerPreferencesEventHandlerFunc is the handler for proposer_preferences events.
+type ProposerPreferencesEventHandlerFunc func(context.Context, *gloas.SignedProposerPreferences)

api/v1/event.go

@@ -45,8 +45,10 @@ var SupportedEventTopics = map[string]bool{
 	"chain_reorg":                 true,
 	"contribution_and_proof":      true,
 	"data_column_sidecar":         true,
+	"execution_payload":           true,
 	"execution_payload_available": true,
 	"execution_payload_bid":       true,
+	"execution_payload_gossip":    true,
 	"finalized_checkpoint":        true,
 	"head":                        true,
 	"inclusion_list":              true,
@@ -121,6 +123,8 @@ func (e *Event) UnmarshalJSON(input []byte) error {
 		e.Data = &altair.SignedContributionAndProof{}
 	case "data_column_sidecar":
 		e.Data = &DataColumnSidecarEvent{}
+	case "execution_payload", "execution_payload_gossip":
+		e.Data = &gloas.SignedExecutionPayloadEnvelope{}
 	case "execution_payload_available":
 		e.Data = &ExecutionPayloadAvailableEvent{}
 	case "execution_payload_bid":

http/events.go

@@ -32,6 +32,7 @@ import (
 	"github.com/ethpandaops/go-eth2-client/spec/altair"
 	"github.com/ethpandaops/go-eth2-client/spec/capella"
 	"github.com/ethpandaops/go-eth2-client/spec/electra"
+	"github.com/ethpandaops/go-eth2-client/spec/gloas"
 	"github.com/ethpandaops/go-eth2-client/spec/phase0"
 	"github.com/r3labs/sse/v2"
 	"github.com/rs/zerolog"
@@ -145,12 +146,24 @@ func (*Service) checkEventSpecificHandler(opts *api.EventsOpts, topic string) er
 		hasHandler = opts.ContributionAndProofHandler != nil
 	case "data_column_sidecar":
 		hasHandler = opts.DataColumnSidecarHandler != nil
+	case "execution_payload":
+		hasHandler = opts.ExecutionPayloadHandler != nil
+	case "execution_payload_available":
+		hasHandler = opts.ExecutionPayloadAvailableHandler != nil
+	case "execution_payload_bid":
+		hasHandler = opts.ExecutionPayloadBidHandler != nil
+	case "execution_payload_gossip":
+		hasHandler = opts.ExecutionPayloadGossipHandler != nil
 	case "finalized_checkpoint":
 		hasHandler = opts.FinalizedCheckpointHandler != nil
 	case "head":
 		hasHandler = opts.HeadHandler != nil
+	case "payload_attestation_message":
+		hasHandler = opts.PayloadAttestationMessageHandler != nil
 	case "payload_attributes":
 		hasHandler = opts.PayloadAttributesHandler != nil
+	case "proposer_preferences":
+		hasHandler = opts.ProposerPreferencesHandler != nil
 	case "proposer_slashing":
 		hasHandler = opts.ProposerSlashingHandler != nil
 	case "single_attestation":
@@ -200,12 +213,24 @@ func (s *Service) handleEvent(ctx context.Context,
 		s.handleContributionAndProofEvent(ctx, msg, opts)
 	case "data_column_sidecar":
 		s.handleDataColumnSidecarEvent(ctx, msg, opts)
+	case "execution_payload":
+		s.handleExecutionPayloadEvent(ctx, msg, opts)
+	case "execution_payload_available":
+		s.handleExecutionPayloadAvailableEvent(ctx, msg, opts)
+	case "execution_payload_bid":
+		s.handleExecutionPayloadBidEvent(ctx, msg, opts)
+	case "execution_payload_gossip":
+		s.handleExecutionPayloadGossipEvent(ctx, msg, opts)
 	case "finalized_checkpoint":
 		s.handleFinalizedCheckpointEvent(ctx, msg, opts)
 	case "head":
 		s.handleHeadEvent(ctx, msg, opts)
+	case "payload_attestation_message":
+		s.handlePayloadAttestationMessageEvent(ctx, msg, opts)
 	case "payload_attributes":
 		s.handlePayloadAttributesEvent(ctx, msg, opts)
+	case "proposer_preferences":
+		s.handleProposerPreferencesEvent(ctx, msg, opts)
 	case "proposer_slashing":
 		s.handleProposerSlashingEvent(ctx, msg, opts)
 	case "single_attestation":
@@ -623,3 +648,165 @@ func (*Service) handleDataColumnSidecarEvent(ctx context.Context,
 		log.Debug().Msg("No specific or generic handler supplied; ignoring")
 	}
 }
+
+func (*Service) handleExecutionPayloadEvent(ctx context.Context,
+	msg *sse.Event,
+	opts *api.EventsOpts,
+) {
+	log := zerolog.Ctx(ctx)
+	data := &gloas.SignedExecutionPayloadEnvelope{}
+
+	err := json.Unmarshal(msg.Data, data)
+	if err != nil {
+		log.Error().Err(err).RawJSON("data", msg.Data).Msg("Failed to parse execution payload event")
+
+		return
+	}
+
+	switch {
+	case opts.ExecutionPayloadHandler != nil:
+		opts.ExecutionPayloadHandler(ctx, data)
+	case opts.Handler != nil:
+		opts.Handler(&apiv1.Event{
+			Topic: string(msg.Event),
+			Data:  data,
+		})
+	default:
+		log.Debug().Msg("No specific or generic handler supplied; ignoring")
+	}
+}
+
+func (*Service) handleExecutionPayloadAvailableEvent(ctx context.Context,
+	msg *sse.Event,
+	opts *api.EventsOpts,
+) {
+	log := zerolog.Ctx(ctx)
+	data := &apiv1.ExecutionPayloadAvailableEvent{}
+
+	err := json.Unmarshal(msg.Data, data)
+	if err != nil {
+		log.Error().Err(err).RawJSON("data", msg.Data).Msg("Failed to parse execution payload available event")
+
+		return
+	}
+
+	switch {
+	case opts.ExecutionPayloadAvailableHandler != nil:
+		opts.ExecutionPayloadAvailableHandler(ctx, data)
+	case opts.Handler != nil:
+		opts.Handler(&apiv1.Event{
+			Topic: string(msg.Event),
+			Data:  data,
+		})
+	default:
+		log.Debug().Msg("No specific or generic handler supplied; ignoring")
+	}
+}
+
+func (*Service) handleExecutionPayloadBidEvent(ctx context.Context,
+	msg *sse.Event,
+	opts *api.EventsOpts,
+) {
+	log := zerolog.Ctx(ctx)
+	data := &gloas.SignedExecutionPayloadBid{}
+
+	err := json.Unmarshal(msg.Data, data)
+	if err != nil {
+		log.Error().Err(err).RawJSON("data", msg.Data).Msg("Failed to parse execution payload bid event")
+
+		return
+	}
+
+	switch {
+	case opts.ExecutionPayloadBidHandler != nil:
+		opts.ExecutionPayloadBidHandler(ctx, data)
+	case opts.Handler != nil:
+		opts.Handler(&apiv1.Event{
+			Topic: string(msg.Event),
+			Data:  data,
+		})
+	default:
+		log.Debug().Msg("No specific or generic handler supplied; ignoring")
+	}
+}
+
+func (*Service) handleExecutionPayloadGossipEvent(ctx context.Context,
+	msg *sse.Event,
+	opts *api.EventsOpts,
+) {
+	log := zerolog.Ctx(ctx)
+	data := &gloas.SignedExecutionPayloadEnvelope{}
+
+	err := json.Unmarshal(msg.Data, data)
+	if err != nil {
+		log.Error().Err(err).RawJSON("data", msg.Data).Msg("Failed to parse execution payload gossip event")
+
+		return
+	}
+
+	switch {
+	case opts.ExecutionPayloadGossipHandler != nil:
+		opts.ExecutionPayloadGossipHandler(ctx, data)
+	case opts.Handler != nil:
+		opts.Handler(&apiv1.Event{
+			Topic: string(msg.Event),
+			Data:  data,
+		})
+	default:
+		log.Debug().Msg("No specific or generic handler supplied; ignoring")
+	}
+}
+
+func (*Service) handlePayloadAttestationMessageEvent(ctx context.Context,
+	msg *sse.Event,
+	opts *api.EventsOpts,
+) {
+	log := zerolog.Ctx(ctx)
+	data := &gloas.PayloadAttestationMessage{}
+
+	err := json.Unmarshal(msg.Data, data)
+	if err != nil {
+		log.Error().Err(err).RawJSON("data", msg.Data).Msg("Failed to parse payload attestation message event")
+
+		return
+	}
+
+	switch {
+	case opts.PayloadAttestationMessageHandler != nil:
+		opts.PayloadAttestationMessageHandler(ctx, data)
+	case opts.Handler != nil:
+		opts.Handler(&apiv1.Event{
+			Topic: string(msg.Event),
+			Data:  data,
+		})
+	default:
+		log.Debug().Msg("No specific or generic handler supplied; ignoring")
+	}
+}
+
+func (*Service) handleProposerPreferencesEvent(ctx context.Context,
+	msg *sse.Event,
+	opts *api.EventsOpts,
+) {
+	log := zerolog.Ctx(ctx)
+	data := &gloas.SignedProposerPreferences{}
+
+	err := json.Unmarshal(msg.Data, data)
+	if err != nil {
+		log.Error().Err(err).RawJSON("data", msg.Data).Msg("Failed to parse proposer preferences event")
+
+		return
+	}
+
+	switch {
+	case opts.ProposerPreferencesHandler != nil:
+		opts.ProposerPreferencesHandler(ctx, data)
+	case opts.Handler != nil:
+		opts.Handler(&apiv1.Event{
+			Topic: string(msg.Event),
+			Data:  data,
+		})
+	default:
+		log.Debug().Msg("No specific or generic handler supplied; ignoring")
+	}
+}

util/proof/proofs.go

@@ -89,7 +89,7 @@ func NumFields(o any) int {
 	t := reflect.TypeOf(o)
 
 	// If it's a pointer, get the underlying type
-	if t.Kind() == reflect.Ptr {
+	if t.Kind() == reflect.Pointer {
 		t = t.Elem()
 	}
 
@@ -111,7 +111,7 @@ func FieldIndex(o any, fieldName string) (int, error) {
 	t := reflect.TypeOf(o)
 
 	// If it's a pointer, get the underlying type
-	if t.Kind() == reflect.Ptr {
+	if t.Kind() == reflect.Pointer {
 		t = t.Elem()
 	}