diff --git a/pkg/transformers/deal/config.go b/pkg/transformers/deal/config.go index 19449c57..443f89b1 100644 --- a/pkg/transformers/deal/config.go +++ b/pkg/transformers/deal/config.go @@ -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, } diff --git a/pkg/transformers/deal/converter.go b/pkg/transformers/deal/converter.go index 5e4505d9..2a7695d7 100644 --- a/pkg/transformers/deal/converter.go +++ b/pkg/transformers/deal/converter.go @@ -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 { diff --git a/pkg/transformers/deal/repository.go b/pkg/transformers/deal/repository.go index 8845af89..6234bbdd 100644 --- a/pkg/transformers/deal/repository.go +++ b/pkg/transformers/deal/repository.go @@ -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 +} diff --git a/pkg/transformers/deal/transformer.go b/pkg/transformers/deal/transformer.go deleted file mode 100644 index 890d36b0..00000000 --- a/pkg/transformers/deal/transformer.go +++ /dev/null @@ -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 -} diff --git a/pkg/transformers/transformers.go b/pkg/transformers/transformers.go index 63c583ea..07578db6 100644 --- a/pkg/transformers/transformers.go +++ b/pkg/transformers/transformers.go @@ -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{