From 48f70d4ddf76f17a91501ebbeb6365c625873538 Mon Sep 17 00:00:00 2001 From: Ian Norden Date: Mon, 3 Feb 2020 12:22:29 -0600 Subject: [PATCH] btc mocks; reorg --- cmd/streamEthSubscribe.go | 3 +- cmd/superNode.go | 10 +- .../eth => ipfs}/mocks/blockservice.go | 0 .../eth => ipfs}/mocks/dag_putters.go | 0 pkg/super_node/backfiller.go | 3 +- pkg/super_node/backfiller_test.go | 13 +- pkg/super_node/btc/mocks/converter.go | 64 ++++++++++ .../mocks/indexer.go} | 28 ++++- pkg/super_node/btc/mocks/publisher.go | 65 ++++++++++ pkg/super_node/btc/subscription_config.go | 118 ++++++++++++++++++ pkg/super_node/constructors.go | 57 ++++----- pkg/super_node/eth/api.go | 4 +- pkg/super_node/eth/backend.go | 3 +- pkg/super_node/eth/filterer.go | 18 ++- pkg/super_node/eth/indexer.go | 3 +- pkg/super_node/eth/ipld_fetcher_test.go | 2 +- pkg/super_node/eth/models.go | 7 ++ pkg/super_node/eth/publisher_test.go | 21 ++-- pkg/super_node/eth/retriever.go | 17 ++- pkg/super_node/eth/retriever_test.go | 109 ++++++++-------- .../subscription_config.go} | 24 ++-- pkg/super_node/service.go | 5 +- pkg/super_node/service_test.go | 29 +++-- .../{config => shared}/chain_type.go | 2 +- pkg/super_node/{config => shared}/config.go | 12 +- pkg/super_node/shared/intefaces.go | 4 +- .../{eth => shared}/mocks/fetcher.go | 8 +- .../{eth => shared}/mocks/retriever.go | 0 .../{eth => shared}/mocks/streamer.go | 11 +- 29 files changed, 452 insertions(+), 188 deletions(-) rename pkg/{super_node/eth => ipfs}/mocks/blockservice.go (100%) rename pkg/{super_node/eth => ipfs}/mocks/dag_putters.go (100%) create mode 100644 pkg/super_node/btc/mocks/converter.go rename pkg/super_node/{config/btc_subscription.go => btc/mocks/indexer.go} (50%) create mode 100644 pkg/super_node/btc/mocks/publisher.go create mode 100644 pkg/super_node/btc/subscription_config.go rename pkg/super_node/{config/eth_subscription.go => eth/subscription_config.go} (88%) rename pkg/super_node/{config => shared}/chain_type.go (98%) rename pkg/super_node/{config => shared}/config.go (95%) rename pkg/super_node/{eth => shared}/mocks/fetcher.go (88%) rename pkg/super_node/{eth => shared}/mocks/retriever.go (100%) rename pkg/super_node/{eth => shared}/mocks/streamer.go (74%) diff --git a/cmd/streamEthSubscribe.go b/cmd/streamEthSubscribe.go index 24c79501..20051b97 100644 --- a/cmd/streamEthSubscribe.go +++ b/cmd/streamEthSubscribe.go @@ -32,7 +32,6 @@ import ( "github.com/vulcanize/vulcanizedb/pkg/eth/client" "github.com/vulcanize/vulcanizedb/pkg/eth/core" "github.com/vulcanize/vulcanizedb/pkg/super_node" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" ) @@ -55,7 +54,7 @@ func init() { func streamEthSubscription() { // Prep the subscription config/filters to be sent to the server - ethSubConfig, err := config.NewEthSubscriptionConfig() + ethSubConfig, err := eth.NewEthSubscriptionConfig() if err != nil { log.Fatal(err) } diff --git a/cmd/superNode.go b/cmd/superNode.go index d2088583..17251535 100644 --- a/cmd/superNode.go +++ b/cmd/superNode.go @@ -18,14 +18,12 @@ package cmd import ( "sync" - "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" - "github.com/ethereum/go-ethereum/rpc" log "github.com/sirupsen/logrus" "github.com/spf13/cobra" "github.com/vulcanize/vulcanizedb/pkg/super_node" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) // superNodeCmd represents the superNode command @@ -85,8 +83,8 @@ func superNode() { wg.Wait() } -func newSuperNode() (super_node.SuperNode, *config.SuperNode, error) { - superNodeConfig, err := config.NewSuperNodeConfig() +func newSuperNode() (super_node.SuperNode, *shared.SuperNodeConfig, error) { + superNodeConfig, err := shared.NewSuperNodeConfig() if err != nil { return nil, nil, err } @@ -97,7 +95,7 @@ func newSuperNode() (super_node.SuperNode, *config.SuperNode, error) { return sn, superNodeConfig, nil } -func startServers(superNode super_node.SuperNode, settings *config.SuperNode) error { +func startServers(superNode super_node.SuperNode, settings *shared.SuperNodeConfig) error { _, _, err := rpc.StartIPCEndpoint(settings.IPCEndpoint, superNode.APIs()) if err != nil { return err diff --git a/pkg/super_node/eth/mocks/blockservice.go b/pkg/ipfs/mocks/blockservice.go similarity index 100% rename from pkg/super_node/eth/mocks/blockservice.go rename to pkg/ipfs/mocks/blockservice.go diff --git a/pkg/super_node/eth/mocks/dag_putters.go b/pkg/ipfs/mocks/dag_putters.go similarity index 100% rename from pkg/super_node/eth/mocks/dag_putters.go rename to pkg/ipfs/mocks/dag_putters.go diff --git a/pkg/super_node/backfiller.go b/pkg/super_node/backfiller.go index a9ad071b..63b695b6 100644 --- a/pkg/super_node/backfiller.go +++ b/pkg/super_node/backfiller.go @@ -25,7 +25,6 @@ import ( log "github.com/sirupsen/logrus" "github.com/vulcanize/vulcanizedb/libraries/shared/storage/utils" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) @@ -59,7 +58,7 @@ type BackFillService struct { } // NewBackFillService returns a new BackFillInterface -func NewBackFillService(settings *config.SuperNode) (BackFillInterface, error) { +func NewBackFillService(settings *shared.SuperNodeConfig) (BackFillInterface, error) { publisher, err := NewIPLDPublisher(settings.Chain, settings.IPFSPath) if err != nil { return nil, err diff --git a/pkg/super_node/backfiller_test.go b/pkg/super_node/backfiller_test.go index c4d539d1..092e664f 100644 --- a/pkg/super_node/backfiller_test.go +++ b/pkg/super_node/backfiller_test.go @@ -27,6 +27,7 @@ import ( "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks" "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" + mocks2 "github.com/vulcanize/vulcanizedb/pkg/super_node/shared/mocks" ) var _ = Describe("BackFiller", func() { @@ -43,7 +44,7 @@ var _ = Describe("BackFiller", func() { ReturnIPLDPayload: []eth.IPLDPayload{mocks.MockIPLDPayload, mocks.MockIPLDPayload}, ReturnErr: nil, } - mockRetriever := &mocks.MockCIDRetriever{ + mockRetriever := &mocks2.MockCIDRetriever{ FirstBlockNumberToReturn: 1, GapsToRetrieve: []shared.Gap{ { @@ -51,7 +52,7 @@ var _ = Describe("BackFiller", func() { }, }, } - mockFetcher := &mocks.StateDiffFetcher{ + mockFetcher := &mocks2.IPLDFetcher{ PayloadsToReturn: map[uint64]shared.RawChainData{ 100: mocks.MockStateDiffPayload, 101: mocks.MockStateDiffPayload, @@ -97,7 +98,7 @@ var _ = Describe("BackFiller", func() { ReturnIPLDPayload: []eth.IPLDPayload{mocks.MockIPLDPayload}, ReturnErr: nil, } - mockRetriever := &mocks.MockCIDRetriever{ + mockRetriever := &mocks2.MockCIDRetriever{ FirstBlockNumberToReturn: 1, GapsToRetrieve: []shared.Gap{ { @@ -105,7 +106,7 @@ var _ = Describe("BackFiller", func() { }, }, } - mockFetcher := &mocks.StateDiffFetcher{ + mockFetcher := &mocks2.IPLDFetcher{ PayloadsToReturn: map[uint64]shared.RawChainData{ 100: mocks.MockStateDiffPayload, }, @@ -147,11 +148,11 @@ var _ = Describe("BackFiller", func() { ReturnIPLDPayload: []eth.IPLDPayload{mocks.MockIPLDPayload, mocks.MockIPLDPayload}, ReturnErr: nil, } - mockRetriever := &mocks.MockCIDRetriever{ + mockRetriever := &mocks2.MockCIDRetriever{ FirstBlockNumberToReturn: 3, GapsToRetrieve: []shared.Gap{}, } - mockFetcher := &mocks.StateDiffFetcher{ + mockFetcher := &mocks2.IPLDFetcher{ PayloadsToReturn: map[uint64]shared.RawChainData{ 1: mocks.MockStateDiffPayload, 2: mocks.MockStateDiffPayload, diff --git a/pkg/super_node/btc/mocks/converter.go b/pkg/super_node/btc/mocks/converter.go new file mode 100644 index 00000000..8f7f2ccd --- /dev/null +++ b/pkg/super_node/btc/mocks/converter.go @@ -0,0 +1,64 @@ +// VulcanizeDB +// Copyright © 2019 Vulcanize + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. + +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package mocks + +import ( + "fmt" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/btc" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" +) + +// PayloadConverter is the underlying struct for the Converter interface +type PayloadConverter struct { + PassedStatediffPayload btc.BlockPayload + ReturnIPLDPayload btc.IPLDPayload + ReturnErr error +} + +// Convert method is used to convert a geth statediff.Payload to a IPLDPayload +func (pc *PayloadConverter) Convert(payload shared.RawChainData) (shared.StreamedIPLDs, error) { + stateDiffPayload, ok := payload.(btc.BlockPayload) + if !ok { + return nil, fmt.Errorf("convert expected payload type %T got %T", btc.BlockPayload{}, payload) + } + pc.PassedStatediffPayload = stateDiffPayload + return pc.ReturnIPLDPayload, pc.ReturnErr +} + +// IterativePayloadConverter is the underlying struct for the Converter interface +type IterativePayloadConverter struct { + PassedStatediffPayload []btc.BlockPayload + ReturnIPLDPayload []btc.IPLDPayload + ReturnErr error + iteration int +} + +// Convert method is used to convert a geth statediff.Payload to a IPLDPayload +func (pc *IterativePayloadConverter) Convert(payload shared.RawChainData) (shared.StreamedIPLDs, error) { + stateDiffPayload, ok := payload.(btc.BlockPayload) + if !ok { + return nil, fmt.Errorf("convert expected payload type %T got %T", btc.BlockPayload{}, payload) + } + pc.PassedStatediffPayload = append(pc.PassedStatediffPayload, stateDiffPayload) + if len(pc.PassedStatediffPayload) < pc.iteration+1 { + return nil, fmt.Errorf("IterativePayloadConverter does not have a payload to return at iteration %d", pc.iteration) + } + returnPayload := pc.ReturnIPLDPayload[pc.iteration] + pc.iteration++ + return returnPayload, pc.ReturnErr +} diff --git a/pkg/super_node/config/btc_subscription.go b/pkg/super_node/btc/mocks/indexer.go similarity index 50% rename from pkg/super_node/config/btc_subscription.go rename to pkg/super_node/btc/mocks/indexer.go index 50d7e290..699b7076 100644 --- a/pkg/super_node/config/btc_subscription.go +++ b/pkg/super_node/btc/mocks/indexer.go @@ -14,4 +14,30 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -package config +package mocks + +import ( + "fmt" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/btc" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" +) + +// CIDIndexer is the underlying struct for the Indexer interface +type CIDIndexer struct { + PassedCIDPayload []*btc.CIDPayload + ReturnErr error +} + +// Index indexes a cidPayload in Postgres +func (repo *CIDIndexer) Index(cids shared.CIDsForIndexing) error { + cidPayload, ok := cids.(*btc.CIDPayload) + if !ok { + return fmt.Errorf("index expected cids type %T got %T", ð.CIDPayload{}, cids) + } + repo.PassedCIDPayload = append(repo.PassedCIDPayload, cidPayload) + return repo.ReturnErr +} diff --git a/pkg/super_node/btc/mocks/publisher.go b/pkg/super_node/btc/mocks/publisher.go new file mode 100644 index 00000000..dc15fe52 --- /dev/null +++ b/pkg/super_node/btc/mocks/publisher.go @@ -0,0 +1,65 @@ +// VulcanizeDB +// Copyright © 2019 Vulcanize + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. + +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package mocks + +import ( + "fmt" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/btc" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" +) + +// IPLDPublisher is the underlying struct for the Publisher interface +type IPLDPublisher struct { + PassedIPLDPayload btc.IPLDPayload + ReturnCIDPayload *btc.CIDPayload + ReturnErr error +} + +// Publish publishes an IPLDPayload to IPFS and returns the corresponding CIDPayload +func (pub *IPLDPublisher) Publish(payload shared.StreamedIPLDs) (shared.CIDsForIndexing, error) { + ipldPayload, ok := payload.(btc.IPLDPayload) + if !ok { + return nil, fmt.Errorf("publish expected payload type %T got %T", &btc.IPLDPayload{}, payload) + } + pub.PassedIPLDPayload = ipldPayload + return pub.ReturnCIDPayload, pub.ReturnErr +} + +// IterativeIPLDPublisher is the underlying struct for the Publisher interface; used in testing +type IterativeIPLDPublisher struct { + PassedIPLDPayload []btc.IPLDPayload + ReturnCIDPayload []*btc.CIDPayload + ReturnErr error + iteration int +} + +// Publish publishes an IPLDPayload to IPFS and returns the corresponding CIDPayload +func (pub *IterativeIPLDPublisher) Publish(payload shared.StreamedIPLDs) (shared.CIDsForIndexing, error) { + ipldPayload, ok := payload.(btc.IPLDPayload) + if !ok { + return nil, fmt.Errorf("publish expected payload type %T got %T", &btc.IPLDPayload{}, payload) + } + pub.PassedIPLDPayload = append(pub.PassedIPLDPayload, ipldPayload) + if len(pub.ReturnCIDPayload) < pub.iteration+1 { + return nil, fmt.Errorf("IterativeIPLDPublisher does not have a payload to return at iteration %d", pub.iteration) + } + returnPayload := pub.ReturnCIDPayload[pub.iteration] + pub.iteration++ + return returnPayload, pub.ReturnErr +} diff --git a/pkg/super_node/btc/subscription_config.go b/pkg/super_node/btc/subscription_config.go new file mode 100644 index 00000000..8e4a59b2 --- /dev/null +++ b/pkg/super_node/btc/subscription_config.go @@ -0,0 +1,118 @@ +// VulcanizeDB +// Copyright © 2019 Vulcanize + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. + +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package btc + +import ( + "math/big" + + "github.com/spf13/viper" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" +) + +/* +// HeaderModel is the db model for btc.header_cids table +// TxInput is the db model for btc.tx_inputs table +type TxInput struct { + ID int64 `db:"id"` + TxID int64 `db:"tx_id"` + Index int64 `db:"index"` + TxWitness [][]byte `db:"tx_witness"` + SignatureScript []byte `db:"sig_script"` + PreviousOutPointHash string `db:"outpoint_hash"` + PreviousOutPointIndex uint32 `db:"outpoint_index"` +} + +// TxOutput is the db model for btc.tx_outputs table +type TxOutput struct { + ID int64 `db:"id"` + TxID int64 `db:"tx_id"` + Index int64 `db:"index"` + Value int64 `db:"value"` + PkScript []byte `db:"pk_script"` +} + +*/ +// SubscriptionSettings config is used by a subscriber to specify what bitcoin data to stream from the super node +type SubscriptionSettings struct { + BackFill bool + BackFillOnly bool + Start *big.Int + End *big.Int // set to 0 or a negative value to have no ending block + HeaderFilter HeaderFilter + TxFilter TxFilter +} + +// HeaderFilter contains filter settings for headers +type HeaderFilter struct { + Off bool +} + +// TxFilter contains filter settings for txs +type TxFilter struct { + Off bool + Index int64 // allow filtering by index so that we can filter for only coinbase transactions (index 0) if we want to + HasWitness bool + WitnessHashes []string +} + +// Init is used to initialize a EthSubscription struct with env variables +func NewEthSubscriptionConfig() (*SubscriptionSettings, error) { + sc := new(SubscriptionSettings) + // Below default to false, which means we do not backfill by default + sc.BackFill = viper.GetBool("superNode.ethSubscription.historicalData") + sc.BackFillOnly = viper.GetBool("superNode.ethSubscription.historicalDataOnly") + // Below default to 0 + // 0 start means we start at the beginning and 0 end means we continue indefinitely + sc.Start = big.NewInt(viper.GetInt64("superNode.ethSubscription.startingBlock")) + sc.End = big.NewInt(viper.GetInt64("superNode.ethSubscription.endingBlock")) + // Below default to false, which means we get all headers and no uncles by default + sc.HeaderFilter = HeaderFilter{ + Off: viper.GetBool("superNode.ethSubscription.off"), + } + // Below defaults to false and two slices of length 0 + // Which means we get all transactions by default + sc.TxFilter = TxFilter{ + Off: viper.GetBool("superNode.ethSubscription.txFilter.off"), + } + return sc, nil +} + +// StartingBlock satisfies the SubscriptionSettings() interface +func (sc *SubscriptionSettings) StartingBlock() *big.Int { + return sc.Start +} + +// EndingBlock satisfies the SubscriptionSettings() interface +func (sc *SubscriptionSettings) EndingBlock() *big.Int { + return sc.End +} + +// HistoricalData satisfies the SubscriptionSettings() interface +func (sc *SubscriptionSettings) HistoricalData() bool { + return sc.BackFill +} + +// HistoricalDataOnly satisfies the SubscriptionSettings() interface +func (sc *SubscriptionSettings) HistoricalDataOnly() bool { + return sc.BackFillOnly +} + +// ChainType satisfies the SubscriptionSettings() interface +func (sc *SubscriptionSettings) ChainType() shared.ChainType { + return shared.Bitcoin +} diff --git a/pkg/super_node/constructors.go b/pkg/super_node/constructors.go index 219c839d..210966b1 100644 --- a/pkg/super_node/constructors.go +++ b/pkg/super_node/constructors.go @@ -20,23 +20,20 @@ import ( "fmt" "github.com/btcsuite/btcd/rpcclient" - "github.com/vulcanize/vulcanizedb/pkg/super_node/btc" - - "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" - "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rpc" "github.com/vulcanize/vulcanizedb/pkg/eth/core" "github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" + "github.com/vulcanize/vulcanizedb/pkg/super_node/btc" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) // NewResponseFilterer constructs a ResponseFilterer for the provided chain type -func NewResponseFilterer(chain config.ChainType) (shared.ResponseFilterer, error) { +func NewResponseFilterer(chain shared.ChainType) (shared.ResponseFilterer, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewResponseFilterer(), nil default: return nil, fmt.Errorf("invalid chain %T for filterer constructor", chain) @@ -44,11 +41,11 @@ func NewResponseFilterer(chain config.ChainType) (shared.ResponseFilterer, error } // NewCIDIndexer constructs a CIDIndexer for the provided chain type -func NewCIDIndexer(chain config.ChainType, db *postgres.DB) (shared.CIDIndexer, error) { +func NewCIDIndexer(chain shared.ChainType, db *postgres.DB) (shared.CIDIndexer, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewCIDIndexer(db), nil - case config.Bitcoin: + case shared.Bitcoin: return btc.NewCIDIndexer(db), nil default: return nil, fmt.Errorf("invalid chain %T for indexer constructor", chain) @@ -56,9 +53,9 @@ func NewCIDIndexer(chain config.ChainType, db *postgres.DB) (shared.CIDIndexer, } // NewCIDRetriever constructs a CIDRetriever for the provided chain type -func NewCIDRetriever(chain config.ChainType, db *postgres.DB) (shared.CIDRetriever, error) { +func NewCIDRetriever(chain shared.ChainType, db *postgres.DB) (shared.CIDRetriever, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewCIDRetriever(db), nil default: return nil, fmt.Errorf("invalid chain %T for retriever constructor", chain) @@ -66,9 +63,9 @@ func NewCIDRetriever(chain config.ChainType, db *postgres.DB) (shared.CIDRetriev } // NewPayloadStreamer constructs a PayloadStreamer for the provided chain type -func NewPayloadStreamer(chain config.ChainType, client interface{}) (shared.PayloadStreamer, chan shared.RawChainData, error) { +func NewPayloadStreamer(chain shared.ChainType, client interface{}) (shared.PayloadStreamer, chan shared.RawChainData, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: ethClient, ok := client.(core.RPCClient) if !ok { var expectedClientType core.RPCClient @@ -76,10 +73,10 @@ func NewPayloadStreamer(chain config.ChainType, client interface{}) (shared.Payl } streamChan := make(chan shared.RawChainData, eth.PayloadChanBufferSize) return eth.NewPayloadStreamer(ethClient), streamChan, nil - case config.Bitcoin: + case shared.Bitcoin: btcClientConn, ok := client.(*rpcclient.ConnConfig) if !ok { - return nil, nil, fmt.Errorf("bitcoin payload streamer constructor expected client config type %T got %T", rpcclient.ConnConfig{}, client) + return nil, nil, fmt.Errorf("bitcoin payload streamer constructor expected client shared type %T got %T", rpcclient.ConnConfig{}, client) } streamChan := make(chan shared.RawChainData, btc.PayloadChanBufferSize) return btc.NewPayloadStreamer(btcClientConn), streamChan, nil @@ -89,9 +86,9 @@ func NewPayloadStreamer(chain config.ChainType, client interface{}) (shared.Payl } // NewPaylaodFetcher constructs a PayloadFetcher for the provided chain type -func NewPaylaodFetcher(chain config.ChainType, client interface{}) (shared.PayloadFetcher, error) { +func NewPaylaodFetcher(chain shared.ChainType, client interface{}) (shared.PayloadFetcher, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: batchClient, ok := client.(eth.BatchClient) if !ok { var expectedClient eth.BatchClient @@ -104,11 +101,11 @@ func NewPaylaodFetcher(chain config.ChainType, client interface{}) (shared.Paylo } // NewPayloadConverter constructs a PayloadConverter for the provided chain type -func NewPayloadConverter(chain config.ChainType) (shared.PayloadConverter, error) { +func NewPayloadConverter(chain shared.ChainType) (shared.PayloadConverter, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewPayloadConverter(params.MainnetChainConfig), nil - case config.Bitcoin: + case shared.Bitcoin: return btc.NewPayloadConverter(), nil default: return nil, fmt.Errorf("invalid chain %T for converter constructor", chain) @@ -116,9 +113,9 @@ func NewPayloadConverter(chain config.ChainType) (shared.PayloadConverter, error } // NewIPLDFetcher constructs an IPLDFetcher for the provided chain type -func NewIPLDFetcher(chain config.ChainType, ipfsPath string) (shared.IPLDFetcher, error) { +func NewIPLDFetcher(chain shared.ChainType, ipfsPath string) (shared.IPLDFetcher, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewIPLDFetcher(ipfsPath) default: return nil, fmt.Errorf("invalid chain %T for fetcher constructor", chain) @@ -126,11 +123,11 @@ func NewIPLDFetcher(chain config.ChainType, ipfsPath string) (shared.IPLDFetcher } // NewIPLDPublisher constructs an IPLDPublisher for the provided chain type -func NewIPLDPublisher(chain config.ChainType, ipfsPath string) (shared.IPLDPublisher, error) { +func NewIPLDPublisher(chain shared.ChainType, ipfsPath string) (shared.IPLDPublisher, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewIPLDPublisher(ipfsPath) - case config.Bitcoin: + case shared.Bitcoin: return btc.NewIPLDPublisher(ipfsPath) default: return nil, fmt.Errorf("invalid chain %T for publisher constructor", chain) @@ -138,9 +135,9 @@ func NewIPLDPublisher(chain config.ChainType, ipfsPath string) (shared.IPLDPubli } // NewIPLDResolver constructs an IPLDResolver for the provided chain type -func NewIPLDResolver(chain config.ChainType) (shared.IPLDResolver, error) { +func NewIPLDResolver(chain shared.ChainType) (shared.IPLDResolver, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: return eth.NewIPLDResolver(), nil default: return nil, fmt.Errorf("invalid chain %T for resolver constructor", chain) @@ -148,9 +145,9 @@ func NewIPLDResolver(chain config.ChainType) (shared.IPLDResolver, error) { } // NewPublicAPI constructs a PublicAPI for the provided chain type -func NewPublicAPI(chain config.ChainType, db *postgres.DB, ipfsPath string) (rpc.API, error) { +func NewPublicAPI(chain shared.ChainType, db *postgres.DB, ipfsPath string) (rpc.API, error) { switch chain { - case config.Ethereum: + case shared.Ethereum: backend, err := eth.NewEthBackend(db, ipfsPath) if err != nil { return rpc.API{}, err diff --git a/pkg/super_node/eth/api.go b/pkg/super_node/eth/api.go index ec712d6d..7418c4d8 100644 --- a/pkg/super_node/eth/api.go +++ b/pkg/super_node/eth/api.go @@ -27,8 +27,6 @@ import ( "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" "github.com/ipfs/go-block-format" - - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" ) // APIName is the namespace for the super node's eth api @@ -72,7 +70,7 @@ func (pea *PublicEthAPI) GetLogs(ctx context.Context, crit ethereum.FilterQuery) topicStrSets[i] = append(topicStrSets[i], topic.String()) } } - filter := config.ReceiptFilter{ + filter := ReceiptFilter{ Contracts: addrStrs, Topics: topicStrSets, } diff --git a/pkg/super_node/eth/backend.go b/pkg/super_node/eth/backend.go index 476745b5..8edc7e2d 100644 --- a/pkg/super_node/eth/backend.go +++ b/pkg/super_node/eth/backend.go @@ -29,7 +29,6 @@ import ( "github.com/sirupsen/logrus" "github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" ) var ( @@ -123,7 +122,7 @@ func (b *Backend) GetLogs(ctx context.Context, hash common.Hash) ([][]*types.Log if err != nil { return nil, err } - receiptCIDs, err := b.retriever.RetrieveRctCIDs(tx, config.ReceiptFilter{}, 0, &hash, nil) + receiptCIDs, err := b.retriever.RetrieveRctCIDs(tx, ReceiptFilter{}, 0, &hash, nil) if err != nil { if err := tx.Rollback(); err != nil { logrus.Error(err) diff --git a/pkg/super_node/eth/filterer.go b/pkg/super_node/eth/filterer.go index 33c9f918..3fb23016 100644 --- a/pkg/super_node/eth/filterer.go +++ b/pkg/super_node/eth/filterer.go @@ -20,14 +20,12 @@ import ( "bytes" "fmt" - "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/rlp" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) // ResponseFilterer satisfies the ResponseFilterer interface for ethereum @@ -40,9 +38,9 @@ func NewResponseFilterer() *ResponseFilterer { // Filter is used to filter through eth data to extract and package requested data into a Payload func (s *ResponseFilterer) Filter(filter shared.SubscriptionSettings, payload shared.StreamedIPLDs) (shared.ServerResponse, error) { - ethFilters, ok := filter.(*config.EthSubscription) + ethFilters, ok := filter.(*SubscriptionSettings) if !ok { - return StreamResponse{}, fmt.Errorf("eth filterer expected filter type %T got %T", &config.EthSubscription{}, filter) + return StreamResponse{}, fmt.Errorf("eth filterer expected filter type %T got %T", &SubscriptionSettings{}, filter) } ethPayload, ok := payload.(IPLDPayload) if !ok { @@ -76,7 +74,7 @@ func (s *ResponseFilterer) Filter(filter shared.SubscriptionSettings, payload sh return StreamResponse{}, nil } -func (s *ResponseFilterer) filterHeaders(headerFilter config.HeaderFilter, response *StreamResponse, payload IPLDPayload) error { +func (s *ResponseFilterer) filterHeaders(headerFilter HeaderFilter, response *StreamResponse, payload IPLDPayload) error { if !headerFilter.Off { headerRLP, err := rlp.EncodeToBytes(payload.Block.Header()) if err != nil { @@ -104,7 +102,7 @@ func checkRange(start, end, actual int64) bool { return false } -func (s *ResponseFilterer) filterTransactions(trxFilter config.TxFilter, response *StreamResponse, payload IPLDPayload) ([]common.Hash, error) { +func (s *ResponseFilterer) filterTransactions(trxFilter TxFilter, response *StreamResponse, payload IPLDPayload) ([]common.Hash, error) { trxHashes := make([]common.Hash, 0, len(payload.Block.Body().Transactions)) if !trxFilter.Off { for i, trx := range payload.Block.Body().Transactions { @@ -139,7 +137,7 @@ func checkTransactions(wantedSrc, wantedDst []string, actualSrc, actualDst strin return false } -func (s *ResponseFilterer) filerReceipts(receiptFilter config.ReceiptFilter, response *StreamResponse, payload IPLDPayload, trxHashes []common.Hash) error { +func (s *ResponseFilterer) filerReceipts(receiptFilter ReceiptFilter, response *StreamResponse, payload IPLDPayload, trxHashes []common.Hash) error { if !receiptFilter.Off { for i, receipt := range payload.Receipts { // topics is always length 4 @@ -223,7 +221,7 @@ func slicesShareString(slice1, slice2 []string) int { return 0 } -func (s *ResponseFilterer) filterState(stateFilter config.StateFilter, response *StreamResponse, payload IPLDPayload) error { +func (s *ResponseFilterer) filterState(stateFilter StateFilter, response *StreamResponse, payload IPLDPayload) error { if !stateFilter.Off { response.StateNodesRlp = make(map[common.Hash][]byte) keyFilters := make([]common.Hash, len(stateFilter.Addresses)) @@ -254,7 +252,7 @@ func checkNodeKeys(wantedKeys []common.Hash, actualKey common.Hash) bool { return false } -func (s *ResponseFilterer) filterStorage(storageFilter config.StorageFilter, response *StreamResponse, payload IPLDPayload) error { +func (s *ResponseFilterer) filterStorage(storageFilter StorageFilter, response *StreamResponse, payload IPLDPayload) error { if !storageFilter.Off { response.StorageNodesRlp = make(map[common.Hash]map[common.Hash][]byte) stateKeyFilters := make([]common.Hash, len(storageFilter.Addresses)) diff --git a/pkg/super_node/eth/indexer.go b/pkg/super_node/eth/indexer.go index 9ea10c84..da0c664a 100644 --- a/pkg/super_node/eth/indexer.go +++ b/pkg/super_node/eth/indexer.go @@ -23,7 +23,6 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/jmoiron/sqlx" - "github.com/lib/pq" log "github.com/sirupsen/logrus" "github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres" @@ -119,7 +118,7 @@ func (in *CIDIndexer) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *CIDPa func (in *CIDIndexer) indexReceiptCID(tx *sqlx.Tx, cidMeta ReceiptModel, txID int64) error { _, err := tx.Exec(`INSERT INTO eth.receipt_cids (tx_id, cid, contract, topic0s, topic1s, topic2s, topic3s) VALUES ($1, $2, $3, $4, $5, $6, $7)`, - txID, cidMeta.CID, cidMeta.Contract, pq.Array(cidMeta.Topic0s), pq.Array(cidMeta.Topic1s), pq.Array(cidMeta.Topic2s), pq.Array(cidMeta.Topic3s)) + txID, cidMeta.CID, cidMeta.Contract, cidMeta.Topic0s, cidMeta.Topic1s, cidMeta.Topic2s, cidMeta.Topic3s) return err } diff --git a/pkg/super_node/eth/ipld_fetcher_test.go b/pkg/super_node/eth/ipld_fetcher_test.go index 1c5869c1..ee0d954c 100644 --- a/pkg/super_node/eth/ipld_fetcher_test.go +++ b/pkg/super_node/eth/ipld_fetcher_test.go @@ -24,8 +24,8 @@ import ( . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" + "github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" - "github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks" ) var ( diff --git a/pkg/super_node/eth/models.go b/pkg/super_node/eth/models.go index 0cc931e5..2a9e7309 100644 --- a/pkg/super_node/eth/models.go +++ b/pkg/super_node/eth/models.go @@ -18,6 +18,7 @@ package eth import "github.com/lib/pq" +// HeaderModel is the db model for eth.header_cids type HeaderModel struct { ID int64 `db:"id"` BlockNumber string `db:"block_number"` @@ -27,6 +28,7 @@ type HeaderModel struct { TotalDifficulty string `db:"td"` } +// UncleModel is the db model for eth.uncle_cids type UncleModel struct { ID int64 `db:"id"` HeaderID int64 `db:"header_id"` @@ -35,6 +37,7 @@ type UncleModel struct { CID string `db:"cid"` } +// TxModel is the db model for eth.transaction_cids type TxModel struct { ID int64 `db:"id"` HeaderID int64 `db:"header_id"` @@ -45,6 +48,7 @@ type TxModel struct { Src string `db:"src"` } +// ReceiptModel is the db model for eth.receipt_cids type ReceiptModel struct { ID int64 `db:"id"` TxID int64 `db:"tx_id"` @@ -56,6 +60,7 @@ type ReceiptModel struct { Topic3s pq.StringArray `db:"topic3s"` } +// StateNodeModel is the db model for eth.state_cids type StateNodeModel struct { ID int64 `db:"id"` HeaderID int64 `db:"header_id"` @@ -64,6 +69,7 @@ type StateNodeModel struct { CID string `db:"cid"` } +// StorageNodeModel is the db model for eth.storage_cids type StorageNodeModel struct { ID int64 `db:"id"` StateID int64 `db:"state_id"` @@ -72,6 +78,7 @@ type StorageNodeModel struct { CID string `db:"cid"` } +// StorageNodeWithStateKeyModel is a db model for eth.storage_cids + eth.state_cids.state_key type StorageNodeWithStateKeyModel struct { ID int64 `db:"id"` StateID int64 `db:"state_id"` diff --git a/pkg/super_node/eth/publisher_test.go b/pkg/super_node/eth/publisher_test.go index 65d9872f..e7292f49 100644 --- a/pkg/super_node/eth/publisher_test.go +++ b/pkg/super_node/eth/publisher_test.go @@ -21,25 +21,26 @@ import ( . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" + mocks2 "github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks" ) var ( - mockHeaderDagPutter *mocks.DagPutter - mockTrxDagPutter *mocks.DagPutter - mockRctDagPutter *mocks.DagPutter - mockStateDagPutter *mocks.MappedDagPutter - mockStorageDagPutter *mocks.DagPutter + mockHeaderDagPutter *mocks2.DagPutter + mockTrxDagPutter *mocks2.DagPutter + mockRctDagPutter *mocks2.DagPutter + mockStateDagPutter *mocks2.MappedDagPutter + mockStorageDagPutter *mocks2.DagPutter ) var _ = Describe("Publisher", func() { BeforeEach(func() { - mockHeaderDagPutter = new(mocks.DagPutter) - mockTrxDagPutter = new(mocks.DagPutter) - mockRctDagPutter = new(mocks.DagPutter) - mockStateDagPutter = new(mocks.MappedDagPutter) - mockStorageDagPutter = new(mocks.DagPutter) + mockHeaderDagPutter = new(mocks2.DagPutter) + mockTrxDagPutter = new(mocks2.DagPutter) + mockRctDagPutter = new(mocks2.DagPutter) + mockStateDagPutter = new(mocks2.MappedDagPutter) + mockStorageDagPutter = new(mocks2.DagPutter) }) Describe("Publish", func() { diff --git a/pkg/super_node/eth/retriever.go b/pkg/super_node/eth/retriever.go index 8b4dc000..12552c16 100644 --- a/pkg/super_node/eth/retriever.go +++ b/pkg/super_node/eth/retriever.go @@ -27,7 +27,6 @@ import ( log "github.com/sirupsen/logrus" "github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) @@ -59,9 +58,9 @@ func (ecr *CIDRetriever) RetrieveLastBlockNumber() (int64, error) { // Retrieve is used to retrieve all of the CIDs which conform to the passed StreamFilters func (ecr *CIDRetriever) Retrieve(filter shared.SubscriptionSettings, blockNumber int64) (shared.CIDsForFetching, bool, error) { - streamFilter, ok := filter.(*config.EthSubscription) + streamFilter, ok := filter.(*SubscriptionSettings) if !ok { - return nil, true, fmt.Errorf("eth retriever expected filter type %T got %T", &config.EthSubscription{}, filter) + return nil, true, fmt.Errorf("eth retriever expected filter type %T got %T", &SubscriptionSettings{}, filter) } log.Debug("retrieving cids") tx, err := ecr.db.Beginx() @@ -173,7 +172,7 @@ func (ecr *CIDRetriever) RetrieveUncleCIDsByHeaderID(tx *sqlx.Tx, headerID int64 // RetrieveTxCIDs retrieves and returns all of the trx cids at the provided blockheight that conform to the provided filter parameters // also returns the ids for the returned transaction cids -func (ecr *CIDRetriever) RetrieveTxCIDs(tx *sqlx.Tx, txFilter config.TxFilter, blockNumber int64) ([]TxModel, error) { +func (ecr *CIDRetriever) RetrieveTxCIDs(tx *sqlx.Tx, txFilter TxFilter, blockNumber int64) ([]TxModel, error) { log.Debug("retrieving transaction cids for block ", blockNumber) args := make([]interface{}, 0, 3) results := make([]TxModel, 0) @@ -196,7 +195,7 @@ func (ecr *CIDRetriever) RetrieveTxCIDs(tx *sqlx.Tx, txFilter config.TxFilter, b // RetrieveRctCIDs retrieves and returns all of the rct cids at the provided blockheight that conform to the provided // filter parameters and correspond to the provided tx ids -func (ecr *CIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter config.ReceiptFilter, blockNumber int64, blockHash *common.Hash, trxIds []int64) ([]ReceiptModel, error) { +func (ecr *CIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter ReceiptFilter, blockNumber int64, blockHash *common.Hash, trxIds []int64) ([]ReceiptModel, error) { log.Debug("retrieving receipt cids for block ", blockNumber) id := 1 args := make([]interface{}, 0, 4) @@ -282,7 +281,7 @@ func (ecr *CIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter config.ReceiptFi } // RetrieveStateCIDs retrieves and returns all of the state node cids at the provided blockheight that conform to the provided filter parameters -func (ecr *CIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter config.StateFilter, blockNumber int64) ([]StateNodeModel, error) { +func (ecr *CIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter StateFilter, blockNumber int64) ([]StateNodeModel, error) { log.Debug("retrieving state cids for block ", blockNumber) args := make([]interface{}, 0, 2) pgStr := `SELECT state_cids.id, state_cids.header_id, @@ -307,7 +306,7 @@ func (ecr *CIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter config.State } // RetrieveStorageCIDs retrieves and returns all of the storage node cids at the provided blockheight that conform to the provided filter parameters -func (ecr *CIDRetriever) RetrieveStorageCIDs(tx *sqlx.Tx, storageFilter config.StorageFilter, blockNumber int64) ([]StorageNodeWithStateKeyModel, error) { +func (ecr *CIDRetriever) RetrieveStorageCIDs(tx *sqlx.Tx, storageFilter StorageFilter, blockNumber int64) ([]StorageNodeWithStateKeyModel, error) { log.Debug("retrieving storage cids for block ", blockNumber) args := make([]interface{}, 0, 3) pgStr := `SELECT storage_cids.id, storage_cids.state_id, storage_cids.storage_key, @@ -472,7 +471,7 @@ func (ecr *CIDRetriever) RetrieveHeaderCIDByHash(tx *sqlx.Tx, blockHash common.H func (ecr *CIDRetriever) RetrieveTxCIDsByHeaderID(tx *sqlx.Tx, headerID int64) ([]TxModel, error) { log.Debug("retrieving tx cids for block id ", headerID) pgStr := `SELECT * FROM eth.transaction_cids - WHERE eth.transaction_cids.header_id = $1` + WHERE header_id = $1` var txCIDs []TxModel return txCIDs, tx.Select(&txCIDs, pgStr, headerID) } @@ -481,7 +480,7 @@ func (ecr *CIDRetriever) RetrieveTxCIDsByHeaderID(tx *sqlx.Tx, headerID int64) ( func (ecr *CIDRetriever) RetrieveReceiptCIDsByTxIDs(tx *sqlx.Tx, txIDs []int64) ([]ReceiptModel, error) { log.Debugf("retrieving receipt cids for tx ids %v", txIDs) pgStr := `SELECT * FROM eth.receipt_cids - WHERE eth.receipt_cids.tx_id = ANY($1::INTEGER[])` + WHERE tx_id = ANY($1::INTEGER[])` var rctCIDs []ReceiptModel return rctCIDs, tx.Select(&rctCIDs, pgStr, pq.Array(txIDs)) } diff --git a/pkg/super_node/eth/retriever_test.go b/pkg/super_node/eth/retriever_test.go index 5311794f..17c1da37 100644 --- a/pkg/super_node/eth/retriever_test.go +++ b/pkg/super_node/eth/retriever_test.go @@ -23,7 +23,6 @@ import ( . "github.com/onsi/gomega" "github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" eth2 "github.com/vulcanize/vulcanizedb/pkg/super_node/eth" "github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks" @@ -31,175 +30,175 @@ import ( ) var ( - openFilter = &config.EthSubscription{ + openFilter = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{}, - TxFilter: config.TxFilter{}, - ReceiptFilter: config.ReceiptFilter{}, - StateFilter: config.StateFilter{}, - StorageFilter: config.StorageFilter{}, + HeaderFilter: eth.HeaderFilter{}, + TxFilter: eth.TxFilter{}, + ReceiptFilter: eth.ReceiptFilter{}, + StateFilter: eth.StateFilter{}, + StorageFilter: eth.StorageFilter{}, } - rctContractFilter = &config.EthSubscription{ + rctContractFilter = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Off: true, }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ Contracts: []string{mocks.AnotherAddress.String()}, }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - rctTopicsFilter = &config.EthSubscription{ + rctTopicsFilter = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Off: true, }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ Topics: [][]string{{"0x0000000000000000000000000000000000000000000000000000000000000004"}}, }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - rctTopicsAndContractFilter = &config.EthSubscription{ + rctTopicsAndContractFilter = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Off: true, }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ Topics: [][]string{ {"0x0000000000000000000000000000000000000000000000000000000000000004"}, {"0x0000000000000000000000000000000000000000000000000000000000000006"}, }, Contracts: []string{mocks.Address.String()}, }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - rctTopicsAndContractFilterFail = &config.EthSubscription{ + rctTopicsAndContractFilterFail = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Off: true, }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ Topics: [][]string{ {"0x0000000000000000000000000000000000000000000000000000000000000004"}, {"0x0000000000000000000000000000000000000000000000000000000000000007"}, // This topic won't match on the mocks.Address.String() contract receipt }, Contracts: []string{mocks.Address.String()}, }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - rctContractsAndTopicFilter = &config.EthSubscription{ + rctContractsAndTopicFilter = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Off: true, }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ Topics: [][]string{{"0x0000000000000000000000000000000000000000000000000000000000000005"}}, Contracts: []string{mocks.Address.String(), mocks.AnotherAddress.String()}, }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - rctsForAllCollectedTrxs = &config.EthSubscription{ + rctsForAllCollectedTrxs = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{}, // Trx filter open so we will collect all trxs, therefore we will also collect all corresponding rcts despite rct filter - ReceiptFilter: config.ReceiptFilter{ + TxFilter: eth.TxFilter{}, // Trx filter open so we will collect all trxs, therefore we will also collect all corresponding rcts despite rct filter + ReceiptFilter: eth.ReceiptFilter{ MatchTxs: true, Topics: [][]string{{"0x0000000000000000000000000000000000000000000000000000000000000006"}}, // Topic0 isn't one of the topic0s we have Contracts: []string{"0x0000000000000000000000000000000000000002"}, // Contract isn't one of the contracts we have }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - rctsForSelectCollectedTrxs = &config.EthSubscription{ + rctsForSelectCollectedTrxs = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Dst: []string{mocks.AnotherAddress.String()}, // We only filter for one of the trxs so we will only get the one corresponding receipt }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ MatchTxs: true, Topics: [][]string{{"0x0000000000000000000000000000000000000000000000000000000000000006"}}, // Topic0 isn't one of the topic0s we have Contracts: []string{"0x0000000000000000000000000000000000000002"}, // Contract isn't one of the contracts we have }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Off: true, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } - stateFilter = &config.EthSubscription{ + stateFilter = ð.SubscriptionSettings{ Start: big.NewInt(0), End: big.NewInt(1), - HeaderFilter: config.HeaderFilter{ + HeaderFilter: eth.HeaderFilter{ Off: true, }, - TxFilter: config.TxFilter{ + TxFilter: eth.TxFilter{ Off: true, }, - ReceiptFilter: config.ReceiptFilter{ + ReceiptFilter: eth.ReceiptFilter{ Off: true, }, - StateFilter: config.StateFilter{ + StateFilter: eth.StateFilter{ Addresses: []string{mocks.Address.Hex()}, }, - StorageFilter: config.StorageFilter{ + StorageFilter: eth.StorageFilter{ Off: true, }, } diff --git a/pkg/super_node/config/eth_subscription.go b/pkg/super_node/eth/subscription_config.go similarity index 88% rename from pkg/super_node/config/eth_subscription.go rename to pkg/super_node/eth/subscription_config.go index 3108990a..368d461a 100644 --- a/pkg/super_node/config/eth_subscription.go +++ b/pkg/super_node/eth/subscription_config.go @@ -14,17 +14,19 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -package config +package eth import ( "errors" "math/big" "github.com/spf13/viper" + + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) -// EthSubscription config is used by a subscriber to specify what eth data to stream from the super node -type EthSubscription struct { +// SubscriptionSettings config is used by a subscriber to specify what eth data to stream from the super node +type SubscriptionSettings struct { BackFill bool BackFillOnly bool Start *big.Int @@ -73,8 +75,8 @@ type StorageFilter struct { } // Init is used to initialize a EthSubscription struct with env variables -func NewEthSubscriptionConfig() (*EthSubscription, error) { - sc := new(EthSubscription) +func NewEthSubscriptionConfig() (*SubscriptionSettings, error) { + sc := new(SubscriptionSettings) // Below default to false, which means we do not backfill by default sc.BackFill = viper.GetBool("superNode.ethSubscription.historicalData") sc.BackFillOnly = viper.GetBool("superNode.ethSubscription.historicalDataOnly") @@ -126,26 +128,26 @@ func NewEthSubscriptionConfig() (*EthSubscription, error) { } // StartingBlock satisfies the SubscriptionSettings() interface -func (sc *EthSubscription) StartingBlock() *big.Int { +func (sc *SubscriptionSettings) StartingBlock() *big.Int { return sc.Start } // EndingBlock satisfies the SubscriptionSettings() interface -func (sc *EthSubscription) EndingBlock() *big.Int { +func (sc *SubscriptionSettings) EndingBlock() *big.Int { return sc.End } // HistoricalData satisfies the SubscriptionSettings() interface -func (sc *EthSubscription) HistoricalData() bool { +func (sc *SubscriptionSettings) HistoricalData() bool { return sc.BackFill } // HistoricalDataOnly satisfies the SubscriptionSettings() interface -func (sc *EthSubscription) HistoricalDataOnly() bool { +func (sc *SubscriptionSettings) HistoricalDataOnly() bool { return sc.BackFillOnly } // ChainType satisfies the SubscriptionSettings() interface -func (sc *EthSubscription) ChainType() ChainType { - return Ethereum +func (sc *SubscriptionSettings) ChainType() shared.ChainType { + return shared.Ethereum } diff --git a/pkg/super_node/service.go b/pkg/super_node/service.go index af0bdb42..c650533f 100644 --- a/pkg/super_node/service.go +++ b/pkg/super_node/service.go @@ -31,7 +31,6 @@ import ( "github.com/vulcanize/vulcanizedb/pkg/eth/core" "github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres" "github.com/vulcanize/vulcanizedb/pkg/ipfs" - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) @@ -90,7 +89,7 @@ type Service struct { // Number of publishAndIndex workers WorkerPoolSize int // chain type for this service - chain config.ChainType + chain shared.ChainType // Path to ipfs data dir ipfsPath string // Underlying db @@ -98,7 +97,7 @@ type Service struct { } // NewSuperNode creates a new super_node.Interface using an underlying super_node.Service struct -func NewSuperNode(settings *config.SuperNode) (SuperNode, error) { +func NewSuperNode(settings *shared.SuperNodeConfig) (SuperNode, error) { if err := ipfs.InitIPFSPlugins(); err != nil { return nil, err } diff --git a/pkg/super_node/service_test.go b/pkg/super_node/service_test.go index 14c39001..20deaa32 100644 --- a/pkg/super_node/service_test.go +++ b/pkg/super_node/service_test.go @@ -20,15 +20,14 @@ import ( "sync" "time" - "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" - "github.com/ethereum/go-ethereum/rpc" - "github.com/ethereum/go-ethereum/statediff" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" "github.com/vulcanize/vulcanizedb/pkg/super_node" - mocks2 "github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks" + "github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks" + "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" + mocks2 "github.com/vulcanize/vulcanizedb/pkg/super_node/shared/mocks" ) var _ = Describe("Service", func() { @@ -37,22 +36,22 @@ var _ = Describe("Service", func() { wg := new(sync.WaitGroup) payloadChan := make(chan shared.RawChainData, 1) quitChan := make(chan bool, 1) - mockCidIndexer := &mocks2.CIDIndexer{ + mockCidIndexer := &mocks.CIDIndexer{ ReturnErr: nil, } - mockPublisher := &mocks2.IPLDPublisher{ - ReturnCIDPayload: mocks2.MockCIDPayload, + mockPublisher := &mocks.IPLDPublisher{ + ReturnCIDPayload: mocks.MockCIDPayload, ReturnErr: nil, } - mockStreamer := &mocks2.StateDiffStreamer{ + mockStreamer := &mocks2.PayloadStreamer{ ReturnSub: &rpc.ClientSubscription{}, - StreamPayloads: []statediff.Payload{ - mocks2.MockStateDiffPayload, + StreamPayloads: []shared.RawChainData{ + mocks.MockStateDiffPayload, }, ReturnErr: nil, } - mockConverter := &mocks2.PayloadConverter{ - ReturnIPLDPayload: mocks2.MockIPLDPayload, + mockConverter := &mocks.PayloadConverter{ + ReturnIPLDPayload: mocks.MockIPLDPayload, ReturnErr: nil, } processor := &super_node.Service{ @@ -69,10 +68,10 @@ var _ = Describe("Service", func() { time.Sleep(2 * time.Second) quitChan <- true wg.Wait() - Expect(mockConverter.PassedStatediffPayload).To(Equal(mocks2.MockStateDiffPayload)) + Expect(mockConverter.PassedStatediffPayload).To(Equal(mocks.MockStateDiffPayload)) Expect(len(mockCidIndexer.PassedCIDPayload)).To(Equal(1)) - Expect(mockCidIndexer.PassedCIDPayload[0]).To(Equal(mocks2.MockCIDPayload)) - Expect(mockPublisher.PassedIPLDPayload).To(Equal(mocks2.MockIPLDPayload)) + Expect(mockCidIndexer.PassedCIDPayload[0]).To(Equal(mocks.MockCIDPayload)) + Expect(mockPublisher.PassedIPLDPayload).To(Equal(mocks.MockIPLDPayload)) Expect(mockStreamer.PassedPayloadChan).To(Equal(payloadChan)) }) }) diff --git a/pkg/super_node/config/chain_type.go b/pkg/super_node/shared/chain_type.go similarity index 98% rename from pkg/super_node/config/chain_type.go rename to pkg/super_node/shared/chain_type.go index 3b4305a9..3aad798c 100644 --- a/pkg/super_node/config/chain_type.go +++ b/pkg/super_node/shared/chain_type.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -package config +package shared import ( "errors" diff --git a/pkg/super_node/config/config.go b/pkg/super_node/shared/config.go similarity index 95% rename from pkg/super_node/config/config.go rename to pkg/super_node/shared/config.go index 75d1a246..5cc9d1e6 100644 --- a/pkg/super_node/config/config.go +++ b/pkg/super_node/shared/config.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -package config +package shared import ( "fmt" @@ -36,8 +36,8 @@ import ( "github.com/vulcanize/vulcanizedb/utils" ) -// SuperNode config struct -type SuperNode struct { +// SuperNodeConfig struct +type SuperNodeConfig struct { // Ubiquitous fields Chain ChainType IPFSPath string @@ -62,8 +62,8 @@ type SuperNode struct { } // NewSuperNodeConfig is used to initialize a SuperNode config from a config .toml file -func NewSuperNodeConfig() (*SuperNode, error) { - sn := new(SuperNode) +func NewSuperNodeConfig() (*SuperNodeConfig, error) { + sn := new(SuperNodeConfig) sn.DBConfig = config.Database{ Name: viper.GetString("superNode.database.name"), Hostname: viper.GetString("superNode.database.hostname"), @@ -128,7 +128,7 @@ func NewSuperNodeConfig() (*SuperNode, error) { } // BackFillFields is used to fill in the BackFill fields of the config -func (sn *SuperNode) BackFillFields() error { +func (sn *SuperNodeConfig) BackFillFields() error { sn.BackFill = true _, httpClient, err := getNodeAndClient(sn.Chain, viper.GetString("superNode.backFill.httpPath")) if err != nil { diff --git a/pkg/super_node/shared/intefaces.go b/pkg/super_node/shared/intefaces.go index 80945c97..2974b7b1 100644 --- a/pkg/super_node/shared/intefaces.go +++ b/pkg/super_node/shared/intefaces.go @@ -18,8 +18,6 @@ package shared import ( "math/big" - - "github.com/vulcanize/vulcanizedb/pkg/super_node/config" ) // PayloadStreamer streams chain-specific payloads to the provided channel @@ -88,7 +86,7 @@ type DagPutter interface { type SubscriptionSettings interface { StartingBlock() *big.Int EndingBlock() *big.Int - ChainType() config.ChainType + ChainType() ChainType HistoricalData() bool HistoricalDataOnly() bool } diff --git a/pkg/super_node/eth/mocks/fetcher.go b/pkg/super_node/shared/mocks/fetcher.go similarity index 88% rename from pkg/super_node/eth/mocks/fetcher.go rename to pkg/super_node/shared/mocks/fetcher.go index dd9bc3cb..e9d6cbfa 100644 --- a/pkg/super_node/eth/mocks/fetcher.go +++ b/pkg/super_node/shared/mocks/fetcher.go @@ -23,16 +23,16 @@ import ( "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) -// StateDiffFetcher mock for tests -type StateDiffFetcher struct { +// IPLDFetcher mock for tests +type IPLDFetcher struct { PayloadsToReturn map[uint64]shared.RawChainData FetchErrs map[uint64]error CalledAtBlockHeights [][]uint64 CalledTimes int64 } -// FetchStateDiffsAt mock method -func (fetcher *StateDiffFetcher) FetchAt(blockHeights []uint64) ([]shared.RawChainData, error) { +// FetchAt mock method +func (fetcher *IPLDFetcher) FetchAt(blockHeights []uint64) ([]shared.RawChainData, error) { if fetcher.PayloadsToReturn == nil { return nil, errors.New("mock StateDiffFetcher needs to be initialized with payloads to return") } diff --git a/pkg/super_node/eth/mocks/retriever.go b/pkg/super_node/shared/mocks/retriever.go similarity index 100% rename from pkg/super_node/eth/mocks/retriever.go rename to pkg/super_node/shared/mocks/retriever.go diff --git a/pkg/super_node/eth/mocks/streamer.go b/pkg/super_node/shared/mocks/streamer.go similarity index 74% rename from pkg/super_node/eth/mocks/streamer.go rename to pkg/super_node/shared/mocks/streamer.go index a4e49aad..6b9d2774 100644 --- a/pkg/super_node/eth/mocks/streamer.go +++ b/pkg/super_node/shared/mocks/streamer.go @@ -18,20 +18,19 @@ package mocks import ( "github.com/ethereum/go-ethereum/rpc" - "github.com/ethereum/go-ethereum/statediff" "github.com/vulcanize/vulcanizedb/pkg/super_node/shared" ) -// StateDiffStreamer is the underlying struct for the Streamer interface -type StateDiffStreamer struct { +// PayloadStreamer mock struct +type PayloadStreamer struct { PassedPayloadChan chan shared.RawChainData ReturnSub *rpc.ClientSubscription ReturnErr error - StreamPayloads []statediff.Payload + StreamPayloads []shared.RawChainData } -// Stream is the main loop for subscribing to data from the Geth state diff process -func (sds *StateDiffStreamer) Stream(payloadChan chan shared.RawChainData) (shared.ClientSubscription, error) { +// Stream mock method +func (sds *PayloadStreamer) Stream(payloadChan chan shared.RawChainData) (shared.ClientSubscription, error) { sds.PassedPayloadChan = payloadChan go func() {