trimming down to ipfs watchers

pkg/shared/chain_type.go

@@ -0,0 +1,71 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"errors"
+	"strings"
+)
+
+// ChainType enum for specifying blockchain
+type ChainType int
+
+const (
+	UnknownChain ChainType = iota
+	Ethereum
+	Bitcoin
+	Omni
+)
+
+func (c ChainType) String() string {
+	switch c {
+	case Ethereum:
+		return "Ethereum"
+	case Bitcoin:
+		return "Bitcoin"
+	case Omni:
+		return "Omni"
+	default:
+		return ""
+	}
+}
+
+func (c ChainType) API() string {
+	switch c {
+	case Ethereum:
+		return "eth"
+	case Bitcoin:
+		return "btc"
+	case Omni:
+		return "omni"
+	default:
+		return ""
+	}
+}
+
+func NewChainType(name string) (ChainType, error) {
+	switch strings.ToLower(name) {
+	case "ethereum", "eth":
+		return Ethereum, nil
+	case "bitcoin", "btc", "xbt":
+		return Bitcoin, nil
+	case "omni":
+		return Omni, nil
+	default:
+		return UnknownChain, errors.New("invalid name for chain")
+	}
+}

pkg/shared/data_type.go

@@ -0,0 +1,144 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"fmt"
+	"strings"
+)
+
+// DataType is an enum to loosely represent type of chain data
+type DataType int
+
+const (
+	UnknownDataType DataType = iota - 1
+	Full
+	Headers
+	Uncles
+	Transactions
+	Receipts
+	State
+	Storage
+)
+
+// String() method to resolve ReSyncType enum
+func (r DataType) String() string {
+	switch r {
+	case Full:
+		return "full"
+	case Headers:
+		return "headers"
+	case Uncles:
+		return "uncles"
+	case Transactions:
+		return "transactions"
+	case Receipts:
+		return "receipts"
+	case State:
+		return "state"
+	case Storage:
+		return "storage"
+	default:
+		return "unknown"
+	}
+}
+
+// GenerateDataTypeFromString
+func GenerateDataTypeFromString(str string) (DataType, error) {
+	switch strings.ToLower(str) {
+	case "full", "f":
+		return Full, nil
+	case "headers", "header", "h":
+		return Headers, nil
+	case "uncles", "u":
+		return Uncles, nil
+	case "transactions", "transaction", "trxs", "txs", "trx", "tx", "t":
+		return Transactions, nil
+	case "receipts", "receipt", "rcts", "rct", "r":
+		return Receipts, nil
+	case "state":
+		return State, nil
+	case "storage":
+		return Storage, nil
+	default:
+		return UnknownDataType, fmt.Errorf("unrecognized resync type: %s", str)
+	}
+}
+
+func SupportedDataType(d DataType, c ChainType) (bool, error) {
+	switch c {
+	case Ethereum:
+		switch d {
+		case Full:
+			return true, nil
+		case Headers:
+			return true, nil
+		case Uncles:
+			return true, nil
+		case Transactions:
+			return true, nil
+		case Receipts:
+			return true, nil
+		case State:
+			return true, nil
+		case Storage:
+			return true, nil
+		default:
+			return true, nil
+		}
+	case Bitcoin:
+		switch d {
+		case Full:
+			return true, nil
+		case Headers:
+			return true, nil
+		case Uncles:
+			return false, nil
+		case Transactions:
+			return true, nil
+		case Receipts:
+			return false, nil
+		case State:
+			return false, nil
+		case Storage:
+			return false, nil
+		default:
+			return false, nil
+		}
+	case Omni:
+		switch d {
+		case Full:
+			return false, nil
+		case Headers:
+			return false, nil
+		case Uncles:
+			return false, nil
+		case Transactions:
+			return false, nil
+		case Receipts:
+			return false, nil
+		case State:
+			return false, nil
+		case Storage:
+			return false, nil
+		default:
+			return false, nil
+		}
+	default:
+		return false, fmt.Errorf("unrecognized chain type %s", c.String())
+	}
+}

