feat: ethereum oracles in datanode

core/evtforward/ethcall/result.go

@@ -1,9 +1,10 @@
 package ethcall
 
 import (
-	"code.vegaprotocol.io/vega/core/types"
 	"encoding/json"
 	"fmt"
+
+	"code.vegaprotocol.io/vega/core/types"
 	"github.com/PaesslerAG/jsonpath"
 )
 

core/oracles/engine.go

@@ -216,6 +216,14 @@ func (e *Engine) sendMatchedOracleData(ctx context.Context, data OracleData, spe
 		})
 	}
 
+	metaData := make([]*datapb.Property, 0, len(data.MetaData))
+	for name, value := range data.MetaData {
+		metaData = append(metaData, &datapb.Property{
+			Name:  name,
+			Value: value,
+		})
+	}
+
 	sort.Slice(payload, func(i, j int) bool {
 		return strings.Compare(payload[i].Name, payload[j].Name) < 0
 	})
@@ -237,6 +245,7 @@ func (e *Engine) sendMatchedOracleData(ctx context.Context, data OracleData, spe
 				Data:           payload,
 				MatchedSpecIds: ids,
 				BroadcastAt:    e.timeService.GetTimeNow().UnixNano(),
+				MetaData:       metaData,
 			},
 		},
 	}

core/oracles/ethereum_oracle_verifier.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -209,6 +210,8 @@ func (s *EthereumOracleVerifier) OnTick(ctx context.Context, t time.Time) {
 		s.oracleEngine.BroadcastData(ctx, OracleData{
 			Signers: nil,
 			Data:    result.Normalised,
+			// For now
+			MetaData: map[string]string{"eth-bloch-height": strconv.FormatUint(callResult.BlockHeight, 10)},
 		})
 	}
 

core/oracles/oracle_data.go

