eth,eth/watcher: Create Chainlink price feed watcher

Makefile

@@ -1,7 +1,10 @@
 SHELL=/bin/bash
 GO_BUILD_DIR?="./"
 
-all: net/lp_rpc.pb.go net/redeemer.pb.go net/redeemer_mock.pb.go core/test_segment.go livepeer livepeer_cli livepeer_router livepeer_bench
+MOCKGEN=go run github.com/golang/mock/mockgen
+ABIGEN=go run github.com/ethereum/go-ethereum/cmd/abigen
+
+all: net/lp_rpc.pb.go net/redeemer.pb.go net/redeemer_mock.pb.go core/test_segment.go eth/contracts/chainlink/AggregatorV3Interface.go livepeer livepeer_cli livepeer_router livepeer_bench
 
 net/lp_rpc.pb.go: net/lp_rpc.proto
 	protoc -I=. --go_out=. --go-grpc_out=. $^
@@ -10,12 +13,21 @@ net/redeemer.pb.go: net/redeemer.proto
 	protoc -I=. --go_out=. --go-grpc_out=. $^
 
 net/redeemer_mock.pb.go net/redeemer_grpc_mock.pb.go: net/redeemer.pb.go net/redeemer_grpc.pb.go
-	@mockgen -source net/redeemer.pb.go -destination net/redeemer_mock.pb.go -package net
-	@mockgen -source net/redeemer_grpc.pb.go -destination net/redeemer_grpc_mock.pb.go -package net
+	@$(MOCKGEN) -source net/redeemer.pb.go -destination net/redeemer_mock.pb.go -package net
+	@$(MOCKGEN) -source net/redeemer_grpc.pb.go -destination net/redeemer_grpc_mock.pb.go -package net
 
 core/test_segment.go:
 	core/test_segment.sh core/test_segment.go
 