pkg/shared/env.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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"os"
+	"path/filepath"
+
+	"github.com/ethereum/go-ethereum/rpc"
+
+	"github.com/btcsuite/btcd/rpcclient"
+	"github.com/spf13/viper"
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/core"
+)
+
+// Env variables
+const (
+	IPFS_PATH    = "IPFS_PATH"
+	IPFS_MODE    = "IPFS_MODE"
+	HTTP_TIMEOUT = "HTTP_TIMEOUT"
+
+	ETH_WS_PATH       = "ETH_WS_PATH"
+	ETH_HTTP_PATH     = "ETH_HTTP_PATH"
+	ETH_NODE_ID       = "ETH_NODE_ID"
+	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
+	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
+	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
+
+	BTC_WS_PATH       = "BTC_WS_PATH"
+	BTC_HTTP_PATH     = "BTC_HTTP_PATH"
+	BTC_NODE_PASSWORD = "BTC_NODE_PASSWORD"
+	BTC_NODE_USER     = "BTC_NODE_USER"
+	BTC_NODE_ID       = "BTC_NODE_ID"
+	BTC_CLIENT_NAME   = "BTC_CLIENT_NAME"
+	BTC_GENESIS_BLOCK = "BTC_GENESIS_BLOCK"
+	BTC_NETWORK_ID    = "BTC_NETWORK_ID"
+)
+
+// GetEthNodeAndClient returns eth node info and client from path url
+func GetEthNodeAndClient(path string) (core.Node, *rpc.Client, error) {
+	viper.BindEnv("ethereum.nodeID", ETH_NODE_ID)
+	viper.BindEnv("ethereum.clientName", ETH_CLIENT_NAME)
+	viper.BindEnv("ethereum.genesisBlock", ETH_GENESIS_BLOCK)
+	viper.BindEnv("ethereum.networkID", ETH_NETWORK_ID)
+
+	rpcClient, err := rpc.Dial(path)
+	if err != nil {
+		return core.Node{}, nil, err
+	}
+	return core.Node{
+		ID:           viper.GetString("ethereum.nodeID"),
+		ClientName:   viper.GetString("ethereum.clientName"),
+		GenesisBlock: viper.GetString("ethereum.genesisBlock"),
+		NetworkID:    viper.GetString("ethereum.networkID"),
+	}, rpcClient, nil
+}
+
+// GetIPFSPath returns the ipfs path from the config or env variable
+func GetIPFSPath() (string, error) {
+	viper.BindEnv("ipfs.path", IPFS_PATH)
+	ipfsPath := viper.GetString("ipfs.path")
+	if ipfsPath == "" {
+		home, err := os.UserHomeDir()
+		if err != nil {
+			return "", err
+		}
+		ipfsPath = filepath.Join(home, ".ipfs")
+	}
+	return ipfsPath, nil
+}
+
+// GetIPFSMode returns the ipfs mode of operation from the config or env variable
+func GetIPFSMode() (IPFSMode, error) {
+	viper.BindEnv("ipfs.mode", IPFS_MODE)
+	ipfsMode := viper.GetString("ipfs.mode")
+	if ipfsMode == "" {
+		return DirectPostgres, nil
+	}
+	return NewIPFSMode(ipfsMode)
+}
+
+// GetBtcNodeAndClient returns btc node info from path url
+func GetBtcNodeAndClient(path string) (core.Node, *rpcclient.ConnConfig) {
+	viper.BindEnv("bitcoin.nodeID", BTC_NODE_ID)
+	viper.BindEnv("bitcoin.clientName", BTC_CLIENT_NAME)
+	viper.BindEnv("bitcoin.genesisBlock", BTC_GENESIS_BLOCK)
+	viper.BindEnv("bitcoin.networkID", BTC_NETWORK_ID)
+	viper.BindEnv("bitcoin.pass", BTC_NODE_PASSWORD)
+	viper.BindEnv("bitcoin.user", BTC_NODE_USER)
+
+	// For bitcoin we load in node info from the config because there is no RPC endpoint to retrieve this from the node
+	return core.Node{
+			ID:           viper.GetString("bitcoin.nodeID"),
+			ClientName:   viper.GetString("bitcoin.clientName"),
+			GenesisBlock: viper.GetString("bitcoin.genesisBlock"),
+			NetworkID:    viper.GetString("bitcoin.networkID"),
+		}, &rpcclient.ConnConfig{
+			Host:         path,
+			HTTPPostMode: true, // Bitcoin core only supports HTTP POST mode
+			DisableTLS:   true, // Bitcoin core does not provide TLS by default
+			Pass:         viper.GetString("bitcoin.pass"),
+			User:         viper.GetString("bitcoin.user"),
+		}
+}

