mv pkg/omni pkg/contract_watcher

pkg/contract_watcher/light/converter/converter.go

@@ -0,0 +1,213 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package converter
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math/big"
+	"strconv"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	gethTypes "github.com/ethereum/go-ethereum/core/types"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
+)
+
+type Converter interface {
+	Convert(logs []gethTypes.Log, event types.Event, headerID int64) ([]types.Log, error)
+	ConvertBatch(logs []gethTypes.Log, events map[string]types.Event, headerID int64) (map[string][]types.Log, error)
+	Update(info *contract.Contract)
+}
+
+type converter struct {
+	ContractInfo *contract.Contract
+}
+
+func NewConverter(info *contract.Contract) *converter {
+	return &converter{
+		ContractInfo: info,
+	}
+}
+
+func (c *converter) Update(info *contract.Contract) {
+	c.ContractInfo = info
+}
+
+// Convert the given watched event log into a types.Log for the given event
+func (c *converter) Convert(logs []gethTypes.Log, event types.Event, headerID int64) ([]types.Log, error) {
+	contract := bind.NewBoundContract(common.HexToAddress(c.ContractInfo.Address), c.ContractInfo.ParsedAbi, nil, nil, nil)
+	returnLogs := make([]types.Log, 0, len(logs))
+	for _, log := range logs {
+		values := make(map[string]interface{})
+		for _, field := range event.Fields {
+			var i interface{}
+			values[field.Name] = i
+		}
+
+		err := contract.UnpackLogIntoMap(values, event.Name, log)
+		if err != nil {
+			return nil, err
+		}
+
+		strValues := make(map[string]string, len(values))
+		seenAddrs := make([]interface{}, 0, len(values))
+		seenHashes := make([]interface{}, 0, len(values))
+		for fieldName, input := range values {
+			// Postgres cannot handle custom types, resolve everything to strings
+			switch input.(type) {
+			case *big.Int:
+				b := input.(*big.Int)
+				strValues[fieldName] = b.String()
+			case common.Address:
+				a := input.(common.Address)
+				strValues[fieldName] = a.String()
+				seenAddrs = append(seenAddrs, a)
+			case common.Hash:
+				h := input.(common.Hash)
+				strValues[fieldName] = h.String()
+				seenHashes = append(seenHashes, h)
+			case string:
+				strValues[fieldName] = input.(string)
+			case bool:
+				strValues[fieldName] = strconv.FormatBool(input.(bool))
+			case []byte:
+				b := input.([]byte)
+				strValues[fieldName] = hexutil.Encode(b)
+				if len(b) == 32 {
+					seenHashes = append(seenHashes, common.HexToHash(strValues[fieldName]))
+				}
+			case byte:
+				b := input.(byte)
+				strValues[fieldName] = string(b)
+			default:
+				return nil, errors.New(fmt.Sprintf("error: unhandled abi type %T", input))
+			}
+		}
+
+		// Only hold onto logs that pass our address filter, if any
+		if c.ContractInfo.PassesEventFilter(strValues) {
+			raw, err := json.Marshal(log)
+			if err != nil {
+				return nil, err
+			}
+
+			returnLogs = append(returnLogs, types.Log{
+				LogIndex:         log.Index,
+				Values:           strValues,
+				Raw:              raw,
+				TransactionIndex: log.TxIndex,
+				Id:               headerID,
+			})
+
+			// Cache emitted values if their caching is turned on
+			if c.ContractInfo.EmittedAddrs != nil {
+				c.ContractInfo.AddEmittedAddr(seenAddrs...)
+			}
+			if c.ContractInfo.EmittedHashes != nil {
+				c.ContractInfo.AddEmittedHash(seenHashes...)
+			}
+		}
+	}
+
+	return returnLogs, nil
+}
+
+// Convert the given watched event logs into types.Logs; returns a map of event names to a slice of their converted logs
+func (c *converter) ConvertBatch(logs []gethTypes.Log, events map[string]types.Event, headerID int64) (map[string][]types.Log, error) {
+	contract := bind.NewBoundContract(common.HexToAddress(c.ContractInfo.Address), c.ContractInfo.ParsedAbi, nil, nil, nil)
+	eventsToLogs := make(map[string][]types.Log)
+	for _, event := range events {
+		eventsToLogs[event.Name] = make([]types.Log, 0, len(logs))
+		// Iterate through all event logs
+		for _, log := range logs {
+			// If the log is of this event type, process it as such
+			if event.Sig() == log.Topics[0] {
+				values := make(map[string]interface{})
+				err := contract.UnpackLogIntoMap(values, event.Name, log)
+				if err != nil {
+					return nil, err
+				}
+				// Postgres cannot handle custom types, so we will resolve everything to strings
+				strValues := make(map[string]string, len(values))
+				// Keep track of addresses and hashes emitted from events
+				seenAddrs := make([]interface{}, 0, len(values))
+				seenHashes := make([]interface{}, 0, len(values))
+				for fieldName, input := range values {
+					switch input.(type) {
+					case *big.Int:
+						b := input.(*big.Int)
+						strValues[fieldName] = b.String()
+					case common.Address:
+						a := input.(common.Address)
+						strValues[fieldName] = a.String()
+						seenAddrs = append(seenAddrs, a)
+					case common.Hash:
+						h := input.(common.Hash)
+						strValues[fieldName] = h.String()
+						seenHashes = append(seenHashes, h)
+					case string:
+						strValues[fieldName] = input.(string)
+					case bool:
+						strValues[fieldName] = strconv.FormatBool(input.(bool))
+					case []byte:
+						b := input.([]byte)
+						strValues[fieldName] = hexutil.Encode(b)
+						if len(b) == 32 { // collect byte arrays of size 32 as hashes
+							seenHashes = append(seenHashes, common.BytesToHash(b))
+						}
+					case byte:
+						b := input.(byte)
+						strValues[fieldName] = string(b)
+					default:
+						return nil, errors.New(fmt.Sprintf("error: unhandled abi type %T", input))
+					}
+				}
+
+				// Only hold onto logs that pass our argument filter, if any
+				if c.ContractInfo.PassesEventFilter(strValues) {
+					raw, err := json.Marshal(log)
+					if err != nil {
+						return nil, err
+					}
+
+					eventsToLogs[event.Name] = append(eventsToLogs[event.Name], types.Log{
+						LogIndex:         log.Index,
+						Values:           strValues,
+						Raw:              raw,
+						TransactionIndex: log.TxIndex,
+						Id:               headerID,
+					})
+
+					// Cache emitted values that pass the argument filter if their caching is turned on
+					if c.ContractInfo.EmittedAddrs != nil {
+						c.ContractInfo.AddEmittedAddr(seenAddrs...)
+					}
+					if c.ContractInfo.EmittedHashes != nil {
+						c.ContractInfo.AddEmittedHash(seenHashes...)
+					}
+				}
+			}
+		}
+	}
+
+	return eventsToLogs, nil
+}

