Refactored for processor injection and restructured code

Author: silaslenihan

pkg/report/datafeeds/evm/decode.go

@@ -1,15 +1,14 @@
-package evm
+package datafeeds
 
 import (
 	"fmt"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
-	"github.com/smartcontractkit/chainlink-evm/pkg/report/datafeeds"
 )
 
 type Report struct {
-	datafeeds.Metadata
+	Metadata
 	Data []byte
 }
 

pkg/report/datafeeds/evm/types.go pkg/report/datafeeds/types.gopkg/report/datafeeds/evm/types.go renamed to pkg/report/datafeeds/types.go

@@ -1,15 +1,15 @@
-package evm_test
+package datafeeds_test
 
 import (
 	"math/big"
 	"testing"
 
-	"github.com/smartcontractkit/chainlink-evm/pkg/report/datafeeds/evm"
+	"github.com/smartcontractkit/chainlink-evm/pkg/report/datafeeds"
 )
 
 func TestDecodeFeedReport(t *testing.T) {
 	// Create some sample records.
-	original := []evm.FeedReport{
+	original := []datafeeds.FeedReport{
 		{
 			// Example feedID: the first byte is 0x01 and the remainder are zeros.
 			FeedID:    [32]byte{0x01},
@@ -24,7 +24,7 @@ func TestDecodeFeedReport(t *testing.T) {
 	}
 
 	// Get the ABI schema from our constructor.
-	schema := evm.GetSchema()
+	schema := datafeeds.GetSchema()
 
 	// Pack the original data using the ABI schema.
 	encoded, err := schema.Pack(original)
@@ -33,7 +33,7 @@ func TestDecodeFeedReport(t *testing.T) {
 	}
 
 	// Decode the data using our Decode function.
-	decoded, err := evm.Decode(encoded)
+	decoded, err := datafeeds.Decode(encoded)
 	if err != nil {
 		t.Fatalf("failed to decode data: %v", err)
 	}

pkg/report/datafeeds/evm/types_test.go pkg/report/datafeeds/types_test.gopkg/report/datafeeds/evm/types_test.go renamed to pkg/report/datafeeds/types_test.go

@@ -1,14 +1,43 @@
 package registry
 
 import (
+	"fmt"
 	"math"
 	"math/big"
 
 	wt_msg "github.com/smartcontractkit/chainlink-evm/pkg/report/pb/platform"
+	"github.com/smartcontractkit/chainlink-evm/pkg/report/platform"
 
 	"github.com/smartcontractkit/chainlink-evm/pkg/report/datafeeds"
 )
 
+func DecodeAsFeedUpdated(m *wt_msg.WriteConfirmed) ([]*FeedUpdated, error) {
+	// Decode the confirmed report (WT -> DF contract event)
+	r, err := platform.Decode(m.Report)
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode report: %w", err)
+	}
+
+	// Decode the underlying Data Feeds reports
+	reports, err := datafeeds.Decode(r.Data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode Data Feeds report: %w", err)
+	}
+
+	// Allocate space for the messages (event per updated feed)
+	msgs := make([]*FeedUpdated, 0, len(*reports))
+
+	// Iterate over the underlying Mercury reports
+	for _, rf := range *reports {
+		feedID := datafeeds.FeedID(rf.FeedID)
+
+		// TODO: unsure if r.Data is correct for Report
+		msgs = append(msgs, NewFeedUpdated(m, feedID, rf.Timestamp, rf.Price, r.Data, false))
+	}
+
+	return msgs, nil
+}
+
 // newFeedUpdated creates a FeedUpdated from the given common parameters.
 // If includeTxInfo is true, TxSender and TxReceiver are set.
 func NewFeedUpdated(
@@ -24,7 +53,7 @@ func NewFeedUpdated(
 		ObservationsTimestamp: observationsTimestamp,
 		Benchmark:             benchmarkPrice.Bytes(),
 		Report:                report,
-		BenchmarkVal:          ToBenchmarkVal(feedID, benchmarkPrice),
+		BenchmarkVal:          toBenchmarkVal(feedID, benchmarkPrice),
 
 		// Head data - when was the event produced on-chain
 		BlockHash:      m.BlockHash,
@@ -71,7 +100,7 @@ func NewFeedUpdated(
 // for most use-cases. For big numbers, benchmark bytes should be used instead.
 //
 // Returns `math.NaN()` if report data type not a number, or `+/-Inf` if number doesn't fit in double.
-func ToBenchmarkVal(feedID datafeeds.FeedID, val *big.Int) float64 {
+func toBenchmarkVal(feedID datafeeds.FeedID, val *big.Int) float64 {
 	// Return NaN if the value is nil
 	if val == nil {
 		return math.NaN()

pkg/report/pb/data-feeds/on-chain/registry/decode.go

@@ -1,5 +1,5 @@
 // evm_feedupdated_test.go
-package evm_test
+package registry_test
 
 import (
 	"encoding/hex"
@@ -9,13 +9,13 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink-evm/pkg/report/datafeeds"
-	"github.com/smartcontractkit/chainlink-evm/pkg/report/datafeeds/evm"
 
+	"github.com/smartcontractkit/chainlink-evm/pkg/report/pb/data-feeds/on-chain/registry"
 	wt_msg "github.com/smartcontractkit/chainlink-evm/pkg/report/pb/platform"
 )
 
 func TestDecodeAsFeedUpdated(t *testing.T) {
-	feedReports := evm.Reports{
+	feedReports := datafeeds.Reports{
 		{
 			FeedID:    [32]byte{0x01},
 			Price:     big.NewInt(1234567890),
@@ -24,7 +24,7 @@ func TestDecodeAsFeedUpdated(t *testing.T) {
 	}
 
 	// ABI‑encode the feed reports using the global schema.
-	feedReportsEncoded, err := evm.GetSchema().Pack(feedReports)
+	feedReportsEncoded, err := datafeeds.GetSchema().Pack(feedReports)
 	require.NoError(t, err)
 
 	metadata := datafeeds.Metadata{
@@ -40,7 +40,7 @@ func TestDecodeAsFeedUpdated(t *testing.T) {
 	}
 
 	// Build the full Report with the metadata and the already ABI‑encoded feed reports.
-	report := evm.Report{
+	report := datafeeds.Report{
 		Metadata: metadata,
 		Data:     feedReportsEncoded,
 	}
@@ -66,7 +66,7 @@ func TestDecodeAsFeedUpdated(t *testing.T) {
 	}
 
 	// Decode the WriteConfirmed message.
-	feedUpdates, err := evm.DecodeAsFeedUpdated(&validMsg)
+	feedUpdates, err := registry.DecodeAsFeedUpdated(&validMsg)
 	require.NoError(t, err)
 
 	require.Len(t, feedUpdates, len(feedReports))
@@ -85,6 +85,6 @@ func TestDecodeAsFeedUpdated(t *testing.T) {
 	invalidMsg := wt_msg.WriteConfirmed{
 		Report: []byte{0x01, 0x02, 0x03, 0x04},
 	}
-	_, err = evm.DecodeAsFeedUpdated(&invalidMsg)
+	_, err = registry.DecodeAsFeedUpdated(&invalidMsg)
 	require.Error(t, err)
 }

pkg/report/datafeeds/evm/decode_test.go …a-feeds/on-chain/registry/decode_test.gopkg/report/datafeeds/evm/decode_test.go renamed to pkg/report/pb/data-feeds/on-chain/registry/decode_test.go pkg/report/datafeeds/evm/decode_test.go renamed to pkg/report/pb/data-feeds/on-chain/registry/decode_test.go

@@ -30,15 +30,10 @@ const (
 // includes decoding messages as specific types and deriving metrics based on the decoded messages.
 // TODO: Report decoding uses the same ABI for EVM and Aptos, however, future chains may need a different
 // decoding scheme. Generalize this in the future to support different chains and decoding schemes.
-func NewMonitor(ctx context.Context, lggr logger.Logger, decodeFn func(m *wt.WriteConfirmed) ([]*registry.FeedUpdated, error)) (*monitor.BeholderClient, error) {
+func NewMonitor(lggr logger.Logger, chainSpecificProcessors []monitor.ProtoProcessor) (*monitor.BeholderClient, error) {
 	// Initialize the Beholder client with a local logger a custom Emitter
 	client := beholder.GetClient().ForPackage("write_target")
 
-	registryMetrics, err := registry.NewMetrics()
-	if err != nil {
-		return nil, fmt.Errorf("failed to create new registry metrics: %w", err)
-	}
-
 	forwarderMetrics, err := forwarder.NewMetrics()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new forwarder metrics: %w", err)
@@ -54,13 +49,9 @@ func NewMonitor(ctx context.Context, lggr logger.Logger, decodeFn func(m *wt.Wri
 
 	// Proxy ProtoEmitter with additional processing
 	protoEmitterProxy := protoEmitter{
-		lggr:    lggr,
-		emitter: emitter,
-		processors: []monitor.ProtoProcessor{
-			&wtProcessor{wtMetrics},
-			&keystoneProcessor{emitter, forwarderMetrics},
-			&dataFeedsProcessor{emitter, registryMetrics, decodeFn},
-		},
+		lggr:       lggr,
+		emitter:    emitter,
+		processors: append(chainSpecificProcessors, &wtProcessor{wtMetrics}, &keystoneProcessor{emitter, forwarderMetrics}),
 	}
 	return &monitor.BeholderClient{Client: &client, ProtoEmitter: &protoEmitterProxy}, nil
 }