pkg/shared/functions.go

@@ -0,0 +1,129 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"bytes"
+
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/ipfs/ipld"
+
+	"github.com/ipfs/go-cid"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ipfs/go-ipfs-blockstore"
+	"github.com/ipfs/go-ipfs-ds-help"
+	node "github.com/ipfs/go-ipld-format"
+	"github.com/jmoiron/sqlx"
+	"github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/ipfs"
+)
+
+// ListContainsString used to check if a list of strings contains a particular string
+func ListContainsString(sss []string, s string) bool {
+	for _, str := range sss {
+		if s == str {
+			return true
+		}
+	}
+	return false
+}
+
+// IPLDsContainBytes used to check if a list of strings contains a particular string
+func IPLDsContainBytes(iplds []ipfs.BlockModel, b []byte) bool {
+	for _, ipld := range iplds {
+		if bytes.Equal(ipld.Data, b) {
+			return true
+		}
+	}
+	return false
+}
+
+// ListContainsGap used to check if a list of Gaps contains a particular Gap
+func ListContainsGap(gapList []Gap, gap Gap) bool {
+	for _, listGap := range gapList {
+		if listGap == gap {
+			return true
+		}
+	}
+	return false
+}
+
+// HandleNullAddrPointer will return an emtpy string for a nil address pointer
+func HandleNullAddrPointer(to *common.Address) string {
+	if to == nil {
+		return ""
+	}
+	return to.Hex()
+}
+
+// HandleNullAddr will return an empty string for a a null address
+func HandleNullAddr(to common.Address) string {
+	if to.Hex() == "0x0000000000000000000000000000000000000000" {
+		return ""
+	}
+	return to.Hex()
+}
+
+// Rollback sql transaction and log any error
+func Rollback(tx *sqlx.Tx) {
+	if err := tx.Rollback(); err != nil {
+		logrus.Error(err)
+	}
+}
+
+// PublishIPLD is used to insert an ipld into Postgres blockstore with the provided tx
+func PublishIPLD(tx *sqlx.Tx, i node.Node) error {
+	dbKey := dshelp.CidToDsKey(i.Cid())
+	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
+	raw := i.RawData()
+	_, err := tx.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, prefixedKey, raw)
+	return err
+}
+
+// FetchIPLD is used to retrieve an ipld from Postgres blockstore with the provided tx
+func FetchIPLD(tx *sqlx.Tx, cid string) ([]byte, error) {
+	mhKey, err := MultihashKeyFromCIDString(cid)
+	if err != nil {
+		return nil, err
+	}
+	pgStr := `SELECT data FROM public.blocks WHERE key = $1`
+	var block []byte
+	return block, tx.Get(&block, pgStr, mhKey)
+}
+
+// MultihashKeyFromCIDString converts a cid string into a blockstore-prefixed multihash db key string
+func MultihashKeyFromCIDString(c string) (string, error) {
+	dc, err := cid.Decode(c)
+	if err != nil {
+		return "", err
+	}
+	dbKey := dshelp.CidToDsKey(dc)
+	return blockstore.BlockPrefix.String() + dbKey.String(), nil
+}
+
+// PublishRaw derives a cid from raw bytes and provided codec and multihash type, and writes it to the db tx
+func PublishRaw(tx *sqlx.Tx, codec, mh uint64, raw []byte) (string, error) {
+	c, err := ipld.RawdataToCid(codec, raw, mh)
+	if err != nil {
+		return "", err
+	}
+	dbKey := dshelp.CidToDsKey(c)
+	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
+	_, err = tx.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, prefixedKey, raw)
+	return c.String(), err
+}

pkg/shared/intefaces.go

