WIP update tests for aggregate fetching

libraries/shared/watcher.go

@@ -34,12 +34,12 @@ func NewWatcher(db postgres.DB, bc core.BlockChain) Watcher {
 	fetcher := shared.NewFetcher(bc)
 
 	return Watcher{
-		DB: db,
+		DB:         db,
 		Blockchain: bc,
-		Fetcher: fetcher,
+		Fetcher:    fetcher,
-		Chunker: chunker,
+		Chunker:    chunker,
-		Addresses: contractAddresses,
+		Addresses:  contractAddresses,
-		Topics: topic0s,
+		Topics:     topic0s,
 	}
 }
 

libraries/shared/watcher_test.go

@@ -2,33 +2,49 @@ package shared_test
 
 import (
 	"errors"
-
+	"github.com/ethereum/go-ethereum/core/types"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/libraries/shared"
 	"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/pkg/fakes"
 	shared2 "github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
+	"github.com/vulcanize/vulcanizedb/test_config"
 )
 
 type MockTransformer struct {
 	executeWasCalled bool
 	executeError     error
+	passedLogs       []types.Log
+	passedHeader     core.Header
 }
 
-func (mh *MockTransformer) Execute() error {
+func (mh *MockTransformer) Execute(logs []types.Log, header core.Header) error {
 	if mh.executeError != nil {
 		return mh.executeError
 	}
 	mh.executeWasCalled = true
+	mh.passedLogs = logs
+	mh.passedHeader = header
 	return nil
 }
 
-func fakeTransformerInitializer(db *postgres.DB, blockchain core.BlockChain) shared2.Transformer {
+func (mh *MockTransformer) Name() string {
+	return "MockTransformer"
+}
+
+func fakeTransformerInitializer(db *postgres.DB) shared2.Transformer {
 	return &MockTransformer{}
 }
 
 var _ = Describe("Watcher", func() {
+	// TODO Add test for watcher setting the BC
+	// TODO Add tests for log chunk delegation
+	// TODO Add tests for aggregate fetching
+	// TODO Add tests for MissingHeaders
+
 	It("Adds transformers", func() {
 		watcher := shared.Watcher{}
 
@@ -47,24 +63,41 @@ var _ = Describe("Watcher", func() {
 		Expect(len(watcher.Transformers)).To(Equal(2))
 	})
 
-	It("Executes each transformer", func() {
+	Describe("with missing headers", func() {
-		watcher := shared.Watcher{}
+		var (
-		fakeTransformer := &MockTransformer{}
+			db               *postgres.DB
-		watcher.Transformers = []shared2.Transformer{fakeTransformer}
+			watcher          shared.Watcher
+			fakeTransformer  *MockTransformer
+			headerRepository repositories.HeaderRepository
+		)
 
-		watcher.Execute()
+		BeforeEach(func() {
+			db = test_config.NewTestDB(test_config.NewTestNode())
+			test_config.CleanTestDB(db)
+			watcher = shared.NewWatcher(*db, nil)
+			headerRepository = repositories.NewHeaderRepository(db)
+			_, err := headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
+			Expect(err).NotTo(HaveOccurred())
+		})
 
-		Expect(fakeTransformer.executeWasCalled).To(BeTrue())
+		It("executes each transformer", func() {
-	})
+			fakeTransformer = &MockTransformer{}
+			watcher.Transformers = []shared2.Transformer{fakeTransformer}
 
-	It("Returns an error if transformer returns an error", func() {
+			err := watcher.Execute()
-		watcher := shared.Watcher{}
-		fakeTransformer := &MockTransformer{executeError: errors.New("Something bad happened")}
-		watcher.Transformers = []shared2.Transformer{fakeTransformer}
 
-		err := watcher.Execute()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeTransformer.executeWasCalled).To(BeTrue())
+		})
 
-		Expect(err).To(HaveOccurred())
+		It("returns an error if transformer returns an error", func() {
-		Expect(fakeTransformer.executeWasCalled).To(BeFalse())
+			fakeTransformer = &MockTransformer{executeError: errors.New("Something bad happened")}
+			watcher.Transformers = []shared2.Transformer{fakeTransformer}
+
+			err := watcher.Execute()
+
+			Expect(err).To(HaveOccurred())
+			Expect(fakeTransformer.executeWasCalled).To(BeFalse())
+		})
 	})
 })

pkg/datastore/postgres/repositories/header_repository.go

@@ -3,7 +3,6 @@ package repositories
 import (
 	"database/sql"
 	"errors"
-
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )

pkg/transformers/factories/log_note_transformer_test.go

@@ -15,7 +15,7 @@
 package factories_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/core"
@@ -31,9 +31,7 @@ var _ = Describe("LogNoteTransformer", func() {
 	var (
 		repository  mocks.MockRepository
 		converter   mocks.MockLogNoteConverter
-		fetcher     mocks.MockLogFetcher
 		headerOne   core.Header
-		headerTwo   core.Header
 		transformer shared.Transformer
 		model       test_data.GenericModel
 		config      = test_data.GenericTestConfig
@@ -43,75 +41,28 @@ var _ = Describe("LogNoteTransformer", func() {
 	BeforeEach(func() {
 		repository = mocks.MockRepository{}
 		converter = mocks.MockLogNoteConverter{}
-		fetcher = mocks.MockLogFetcher{}
 		transformer = factories.LogNoteTransformer{
 			Config:     config,
 			Converter:  &converter,
 			Repository: &repository,
-			Fetcher:    &fetcher,
+		}.NewLogNoteTransformer(nil)
-		}.NewLogNoteTransformer(nil, nil)
 
 		headerOne = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
-		headerTwo = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
 	})
 
-	It("sets the blockchain and database", func() {
+	It("sets the database", func() {
-		Expect(fetcher.SetBcCalled).To(BeTrue())
 		Expect(repository.SetDbCalled).To(BeTrue())
 	})
 
-	It("gets missing headers for block numbers specified in config", func() {
+	It("marks header checked if no logs are provided", func() {
-		err := transformer.Execute()
+		err := transformer.Execute([]types.Log{}, headerOne)
-
-		Expect(err).NotTo(HaveOccurred())
-		Expect(repository.PassedStartingBlockNumber).To(Equal(config.StartingBlockNumber))
-		Expect(repository.PassedEndingBlockNumber).To(Equal(config.EndingBlockNumber))
-	})
-
-	It("returns error if repository returns error for missing headers", func() {
-		repository.SetMissingHeadersError(fakes.FakeError)
-
-		err := transformer.Execute()
-
-		Expect(err).To(HaveOccurred())
-		Expect(err).To(MatchError(fakes.FakeError))
-	})
-
-	It("fetches logs for missing headers", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
-
-		err := transformer.Execute()
-
-		Expect(err).NotTo(HaveOccurred())
-		Expect(fetcher.FetchedBlocks).To(Equal([]int64{headerOne.BlockNumber, headerTwo.BlockNumber}))
-		Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{config.ContractAddresses, config.ContractAddresses}))
-		expectedTopic := common.HexToHash(config.Topic)
-		Expect(fetcher.FetchedTopics).To(Equal([][]common.Hash{{expectedTopic}}))
-	})
-
-	It("returns error if fetcher returns error", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetcherError(fakes.FakeError)
-
-		err := transformer.Execute()
-
-		Expect(err).To(HaveOccurred())
-		Expect(err).To(MatchError(fakes.FakeError))
-	})
-
-	It("marks header checked if no logs returned", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		repository.AssertMarkHeaderCheckedCalledWith(headerOne.Id)
 	})
 
 	It("doesn't attempt to convert or persist an empty collection when there are no logs", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
+		err := transformer.Execute([]types.Log{}, headerOne)
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(converter.ToModelsCalledCounter).To(Equal(0))
@@ -119,30 +70,23 @@ var _ = Describe("LogNoteTransformer", func() {
 	})
 
 	It("does not call repository.MarkCheckedHeader when there are logs", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
+		err := transformer.Execute(logs, headerOne)
-		fetcher.SetFetchedLogs(logs)
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		repository.AssertMarkHeaderCheckedNotCalled()
 	})
 
 	It("returns error if marking header checked returns err", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
 		repository.SetMarkHeaderCheckedError(fakes.FakeError)
 
-		err := transformer.Execute()
+		err := transformer.Execute([]types.Log{}, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))
 	})
 
 	It("converts matching logs to models", func() {
-		fetcher.SetFetchedLogs(logs)
+		err := transformer.Execute(logs, headerOne)
-		repository.SetMissingHeaders([]core.Header{headerOne})
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(converter.PassedLogs).To(Equal(logs))
@@ -150,20 +94,16 @@ var _ = Describe("LogNoteTransformer", func() {
 
 	It("returns error if converter returns error", func() {
 		converter.SetConverterError(fakes.FakeError)
-		fetcher.SetFetchedLogs(logs)
-		repository.SetMissingHeaders([]core.Header{headerOne})
 
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))
 	})
 
 	It("persists the model", func() {
-		fetcher.SetFetchedLogs(logs)
-		repository.SetMissingHeaders([]core.Header{headerOne})
 		converter.SetReturnModels([]interface{}{model})
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(repository.PassedHeaderID).To(Equal(headerOne.Id))
@@ -171,11 +111,9 @@ var _ = Describe("LogNoteTransformer", func() {
 	})
 
 	It("returns error if repository returns error for create", func() {
-		fetcher.SetFetchedLogs(logs)
-		repository.SetMissingHeaders([]core.Header{headerOne})
 		repository.SetCreateError(fakes.FakeError)
 
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))

pkg/transformers/factories/transformer_test.go

@@ -15,7 +15,7 @@
 package factories_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/core"
@@ -30,86 +30,39 @@ import (
 var _ = Describe("Transformer", func() {
 	var (
 		repository  mocks.MockRepository
-		fetcher     mocks.MockLogFetcher
 		converter   mocks.MockConverter
 		transformer shared.Transformer
 		headerOne   core.Header
-		headerTwo   core.Header
 		config      = test_data.GenericTestConfig
 		logs        = test_data.GenericTestLogs
 	)
 
 	BeforeEach(func() {
 		repository = mocks.MockRepository{}
-		fetcher = mocks.MockLogFetcher{}
 		converter = mocks.MockConverter{}
 
 		transformer = factories.Transformer{
 			Repository: &repository,
-			Fetcher:    &fetcher,
 			Converter:  &converter,
 			Config:     config,
-		}.NewTransformer(nil, nil)
+		}.NewTransformer(nil)
 
 		headerOne = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
-		headerTwo = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
 	})
 
-	It("sets the blockchain and db", func() {
+	It("sets the db", func() {
-		Expect(fetcher.SetBcCalled).To(BeTrue())
 		Expect(repository.SetDbCalled).To(BeTrue())
 	})
 
-	It("gets missing headers for blocks in the configured range", func() {
-		err := transformer.Execute()
-
-		Expect(err).NotTo(HaveOccurred())
-		Expect(repository.PassedStartingBlockNumber).To(Equal(config.StartingBlockNumber))
-		Expect(repository.PassedEndingBlockNumber).To(Equal(config.EndingBlockNumber))
-	})
-
-	It("returns an error if it fails to get missing headers", func() {
-		repository.SetMissingHeadersError(fakes.FakeError)
-		err := transformer.Execute()
-
-		Expect(err).To(HaveOccurred())
-		Expect(err).To(MatchError(fakes.FakeError))
-	})
-
-	It("fetches eth logs for each missing header", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
-		expectedTopics := [][]common.Hash{{common.HexToHash(config.Topic)}}
-		err := transformer.Execute()
-
-		Expect(err).NotTo(HaveOccurred())
-		Expect(fetcher.FetchedBlocks).To(Equal([]int64{headerOne.BlockNumber, headerTwo.BlockNumber}))
-		Expect(fetcher.FetchedTopics).To(Equal(expectedTopics))
-		Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{config.ContractAddresses, config.ContractAddresses}))
-	})
-
-	It("returns an error if fetching logs fails", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetcherError(fakes.FakeError)
-		err := transformer.Execute()
-
-		Expect(err).To(HaveOccurred())
-		Expect(err).To(MatchError(fakes.FakeError))
-	})
-
 	It("marks header checked if no logs returned", func() {
-		headerID := int64(123)
+		err := transformer.Execute([]types.Log{}, headerOne)
-		repository.SetMissingHeaders([]core.Header{{Id: headerID}})
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
-		repository.AssertMarkHeaderCheckedCalledWith(headerID)
+		repository.AssertMarkHeaderCheckedCalledWith(headerOne.Id)
 	})
 
 	It("doesn't attempt to convert or persist an empty collection when there are no logs", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
+		err := transformer.Execute([]types.Log{}, headerOne)
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(converter.ToEntitiesCalledCounter).To(Equal(0))
@@ -118,29 +71,23 @@ var _ = Describe("Transformer", func() {
 	})
 
 	It("does not call repository.MarkCheckedHeader when there are logs", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
+		err := transformer.Execute(logs, headerOne)
-		fetcher.SetFetchedLogs(logs)
-
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		repository.AssertMarkHeaderCheckedNotCalled()
 	})
 
 	It("returns error if marking header checked returns err", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
 		repository.SetMarkHeaderCheckedError(fakes.FakeError)
 
-		err := transformer.Execute()
+		err := transformer.Execute([]types.Log{}, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))
 	})
 
 	It("converts an eth log to an entity", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
+		err := transformer.Execute(logs, headerOne)
-		fetcher.SetFetchedLogs(logs)
-		err := transformer.Execute()
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(converter.ContractAbi).To(Equal(config.ContractAbi))
@@ -148,45 +95,37 @@ var _ = Describe("Transformer", func() {
 	})
 
 	It("returns an error if converter fails", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetchedLogs(logs)
 		converter.ToEntitiesError = fakes.FakeError
 
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))
 	})
 
 	It("converts an entity to a model", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetchedLogs(logs)
 		converter.EntitiesToReturn = []interface{}{test_data.GenericEntity{}}
 
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(converter.EntitiesToConvert[0]).To(Equal(test_data.GenericEntity{}))
 	})
 
 	It("returns an error if converting to models fails", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetchedLogs(logs)
 		converter.EntitiesToReturn = []interface{}{test_data.GenericEntity{}}
 		converter.ToModelsError = fakes.FakeError
 
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))
 	})
 
 	It("persists the record", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetchedLogs(logs)
 		converter.ModelsToReturn = []interface{}{test_data.GenericModel{}}
 
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(repository.PassedHeaderID).To(Equal(headerOne.Id))
@@ -194,10 +133,8 @@ var _ = Describe("Transformer", func() {
 	})
 
 	It("returns error if persisting the record fails", func() {
-		repository.SetMissingHeaders([]core.Header{headerOne})
-		fetcher.SetFetchedLogs(logs)
 		repository.SetCreateError(fakes.FakeError)
-		err := transformer.Execute()
+		err := transformer.Execute(logs, headerOne)
 
 		Expect(err).To(HaveOccurred())
 		Expect(err).To(MatchError(fakes.FakeError))

pkg/transformers/integration_tests/bite.go

@@ -45,17 +45,24 @@ var _ = Describe("Bite Transformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		initializer := factories.Transformer{
 			Config:     config,
 			Converter:  &bite.BiteConverter{},
 			Repository: &bite.BiteRepository{},
-			Fetcher:    &shared.Fetcher{},
 		}
-		transformer := initializer.NewTransformer(db, blockChain)
+		transformer := initializer.NewTransformer(db)
-		err = transformer.Execute()
+
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			[]common.Address{common.HexToAddress(config.ContractAddresses[0])},
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []bite.BiteModel

pkg/transformers/integration_tests/cat_file.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 	"sort"
@@ -39,6 +40,7 @@ var _ = Describe("Cat File transformer", func() {
 		rpcClient  client.RpcClient
 		err        error
 		ethClient  *ethclient.Client
+		fetcher    shared.Fetcher
 	)
 
 	BeforeEach(func() {
@@ -48,13 +50,15 @@ var _ = Describe("Cat File transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+
+		fetcher = shared.NewFetcher(blockChain)
 	})
 
 	// Cat contract Kovan address: 0x2f34f22a00ee4b7a8f8bbc4eaee1658774c624e0
 	It("persists a chop lump event", func() {
 		// transaction: 0x98574bfba4d05c3875be10d2376e678d005dbebe9a4520363407508fd21f4014
 		chopLumpBlockNumber := int64(8762253)
-		err = persistHeader(db, chopLumpBlockNumber, blockChain)
+		header, err := persistHeader(db, chopLumpBlockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		config := chop_lump.CatFileChopLumpConfig
@@ -65,10 +69,16 @@ var _ = Describe("Cat File transformer", func() {
 			Config:     config,
 			Converter:  &chop_lump.CatFileChopLumpConverter{},
 			Repository: &chop_lump.CatFileChopLumpRepository{},
-			Fetcher:    &shared.Fetcher{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
-		err := transformer.Execute()
+
+		logs, err := fetcher.FetchLogs(
+			[]common.Address{common.HexToAddress(config.ContractAddresses[0])},
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []chop_lump.CatFileChopLumpModel
@@ -92,7 +102,7 @@ var _ = Describe("Cat File transformer", func() {
 	It("persists a flip event", func() {
 		// transaction: 0x44bc18fdb1a5a263db114e7879653304db3e19ceb4e4496f21bc0a76c5faccbe
 		flipBlockNumber := int64(8751794)
-		err = persistHeader(db, flipBlockNumber, blockChain)
+		header, err := persistHeader(db, flipBlockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		config := flip.CatFileFlipConfig
@@ -103,10 +113,17 @@ var _ = Describe("Cat File transformer", func() {
 			Config:     config,
 			Converter:  &flip.CatFileFlipConverter{},
 			Repository: &flip.CatFileFlipRepository{},
-			Fetcher:    &shared.Fetcher{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+
-		err := transformer.Execute()
+		transformer := initializer.NewLogNoteTransformer(db)
+
+		logs, err := fetcher.FetchLogs(
+			[]common.Address{common.HexToAddress(config.ContractAddresses[0])},
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []flip.CatFileFlipModel
@@ -122,7 +139,7 @@ var _ = Describe("Cat File transformer", func() {
 	It("persists a pit vow event", func() {
 		// transaction: 0x44bc18fdb1a5a263db114e7879653304db3e19ceb4e4496f21bc0a76c5faccbe
 		pitVowBlockNumber := int64(8751794)
-		err = persistHeader(db, pitVowBlockNumber, blockChain)
+		header, err := persistHeader(db, pitVowBlockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		config := pit_vow.CatFilePitVowConfig
@@ -133,10 +150,16 @@ var _ = Describe("Cat File transformer", func() {
 			Config:     config,
 			Converter:  &pit_vow.CatFilePitVowConverter{},
 			Repository: &pit_vow.CatFilePitVowRepository{},
-			Fetcher:    &shared.Fetcher{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
-		err := transformer.Execute()
+
+		logs, err := fetcher.FetchLogs(
+			[]common.Address{common.HexToAddress(config.ContractAddresses[0])},
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []pit_vow.CatFilePitVowModel

pkg/transformers/integration_tests/deal.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -29,8 +30,13 @@ import (
 
 var _ = Describe("Deal transformer", func() {
 	var (
-		db         *postgres.DB
+		db          *postgres.DB
-		blockChain core.BlockChain
+		blockChain  core.BlockChain
+		config      shared.TransformerConfig
+		initializer factories.LogNoteTransformer
+		fetcher     shared.Fetcher
+		addresses   []common.Address
+		topics      []common.Hash
 	)
 
 	BeforeEach(func() {
@@ -40,26 +46,35 @@ var _ = Describe("Deal transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+
+		config = deal.DealConfig
+
+		initializer = factories.LogNoteTransformer{
+			Config:     config,
+			Converter:  &deal.DealConverter{},
+			Repository: &deal.DealRepository{},
+		}
+
+		fetcher = shared.NewFetcher(blockChain)
+		addresses = shared.HexStringsToAddresses(config.ContractAddresses)
+		topics = []common.Hash{common.HexToHash(config.Topic)}
+
 	})
 
 	It("persists a flip deal log event", func() {
 		// transaction: 0x05b5eabac2ace136f0f7e0efc61d7d42abe8e8938cc0f04fbf1a6ba545d59e58
 		flipBlockNumber := int64(8958007)
-		err := persistHeader(db, flipBlockNumber, blockChain)
+		header, err := persistHeader(db, flipBlockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		config := deal.DealConfig
+		initializer.Config.StartingBlockNumber = flipBlockNumber
-		config.StartingBlockNumber = flipBlockNumber
+		initializer.Config.EndingBlockNumber = flipBlockNumber
-		config.EndingBlockNumber = flipBlockNumber
 
-		initializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Converter:  &deal.DealConverter{},
+
-			Repository: &deal.DealRepository{},
+		transformer := initializer.NewLogNoteTransformer(db)
-			Fetcher:    &shared.Fetcher{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []deal.DealModel
@@ -77,21 +92,17 @@ var _ = Describe("Deal transformer", func() {
 
 	It("persists a flap deal log event", func() {
 		flapBlockNumber := int64(9004628)
-		err := persistHeader(db, flapBlockNumber, blockChain)
+		header, err := persistHeader(db, flapBlockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		config := deal.DealConfig
+		initializer.Config.StartingBlockNumber = flapBlockNumber
-		config.StartingBlockNumber = flapBlockNumber
+		initializer.Config.EndingBlockNumber = flapBlockNumber
-		config.EndingBlockNumber = flapBlockNumber
 
-		initializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Converter:  &deal.DealConverter{},
+
-			Repository: &deal.DealRepository{},
+		transformer := initializer.NewLogNoteTransformer(db)
-			Fetcher:    &shared.Fetcher{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []deal.DealModel

pkg/transformers/integration_tests/dent.go

@@ -1,6 +1,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/constants"
@@ -15,8 +16,14 @@ import (
 
 var _ = Describe("Dent transformer", func() {
 	var (
-		db         *postgres.DB
+		db          *postgres.DB
-		blockChain core.BlockChain
+		blockChain  core.BlockChain
+		fetcher     shared.Fetcher
+		transformer shared.Transformer
+		config      shared.TransformerConfig
+		addresses   []common.Address
+		topics      []common.Hash
+		initializer factories.LogNoteTransformer
 	)
 
 	BeforeEach(func() {
@@ -26,25 +33,31 @@ var _ = Describe("Dent transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+
+		config = dent.DentConfig
+		addresses = shared.HexStringsToAddresses(config.ContractAddresses)
+		topics = []common.Hash{common.HexToHash(config.Topic)}
+		fetcher = shared.NewFetcher(blockChain)
+
+		initializer = factories.LogNoteTransformer{
+			Config:     config,
+			Converter:  &dent.DentConverter{},
+			Repository: &dent.DentRepository{},
+		}
 	})
 
 	It("persists a flop dent log event", func() {
 		blockNumber := int64(8955613)
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		config := dent.DentConfig
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		initializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Converter:  &dent.DentConverter{},
+
-			Repository: &dent.DentRepository{},
+		err = transformer.Execute(logs, header)
-			Fetcher:    &shared.Fetcher{},
-		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []dent.DentModel

pkg/transformers/integration_tests/drip_drip.go

@@ -15,8 +15,11 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
@@ -25,32 +28,43 @@ import (
 )
 
 var _ = Describe("DripDrip Transformer", func() {
+	var (
+		db         *postgres.DB
+		blockChain core.BlockChain
+	)
+
+	BeforeEach(func() {
+		rpcClient, ethClient, err := getClients(ipc)
+		Expect(err).NotTo(HaveOccurred())
+		blockChain, err = getBlockChain(rpcClient, ethClient)
+		Expect(err).NotTo(HaveOccurred())
+		db = test_config.NewTestDB(blockChain.Node())
+		test_config.CleanTestDB(db)
+	})
 	It("transforms DripDrip log events", func() {
 		blockNumber := int64(8934775)
 		config := drip_drip.DripDripConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		rpcClient, ethClient, err := getClients(ipc)
+		header, err := persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-		blockChain, err := getBlockChain(rpcClient, ethClient)
-		Expect(err).NotTo(HaveOccurred())
-
-		db := test_config.NewTestDB(blockChain.Node())
-		test_config.CleanTestDB(db)
-
-		err = persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		initializer := factories.LogNoteTransformer{
 			Config:     config,
 			Converter:  &drip_drip.DripDripConverter{},
 			Repository: &drip_drip.DripDripRepository{},
-			Fetcher:    &shared.Fetcher{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
 
-		err = transformer.Execute()
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResults []drip_drip.DripDripModel

pkg/transformers/integration_tests/drip_file_vow.go

@@ -15,8 +15,11 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
@@ -25,31 +28,44 @@ import (
 )
 
 var _ = Describe("Drip File Vow LogNoteTransformer", func() {
+	var (
+		db         *postgres.DB
+		blockChain core.BlockChain
+	)
+
+	BeforeEach(func() {
+		rpcClient, ethClient, err := getClients(ipc)
+		Expect(err).NotTo(HaveOccurred())
+		blockChain, err = getBlockChain(rpcClient, ethClient)
+		Expect(err).NotTo(HaveOccurred())
+		db = test_config.NewTestDB(blockChain.Node())
+		test_config.CleanTestDB(db)
+	})
+
 	It("transforms DripFileVow log events", func() {
 		blockNumber := int64(8762197)
 		config := vow.DripFileVowConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		rpcClient, ethClient, err := getClients(ipc)
+		header, err := persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-		blockChain, err := getBlockChain(rpcClient, ethClient)
-		Expect(err).NotTo(HaveOccurred())
-
-		db := test_config.NewTestDB(blockChain.Node())
-		test_config.CleanTestDB(db)
-
-		err = persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		initializer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vow.DripFileVowConverter{},
 			Repository: &vow.DripFileVowRepository{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
-		err = transformer.Execute()
+
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []vow.DripFileVowModel

pkg/transformers/integration_tests/flap_kick.go

@@ -15,6 +15,10 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/test_config"
 	"time"
 
 	. "github.com/onsi/ginkgo"
@@ -23,35 +27,46 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/flap_kick"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
-	"github.com/vulcanize/vulcanizedb/test_config"
 )
 
 var _ = Describe("FlapKick Transformer", func() {
+	var (
+		db         *postgres.DB
+		blockChain core.BlockChain
+	)
+
+	BeforeEach(func() {
+		rpcClient, ethClient, err := getClients(ipc)
+		Expect(err).NotTo(HaveOccurred())
+		blockChain, err = getBlockChain(rpcClient, ethClient)
+		Expect(err).NotTo(HaveOccurred())
+		db = test_config.NewTestDB(blockChain.Node())
+		test_config.CleanTestDB(db)
+	})
+
 	It("fetches and transforms a FlapKick event from Kovan chain", func() {
 		blockNumber := int64(9002933)
 		config := flap_kick.FlapKickConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		rpcClient, ethClient, err := getClients(ipc)
+		header, err := persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-		blockChain, err := getBlockChain(rpcClient, ethClient)
 		Expect(err).NotTo(HaveOccurred())
 
-		db := test_config.NewTestDB(blockChain.Node())
+		transformer := factories.Transformer{
-		test_config.CleanTestDB(db)
-
-		err = persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-
-		initializer := factories.Transformer{
 			Config:     config,
 			Converter:  &flap_kick.FlapKickConverter{},
 			Repository: &flap_kick.FlapKickRepository{},
-			Fetcher:    &shared.Fetcher{},
+		}.NewTransformer(db)
-		}
+
-		transformer := initializer.NewTransformer(db, blockChain)
+		fetcher := shared.NewFetcher(blockChain)
-		err = transformer.Execute()
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []flap_kick.FlapKickModel

pkg/transformers/integration_tests/flip_kick.go

@@ -65,21 +65,26 @@ var _ = Describe("FlipKick Transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		blockChain, err := getBlockChain(rpcClient, ethClient)
 		Expect(err).NotTo(HaveOccurred())
-
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.Transformer{
+		transformer := factories.Transformer{
 			Config:     config,
 			Converter:  &flip_kick.FlipKickConverter{},
 			Repository: &flip_kick.FlipKickRepository{},
-			Fetcher:    &shared.Fetcher{},
+		}.NewTransformer(db)
-		}
+
-		transformer := initializer.NewTransformer(db, blockChain)
+		fetcher := shared.NewFetcher(blockChain)
-		err = transformer.Execute()
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []flip_kick.FlipKickModel

pkg/transformers/integration_tests/flop_kick.go

@@ -35,8 +35,13 @@ import (
 
 var _ = Describe("FlopKick Transformer", func() {
 	var (
-		db         *postgres.DB
+		db          *postgres.DB
-		blockChain core.BlockChain
+		blockChain  core.BlockChain
+		config      shared.TransformerConfig
+		initializer factories.Transformer
+		fetcher     shared.LogFetcher
+		addresses   []common.Address
+		topics      []common.Hash
 	)
 
 	BeforeEach(func() {
@@ -46,25 +51,33 @@ var _ = Describe("FlopKick Transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+
+		config = flop_kick.Config
+
+		initializer = factories.Transformer{
+			Config:     config,
+			Converter:  &flop_kick.FlopKickConverter{},
+			Repository: &flop_kick.FlopKickRepository{},
+		}
+
+		fetcher = shared.NewFetcher(blockChain)
+		addresses = shared.HexStringsToAddresses(config.ContractAddresses)
+		topics = []common.Hash{common.HexToHash(config.Topic)}
 	})
 
 	It("fetches and transforms a FlopKick event from Kovan chain", func() {
 		blockNumber := int64(8672119)
-		config := flop_kick.Config
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.Transformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Converter:  &flop_kick.FlopKickConverter{},
+
-			Repository: &flop_kick.FlopKickRepository{},
+		transformer := initializer.NewTransformer(db)
-			Fetcher:    &shared.Fetcher{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []flop_kick.Model
@@ -82,21 +95,17 @@ var _ = Describe("FlopKick Transformer", func() {
 
 	It("fetches and transforms another FlopKick event from Kovan chain", func() {
 		blockNumber := int64(8955611)
-		config := flop_kick.Config
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.Transformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Converter:  &flop_kick.FlopKickConverter{},
+
-			Repository: &flop_kick.FlopKickRepository{},
+		transformer := initializer.NewTransformer(db)
-			Fetcher:    &shared.Fetcher{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []flop_kick.Model

pkg/transformers/integration_tests/frob.go

@@ -33,8 +33,11 @@ import (
 
 var _ = Describe("Frob Transformer", func() {
 	var (
-		db         *postgres.DB
+		db          *postgres.DB
-		blockChain core.BlockChain
+		blockChain  core.BlockChain
+		fetcher     shared.Fetcher
+		config      shared.TransformerConfig
+		initializer factories.Transformer
 	)
 
 	BeforeEach(func() {
@@ -44,25 +47,32 @@ var _ = Describe("Frob Transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+
+		fetcher = shared.NewFetcher(blockChain)
+		config = frob.FrobConfig
+		initializer = factories.Transformer{
+			Config:     config,
+			Converter:  &frob.FrobConverter{},
+			Repository: &frob.FrobRepository{},
+		}
 	})
 
 	It("fetches and transforms a Frob event from Kovan chain", func() {
 		blockNumber := int64(8935258)
-		config := frob.FrobConfig
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.Transformer{
+		logs, err := fetcher.FetchLogs(
-			Config:     config,
+			shared.HexStringsToAddresses(config.ContractAddresses),
-			Converter:  &frob.FrobConverter{},
+			[]common.Hash{common.HexToHash(config.Topic)},
-			Repository: &frob.FrobRepository{},
+			header)
-			Fetcher:    &shared.Fetcher{},
+		Expect(err).NotTo(HaveOccurred())
-		}
+
-		transformer := initializer.NewTransformer(db, blockChain)
+		transformer := initializer.NewTransformer(db)
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []frob.FrobModel

pkg/transformers/integration_tests/helpers.go

@@ -42,12 +42,13 @@ func getBlockChain(rpcClient client.RpcClient, ethClient *ethclient.Client) (cor
 	return blockChain, nil
 }
 
-func persistHeader(db *postgres.DB, blockNumber int64, blockChain core.BlockChain) error {
+// Persist the header for a given block to postgres. Returns the header if successful.
+func persistHeader(db *postgres.DB, blockNumber int64, blockChain core.BlockChain) (core.Header, error) {
 	header, err := blockChain.GetHeaderByNumber(blockNumber)
 	if err != nil {
-		return err
+		return core.Header{}, err
 	}
 	headerRepository := repositories.NewHeaderRepository(db)
 	_, err = headerRepository.CreateOrUpdateHeader(header)
-	return err
+	return header, err
 }

pkg/transformers/integration_tests/pit_file_debt_ceiling.go

@@ -15,8 +15,11 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
@@ -25,31 +28,44 @@ import (
 )
 
 var _ = Describe("PitFileDebtCeiling LogNoteTransformer", func() {
+	var (
+		db         *postgres.DB
+		blockChain core.BlockChain
+	)
+
+	BeforeEach(func() {
+		rpcClient, ethClient, err := getClients(ipc)
+		Expect(err).NotTo(HaveOccurred())
+		blockChain, err = getBlockChain(rpcClient, ethClient)
+		Expect(err).NotTo(HaveOccurred())
+		db = test_config.NewTestDB(blockChain.Node())
+		test_config.CleanTestDB(db)
+	})
+
 	It("fetches and transforms a PitFileDebtCeiling event from Kovan chain", func() {
 		blockNumber := int64(8535578)
 		config := debt_ceiling.DebtCeilingFileConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		rpcClient, ethClient, err := getClients(ipc)
+		header, err := persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-		blockChain, err := getBlockChain(rpcClient, ethClient)
 		Expect(err).NotTo(HaveOccurred())
 
-		db := test_config.NewTestDB(blockChain.Node())
+		fetcher := shared.NewFetcher(blockChain)
-		test_config.CleanTestDB(db)
+		logs, err := fetcher.FetchLogs(
-
+			shared.HexStringsToAddresses(config.ContractAddresses),
-		err = persistHeader(db, blockNumber, blockChain)
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
 		Expect(err).NotTo(HaveOccurred())
 
 		initializer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &debt_ceiling.PitFileDebtCeilingConverter{},
 			Repository: &debt_ceiling.PitFileDebtCeilingRepository{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
-		err = transformer.Execute()
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []debt_ceiling.PitFileDebtCeilingModel

pkg/transformers/integration_tests/pit_file_ilk.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
@@ -39,17 +40,24 @@ var _ = Describe("PitFileIlk LogNoteTransformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
 		initializer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &ilk.PitFileIlkConverter{},
 			Repository: &ilk.PitFileIlkRepository{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
-		err = transformer.Execute()
+
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []ilk.PitFileIlkModel

pkg/transformers/integration_tests/price_feeds.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -29,8 +30,11 @@ import (
 
 var _ = Describe("Price feeds transformer", func() {
 	var (
-		db         *postgres.DB
+		db          *postgres.DB
-		blockChain core.BlockChain
+		blockChain  core.BlockChain
+		config      shared.TransformerConfig
+		fetcher     shared.Fetcher
+		initializer factories.LogNoteTransformer
 	)
 
 	BeforeEach(func() {
@@ -40,30 +44,36 @@ var _ = Describe("Price feeds transformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+		config = price_feeds.PriceFeedConfig
+		fetcher = shared.NewFetcher(blockChain)
+
+		initializer = factories.LogNoteTransformer{
+			Config:     config,
+			Converter:  &price_feeds.PriceFeedConverter{},
+			Repository: &price_feeds.PriceFeedRepository{},
+		}
 	})
 
 	It("persists a ETH/USD price feed event", func() {
 		blockNumber := int64(8763054)
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
-		config := price_feeds.PriceFeedConfig
+		initializer.Config.ContractAddresses = []string{constants.PipContractAddress}
-		config.ContractAddresses = []string{constants.PipContractAddress}
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		transformerInitializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(
-			Config:     config,
+			shared.HexStringsToAddresses(initializer.Config.ContractAddresses),
-			Converter:  &price_feeds.PriceFeedConverter{},
+			[]common.Hash{common.HexToHash(config.Topic)},
-			Repository: &price_feeds.PriceFeedRepository{},
+			header)
-			Fetcher:    &shared.Fetcher{},
+		Expect(err).NotTo(HaveOccurred())
-		}
-		transformer := transformerInitializer.NewLogNoteTransformer(db, blockChain)
 
-		err = transformer.Execute()
+		transformer := initializer.NewLogNoteTransformer(db)
+		err = transformer.Execute(logs, header)
 
 		Expect(err).NotTo(HaveOccurred())
 		var model price_feeds.PriceFeedModel
-		err = db.Get(&model, `SELECT block_number, medianizer_address, usd_value, tx_idx, raw_log FROM maker.price_feeds WHERE block_number = $1`, config.StartingBlockNumber)
+		err = db.Get(&model, `SELECT block_number, medianizer_address, usd_value, tx_idx, raw_log FROM maker.price_feeds WHERE block_number = $1`, initializer.Config.StartingBlockNumber)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(model.UsdValue).To(Equal("207.314891143000011198"))
 		Expect(model.MedianizerAddress).To(Equal(config.ContractAddresses[0]))
@@ -71,26 +81,24 @@ var _ = Describe("Price feeds transformer", func() {
 
 	It("persists a MKR/USD price feed event", func() {
 		blockNumber := int64(8763059)
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
-		config := price_feeds.PriceFeedConfig
+		initializer.Config.ContractAddresses = []string{constants.PepContractAddress}
-		config.ContractAddresses = []string{constants.PepContractAddress}
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		transformerInitializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(
-			Config:     config,
+			shared.HexStringsToAddresses(initializer.Config.ContractAddresses),
-			Converter:  &price_feeds.PriceFeedConverter{},
+			[]common.Hash{common.HexToHash(config.Topic)},
-			Repository: &price_feeds.PriceFeedRepository{},
+			header)
-			Fetcher:    &shared.Fetcher{},
+		Expect(err).NotTo(HaveOccurred())
-		}
-		transformer := transformerInitializer.NewLogNoteTransformer(db, blockChain)
 
-		err = transformer.Execute()
+		transformer := initializer.NewLogNoteTransformer(db)
+		err = transformer.Execute(logs, header)
 
 		Expect(err).NotTo(HaveOccurred())
 		var model price_feeds.PriceFeedModel
-		err = db.Get(&model, `SELECT block_number, medianizer_address, usd_value, tx_idx, raw_log FROM maker.price_feeds WHERE block_number = $1`, config.StartingBlockNumber)
+		err = db.Get(&model, `SELECT block_number, medianizer_address, usd_value, tx_idx, raw_log FROM maker.price_feeds WHERE block_number = $1`, initializer.Config.StartingBlockNumber)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(model.UsdValue).To(Equal("391.803979212000001553"))
 		Expect(model.MedianizerAddress).To(Equal(config.ContractAddresses[0]))
@@ -98,26 +106,24 @@ var _ = Describe("Price feeds transformer", func() {
 
 	It("persists a REP/USD price feed event", func() {
 		blockNumber := int64(8763062)
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
-		config := price_feeds.PriceFeedConfig
+		initializer.Config.ContractAddresses = []string{constants.RepContractAddress}
-		config.ContractAddresses = []string{constants.RepContractAddress}
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		transformerInitializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(
-			Config:     config,
+			shared.HexStringsToAddresses(config.ContractAddresses),
-			Converter:  &price_feeds.PriceFeedConverter{},
+			[]common.Hash{common.HexToHash(config.Topic)},
-			Repository: &price_feeds.PriceFeedRepository{},
+			header)
-			Fetcher:    &shared.Fetcher{},
+		Expect(err).NotTo(HaveOccurred())
-		}
-		transformer := transformerInitializer.NewLogNoteTransformer(db, blockChain)
 
-		err = transformer.Execute()
+		transformer := initializer.NewLogNoteTransformer(db)
+		err = transformer.Execute(logs, header)
 
 		Expect(err).NotTo(HaveOccurred())
 		var model price_feeds.PriceFeedModel
-		err = db.Get(&model, `SELECT block_number, medianizer_address, usd_value, tx_idx, raw_log FROM maker.price_feeds WHERE block_number = $1`, config.StartingBlockNumber)
+		err = db.Get(&model, `SELECT block_number, medianizer_address, usd_value, tx_idx, raw_log FROM maker.price_feeds WHERE block_number = $1`, initializer.Config.StartingBlockNumber)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(model.UsdValue).To(Equal("12.816928482699999847"))
 		Expect(model.MedianizerAddress).To(Equal(config.ContractAddresses[0]))

pkg/transformers/integration_tests/tend.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
@@ -28,8 +29,13 @@ import (
 
 var _ = Describe("Tend LogNoteTransformer", func() {
 	var (
-		db         *postgres.DB
+		db          *postgres.DB
-		blockChain core.BlockChain
+		blockChain  core.BlockChain
+		config      shared.TransformerConfig
+		fetcher     shared.Fetcher
+		initializer factories.LogNoteTransformer
+		addresses   []common.Address
+		topics      []common.Hash
 	)
 
 	BeforeEach(func() {
@@ -39,25 +45,31 @@ var _ = Describe("Tend LogNoteTransformer", func() {
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+
+		fetcher = shared.NewFetcher(blockChain)
+		addresses = shared.HexStringsToAddresses(config.ContractAddresses)
+		topics = []common.Hash{common.HexToHash(config.Topic)}
+
+		initializer = factories.LogNoteTransformer{
+			Config:     tend.TendConfig,
+			Converter:  &tend.TendConverter{},
+			Repository: &tend.TendRepository{},
+		}
 	})
 
 	It("fetches and transforms a Flip Tend event from Kovan chain", func() {
 		blockNumber := int64(8935601)
-		config := tend.TendConfig
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Fetcher:    &shared.Fetcher{},
+
-			Converter:  &tend.TendConverter{},
+		transformer := initializer.NewLogNoteTransformer(db)
-			Repository: &tend.TendRepository{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []tend.TendModel
@@ -80,21 +92,17 @@ var _ = Describe("Tend LogNoteTransformer", func() {
 
 	It("fetches and transforms a subsequent Flip Tend event from Kovan chain for the same auction", func() {
 		blockNumber := int64(8935731)
-		config := tend.TendConfig
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Fetcher:    &shared.Fetcher{},
+
-			Converter:  &tend.TendConverter{},
+		transformer := initializer.NewLogNoteTransformer(db)
-			Repository: &tend.TendRepository{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []tend.TendModel
@@ -117,21 +125,17 @@ var _ = Describe("Tend LogNoteTransformer", func() {
 
 	It("fetches and transforms a Flap Tend event from the Kovan chain", func() {
 		blockNumber := int64(9003177)
-		config := tend.TendConfig
+		initializer.Config.StartingBlockNumber = blockNumber
-		config.StartingBlockNumber = blockNumber
+		initializer.Config.EndingBlockNumber = blockNumber
-		config.EndingBlockNumber = blockNumber
 
-		err := persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		logs, err := fetcher.FetchLogs(addresses, topics, header)
-			Config:     config,
+		Expect(err).NotTo(HaveOccurred())
-			Fetcher:    &shared.Fetcher{},
+
-			Converter:  &tend.TendConverter{},
+		transformer := initializer.NewLogNoteTransformer(db)
-			Repository: &tend.TendRepository{},
+		err = transformer.Execute(logs, header)
-		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
-		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []tend.TendModel

pkg/transformers/integration_tests/vat_flux.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -39,17 +40,24 @@ var _ = Describe("VatFlux LogNoteTransformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
+		Expect(err).NotTo(HaveOccurred())
+
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
 		Expect(err).NotTo(HaveOccurred())
 
 		initializer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vat_flux.VatFluxConverter{},
 			Repository: &vat_flux.VatFluxRepository{},
 		}
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+		transformer := initializer.NewLogNoteTransformer(db)
-		err = transformer.Execute()
+
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []vat_flux.VatFluxModel

pkg/transformers/integration_tests/vat_fold.go

@@ -18,39 +18,53 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/test_config"
 
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_fold"
-	"github.com/vulcanize/vulcanizedb/test_config"
 )
 
 var _ = Describe("VatFold Transformer", func() {
+	var (
+		db         *postgres.DB
+		blockChain core.BlockChain
+	)
+
+	BeforeEach(func() {
+		rpcClient, ethClient, err := getClients(ipc)
+		Expect(err).NotTo(HaveOccurred())
+		blockChain, err = getBlockChain(rpcClient, ethClient)
+		Expect(err).NotTo(HaveOccurred())
+		db = test_config.NewTestDB(blockChain.Node())
+		test_config.CleanTestDB(db)
+	})
+
 	It("transforms VatFold log events", func() {
 		blockNumber := int64(9051149)
 		config := vat_fold.VatFoldConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		rpcClient, ethClient, err := getClients(ipc)
+		header, err := persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-		blockChain, err := getBlockChain(rpcClient, ethClient)
 		Expect(err).NotTo(HaveOccurred())
 
-		db := test_config.NewTestDB(blockChain.Node())
+		fetcher := shared.NewFetcher(blockChain)
-		test_config.CleanTestDB(db)
+		logs, err := fetcher.FetchLogs(
-
+			shared.HexStringsToAddresses(config.ContractAddresses),
-		err = persistHeader(db, blockNumber, blockChain)
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		transformer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vat_fold.VatFoldConverter{},
 			Repository: &vat_fold.VatFoldRepository{},
-		}
+		}.NewLogNoteTransformer(db)
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResults []vat_fold.VatFoldModel

pkg/transformers/integration_tests/vat_grab.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	"math/big"
 
 	. "github.com/onsi/ginkgo"
@@ -41,17 +42,23 @@ var _ = Describe("Vat Grab Transformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
+		Expect(err).NotTo(HaveOccurred())
+
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
 		Expect(err).NotTo(HaveOccurred())
 
 		transformer := factories.LogNoteTransformer{
 			Config:     config,
 			Converter:  &vat_grab.VatGrabConverter{},
 			Repository: &vat_grab.VatGrabRepository{},
-			Fetcher:    &shared.Fetcher{},
+		}.NewLogNoteTransformer(db)
-		}.NewLogNoteTransformer(db, blockChain)
 
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []vat_grab.VatGrabModel

pkg/transformers/integration_tests/vat_heal.go

@@ -40,17 +40,23 @@ var _ = Describe("VatHeal Transformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
+		Expect(err).NotTo(HaveOccurred())
+
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
 		Expect(err).NotTo(HaveOccurred())
 
 		transformer := factories.LogNoteTransformer{
 			Config:     config,
 			Converter:  &vat_heal.VatHealConverter{},
 			Repository: &vat_heal.VatHealRepository{},
-			Fetcher:    &shared.Fetcher{},
+		}.NewLogNoteTransformer(db)
-		}.NewLogNoteTransformer(db, blockChain)
 
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResults []vat_heal.VatHealModel

pkg/transformers/integration_tests/vat_init.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
@@ -39,17 +40,23 @@ var _ = Describe("VatInit LogNoteTransformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		transformer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vat_init.VatInitConverter{},
 			Repository: &vat_init.VatInitRepository{},
-		}
+		}.NewLogNoteTransformer(db)
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResults []vat_init.VatInitModel

pkg/transformers/integration_tests/vat_move.go

@@ -40,17 +40,23 @@ var _ = Describe("VatMove LogNoteTransformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		transformer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vat_move.VatMoveConverter{},
 			Repository: &vat_move.VatMoveRepository{},
-		}
+		}.NewLogNoteTransformer(db)
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResults []vat_move.VatMoveModel

pkg/transformers/integration_tests/vat_slip.go

@@ -1,6 +1,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -18,31 +19,38 @@ var _ = Describe("Vat slip transformer", func() {
 		blockChain core.BlockChain
 	)
 
-	It("persists vat slip event", func() {
+	BeforeEach(func() {
 		rpcClient, ethClient, err := getClients(ipc)
 		Expect(err).NotTo(HaveOccurred())
 		blockChain, err = getBlockChain(rpcClient, ethClient)
 		Expect(err).NotTo(HaveOccurred())
 		db = test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
+	})
 
+	It("persists vat slip event", func() {
 		blockNumber := int64(8953655)
 		config := vat_slip.VatSlipConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		transformer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vat_slip.VatSlipConverter{},
 			Repository: &vat_slip.VatSlipRepository{},
-		}
+		}.NewLogNoteTransformer(db)
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
 
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 
 		Expect(err).NotTo(HaveOccurred())
 		var headerID int64

pkg/transformers/integration_tests/vat_tune.go

@@ -15,6 +15,7 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	"math/big"
 
 	. "github.com/onsi/ginkgo"
@@ -41,17 +42,23 @@ var _ = Describe("VatTune LogNoteTransformer", func() {
 		db := test_config.NewTestDB(blockChain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber, blockChain)
+		header, err := persistHeader(db, blockNumber, blockChain)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := factories.LogNoteTransformer{
+		fetcher := shared.NewFetcher(blockChain)
+		logs, err := fetcher.FetchLogs(
+			shared.HexStringsToAddresses(config.ContractAddresses),
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
+		Expect(err).NotTo(HaveOccurred())
+
+		transformer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vat_tune.VatTuneConverter{},
 			Repository: &vat_tune.VatTuneRepository{},
-		}
+		}.NewLogNoteTransformer(db)
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []vat_tune.VatTuneModel

pkg/transformers/integration_tests/vow_flog.go

@@ -15,8 +15,11 @@
 package integration_tests
 
 import (
+	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
@@ -25,32 +28,43 @@ import (
 )
 
 var _ = Describe("VowFlog LogNoteTransformer", func() {
+	var (
+		db         *postgres.DB
+		blockChain core.BlockChain
+	)
+
+	BeforeEach(func() {
+		rpcClient, ethClient, err := getClients(ipc)
+		Expect(err).NotTo(HaveOccurred())
+		blockChain, err = getBlockChain(rpcClient, ethClient)
+		Expect(err).NotTo(HaveOccurred())
+		db = test_config.NewTestDB(blockChain.Node())
+		test_config.CleanTestDB(db)
+	})
+
 	It("transforms VowFlog log events", func() {
 		blockNumber := int64(8946819)
 		config := vow_flog.VowFlogConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
-		rpcClient, ethClient, err := getClients(ipc)
+		header, err := persistHeader(db, blockNumber, blockChain)
-		Expect(err).NotTo(HaveOccurred())
-		blockChain, err := getBlockChain(rpcClient, ethClient)
 		Expect(err).NotTo(HaveOccurred())
 
-		db := test_config.NewTestDB(blockChain.Node())
+		fetcher := shared.NewFetcher(blockChain)
-		test_config.CleanTestDB(db)
+		logs, err := fetcher.FetchLogs(
-
+			shared.HexStringsToAddresses(config.ContractAddresses),
-		err = persistHeader(db, blockNumber, blockChain)
+			[]common.Hash{common.HexToHash(config.Topic)},
+			header)
 		Expect(err).NotTo(HaveOccurred())
-		Expect(1).To(Equal(1))
 
-		initializer := factories.LogNoteTransformer{
+		transformer := factories.LogNoteTransformer{
 			Config:     config,
-			Fetcher:    &shared.Fetcher{},
 			Converter:  &vow_flog.VowFlogConverter{},
 			Repository: &vow_flog.VowFlogRepository{},
-		}
+		}.NewLogNoteTransformer(db)
-		transformer := initializer.NewLogNoteTransformer(db, blockChain)
+
-		err = transformer.Execute()
+		err = transformer.Execute(logs, header)
 		Expect(err).NotTo(HaveOccurred())
 
 		var dbResult []vow_flog.VowFlogModel

pkg/transformers/shared/log_chunker.go

@@ -27,7 +27,7 @@ type LogChunker struct {
 // Initialises a chunker by creating efficient lookup maps
 func NewLogChunker(transformerConfigs []TransformerConfig) LogChunker {
 	addressToNames := map[string][]string{}
-	nameToTopic0   := map[string]common.Hash{}
+	nameToTopic0 := map[string]common.Hash{}
 
 	for _, config := range transformerConfigs {
 		for _, address := range config.ContractAddresses {

pkg/transformers/shared/repository_utility_test.go

@@ -87,4 +87,4 @@ func getExpectedColumnNames() []string {
 		"vow_flog_checked",
 		"flap_kick_checked",
 	}
-}
+}

pkg/transformers/shared/transformer.go

@@ -47,3 +47,10 @@ func HexToString(byteString string) string {
 	value := common.HexToHash(byteString)
 	return value.Big().String()
 }
+
+func HexStringsToAddresses(strings []string) (addresses []common.Address) {
+	for _, hexString := range strings {
+		addresses = append(addresses, common.HexToAddress(hexString))
+	}
+	return
+}

pkg/transformers/test_data/mocks/repository.go

@@ -3,7 +3,6 @@ package mocks
 import (
 	. "github.com/onsi/gomega"
 
-	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
 
@@ -12,8 +11,6 @@ type MockRepository struct {
 	markHeaderCheckedError           error
 	MarkHeaderCheckedPassedHeaderIDs []int64
 	CreatedHeaderIds                 []int64
-	missingHeaders                   []core.Header
-	missingHeadersError              error
 	PassedStartingBlockNumber        int64
 	PassedEndingBlockNumber          int64
 	PassedHeaderID                   int64
@@ -36,24 +33,10 @@ func (repository *MockRepository) MarkHeaderChecked(headerID int64) error {
 	return repository.markHeaderCheckedError
 }
 
-func (repository *MockRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error) {
-	repository.PassedStartingBlockNumber = startingBlockNumber
-	repository.PassedEndingBlockNumber = endingBlockNumber
-	return repository.missingHeaders, repository.missingHeadersError
-}
-
 func (repository *MockRepository) SetDB(db *postgres.DB) {
 	repository.SetDbCalled = true
 }
 
-func (repository *MockRepository) SetMissingHeadersError(e error) {
-	repository.missingHeadersError = e
-}
-
-func (repository *MockRepository) SetMissingHeaders(headers []core.Header) {
-	repository.missingHeaders = headers
-}
-
 func (repository *MockRepository) SetMarkHeaderCheckedError(e error) {
 	repository.markHeaderCheckedError = e
 }

pkg/transformers/transformers.go

@@ -79,7 +79,7 @@ var (
 		Repository: &flop_kick.FlopKickRepository{},
 	}
 
-	customEventTransformers = []factories.Transformer {
+	customEventTransformers = []factories.Transformer{
 		BiteTransformer,
 		FlapKickTransformer,
 		FlipKickTransformer,
@@ -228,7 +228,7 @@ var (
 		Repository: &vat_flux.VatFluxRepository{},
 	}
 
-	logNoteTransformers = []factories.LogNoteTransformer {
+	logNoteTransformers = []factories.LogNoteTransformer{
 		CatFileChopLumpTransformer,
 		CatFileFlipTransformer,
 		CatFilePitVowTransformer,