+eth/contracts/chainlink/AggregatorV3Interface.go:
+	solc --version | grep 0.7.6+commit.7338295f
+	@set -ex; \
+	for sol_file in eth/contracts/chainlink/*.sol; do \
+		contract_name=$$(basename "$$sol_file" .sol); \
+		solc --abi --optimize --overwrite -o $$(dirname "$$sol_file") $$sol_file; \
+		$(ABIGEN) --abi=$${sol_file%.sol}.abi --pkg=chainlink --type=$$contract_name --out=$${sol_file%.sol}.go; \
+	done
+
 version=$(shell cat VERSION)
 
 ldflags := -X github.com/livepeer/go-livepeer/core.LivepeerVersion=$(shell ./print_version.sh)

eth/contracts/chainlink/AggregatorV3Interface.abi

@@ -0,0 +1 @@
+[{"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"description","outputs":[{"internalType":"string","name":"","type":"string"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"uint80","name":"_roundId","type":"uint80"}],"name":"getRoundData","outputs":[{"internalType":"uint80","name":"roundId","type":"uint80"},{"internalType":"int256","name":"answer","type":"int256"},{"internalType":"uint256","name":"startedAt","type":"uint256"},{"internalType":"uint256","name":"updatedAt","type":"uint256"},{"internalType":"uint80","name":"answeredInRound","type":"uint80"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"latestRoundData","outputs":[{"internalType":"uint80","name":"roundId","type":"uint80"},{"internalType":"int256","name":"answer","type":"int256"},{"internalType":"uint256","name":"startedAt","type":"uint256"},{"internalType":"uint256","name":"updatedAt","type":"uint256"},{"internalType":"uint80","name":"answeredInRound","type":"uint80"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"version","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"stateMutability":"view","type":"function"}]

eth/contracts/chainlink/AggregatorV3Interface.sol

@@ -0,0 +1,36 @@
+// SPDX-License-Identifier: MIT
+// https://github.com/smartcontractkit/chainlink/blob/v2.9.1/contracts/src/v0.7/interfaces/AggregatorV3Interface.sol
+pragma solidity ^0.7.0;
+
+interface AggregatorV3Interface {
+    function decimals() external view returns (uint8);
+
+    function description() external view returns (string memory);
+
+    function version() external view returns (uint256);
+
+    // getRoundData and latestRoundData should both raise "No data present"
+    // if they do not have data to report, instead of returning unset values
+    // which could be misinterpreted as actual reported values.
+    function getRoundData(uint80 _roundId)
+        external
+        view
+        returns (
+            uint80 roundId,
+            int256 answer,
+            uint256 startedAt,
+            uint256 updatedAt,
+            uint80 answeredInRound
+        );
+
+    function latestRoundData()
+        external
+        view
+        returns (
+            uint80 roundId,
+            int256 answer,
+            uint256 startedAt,
+            uint256 updatedAt,
+            uint80 answeredInRound
+        );
+}

eth/pricefeed.go

@@ -0,0 +1,117 @@
+package eth
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/big"
+	"regexp"
+	"time"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/livepeer/go-livepeer/eth/contracts/chainlink"
+)
+
+type PriceData struct {
+	RoundID   int64
+	Price     *big.Rat
+	UpdatedAt time.Time
+}
+
+// PriceFeedEthClient is an interface for fetching price data from a Chainlink
+// PriceFeed contract.
+type PriceFeedEthClient interface {
+	Description() (string, error)
+	FetchPriceData() (PriceData, error)
+}
+
+func NewPriceFeedEthClient(ctx context.Context, rpcUrl, priceFeedAddr string) (PriceFeedEthClient, error) {
+	client, err := ethclient.DialContext(ctx, rpcUrl)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize client: %w", err)
+	}
+
+	ok := isContractAddress(priceFeedAddr, client)
+	if !ok {
+		return nil, fmt.Errorf("not a contract address: %s", priceFeedAddr)
+	}
+
+	addr := common.HexToAddress(priceFeedAddr)
+	priceFeed, err := chainlink.NewAggregatorV3Interface(addr, client)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create mock aggregator proxy: %w", err)
+	}
+
+	return &priceFeedClient{
+		client:    client,
+		priceFeed: priceFeed,
+	}, nil
+}
+
+type priceFeedClient struct {
+	client    *ethclient.Client
+	priceFeed *chainlink.AggregatorV3Interface
+}
+
+func (c *priceFeedClient) Description() (string, error) {
+	return c.priceFeed.Description(&bind.CallOpts{})
+}
+
+func (c *priceFeedClient) FetchPriceData() (PriceData, error) {
+	data, err := c.priceFeed.LatestRoundData(&bind.CallOpts{})
+	if err != nil {
+		return PriceData{}, errors.New("failed to get latest round data: " + err.Error())
+	}
+
+	decimals, err := c.priceFeed.Decimals(&bind.CallOpts{})
+	if err != nil {
+		return PriceData{}, errors.New("failed to get decimals: " + err.Error())
+	}
+
+	return computePriceData(data.RoundId, data.UpdatedAt, data.Answer, decimals), nil
+}
+
+// computePriceData transforms the raw data from the PriceFeed into the higher
+// level PriceData struct, more easily usable by the rest of the system.
+func computePriceData(roundID, updatedAt, answer *big.Int, decimals uint8) PriceData {
+	// Compute a big.int which is 10^decimals.
+	divisor := new(big.Int).Exp(
+		big.NewInt(10),
+		big.NewInt(int64(decimals)),
+		nil)
+
+	return PriceData{
+		RoundID:   roundID.Int64(),
+		Price:     new(big.Rat).SetFrac(answer, divisor),
+		UpdatedAt: time.Unix(updatedAt.Int64(), 0),
+	}
+}
+
+type ethClient interface {
+	CodeAt(ctx context.Context, contract common.Address, blockNumber *big.Int) ([]byte, error)
+}
+
+// isContractAddress checks if the given address is an address of a contract
+// deployed on the corresponding blockchain.
+func isContractAddress(addr string, client ethClient) bool {
+	if len(addr) == 0 {
+		return false
+	}
+
+	// Ensure it is an Ethereum address: 0x followed by 40 hexadecimal characters.
+	re := regexp.MustCompile("^0x[0-9a-fA-F]{40}$")
+	if !re.MatchString(addr) {
+		return false
+	}
+
+	// Ensure it is a contract address.
+	address := common.HexToAddress(addr)
+	bytecode, err := client.CodeAt(context.Background(), address, nil) // nil is latest block
+	if err != nil {
+		return false
+	}
+	isContract := len(bytecode) > 0
+	return isContract
+}

eth/pricefeed_test.go

@@ -0,0 +1,93 @@
+package eth
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+type mockEthClient struct {
+	mock.Mock
+}
+
+func (m *mockEthClient) CodeAt(ctx context.Context, contract common.Address, blockNumber *big.Int) ([]byte, error) {
+	args := m.Called(ctx, contract, blockNumber)
+	return args.Get(0).([]byte), args.Error(1)
+}
+
+func TestIsContractAddress(t *testing.T) {
+	assert := assert.New(t)
+
+	t.Run("invalid address", func(t *testing.T) {
+		addr := "0x123"
+		mockClient := new(mockEthClient)
+		assert.False(isContractAddress(addr, mockClient))
+	})
+
+	t.Run("empty address", func(t *testing.T) {
+		addr := ""
+		mockClient := new(mockEthClient)
+		assert.False(isContractAddress(addr, mockClient))
+	})
+
+	t.Run("valid address but not contract", func(t *testing.T) {
+		addr := "0xAb5801a7D398351b8bE11C439e05C5B3259aeC9B"
+		mockClient := new(mockEthClient)
+		mockClient.On("CodeAt", mock.Anything, common.HexToAddress(addr), mock.Anything).Return([]byte{}, nil)
+		assert.False(isContractAddress(addr, mockClient))
+	})
+
+	t.Run("valid contract address", func(t *testing.T) {
+		addr := "0xAb5801a7D398351b8bE11C439e05C5B3259aeC9B"
+		mockClient := new(mockEthClient)
+		mockClient.On("CodeAt", mock.Anything, common.HexToAddress(addr), mock.Anything).Return([]byte{0x1}, nil)
+		assert.True(isContractAddress(addr, mockClient))
+	})
+}
+
+func TestComputePriceData(t *testing.T) {
+	assert := assert.New(t)
+
+	t.Run("valid data", func(t *testing.T) {
+		roundID := big.NewInt(1)
+		updatedAt := big.NewInt(1626192000)
+		answer := big.NewInt(420666000)
+		decimals := uint8(6)
+
+		data := computePriceData(roundID, updatedAt, answer, decimals)
+
+		assert.EqualValues(int64(1), data.RoundID, "Round ID didn't match")
+		assert.Equal("210333/500", data.Price.RatString(), "The Price Rat didn't match")
+		assert.Equal("2021-07-13 16:00:00 +0000 UTC", data.UpdatedAt.UTC().String(), "The updated at time did not match")
+	})
+
+	t.Run("zero answer", func(t *testing.T) {
+		roundID := big.NewInt(2)
+		updatedAt := big.NewInt(1626192000)
+		answer := big.NewInt(0)
+		decimals := uint8(18)
+
+		data := computePriceData(roundID, updatedAt, answer, decimals)
+
+		assert.EqualValues(int64(2), data.RoundID, "Round ID didn't match")
+		assert.Equal("0", data.Price.RatString(), "The Price Rat didn't match")
+		assert.Equal("2021-07-13 16:00:00 +0000 UTC", data.UpdatedAt.UTC().String(), "The updated at time did not match")
+	})
+
+	t.Run("zero decimals", func(t *testing.T) {
+		roundID := big.NewInt(3)
+		updatedAt := big.NewInt(1626192000)
+		answer := big.NewInt(13)
+		decimals := uint8(0)
+
+		data := computePriceData(roundID, updatedAt, answer, decimals)
+
+		assert.EqualValues(int64(3), data.RoundID, "Round ID didn't match")
+		assert.Equal("13", data.Price.RatString(), "The Price Rat didn't match")
+		assert.Equal("2021-07-13 16:00:00 +0000 UTC", data.UpdatedAt.UTC().String(), "The updated at time did not match")
+	})
+}