@@ -0,0 +1,88 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"math/big"
+)
+
+// PayloadStreamer streams chain-specific payloads to the provided channel
+type PayloadStreamer interface {
+	Stream(payloadChan chan RawChainData) (ClientSubscription, error)
+}
+
+// PayloadFetcher fetches chain-specific payloads
+type PayloadFetcher interface {
+	FetchAt(blockHeights []uint64) ([]RawChainData, error)
+}
+
+// PayloadConverter converts chain-specific payloads into IPLD payloads for publishing
+type PayloadConverter interface {
+	Convert(payload RawChainData) (ConvertedData, error)
+}
+
+// IPLDPublisher publishes IPLD payloads and returns a CID payload for indexing
+type IPLDPublisher interface {
+	Publish(payload ConvertedData) (CIDsForIndexing, error)
+}
+
+// CIDIndexer indexes a CID payload in Postgres
+type CIDIndexer interface {
+	Index(cids CIDsForIndexing) error
+}
+
+// ResponseFilterer applies a filter to an IPLD payload to return a subscription response packet
+type ResponseFilterer interface {
+	Filter(filter SubscriptionSettings, payload ConvertedData) (response IPLDs, err error)
+}
+
+// CIDRetriever retrieves cids according to a provided filter and returns a CID wrapper
+type CIDRetriever interface {
+	Retrieve(filter SubscriptionSettings, blockNumber int64) ([]CIDsForFetching, bool, error)
+	RetrieveFirstBlockNumber() (int64, error)
+	RetrieveLastBlockNumber() (int64, error)
+	RetrieveGapsInData(validationLevel int) ([]Gap, error)
+}
+
+// IPLDFetcher uses a CID wrapper to fetch an IPLD wrapper
+type IPLDFetcher interface {
+	Fetch(cids CIDsForFetching) (IPLDs, error)
+}
+
+// ClientSubscription is a general interface for chain data subscriptions
+type ClientSubscription interface {
+	Err() <-chan error
+	Unsubscribe()
+}
+
+// Cleaner is for cleaning out data from the cache within the given ranges
+type Cleaner interface {
+	Clean(rngs [][2]uint64, t DataType) error
+	ResetValidation(rngs [][2]uint64) error
+}
+
+// SubscriptionSettings is the interface every subscription filter type needs to satisfy, no matter the chain
+// Further specifics of the underlying filter type depend on the internal needs of the types
+// which satisfy the ResponseFilterer and CIDRetriever interfaces for a specific chain
+// The underlying type needs to be rlp serializable
+type SubscriptionSettings interface {
+	StartingBlock() *big.Int
+	EndingBlock() *big.Int
+	ChainType() ChainType
+	HistoricalData() bool
+	HistoricalDataOnly() bool
+}

pkg/shared/ipfs_mode.go

@@ -0,0 +1,58 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"errors"
+	"strings"
+)
+
+// IPFSMode enum for specifying how we want to interface and publish objects to IPFS
+type IPFSMode int
+
+const (
+	Unknown IPFSMode = iota
+	LocalInterface
+	RemoteClient
+	DirectPostgres
+)
+
+func (c IPFSMode) String() string {
+	switch c {
+	case LocalInterface:
+		return "Local"
+	case RemoteClient:
+		return "Remote"
+	case DirectPostgres:
+		return "Postgres"
+	default:
+		return ""
+	}
+}
+
+func NewIPFSMode(name string) (IPFSMode, error) {
+	switch strings.ToLower(name) {
+	case "local", "interface", "minimal":
+		return LocalInterface, nil
+	case "remote", "client":
+		return RemoteClient, errors.New("remote IPFS client mode is not currently supported")
+	case "postgres", "direct":
+		return DirectPostgres, nil
+	default:
+		return Unknown, errors.New("unrecognized name for ipfs mode")
+	}
+}

pkg/shared/mocks/payload_fetcher.go

@@ -0,0 +1,50 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package mocks
+
+import (
+	"errors"
+	"sync/atomic"
+
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/shared"
+)
+
+// PayloadFetcher mock for tests
+type PayloadFetcher struct {
+	PayloadsToReturn     map[uint64]shared.RawChainData
+	FetchErrs            map[uint64]error
+	CalledAtBlockHeights [][]uint64
+	CalledTimes          int64
+}
+
+// FetchAt mock method
+func (fetcher *PayloadFetcher) FetchAt(blockHeights []uint64) ([]shared.RawChainData, error) {
+	if fetcher.PayloadsToReturn == nil {
+		return nil, errors.New("mock StateDiffFetcher needs to be initialized with payloads to return")
+	}
+	atomic.AddInt64(&fetcher.CalledTimes, 1) // thread-safe increment
+	fetcher.CalledAtBlockHeights = append(fetcher.CalledAtBlockHeights, blockHeights)
+	results := make([]shared.RawChainData, 0, len(blockHeights))
+	for _, height := range blockHeights {
+		results = append(results, fetcher.PayloadsToReturn[height])
+		err, ok := fetcher.FetchErrs[height]
+		if ok && err != nil {
+			return nil, err
+		}
+	}
+	return results, nil
+}

