Extract price feed contract addresses to config

README.md

@@ -29,7 +29,8 @@ Vulcanize DB is a set of tools that make it easier for developers to write appli
 
 ## Configuration
 - To use a local Ethereum node, copy `environments/public.toml.example` to
-  `environments/public.toml` and update the `ipcPath` and `levelDbPath`.
+  `environments/public.toml` and update the `ipcPath`, `levelDbPath`, 
+  `pipContractAddress`, `pepContractAddress`, and `repContractAddress`.
   - `ipcPath` should match the local node's IPC filepath:
       - when using geth:
         - The IPC file is called `geth.ipc`.
@@ -51,7 +52,10 @@ Vulcanize DB is a set of tools that make it easier for developers to write appli
           - Linux: `$HOME/.ethereum/geth/chaindata`
       - `levelDbPath` is irrelevant (and `coldImport` is currently unavailable) if only running parity.
 
-- See `environments/infura.toml` to configure commands to run against infura, if a local node is unavailable.
+  - `pepContractAddress`, `pipContractAddress`, and `repContractAddress` should match that medianizer
+     addresses for each pair on the chain you're tracking. See https://makerdao.com/feeds/
+     
+- See `environments/infura.toml` to configure commands to run against infura, if a local node is unavailable
 - Copy `environments/local.toml.example` to `environments/local.toml` to configure commands to run against a local node such as [Ganache](https://truffleframework.com/ganache) or [ganache-cli](https://github.com/trufflesuite/ganache-clihttps://github.com/trufflesuite/ganache-cli).
 
 ## Start syncing with postgres
@@ -75,7 +79,7 @@ Sync VulcanizeDB from the LevelDB underlying a Geth node.
 Syncs VulcanizeDB with the configured Ethereum node, populating only block headers.
 This command is useful when you want a minimal baseline from which to track targeted data on the blockchain (e.g. individual smart contract storage values).
 1. Start Ethereum node
-2. In a separate terminal start VulcanizeDB:
+1. In a separate terminal start VulcanizeDB:
     - `./vulcanizedb lightSync --config <config.toml> --starting-block-number <block-number>`
 
 ## Backfill Auction event logs from light sync
@@ -87,6 +91,12 @@ _Since auction contracts have not yet been deployed, this command will need to b
 1. Start Ethereum node
 1. In a separate terminal run the backfill command:
   - `./vulcanizedb backfillAuctionLogs --config <config.toml>`
+  
+## Sync in light mode with MakerDAO price feeds
+Sync VulcanizeDB with the configured Ethereum node, populating block headers as well as price feeds for MKR/USD, ETH/USD, and REP/USD.
+1. Start Ethereum node
+1. In a separate terminal window start VulcanizeDB
+    - `./vulcanizedb syncPriceFeeds --config <config.toml> --starting-block-number <block-number>`
 
 ## Start full environment in docker by single command
 

cmd/lightSync.go

@@ -76,26 +76,10 @@ func backFillAllHeaders(blockchain core.BlockChain, headerRepository datastore.H
 func lightSync() {
 	ticker := time.NewTicker(pollingInterval)
 	defer ticker.Stop()
-	rawRpcClient, err := rpc.Dial(ipc)
+	blockChain := getBlockChain()
-	if err != nil {
+	validateArgs(blockChain)
-		log.Fatal(err)
-	}
-	rpcClient := client.NewRpcClient(rawRpcClient, ipc)
-	ethClient := ethclient.NewClient(rawRpcClient)
-	client := client.NewEthClient(ethClient)
-	node := node.MakeNode(rpcClient)
-	transactionConverter := vRpc.NewRpcTransactionConverter(client)
-	blockChain := geth.NewBlockChain(client, node, transactionConverter)
-
-	lastBlock := blockChain.LastBlock().Int64()
-	if lastBlock == 0 {
-		log.Fatal("geth initial: state sync not finished")
-	}
-	if startingBlockNumber > lastBlock {
-		log.Fatal("starting block number > current block number")
-	}
-
 	db := utils.LoadPostgres(databaseConfig, blockChain.Node())
+
 	headerRepository := repositories.NewHeaderRepository(&db)
 	validator := history.NewHeaderValidator(blockChain, headerRepository, validationWindow, []transformers.Transformer{})
 	missingBlocksPopulated := make(chan int)
@@ -111,3 +95,27 @@ func lightSync() {
 		}
 	}
 }
+
+func validateArgs(blockChain *geth.BlockChain) {
+	lastBlock := blockChain.LastBlock().Int64()
+	if lastBlock == 0 {
+		log.Fatal("geth initial: state sync not finished")
+	}
+	if startingBlockNumber > lastBlock {
+		log.Fatal("starting block number > current block number")
+	}
+}
+
+func getBlockChain() *geth.BlockChain {
+	rawRpcClient, err := rpc.Dial(ipc)
+	if err != nil {
+		log.Fatal(err)
+	}
+	rpcClient := client.NewRpcClient(rawRpcClient, ipc)
+	ethClient := ethclient.NewClient(rawRpcClient)
+	client := client.NewEthClient(ethClient)
+	node := node.MakeNode(rpcClient)
+	transactionConverter := vRpc.NewRpcTransactionConverter(client)
+	blockChain := geth.NewBlockChain(client, node, transactionConverter)
+	return blockChain
+}

cmd/root.go

@@ -29,6 +29,9 @@ var (
 	databaseConfig      config.Database
 	ipc                 string
 	levelDbPath         string
+	pepContractAddress  string
+	pipContractAddress  string
+	repContractAddress  string
 	startingBlockNumber int64
 	syncAll             bool
 	endingBlockNumber   int64
@@ -49,6 +52,9 @@ func Execute() {
 func database(cmd *cobra.Command, args []string) {
 	ipc = viper.GetString("client.ipcpath")
 	levelDbPath = viper.GetString("client.leveldbpath")
+	pepContractAddress = viper.GetString("client.pepcontractaddress")
+	pipContractAddress = viper.GetString("client.pipcontractaddress")
+	repContractAddress = viper.GetString("client.repcontractaddress")
 	databaseConfig = config.Database{
 		Name:     viper.GetString("database.name"),
 		Hostname: viper.GetString("database.hostname"),
@@ -70,6 +76,9 @@ func init() {
 	rootCmd.PersistentFlags().String("database-password", "", "database password")
 	rootCmd.PersistentFlags().String("client-ipcPath", "", "location of geth.ipc file")
 	rootCmd.PersistentFlags().String("client-levelDbPath", "", "location of levelDb chaindata")
+	rootCmd.PersistentFlags().String("client-pepContractAddress", "", "mkr/usd price feed contract address")
+	rootCmd.PersistentFlags().String("client-pipContractAddress", "", "eth/usd price feed contract address")
+	rootCmd.PersistentFlags().String("client-repContractAddress", "", "rep/usd price feed contract address")
 
 	viper.BindPFlag("database.name", rootCmd.PersistentFlags().Lookup("database-name"))
 	viper.BindPFlag("database.port", rootCmd.PersistentFlags().Lookup("database-port"))
@@ -78,6 +87,9 @@ func init() {
 	viper.BindPFlag("database.password", rootCmd.PersistentFlags().Lookup("database-password"))
 	viper.BindPFlag("client.ipcPath", rootCmd.PersistentFlags().Lookup("client-ipcPath"))
 	viper.BindPFlag("client.levelDbPath", rootCmd.PersistentFlags().Lookup("client-levelDbPath"))
+	viper.BindPFlag("client.pepContractAddress", rootCmd.PersistentFlags().Lookup("client-pepContractAddress"))
+	viper.BindPFlag("client.pipContractAddress", rootCmd.PersistentFlags().Lookup("client-pipContractAddress"))
+	viper.BindPFlag("client.repContractAddress", rootCmd.PersistentFlags().Lookup("client-repContractAddress"))
 }
 
 func initConfig() {

cmd/syncPriceFeeds.go

@@ -1,4 +1,4 @@
-// Copyright © 2018 NAME HERE <EMAIL ADDRESS>
+// Copyright © 2018 Vulcanize
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -19,17 +19,11 @@ import (
 	"os"
 	"time"
 
-	"github.com/ethereum/go-ethereum/ethclient"
-	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/spf13/cobra"
 
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
-	"github.com/vulcanize/vulcanizedb/pkg/geth"
-	"github.com/vulcanize/vulcanizedb/pkg/geth/client"
-	vRpc "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
-	"github.com/vulcanize/vulcanizedb/pkg/geth/node"
 	"github.com/vulcanize/vulcanizedb/pkg/history"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/price_feeds/pep"
@@ -68,33 +62,17 @@ func backFillPriceFeeds(blockchain core.BlockChain, headerRepository datastore.H
 func syncPriceFeeds() {
 	ticker := time.NewTicker(pollingInterval)
 	defer ticker.Stop()
-	rawRpcClient, err := rpc.Dial(ipc)
+	blockChain := getBlockChain()
-	if err != nil {
+	validateArgs(blockChain)
-		log.Fatal(err)
-	}
-	rpcClient := client.NewRpcClient(rawRpcClient, ipc)
-	ethClient := ethclient.NewClient(rawRpcClient)
-	client := client.NewEthClient(ethClient)
-	node := node.MakeNode(rpcClient)
-	transactionConverter := vRpc.NewRpcTransactionConverter(client)
-	blockChain := geth.NewBlockChain(client, node, transactionConverter)
-
-	lastBlock := blockChain.LastBlock().Int64()
-	if lastBlock == 0 {
-		log.Fatal("geth initial: state sync not finished")
-	}
-	if startingBlockNumber > lastBlock {
-		log.Fatal("starting block number > current block number")
-	}
-
 	db := utils.LoadPostgres(databaseConfig, blockChain.Node())
+
+	transformers := []transformers.Transformer{
+		pep.NewPepTransformer(blockChain, &db, pepContractAddress),
+		pip.NewPipTransformer(blockChain, &db, pipContractAddress),
+		rep.NewRepTransformer(blockChain, &db, repContractAddress),
+	}
 	headerRepository := repositories.NewHeaderRepository(&db)
 	missingBlocksPopulated := make(chan int)
-	transformers := []transformers.Transformer{
-		pep.NewPepTransformer(blockChain, &db),
-		pip.NewPipTransformer(blockChain, &db),
-		rep.NewRepTransformer(blockChain, &db),
-	}
 	validator := history.NewHeaderValidator(blockChain, headerRepository, validationWindow, transformers)
 	go backFillPriceFeeds(blockChain, headerRepository, missingBlocksPopulated, startingBlockNumber, transformers)
 

environments/infura.toml

@@ -5,3 +5,6 @@ port = 5432
 
 [client]
 ipcPath = "https://mainnet.infura.io/J5Vd2fRtGsw0zZ0Ov3BL"
+pepContractAddress = "0x99041F808D598B782D5a3e498681C2452A31da08"
+pipContractAddress = "0x729D19f657BD0614b4985Cf1D82531c67569197B"
+repContractAddress = "0xF5f94b7F9De14D43112e713835BCef2d55b76c1C"

environments/private.toml

@@ -4,4 +4,7 @@ hostname = "localhost"
 port = 5432
 
 [client]
-ipcPath = "http://127.0.0.1:7545"
+ipcPath = "http://127.0.0.1:7545"
+pepContractAddress = "0x99041F808D598B782D5a3e498681C2452A31da08"
+pipContractAddress = "0x729D19f657BD0614b4985Cf1D82531c67569197B"
+repContractAddress = "0xF5f94b7F9De14D43112e713835BCef2d55b76c1C"

environments/public.toml.example

@@ -6,3 +6,6 @@ port = 5432
 [client]
 ipcPath = <local node's IPC filepath>
 levelDbPath = <local node's LevelDB chaindata filepath>
+pepContractAddress = <pep medianizer's address - see https://makerdao.com/feeds/ >
+pipContractAddress = <pip medianizer's address - see https://makerdao.com/feeds/ >
+repContractAddress = <rep medianizer's address - see https://makerdao.com/feeds/ >

pkg/fakes/mock_blockchain.go

@@ -18,11 +18,6 @@ type MockBlockChain struct {
 	fetchContractDataPassedMethodArg   interface{}
 	fetchContractDataPassedResult      interface{}
 	fetchContractDataPassedBlockNumber int64
-	getLogsPassedContract              core.Contract
-	getLogsPassedStartingBlockNumber   *big.Int
-	getLogsPassedEndingBlockNumber     *big.Int
-	getLogsReturnLogs                  []core.Log
-	getLogsReturnErr                   error
 	getBlockByNumberErr                error
 	logQuery                           ethereum.FilterQuery
 	logQueryErr                        error
@@ -57,14 +52,6 @@ func (chain *MockBlockChain) SetGetEthLogsWithCustomQueryReturnLogs(logs []types
 	chain.logQueryReturnLogs = logs
 }
 
-func (chain *MockBlockChain) SetGetLogsReturnErr(err error) {
-	chain.getLogsReturnErr = err
-}
-
-func (chain *MockBlockChain) SetGetLogsReturnLogs(logs []core.Log) {
-	chain.getLogsReturnLogs = logs
-}
-
 func (chain *MockBlockChain) FetchContractData(abiJSON, address, method string, methodArg, result interface{}, blockNumber int64) error {
 	chain.fetchContractDataPassedAbi = abiJSON
 	chain.fetchContractDataPassedAddress = address
@@ -89,10 +76,7 @@ func (chain *MockBlockChain) GetHeaderByNumber(blockNumber int64) (core.Header,
 }
 
 func (chain *MockBlockChain) GetLogs(contract core.Contract, startingBlockNumber, endingBlockNumber *big.Int) ([]core.Log, error) {
-	chain.getLogsPassedContract = contract
+	return []core.Log{}, nil
-	chain.getLogsPassedStartingBlockNumber = startingBlockNumber
-	chain.getLogsPassedEndingBlockNumber = endingBlockNumber
-	return chain.getLogsReturnLogs, chain.getLogsReturnErr
 }
 
 func (chain *MockBlockChain) CallContract(contractHash string, input []byte, blockNumber *big.Int) ([]byte, error) {

pkg/transformers/price_feeds/constants.go

@@ -10,12 +10,9 @@ var (
 	ErrNoMatchingLog = errors.New("no matching log")
 	Ether            = big.NewFloat(1e18)
 	PeekMethodName   = "peek"
-	PepAddress       = "0x99041F808D598B782D5a3e498681C2452A31da08"
 	PepLogTopic0     = "0x296ba4ca62c6c21c95e828080cb8aec7481b71390585605300a8a76f9e95b527"
-	PipAddress       = "0x729D19f657BD0614b4985Cf1D82531c67569197B"
 	PipLogTopic0     = "0x1817835800000000000000000000000000000000000000000000000000000000"
 	PipMedianizerABI = `[{"constant":false,"inputs":[{"name":"owner_","type":"address"}],"name":"setOwner","outputs":[],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"","type":"bytes32"}],"name":"poke","outputs":[],"payable":false,"type":"function"},{"constant":false,"inputs":[],"name":"poke","outputs":[],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"compute","outputs":[{"name":"","type":"bytes32"},{"name":"","type":"bool"}],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"wat","type":"address"}],"name":"set","outputs":[],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"wat","type":"address"}],"name":"unset","outputs":[],"payable":false,"type":"function"},{"constant":true,"inputs":[{"name":"","type":"address"}],"name":"indexes","outputs":[{"name":"","type":"bytes12"}],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"next","outputs":[{"name":"","type":"bytes12"}],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"read","outputs":[{"name":"","type":"bytes32"}],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"peek","outputs":[{"name":"","type":"bytes32"},{"name":"","type":"bool"}],"payable":false,"type":"function"},{"constant":true,"inputs":[{"name":"","type":"bytes12"}],"name":"values","outputs":[{"name":"","type":"address"}],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"min_","type":"uint96"}],"name":"setMin","outputs":[],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"authority_","type":"address"}],"name":"setAuthority","outputs":[],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"owner","outputs":[{"name":"","type":"address"}],"payable":false,"type":"function"},{"constant":false,"inputs":[],"name":"void","outputs":[],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"pos","type":"bytes12"},{"name":"wat","type":"address"}],"name":"set","outputs":[],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"authority","outputs":[{"name":"","type":"address"}],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"pos","type":"bytes12"}],"name":"unset","outputs":[],"payable":false,"type":"function"},{"constant":false,"inputs":[{"name":"next_","type":"bytes12"}],"name":"setNext","outputs":[],"payable":false,"type":"function"},{"constant":true,"inputs":[],"name":"min","outputs":[{"name":"","type":"uint96"}],"payable":false,"type":"function"},{"anonymous":true,"inputs":[{"indexed":true,"name":"sig","type":"bytes4"},{"indexed":true,"name":"guy","type":"address"},{"indexed":true,"name":"foo","type":"bytes32"},{"indexed":true,"name":"bar","type":"bytes32"},{"indexed":false,"name":"wad","type":"uint256"},{"indexed":false,"name":"fax","type":"bytes"}],"name":"LogNote","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"authority","type":"address"}],"name":"LogSetAuthority","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"owner","type":"address"}],"name":"LogSetOwner","type":"event"}]`
 	Ray              = big.NewFloat(1e27)
-	RepAddress       = "0xF5f94b7F9De14D43112e713835BCef2d55b76c1C"
 	RepLogTopic0     = "0x296ba4ca62c6c21c95e828080cb8aec7481b71390585605300a8a76f9e95b527"
 )

pkg/transformers/price_feeds/pep/fetcher.go

@@ -16,12 +16,14 @@ type IPepFetcher interface {
 }
 
 type PepFetcher struct {
-	blockChain core.BlockChain
+	blockChain      core.BlockChain
+	contractAddress string
 }
 
-func NewPepFetcher(chain core.BlockChain) PepFetcher {
+func NewPepFetcher(chain core.BlockChain, contractAddress string) PepFetcher {
 	return PepFetcher{
-		blockChain: chain,
+		blockChain:      chain,
+		contractAddress: contractAddress,
 	}
 }
 
@@ -30,7 +32,7 @@ func (fetcher PepFetcher) FetchPepValue(header core.Header) (string, error) {
 	query := ethereum.FilterQuery{
 		FromBlock: blockNumber,
 		ToBlock:   blockNumber,
-		Addresses: []common.Address{common.HexToAddress(price_feeds.PepAddress)},
+		Addresses: []common.Address{common.HexToAddress(fetcher.contractAddress)},
 		Topics:    [][]common.Hash{{common.HexToHash(price_feeds.PepLogTopic0)}},
 	}
 	logs, err := fetcher.blockChain.GetEthLogsWithCustomQuery(query)

pkg/transformers/price_feeds/pep/fetcher_test.go

@@ -18,8 +18,9 @@ var _ = Describe("Pep fetcher", func() {
 	It("gets logs describing updated mkr/usd value", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
 		mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{}})
-		fetcher := pep.NewPepFetcher(mockBlockChain)
+		contractAddress := "pep-contract-address"
-		blockNumber := int64(100)
+		fetcher := pep.NewPepFetcher(mockBlockChain, contractAddress)
+		blockNumber := int64(12345)
 		header := core.Header{
 			BlockNumber: blockNumber,
 			Hash:        "",
@@ -32,7 +33,7 @@ var _ = Describe("Pep fetcher", func() {
 		expectedQuery := ethereum.FilterQuery{
 			FromBlock: big.NewInt(blockNumber),
 			ToBlock:   big.NewInt(blockNumber),
-			Addresses: []common.Address{common.HexToAddress(price_feeds.PepAddress)},
+			Addresses: []common.Address{common.HexToAddress(contractAddress)},
 			Topics:    [][]common.Hash{{common.HexToHash(price_feeds.PepLogTopic0)}},
 		}
 		mockBlockChain.AssertGetEthLogsWithCustomQueryCalledWith(expectedQuery)
@@ -42,7 +43,7 @@ var _ = Describe("Pep fetcher", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
 		mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{}})
 		mockBlockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
-		fetcher := pep.NewPepFetcher(mockBlockChain)
+		fetcher := pep.NewPepFetcher(mockBlockChain, "pep-contract-address")
 
 		_, err := fetcher.FetchPepValue(core.Header{})
 
@@ -52,7 +53,7 @@ var _ = Describe("Pep fetcher", func() {
 
 	It("returns no matching logs error if no logs returned", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
-		fetcher := pep.NewPepFetcher(mockBlockChain)
+		fetcher := pep.NewPepFetcher(mockBlockChain, "pep-contract-address")
 
 		_, err := fetcher.FetchPepValue(core.Header{})
 
@@ -63,7 +64,7 @@ var _ = Describe("Pep fetcher", func() {
 	It("returns error if more than one matching logs returned", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
 		mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{}, {}})
-		fetcher := pep.NewPepFetcher(mockBlockChain)
+		fetcher := pep.NewPepFetcher(mockBlockChain, "pep-contract-address")
 
 		_, err := fetcher.FetchPepValue(core.Header{})
 

pkg/transformers/price_feeds/pep/transformer.go

@@ -11,8 +11,8 @@ type PepTransformer struct {
 	repository IPepRepository
 }
 
-func NewPepTransformer(chain core.BlockChain, db *postgres.DB) PepTransformer {
+func NewPepTransformer(chain core.BlockChain, db *postgres.DB, contractAddress string) PepTransformer {
-	fetcher := NewPepFetcher(chain)
+	fetcher := NewPepFetcher(chain, contractAddress)
 	repository := NewPepRepository(db)
 	return PepTransformer{
 		fetcher:    fetcher,

pkg/transformers/price_feeds/pep/transformer_test.go

@@ -17,7 +17,7 @@ var _ = Describe("Pep transformer", func() {
 	It("returns nil if no logs found", func() {
 		chain := fakes.NewMockBlockChain()
 		db := test_config.NewTestDB(core.Node{})
-		transformer := pep.NewPepTransformer(chain, db)
+		transformer := pep.NewPepTransformer(chain, db, "pep-contract-address")
 
 		err := transformer.Execute(core.Header{}, 123)
 
@@ -33,7 +33,7 @@ var _ = Describe("Pep transformer", func() {
 		header := core.Header{BlockNumber: 12345}
 		headerID, err := headerRepository.CreateOrUpdateHeader(header)
 		Expect(err).NotTo(HaveOccurred())
-		transformer := pep.NewPepTransformer(chain, db)
+		transformer := pep.NewPepTransformer(chain, db, "pep-contract-address")
 
 		err = transformer.Execute(header, headerID)
 

pkg/transformers/price_feeds/pip/fetcher.go

@@ -14,12 +14,14 @@ type IPipFetcher interface {
 }
 
 type PipFetcher struct {
-	blockChain core.BlockChain
+	blockChain      core.BlockChain
+	contractAddress string
 }
 
-func NewPipFetcher(chain core.BlockChain) PipFetcher {
+func NewPipFetcher(chain core.BlockChain, contractAddress string) PipFetcher {
 	return PipFetcher{
-		blockChain: chain,
+		blockChain:      chain,
+		contractAddress: contractAddress,
 	}
 }
 
@@ -28,7 +30,7 @@ func (fetcher PipFetcher) FetchPipValue(header core.Header) (string, error) {
 	query := ethereum.FilterQuery{
 		FromBlock: blockNumber,
 		ToBlock:   blockNumber,
-		Addresses: []common.Address{common.HexToAddress(price_feeds.PipAddress)},
+		Addresses: []common.Address{common.HexToAddress(fetcher.contractAddress)},
 		Topics:    [][]common.Hash{{common.HexToHash(price_feeds.PipLogTopic0)}},
 	}
 	logs, err := fetcher.blockChain.GetEthLogsWithCustomQuery(query)
@@ -50,7 +52,7 @@ func (fetcher PipFetcher) getLogValue(logs []types.Log, err error) (string, erro
 		ret0,
 		ret1,
 	}
-	err = fetcher.blockChain.FetchContractData(price_feeds.PipMedianizerABI, price_feeds.PipAddress, price_feeds.PeekMethodName, nil, r, int64(logs[0].BlockNumber))
+	err = fetcher.blockChain.FetchContractData(price_feeds.PipMedianizerABI, fetcher.contractAddress, price_feeds.PeekMethodName, nil, r, int64(logs[0].BlockNumber))
 	if err != nil {
 		return "", err
 	}

pkg/transformers/price_feeds/pip/fetcher_test.go

@@ -14,7 +14,7 @@ var _ = Describe("Pip fetcher", func() {
 	It("returns error if fetching logs fails", func() {
 		chain := fakes.NewMockBlockChain()
 		chain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
-		fetcher := pip.NewPipFetcher(chain)
+		fetcher := pip.NewPipFetcher(chain, "pip-contract-address")
 
 		_, err := fetcher.FetchPipValue(core.Header{})
 
@@ -24,7 +24,7 @@ var _ = Describe("Pip fetcher", func() {
 
 	It("returns no matching logs error if no logs returned", func() {
 		chain := fakes.NewMockBlockChain()
-		fetcher := pip.NewPipFetcher(chain)
+		fetcher := pip.NewPipFetcher(chain, "pip-contract-address")
 
 		_, err := fetcher.FetchPipValue(core.Header{})
 
@@ -37,19 +37,20 @@ var _ = Describe("Pip fetcher", func() {
 			blockNumber := uint64(12345)
 			chain := fakes.NewMockBlockChain()
 			chain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{BlockNumber: blockNumber}})
-			fetcher := pip.NewPipFetcher(chain)
+			contractAddress := "pip-contract-address"
+			fetcher := pip.NewPipFetcher(chain, contractAddress)
 
 			_, err := fetcher.FetchPipValue(core.Header{})
 
 			Expect(err).NotTo(HaveOccurred())
-			chain.AssertFetchContractDataCalledWith(price_feeds.PipMedianizerABI, price_feeds.PipAddress, price_feeds.PeekMethodName, nil, &[]interface{}{[32]byte{}, false}, int64(blockNumber))
+			chain.AssertFetchContractDataCalledWith(price_feeds.PipMedianizerABI, contractAddress, price_feeds.PeekMethodName, nil, &[]interface{}{[32]byte{}, false}, int64(blockNumber))
 		})
 
 		It("returns error if contract call fails", func() {
 			chain := fakes.NewMockBlockChain()
 			chain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{BlockNumber: uint64(12345)}})
 			chain.SetFetchContractDataErr(fakes.FakeError)
-			fetcher := pip.NewPipFetcher(chain)
+			fetcher := pip.NewPipFetcher(chain, "pip-contract-address")
 
 			_, err := fetcher.FetchPipValue(core.Header{})
 

pkg/transformers/price_feeds/pip/transformer.go

@@ -11,8 +11,8 @@ type PipTransformer struct {
 	repository IPipRepository
 }
 
-func NewPipTransformer(chain core.BlockChain, db *postgres.DB) PipTransformer {
+func NewPipTransformer(chain core.BlockChain, db *postgres.DB, contractAddress string) PipTransformer {
-	fetcher := NewPipFetcher(chain)
+	fetcher := NewPipFetcher(chain, contractAddress)
 	repository := NewPipRepository(db)
 	return PipTransformer{
 		fetcher:    fetcher,

pkg/transformers/price_feeds/pip/transformer_test.go

@@ -16,7 +16,7 @@ var _ = Describe("Pip transformer", func() {
 	It("returns nil if no logs found", func() {
 		chain := fakes.NewMockBlockChain()
 		db := test_config.NewTestDB(core.Node{})
-		transformer := pip.NewPipTransformer(chain, db)
+		transformer := pip.NewPipTransformer(chain, db, "pip-contract-address")
 
 		err := transformer.Execute(core.Header{}, 123)
 
@@ -32,7 +32,7 @@ var _ = Describe("Pip transformer", func() {
 		header := core.Header{BlockNumber: 12345}
 		headerID, err := headerRepository.CreateOrUpdateHeader(header)
 		Expect(err).NotTo(HaveOccurred())
-		transformer := pip.NewPipTransformer(chain, db)
+		transformer := pip.NewPipTransformer(chain, db, "pip-contract-address")
 
 		err = transformer.Execute(header, headerID)
 

pkg/transformers/price_feeds/rep/fetcher.go

@@ -1,6 +1,9 @@
 package rep
 
 import (
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/price_feeds"
 	"math/big"
@@ -11,22 +14,30 @@ type IRepFetcher interface {
 }
 
 type RepFetcher struct {
-	chain core.BlockChain
+	chain           core.BlockChain
+	contractAddress string
 }
 
-func NewRepFetcher(chain core.BlockChain) RepFetcher {
+func NewRepFetcher(chain core.BlockChain, contractAddress string) RepFetcher {
 	return RepFetcher{
-		chain: chain,
+		chain:           chain,
+		contractAddress: contractAddress,
 	}
 }
 
 func (fetcher RepFetcher) FetchRepValue(header core.Header) (string, error) {
 	blockNumber := big.NewInt(header.BlockNumber)
-	logs, err := fetcher.chain.GetLogs(price_feeds.RepAddress, price_feeds.RepLogTopic0, blockNumber, blockNumber)
+	query := ethereum.FilterQuery{
+		FromBlock: blockNumber,
+		ToBlock:   blockNumber,
+		Addresses: []common.Address{common.HexToAddress(fetcher.contractAddress)},
+		Topics:    [][]common.Hash{{common.HexToHash(price_feeds.RepLogTopic0)}},
+	}
+	logs, err := fetcher.chain.GetEthLogsWithCustomQuery(query)
 	return fetcher.getLogValue(logs, err)
 }
 
-func (fetcher RepFetcher) getLogValue(logs []core.Log, err error) (string, error) {
+func (fetcher RepFetcher) getLogValue(logs []types.Log, err error) (string, error) {
 	if err != nil {
 		return "", err
 	}
@@ -36,5 +47,5 @@ func (fetcher RepFetcher) getLogValue(logs []core.Log, err error) (string, error
 	if len(logs) > 1 {
 		return "", price_feeds.ErrMultipleLogs
 	}
-	return logs[0].Data, nil
+	return string(logs[0].Data), nil
 }

pkg/transformers/price_feeds/rep/fetcher_test.go

@@ -1,6 +1,9 @@
 package rep_test
 
 import (
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
@@ -13,10 +16,12 @@ import (
 var _ = Describe("Rep fetcher", func() {
 	It("gets logs describing updated rep/usd value", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
-		mockBlockChain.SetGetLogsReturnLogs([]core.Log{{}})
+		mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{}})
-		fetcher := rep.NewRepFetcher(mockBlockChain)
+		contractAddress := "rep-contract-address"
+		fetcher := rep.NewRepFetcher(mockBlockChain, contractAddress)
+		blockNumber := int64(100)
 		header := core.Header{
-			BlockNumber: 100,
+			BlockNumber: blockNumber,
 			Hash:        "",
 			Raw:         nil,
 		}
@@ -24,14 +29,19 @@ var _ = Describe("Rep fetcher", func() {
 		_, err := fetcher.FetchRepValue(header)
 
 		Expect(err).NotTo(HaveOccurred())
-		mockBlockChain.AssertGetLogsCalledWith(price_feeds.RepAddress, price_feeds.RepLogTopic0, big.NewInt(header.BlockNumber), big.NewInt(header.BlockNumber))
+		expectedQuery := ethereum.FilterQuery{
+			FromBlock: big.NewInt(blockNumber),
+			ToBlock:   big.NewInt(blockNumber),
+			Addresses: []common.Address{common.HexToAddress(contractAddress)},
+			Topics:    [][]common.Hash{{common.HexToHash(price_feeds.RepLogTopic0)}},
+		}
+		mockBlockChain.AssertGetEthLogsWithCustomQueryCalledWith(expectedQuery)
 	})
 
 	It("returns error if getting logs fails", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
-		mockBlockChain.SetGetLogsReturnLogs([]core.Log{{}})
+		mockBlockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
-		mockBlockChain.SetGetLogsReturnErr(fakes.FakeError)
+		fetcher := rep.NewRepFetcher(mockBlockChain, "rep-contract-address")
-		fetcher := rep.NewRepFetcher(mockBlockChain)
 
 		_, err := fetcher.FetchRepValue(core.Header{})
 
@@ -41,7 +51,7 @@ var _ = Describe("Rep fetcher", func() {
 
 	It("returns no matching logs error if no logs returned", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
-		fetcher := rep.NewRepFetcher(mockBlockChain)
+		fetcher := rep.NewRepFetcher(mockBlockChain, "rep-contract-address")
 
 		_, err := fetcher.FetchRepValue(core.Header{})
 
@@ -51,8 +61,8 @@ var _ = Describe("Rep fetcher", func() {
 
 	It("returns error if more than one matching logs returned", func() {
 		mockBlockChain := fakes.NewMockBlockChain()
-		mockBlockChain.SetGetLogsReturnLogs([]core.Log{{}, {}})
+		mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{}, {}})
-		fetcher := rep.NewRepFetcher(mockBlockChain)
+		fetcher := rep.NewRepFetcher(mockBlockChain, "rep-contract-address")
 
 		_, err := fetcher.FetchRepValue(core.Header{})
 

pkg/transformers/price_feeds/rep/repository_test.go

@@ -27,6 +27,7 @@ var _ = Describe("Rep repository", func() {
 
 	It("creates a rep when matching header exists", func() {
 		db := test_config.NewTestDB(core.Node{})
+		test_config.CleanTestDB(db)
 		repository := rep.NewRepRepository(db)
 		header := core.Header{BlockNumber: 12345}
 		headerRepository := repositories.NewHeaderRepository(db)

pkg/transformers/price_feeds/rep/transformer.go

@@ -11,8 +11,8 @@ type RepTransformer struct {
 	repository IRepRepository
 }
 
-func NewRepTransformer(chain core.BlockChain, db *postgres.DB) RepTransformer {
+func NewRepTransformer(chain core.BlockChain, db *postgres.DB, contractAddress string) RepTransformer {
-	fetcher := NewRepFetcher(chain)
+	fetcher := NewRepFetcher(chain, contractAddress)
 	repository := NewRepRepository(db)
 	return RepTransformer{
 		fetcher:    fetcher,

pkg/transformers/price_feeds/rep/transformer_test.go

@@ -1,10 +1,10 @@
 package rep_test
 
 import (
-	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
 	"github.com/vulcanize/vulcanizedb/pkg/fakes"
@@ -17,7 +17,7 @@ var _ = Describe("Rep transformer", func() {
 	It("returns nil if no logs found", func() {
 		chain := fakes.NewMockBlockChain()
 		db := test_config.NewTestDB(core.Node{})
-		transformer := rep.NewRepTransformer(chain, db)
+		transformer := rep.NewRepTransformer(chain, db, "rep-contract-address")
 
 		err := transformer.Execute(core.Header{}, 123)
 
@@ -26,13 +26,14 @@ var _ = Describe("Rep transformer", func() {
 
 	It("creates rep row for found log", func() {
 		chain := fakes.NewMockBlockChain()
-		chain.SetGetLogsReturnLogs([]core.Log{{Data: common.ToHex([]byte{1, 2, 3, 4, 5})}})
+		chain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{Data: []byte{1, 2, 3, 4, 5}}})
 		db := test_config.NewTestDB(core.Node{})
+		test_config.CleanTestDB(db)
 		headerRepository := repositories.NewHeaderRepository(db)
 		header := core.Header{BlockNumber: 12345}
 		headerID, err := headerRepository.CreateOrUpdateHeader(header)
 		Expect(err).NotTo(HaveOccurred())
-		transformer := rep.NewRepTransformer(chain, db)
+		transformer := rep.NewRepTransformer(chain, db, "rep-contract-address")
 
 		err = transformer.Execute(header, headerID)
 

test_config/test_config.go

@@ -79,6 +79,7 @@ func CleanTestDB(db *postgres.DB) {
 	db.MustExec("DELETE FROM logs")
 	db.MustExec("DELETE FROM maker.peps")
 	db.MustExec("DELETE FROM maker.pips")
+	db.MustExec("DELETE FROM maker.reps")
 	db.MustExec("DELETE FROM receipts")
 	db.MustExec("DELETE FROM transactions")
 	db.MustExec("DELETE FROM watched_contracts")