Add ERC20 token watcher example

- starting with the totalSupply function
- sets contract config on transformer by passing it into the transformer
initializer
- handles block records with the same number for different nodes for both creating token_supply records, and finding missing blocks

examples/erc20_watcher/every_block/every_block_suite_test.go

@@ -0,0 +1,33 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"io/ioutil"
+	"log"
+)
+
+func TestEveryBlock(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "EveryBlock Suite")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

examples/erc20_watcher/every_block/fetcher.go

@@ -0,0 +1,70 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block
+
+import (
+	"fmt"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"log"
+	"math/big"
+)
+
+type ERC20FetcherInterface interface {
+	FetchSupplyOf(contractAbi string, contractAddress string, blockNumber int64) (big.Int, error)
+	GetBlockchain() core.Blockchain
+}
+
+func NewFetcher(blockchain core.Blockchain) Fetcher {
+	return Fetcher{
+		Blockchain: blockchain,
+	}
+}
+
+type Fetcher struct {
+	Blockchain      core.Blockchain
+	ContractAbi     string
+	ContractAddress string
+}
+
+type fetcherError struct {
+	err         string
+	fetchMethod string
+}
+
+func (fe *fetcherError) Error() string {
+	return fmt.Sprintf("Error fetching %s: %s", fe.fetchMethod, fe.err)
+}
+
+func newFetcherError(err error, fetchMethod string) *fetcherError {
+	e := fetcherError{err.Error(), fetchMethod}
+	log.Println(e.Error())
+	return &e
+}
+
+func (f Fetcher) FetchSupplyOf(contractAbi string, contractAddress string, blockNumber int64) (big.Int, error) {
+	method := "totalSupply"
+	var result = new(big.Int)
+	err := f.Blockchain.FetchContractData(contractAbi, contractAddress, method, nil, &result, blockNumber)
+
+	if err != nil {
+		return *result, newFetcherError(err, method)
+	}
+
+	return *result, nil
+}
+
+func (f Fetcher) GetBlockchain() core.Blockchain {
+	return f.Blockchain
+}

examples/erc20_watcher/every_block/fetcher_test.go

@@ -0,0 +1,75 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/examples/constants"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher/every_block"
+	"github.com/vulcanize/vulcanizedb/examples/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/geth"
+	"math/big"
+)
+
+var _ = Describe("ERC20 Fetcher", func() {
+	blockNumber := int64(5502914)
+
+	infuraIPC := "https://mainnet.infura.io/J5Vd2fRtGsw0zZ0Ov3BL"
+	realBlockchain := geth.NewBlockchain(infuraIPC)
+	realFetcher := every_block.NewFetcher(realBlockchain)
+
+	fakeBlockchain := &mocks.Blockchain{}
+	testFetcher := every_block.NewFetcher(fakeBlockchain)
+	testAbi := "testAbi"
+	testContractAddress := "testContractAddress"
+
+	errorBlockchain := &mocks.FailureBlockchain{}
+	errorFetcher := every_block.NewFetcher(errorBlockchain)
+
+	Describe("FetchSupplyOf", func() {
+		It("fetches data from the blockchain with the correct arguments", func() {
+			_, err := testFetcher.FetchSupplyOf(testAbi, testContractAddress, blockNumber)
+
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeBlockchain.FetchedAbi).To(Equal(testAbi))
+			Expect(fakeBlockchain.FetchedContractAddress).To(Equal(testContractAddress))
+			Expect(fakeBlockchain.FetchedMethod).To(Equal("totalSupply"))
+			Expect(fakeBlockchain.FetchedMethodArg).To(BeNil())
+			expectedResult := big.Int{}
+			expected := &expectedResult
+			Expect(fakeBlockchain.FetchedResult).To(Equal(&expected))
+			Expect(fakeBlockchain.FetchedBlockNumber).To(Equal(blockNumber))
+		})
+
+		It("fetches a token's total supply at the given block height", func() {
+			result, err := realFetcher.FetchSupplyOf(constants.DaiAbiString, constants.DaiContractAddress, blockNumber)
+
+			Expect(err).NotTo(HaveOccurred())
+			expectedResult := big.Int{}
+			expectedResult.SetString("27647235749155415536952630", 10)
+			Expect(result).To(Equal(expectedResult))
+		})
+
+		It("returns an error if the call to the blockchain fails", func() {
+			result, err := errorFetcher.FetchSupplyOf("", "", 0)
+
+			Expect(result.String()).To(Equal("0"))
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("totalSupply"))
+			Expect(err.Error()).To(ContainSubstring(mocks.TestError.Error()))
+		})
+	})
+})

