deep-stack/ipld-eth-server
6
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

VDB-184 Refactor Dent (#91)

Browse Source 
* Factorise dent transformer

* Update dent transformer tests

* Add additional tests to dent repository test

* Update dent to use LogNoteTransformer factory
This commit is contained in:
Edvard Hübinette 2018-10-26 20:56:49 +02:00 committed by Elizabeth
parent d1ee501f12
commit f3a82e6ba1
10 changed files with 113 additions and 281 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -16,10 +16,11 @@ package dent
import "github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
var DentConfig = shared.TransformerConfig{
var DentConfig = shared.SingleTransformerConfig{
TransformerName: shared.DentLabel,
ContractAddresses: []string{shared.FlipperContractAddress},
ContractAbi: shared.FlipperABI,
Topics: []string{shared.DentFunctionSignature},
Topic: shared.DentFunctionSignature,
StartingBlockNumber: 0,
EndingBlockNumber: 10000000,
}

6
pkg/transformers/dent/converter.go
View File

@ -23,17 +23,13 @@ import (
"github.com/ethereum/go-ethereum/core/types"
)
type Converter interface {
ToModels(ethLogs []types.Log) ([]DentModel, error)
}
type DentConverter struct{}
func NewDentConverter() DentConverter {
return DentConverter{}
}
func (c DentConverter) ToModels(ethLogs []types.Log) (result []DentModel, err error) {
func (c DentConverter) ToModels(ethLogs []types.Log) (result []interface{}, err error) {
for _, log := range ethLogs {
err := validateLog(log)
if err != nil {

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

@ -36,7 +36,7 @@ var _ = Describe("Dent Converter", func() {
Expect(err).NotTo(HaveOccurred())
Expect(len(models)).To(Equal(1))
Expect(models[0]).To(Equal(test_data.DentModel))
Expect(models[0].(dent.DentModel)).To(Equal(test_data.DentModel))
})
It("returns an error if the expected amount of topics aren't in the log", func() {

45
pkg/transformers/dent/repository.go
View File

@ -15,44 +15,43 @@
package dent
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type Repository interface {
Create(headerId int64, models []DentModel) error
MarkHeaderChecked(headerId int64) error
MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error)
}
type DentRepository struct {
db *postgres.DB
}
func NewDentRepository(database *postgres.DB) DentRepository {
return DentRepository{db: database}
}
func (r DentRepository) Create(headerId int64, models []DentModel) error {
func (r DentRepository) Create(headerId int64, models []interface{}) error {
tx, err := r.db.Begin()
if err != nil {
return err
}
for _, model := range models {
dent, ok := model.(DentModel)
if !ok {
tx.Rollback()
return fmt.Errorf("model of type %T, not %T", model, DentModel{})
}
_, err = tx.Exec(
`INSERT into maker.dent (header_id, bid_id, lot, bid, guy, tic, log_idx, tx_idx, raw_log)
VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
headerId, model.BidId, model.Lot, model.Bid, model.Guy, model.Tic, model.LogIndex, model.TransactionIndex, model.Raw,
VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9)`,
headerId, dent.BidId, dent.Lot, dent.Bid, dent.Guy, dent.Tic, dent.LogIndex, dent.TransactionIndex, dent.Raw,
)
if err != nil {
tx.Rollback()
return err
}
}
_, err = tx.Exec(`INSERT INTO public.checked_headers (header_id, dent_checked)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET dent_checked = $2`, headerId, true)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET dent_checked = $2`, headerId, true)
if err != nil {
tx.Rollback()
return err
@ -74,11 +73,11 @@ func (r DentRepository) MissingHeaders(startingBlockNumber, endingBlockNumber in
err := r.db.Select(
&missingHeaders,
`SELECT headers.id, headers.block_number FROM headers
LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL OR dent_checked IS FALSE)
AND headers.block_number >= $1
AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3`,
LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL OR dent_checked IS FALSE)
AND headers.block_number >= $1
AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3`,
startingBlockNumber,
endingBlockNumber,
r.db.Node.ID,
@ -86,3 +85,7 @@ func (r DentRepository) MissingHeaders(startingBlockNumber, endingBlockNumber in
return missingHeaders, err
}
func (repository *DentRepository) SetDB(db *postgres.DB) {
repository.db = db
}

45
pkg/transformers/dent/repository_test.go
View File

@ -38,7 +38,8 @@ var _ = Describe("Dent Repository", func() {
BeforeEach(func() {
db = test_config.NewTestDB(test_config.NewTestNode())
test_config.CleanTestDB(db)
dentRepository = dent.NewDentRepository(db)
dentRepository = dent.DentRepository{}
dentRepository.SetDB(db)
headerRepository = repositories.NewHeaderRepository(db)
})
@ -48,12 +49,12 @@ var _ = Describe("Dent Repository", func() {
BeforeEach(func() {
headerId, err = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
Expect(err).NotTo(HaveOccurred())
err := dentRepository.Create(headerId, []dent.DentModel{test_data.DentModel})
Expect(err).NotTo(HaveOccurred())
})
It("persists a dent record", func() {
err := dentRepository.Create(headerId, []interface{}{test_data.DentModel})
Expect(err).NotTo(HaveOccurred())
var count int
db.QueryRow(`SELECT count(*) FROM maker.dent`).Scan(&count)
Expect(count).To(Equal(1))
@ -72,6 +73,22 @@ var _ = Describe("Dent Repository", func() {
})
It("marks header as checked for logs", func() {
err := dentRepository.Create(headerId, []interface{}{test_data.DentModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
err = db.Get(&headerChecked, `SELECT dent_checked FROM public.checked_headers WHERE header_id = $1`, headerId)
Expect(err).NotTo(HaveOccurred())
Expect(headerChecked).To(BeTrue())
})
It("updates the checked_header row if it already exists", func() {
_, err := db.Exec(`INSERT INTO public.checked_headers (header_id) VALUES ($1)`, headerId)
Expect(err).NotTo(HaveOccurred())
err = dentRepository.Create(headerId, []interface{}{test_data.DentModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
err = db.Get(&headerChecked, `SELECT dent_checked FROM public.checked_headers WHERE header_id = $1`, headerId)
Expect(err).NotTo(HaveOccurred())
@ -79,12 +96,25 @@ var _ = Describe("Dent Repository", func() {
})
It("returns an error if inserting a dent record fails", func() {
err = dentRepository.Create(headerId, []dent.DentModel{test_data.DentModel})
err := dentRepository.Create(headerId, []interface{}{test_data.DentModel})
Expect(err).NotTo(HaveOccurred())
err = dentRepository.Create(headerId, []interface{}{test_data.DentModel})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
})
It("deletes the tend record if its corresponding header record is deleted", func() {
It("returns an error if model is of wrong type", func() {
err = dentRepository.Create(headerId, []interface{}{test_data.WrongModel{}})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("model of type"))
})
It("deletes the dent record if its corresponding header record is deleted", func() {
err := dentRepository.Create(headerId, []interface{}{test_data.DentModel})
Expect(err).NotTo(HaveOccurred())
var count int
err = db.QueryRow(`SELECT count(*) from maker.dent`).Scan(&count)
Expect(err).NotTo(HaveOccurred())
@ -180,7 +210,8 @@ var _ = Describe("Dent Repository", func() {
dentRepository.MarkHeaderChecked(headerIds[1])
node2 := core.Node{}
db2 := test_config.NewTestDB(node2)
dentRepository2 := dent.NewDentRepository(db2)
dentRepository2 := dent.DentRepository{}
dentRepository2.SetDB(db2)
headerRepository2 := repositories.NewHeaderRepository(db2)
var node2HeaderIds []int64
for _, number := range blockNumbers {

83
pkg/transformers/dent/transformer.go
View File

@ -1,83 +0,0 @@
// Copyright 2018 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package dent
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 DentTransformer struct {
Config shared.TransformerConfig
Converter Converter
Fetcher shared.LogFetcher
Repository Repository
}
type DentTransformerInitializer struct {
Config shared.TransformerConfig
}
func (i DentTransformerInitializer) NewDentTransformer(db *postgres.DB, blockChain core.BlockChain) shared.Transformer {
converter := NewDentConverter()
fetcher := shared.NewFetcher(blockChain)
repository := NewDentRepository(db)
return DentTransformer{
Config: i.Config,
Converter: converter,
Fetcher: fetcher,
Repository: repository,
}
}
func (t DentTransformer) Execute() error {
config := t.Config
topics := [][]common.Hash{{common.HexToHash(shared.DentFunctionSignature)}}
headers, err := t.Repository.MissingHeaders(config.StartingBlockNumber, config.EndingBlockNumber)
log.Printf("Fetching dent event logs for %d headers \n", len(headers))
for _, header := range headers {
ethLogs, err := t.Fetcher.FetchLogs(config.ContractAddresses, topics, header.BlockNumber)
if err != nil {
log.Println("Error fetching dent logs:", err)
return err
}
if len(ethLogs) < 1 {
err := t.Repository.MarkHeaderChecked(header.Id)
if err != nil {
return err
}
}
models, err := t.Converter.ToModels(ethLogs)
if err != nil {
log.Println("Error converting dent log", err)
return err
}
err = t.Repository.Create(header.Id, models)
if err != nil {
log.Println("Error persisting dent record", err)
return err
}
}
return err
}

89
pkg/transformers/dent/transformer_test.go
View File

@ -15,6 +15,7 @@
package dent_test
import (
"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
"math/rand"
"github.com/ethereum/go-ethereum/common"
@ -28,38 +29,46 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks"
dent_mocks "github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks/dent"
)
var _ = Describe("DentTransformer", func() {
var config = dent.DentConfig
var dentRepository dent_mocks.MockDentRepository
var repository mocks.MockRepository
var fetcher mocks.MockLogFetcher
var converter dent_mocks.MockDentConverter
var transformer dent.DentTransformer
var converter mocks.MockLogNoteConverter
var transformer shared.Transformer
var headerOne core.Header
var headerTwo core.Header
BeforeEach(func() {
dentRepository = dent_mocks.MockDentRepository{}
repository = mocks.MockRepository{}
fetcher = mocks.MockLogFetcher{}
converter = dent_mocks.MockDentConverter{}
transformer = dent.DentTransformer{
Repository: &dentRepository,
converter = mocks.MockLogNoteConverter{}
transformer = factories.LogNoteTransformer{
Config: config,
Fetcher: &fetcher,
Converter: &converter,
}
Repository: &repository,
Fetcher: &fetcher,
}.NewLogNoteTransformer(nil, nil)
headerOne = core.Header{BlockNumber: rand.Int63(), Id: rand.Int63()}
headerTwo = core.Header{BlockNumber: rand.Int63(), Id: rand.Int63()}
})
It("sets the blockchain and database", func() {
Expect(fetcher.SetBcCalled).To(BeTrue())
Expect(repository.SetDbCalled).To(BeTrue())
})
It("gets missing headers", func() {
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(dentRepository.PassedStartingBlockNumber).To(Equal(config.StartingBlockNumber))
Expect(dentRepository.PassedEndingBlockNumber).To(Equal(config.EndingBlockNumber))
Expect(repository.PassedStartingBlockNumber).To(Equal(config.StartingBlockNumber))
Expect(repository.PassedEndingBlockNumber).To(Equal(config.EndingBlockNumber))
})
It("returns an error if fetching the missing headers fails", func() {
dentRepository.SetMissingHeadersError(fakes.FakeError)
repository.SetMissingHeadersError(fakes.FakeError)
err := transformer.Execute()
Expect(err).To(HaveOccurred())
@ -67,20 +76,18 @@ var _ = Describe("DentTransformer", func() {
})
It("fetches logs for each missing header", func() {
header1 := core.Header{BlockNumber: rand.Int63()}
header2 := core.Header{BlockNumber: rand.Int63()}
dentRepository.SetMissingHeaders([]core.Header{header1, header2})
repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{config.ContractAddresses, config.ContractAddresses}))
expectedTopics := [][]common.Hash{{common.HexToHash(shared.DentFunctionSignature)}}
Expect(fetcher.FetchedTopics).To(Equal(expectedTopics))
Expect(fetcher.FetchedBlocks).To(Equal([]int64{header1.BlockNumber, header2.BlockNumber}))
Expect(fetcher.FetchedBlocks).To(Equal([]int64{headerOne.BlockNumber, headerTwo.BlockNumber}))
})
It("returns an error if fetching logs fails", func() {
dentRepository.SetMissingHeaders([]core.Header{{}})
repository.SetMissingHeaders([]core.Header{headerOne})
fetcher.SetFetcherError(fakes.FakeError)
err := transformer.Execute()
@ -89,34 +96,17 @@ var _ = Describe("DentTransformer", func() {
})
It("marks header checked if no logs returned", func() {
mockConverter := &dent_mocks.MockDentConverter{}
mockRepository := &dent_mocks.MockDentRepository{}
headerID := int64(123)
mockRepository.SetMissingHeaders([]core.Header{{Id: headerID}})
mockFetcher := &mocks.MockLogFetcher{}
transformer := dent.DentTransformer{
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 := &dent_mocks.MockDentConverter{}
mockRepository := &dent_mocks.MockDentRepository{}
mockRepository.SetMissingHeaders([]core.Header{{Id: int64(123)}})
mockRepository.SetMarkHeaderCheckedErr(fakes.FakeError)
mockFetcher := &mocks.MockLogFetcher{}
transformer := dent.DentTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetMarkHeaderCheckedError(fakes.FakeError)
err := transformer.Execute()
@ -125,16 +115,16 @@ var _ = Describe("DentTransformer", func() {
})
It("converts each eth log to a Model", func() {
dentRepository.SetMissingHeaders([]core.Header{{}})
repository.SetMissingHeaders([]core.Header{headerOne})
fetcher.SetFetchedLogs([]types.Log{test_data.DentLog})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(converter.LogsToConvert).To(Equal([]types.Log{test_data.DentLog}))
Expect(converter.PassedLogs).To(Equal([]types.Log{test_data.DentLog}))
})
It("returns an error if converting the eth log fails", func() {
dentRepository.SetMissingHeaders([]core.Header{{}})
repository.SetMissingHeaders([]core.Header{headerOne})
fetcher.SetFetchedLogs([]types.Log{test_data.DentLog})
converter.SetConverterError(fakes.FakeError)
err := transformer.Execute()
@ -143,21 +133,20 @@ var _ = Describe("DentTransformer", func() {
Expect(err).To(MatchError(fakes.FakeError))
})
It("persists each model as a Dent record", func() {
header1 := core.Header{Id: rand.Int63()}
header2 := core.Header{Id: rand.Int63()}
dentRepository.SetMissingHeaders([]core.Header{header1, header2})
It("persists the model as a Dent record", func() {
repository.SetMissingHeaders([]core.Header{headerOne})
fetcher.SetFetchedLogs([]types.Log{test_data.DentLog})
converter.SetReturnModels([]interface{}{test_data.DentModel})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(dentRepository.PassedDentModels).To(Equal([]dent.DentModel{test_data.DentModel, test_data.DentModel}))
Expect(dentRepository.PassedHeaderIds).To(Equal([]int64{header1.Id, header2.Id}))
Expect(repository.PassedModels).To(Equal([]interface{}{test_data.DentModel}))
Expect(repository.PassedHeaderID).To(Equal(headerOne.Id))
})
It("returns an error if persisting dent record fails", func() {
dentRepository.SetMissingHeaders([]core.Header{{}})
dentRepository.SetCreateError(fakes.FakeError)
repository.SetMissingHeaders([]core.Header{{}})
repository.SetCreateError(fakes.FakeError)
fetcher.SetFetchedLogs([]types.Log{test_data.DentLog})
err := transformer.Execute()

38
pkg/transformers/test_data/mocks/dent/converter.go
View File

@ -1,38 +0,0 @@
// Copyright 2018 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package dent
import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/transformers/dent"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
)
type MockDentConverter struct {
converterError error
PassedContractAddress string
PassedContractAbi string
LogsToConvert []types.Log
}
func (c *MockDentConverter) ToModels(ethLogs []types.Log) ([]dent.DentModel, error) {
c.LogsToConvert = append(c.LogsToConvert, ethLogs...)
return []dent.DentModel{test_data.DentModel}, c.converterError
}
func (c *MockDentConverter) SetConverterError(err error) {
c.converterError = err
}

73
pkg/transformers/test_data/mocks/dent/repository.go
View File

@ -1,73 +0,0 @@
// Copyright 2018 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package dent
import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/transformers/dent"
)
type MockDentRepository struct {
PassedStartingBlockNumber int64
PassedEndingBlockNumber int64
PassedDentModels []dent.DentModel
PassedHeaderIds []int64
markHeaderCheckedErr error
markHeaderCheckedPassedHeaderId int64
missingHeaders []core.Header
missingHeadersError error
createError error
}
func (r *MockDentRepository) Create(headerId int64, models []dent.DentModel) error {
r.PassedHeaderIds = append(r.PassedHeaderIds, headerId)
r.PassedDentModels = append(r.PassedDentModels, models...)
return r.createError
}
func (r *MockDentRepository) MarkHeaderChecked(headerId int64) error {
r.markHeaderCheckedPassedHeaderId = headerId
return r.markHeaderCheckedErr
}
func (r *MockDentRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error) {
r.PassedStartingBlockNumber = startingBlockNumber
r.PassedEndingBlockNumber = endingBlockNumber
return r.missingHeaders, r.missingHeadersError
}
func (r *MockDentRepository) SetMarkHeaderCheckedErr(err error) {
r.markHeaderCheckedErr = err
}
func (r *MockDentRepository) SetMissingHeadersError(err error) {
r.missingHeadersError = err
}
func (r *MockDentRepository) SetMissingHeaders(headers []core.Header) {
r.missingHeaders = headers
}
func (r *MockDentRepository) SetCreateError(err error) {
r.createError = err
}
func (r *MockDentRepository) AssertMarkHeaderCheckedCalledWith(headerId int64) {
Expect(r.markHeaderCheckedPassedHeaderId).To(Equal(headerId))
}

8
pkg/transformers/transformers.go
View File

@ -83,7 +83,13 @@ var (
Fetcher: &shared.Fetcher{},
}.NewLogNoteTransformer
DentTransformerInitializer = dent.DentTransformerInitializer{Config: dent.DentConfig}.NewDentTransformer
DentTransformerInitializer = factories.Transformer{
Config: dent.DentConfig,
Converter: &dent.DentConverter{},
Repository: &dent.DentRepository{},
Fetcher: &shared.Fetcher{},
}.NewTransformer
DripDripTransformerInitializer = drip_drip.DripDripTransformerInitializer{Config: drip_drip.DripDripConfig}.NewDripDripTransformer
DripFileIlkTransformerInitializer = factories.LogNoteTransformer{