pkg/shared/mocks/retriever.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 <http://www.gnu.org/licenses/>.
+
+package mocks
+
+import (
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/postgres"
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/shared"
+)
+
+// CIDRetriever is a mock CID retriever for use in tests
+type CIDRetriever struct {
+	GapsToRetrieve              []shared.Gap
+	GapsToRetrieveErr           error
+	CalledTimes                 int
+	FirstBlockNumberToReturn    int64
+	RetrieveFirstBlockNumberErr error
+}
+
+// RetrieveCIDs mock method
+func (*CIDRetriever) Retrieve(filter shared.SubscriptionSettings, blockNumber int64) ([]shared.CIDsForFetching, bool, error) {
+	panic("implement me")
+}
+
+// RetrieveLastBlockNumber mock method
+func (*CIDRetriever) RetrieveLastBlockNumber() (int64, error) {
+	panic("implement me")
+}
+
+// RetrieveFirstBlockNumber mock method
+func (mcr *CIDRetriever) RetrieveFirstBlockNumber() (int64, error) {
+	return mcr.FirstBlockNumberToReturn, mcr.RetrieveFirstBlockNumberErr
+}
+
+// RetrieveGapsInData mock method
+func (mcr *CIDRetriever) RetrieveGapsInData(int) ([]shared.Gap, error) {
+	mcr.CalledTimes++
+	return mcr.GapsToRetrieve, mcr.GapsToRetrieveErr
+}
+
+// SetGapsToRetrieve mock method
+func (mcr *CIDRetriever) SetGapsToRetrieve(gaps []shared.Gap) {
+	if mcr.GapsToRetrieve == nil {
+		mcr.GapsToRetrieve = make([]shared.Gap, 0)
+	}
+	mcr.GapsToRetrieve = append(mcr.GapsToRetrieve, gaps...)
+}
+
+func (mcr *CIDRetriever) Database() *postgres.DB {
+	panic("implement me")
+}

pkg/shared/mocks/streamer.go

@@ -0,0 +1,43 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package mocks
+
+import (
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/shared"
+)
+
+// PayloadStreamer mock struct
+type PayloadStreamer struct {
+	PassedPayloadChan chan shared.RawChainData
+	ReturnSub         *rpc.ClientSubscription
+	ReturnErr         error
+	StreamPayloads    []shared.RawChainData
+}
+
+// Stream mock method
+func (sds *PayloadStreamer) Stream(payloadChan chan shared.RawChainData) (shared.ClientSubscription, error) {
+	sds.PassedPayloadChan = payloadChan
+
+	go func() {
+		for _, payload := range sds.StreamPayloads {
+			sds.PassedPayloadChan <- payload
+		}
+	}()
+
+	return sds.ReturnSub, sds.ReturnErr
+}

pkg/shared/test_helpers.go

@@ -0,0 +1,32 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+import (
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/config"
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/core"
+	"github.com/vulcanize/ipfs-chain-watcher/pkg/postgres"
+)
+
+// SetupDB is use to setup a db for super node tests
+func SetupDB() (*postgres.DB, error) {
+	return postgres.NewDB(config.Database{
+		Hostname: "localhost",
+		Name:     "vulcanize_testing",
+		Port:     5432,
+	}, core.Node{})
+}

pkg/shared/types.go

@@ -0,0 +1,38 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package shared
+
+// These types serve as very loose wrappers around a generic underlying interface{}
+type RawChainData interface{}
+
+// The concrete type underneath StreamedIPLDs should not be a pointer
+type ConvertedData interface {
+	Height() int64
+}
+
+type CIDsForIndexing interface{}
+
+type CIDsForFetching interface{}
+
+type IPLDs interface {
+	Height() int64
+}
+
+type Gap struct {
+	Start uint64
+	Stop  uint64
+}