@@ -24,8 +24,9 @@ import (
 
 // OracleData holds normalized data coming from an oracle.
 type OracleData struct {
-	Signers []*types.Signer
-	Data    map[string]string
+	Signers  []*types.Signer
+	Data     map[string]string
+	MetaData map[string]string
 }
 
 func (d OracleData) GetUint(name string) (*num.Uint, error) {

core/types/data_source_external_eth_oracle_test.go

@@ -50,7 +50,7 @@ func TestEthCallSpecFromProto(t *testing.T) {
 		assert.Nil(t, s.Trigger)
 	})
 
-	t.Run("non-empty", func(t *testing.T) {
+	t.Run("non-empty with error", func(t *testing.T) {
 		timeNow := uint64(time.Now().UnixNano())
 		protoSource := &vegapb.EthCallSpec{
 			Address: "test-eth-address",
@@ -113,6 +113,57 @@ func TestEthCallSpecFromProto(t *testing.T) {
 		assert.NotNil(t, ds.Trigger)
 		assert.IsType(t, &vegapb.EthCallTrigger_TimeTrigger{}, ds.Trigger.IntoEthCallTriggerProto().Trigger)
 	})
+
+	t.Run("non-empty", func(t *testing.T) {
+		timeNow := uint64(time.Now().UnixNano())
+		protoSource := &vegapb.EthCallSpec{
+			Address: "test-eth-address",
+			Abi: &structpb.ListValue{
+				Values: []*structpb.Value{
+					{
+						Kind: &structpb.Value_NumberValue{
+							NumberValue: float64(5),
+						},
+					},
+				},
+			},
+			Method: "test-method",
+			Args: []*structpb.Value{
+				structpb.NewStringValue("test-arg-value"),
+			},
+			Trigger: &vegapb.EthCallTrigger{
+				Trigger: &vegapb.EthCallTrigger_TimeTrigger{
+					TimeTrigger: &vegapb.EthTimeTrigger{
+						Initial: &timeNow,
+					},
+				},
+			},
+			Filters: []*v1.Filter{
+				{
+					Key: &v1.PropertyKey{
+						Name: "test-key",
+						Type: v1.PropertyKey_Type(1),
+					},
+				},
+			},
+		}
+
+		ds, err := types.EthCallSpecFromProto(protoSource)
+		assert.Nil(t, err)
+		assert.IsType(t, types.EthCallSpec{}, ds)
+
+		assert.Equal(t, "test-eth-address", ds.Address)
+		assert.Equal(t, []byte{91, 53, 93}, ds.AbiJson)
+		assert.Equal(t, "test-method", ds.Method)
+		assert.Equal(t, []string{"\"test-arg-value\""}, ds.ArgsJson)
+		filters := ds.Filters
+		assert.Equal(t, 1, len(filters))
+		assert.Equal(t, 0, len(filters[0].Conditions))
+		assert.Equal(t, "test-key", filters[0].Key.Name)
+		assert.Equal(t, v1.PropertyKey_Type(1), filters[0].Key.Type)
+		assert.NotNil(t, ds.Trigger)
+		assert.IsType(t, &vegapb.EthCallTrigger_TimeTrigger{}, ds.Trigger.IntoEthCallTriggerProto().Trigger)
+	})
 }
 
 func TestEthCallSpecIntoProto(t *testing.T) {

datanode/entities/data.go

@@ -74,6 +74,7 @@ func DeserializeSigners(data Signers) []*types.Signer {
 type Data struct {
 	Signers        Signers
 	Data           []Property
+	MetaData       []Property
 	MatchedSpecIds [][]byte // pgx automatically handles [][]byte to Postgres ByteaArray mappings
 	BroadcastAt    time.Time
 	TxHash         TxHash
@@ -89,18 +90,28 @@ func ExternalDataFromProto(data *datapb.ExternalData, txHash TxHash, vegaTime ti
 	properties := []Property{}
 	specIDs := [][]byte{}
 	signers := Signers{}
+	metaDataProperties := []Property{}
 
 	if data.Data != nil {
 		properties = make([]Property, 0, len(data.Data.Data))
 		specIDs = make([][]byte, 0, len(data.Data.MatchedSpecIds))
 
+		metaDataProperties = make([]Property, 0, len(data.Data.MetaData))
+
 		for _, property := range data.Data.Data {
 			properties = append(properties, Property{
 				Name:  property.Name,
 				Value: property.Value,
 			})
 		}
 
+		for _, m := range data.Data.MetaData {
+			metaDataProperties = append(metaDataProperties, Property{
+				Name:  m.Name,
+				Value: m.Value,
+			})
+		}
+
 		for _, specID := range data.Data.MatchedSpecIds {
 			id := SpecID(specID)
 			idBytes, err := id.Bytes()
@@ -121,6 +132,7 @@ func ExternalDataFromProto(data *datapb.ExternalData, txHash TxHash, vegaTime ti
 		Data: &Data{
 			Signers:        signers,
 			Data:           properties,
+			MetaData:       metaDataProperties,
 			MatchedSpecIds: specIDs,
 			BroadcastAt:    NanosToPostgresTimestamp(data.Data.BroadcastAt),
 			TxHash:         txHash,
@@ -134,11 +146,13 @@ func (od *ExternalData) ToProto() *datapb.ExternalData {
 	properties := []*datapb.Property{}
 	specIDs := []string{}
 	signersAsProto := []*datapb.Signer{}
+	metaDataProperties := []*datapb.Property{}
 
 	if od.Data != nil {
 		if od.Data.Data != nil {
 			properties = make([]*datapb.Property, 0, len(od.Data.Data))
 			specIDs = make([]string, 0, len(od.Data.MatchedSpecIds))
+			metaDataProperties = make([]*datapb.Property, 0, len(od.Data.MetaData))
 
 			for _, prop := range od.Data.Data {
 				properties = append(properties, &datapb.Property{
@@ -147,6 +161,13 @@ func (od *ExternalData) ToProto() *datapb.ExternalData {
 				})
 			}
 
+			for _, m := range od.Data.MetaData {
+				metaDataProperties = append(metaDataProperties, &datapb.Property{
+					Name:  m.Name,
+					Value: m.Value,
+				})
+			}
+
 			for _, id := range od.Data.MatchedSpecIds {
 				hexID := hex.EncodeToString(id)
 				specIDs = append(specIDs, hexID)
@@ -161,6 +182,7 @@ func (od *ExternalData) ToProto() *datapb.ExternalData {
 		Data: &datapb.Data{
 			Signers:        signersAsProto,
 			Data:           properties,
+			MetaData:       metaDataProperties,
 			MatchedSpecIds: specIDs,
 			BroadcastAt:    od.Data.BroadcastAt.UnixNano(),
 		},

datanode/entities/data_source_definition.go

@@ -41,6 +41,38 @@ func (s *DataSourceDefinition) GetOracle() (*DataSourceSpecConfiguration, error)
 	return ds, nil
 }
 
+func (s *DataSourceDefinition) GetEthOracle() (*EthCallSpec, error) {
+	ds := &EthCallSpec{}
+	data := s.Content()
+	if data != nil {
+		switch tp := data.(type) {
+		case *vega.EthCallSpec:
+			ds.Address = tp.Address
+			abi, _ := tp.GetAbi().MarshalJSON()
+			//if err != nil {
+			//}
+			ds.Abi = abi
+			ds.Method = tp.Method
+			args := tp.GetArgs()
+			for _, arg := range args {
+				jsonArg, err := arg.MarshalJSON()
+				if err != nil {
+					return nil, err // TODO: Fix all of the errors
+				}
+				ds.ArgsJson = append(ds.ArgsJson, string(jsonArg))
+			}
+			trigger, _ := types.EthCallTriggerFromProto(tp.Trigger)
+			//if err != nil {
+			//}
+			ds.Trigger = EthCallTrigger{EthCallTrigger: trigger}
+			ds.RequiredConfirmations = tp.RequiredConfirmations
+			ds.Filters = s.GetFilters()
+		}
+	}
+
+	return ds, nil
+}
+
 func (s *DataSourceDefinition) GetInternalTimeTrigger() *DataSourceSpecConfigurationTime {
 	ds := &DataSourceSpecConfigurationTime{
 		Conditions: []Condition{},
@@ -84,7 +116,9 @@ func (s *DataSourceDefinition) GetFilters() []Filter {
 	if data != nil {
 		switch tp := data.(type) {
 		case *vega.DataSourceSpecConfiguration:
-			filters = FiltersFromProto(tp.GetFilters())
+			filters = FiltersFromProto(tp.Filters)
+		case *vega.EthCallSpec:
+			filters = FiltersFromProto(tp.Filters)
 		}
 	}
 
@@ -118,6 +152,76 @@ type DataSourceSpecConfiguration struct {
 	Filters []Filter
 }
 
+type EthCallTrigger struct {
+	types.EthCallTrigger
+}
+
+type EthCallSpec struct {
+	Address               string
+	Abi                   []byte
+	Method                string
+	ArgsJson              []string
+	Trigger               EthCallTrigger
+	RequiredConfirmations uint64
+	Filters               []Filter
+}
+
+func (es *EthCallSpec) GetFilters() []Filter {
+	if es != nil {
+		return es.Filters
+	}
+
+	return []Filter{}
+}
+
+func (es *EthCallSpec) GetAddress() string {
+	if es != nil {
+		return es.Address
+	}
+
+	return ""
+}
+
+func (es *EthCallSpec) GetAbi() []byte {
+	if es != nil {
+		return es.Abi
+	}
+
+	return nil
+}
+
+func (es *EthCallSpec) GetMethod() string {
+	if es != nil {
+		return es.Method
+	}
+
+	return ""
+}
+
+func (es *EthCallSpec) GetArgs() []string {
+	if es != nil {
+		return es.ArgsJson
+	}
+
+	return []string{}
+}
+
+func (es *EthCallSpec) GetTrigger() EthCallTrigger {
+	if es != nil {
+		return es.Trigger
+	}
+
+	return EthCallTrigger{}
+}
+
+func (es *EthCallSpec) GetRequiredConfirmations() uint64 {
+	if es != nil {
+		return es.RequiredConfirmations
+	}
+
+	return uint64(0)
+}
+
 // DataSourceSpecConfigurationTime is a simplified version of the internal time
 // termination data source; only for internal use;
 // New internal types will be created for Cosmic Elevator new internal terminations.