pkg/contract_watcher/light/converter/converter_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package converter_test
+
+import (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestConverter(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Converter Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/contract_watcher/light/converter/converter_test.go

@@ -0,0 +1,151 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package converter_test
+
+import (
+	"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/contract_watcher/light/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
+)
+
+var _ = Describe("Converter", func() {
+	var con *contract.Contract
+	var tusdWantedEvents = []string{"Transfer", "Mint"}
+	var ensWantedEvents = []string{"NewOwner"}
+	var err error
+
+	Describe("Update", func() {
+		It("Updates contract info held by the converter", func() {
+			con = test_helpers.SetupTusdContract(tusdWantedEvents, []string{})
+			c := converter.NewConverter(con)
+			Expect(c.ContractInfo).To(Equal(con))
+
+			info := test_helpers.SetupTusdContract([]string{}, []string{})
+			c.Update(info)
+			Expect(c.ContractInfo).To(Equal(info))
+		})
+	})
+
+	Describe("Convert", func() {
+		It("Converts a watched event log to mapping of event input names to values", func() {
+			con = test_helpers.SetupTusdContract(tusdWantedEvents, []string{})
+			_, ok := con.Events["Approval"]
+			Expect(ok).To(Equal(false))
+
+			event, ok := con.Events["Transfer"]
+			Expect(ok).To(Equal(true))
+
+			c := converter.NewConverter(con)
+			logs, err := c.Convert([]types.Log{mocks.MockTransferLog1, mocks.MockTransferLog2}, event, 232)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(logs)).To(Equal(2))
+
+			sender1 := common.HexToAddress("0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391")
+			sender2 := common.HexToAddress("0x000000000000000000000000000000000000000000000000000000000000af21")
+			value := helpers.BigFromString("1097077688018008265106216665536940668749033598146")
+
+			Expect(logs[0].Values["to"]).To(Equal(sender1.String()))
+			Expect(logs[0].Values["from"]).To(Equal(sender2.String()))
+			Expect(logs[0].Values["value"]).To(Equal(value.String()))
+			Expect(logs[0].Id).To(Equal(int64(232)))
+			Expect(logs[1].Values["to"]).To(Equal(sender2.String()))
+			Expect(logs[1].Values["from"]).To(Equal(sender1.String()))
+			Expect(logs[1].Values["value"]).To(Equal(value.String()))
+			Expect(logs[1].Id).To(Equal(int64(232)))
+		})
+
+		It("Keeps track of addresses it sees if they will be used for method polling", func() {
+			con = test_helpers.SetupTusdContract(tusdWantedEvents, []string{"balanceOf"})
+			event, ok := con.Events["Transfer"]
+			Expect(ok).To(Equal(true))
+
+			c := converter.NewConverter(con)
+			_, err := c.Convert([]types.Log{mocks.MockTransferLog1, mocks.MockTransferLog2}, event, 232)
+			Expect(err).ToNot(HaveOccurred())
+
+			b, ok := con.EmittedAddrs[common.HexToAddress("0x000000000000000000000000000000000000Af21")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = con.EmittedAddrs[common.HexToAddress("0x09BbBBE21a5975cAc061D82f7b843bCE061BA391")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			_, ok = con.EmittedAddrs[common.HexToAddress("0x")]
+			Expect(ok).To(Equal(false))
+
+			_, ok = con.EmittedAddrs[""]
+			Expect(ok).To(Equal(false))
+
+			_, ok = con.EmittedAddrs[common.HexToAddress("0x09THISE21a5IS5cFAKE1D82fAND43bCE06MADEUP")]
+			Expect(ok).To(Equal(false))
+
+			_, ok = con.EmittedHashes[common.HexToHash("0x000000000000000000000000c02aaa39b223helloa0e5c4f27ead9083c752553")]
+			Expect(ok).To(Equal(false))
+		})
+
+		It("Keeps track of hashes it sees if they will be used for method polling", func() {
+			con = test_helpers.SetupENSContract(ensWantedEvents, []string{"owner"})
+			event, ok := con.Events["NewOwner"]
+			Expect(ok).To(Equal(true))
+
+			c := converter.NewConverter(con)
+			_, err := c.Convert([]types.Log{mocks.MockNewOwnerLog1, mocks.MockNewOwnerLog2}, event, 232)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(con.EmittedHashes)).To(Equal(3))
+
+			b, ok := con.EmittedHashes[common.HexToHash("0x000000000000000000000000c02aaa39b223helloa0e5c4f27ead9083c752553")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = con.EmittedHashes[common.HexToHash("0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba391")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = con.EmittedHashes[common.HexToHash("0x9dd48110dcc444fdc242510c09bbbbe21a5975cac061d82f7b843bce061ba400")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			_, ok = con.EmittedHashes[common.HexToHash("0x9dd48thiscc444isc242510c0made03upa5975cac061dhashb843bce061ba400")]
+			Expect(ok).To(Equal(false))
+
+			_, ok = con.EmittedHashes[common.HexToAddress("0x")]
+			Expect(ok).To(Equal(false))
+
+			_, ok = con.EmittedHashes[""]
+			Expect(ok).To(Equal(false))
+
+			// Does not keep track of emitted addresses if the methods provided will not use them
+			_, ok = con.EmittedAddrs[common.HexToAddress("0x000000000000000000000000000000000000Af21")]
+			Expect(ok).To(Equal(false))
+		})
+
+		It("Fails with an empty contract", func() {
+			event := con.Events["Transfer"]
+			c := converter.NewConverter(&contract.Contract{})
+			_, err = c.Convert([]types.Log{mocks.MockTransferLog1}, event, 232)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+})

pkg/contract_watcher/light/fetcher/fetcher.go

@@ -0,0 +1,69 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package fetcher
+
+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"
+)
+
+type Fetcher interface {
+	FetchLogs(contractAddresses []string, topics []common.Hash, missingHeader core.Header) ([]types.Log, error)
+}
+
+type fetcher struct {
+	blockChain core.BlockChain
+}
+
+func NewFetcher(blockchain core.BlockChain) *fetcher {
+	return &fetcher{
+		blockChain: blockchain,
+	}
+}
+
+// Checks all topic0s, on all addresses, fetching matching logs for the given header
+func (fetcher *fetcher) FetchLogs(contractAddresses []string, topic0s []common.Hash, header core.Header) ([]types.Log, error) {
+	addresses := hexStringsToAddresses(contractAddresses)
+	blockHash := common.HexToHash(header.Hash)
+	query := ethereum.FilterQuery{
+		BlockHash: &blockHash,
+		Addresses: addresses,
+		// Search for _any_ of the topics in topic0 position; see docs on `FilterQuery`
+		Topics: [][]common.Hash{topic0s},
+	}
+
+	logs, err := fetcher.blockChain.GetEthLogsWithCustomQuery(query)
+	if err != nil {
+		// TODO review aggregate fetching error handling
+		return []types.Log{}, err
+	}
+
+	return logs, nil
+}
+
+func hexStringsToAddresses(hexStrings []string) []common.Address {
+	var addresses []common.Address
+	for _, hexString := range hexStrings {
+		address := common.HexToAddress(hexString)
+		addresses = append(addresses, address)
+	}
+
+	return addresses
+}

pkg/contract_watcher/light/fetcher/fetcher_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package fetcher_test
+
+import (
+	"io/ioutil"
+	"log"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestFetcher(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Fetcher Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	log.SetOutput(ioutil.Discard)
+})

pkg/contract_watcher/light/fetcher/fetcher_test.go

@@ -0,0 +1,66 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package fetcher_test
+
+import (
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/fetcher"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/fakes"
+)
+
+var _ = Describe("Fetcher", func() {
+	Describe("FetchLogs", func() {
+		It("fetches logs based on the given query", func() {
+			blockChain := fakes.NewMockBlockChain()
+			fetcher := fetcher.NewFetcher(blockChain)
+			header := fakes.FakeHeader
+
+			addresses := []string{"0xfakeAddress", "0xanotherFakeAddress"}
+			topicZeros := [][]common.Hash{{common.BytesToHash([]byte{1, 2, 3, 4, 5})}}
+
+			_, err := fetcher.FetchLogs(addresses, []common.Hash{common.BytesToHash([]byte{1, 2, 3, 4, 5})}, header)
+
+			address1 := common.HexToAddress("0xfakeAddress")
+			address2 := common.HexToAddress("0xanotherFakeAddress")
+			Expect(err).NotTo(HaveOccurred())
+
+			blockHash := common.HexToHash(header.Hash)
+			expectedQuery := ethereum.FilterQuery{
+				BlockHash: &blockHash,
+				Addresses: []common.Address{address1, address2},
+				Topics:    topicZeros,
+			}
+			blockChain.AssertGetEthLogsWithCustomQueryCalledWith(expectedQuery)
+		})
+
+		It("returns an error if fetching the logs fails", func() {
+			blockChain := fakes.NewMockBlockChain()
+			blockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
+			fetcher := fetcher.NewFetcher(blockChain)
+
+			_, err := fetcher.FetchLogs([]string{}, []common.Hash{}, core.Header{})
+
+			Expect(err).To(HaveOccurred())
+			Expect(err).To(MatchError(fakes.FakeError))
+		})
+	})
+})

pkg/contract_watcher/light/repository/header_repository.go

@@ -0,0 +1,259 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package repository
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/hashicorp/golang-lru"
+
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+const columnCacheSize = 1000
+
+type HeaderRepository interface {
+	AddCheckColumn(id string) error
+	AddCheckColumns(ids []string) error
+	MarkHeaderChecked(headerID int64, eventID string) error
+	MarkHeaderCheckedForAll(headerID int64, ids []string) error
+	MarkHeadersCheckedForAll(headers []core.Header, ids []string) error
+	MissingHeaders(startingBlockNumber int64, endingBlockNumber int64, eventID string) ([]core.Header, error)
+	MissingMethodsCheckedEventsIntersection(startingBlockNumber, endingBlockNumber int64, methodIds, eventIds []string) ([]core.Header, error)
+	MissingHeadersForAll(startingBlockNumber, endingBlockNumber int64, ids []string) ([]core.Header, error)
+	CheckCache(key string) (interface{}, bool)
+}
+
+type headerRepository struct {
+	db      *postgres.DB
+	columns *lru.Cache // Cache created columns to minimize db connections
+}
+
+func NewHeaderRepository(db *postgres.DB) *headerRepository {
+	ccs, _ := lru.New(columnCacheSize)
+	return &headerRepository{
+		db:      db,
+		columns: ccs,
+	}
+}
+
+func (r *headerRepository) AddCheckColumn(id string) error {
+	// Check cache to see if column already exists before querying pg
+	_, ok := r.columns.Get(id)
+	if ok {
+		return nil
+	}
+
+	pgStr := "ALTER TABLE public.checked_headers ADD COLUMN IF NOT EXISTS "
+	pgStr = pgStr + id + " INTEGER NOT NULL DEFAULT 0"
+	_, err := r.db.Exec(pgStr)
+	if err != nil {
+		return err
+	}
+
+	// Add column name to cache
+	r.columns.Add(id, true)
+
+	return nil
+}
+
+func (r *headerRepository) AddCheckColumns(ids []string) error {
+	var err error
+	baseQuery := "ALTER TABLE public.checked_headers"
+	input := make([]string, 0, len(ids))
+	for _, id := range ids {
+		_, ok := r.columns.Get(id)
+		if !ok {
+			baseQuery += " ADD COLUMN IF NOT EXISTS " + id + " INTEGER NOT NULL DEFAULT 0,"
+			input = append(input, id)
+		}
+	}
+	if len(input) > 0 {
+		_, err = r.db.Exec(baseQuery[:len(baseQuery)-1])
+		if err == nil {
+			for _, id := range input {
+				r.columns.Add(id, true)
+			}
+		}
+	}
+
+	return err
+}
+
+func (r *headerRepository) MarkHeaderChecked(headerID int64, id string) error {
+	_, err := r.db.Exec(`INSERT INTO public.checked_headers (header_id, `+id+`)
+		VALUES ($1, $2) 
+		ON CONFLICT (header_id) DO
+			UPDATE SET `+id+` = checked_headers.`+id+` + 1`, headerID, 1)
+
+	return err
+}
+
+func (r *headerRepository) MarkHeaderCheckedForAll(headerID int64, ids []string) error {
+	pgStr := "INSERT INTO public.checked_headers (header_id, "
+	for _, id := range ids {
+		pgStr += id + ", "
+	}
+	pgStr = pgStr[:len(pgStr)-2] + ") VALUES ($1, "
+	for i := 0; i < len(ids); i++ {
+		pgStr += "1, "
+	}
+	pgStr = pgStr[:len(pgStr)-2] + ") ON CONFLICT (header_id) DO UPDATE SET "
+	for _, id := range ids {
+		pgStr += id + `= checked_headers.` + id + ` + 1, `
+	}
+	pgStr = pgStr[:len(pgStr)-2]
+	_, err := r.db.Exec(pgStr, headerID)
+
+	return err
+}
+
+func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids []string) error {
+	tx, err := r.db.Begin()
+	if err != nil {
+		return err
+	}
+
+	for _, header := range headers {
+		pgStr := "INSERT INTO public.checked_headers (header_id, "
+		for _, id := range ids {
+			pgStr += id + ", "
+		}
+		pgStr = pgStr[:len(pgStr)-2] + ") VALUES ($1, "
+		for i := 0; i < len(ids); i++ {
+			pgStr += "1, "
+		}
+		pgStr = pgStr[:len(pgStr)-2] + ") ON CONFLICT (header_id) DO UPDATE SET "
+		for _, id := range ids {
+			pgStr += fmt.Sprintf("%s = checked_headers.%s + 1, ", id, id)
+		}
+		pgStr = pgStr[:len(pgStr)-2]
+		_, err = tx.Exec(pgStr, header.Id)
+		if err != nil {
+			tx.Rollback()
+			return err
+		}
+	}
+
+	return tx.Commit()
+}
+
+func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64, id string) ([]core.Header, error) {
+	var result []core.Header
+	var query string
+	var err error
+
+	if endingBlockNumber == -1 {
+		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				LEFT JOIN checked_headers on headers.id = header_id
+				WHERE (header_id ISNULL OR checked_headers.` + id + `=0)
+				AND headers.block_number >= $1
+				AND headers.eth_node_fingerprint = $2
+				ORDER BY headers.block_number`
+		err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
+	} else {
+		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				LEFT JOIN checked_headers on headers.id = header_id
+				WHERE (header_id ISNULL OR checked_headers.` + id + `=0)
+				AND headers.block_number >= $1
+				AND headers.block_number <= $2
+				AND headers.eth_node_fingerprint = $3
+				ORDER BY headers.block_number`
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+
+	return result, err
+}
+
+func (r *headerRepository) MissingHeadersForAll(startingBlockNumber, endingBlockNumber int64, ids []string) ([]core.Header, error) {
+	var result []core.Header
+	var query string
+	var err error
+
+	baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				  LEFT JOIN checked_headers on headers.id = header_id
+				  WHERE (header_id ISNULL`
+	for _, id := range ids {
+		baseQuery += ` OR checked_headers.` + id + `= 0`
+	}
+
+	if endingBlockNumber == -1 {
+		endStr := `) AND headers.block_number >= $1
+				  AND headers.eth_node_fingerprint = $2
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
+	} else {
+		endStr := `) AND headers.block_number >= $1
+				  AND headers.block_number <= $2
+				  AND headers.eth_node_fingerprint = $3
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+
+	return result, err
+}
+
+func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlockNumber, endingBlockNumber int64, methodIds, eventIds []string) ([]core.Header, error) {
+	var result []core.Header
+	var query string
+	var err error
+
+	baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
+				  LEFT JOIN checked_headers on headers.id = header_id
+				  WHERE (header_id IS NOT NULL`
+	for _, id := range eventIds {
+		baseQuery += ` AND ` + id + `!=0`
+	}
+	baseQuery += `) AND (`
+	for _, id := range methodIds {
+		baseQuery += id + ` =0 AND `
+	}
+	baseQuery = baseQuery[:len(baseQuery)-5] + `) `
+
+	if endingBlockNumber == -1 {
+		endStr := `AND headers.block_number >= $1
+				  AND headers.eth_node_fingerprint = $2
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
+	} else {
+		endStr := `AND headers.block_number >= $1
+				  AND headers.block_number <= $2
+				  AND headers.eth_node_fingerprint = $3
+				  ORDER BY headers.block_number`
+		query = baseQuery + endStr
+		err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
+	}
+
+	return result, err
+}
+
+func (r *headerRepository) CheckCache(key string) (interface{}, bool) {
+	return r.columns.Get(key)
+}
+
+func MarkHeaderCheckedInTransaction(headerID int64, tx *sql.Tx, eventID string) error {
+	_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`)
+		VALUES ($1, $2) 
+		ON CONFLICT (header_id) DO
+			UPDATE SET `+eventID+` = checked_headers.`+eventID+` + 1`, headerID, 1)
+	return err
+}

pkg/contract_watcher/light/repository/header_repository_test.go

@@ -0,0 +1,317 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package repository_test
+
+import (
+	"fmt"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+)
+
+var _ = Describe("Repository", func() {
+	var db *postgres.DB
+	var omniHeaderRepo repository.HeaderRepository   // omni/light header repository
+	var coreHeaderRepo repositories.HeaderRepository // pkg/datastore header repository
+	var eventIDs = []string{
+		"eventName_contractAddr",
+		"eventName_contractAddr2",
+		"eventName_contractAddr3",
+	}
+	var methodIDs = []string{
+		"methodName_contractAddr",
+		"methodName_contractAddr2",
+		"methodName_contractAddr3",
+	}
+
+	BeforeEach(func() {
+		db, _ = test_helpers.SetupDBandBC()
+		omniHeaderRepo = repository.NewHeaderRepository(db)
+		coreHeaderRepo = repositories.NewHeaderRepository(db)
+	})
+
+	AfterEach(func() {
+		test_helpers.TearDown(db)
+	})
+
+	Describe("AddCheckColumn", func() {
+		It("Creates a column for the given eventID to mark if the header has been checked for that event", func() {
+			query := fmt.Sprintf("SELECT %s FROM checked_headers", eventIDs[0])
+			_, err := db.Exec(query)
+			Expect(err).To(HaveOccurred())
+
+			err = omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			_, err = db.Exec(query)
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("Caches column it creates so that it does not need to repeatedly query the database to check for it's existence", func() {
+			_, ok := omniHeaderRepo.CheckCache(eventIDs[0])
+			Expect(ok).To(Equal(false))
+
+			err := omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			v, ok := omniHeaderRepo.CheckCache(eventIDs[0])
+			Expect(ok).To(Equal(true))
+			Expect(v).To(Equal(true))
+		})
+	})
+
+	Describe("AddCheckColumns", func() {
+		It("Creates a column for the given eventIDs to mark if the header has been checked for those events", func() {
+			for _, id := range eventIDs {
+				_, err := db.Exec(fmt.Sprintf("SELECT %s FROM checked_headers", id))
+				Expect(err).To(HaveOccurred())
+			}
+
+			err := omniHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			for _, id := range eventIDs {
+				_, err := db.Exec(fmt.Sprintf("SELECT %s FROM checked_headers", id))
+				Expect(err).ToNot(HaveOccurred())
+			}
+		})
+
+		It("Caches columns it creates so that it does not need to repeatedly query the database to check for it's existence", func() {
+			for _, id := range eventIDs {
+				_, ok := omniHeaderRepo.CheckCache(id)
+				Expect(ok).To(Equal(false))
+			}
+
+			err := omniHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			for _, id := range eventIDs {
+				v, ok := omniHeaderRepo.CheckCache(id)
+				Expect(ok).To(Equal(true))
+				Expect(v).To(Equal(true))
+			}
+		})
+	})
+
+	Describe("MissingHeaders", func() {
+		It("Returns all unchecked headers for the given eventID", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+		})
+
+		It("Returns unchecked headers in ascending order", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			h1 := missingHeaders[0]
+			h2 := missingHeaders[1]
+			h3 := missingHeaders[2]
+			Expect(h1.BlockNumber).To(Equal(int64(6194632)))
+			Expect(h2.BlockNumber).To(Equal(int64(6194633)))
+			Expect(h3.BlockNumber).To(Equal(int64(6194634)))
+		})
+
+		It("Fails if eventID does not yet exist in check_headers table", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			_, err = omniHeaderRepo.MissingHeaders(6194630, 6194635, "notEventId")
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("MissingHeadersForAll", func() { // HERE
+		It("Returns all headers that have not been checked for all of the ids provided", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := omniHeaderRepo.MissingHeadersForAll(6194630, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			err = omniHeaderRepo.MarkHeaderChecked(missingHeaders[0].Id, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = omniHeaderRepo.MissingHeadersForAll(6194630, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			err = omniHeaderRepo.MarkHeaderChecked(missingHeaders[0].Id, eventIDs[1])
+			Expect(err).ToNot(HaveOccurred())
+			err = omniHeaderRepo.MarkHeaderChecked(missingHeaders[0].Id, eventIDs[2])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = omniHeaderRepo.MissingHeadersForAll(6194630, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+		})
+
+		It("Fails if one of the eventIDs does not yet exist in check_headers table", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			badEventIDs := append(eventIDs, "notEventId")
+
+			_, err = omniHeaderRepo.MissingHeadersForAll(6194630, 6194635, badEventIDs)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("MarkHeaderChecked", func() {
+		It("Marks the header checked for the given eventID", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			err = omniHeaderRepo.MarkHeaderChecked(headerID, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+		})
+
+		It("Fails if eventID does not yet exist in check_headers table", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumn(eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			err = omniHeaderRepo.MarkHeaderChecked(headerID, "notEventId")
+			Expect(err).To(HaveOccurred())
+
+			missingHeaders, err = omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+		})
+	})
+
+	Describe("MarkHeaderCheckedForAll", func() {
+		It("Marks the header checked for all provided column ids", func() {
+			addHeaders(coreHeaderRepo)
+			err := omniHeaderRepo.AddCheckColumns(eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err := omniHeaderRepo.MissingHeadersForAll(6194630, 6194635, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			err = omniHeaderRepo.MarkHeaderCheckedForAll(headerID, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+
+			missingHeaders, err = omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(2))
+		})
+	})
+
+	Describe("MarkHeadersCheckedForAll", func() {
+		It("Marks the headers checked for all provided column ids", func() {
+			addHeaders(coreHeaderRepo)
+			methodIDs := []string{
+				"methodName_contractAddr",
+				"methodName_contractAddr2",
+				"methodName_contractAddr3",
+			}
+
+			var missingHeaders []core.Header
+			for _, id := range methodIDs {
+				err := omniHeaderRepo.AddCheckColumn(id)
+				Expect(err).ToNot(HaveOccurred())
+				missingHeaders, err = omniHeaderRepo.MissingHeaders(6194630, 6194635, id)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(len(missingHeaders)).To(Equal(3))
+			}
+
+			err := omniHeaderRepo.MarkHeadersCheckedForAll(missingHeaders, methodIDs)
+			Expect(err).ToNot(HaveOccurred())
+			for _, id := range methodIDs {
+				missingHeaders, err = omniHeaderRepo.MissingHeaders(6194630, 6194635, id)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(len(missingHeaders)).To(Equal(0))
+			}
+		})
+	})
+
+	Describe("MissingMethodsCheckedEventsIntersection", func() {
+		It("Returns headers that have been checked for all the provided events but have not been checked for all the provided methods", func() {
+			addHeaders(coreHeaderRepo)
+			for i, id := range eventIDs {
+				err := omniHeaderRepo.AddCheckColumn(id)
+				Expect(err).ToNot(HaveOccurred())
+				err = omniHeaderRepo.AddCheckColumn(methodIDs[i])
+				Expect(err).ToNot(HaveOccurred())
+			}
+
+			missingHeaders, err := omniHeaderRepo.MissingHeaders(6194630, 6194635, eventIDs[0])
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(missingHeaders)).To(Equal(3))
+
+			headerID := missingHeaders[0].Id
+			headerID2 := missingHeaders[1].Id
+			for i, id := range eventIDs {
+				err = omniHeaderRepo.MarkHeaderChecked(headerID, id)
+				Expect(err).ToNot(HaveOccurred())
+				err = omniHeaderRepo.MarkHeaderChecked(headerID2, id)
+				Expect(err).ToNot(HaveOccurred())
+				err = omniHeaderRepo.MarkHeaderChecked(headerID, methodIDs[i])
+				Expect(err).ToNot(HaveOccurred())
+			}
+
+			intersectionHeaders, err := omniHeaderRepo.MissingMethodsCheckedEventsIntersection(6194630, 6194635, methodIDs, eventIDs)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(intersectionHeaders)).To(Equal(1))
+			Expect(intersectionHeaders[0].Id).To(Equal(headerID2))
+
+		})
+	})
+})
+
+func addHeaders(coreHeaderRepo repositories.HeaderRepository) {
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader1)
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader2)
+	coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader3)
+}

pkg/contract_watcher/light/repository/repository_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package repository_test
+
+import (
+	"github.com/sirupsen/logrus"
+	"io/ioutil"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestRepository(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Repository Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	logrus.SetOutput(ioutil.Discard)
+})

pkg/contract_watcher/light/retriever/block_retriever.go

@@ -0,0 +1,60 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package retriever
+
+import (
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+// Block retriever is used to retrieve the first block for a given contract and the most recent block
+// It requires a vDB synced database with blocks, transactions, receipts, and logs
+type BlockRetriever interface {
+	RetrieveFirstBlock() (int64, error)
+	RetrieveMostRecentBlock() (int64, error)
+}
+
+type blockRetriever struct {
+	db *postgres.DB
+}
+
+func NewBlockRetriever(db *postgres.DB) (r *blockRetriever) {
+	return &blockRetriever{
+		db: db,
+	}
+}
+
+// Retrieve block number of earliest header in repo
+func (r *blockRetriever) RetrieveFirstBlock() (int64, error) {
+	var firstBlock int
+	err := r.db.Get(
+		&firstBlock,
+		"SELECT block_number FROM headers ORDER BY block_number LIMIT 1",
+	)
+
+	return int64(firstBlock), err
+}
+
+// Retrieve block number of latest header in repo
+func (r *blockRetriever) RetrieveMostRecentBlock() (int64, error) {
+	var lastBlock int
+	err := r.db.Get(
+		&lastBlock,
+		"SELECT block_number FROM headers ORDER BY block_number DESC LIMIT 1",
+	)
+
+	return int64(lastBlock), err
+}

pkg/contract_watcher/light/retriever/block_retriever_test.go

@@ -0,0 +1,79 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package retriever_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+)
+
+var _ = Describe("Block Retriever", func() {
+	var db *postgres.DB
+	var r retriever.BlockRetriever
+	var headerRepository repositories.HeaderRepository
+
+	BeforeEach(func() {
+		db, _ = test_helpers.SetupDBandBC()
+		headerRepository = repositories.NewHeaderRepository(db)
+		r = retriever.NewBlockRetriever(db)
+	})
+
+	AfterEach(func() {
+		test_helpers.TearDown(db)
+	})
+
+	Describe("RetrieveFirstBlock", func() {
+		It("Retrieves block number of earliest header in the database", func() {
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader2)
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
+
+			i, err := r.RetrieveFirstBlock()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(i).To(Equal(int64(6194632)))
+		})
+
+		It("Fails if no headers can be found in the database", func() {
+			_, err := r.RetrieveFirstBlock()
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("RetrieveMostRecentBlock", func() {
+		It("Retrieves the latest header's block number", func() {
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader2)
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
+
+			i, err := r.RetrieveMostRecentBlock()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(i).To(Equal(int64(6194634)))
+		})
+
+		It("Fails if no headers can be found in the database", func() {
+			i, err := r.RetrieveMostRecentBlock()
+			Expect(err).To(HaveOccurred())
+			Expect(i).To(Equal(int64(0)))
+		})
+	})
+})

pkg/contract_watcher/light/retriever/retriever_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package retriever_test
+
+import (
+	"github.com/sirupsen/logrus"
+	"io/ioutil"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestRetriever(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light BLock Number Retriever Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	logrus.SetOutput(ioutil.Discard)
+})

pkg/contract_watcher/light/transformer/transformer.go

@@ -0,0 +1,320 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package transformer
+
+import (
+	"errors"
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/common"
+	gethTypes "github.com/ethereum/go-ethereum/core/types"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/converter"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/fetcher"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/retriever"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/contract"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/parser"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/poller"
+	srep "github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/repository"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+// Requires a light synced vDB (headers) and a running eth node (or infura)
+type transformer struct {
+	// Database interfaces
+	srep.EventRepository        // Holds transformed watched event log data
+	repository.HeaderRepository // Interface for interaction with header repositories
+
+	// Pre-processing interfaces
+	parser.Parser            // Parses events and methods out of contract abi fetched using contract address
+	retriever.BlockRetriever // Retrieves first block for contract and current block height
+
+	// Processing interfaces
+	fetcher.Fetcher     // Fetches event logs, using header hashes
+	converter.Converter // Converts watched event logs into custom log
+	poller.Poller       // Polls methods using arguments collected from events and persists them using a method datastore
+
+	// Store contract configuration information
+	Config config.ContractConfig
+
+	// Store contract info as mapping to contract address
+	Contracts map[string]*contract.Contract
+
+	// Internally configured transformer variables
+	contractAddresses []string            // Holds all contract addresses, for batch fetching of logs
+	sortedEventIds    map[string][]string // Map to sort event column ids by contract, for post fetch processing and persisting of logs
+	sortedMethodIds   map[string][]string // Map to sort method column ids by contract, for post fetch method polling
+	eventIds          []string            // Holds event column ids across all contract, for batch fetching of headers
+	eventFilters      []common.Hash       // Holds topic0 hashes across all contracts, for batch fetching of logs
+	start             int64               // Hold the lowest starting block and the highest ending block
+}
+
+// Order-of-operations:
+// 1. Create new transformer
+// 2. Load contract addresses and their parameters
+// 3. Init
+// 4. Execute
+
+// Transformer takes in config for blockchain, database, and network id
+func NewTransformer(con config.ContractConfig, bc core.BlockChain, db *postgres.DB) *transformer {
+
+	return &transformer{
+		Poller:           poller.NewPoller(bc, db, types.LightSync),
+		Fetcher:          fetcher.NewFetcher(bc),
+		Parser:           parser.NewParser(con.Network),
+		HeaderRepository: repository.NewHeaderRepository(db),
+		BlockRetriever:   retriever.NewBlockRetriever(db),
+		Converter:        converter.NewConverter(&contract.Contract{}),
+		Contracts:        map[string]*contract.Contract{},
+		EventRepository:  srep.NewEventRepository(db, types.LightSync),
+		Config:           con,
+	}
+}
+
+// Use after creating and setting transformer
+// Loops over all of the addr => filter sets
+// Uses parser to pull event info from abi
+// Use this info to generate event filters
+func (tr *transformer) Init() error {
+	// Initialize internally configured transformer settings
+	tr.contractAddresses = make([]string, 0)       // Holds all contract addresses, for batch fetching of logs
+	tr.sortedEventIds = make(map[string][]string)  // Map to sort event column ids by contract, for post fetch processing and persisting of logs
+	tr.sortedMethodIds = make(map[string][]string) // Map to sort method column ids by contract, for post fetch method polling
+	tr.eventIds = make([]string, 0)                // Holds event column ids across all contract, for batch fetching of headers
+	tr.eventFilters = make([]common.Hash, 0)       // Holds topic0 hashes across all contracts, for batch fetching of logs
+	tr.start = 100000000000                        // Hold the lowest starting block and the highest ending block
+
+	// Iterate through all internal contract addresses
+	for contractAddr := range tr.Config.Addresses {
+		// Configure Abi
+		if tr.Config.Abis[contractAddr] == "" {
+			// If no abi is given in the config, this method will try fetching from internal look-up table and etherscan
+			err := tr.Parser.Parse(contractAddr)
+			if err != nil {
+				return err
+			}
+		} else {
+			// If we have an abi from the config, load that into the parser
+			err := tr.Parser.ParseAbiStr(tr.Config.Abis[contractAddr])
+			if err != nil {
+				return err
+			}
+		}
+
+		// Get first block and most recent block number in the header repo
+		firstBlock, err := tr.BlockRetriever.RetrieveFirstBlock()
+		if err != nil {
+			return err
+		}
+		lastBlock, err := tr.BlockRetriever.RetrieveMostRecentBlock()
+		if err != nil {
+			return err
+		}
+
+		// Set to specified range if it falls within the bounds
+		if firstBlock < tr.Config.StartingBlocks[contractAddr] {
+			firstBlock = tr.Config.StartingBlocks[contractAddr]
+		}
+
+		// Get contract name if it has one
+		var name = new(string)
+		tr.FetchContractData(tr.Abi(), contractAddr, "name", nil, &name, lastBlock)
+
+		// Remove any potential accidental duplicate inputs
+		eventArgs := map[string]bool{}
+		for _, arg := range tr.Config.EventArgs[contractAddr] {
+			eventArgs[arg] = true
+		}
+		methodArgs := map[string]bool{}
+		for _, arg := range tr.Config.MethodArgs[contractAddr] {
+			methodArgs[arg] = true
+		}
+
+		// Aggregate info into contract object and store for execution
+		con := contract.Contract{
+			Name:          *name,
+			Network:       tr.Config.Network,
+			Address:       contractAddr,
+			Abi:           tr.Parser.Abi(),
+			ParsedAbi:     tr.Parser.ParsedAbi(),
+			StartingBlock: firstBlock,
+			LastBlock:     -1,
+			Events:        tr.Parser.GetEvents(tr.Config.Events[contractAddr]),
+			Methods:       tr.Parser.GetSelectMethods(tr.Config.Methods[contractAddr]),
+			FilterArgs:    eventArgs,
+			MethodArgs:    methodArgs,
+			Piping:        tr.Config.Piping[contractAddr],
+		}.Init()
+		tr.Contracts[contractAddr] = con
+		tr.contractAddresses = append(tr.contractAddresses, con.Address)
+
+		// Create checked_headers columns for each event id and append to list of all event ids
+		tr.sortedEventIds[con.Address] = make([]string, 0, len(con.Events))
+		for _, event := range con.Events {
+			eventId := strings.ToLower(event.Name + "_" + con.Address)
+			err := tr.HeaderRepository.AddCheckColumn(eventId)
+			if err != nil {
+				return err
+			}
+			// Keep track of this event id; sorted and unsorted
+			tr.sortedEventIds[con.Address] = append(tr.sortedEventIds[con.Address], eventId)
+			tr.eventIds = append(tr.eventIds, eventId)
+			// Append this event sig to the filters
+			tr.eventFilters = append(tr.eventFilters, event.Sig())
+		}
+
+		// Create checked_headers columns for each method id and append list of all method ids
+		tr.sortedMethodIds[con.Address] = make([]string, 0, len(con.Methods))
+		for _, m := range con.Methods {
+			methodId := strings.ToLower(m.Name + "_" + con.Address)
+			err := tr.HeaderRepository.AddCheckColumn(methodId)
+			if err != nil {
+				return err
+			}
+			tr.sortedMethodIds[con.Address] = append(tr.sortedMethodIds[con.Address], methodId)
+		}
+
+		// Update start to the lowest block
+		if con.StartingBlock < tr.start {
+			tr.start = con.StartingBlock
+		}
+	}
+
+	return nil
+}
+
+func (tr *transformer) Execute() error {
+	if len(tr.Contracts) == 0 {
+		return errors.New("error: transformer has no initialized contracts")
+	}
+
+	// Map to sort batch fetched logs by which contract they belong to, for post fetch processing
+	sortedLogs := make(map[string][]gethTypes.Log)
+	for _, con := range tr.Contracts {
+		sortedLogs[con.Address] = []gethTypes.Log{}
+	}
+
+	// Find unchecked headers for all events across all contracts; these are returned in asc order
+	missingHeaders, err := tr.HeaderRepository.MissingHeadersForAll(tr.start, -1, tr.eventIds)
+	if err != nil {
+		return err
+	}
+
+	// Iterate over headers
+	for _, header := range missingHeaders {
+		// And fetch all event logs across contracts at this header
+		allLogs, err := tr.Fetcher.FetchLogs(tr.contractAddresses, tr.eventFilters, header)
+		if err != nil {
+			return err
+		}
+
+		// If no logs are found mark the header checked for all of these eventIDs
+		// and continue to method polling and onto the next iteration
+		if len(allLogs) < 1 {
+			err = tr.HeaderRepository.MarkHeaderCheckedForAll(header.Id, tr.eventIds)
+			if err != nil {
+				return err
+			}
+			err = tr.methodPolling(header, tr.sortedMethodIds)
+			if err != nil {
+				return err
+			}
+			continue
+		}
+
+		// Sort logs by the contract they belong to
+		for _, log := range allLogs {
+			addr := strings.ToLower(log.Address.Hex())
+			sortedLogs[addr] = append(sortedLogs[addr], log)
+		}
+
+		// Process logs for each contract
+		for conAddr, logs := range sortedLogs {
+			if logs == nil {
+				continue
+			}
+			// Configure converter with this contract
+			con := tr.Contracts[conAddr]
+			tr.Converter.Update(con)
+
+			// Convert logs into batches of log mappings (eventName => []types.Logs
+			convertedLogs, err := tr.Converter.ConvertBatch(logs, con.Events, header.Id)
+			if err != nil {
+				return err
+			}
+			// Cycle through each type of event log and persist them
+			for eventName, logs := range convertedLogs {
+				// If logs for this event are empty, mark them checked at this header and continue
+				if len(logs) < 1 {
+					eventId := strings.ToLower(eventName + "_" + con.Address)
+					err = tr.HeaderRepository.MarkHeaderChecked(header.Id, eventId)
+					if err != nil {
+						return err
+					}
+					continue
+				}
+				// If logs aren't empty, persist them
+				// Header is marked checked in the transactions
+				err = tr.EventRepository.PersistLogs(logs, con.Events[eventName], con.Address, con.Name)
+				if err != nil {
+					return err
+				}
+			}
+		}
+
+		// Poll contracts at this block height
+		err = tr.methodPolling(header, tr.sortedMethodIds)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// Used to poll contract methods at a given header
+func (tr *transformer) methodPolling(header core.Header, sortedMethodIds map[string][]string) error {
+	for _, con := range tr.Contracts {
+		// Skip method polling processes if no methods are specified
+		// Also don't try to poll methods below this contract's specified starting block
+		if len(con.Methods) == 0 || header.BlockNumber < con.StartingBlock {
+			continue
+		}
+
+		// Poll all methods for this contract at this header
+		err := tr.Poller.PollContractAt(*con, header.BlockNumber)
+		if err != nil {
+			return err
+		}
+
+		// Mark this header checked for the methods
+		err = tr.HeaderRepository.MarkHeaderCheckedForAll(header.Id, sortedMethodIds[con.Address])
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (tr *transformer) GetConfig() config.ContractConfig {
+	return tr.Config
+}

pkg/contract_watcher/light/transformer/transformer_suite_test.go

@@ -0,0 +1,35 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package transformer_test
+
+import (
+	"github.com/sirupsen/logrus"
+	"io/ioutil"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestTransformer(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Light Transformer Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	logrus.SetOutput(ioutil.Discard)
+})

pkg/contract_watcher/light/transformer/transformer_test.go

@@ -0,0 +1,462 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package transformer_test
+
+import (
+	"fmt"
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/common"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/transformer"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/constants"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers"
+	"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/helpers/test_helpers/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
+)
+
+var _ = Describe("Transformer", func() {
+	var db *postgres.DB
+	var err error
+	var blockChain core.BlockChain
+	var headerRepository repositories.HeaderRepository
+	var headerID, headerID2 int64
+	var ensAddr = strings.ToLower(constants.EnsContractAddress)
+	var tusdAddr = strings.ToLower(constants.TusdContractAddress)
+
+	BeforeEach(func() {
+		db, blockChain = test_helpers.SetupDBandBC()
+		headerRepository = repositories.NewHeaderRepository(db)
+	})
+
+	AfterEach(func() {
+		test_helpers.TearDown(db)
+	})
+
+	Describe("Init", func() {
+		It("Initializes transformer's contract objects", func() {
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
+			t := transformer.NewTransformer(mocks.TusdConfig, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+
+			c, ok := t.Contracts[tusdAddr]
+			Expect(ok).To(Equal(true))
+
+			Expect(c.StartingBlock).To(Equal(int64(6194632)))
+			Expect(c.LastBlock).To(Equal(int64(-1)))
+			Expect(c.Abi).To(Equal(constants.TusdAbiString))
+			Expect(c.Name).To(Equal("TrueUSD"))
+			Expect(c.Address).To(Equal(tusdAddr))
+		})
+
+		It("Fails to initialize if first and most recent block numbers cannot be fetched from vDB headers table", func() {
+			t := transformer.NewTransformer(mocks.TusdConfig, blockChain, db)
+			err = t.Init()
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("Does nothing if nothing if no addresses are configured", func() {
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
+			headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
+			var testConf config.ContractConfig
+			testConf = mocks.TusdConfig
+			testConf.Addresses = nil
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+
+			_, ok := t.Contracts[tusdAddr]
+			Expect(ok).To(Equal(false))
+		})
+	})
+
+	Describe("Execute- against TrueUSD contract", func() {
+		BeforeEach(func() {
+			header1, err := blockChain.GetHeaderByNumber(6791668)
+			Expect(err).ToNot(HaveOccurred())
+			header2, err := blockChain.GetHeaderByNumber(6791669)
+			Expect(err).ToNot(HaveOccurred())
+			header3, err := blockChain.GetHeaderByNumber(6791670)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header1)
+			headerID, err = headerRepository.CreateOrUpdateHeader(header2)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header3)
+		})
+
+		It("Transforms watched contract data into custom repositories", func() {
+			t := transformer.NewTransformer(mocks.TusdConfig, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			log := test_helpers.LightTransferLog{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.transfer_event", tusdAddr)).StructScan(&log)
+			Expect(err).ToNot(HaveOccurred())
+			// We don't know vulcID, so compare individual fields instead of complete structures
+			Expect(log.HeaderID).To(Equal(headerID))
+			Expect(log.From).To(Equal("0x1062a747393198f70F71ec65A582423Dba7E5Ab3"))
+			Expect(log.To).To(Equal("0x2930096dB16b4A44Ecd4084EA4bd26F7EeF1AEf0"))
+			Expect(log.Value).To(Equal("9998940000000000000000"))
+		})
+
+		It("Keeps track of contract-related addresses while transforming event data if they need to be used for later method polling", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.TusdConfig
+			testConf.Methods = map[string][]string{
+				tusdAddr: {"balanceOf"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			c, ok := t.Contracts[tusdAddr]
+			Expect(ok).To(Equal(true))
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(c.EmittedAddrs)).To(Equal(4))
+			Expect(len(c.EmittedHashes)).To(Equal(0))
+
+			b, ok := c.EmittedAddrs[common.HexToAddress("0x1062a747393198f70F71ec65A582423Dba7E5Ab3")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = c.EmittedAddrs[common.HexToAddress("0x2930096dB16b4A44Ecd4084EA4bd26F7EeF1AEf0")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = c.EmittedAddrs[common.HexToAddress("0x571A326f5B15E16917dC17761c340c1ec5d06f6d")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = c.EmittedAddrs[common.HexToAddress("0xFBb1b73C4f0BDa4f67dcA266ce6Ef42f520fBB98")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			_, ok = c.EmittedAddrs[common.HexToAddress("0x09BbBBE21a5975cAc061D82f7b843b1234567890")]
+			Expect(ok).To(Equal(false))
+
+			_, ok = c.EmittedAddrs[common.HexToAddress("0x")]
+			Expect(ok).To(Equal(false))
+
+			_, ok = c.EmittedAddrs[""]
+			Expect(ok).To(Equal(false))
+
+			_, ok = c.EmittedAddrs[common.HexToAddress("0x09THISE21a5IS5cFAKE1D82fAND43bCE06MADEUP")]
+			Expect(ok).To(Equal(false))
+		})
+
+		It("Polls given methods using generated token holder address", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.TusdConfig
+			testConf.Methods = map[string][]string{
+				tusdAddr: {"balanceOf"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			res := test_helpers.BalanceOf{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.balanceof_method WHERE who_ = '0x1062a747393198f70F71ec65A582423Dba7E5Ab3' AND block = '6791669'", tusdAddr)).StructScan(&res)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Balance).To(Equal("55849938025000000000000"))
+			Expect(res.TokenName).To(Equal("TrueUSD"))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.balanceof_method WHERE who_ = '0x09BbBBE21a5975cAc061D82f7b843b1234567890' AND block = '6791669'", tusdAddr)).StructScan(&res)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("Fails if initialization has not been done", func() {
+			t := transformer.NewTransformer(mocks.TusdConfig, blockChain, db)
+			err = t.Execute()
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("Execute- against ENS registry contract", func() {
+		BeforeEach(func() {
+			header1, err := blockChain.GetHeaderByNumber(6885695)
+			Expect(err).ToNot(HaveOccurred())
+			header2, err := blockChain.GetHeaderByNumber(6885696)
+			Expect(err).ToNot(HaveOccurred())
+			header3, err := blockChain.GetHeaderByNumber(6885697)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header1)
+			headerID, err = headerRepository.CreateOrUpdateHeader(header2)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header3)
+		})
+
+		It("Transforms watched contract data into custom repositories", func() {
+			t := transformer.NewTransformer(mocks.ENSConfig, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			log := test_helpers.LightNewOwnerLog{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.newowner_event", ensAddr)).StructScan(&log)
+			Expect(err).ToNot(HaveOccurred())
+			// We don't know vulcID, so compare individual fields instead of complete structures
+			Expect(log.HeaderID).To(Equal(headerID))
+			Expect(log.Node).To(Equal("0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae"))
+			Expect(log.Label).To(Equal("0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047"))
+			Expect(log.Owner).To(Equal("0x6090A6e47849629b7245Dfa1Ca21D94cd15878Ef"))
+		})
+
+		It("Keeps track of contract-related hashes while transforming event data if they need to be used for later method polling", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.ENSConfig
+			testConf.Methods = map[string][]string{
+				ensAddr: {"owner"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			c, ok := t.Contracts[ensAddr]
+			Expect(ok).To(Equal(true))
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(c.EmittedHashes)).To(Equal(2))
+			Expect(len(c.EmittedAddrs)).To(Equal(0))
+
+			b, ok := c.EmittedHashes[common.HexToHash("0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = c.EmittedHashes[common.HexToHash("0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			// Doesn't keep track of address since it wouldn't be used in calling the 'owner' method
+			_, ok = c.EmittedAddrs[common.HexToAddress("0x6090A6e47849629b7245Dfa1Ca21D94cd15878Ef")]
+			Expect(ok).To(Equal(false))
+		})
+
+		It("Polls given method using list of collected hashes", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.ENSConfig
+			testConf.Methods = map[string][]string{
+				ensAddr: {"owner"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			res := test_helpers.Owner{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae' AND block = '6885696'", ensAddr)).StructScan(&res)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Address).To(Equal("0x6090A6e47849629b7245Dfa1Ca21D94cd15878Ef"))
+			Expect(res.TokenName).To(Equal(""))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047' AND block = '6885696'", ensAddr)).StructScan(&res)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Address).To(Equal("0x0000000000000000000000000000000000000000"))
+			Expect(res.TokenName).To(Equal(""))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x9THIS110dcc444fIS242510c09bbAbe21aFAKEcacNODE82f7b843HASH61ba391' AND block = '6885696'", ensAddr)).StructScan(&res)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("It does not persist events if they do not pass the emitted arg filter", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.ENSConfig
+			testConf.EventArgs = map[string][]string{
+				ensAddr: {"fake_filter_value"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			log := test_helpers.LightNewOwnerLog{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.newowner_event", ensAddr)).StructScan(&log)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("If a method arg filter is applied, only those arguments are used in polling", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.ENSConfig
+			testConf.Methods = map[string][]string{
+				ensAddr: {"owner"},
+			}
+			testConf.MethodArgs = map[string][]string{
+				ensAddr: {"0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			res := test_helpers.Owner{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae' AND block = '6885696'", ensAddr)).StructScan(&res)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Address).To(Equal("0x6090A6e47849629b7245Dfa1Ca21D94cd15878Ef"))
+			Expect(res.TokenName).To(Equal(""))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047' AND block = '6885696'", ensAddr)).StructScan(&res)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("Execute- against both ENS and TrueUSD", func() {
+		BeforeEach(func() {
+			header1, err := blockChain.GetHeaderByNumber(6791668)
+			Expect(err).ToNot(HaveOccurred())
+			header2, err := blockChain.GetHeaderByNumber(6791669)
+			Expect(err).ToNot(HaveOccurred())
+			header3, err := blockChain.GetHeaderByNumber(6791670)
+			Expect(err).ToNot(HaveOccurred())
+			header4, err := blockChain.GetHeaderByNumber(6885695)
+			Expect(err).ToNot(HaveOccurred())
+			header5, err := blockChain.GetHeaderByNumber(6885696)
+			Expect(err).ToNot(HaveOccurred())
+			header6, err := blockChain.GetHeaderByNumber(6885697)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header1)
+			headerID, err = headerRepository.CreateOrUpdateHeader(header2)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header3)
+			headerRepository.CreateOrUpdateHeader(header4)
+			headerID2, err = headerRepository.CreateOrUpdateHeader(header5)
+			Expect(err).ToNot(HaveOccurred())
+			headerRepository.CreateOrUpdateHeader(header6)
+		})
+
+		It("Transforms watched contract data into custom repositories", func() {
+			t := transformer.NewTransformer(mocks.ENSandTusdConfig, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			newOwnerLog := test_helpers.LightNewOwnerLog{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.newowner_event", ensAddr)).StructScan(&newOwnerLog)
+			Expect(err).ToNot(HaveOccurred())
+			// We don't know vulcID, so compare individual fields instead of complete structures
+			Expect(newOwnerLog.HeaderID).To(Equal(headerID2))
+			Expect(newOwnerLog.Node).To(Equal("0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae"))
+			Expect(newOwnerLog.Label).To(Equal("0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047"))
+			Expect(newOwnerLog.Owner).To(Equal("0x6090A6e47849629b7245Dfa1Ca21D94cd15878Ef"))
+
+			transferLog := test_helpers.LightTransferLog{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.transfer_event", tusdAddr)).StructScan(&transferLog)
+			Expect(err).ToNot(HaveOccurred())
+			// We don't know vulcID, so compare individual fields instead of complete structures
+			Expect(transferLog.HeaderID).To(Equal(headerID))
+			Expect(transferLog.From).To(Equal("0x1062a747393198f70F71ec65A582423Dba7E5Ab3"))
+			Expect(transferLog.To).To(Equal("0x2930096dB16b4A44Ecd4084EA4bd26F7EeF1AEf0"))
+			Expect(transferLog.Value).To(Equal("9998940000000000000000"))
+		})
+
+		It("Keeps track of contract-related hashes and addresses while transforming event data if they need to be used for later method polling", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.ENSandTusdConfig
+			testConf.Methods = map[string][]string{
+				ensAddr:  {"owner"},
+				tusdAddr: {"balanceOf"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			ens, ok := t.Contracts[ensAddr]
+			Expect(ok).To(Equal(true))
+			tusd, ok := t.Contracts[tusdAddr]
+			Expect(ok).To(Equal(true))
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(ens.EmittedHashes)).To(Equal(2))
+			Expect(len(ens.EmittedAddrs)).To(Equal(0))
+			Expect(len(tusd.EmittedAddrs)).To(Equal(4))
+			Expect(len(tusd.EmittedHashes)).To(Equal(0))
+
+			b, ok := ens.EmittedHashes[common.HexToHash("0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = ens.EmittedHashes[common.HexToHash("0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = tusd.EmittedAddrs[common.HexToAddress("0x1062a747393198f70F71ec65A582423Dba7E5Ab3")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = tusd.EmittedAddrs[common.HexToAddress("0x2930096dB16b4A44Ecd4084EA4bd26F7EeF1AEf0")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = tusd.EmittedAddrs[common.HexToAddress("0x571A326f5B15E16917dC17761c340c1ec5d06f6d")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+
+			b, ok = tusd.EmittedAddrs[common.HexToAddress("0xFBb1b73C4f0BDa4f67dcA266ce6Ef42f520fBB98")]
+			Expect(ok).To(Equal(true))
+			Expect(b).To(Equal(true))
+		})
+
+		It("Polls given methods for each contract, using list of collected values", func() {
+			var testConf config.ContractConfig
+			testConf = mocks.ENSandTusdConfig
+			testConf.Methods = map[string][]string{
+				ensAddr:  {"owner"},
+				tusdAddr: {"balanceOf"},
+			}
+			t := transformer.NewTransformer(testConf, blockChain, db)
+			err = t.Init()
+			Expect(err).ToNot(HaveOccurred())
+			err = t.Execute()
+			Expect(err).ToNot(HaveOccurred())
+
+			owner := test_helpers.Owner{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x93cdeb708b7545dc668eb9280176169d1c33cfd8ed6f04690a0bcc88a93fc4ae' AND block = '6885696'", ensAddr)).StructScan(&owner)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(owner.Address).To(Equal("0x6090A6e47849629b7245Dfa1Ca21D94cd15878Ef"))
+			Expect(owner.TokenName).To(Equal(""))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x95832c7a47ff8a7840e28b78ce695797aaf402b1c186bad9eca28842625b5047' AND block = '6885696'", ensAddr)).StructScan(&owner)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(owner.Address).To(Equal("0x0000000000000000000000000000000000000000"))
+			Expect(owner.TokenName).To(Equal(""))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.owner_method WHERE node_ = '0x95832c7a47ff8a7840e28b78ceMADEUPaaf4HASHc186badTHItransformers.8IS625bFAKE' AND block = '6885696'", ensAddr)).StructScan(&owner)
+			Expect(err).To(HaveOccurred())
+
+			bal := test_helpers.BalanceOf{}
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.balanceof_method WHERE who_ = '0x1062a747393198f70F71ec65A582423Dba7E5Ab3' AND block = '6791669'", tusdAddr)).StructScan(&bal)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(bal.Balance).To(Equal("55849938025000000000000"))
+			Expect(bal.TokenName).To(Equal("TrueUSD"))
+
+			err = db.QueryRowx(fmt.Sprintf("SELECT * FROM light_%s.balanceof_method WHERE who_ = '0x09BbBBE21a5975cAc061D82f7b843b1234567890' AND block = '6791669'", tusdAddr)).StructScan(&bal)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+})