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 slip

Browse Source
This commit is contained in:
Rob Mulholand 2018-10-24 19:42:42 -05:00
parent 1e7ee08af5
commit f5b9a21a8c
11 changed files with 102 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
pkg/transformers/integration_tests/vat_slip.go
View File

@ -3,8 +3,10 @@ package integration_tests
import (
. "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"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_slip"
"github.com/vulcanize/vulcanizedb/test_config"
@ -29,13 +31,18 @@ var _ = Describe("Vat slip transformer", func() {
})
It("persists vat slip event", func(done Done) {
config := shared.TransformerConfig{
config := shared.SingleTransformerConfig{
ContractAddresses: []string{"0xcd726790550afcd77e9a7a47e86a3f9010af126b"},
StartingBlockNumber: 8953655,
EndingBlockNumber: 8953655,
}
transformerInitializer := vat_slip.VatSlipTransformerInitializer{Config: config}
transformer := transformerInitializer.NewVatSlipTransformer(db, blockChain)
initializer := factories.Transformer{
Config: config,
Fetcher: &shared.Fetcher{},
Converter: &vat_slip.VatSlipConverter{},
Repository: &vat_slip.VatSlipRepository{},
}
transformer := initializer.NewTransformer(db, blockChain)
err := transformer.Execute()

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

@ -17,7 +17,6 @@ package vat_slip
import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_slip"
)
type MockVatSlipConverter struct {
@ -25,9 +24,9 @@ type MockVatSlipConverter struct {
PassedLogs []types.Log
}
func (converter *MockVatSlipConverter) ToModels(ethLogs []types.Log) ([]vat_slip.VatSlipModel, error) {
func (converter *MockVatSlipConverter) ToModels(ethLogs []types.Log) ([]interface{}, error) {
converter.PassedLogs = ethLogs
return []vat_slip.VatSlipModel{test_data.VatSlipModel}, converter.err
return []interface{}{test_data.VatSlipModel}, converter.err
}
func (converter *MockVatSlipConverter) SetConverterError(e error) {

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

@ -18,7 +18,7 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_slip"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type MockVatSlipRepository struct {
@ -30,10 +30,11 @@ type MockVatSlipRepository struct {
PassedStartingBlockNumber int64
PassedEndingBlockNumber int64
PassedHeaderID int64
PassedModels []vat_slip.VatSlipModel
PassedModels []interface{}
SetDbCalled bool
}
func (repository *MockVatSlipRepository) Create(headerID int64, models []vat_slip.VatSlipModel) error {
func (repository *MockVatSlipRepository) Create(headerID int64, models []interface{}) error {
repository.PassedHeaderID = headerID
repository.PassedModels = models
return repository.createError
@ -50,6 +51,10 @@ func (repository *MockVatSlipRepository) MissingHeaders(startingBlockNumber, end
return repository.missingHeaders, repository.missingHeadersError
}
func (repository *MockVatSlipRepository) SetDB(db *postgres.DB) {
repository.SetDbCalled = true
}
func (repository *MockVatSlipRepository) SetCreateError(e error) {
repository.createError = e
}

7
pkg/transformers/transformers.go
View File

@ -132,7 +132,12 @@ var (
Repository: &vat_move.VatMoveRepository{},
Fetcher: &shared.Fetcher{},
}.NewTransformer
VatSlipTransformerInitializer = vat_slip.VatSlipTransformerInitializer{Config: vat_slip.VatSlipConfig}.NewVatSlipTransformer
VatSlipTransformerInitializer = factories.Transformer{
Config: vat_slip.VatSlipConfig,
Converter: &vat_slip.VatSlipConverter{},
Repository: &vat_slip.VatSlipRepository{},
Fetcher: &shared.Fetcher{},
}.NewTransformer
VatTollTransformerInitializer = factories.Transformer{
Config: vat_toll.VatTollConfig,
Converter: &vat_toll.VatTollConverter{},

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

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

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

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

2
pkg/transformers/vat_slip/converter_test.go
View File

@ -35,7 +35,7 @@ var _ = Describe("Vat slip converter", func() {
Expect(err).To(HaveOccurred())
})
It("converts a log to an model", func() {
It("converts a log to a model", func() {
converter := vat_slip.VatSlipConverter{}
models, err := converter.ToModels([]types.Log{test_data.EthVatSlipLog})

28
pkg/transformers/vat_slip/repository.go
View File

@ -1,36 +1,31 @@
package vat_slip
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type Repository interface {
Create(headerID int64, models []VatSlipModel) error
MarkHeaderChecked(headerID int64) error
MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error)
}
type VatSlipRepository struct {
db *postgres.DB
}
func NewVatSlipRepository(db *postgres.DB) VatSlipRepository {
return VatSlipRepository{
db: db,
}
}
func (repository VatSlipRepository) Create(headerID int64, models []VatSlipModel) error {
func (repository VatSlipRepository) Create(headerID int64, models []interface{}) error {
tx, err := repository.db.Begin()
if err != nil {
return err
}
for _, model := range models {
vatSlip, ok := model.(VatSlipModel)
if !ok {
tx.Rollback()
return fmt.Errorf("model of type %T, not %T", model, VatSlipModel{})
}
_, err = tx.Exec(
`INSERT into maker.vat_slip (header_id, ilk, guy, rad, tx_idx, log_idx, raw_log)
VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,
headerID, model.Ilk, model.Guy, model.Rad, model.TransactionIndex, model.LogIndex, model.Raw,
headerID, vatSlip.Ilk, vatSlip.Guy, vatSlip.Rad, vatSlip.TransactionIndex, vatSlip.LogIndex, vatSlip.Raw,
)
if err != nil {
tx.Rollback()
@ -45,6 +40,7 @@ func (repository VatSlipRepository) Create(headerID int64, models []VatSlipModel
tx.Rollback()
return err
}
return tx.Commit()
}
@ -72,3 +68,7 @@ func (repository VatSlipRepository) MissingHeaders(startingBlockNumber, endingBl
)
return result, err
}
func (repository *VatSlipRepository) SetDB(db *postgres.DB) {
repository.db = db
}

30
pkg/transformers/vat_slip/repository_test.go
View File

@ -19,7 +19,7 @@ import (
var _ = Describe("Vat slip repository", func() {
var (
db *postgres.DB
vatSlipRepository vat_slip.Repository
vatSlipRepository vat_slip.VatSlipRepository
err error
headerRepository datastore.HeaderRepository
)
@ -28,7 +28,8 @@ var _ = Describe("Vat slip repository", func() {
db = test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
headerRepository = repositories.NewHeaderRepository(db)
vatSlipRepository = vat_slip.NewVatSlipRepository(db)
vatSlipRepository = vat_slip.VatSlipRepository{}
vatSlipRepository.SetDB(db)
})
Describe("Create", func() {
@ -40,7 +41,7 @@ var _ = Describe("Vat slip repository", func() {
})
It("adds a vat slip event", func() {
err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).NotTo(HaveOccurred())
var dbVatSlip vat_slip.VatSlipModel
@ -55,7 +56,7 @@ var _ = Describe("Vat slip repository", func() {
})
It("marks header as checked for logs", func() {
err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
@ -67,7 +68,7 @@ var _ = Describe("Vat slip 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 = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
@ -77,28 +78,28 @@ var _ = Describe("Vat slip repository", func() {
})
It("does not duplicate vat slip events", func() {
err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).NotTo(HaveOccurred())
err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
})
It("allows for multiple vat slip events in one transaction if they have different log indexes", func() {
err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).NotTo(HaveOccurred())
newVatSlip := test_data.VatSlipModel
newVatSlip.LogIndex = newVatSlip.LogIndex + 1
err := vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{newVatSlip})
err := vatSlipRepository.Create(headerID, []interface{}{newVatSlip})
Expect(err).NotTo(HaveOccurred())
})
It("removes vat slip if corresponding header is deleted", func() {
err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
err = vatSlipRepository.Create(headerID, []interface{}{test_data.VatSlipModel})
Expect(err).NotTo(HaveOccurred())
_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
@ -109,6 +110,12 @@ var _ = Describe("Vat slip repository", func() {
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(sql.ErrNoRows))
})
It("returns an error if model is of wrong type", func() {
err = vatSlipRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("model of type"))
})
})
Describe("MarkHeaderChecked", func() {
@ -195,7 +202,8 @@ var _ = Describe("Vat slip repository", func() {
_, err = headerRepositoryTwo.CreateOrUpdateHeader(fakes.GetFakeHeader(n))
Expect(err).NotTo(HaveOccurred())
}
vatSlipRepositoryTwo := vat_slip.NewVatSlipRepository(dbTwo)
vatSlipRepositoryTwo := vat_slip.VatSlipRepository{}
vatSlipRepositoryTwo.SetDB(dbTwo)
err := vatSlipRepository.MarkHeaderChecked(headerIDs[0])
Expect(err).NotTo(HaveOccurred())

64
pkg/transformers/vat_slip/transformer.go
View File

@ -1,64 +0,0 @@
package vat_slip
import (
"log"
"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"
)
type VatSlipTransformer struct {
Config shared.TransformerConfig
Converter Converter
Fetcher shared.LogFetcher
Repository Repository
}
type VatSlipTransformerInitializer struct {
Config shared.TransformerConfig
}
func (initializer VatSlipTransformerInitializer) NewVatSlipTransformer(db *postgres.DB, blockChain core.BlockChain) shared.Transformer {
converter := VatSlipConverter{}
fetcher := shared.NewFetcher(blockChain)
repository := NewVatSlipRepository(db)
return VatSlipTransformer{
Config: initializer.Config,
Converter: converter,
Fetcher: fetcher,
Repository: repository,
}
}
func (transformer VatSlipTransformer) Execute() error {
missingHeaders, err := transformer.Repository.MissingHeaders(transformer.Config.StartingBlockNumber, transformer.Config.EndingBlockNumber)
if err != nil {
return err
}
log.Printf("Fetching vat slip event logs for %d headers \n", len(missingHeaders))
for _, header := range missingHeaders {
topics := [][]common.Hash{{common.HexToHash(shared.VatSlipSignature)}}
matchingLogs, err := transformer.Fetcher.FetchLogs(VatSlipConfig.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
}

68
pkg/transformers/vat_slip/transformer_test.go
View File

@ -8,41 +8,51 @@ 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_slip_mocks "github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks/vat_slip"
"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_slip"
"math/rand"
)
var _ = Describe("Vat slip transformer", func() {
var (
config shared.TransformerConfig
converter *vat_slip_mocks.MockVatSlipConverter
fetcher *mocks.MockLogFetcher
repository *vat_slip_mocks.MockVatSlipRepository
transformer vat_slip.VatSlipTransformer
config = vat_slip.VatSlipConfig
fetcher mocks.MockLogFetcher
converter vat_slip_mocks.MockVatSlipConverter
repository vat_slip_mocks.MockVatSlipRepository
transformer shared.Transformer
headerOne core.Header
headerTwo core.Header
)
BeforeEach(func() {
config = vat_slip.VatSlipConfig
converter = &vat_slip_mocks.MockVatSlipConverter{}
fetcher = &mocks.MockLogFetcher{}
repository = &vat_slip_mocks.MockVatSlipRepository{}
transformer = vat_slip.VatSlipTransformer{
fetcher = mocks.MockLogFetcher{}
converter = vat_slip_mocks.MockVatSlipConverter{}
repository = vat_slip_mocks.MockVatSlipRepository{}
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,
}
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_slip.VatSlipConfig.StartingBlockNumber))
Expect(repository.PassedEndingBlockNumber).To(Equal(vat_slip.VatSlipConfig.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() {
@ -55,21 +65,19 @@ var _ = Describe("Vat slip transformer", func() {
})
It("fetches logs for missing headers", func() {
headerOne := core.Header{BlockNumber: GinkgoRandomSeed()}
headerTwo := core.Header{BlockNumber: GinkgoRandomSeed()}
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{vat_slip.VatSlipConfig.ContractAddresses, vat_slip.VatSlipConfig.ContractAddresses}))
Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{config.ContractAddresses, config.ContractAddresses}))
Expect(fetcher.FetchedTopics).To(Equal([][]common.Hash{{common.HexToHash(shared.VatSlipSignature)}}))
})
It("returns error if fetcher returns error", func() {
fetcher.SetFetcherError(fakes.FakeError)
repository.SetMissingHeaders([]core.Header{{BlockNumber: GinkgoRandomSeed()}})
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@ -78,17 +86,16 @@ var _ = Describe("Vat slip transformer", func() {
})
It("marks header checked if no logs returned", func() {
headerID := GinkgoRandomSeed()
repository.SetMissingHeaders([]core.Header{{Id: headerID}})
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
repository.AssertMarkHeaderCheckedCalledWith(headerID)
repository.AssertMarkHeaderCheckedCalledWith(headerOne.Id)
})
It("returns error if marking header checked returns err", func() {
repository.SetMissingHeaders([]core.Header{{Id: GinkgoRandomSeed()}})
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetMarkHeaderCheckedErr(fakes.FakeError)
err := transformer.Execute()
@ -99,7 +106,7 @@ var _ = Describe("Vat slip transformer", func() {
It("converts matching logs", func() {
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatSlipLog})
repository.SetMissingHeaders([]core.Header{{BlockNumber: GinkgoRandomSeed()}})
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@ -110,7 +117,7 @@ var _ = Describe("Vat slip transformer", func() {
It("returns error if converter returns error", func() {
converter.SetConverterError(fakes.FakeError)
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatSlipLog})
repository.SetMissingHeaders([]core.Header{{BlockNumber: GinkgoRandomSeed()}})
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@ -120,19 +127,18 @@ var _ = Describe("Vat slip transformer", func() {
It("persists vat slip model", func() {
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatSlipLog})
fakeHeader := core.Header{BlockNumber: GinkgoRandomSeed(), Id: GinkgoRandomSeed()}
repository.SetMissingHeaders([]core.Header{fakeHeader})
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_slip.VatSlipModel{test_data.VatSlipModel}))
Expect(repository.PassedHeaderID).To(Equal(headerOne.Id))
Expect(repository.PassedModels).To(Equal([]interface{}{test_data.VatSlipModel}))
})
It("returns error if repository returns error for create", func() {
fetcher.SetFetchedLogs([]types.Log{test_data.EthVatSlipLog})
repository.SetMissingHeaders([]core.Header{{BlockNumber: GinkgoRandomSeed(), Id: GinkgoRandomSeed()}})
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetCreateError(fakes.FakeError)
err := transformer.Execute()