WIP: First steps in refactoring to aggregate fetching
This commit is contained in:
parent
423fdf01b5
commit
45a087f5d1
@ -1,6 +1,7 @@
|
||||
package shared
|
||||
|
||||
import (
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
|
||||
@ -14,15 +15,22 @@ type Watcher struct {
|
||||
|
||||
func (watcher *Watcher) AddTransformers(us []shared.TransformerInitializer) {
|
||||
for _, transformerInitializer := range us {
|
||||
transformer := transformerInitializer(&watcher.DB, watcher.Blockchain)
|
||||
transformer := transformerInitializer(&watcher.DB)
|
||||
watcher.Transformers = append(watcher.Transformers, transformer)
|
||||
}
|
||||
}
|
||||
|
||||
func (watcher *Watcher) Execute() error {
|
||||
// TODO Solve checkedHeadersColumn issue
|
||||
// TODO Handle start and end numbers in transformers?
|
||||
var missingHeaders []core.Header
|
||||
|
||||
// TODO Get contract addresses and topic0s
|
||||
var logs []types.Log
|
||||
|
||||
var err error
|
||||
for _, transformer := range watcher.Transformers {
|
||||
err = transformer.Execute()
|
||||
err = transformer.Execute(logs, missingHeaders)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
@ -15,10 +15,9 @@
|
||||
package factories
|
||||
|
||||
import (
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"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"
|
||||
@ -28,38 +27,19 @@ type LogNoteTransformer struct {
|
||||
Config shared.TransformerConfig
|
||||
Converter LogNoteConverter
|
||||
Repository Repository
|
||||
Fetcher shared.SettableLogFetcher
|
||||
}
|
||||
|
||||
func (transformer LogNoteTransformer) NewLogNoteTransformer(db *postgres.DB, bc core.BlockChain) shared.Transformer {
|
||||
func (transformer LogNoteTransformer) NewLogNoteTransformer(db *postgres.DB) shared.Transformer {
|
||||
transformer.Repository.SetDB(db)
|
||||
transformer.Fetcher.SetBC(bc)
|
||||
return transformer
|
||||
}
|
||||
|
||||
func (transformer LogNoteTransformer) Execute() error {
|
||||
func (transformer LogNoteTransformer) Execute(logs []types.Log, missingHeaders []core.Header) error {
|
||||
transformerName := transformer.Config.TransformerName
|
||||
missingHeaders, err := transformer.Repository.MissingHeaders(transformer.Config.StartingBlockNumber, transformer.Config.EndingBlockNumber)
|
||||
if err != nil {
|
||||
log.Printf("Error fetching missing headers in %v transformer: %v", transformerName, err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Grab event signature from transformer config
|
||||
// (Double-array structure required for go-ethereum FilterQuery)
|
||||
var topic = [][]common.Hash{{common.HexToHash(transformer.Config.Topic)}}
|
||||
|
||||
log.Printf("Fetching %v event logs for %d headers", transformerName, len(missingHeaders))
|
||||
for _, header := range missingHeaders {
|
||||
// Fetch the missing logs for a given header
|
||||
matchingLogs, err := transformer.Fetcher.FetchLogs(transformer.Config.ContractAddresses, topic, header)
|
||||
if err != nil {
|
||||
log.Printf("Error fetching matching logs in %v transformer: %v", transformerName, err)
|
||||
return err
|
||||
}
|
||||
|
||||
// No matching logs, mark the header as checked for this type of logs
|
||||
if len(matchingLogs) < 1 {
|
||||
if len(logs) < 1 {
|
||||
err := transformer.Repository.MarkHeaderChecked(header.Id)
|
||||
if err != nil {
|
||||
log.Printf("Error marking header as checked in %v: %v", transformerName, err)
|
||||
@ -69,7 +49,7 @@ func (transformer LogNoteTransformer) Execute() error {
|
||||
continue
|
||||
}
|
||||
|
||||
models, err := transformer.Converter.ToModels(matchingLogs)
|
||||
models, err := transformer.Converter.ToModels(logs)
|
||||
if err != nil {
|
||||
log.Printf("Error converting logs in %v: %v", transformerName, err)
|
||||
return err
|
||||
|
@ -15,10 +15,9 @@
|
||||
package factories
|
||||
|
||||
import (
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"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"
|
||||
@ -28,34 +27,20 @@ type Transformer struct {
|
||||
Config shared.TransformerConfig
|
||||
Converter Converter
|
||||
Repository Repository
|
||||
Fetcher shared.SettableLogFetcher
|
||||
}
|
||||
|
||||
func (transformer Transformer) NewTransformer(db *postgres.DB, bc core.BlockChain) shared.Transformer {
|
||||
func (transformer Transformer) NewTransformer(db *postgres.DB) shared.Transformer {
|
||||
transformer.Repository.SetDB(db)
|
||||
transformer.Fetcher.SetBC(bc)
|
||||
return transformer
|
||||
}
|
||||
|
||||
func (transformer Transformer) Execute() error {
|
||||
func (transformer Transformer) Execute(logs []types.Log, missingHeaders []core.Header) error {
|
||||
transformerName := transformer.Config.TransformerName
|
||||
config := transformer.Config
|
||||
topics := [][]common.Hash{{common.HexToHash(config.Topic)}}
|
||||
missingHeaders, err := transformer.Repository.MissingHeaders(config.StartingBlockNumber, config.EndingBlockNumber)
|
||||
if err != nil {
|
||||
log.Printf("Error fetching missing headers in %v transformer: %v \n", transformerName, err)
|
||||
return err
|
||||
}
|
||||
log.Printf("Fetching %v event logs for %d headers \n", transformerName, len(missingHeaders))
|
||||
for _, header := range missingHeaders {
|
||||
logs, err := transformer.Fetcher.FetchLogs(config.ContractAddresses, topics, header)
|
||||
if err != nil {
|
||||
log.Printf("Error fetching matching logs in %v transformer: %v", transformerName, err)
|
||||
return err
|
||||
}
|
||||
|
||||
for _, header := range missingHeaders {
|
||||
if len(logs) < 1 {
|
||||
err = transformer.Repository.MarkHeaderChecked(header.Id)
|
||||
err := transformer.Repository.MarkHeaderChecked(header.Id)
|
||||
if err != nil {
|
||||
log.Printf("Error marking header as checked in %v: %v", transformerName, err)
|
||||
return err
|
||||
|
@ -16,16 +16,17 @@ package shared
|
||||
|
||||
import (
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
|
||||
)
|
||||
|
||||
type Transformer interface {
|
||||
Execute() error
|
||||
Execute(logs []types.Log, missingHeaders []core.Header) error
|
||||
}
|
||||
|
||||
type TransformerInitializer func(db *postgres.DB, blockChain core.BlockChain) Transformer
|
||||
type TransformerInitializer func(db *postgres.DB) Transformer
|
||||
|
||||
type TransformerConfig struct {
|
||||
TransformerName string
|
||||
|
@ -52,196 +52,168 @@ var (
|
||||
Config: bite.BiteConfig,
|
||||
Converter: &bite.BiteConverter{},
|
||||
Repository: &bite.BiteRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewTransformer
|
||||
|
||||
CatFileChopLumpTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: chop_lump.CatFileChopLumpConfig,
|
||||
Converter: &chop_lump.CatFileChopLumpConverter{},
|
||||
Repository: &chop_lump.CatFileChopLumpRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
CatFileFlipTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: flip.CatFileFlipConfig,
|
||||
Converter: &flip.CatFileFlipConverter{},
|
||||
Repository: &flip.CatFileFlipRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
CatFilePitVowTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: pit_vow.CatFilePitVowConfig,
|
||||
Converter: &pit_vow.CatFilePitVowConverter{},
|
||||
Repository: &pit_vow.CatFilePitVowRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
DealTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: deal.DealConfig,
|
||||
Converter: &deal.DealConverter{},
|
||||
Repository: &deal.DealRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
DentTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: dent.DentConfig,
|
||||
Converter: &dent.DentConverter{},
|
||||
Repository: &dent.DentRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
DripDripTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: drip_drip.DripDripConfig,
|
||||
Converter: &drip_drip.DripDripConverter{},
|
||||
Repository: &drip_drip.DripDripRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
DripFileIlkTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: ilk2.DripFileIlkConfig,
|
||||
Converter: &ilk2.DripFileIlkConverter{},
|
||||
Repository: &ilk2.DripFileIlkRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
DripFileRepoTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: repo.DripFileRepoConfig,
|
||||
Converter: &repo.DripFileRepoConverter{},
|
||||
Repository: &repo.DripFileRepoRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
DripFileVowTransfromerInitializer = factories.LogNoteTransformer{
|
||||
Config: vow.DripFileVowConfig,
|
||||
Converter: &vow.DripFileVowConverter{},
|
||||
Repository: &vow.DripFileVowRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
FlapKickTransformerInitializer = factories.Transformer{
|
||||
Config: flap_kick.FlapKickConfig,
|
||||
Converter: &flap_kick.FlapKickConverter{},
|
||||
Repository: &flap_kick.FlapKickRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewTransformer
|
||||
|
||||
FlipKickTransformerInitializer = factories.Transformer{
|
||||
Config: flip_kick.FlipKickConfig,
|
||||
Converter: &flip_kick.FlipKickConverter{},
|
||||
Repository: &flip_kick.FlipKickRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewTransformer
|
||||
|
||||
FlogTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vow_flog.VowFlogConfig,
|
||||
Converter: &vow_flog.VowFlogConverter{},
|
||||
Repository: &vow_flog.VowFlogRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
FrobTransformerInitializer = factories.Transformer{
|
||||
Config: frob.FrobConfig,
|
||||
Converter: &frob.FrobConverter{},
|
||||
Repository: &frob.FrobRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewTransformer
|
||||
|
||||
FlopKickTransformerInitializer = factories.Transformer{
|
||||
Config: flop_kick.Config,
|
||||
Converter: &flop_kick.FlopKickConverter{},
|
||||
Repository: &flop_kick.FlopKickRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewTransformer
|
||||
|
||||
PitFileDebtCeilingTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: debt_ceiling.DebtCeilingFileConfig,
|
||||
Converter: &debt_ceiling.PitFileDebtCeilingConverter{},
|
||||
Repository: &debt_ceiling.PitFileDebtCeilingRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
PitFileIlkTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: ilk.IlkFileConfig,
|
||||
Converter: &ilk.PitFileIlkConverter{},
|
||||
Repository: &ilk.PitFileIlkRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
PriceFeedTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: price_feeds.PriceFeedConfig,
|
||||
Converter: &price_feeds.PriceFeedConverter{},
|
||||
Repository: &price_feeds.PriceFeedRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
TendTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: tend.TendConfig,
|
||||
Converter: &tend.TendConverter{},
|
||||
Repository: &tend.TendRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatInitTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_init.VatInitConfig,
|
||||
Converter: &vat_init.VatInitConverter{},
|
||||
Repository: &vat_init.VatInitRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatGrabTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_grab.VatGrabConfig,
|
||||
Converter: &vat_grab.VatGrabConverter{},
|
||||
Repository: &vat_grab.VatGrabRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatFoldTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_fold.VatFoldConfig,
|
||||
Converter: &vat_fold.VatFoldConverter{},
|
||||
Repository: &vat_fold.VatFoldRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatHealTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_heal.VatHealConfig,
|
||||
Converter: &vat_heal.VatHealConverter{},
|
||||
Repository: &vat_heal.VatHealRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatMoveTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_move.VatMoveConfig,
|
||||
Converter: &vat_move.VatMoveConverter{},
|
||||
Repository: &vat_move.VatMoveRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatSlipTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_slip.VatSlipConfig,
|
||||
Converter: &vat_slip.VatSlipConverter{},
|
||||
Repository: &vat_slip.VatSlipRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatTollTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_toll.VatTollConfig,
|
||||
Converter: &vat_toll.VatTollConverter{},
|
||||
Repository: &vat_toll.VatTollRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatTuneTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_tune.VatTuneConfig,
|
||||
Converter: &vat_tune.VatTuneConverter{},
|
||||
Repository: &vat_tune.VatTuneRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
|
||||
VatFluxTransformerInitializer = factories.LogNoteTransformer{
|
||||
Config: vat_flux.VatFluxConfig,
|
||||
Converter: &vat_flux.VatFluxConverter{},
|
||||
Repository: &vat_flux.VatFluxRepository{},
|
||||
Fetcher: &shared.Fetcher{},
|
||||
}.NewLogNoteTransformer
|
||||
)
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user