cerc-io/ipld-eth-server
18
0
Fork 1
Code Issues 10 Pull Requests 1 Actions Packages Projects Releases 37 Wiki Activity

Use transformer factory for Vat toll

Browse Source
This commit is contained in:
Rob Mulholand 2018-10-24 19:21:14 -05:00
parent 6d662a6ce7
commit 1e7ee08af5
9 changed files with 101 additions and 196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pkg/transformers/test_data/mocks/vat_toll/converter.go
View File

@ -3,7 +3,6 @@ package vat_toll
import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_toll"
)
type MockVatTollConverter struct {
@ -11,9 +10,9 @@ type MockVatTollConverter struct {
PassedLogs []types.Log
}
func (converter *MockVatTollConverter) ToModels(ethLogs []types.Log) ([]vat_toll.VatTollModel, error) {
func (converter *MockVatTollConverter) ToModels(ethLogs []types.Log) ([]interface{}, error) {
converter.PassedLogs = ethLogs
return []vat_toll.VatTollModel{test_data.VatTollModel}, converter.err
return []interface{}{test_data.VatTollModel}, converter.err
}
func (converter *MockVatTollConverter) SetConverterError(e error) {

11
pkg/transformers/test_data/mocks/vat_toll/repository.go
View File

@ -4,7 +4,7 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_toll"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type MockVatTollRepository struct {
@ -16,10 +16,11 @@ type MockVatTollRepository struct {
PassedStartingBlockNumber int64
PassedEndingBlockNumber int64
PassedHeaderID int64
PassedModels []vat_toll.VatTollModel
PassedModels []interface{}
SetDbCalled bool
}
func (repository *MockVatTollRepository) Create(headerID int64, models []vat_toll.VatTollModel) error {
func (repository *MockVatTollRepository) Create(headerID int64, models []interface{}) error {
repository.PassedHeaderID = headerID
repository.PassedModels = models
return repository.createErr
@ -36,6 +37,10 @@ func (repository *MockVatTollRepository) MissingHeaders(startingBlockNumber, end
return repository.missingHeaders, repository.missingHeadersErr
}
func (repository *MockVatTollRepository) SetDB(db *postgres.DB) {
repository.SetDbCalled = true
}
func (repository *MockVatTollRepository) SetCreateError(e error) {
repository.createErr = e
}

7
pkg/transformers/transformers.go
View File

@ -133,7 +133,12 @@ var (
Fetcher: &shared.Fetcher{},
}.NewTransformer
VatSlipTransformerInitializer = vat_slip.VatSlipTransformerInitializer{Config: vat_slip.VatSlipConfig}.NewVatSlipTransformer
VatTollTransformerInitializer = vat_toll.VatTollTransformerInitializer{Config: vat_toll.VatTollConfig}.NewVatTollTransformer
VatTollTransformerInitializer = factories.Transformer{
Config: vat_toll.VatTollConfig,
Converter: &vat_toll.VatTollConverter{},
Repository: &vat_toll.VatTollRepository{},
Fetcher: &shared.Fetcher{},
}.NewTransformer
VatTuneTransformerInitializer = factories.Transformer{
Config: vat_tune.VatTuneConfig,
Converter: &vat_tune.VatTuneConverter{},

5
pkg/transformers/vat_toll/config.go
View File

@ -2,10 +2,11 @@ package vat_toll
import "github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
var VatTollConfig = shared.TransformerConfig{
var VatTollConfig = shared.SingleTransformerConfig{
TransformerName: shared.VatTollLabel,
ContractAddresses: []string{shared.VatContractAddress},
ContractAbi: shared.VatABI,
Topics: []string{shared.VatTollSignature},
Topic: shared.VatTollSignature,
StartingBlockNumber: 0,
EndingBlockNumber: 10000000,
}

8
pkg/transformers/vat_toll/converter.go
View File

@ -9,14 +9,10 @@ import (
"github.com/ethereum/go-ethereum/core/types"
)
type Converter interface {
ToModels(ethLogs []types.Log) ([]VatTollModel, error)
}
type VatTollConverter struct{}
func (VatTollConverter) ToModels(ethLogs []types.Log) ([]VatTollModel, error) {
var models []VatTollModel
func (VatTollConverter) ToModels(ethLogs []types.Log) ([]interface{}, error) {
var models []interface{}
for _, ethLog := range ethLogs {
err := verifyLog(ethLog)
if err != nil {

26
pkg/transformers/vat_toll/repository.go
View File

@ -1,36 +1,30 @@
package vat_toll
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type Repository interface {
Create(headerID int64, models []VatTollModel) error
MarkHeaderChecked(headerID int64) error
MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error)
}
type VatTollRepository struct {
db *postgres.DB
}
func NewVatTollRepository(db *postgres.DB) VatTollRepository {
return VatTollRepository{
db: db,
}
}
func (repository VatTollRepository) Create(headerID int64, models []VatTollModel) error {
func (repository VatTollRepository) Create(headerID int64, models []interface{}) error {
tx, err := repository.db.Begin()
if err != nil {
return err
}
for _, model := range models {
vatToll, ok := model.(VatTollModel)
if !ok {
tx.Rollback()
return fmt.Errorf("model of type %T, not %T", model, VatTollModel{})
}
_, err = tx.Exec(
`INSERT into maker.vat_toll (header_id, ilk, urn, take, tx_idx, log_idx, raw_log)
VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,
headerID, model.Ilk, model.Urn, model.Take, model.TransactionIndex, model.LogIndex, model.Raw,
headerID, vatToll.Ilk, vatToll.Urn, vatToll.Take, vatToll.TransactionIndex, vatToll.LogIndex, vatToll.Raw,
)
if err != nil {
tx.Rollback()
@ -72,3 +66,7 @@ func (repository VatTollRepository) MissingHeaders(startingBlockNumber, endingBl
)
return result, err
}
func (repository *VatTollRepository) SetDB(db *postgres.DB) {
repository.db = db
}

32
pkg/transformers/vat_toll/repository_test.go
View File

@ -14,12 +14,13 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_toll"
"github.com/vulcanize/vulcanizedb/test_config"
"math/rand"
)
var _ = Describe("Vat toll repository", func() {
var (
db *postgres.DB
vatTollRepository vat_toll.Repository
vatTollRepository vat_toll.VatTollRepository
err error
headerRepository datastore.HeaderRepository
)
@ -28,7 +29,8 @@ var _ = Describe("Vat toll repository", func() {
db = test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
headerRepository = repositories.NewHeaderRepository(db)
vatTollRepository = vat_toll.NewVatTollRepository(db)
vatTollRepository = vat_toll.VatTollRepository{}
vatTollRepository.SetDB(db)
})
Describe("Create", func() {
@ -40,7 +42,7 @@ var _ = Describe("Vat toll repository", func() {
})
It("adds a vat toll event", func() {
err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
err = vatTollRepository.Create(headerID, []interface{}{test_data.VatTollModel})
Expect(err).NotTo(HaveOccurred())
var dbVatToll vat_toll.VatTollModel
@ -55,7 +57,7 @@ var _ = Describe("Vat toll repository", func() {
})
It("marks header as checked for logs", func() {
err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
err = vatTollRepository.Create(headerID, []interface{}{test_data.VatTollModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
@ -67,7 +69,7 @@ var _ = Describe("Vat toll repository", func() {
It("updates the header to checked if checked headers row already exists", func() {
_, err := db.Exec(`INSERT INTO public.checked_headers (header_id) VALUES ($1)`, headerID)
Expect(err).NotTo(HaveOccurred())
err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
err = vatTollRepository.Create(headerID, []interface{}{test_data.VatTollModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
@ -77,21 +79,21 @@ var _ = Describe("Vat toll repository", func() {
})
It("does not duplicate vat toll events", func() {
err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
err = vatTollRepository.Create(headerID, []interface{}{test_data.VatTollModel})
Expect(err).NotTo(HaveOccurred())
err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
err = vatTollRepository.Create(headerID, []interface{}{test_data.VatTollModel})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
})
It("allows for multiple vat toll events in one transaction if they have different log indexes", func() {
err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
err = vatTollRepository.Create(headerID, []interface{}{test_data.VatTollModel})
Expect(err).NotTo(HaveOccurred())
newVatToll := test_data.VatTollModel
newVatToll.LogIndex = newVatToll.LogIndex + 1
err := vatTollRepository.Create(headerID, []vat_toll.VatTollModel{newVatToll})
err := vatTollRepository.Create(headerID, []interface{}{newVatToll})
Expect(err).NotTo(HaveOccurred())
})
@ -105,6 +107,13 @@ var _ = Describe("Vat toll repository", func() {
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(sql.ErrNoRows))
})
It("returns an error if model is of wrong type", func() {
err = vatTollRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("model of type"))
})
})
Describe("MarkHeaderChecked", func() {
@ -145,7 +154,7 @@ var _ = Describe("Vat toll repository", func() {
)
BeforeEach(func() {
startingBlock = GinkgoRandomSeed()
startingBlock = rand.Int63()
vatTollBlock = startingBlock + 1
endingBlock = startingBlock + 2
@ -191,7 +200,8 @@ var _ = Describe("Vat toll repository", func() {
_, err = headerRepositoryTwo.CreateOrUpdateHeader(fakes.GetFakeHeader(n))
Expect(err).NotTo(HaveOccurred())
}
vatTollRepositoryTwo := vat_toll.NewVatTollRepository(dbTwo)
vatTollRepositoryTwo := vat_toll.VatTollRepository{}
vatTollRepositoryTwo.SetDB(dbTwo)
err := vatTollRepository.MarkHeaderChecked(headerIDs[0])
Expect(err).NotTo(HaveOccurred())

62
pkg/transformers/vat_toll/transformer.go
View File

@ -1,62 +0,0 @@
package vat_toll
import (
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
"log"
)
type VatTollTransformer struct {
Config shared.TransformerConfig
Converter Converter
Fetcher shared.LogFetcher
Repository Repository
}
type VatTollTransformerInitializer struct {
Config shared.TransformerConfig
}
func (initializer VatTollTransformerInitializer) NewVatTollTransformer(db *postgres.DB, blockChain core.BlockChain) shared.Transformer {
converter := VatTollConverter{}
fetcher := shared.NewFetcher(blockChain)
repository := NewVatTollRepository(db)
return VatTollTransformer{
Config: initializer.Config,
Converter: converter,
Fetcher: fetcher,
Repository: repository,
}
}
func (transformer VatTollTransformer) Execute() error {
missingHeaders, err := transformer.Repository.MissingHeaders(transformer.Config.StartingBlockNumber, transformer.Config.EndingBlockNumber)
if err != nil {
return err
}
log.Printf("Fetching vat toll event logs for %d headers \n", len(missingHeaders))
for _, header := range missingHeaders {
topics := [][]common.Hash{{common.HexToHash(shared.VatTollSignature)}}
matchingLogs, err := transformer.Fetcher.FetchLogs(VatTollConfig.ContractAddresses, topics, header.BlockNumber)
if err != nil {
return err
}
if len(matchingLogs) < 1 {
err = transformer.Repository.MarkHeaderChecked(header.Id)
if err != nil {
return err
}
}
models, err := transformer.Converter.ToModels(matchingLogs)
if err != nil {
return err
}
err = transformer.Repository.Create(header.Id, models)
if err != nil {
return err
}
}
return nil
}

141
pkg/transformers/vat_toll/transformer_test.go
View File

@ -8,38 +8,55 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks"
vat_toll_mocks "github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks/vat_toll"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_toll"
"math/rand"
)
var _ = Describe("Vat toll transformer", func() {
It("gets missing headers for block numbers specified in config", func() {
repository := &vat_toll_mocks.MockVatTollRepository{}
transformer := vat_toll.VatTollTransformer{
Config: vat_toll.VatTollConfig,
Fetcher: &mocks.MockLogFetcher{},
Converter: &vat_toll_mocks.MockVatTollConverter{},
Repository: repository,
}
var (
config = vat_toll.VatTollConfig
fetcher mocks.MockLogFetcher
converter vat_toll_mocks.MockVatTollConverter
repository vat_toll_mocks.MockVatTollRepository
transformer shared.Transformer
headerOne core.Header
headerTwo core.Header
)
BeforeEach(func() {
fetcher = mocks.MockLogFetcher{}
converter = vat_toll_mocks.MockVatTollConverter{}
repository = vat_toll_mocks.MockVatTollRepository{}
headerOne = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
headerTwo = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
transformer = factories.Transformer{
Config: config,
Converter: &converter,
Fetcher: &fetcher,
Repository: &repository,
}.NewTransformer(nil, nil)
})
It("sets the blockchain and database", func() {
Expect(fetcher.SetBcCalled).To(BeTrue())
Expect(repository.SetDbCalled).To(BeTrue())
})
It("gets missing headers for block numbers specified in config", func() {
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(repository.PassedStartingBlockNumber).To(Equal(vat_toll.VatTollConfig.StartingBlockNumber))
Expect(repository.PassedEndingBlockNumber).To(Equal(vat_toll.VatTollConfig.EndingBlockNumber))
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 := &vat_toll_mocks.MockVatTollRepository{}
repository.SetMissingHeadersErr(fakes.FakeError)
transformer := vat_toll.VatTollTransformer{
Fetcher: &mocks.MockLogFetcher{},
Converter: &vat_toll_mocks.MockVatTollConverter{},
Repository: repository,
}
err := transformer.Execute()
@ -48,33 +65,19 @@ var _ = Describe("Vat toll transformer", func() {
})
It("fetches logs for missing headers", func() {
fetcher := &mocks.MockLogFetcher{}
repository := &vat_toll_mocks.MockVatTollRepository{}
repository.SetMissingHeaders([]core.Header{{BlockNumber: 1}, {BlockNumber: 2}})
transformer := vat_toll.VatTollTransformer{
Fetcher: fetcher,
Converter: &vat_toll_mocks.MockVatTollConverter{},
Repository: repository,
}
repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(fetcher.FetchedBlocks).To(Equal([]int64{1, 2}))
Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{vat_toll.VatTollConfig.ContractAddresses, vat_toll.VatTollConfig.ContractAddresses}))
Expect(fetcher.FetchedBlocks).To(Equal([]int64{headerOne.BlockNumber, headerTwo.BlockNumber}))
Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{config.ContractAddresses, config.ContractAddresses}))
Expect(fetcher.FetchedTopics).To(Equal([][]common.Hash{{common.HexToHash(shared.VatTollSignature)}}))
})
It("returns error if fetcher returns error", func() {
fetcher := &mocks.MockLogFetcher{}
fetcher.SetFetcherError(fakes.FakeError)
repository := &vat_toll_mocks.MockVatTollRepository{}
repository.SetMissingHeaders([]core.Header{{BlockNumber: 1}})
transformer := vat_toll.VatTollTransformer{
Fetcher: fetcher,
Converter: &vat_toll_mocks.MockVatTollConverter{},
Repository: repository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@ -83,34 +86,17 @@ var _ = Describe("Vat toll transformer", func() {
})
It("marks header checked if no logs returned", func() {
mockConverter := &vat_toll_mocks.MockVatTollConverter{}
mockRepository := &vat_toll_mocks.MockVatTollRepository{}
headerID := int64(123)
mockRepository.SetMissingHeaders([]core.Header{{Id: headerID}})
mockFetcher := &mocks.MockLogFetcher{}
transformer := vat_toll.VatTollTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
mockRepository.AssertMarkHeaderCheckedCalledWith(headerID)
repository.AssertMarkHeaderCheckedCalledWith(headerOne.Id)
})
It("returns error if marking header checked returns err", func() {
mockConverter := &vat_toll_mocks.MockVatTollConverter{}
mockRepository := &vat_toll_mocks.MockVatTollRepository{}
mockRepository.SetMissingHeaders([]core.Header{{Id: int64(123)}})
mockRepository.SetMarkHeaderCheckedErr(fakes.FakeError)
mockFetcher := &mocks.MockLogFetcher{}
transformer := vat_toll.VatTollTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetMarkHeaderCheckedErr(fakes.FakeError)
err := transformer.Execute()
@ -119,16 +105,8 @@ var _ = Describe("Vat toll transformer", func() {
})
It("converts matching logs", func() {
converter := &vat_toll_mocks.MockVatTollConverter{}
fetcher := &mocks.MockLogFetcher{}
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatTollLog})
repository := &vat_toll_mocks.MockVatTollRepository{}
repository.SetMissingHeaders([]core.Header{{BlockNumber: 1}})
transformer := vat_toll.VatTollTransformer{
Fetcher: fetcher,
Converter: converter,
Repository: repository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@ -137,17 +115,9 @@ var _ = Describe("Vat toll transformer", func() {
})
It("returns error if converter returns error", func() {
converter := &vat_toll_mocks.MockVatTollConverter{}
converter.SetConverterError(fakes.FakeError)
fetcher := &mocks.MockLogFetcher{}
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatTollLog})
repository := &vat_toll_mocks.MockVatTollRepository{}
repository.SetMissingHeaders([]core.Header{{BlockNumber: 1}})
transformer := vat_toll.VatTollTransformer{
Fetcher: fetcher,
Converter: converter,
Repository: repository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@ -156,37 +126,20 @@ var _ = Describe("Vat toll transformer", func() {
})
It("persists vat toll model", func() {
converter := &vat_toll_mocks.MockVatTollConverter{}
fetcher := &mocks.MockLogFetcher{}
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatTollLog})
repository := &vat_toll_mocks.MockVatTollRepository{}
fakeHeader := core.Header{BlockNumber: 1, Id: 2}
repository.SetMissingHeaders([]core.Header{fakeHeader})
transformer := vat_toll.VatTollTransformer{
Fetcher: fetcher,
Converter: converter,
Repository: repository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(repository.PassedHeaderID).To(Equal(fakeHeader.Id))
Expect(repository.PassedModels).To(Equal([]vat_toll.VatTollModel{test_data.VatTollModel}))
Expect(repository.PassedHeaderID).To(Equal(headerOne.Id))
Expect(repository.PassedModels).To(Equal([]interface{}{test_data.VatTollModel}))
})
It("returns error if repository returns error for create", func() {
converter := &vat_toll_mocks.MockVatTollConverter{}
fetcher := &mocks.MockLogFetcher{}
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatTollLog})
repository := &vat_toll_mocks.MockVatTollRepository{}
repository.SetMissingHeaders([]core.Header{{BlockNumber: 1, Id: 2}})
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetCreateError(fakes.FakeError)
transformer := vat_toll.VatTollTransformer{
Fetcher: fetcher,
Converter: converter,
Repository: repository,
}
err := transformer.Execute()