examples/erc20_watcher/every_block/integration_test.go

@@ -0,0 +1,77 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/examples/constants"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher/every_block"
+	"github.com/vulcanize/vulcanizedb/examples/mocks"
+	"github.com/vulcanize/vulcanizedb/examples/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+	"math/big"
+	"strconv"
+)
+
+func setLastBlockOnChain(blockchain *mocks.Blockchain, blockNumber int64) {
+	blockNumberString := strconv.FormatInt(blockNumber, 10)
+	lastBlockOnChain := big.Int{}
+	lastBlockOnChain.SetString(blockNumberString, 10)
+	blockchain.SetLastBlock(&lastBlockOnChain)
+}
+
+var _ = Describe("Everyblock transformers", func() {
+	var db *postgres.DB
+	var blockchain mocks.Blockchain
+	var blockNumber int64
+	var blockId int64
+	var err error
+
+	BeforeEach(func() {
+		blockNumber = erc20_watcher.DaiConfig.FirstBlock
+		lastBlockNumber := blockNumber + 1
+		db = test_helpers.CreateNewDatabase()
+		setLastBlockOnChain(&blockchain, lastBlockNumber)
+
+		blockRepository := repositories.NewBlockRepository(db)
+
+		blockId, err = blockRepository.CreateOrUpdateBlock(core.Block{Number: blockNumber})
+		Expect(err).NotTo(HaveOccurred())
+		_, err = blockRepository.CreateOrUpdateBlock(core.Block{Number: lastBlockNumber})
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("creates a token_supply record for each block in the given range", func() {
+		initializer := every_block.TokenSupplyTransformerInitializer{erc20_watcher.DaiConfig}
+		transformer := initializer.NewTokenSupplyTransformer(db, &blockchain)
+		transformer.Execute()
+
+		var tokenSupplyCount int
+		err := db.QueryRow(`SELECT COUNT(*) FROM token_supply where block_id = $1`, blockId).Scan(&tokenSupplyCount)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(tokenSupplyCount).To(Equal(1))
+
+		var tokenSupply test_helpers.TokenSupplyDBRow
+		err = db.Get(&tokenSupply, `SELECT * from token_supply where block_id = $1`, blockId)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(tokenSupply.BlockID).To(Equal(blockId))
+		Expect(tokenSupply.TokenAddress).To(Equal(constants.DaiContractAddress))
+		Expect(tokenSupply.Supply).To(Equal(int64(0)))
+	})
+})

examples/erc20_watcher/every_block/model.go

@@ -0,0 +1,21 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block
+
+type TokenSupply struct {
+	Value        string
+	TokenAddress string
+	BlockNumber  int64
+}

examples/erc20_watcher/every_block/repository.go

@@ -0,0 +1,91 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block
+
+import (
+	"fmt"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"log"
+)
+
+type ERC20RepositoryInterface interface {
+	Create(supply TokenSupply) error
+	MissingBlocks(startingBlock int64, highestBlock int64) ([]int64, error)
+}
+
+type TokenSupplyRepository struct {
+	*postgres.DB
+}
+
+type repositoryError struct {
+	err         string
+	msg         string
+	blockNumber int64
+}
+
+func (re *repositoryError) Error() string {
+	return fmt.Sprintf(re.msg, re.blockNumber, re.err)
+}
+
+func newRepositoryError(err error, msg string, blockNumber int64) error {
+	e := repositoryError{err.Error(), msg, blockNumber}
+	log.Println(e.Error())
+	return &e
+}
+
+const (
+	GetBlockError          = "Error fetching block number %d: %s"
+	InsertTokenSupplyError = "Error inserting token_supply for block number %d: %s"
+	MissingBlockError      = "Error finding missing token_supply records starting at block %d: %s"
+)
+
+func (tsp *TokenSupplyRepository) Create(supply TokenSupply) error {
+	var blockId int
+	err := tsp.DB.Get(&blockId, `SELECT id FROM blocks WHERE number = $1 AND eth_node_id = $2`, supply.BlockNumber, tsp.NodeID)
+	if err != nil {
+		return newRepositoryError(err, GetBlockError, supply.BlockNumber)
+	}
+
+	_, err = tsp.DB.Exec(
+		`INSERT INTO token_supply (supply, token_address, block_id)
+                VALUES($1, $2, $3)`,
+		supply.Value, supply.TokenAddress, blockId)
+	if err != nil {
+		return newRepositoryError(err, InsertTokenSupplyError, supply.BlockNumber)
+	}
+	return nil
+}
+
+func (tsp *TokenSupplyRepository) MissingBlocks(startingBlock int64, highestBlock int64) ([]int64, error) {
+	blockNumbers := make([]int64, 0)
+
+	err := tsp.DB.Select(
+		&blockNumbers,
+		`SELECT number FROM BLOCKS
+               LEFT JOIN token_supply ON blocks.id = block_id
+               WHERE block_id ISNULL
+               AND eth_node_id = $1
+               AND number >= $2
+               AND number <= $3
+               LIMIT 20`,
+		tsp.NodeID,
+		startingBlock,
+		highestBlock,
+	)
+	if err != nil {
+		return []int64{}, newRepositoryError(err, MissingBlockError, startingBlock)
+	}
+	return blockNumbers, err
+}

examples/erc20_watcher/every_block/repository_test.go

@@ -0,0 +1,207 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher/every_block"
+	"github.com/vulcanize/vulcanizedb/examples/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+	"github.com/vulcanize/vulcanizedb/test_config"
+	"math/rand"
+)
+
+var _ = Describe("ERC20 Token Repository", func() {
+	var db *postgres.DB
+	var blockId int64
+	var blockNumber int64
+	var repository every_block.TokenSupplyRepository
+	var blockRepository repositories.BlockRepository
+	testAddress := "abc"
+
+	BeforeEach(func() {
+		db = test_helpers.CreateNewDatabase()
+		repository = every_block.TokenSupplyRepository{DB: db}
+		_, err := db.Query(`DELETE FROM token_supply`)
+		Expect(err).NotTo(HaveOccurred())
+
+		blockRepository = *repositories.NewBlockRepository(db)
+		blockNumber = rand.Int63()
+		blockId = test_helpers.CreateBlock(blockNumber, blockRepository)
+	})
+
+	Describe("Create", func() {
+		It("creates a token supply record", func() {
+			supply := supplyModel(blockNumber, testAddress, "100")
+			err := repository.Create(supply)
+			Expect(err).NotTo(HaveOccurred())
+
+			dbResult := test_helpers.TokenSupplyDBRow{}
+			expectedTokenSupply := test_helpers.TokenSupplyDBRow{
+				Supply:       int64(100),
+				BlockID:      blockId,
+				TokenAddress: testAddress,
+			}
+
+			var count int
+			err = repository.DB.QueryRowx(`SELECT count(*) FROM token_supply`).Scan(&count)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(count).To(Equal(1))
+
+			err = repository.DB.QueryRowx(`SELECT * FROM token_supply`).StructScan(&dbResult)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(dbResult.Supply).To(Equal(expectedTokenSupply.Supply))
+			Expect(dbResult.BlockID).To(Equal(expectedTokenSupply.BlockID))
+			Expect(dbResult.TokenAddress).To(Equal(expectedTokenSupply.TokenAddress))
+		})
+
+		It("returns an error if fetching the block's id from the database fails", func() {
+			errorSupply := supplyModel(-1, "", "")
+			err := repository.Create(errorSupply)
+
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("sql"))
+			Expect(err.Error()).To(ContainSubstring("block number -1"))
+		})
+
+		It("returns an error if inserting the token_supply fails", func() {
+			errorSupply := supplyModel(blockNumber, "", "")
+			err := repository.Create(errorSupply)
+
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("pq"))
+			Expect(err.Error()).To(ContainSubstring("token_supply for block number"))
+		})
+	})
+
+	Describe("When there are multiple nodes", func() {
+		var node2DB *postgres.DB
+		var node2BlockRepo *repositories.BlockRepository
+		var node2BlockId int64
+		var node2TokenSupplyRepo every_block.TokenSupplyRepository
+		var tokenSupply every_block.TokenSupply
+
+		BeforeEach(func() {
+			node2DB = createDbForAnotherNode()
+
+			//create another block with the same number on node2
+			node2BlockRepo = repositories.NewBlockRepository(node2DB)
+			node2BlockId = test_helpers.CreateBlock(blockNumber, *node2BlockRepo)
+
+			tokenSupply = supplyModel(blockNumber, "abc", "100")
+			node2TokenSupplyRepo = every_block.TokenSupplyRepository{DB: node2DB}
+		})
+
+		It("only creates token_supply records for the current node (node2)", func() {
+			err := node2TokenSupplyRepo.Create(tokenSupply)
+			Expect(err).NotTo(HaveOccurred())
+
+			var tokenSupplies []test_helpers.TokenSupplyDBRow
+			err = node2TokenSupplyRepo.DB.Select(&tokenSupplies, `SELECT * FROM token_supply`)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(tokenSupplies)).To(Equal(1))
+			Expect(tokenSupplies[0].BlockID).To(Equal(node2BlockId))
+		})
+
+		It("only includes missing block numbers for the current node", func() {
+			//create token_supply on original node
+			err := repository.Create(tokenSupply)
+			Expect(err).NotTo(HaveOccurred())
+
+			originalNodeMissingBlocks, err := repository.MissingBlocks(blockNumber, blockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(originalNodeMissingBlocks)).To(Equal(0))
+
+			node2MissingBlocks, err := node2TokenSupplyRepo.MissingBlocks(blockNumber, blockNumber)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(node2MissingBlocks)).To(Equal(1))
+		})
+	})
+
+	Describe("MissingBlocks", func() {
+		It("returns the block numbers for which an associated TokenSupply record hasn't been created", func() {
+			createTokenSupplyFor(repository, blockNumber)
+
+			newBlockNumber := blockNumber + 1
+			test_helpers.CreateBlock(newBlockNumber, blockRepository)
+			blocks, err := repository.MissingBlocks(blockNumber, newBlockNumber)
+
+			Expect(blocks).To(ConsistOf(newBlockNumber))
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("only returns blocks within the given range", func() {
+			newBlockNumber := blockNumber + 1
+			test_helpers.CreateBlock(newBlockNumber, blockRepository)
+			blocks, err := repository.MissingBlocks(blockNumber, blockNumber)
+
+			Expect(blocks).NotTo(ConsistOf(newBlockNumber))
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("does not return numbers that already have an associated TokenSupply record", func() {
+			createTokenSupplyFor(repository, blockNumber)
+			blocks, err := repository.MissingBlocks(blockNumber, blockNumber)
+
+			Expect(blocks).To(BeEmpty())
+			Expect(err).NotTo(HaveOccurred())
+		})
+	})
+
+	It("deletes the token supply record when the associated block is deleted", func() {
+		err := repository.Create(every_block.TokenSupply{BlockNumber: blockNumber, Value: "0"})
+		Expect(err).NotTo(HaveOccurred())
+
+		var count int
+		err = repository.DB.QueryRowx(`SELECT count(*) FROM token_supply`).Scan(&count)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(count).To(Equal(1))
+
+		_, err = db.Query(`DELETE FROM blocks`)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = repository.DB.QueryRowx(`SELECT count(*) FROM token_supply`).Scan(&count)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(count).To(Equal(0))
+	})
+})
+
+func supplyModel(blockNumber int64, tokenAddress string, supplyValue string) every_block.TokenSupply {
+	return every_block.TokenSupply{
+		Value:        supplyValue,
+		TokenAddress: tokenAddress,
+		BlockNumber:  int64(blockNumber),
+	}
+}
+
+func createTokenSupplyFor(repository every_block.TokenSupplyRepository, blockNumber int64) {
+	err := repository.Create(every_block.TokenSupply{BlockNumber: blockNumber, Value: "0"})
+	Expect(err).NotTo(HaveOccurred())
+}
+
+func createDbForAnotherNode() *postgres.DB {
+	anotherNode := core.Node{
+		GenesisBlock: "GENESIS",
+		NetworkID:    1,
+		ID:           "testNodeId",
+		ClientName:   "Geth/v1.7.2-stable-1db4ecdc/darwin-amd64/go1.9",
+	}
+
+	return test_config.NewTestDBWithoutDeletingRecords(anotherNode)
+}

examples/erc20_watcher/every_block/transformer.go

@@ -0,0 +1,116 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block
+
+import (
+	"fmt"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher"
+	"github.com/vulcanize/vulcanizedb/libraries/shared"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"log"
+	"math/big"
+)
+
+type Transformer struct {
+	Fetcher    ERC20FetcherInterface
+	Repository ERC20RepositoryInterface
+	Config     erc20_watcher.ContractConfig
+}
+
+func (t *Transformer) SetConfiguration(config erc20_watcher.ContractConfig) {
+	t.Config = config
+}
+
+type TokenSupplyTransformerInitializer struct {
+	Config erc20_watcher.ContractConfig
+}
+
+func (i TokenSupplyTransformerInitializer) NewTokenSupplyTransformer(db *postgres.DB, blockchain core.Blockchain) shared.Transformer {
+	fetcher := NewFetcher(blockchain)
+	repository := TokenSupplyRepository{DB: db}
+	transformer := Transformer{
+		Fetcher:    &fetcher,
+		Repository: &repository,
+		Config:     i.Config,
+	}
+
+	return transformer
+}
+
+const (
+	FetchingBlocksError = "Error fetching missing blocks starting at block number %d: %s"
+	FetchingSupplyError = "Error fetching supply for block %d: %s"
+	CreateSupplyError   = "Error inserting token_supply for block %d: %s"
+)
+
+type transformerError struct {
+	err         string
+	blockNumber int64
+	msg         string
+}
+
+func (te *transformerError) Error() string {
+	return fmt.Sprintf(te.msg, te.blockNumber, te.err)
+}
+
+func newTransformerError(err error, blockNumber int64, msg string) error {
+	e := transformerError{err.Error(), blockNumber, msg}
+	log.Println(e.Error())
+	return &e
+}
+
+func (t Transformer) Execute() error {
+	var upperBoundBlock int64
+	blockchain := t.Fetcher.GetBlockchain()
+	lastBlock := blockchain.LastBlock().Int64()
+
+	if t.Config.LastBlock == -1 {
+		upperBoundBlock = lastBlock
+	} else {
+		upperBoundBlock = t.Config.LastBlock
+	}
+
+	blocks, err := t.Repository.MissingBlocks(t.Config.FirstBlock, upperBoundBlock)
+
+	if err != nil {
+		return newTransformerError(err, t.Config.FirstBlock, FetchingBlocksError)
+	}
+
+	log.Printf("Fetching totalSupply for %d blocks", len(blocks))
+	for _, blockNumber := range blocks {
+		totalSupply, err := t.Fetcher.FetchSupplyOf(t.Config.Abi, t.Config.Address, blockNumber)
+
+		if err != nil {
+			return newTransformerError(err, blockNumber, FetchingSupplyError)
+		}
+		model := createTokenSupplyModel(totalSupply, t.Config.Address, blockNumber)
+		err = t.Repository.Create(model)
+
+		if err != nil {
+			return newTransformerError(err, blockNumber, CreateSupplyError)
+		}
+	}
+
+	return nil
+}
+
+func createTokenSupplyModel(totalSupply big.Int, address string, blockNumber int64) TokenSupply {
+	return TokenSupply{
+		Value:        totalSupply.String(),
+		TokenAddress: address,
+		BlockNumber:  blockNumber,
+	}
+}

examples/erc20_watcher/every_block/transformer_test.go

@@ -0,0 +1,184 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/examples/constants"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher/every_block"
+	"github.com/vulcanize/vulcanizedb/examples/mocks"
+	"math/big"
+	"math/rand"
+	"strconv"
+)
+
+var testContractConfig = erc20_watcher.ContractConfig{
+	Address:    constants.DaiContractAddress,
+	Abi:        constants.DaiAbiString,
+	FirstBlock: int64(4752008),
+	LastBlock:  int64(5750050),
+	Name:       "Dai",
+}
+
+var config = testContractConfig
+
+var _ = Describe("Everyblock transformer", func() {
+	var fetcher mocks.Fetcher
+	var repository mocks.TotalSupplyRepository
+	var transformer every_block.Transformer
+	var blockchain mocks.Blockchain
+	var initialSupply = "27647235749155415536952630"
+	var initialSupplyPlusOne = "27647235749155415536952631"
+	var initialSupplyPlusTwo = "27647235749155415536952632"
+	var initialSupplyPlusThree = "27647235749155415536952633"
+	var defaultLastBlock = big.Int{}
+
+	BeforeEach(func() {
+		blockchain = mocks.Blockchain{}
+		blockchain.SetLastBlock(&defaultLastBlock)
+		fetcher = mocks.Fetcher{Blockchain: &blockchain}
+		fetcher.SetSupply(initialSupply)
+		repository = mocks.TotalSupplyRepository{}
+		repository.SetMissingBlocks([]int64{config.FirstBlock})
+		//setting the mock repository to return the first block as the missing blocks
+
+		transformer = every_block.Transformer{
+			Fetcher:    &fetcher,
+			Repository: &repository,
+		}
+		transformer.SetConfiguration(config)
+	})
+
+	It("fetches and persists the total supply of a token for a single block", func() {
+		err := transformer.Execute()
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(len(fetcher.FetchedBlocks)).To(Equal(1))
+		Expect(fetcher.FetchedBlocks).To(ConsistOf(config.FirstBlock))
+		Expect(fetcher.Abi).To(Equal(config.Abi))
+		Expect(fetcher.ContractAddress).To(Equal(config.Address))
+
+		Expect(repository.StartingBlock).To(Equal(config.FirstBlock))
+		Expect(repository.EndingBlock).To(Equal(config.LastBlock))
+		Expect(len(repository.TotalSuppliesCreated)).To(Equal(1))
+		expectedSupply := big.Int{}
+		expectedSupply.SetString(initialSupply, 10)
+		expectedSupply.Add(&expectedSupply, big.NewInt(1))
+
+		Expect(repository.TotalSuppliesCreated[0].Value).To(Equal(expectedSupply.String()))
+	})
+
+	It("retrieves the total supply for every missing block", func() {
+		missingBlocks := []int64{
+			config.FirstBlock,
+			config.FirstBlock + 1,
+			config.FirstBlock + 2,
+		}
+		repository.SetMissingBlocks(missingBlocks)
+		transformer.Execute()
+
+		Expect(len(fetcher.FetchedBlocks)).To(Equal(3))
+		Expect(fetcher.FetchedBlocks).To(ConsistOf(config.FirstBlock, config.FirstBlock+1, config.FirstBlock+2))
+		Expect(fetcher.Abi).To(Equal(config.Abi))
+		Expect(fetcher.ContractAddress).To(Equal(config.Address))
+
+		Expect(len(repository.TotalSuppliesCreated)).To(Equal(3))
+		Expect(repository.TotalSuppliesCreated[0].Value).To(Equal(initialSupplyPlusOne))
+		Expect(repository.TotalSuppliesCreated[1].Value).To(Equal(initialSupplyPlusTwo))
+		Expect(repository.TotalSuppliesCreated[2].Value).To(Equal(initialSupplyPlusThree))
+	})
+
+	It("uses the set contract configuration", func() {
+		repository.SetMissingBlocks([]int64{testContractConfig.FirstBlock})
+		transformer.SetConfiguration(testContractConfig)
+		err := transformer.Execute()
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(fetcher.FetchedBlocks).To(ConsistOf(testContractConfig.FirstBlock))
+		Expect(fetcher.Abi).To(Equal(testContractConfig.Abi))
+		Expect(fetcher.ContractAddress).To(Equal(testContractConfig.Address))
+
+		Expect(repository.StartingBlock).To(Equal(testContractConfig.FirstBlock))
+		Expect(repository.EndingBlock).To(Equal(testContractConfig.LastBlock))
+		Expect(len(repository.TotalSuppliesCreated)).To(Equal(1))
+		expectedTokenSupply := every_block.TokenSupply{
+			Value:        initialSupplyPlusOne,
+			TokenAddress: testContractConfig.Address,
+			BlockNumber:  testContractConfig.FirstBlock,
+		}
+		Expect(repository.TotalSuppliesCreated[0]).To(Equal(expectedTokenSupply))
+	})
+
+	It("uses the most recent block if the Config.LastBlock is -1", func() {
+		testContractConfig.LastBlock = -1
+		transformer.SetConfiguration(testContractConfig)
+
+		randomBlockNumber := rand.Int63()
+		numberToString := strconv.FormatInt(randomBlockNumber, 10)
+		mostRecentBlock := big.Int{}
+		mostRecentBlock.SetString(numberToString, 10)
+
+		blockchain.SetLastBlock(&mostRecentBlock)
+
+		err := transformer.Execute()
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(repository.EndingBlock).To(Equal(randomBlockNumber))
+	})
+
+	It("returns an error if the call to get missing blocks fails", func() {
+		failureRepository := mocks.FailureRepository{}
+		failureRepository.SetMissingBlocksFail(true)
+		transformer = every_block.Transformer{
+			Fetcher:    &fetcher,
+			Repository: &failureRepository,
+		}
+		err := transformer.Execute()
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("TestError"))
+		Expect(err.Error()).To(ContainSubstring("fetching missing blocks"))
+	})
+
+	It("returns an error if the call to the blockchain fails", func() {
+		failureBlockchain := mocks.FailureBlockchain{}
+		failureBlockchain.SetLastBlock(&defaultLastBlock)
+		fetcher := every_block.NewFetcher(failureBlockchain)
+		transformer = every_block.Transformer{
+			Fetcher:    &fetcher,
+			Repository: &repository,
+		}
+		err := transformer.Execute()
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("TestError"))
+		Expect(err.Error()).To(ContainSubstring("supply"))
+	})
+
+	It("returns an error if the call to save the token_supply fails", func() {
+		failureRepository := mocks.FailureRepository{}
+		failureRepository.SetMissingBlocks([]int64{config.FirstBlock})
+		failureRepository.SetCreateFail(true)
+
+		transformer = every_block.Transformer{
+			Fetcher:    &fetcher,
+			Repository: &failureRepository,
+		}
+		err := transformer.Execute()
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("TestError"))
+		Expect(err.Error()).To(ContainSubstring("supply"))
+	})
+})

examples/erc20_watcher/every_block/transformers.go

@@ -0,0 +1,28 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package every_block
+
+import (
+	"github.com/vulcanize/vulcanizedb/libraries/shared"
+	"github.com/vulcanize/vulcanizedb/examples/erc20_watcher"
+)
+
+func TransformerInitializers() []shared.TransformerInitializer {
+	config := erc20_watcher.DaiConfig
+	initializer := TokenSupplyTransformerInitializer{config}
+	return []shared.TransformerInitializer{
+		initializer.NewTokenSupplyTransformer,
+	}
+}