[FUN-1234] refactor logpoller_wrapper by injecting a coordinator dep (#…

…12066) * fix: test race * Revert "(test): Run LogPollerWrapperTest v1/v2 tests sequentially" This reverts commit 1ba2c06. * chore: refactor logpoller_wrapper by injecting a coordinator dep * fix: remove duplicates abitype definitions * fix: lint sonarqube reliability issue * fix: race conditions due to share variables --------- Co-authored-by: Justin Kaseman <[email protected]>
smartcontractkit · Feb 19, 2024 · 2adfc1b · 2adfc1b
1 parent f5afdb8
commit 2adfc1b
Show file tree

Hide file tree

Showing 4 changed files with 529 additions and 377 deletions.
diff --git a/core/services/relay/evm/functions/coordinator_v1.go b/core/services/relay/evm/functions/coordinator_v1.go
@@ -0,0 +1,147 @@
+package functions
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/client"
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
+	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/functions/generated/functions_coordinator_1_1_0"
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	evmRelayTypes "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/types"
+)
+
+type CoordinatorV1 struct {
+	address common.Address
+
+	client    client.Client
+	logPoller logpoller.LogPoller
+	lggr      logger.Logger
+}
+
+func NewCoordinatorV1(address common.Address, client client.Client, logPoller logpoller.LogPoller, lggr logger.Logger) *CoordinatorV1 {
+	return &CoordinatorV1{
+		address:   address,
+		client:    client,
+		logPoller: logPoller,
+		lggr:      lggr,
+	}
+}
+
+func (c *CoordinatorV1) Address() common.Address {
+	return c.address
+}
+
+func (c *CoordinatorV1) RegisterFilters() error {
+	if (c.address == common.Address{}) {
+		return nil
+	}
+
+	return c.logPoller.RegisterFilter(
+		logpoller.Filter{
+			Name: logpoller.FilterName("FunctionsLogPollerWrapper", c.address.String(), "-v", "1"),
+			EventSigs: []common.Hash{
+				functions_coordinator_1_1_0.FunctionsCoordinator110OracleRequest{}.Topic(),
+				functions_coordinator_1_1_0.FunctionsCoordinator110OracleResponse{}.Topic(),
+			},
+			Addresses: []common.Address{c.address},
+		})
+}
+
+func (c *CoordinatorV1) OracleRequestLogTopic() (common.Hash, error) {
+	return functions_coordinator_1_1_0.FunctionsCoordinator110OracleRequest{}.Topic(), nil
+}
+
+func (c *CoordinatorV1) OracleResponseLogTopic() (common.Hash, error) {
+	return functions_coordinator_1_1_0.FunctionsCoordinator110OracleResponse{}.Topic(), nil
+}
+
+func (c *CoordinatorV1) LogsToRequests(requestLogs []logpoller.Log) ([]evmRelayTypes.OracleRequest, error) {
+	var requests []evmRelayTypes.OracleRequest
+
+	parsingContract, err := functions_coordinator_1_1_0.NewFunctionsCoordinator110(c.address, c.client)
+	if err != nil {
+		return nil, fmt.Errorf("LogsToRequests: creating a contract instance for NewFunctionsCoordinator110 parsing failed: %w", err)
+	}
+
+	for _, log := range requestLogs {
+		gethLog := log.ToGethLog()
+		oracleRequest, err := parsingContract.ParseOracleRequest(gethLog)
+		if err != nil {
+			c.lggr.Errorw("LogsToRequests: failed to parse a request log, skipping", "err", err)
+			continue
+		}
+
+		commitmentABIV1 := abi.Arguments{
+			{Type: bytes32Type}, // RequestId
+			{Type: addressType}, // Coordinator
+			{Type: uint96Type},  // EstimatedTotalCostJuels
+			{Type: addressType}, // Client
+			{Type: uint64Type},  // SubscriptionId
+			{Type: uint32Type},  // CallbackGasLimit
+			{Type: uint72Type},  // AdminFee
+			{Type: uint72Type},  // DonFee
+			{Type: uint40Type},  // GasOverheadBeforeCallback
+			{Type: uint40Type},  // GasOverheadAfterCallback
+			{Type: uint32Type},  // TimeoutTimestamp
+		}
+
+		commitmentBytesV1, err := commitmentABIV1.Pack(
+			oracleRequest.Commitment.RequestId,
+			oracleRequest.Commitment.Coordinator,
+			oracleRequest.Commitment.EstimatedTotalCostJuels,
+			oracleRequest.Commitment.Client,
+			oracleRequest.Commitment.SubscriptionId,
+			oracleRequest.Commitment.CallbackGasLimit,
+			oracleRequest.Commitment.AdminFee,
+			oracleRequest.Commitment.DonFee,
+			oracleRequest.Commitment.GasOverheadBeforeCallback,
+			oracleRequest.Commitment.GasOverheadAfterCallback,
+			oracleRequest.Commitment.TimeoutTimestamp,
+		)
+		if err != nil {
+			c.lggr.Errorw("LogsToRequests: failed to pack Coordinator v1 commitment bytes, skipping", err)
+		}
+
+		OracleRequestV1 := evmRelayTypes.OracleRequest{
+			RequestId:           oracleRequest.RequestId,
+			RequestingContract:  oracleRequest.RequestingContract,
+			RequestInitiator:    oracleRequest.RequestInitiator,
+			SubscriptionId:      oracleRequest.SubscriptionId,
+			SubscriptionOwner:   oracleRequest.SubscriptionOwner,
+			Data:                oracleRequest.Data,
+			DataVersion:         oracleRequest.DataVersion,
+			Flags:               oracleRequest.Flags,
+			CallbackGasLimit:    oracleRequest.CallbackGasLimit,
+			TxHash:              oracleRequest.Raw.TxHash,
+			OnchainMetadata:     commitmentBytesV1,
+			CoordinatorContract: c.address,
+		}
+
+		requests = append(requests, OracleRequestV1)
+	}
+	return requests, nil
+}
+
+func (c *CoordinatorV1) LogsToResponses(responseLogs []logpoller.Log) ([]evmRelayTypes.OracleResponse, error) {
+	var responses []evmRelayTypes.OracleResponse
+
+	parsingContract, err := functions_coordinator_1_1_0.NewFunctionsCoordinator110(c.address, c.client)
+	if err != nil {
+		return nil, fmt.Errorf("LogsToResponses: creating a contract instance for parsing failed: %w", err)
+	}
+	for _, log := range responseLogs {
+		gethLog := log.ToGethLog()
+		oracleResponse, err := parsingContract.ParseOracleResponse(gethLog)
+		if err != nil {
+			c.lggr.Errorw("LogsToResponses: failed to parse a response log, skipping")
+			continue
+		}
+		responses = append(responses, evmRelayTypes.OracleResponse{
+			RequestId: oracleResponse.RequestId,
+		})
+	}
+	return responses, nil
+}
diff --git a/core/services/relay/evm/functions/coordinator_v2.go b/core/services/relay/evm/functions/coordinator_v2.go
@@ -0,0 +1,146 @@
+package functions
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/client"
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller"
+	"github.com/smartcontractkit/chainlink/v2/core/gethwrappers/functions/generated/functions_coordinator"
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	evmRelayTypes "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/types"
+)
+
+type CoordinatorV2 struct {
+	address common.Address
+
+	client    client.Client
+	logPoller logpoller.LogPoller
+	lggr      logger.Logger
+}
+
+func NewCoordinatorV2(address common.Address, client client.Client, logPoller logpoller.LogPoller, lggr logger.Logger) *CoordinatorV2 {
+	return &CoordinatorV2{
+		address:   address,
+		client:    client,
+		logPoller: logPoller,
+		lggr:      lggr,
+	}
+}
+
+func (c *CoordinatorV2) Address() common.Address {
+	return c.address
+}
+
+func (c *CoordinatorV2) RegisterFilters() error {
+	if (c.address == common.Address{}) {
+		return nil
+	}
+
+	return c.logPoller.RegisterFilter(
+		logpoller.Filter{
+			Name: logpoller.FilterName("FunctionsLogPollerWrapper", c.address.String(), "-v", "2"),
+			EventSigs: []common.Hash{
+				functions_coordinator.FunctionsCoordinatorOracleRequest{}.Topic(),
+				functions_coordinator.FunctionsCoordinatorOracleResponse{}.Topic(),
+			},
+			Addresses: []common.Address{c.address},
+		})
+}
+func (c *CoordinatorV2) OracleRequestLogTopic() (common.Hash, error) {
+	return functions_coordinator.FunctionsCoordinatorOracleRequest{}.Topic(), nil
+}
+func (c *CoordinatorV2) OracleResponseLogTopic() (common.Hash, error) {
+	return functions_coordinator.FunctionsCoordinatorOracleResponse{}.Topic(), nil
+}
+func (c *CoordinatorV2) LogsToRequests(requestLogs []logpoller.Log) ([]evmRelayTypes.OracleRequest, error) {
+	var requests []evmRelayTypes.OracleRequest
+
+	parsingContract, err := functions_coordinator.NewFunctionsCoordinator(c.address, c.client)
+	if err != nil {
+		return nil, fmt.Errorf("LogsToRequests: creating a contract instance for NewFunctionsCoordinator parsing failed: %w", err)
+	}
+
+	for _, log := range requestLogs {
+		gethLog := log.ToGethLog()
+		oracleRequest, err := parsingContract.ParseOracleRequest(gethLog)
+		if err != nil {
+			c.lggr.Errorw("LogsToRequests: failed to parse a request log, skipping", "err", err)
+			continue
+		}
+
+		commitmentABIV2 := abi.Arguments{
+			{Type: bytes32Type}, // RequestId
+			{Type: addressType}, // Coordinator
+			{Type: uint96Type},  // EstimatedTotalCostJuels
+			{Type: addressType}, // Client
+			{Type: uint64Type},  // SubscriptionId
+			{Type: uint32Type},  // CallbackGasLimit
+			{Type: uint72Type},  // AdminFee
+			{Type: uint72Type},  // DonFee
+			{Type: uint40Type},  // GasOverheadBeforeCallback
+			{Type: uint40Type},  // GasOverheadAfterCallback
+			{Type: uint32Type},  // TimeoutTimestamp
+			{Type: uint72Type},  // OperationFee
+		}
+
+		commitmentBytesV2, err := commitmentABIV2.Pack(
+			oracleRequest.Commitment.RequestId,
+			oracleRequest.Commitment.Coordinator,
+			oracleRequest.Commitment.EstimatedTotalCostJuels,
+			oracleRequest.Commitment.Client,
+			oracleRequest.Commitment.SubscriptionId,
+			oracleRequest.Commitment.CallbackGasLimit,
+			oracleRequest.Commitment.AdminFeeJuels,
+			oracleRequest.Commitment.DonFeeJuels,
+			oracleRequest.Commitment.GasOverheadBeforeCallback,
+			oracleRequest.Commitment.GasOverheadAfterCallback,
+			oracleRequest.Commitment.TimeoutTimestamp,
+			oracleRequest.Commitment.OperationFeeJuels,
+		)
+		if err != nil {
+			c.lggr.Errorw("LogsToRequests: failed to pack Coordinator v2 commitment bytes, skipping", err)
+		}
+
+		OracleRequestV2 := evmRelayTypes.OracleRequest{
+			RequestId:           oracleRequest.RequestId,
+			RequestingContract:  oracleRequest.RequestingContract,
+			RequestInitiator:    oracleRequest.RequestInitiator,
+			SubscriptionId:      oracleRequest.SubscriptionId,
+			SubscriptionOwner:   oracleRequest.SubscriptionOwner,
+			Data:                oracleRequest.Data,
+			DataVersion:         oracleRequest.DataVersion,
+			Flags:               oracleRequest.Flags,
+			CallbackGasLimit:    oracleRequest.CallbackGasLimit,
+			TxHash:              oracleRequest.Raw.TxHash,
+			OnchainMetadata:     commitmentBytesV2,
+			CoordinatorContract: c.address,
+		}
+
+		requests = append(requests, OracleRequestV2)
+	}
+	return requests, nil
+}
+
+func (c *CoordinatorV2) LogsToResponses(responseLogs []logpoller.Log) ([]evmRelayTypes.OracleResponse, error) {
+	var responses []evmRelayTypes.OracleResponse
+
+	parsingContract, err := functions_coordinator.NewFunctionsCoordinator(c.address, c.client)
+	if err != nil {
+		return nil, fmt.Errorf("LogsToResponses: creating a contract instance for parsing failed: %w", err)
+	}
+	for _, log := range responseLogs {
+		gethLog := log.ToGethLog()
+		oracleResponse, err := parsingContract.ParseOracleResponse(gethLog)
+		if err != nil {
+			c.lggr.Errorw("LogsToResponses: failed to parse a response log, skipping")
+			continue
+		}
+		responses = append(responses, evmRelayTypes.OracleResponse{
+			RequestId: oracleResponse.RequestId,
+		})
+	}
+	return responses, nil
+}