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

Refactor deal transformer to use factory

Browse Source
This commit is contained in:
Edvard 2018-10-26 14:31:25 +02:00 committed by Rob Mulholand
parent d522d33531
commit 35a8009410
5 changed files with 41 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

10
pkg/transformers/deal/converter.go
View File

@ -21,17 +21,9 @@ import (
"github.com/ethereum/go-ethereum/core/types"
)
type Converter interface {
ToModels(ethLog []types.Log) ([]DealModel, error)
}
type DealConverter struct{}
func NewDealConverter() DealConverter {
return DealConverter{}
}
func (DealConverter) ToModels(ethLogs []types.Log) (result []DealModel, err error) {
func (DealConverter) ToModels(ethLogs []types.Log) (result []interface{}, err error) {
for _, log := range ethLogs {
err := validateLog(log)
if err != nil {

53
pkg/transformers/deal/repository.go
View File

@ -15,32 +15,32 @@
package deal
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type Repository interface {
Create(headerId int64, models []DealModel) error
MarkHeaderChecked(headerID int64) error
MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error)
}
type DealRepository struct {
db *postgres.DB
}
func NewDealRepository(database *postgres.DB) DealRepository {
return DealRepository{db: database}
}
func (r DealRepository) Create(headerId int64, models []DealModel) error {
tx, err := r.db.Begin()
func (repository DealRepository) Create(headerId int64, models []interface{}) error {
tx, err := repository.db.Begin()
if err != nil {
return err
}
for _, model := range models {
deal, ok := model.(DealModel)
if !ok {
tx.Rollback()
return fmt.Errorf("model of type %T, not %T", model, DealModel{})
}
_, err = tx.Exec(
`INSERT into maker.deal (header_id, bid_id, contract_address, log_idx, tx_idx, raw_log)
VALUES($1, $2, $3, $4, $5, $6)`,
headerId, model.BidId, model.ContractAddress, model.LogIndex, model.TransactionIndex, model.Raw,
headerId, deal.BidId, deal.ContractAddress, deal.LogIndex, deal.TransactionIndex, deal.Raw,
)
if err != nil {
tx.Rollback()
@ -51,6 +51,7 @@ func (r DealRepository) Create(headerId int64, models []DealModel) error {
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET deal_checked = $2`, headerId, true)
if err != nil {
tx.Rollback()
return err
@ -58,26 +59,30 @@ func (r DealRepository) Create(headerId int64, models []DealModel) error {
return tx.Commit()
}
func (r DealRepository) MarkHeaderChecked(headerID int64) error {
_, err := r.db.Exec(`INSERT INTO public.checked_headers (header_id, deal_checked)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET deal_checked = $2`, headerID, true)
func (repository DealRepository) MarkHeaderChecked(headerID int64) error {
_, err := repository.db.Exec(`INSERT INTO public.checked_headers (header_id, deal_checked)
VALUES ($1, $2)
ON CONFLICT (header_id) DO
UPDATE SET deal_checked = $2`, headerID, true)
return err
}
func (r DealRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error) {
func (repository DealRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error) {
var missingHeaders []core.Header
err := r.db.Select(&missingHeaders,
err := repository.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 deal_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 deal_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,
repository.db.Node.ID,
)
return missingHeaders, err
}
func (repository *DealRepository) SetDB(db *postgres.DB) {
repository.db = db
}

84
pkg/transformers/deal/transformer.go
View File

@ -1,84 +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 deal
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 DealTransformer struct {
Config shared.TransformerConfig
Converter Converter
Fetcher shared.LogFetcher
Repository Repository
}
type DealTransformerInitializer struct {
Config shared.TransformerConfig
}
func (i DealTransformerInitializer) NewDealTransformer(db *postgres.DB, blockChain core.BlockChain) shared.Transformer {
converter := NewDealConverter()
fetcher := shared.NewFetcher(blockChain)
repository := NewDealRepository(db)
return DealTransformer{
Config: i.Config,
Converter: converter,
Fetcher: fetcher,
Repository: repository,
}
}
func (t DealTransformer) Execute() error {
config := t.Config
topics := [][]common.Hash{{common.HexToHash(shared.DealSignature)}}
headers, err := t.Repository.MissingHeaders(config.StartingBlockNumber, config.EndingBlockNumber)
if err != nil {
return err
}
log.Printf("Fetching deal 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 deal 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 deal log", err)
return err
}
err = t.Repository.Create(header.Id, models)
if err != nil {
log.Println("Error persisting deal record", err)
return err
}
}
return err
}

9
pkg/transformers/transformers.go
View File

@ -49,6 +49,7 @@ import (
var (
BiteTransformerInitializer = bite.BiteTransformerInitializer{Config: bite.BiteConfig}.NewBiteTransformer
CatFileChopLumpTransformerInitializer = factories.Transformer{
Config: chop_lump.CatFileChopLumpConfig,
Converter: &chop_lump.CatFileChopLumpConverter{},
@ -70,7 +71,13 @@ var (
Fetcher: &shared.Fetcher{},
}.NewTransformer
DealTransformerInitializer = deal.DealTransformerInitializer{Config: deal.Config}.NewDealTransformer
DealTransformerInitializer = factories.Transformer{
Config: deal.DealConfig,
Converter: &deal.DealConverter{},
Repository: &deal.DealRepository{},
Fetcher: &shared.Fetcher{},
}.NewTransformer
DentTransformerInitializer = dent.DentTransformerInitializer{Config: dent.DentConfig}.NewDentTransformer
DripDripTransformerInitializer = drip_drip.DripDripTransformerInitializer{Config: drip_drip.DripDripConfig}.NewDripDripTransformer
DripFileIlkTransformerInitializer = factories.Transformer{