seed => super; port 80 => port 8080; backfill process for the super_node

pkg/super_node/api.go

@@ -0,0 +1,90 @@
+// 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 super_node
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+)
+
+// APIName is the namespace used for the state diffing service API
+const APIName = "vdb"
+
+// APIVersion is the version of the state diffing service API
+const APIVersion = "0.0.1"
+
+// PublicSuperNodeAPI is the public api for the super node
+type PublicSuperNodeAPI struct {
+	sni NodeInterface
+}
+
+// NewPublicSuperNodeAPI creates a new PublicSuperNodeAPI with the provided underlying SyncPublishScreenAndServe process
+func NewPublicSuperNodeAPI(superNodeInterface NodeInterface) *PublicSuperNodeAPI {
+	return &PublicSuperNodeAPI{
+		sni: superNodeInterface,
+	}
+}
+
+// Stream is the public method to setup a subscription that fires off SyncPublishScreenAndServe payloads as they are created
+func (api *PublicSuperNodeAPI) Stream(ctx context.Context, streamFilters config.Subscription) (*rpc.Subscription, error) {
+	// ensure that the RPC connection supports subscriptions
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return nil, rpc.ErrNotificationsUnsupported
+	}
+
+	// create subscription and start waiting for statediff events
+	rpcSub := notifier.CreateSubscription()
+
+	go func() {
+		// subscribe to events from the SyncPublishScreenAndServe service
+		payloadChannel := make(chan streamer.SuperNodePayload, payloadChanBufferSize)
+		quitChan := make(chan bool, 1)
+		go api.sni.Subscribe(rpcSub.ID, payloadChannel, quitChan, streamFilters)
+
+		// loop and await state diff payloads and relay them to the subscriber with then notifier
+		for {
+			select {
+			case packet := <-payloadChannel:
+				if notifyErr := notifier.Notify(rpcSub.ID, packet); notifyErr != nil {
+					log.Error("Failed to send state diff packet", "err", notifyErr)
+					api.sni.Unsubscribe(rpcSub.ID)
+					return
+				}
+			case <-rpcSub.Err():
+				api.sni.Unsubscribe(rpcSub.ID)
+				return
+			case <-quitChan:
+				// don't need to unsubscribe, SyncPublishScreenAndServe service does so before sending the quit signal
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
+
+// Node is a public rpc method to allow transformers to fetch the Geth node info for the super node
+func (api *PublicSuperNodeAPI) Node() core.Node {
+	return api.sni.Node()
+}

pkg/super_node/backfiller.go

@@ -0,0 +1,135 @@
+// 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 super_node
+
+import (
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/params"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/fetcher"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// BackFillInterface for filling in gaps in the super node
+type BackFillInterface interface {
+	// Method for the super node to periodically check for and fill in gaps in its data using an archival node
+	FillGaps(wg *sync.WaitGroup, quitChan <-chan bool)
+}
+
+// BackFillService for filling in gaps in the super node
+type BackFillService struct {
+	// Interface for converting statediff payloads into ETH-IPLD object payloads
+	Converter ipfs.PayloadConverter
+	// Interface for publishing the ETH-IPLD payloads to IPFS
+	Publisher ipfs.IPLDPublisher
+	// Interface for indexing the CIDs of the published ETH-IPLDs in Postgres
+	Repository CIDRepository
+	// Interface for searching and retrieving CIDs from Postgres index
+	Retriever CIDRetriever
+	// State-diff fetcher; needs to be configured with an archival core.RpcClient
+	StateDiffFetcher fetcher.IStateDiffFetcher
+	// Check frequency
+	GapCheckFrequency time.Duration
+}
+
+// NewBackFillService returns a new BackFillInterface
+func NewBackFillService(ipfsPublisher ipfs.IPLDPublisher, db *postgres.DB, archivalNodeRpcClient core.RpcClient, freq time.Duration) BackFillInterface {
+	return &BackFillService{
+		Repository:        NewCIDRepository(db),
+		Converter:         ipfs.NewPayloadConverter(params.MainnetChainConfig),
+		Publisher:         ipfsPublisher,
+		Retriever:         NewCIDRetriever(db),
+		StateDiffFetcher:  fetcher.NewStateDiffFetcher(archivalNodeRpcClient),
+		GapCheckFrequency: freq,
+	}
+}
+
+// FillGaps periodically checks for and fills in gaps in the super node db
+// this requires a core.RpcClient that is pointed at an archival node with the StateDiffAt method exposed
+func (bfs *BackFillService) FillGaps(wg *sync.WaitGroup, quitChan <-chan bool) {
+	ticker := time.NewTicker(bfs.GapCheckFrequency)
+	wg.Add(1)
+
+	go func() {
+		for {
+			select {
+			case <-quitChan:
+				log.Info("quiting FillGaps process")
+				wg.Done()
+				return
+			case <-ticker.C:
+				log.Info("searching for gaps in the super node database")
+				startingBlock, firstBlockErr := bfs.Retriever.RetrieveFirstBlockNumber()
+				if firstBlockErr != nil {
+					log.Error(firstBlockErr)
+					continue
+				}
+				if startingBlock != 1 {
+					startingGap := [2]int64{
+						1,
+						startingBlock - 1,
+					}
+					log.Info("found gap at the beginning of the sync")
+					bfs.fillGaps(startingGap)
+				}
+
+				gaps, gapErr := bfs.Retriever.RetrieveGapsInData()
+				if gapErr != nil {
+					log.Error(gapErr)
+					continue
+				}
+				for _, gap := range gaps {
+					bfs.fillGaps(gap)
+				}
+			}
+		}
+	}()
+}
+
+func (bfs *BackFillService) fillGaps(gap [2]int64) {
+	log.Infof("filling in gap from block %d to block %d", gap[0], gap[1])
+	blockHeights := make([]uint64, 0, gap[1]-gap[0]+1)
+	for i := gap[0]; i <= gap[1]; i++ {
+		blockHeights = append(blockHeights, uint64(i))
+	}
+	payloads, fetchErr := bfs.StateDiffFetcher.FetchStateDiffsAt(blockHeights)
+	if fetchErr != nil {
+		log.Error(fetchErr)
+		return
+	}
+	for _, payload := range payloads {
+		ipldPayload, convertErr := bfs.Converter.Convert(*payload)
+		if convertErr != nil {
+			log.Error(convertErr)
+			continue
+		}
+		cidPayload, publishErr := bfs.Publisher.Publish(ipldPayload)
+		if publishErr != nil {
+			log.Error(publishErr)
+			continue
+		}
+		indexErr := bfs.Repository.Index(cidPayload)
+		if indexErr != nil {
+			log.Error(indexErr)
+		}
+	}
+}

pkg/super_node/backfiller_test.go

@@ -0,0 +1,187 @@
+// 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 super_node_test
+
+import (
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/statediff"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	mocks2 "github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node"
+	mocks3 "github.com/vulcanize/vulcanizedb/pkg/super_node/mocks"
+)
+
+var _ = Describe("BackFiller", func() {
+	Describe("FillGaps", func() {
+		It("Periodically checks for and fills in gaps in the super node's data", func() {
+			mockCidRepo := &mocks3.CIDRepository{
+				ReturnErr: nil,
+			}
+			mockPublisher := &mocks.IterativeIPLDPublisher{
+				ReturnCIDPayload: []*ipfs.CIDPayload{mocks.MockCIDPayload, mocks.MockCIDPayload},
+				ReturnErr:        nil,
+			}
+			mockConverter := &mocks.IterativePayloadConverter{
+				ReturnIPLDPayload: []*ipfs.IPLDPayload{mocks.MockIPLDPayload, mocks.MockIPLDPayload},
+				ReturnErr:         nil,
+			}
+			mockRetriever := &mocks3.MockCIDRetriever{
+				FirstBlockNumberToReturn: 1,
+				GapsToRetrieve: [][2]int64{
+					{
+						100, 101,
+					},
+				},
+			}
+			mockFetcher := &mocks2.MockStateDiffFetcher{
+				StateDiffsToReturn: map[uint64]*statediff.Payload{
+					100: &mocks.MockStateDiffPayload,
+					101: &mocks.MockStateDiffPayload,
+				},
+			}
+			backfiller := &super_node.BackFillService{
+				Repository:        mockCidRepo,
+				Publisher:         mockPublisher,
+				Converter:         mockConverter,
+				StateDiffFetcher:  mockFetcher,
+				Retriever:         mockRetriever,
+				GapCheckFrequency: time.Second * 10,
+			}
+			wg := &sync.WaitGroup{}
+			quitChan := make(chan bool, 1)
+			backfiller.FillGaps(wg, quitChan)
+			time.Sleep(time.Second * 15)
+			quitChan <- true
+			Expect(len(mockCidRepo.PassedCIDPayload)).To(Equal(2))
+			Expect(mockCidRepo.PassedCIDPayload[0]).To(Equal(mocks.MockCIDPayload))
+			Expect(mockCidRepo.PassedCIDPayload[1]).To(Equal(mocks.MockCIDPayload))
+			Expect(len(mockPublisher.PassedIPLDPayload)).To(Equal(2))
+			Expect(mockPublisher.PassedIPLDPayload[0]).To(Equal(mocks.MockIPLDPayload))
+			Expect(mockPublisher.PassedIPLDPayload[1]).To(Equal(mocks.MockIPLDPayload))
+			Expect(len(mockConverter.PassedStatediffPayload)).To(Equal(2))
+			Expect(mockConverter.PassedStatediffPayload[0]).To(Equal(mocks.MockStateDiffPayload))
+			Expect(mockConverter.PassedStatediffPayload[1]).To(Equal(mocks.MockStateDiffPayload))
+			Expect(mockRetriever.CalledTimes).To(Equal(1))
+			Expect(len(mockFetcher.CalledAtBlockHeights)).To(Equal(1))
+			Expect(mockFetcher.CalledAtBlockHeights[0]).To(Equal([]uint64{100, 101}))
+		})
+
+		It("Works for single block `ranges`", func() {
+			mockCidRepo := &mocks3.CIDRepository{
+				ReturnErr: nil,
+			}
+			mockPublisher := &mocks.IterativeIPLDPublisher{
+				ReturnCIDPayload: []*ipfs.CIDPayload{mocks.MockCIDPayload},
+				ReturnErr:        nil,
+			}
+			mockConverter := &mocks.IterativePayloadConverter{
+				ReturnIPLDPayload: []*ipfs.IPLDPayload{mocks.MockIPLDPayload},
+				ReturnErr:         nil,
+			}
+			mockRetriever := &mocks3.MockCIDRetriever{
+				FirstBlockNumberToReturn: 1,
+				GapsToRetrieve: [][2]int64{
+					{
+						100, 100,
+					},
+				},
+			}
+			mockFetcher := &mocks2.MockStateDiffFetcher{
+				StateDiffsToReturn: map[uint64]*statediff.Payload{
+					100: &mocks.MockStateDiffPayload,
+				},
+			}
+			backfiller := &super_node.BackFillService{
+				Repository:        mockCidRepo,
+				Publisher:         mockPublisher,
+				Converter:         mockConverter,
+				StateDiffFetcher:  mockFetcher,
+				Retriever:         mockRetriever,
+				GapCheckFrequency: time.Second * 10,
+			}
+			wg := &sync.WaitGroup{}
+			quitChan := make(chan bool, 1)
+			backfiller.FillGaps(wg, quitChan)
+			time.Sleep(time.Second * 15)
+			quitChan <- true
+			Expect(len(mockCidRepo.PassedCIDPayload)).To(Equal(1))
+			Expect(mockCidRepo.PassedCIDPayload[0]).To(Equal(mocks.MockCIDPayload))
+			Expect(len(mockPublisher.PassedIPLDPayload)).To(Equal(1))
+			Expect(mockPublisher.PassedIPLDPayload[0]).To(Equal(mocks.MockIPLDPayload))
+			Expect(len(mockConverter.PassedStatediffPayload)).To(Equal(1))
+			Expect(mockConverter.PassedStatediffPayload[0]).To(Equal(mocks.MockStateDiffPayload))
+			Expect(mockRetriever.CalledTimes).To(Equal(1))
+			Expect(len(mockFetcher.CalledAtBlockHeights)).To(Equal(1))
+			Expect(mockFetcher.CalledAtBlockHeights[0]).To(Equal([]uint64{100}))
+		})
+
+		It("Finds beginning gap", func() {
+			mockCidRepo := &mocks3.CIDRepository{
+				ReturnErr: nil,
+			}
+			mockPublisher := &mocks.IterativeIPLDPublisher{
+				ReturnCIDPayload: []*ipfs.CIDPayload{mocks.MockCIDPayload, mocks.MockCIDPayload},
+				ReturnErr:        nil,
+			}
+			mockConverter := &mocks.IterativePayloadConverter{
+				ReturnIPLDPayload: []*ipfs.IPLDPayload{mocks.MockIPLDPayload, mocks.MockIPLDPayload},
+				ReturnErr:         nil,
+			}
+			mockRetriever := &mocks3.MockCIDRetriever{
+				FirstBlockNumberToReturn: 3,
+				GapsToRetrieve:           [][2]int64{},
+			}
+			mockFetcher := &mocks2.MockStateDiffFetcher{
+				StateDiffsToReturn: map[uint64]*statediff.Payload{
+					1: &mocks.MockStateDiffPayload,
+					2: &mocks.MockStateDiffPayload,
+				},
+			}
+			backfiller := &super_node.BackFillService{
+				Repository:        mockCidRepo,
+				Publisher:         mockPublisher,
+				Converter:         mockConverter,
+				StateDiffFetcher:  mockFetcher,
+				Retriever:         mockRetriever,
+				GapCheckFrequency: time.Second * 10,
+			}
+			wg := &sync.WaitGroup{}
+			quitChan := make(chan bool, 1)
+			backfiller.FillGaps(wg, quitChan)
+			time.Sleep(time.Second * 15)
+			quitChan <- true
+			Expect(len(mockCidRepo.PassedCIDPayload)).To(Equal(2))
+			Expect(mockCidRepo.PassedCIDPayload[0]).To(Equal(mocks.MockCIDPayload))
+			Expect(mockCidRepo.PassedCIDPayload[1]).To(Equal(mocks.MockCIDPayload))
+			Expect(len(mockPublisher.PassedIPLDPayload)).To(Equal(2))
+			Expect(mockPublisher.PassedIPLDPayload[0]).To(Equal(mocks.MockIPLDPayload))
+			Expect(mockPublisher.PassedIPLDPayload[1]).To(Equal(mocks.MockIPLDPayload))
+			Expect(len(mockConverter.PassedStatediffPayload)).To(Equal(2))
+			Expect(mockConverter.PassedStatediffPayload[0]).To(Equal(mocks.MockStateDiffPayload))
+			Expect(mockConverter.PassedStatediffPayload[1]).To(Equal(mocks.MockStateDiffPayload))
+			Expect(mockRetriever.CalledTimes).To(Equal(1))
+			Expect(len(mockFetcher.CalledAtBlockHeights)).To(Equal(1))
+			Expect(mockFetcher.CalledAtBlockHeights[0]).To(Equal([]uint64{1, 2}))
+		})
+	})
+})

pkg/super_node/filterer.go

@@ -0,0 +1,246 @@
+// 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 super_node
+
+import (
+	"bytes"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/rlp"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// ResponseFilterer is the inteface used to screen eth data and package appropriate data into a response payload
+type ResponseFilterer interface {
+	FilterResponse(streamFilters config.Subscription, payload ipfs.IPLDPayload) (streamer.SuperNodePayload, error)
+}
+
+// Filterer is the underlying struct for the ResponseFilterer interface
+type Filterer struct{}
+
+// NewResponseFilterer creates a new Filterer satisfying the ResponseFilterer interface
+func NewResponseFilterer() *Filterer {
+	return &Filterer{}
+}
+
+// FilterResponse is used to filter through eth data to extract and package requested data into a Payload
+func (s *Filterer) FilterResponse(streamFilters config.Subscription, payload ipfs.IPLDPayload) (streamer.SuperNodePayload, error) {
+	response := new(streamer.SuperNodePayload)
+	err := s.filterHeaders(streamFilters, response, payload)
+	if err != nil {
+		return streamer.SuperNodePayload{}, err
+	}
+	txHashes, err := s.filterTransactions(streamFilters, response, payload)
+	if err != nil {
+		return streamer.SuperNodePayload{}, err
+	}
+	err = s.filerReceipts(streamFilters, response, payload, txHashes)
+	if err != nil {
+		return streamer.SuperNodePayload{}, err
+	}
+	err = s.filterState(streamFilters, response, payload)
+	if err != nil {
+		return streamer.SuperNodePayload{}, err
+	}
+	err = s.filterStorage(streamFilters, response, payload)
+	if err != nil {
+		return streamer.SuperNodePayload{}, err
+	}
+	response.BlockNumber = payload.BlockNumber
+	return *response, nil
+}
+
+func (s *Filterer) filterHeaders(streamFilters config.Subscription, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) error {
+	if !streamFilters.HeaderFilter.Off && checkRange(streamFilters.StartingBlock.Int64(), streamFilters.EndingBlock.Int64(), payload.BlockNumber.Int64()) {
+		response.HeadersRlp = append(response.HeadersRlp, payload.HeaderRLP)
+		if !streamFilters.HeaderFilter.FinalOnly {
+			for _, uncle := range payload.BlockBody.Uncles {
+				uncleRlp, err := rlp.EncodeToBytes(uncle)
+				if err != nil {
+					return err
+				}
+				response.UnclesRlp = append(response.UnclesRlp, uncleRlp)
+			}
+		}
+	}
+	return nil
+}
+
+func checkRange(start, end, actual int64) bool {
+	if (end <= 0 || end >= actual) && start <= actual {
+		return true
+	}
+	return false
+}
+
+func (s *Filterer) filterTransactions(streamFilters config.Subscription, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) ([]common.Hash, error) {
+	trxHashes := make([]common.Hash, 0, len(payload.BlockBody.Transactions))
+	if !streamFilters.TrxFilter.Off && checkRange(streamFilters.StartingBlock.Int64(), streamFilters.EndingBlock.Int64(), payload.BlockNumber.Int64()) {
+		for i, trx := range payload.BlockBody.Transactions {
+			if checkTransactions(streamFilters.TrxFilter.Src, streamFilters.TrxFilter.Dst, payload.TrxMetaData[i].Src, payload.TrxMetaData[i].Dst) {
+				trxBuffer := new(bytes.Buffer)
+				err := trx.EncodeRLP(trxBuffer)
+				if err != nil {
+					return nil, err
+				}
+				trxHashes = append(trxHashes, trx.Hash())
+				response.TransactionsRlp = append(response.TransactionsRlp, trxBuffer.Bytes())
+			}
+		}
+	}
+	return trxHashes, nil
+}
+
+func checkTransactions(wantedSrc, wantedDst []string, actualSrc, actualDst string) bool {
+	// If we aren't filtering for any addresses, every transaction is a go
+	if len(wantedDst) == 0 && len(wantedSrc) == 0 {
+		return true
+	}
+	for _, src := range wantedSrc {
+		if src == actualSrc {
+			return true
+		}
+	}
+	for _, dst := range wantedDst {
+		if dst == actualDst {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *Filterer) filerReceipts(streamFilters config.Subscription, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload, trxHashes []common.Hash) error {
+	if !streamFilters.ReceiptFilter.Off && checkRange(streamFilters.StartingBlock.Int64(), streamFilters.EndingBlock.Int64(), payload.BlockNumber.Int64()) {
+		for i, receipt := range payload.Receipts {
+			if checkReceipts(receipt, streamFilters.ReceiptFilter.Topic0s, payload.ReceiptMetaData[i].Topic0s, streamFilters.ReceiptFilter.Contracts, payload.ReceiptMetaData[i].ContractAddress, trxHashes) {
+				receiptForStorage := (*types.ReceiptForStorage)(receipt)
+				receiptBuffer := new(bytes.Buffer)
+				err := receiptForStorage.EncodeRLP(receiptBuffer)
+				if err != nil {
+					return err
+				}
+				response.ReceiptsRlp = append(response.ReceiptsRlp, receiptBuffer.Bytes())
+			}
+		}
+	}
+	return nil
+}
+
+func checkReceipts(rct *types.Receipt, wantedTopics, actualTopics, wantedContracts []string, actualContract string, wantedTrxHashes []common.Hash) bool {
+	// If we aren't filtering for any topics or contracts, all topics are a go
+	if len(wantedTopics) == 0 && len(wantedContracts) == 0 {
+		return true
+	}
+	// No matter what filters we have, we keep receipts for the trxs we are interested in
+	for _, wantedTrxHash := range wantedTrxHashes {
+		if bytes.Equal(wantedTrxHash.Bytes(), rct.TxHash.Bytes()) {
+			return true
+		}
+	}
+
+	if len(wantedContracts) == 0 {
+		// We keep all receipts that have logs we are interested in
+		for _, wantedTopic := range wantedTopics {
+			for _, actualTopic := range actualTopics {
+				if wantedTopic == actualTopic {
+					return true
+				}
+			}
+		}
+	} else { // We keep receipts that belong to one of the specified contracts and have logs with topics if we aren't filtering on topics
+		for _, wantedContract := range wantedContracts {
+			if wantedContract == actualContract {
+				if len(wantedTopics) == 0 {
+					return true
+				} else { // Or if we have contracts and topics to filter on we only keep receipts that satisfy both conditions
+					for _, wantedTopic := range wantedTopics {
+						for _, actualTopic := range actualTopics {
+							if wantedTopic == actualTopic {
+								return true
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+
+	return false
+}
+
+func (s *Filterer) filterState(streamFilters config.Subscription, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) error {
+	if !streamFilters.StateFilter.Off && checkRange(streamFilters.StartingBlock.Int64(), streamFilters.EndingBlock.Int64(), payload.BlockNumber.Int64()) {
+		response.StateNodesRlp = make(map[common.Hash][]byte)
+		keyFilters := make([]common.Hash, 0, len(streamFilters.StateFilter.Addresses))
+		for _, addr := range streamFilters.StateFilter.Addresses {
+			keyFilter := ipfs.AddressToKey(common.HexToAddress(addr))
+			keyFilters = append(keyFilters, keyFilter)
+		}
+		for key, stateNode := range payload.StateNodes {
+			if checkNodeKeys(keyFilters, key) {
+				if stateNode.Leaf || streamFilters.StateFilter.IntermediateNodes {
+					response.StateNodesRlp[key] = stateNode.Value
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func checkNodeKeys(wantedKeys []common.Hash, actualKey common.Hash) bool {
+	// If we aren't filtering for any specific keys, all nodes are a go
+	if len(wantedKeys) == 0 {
+		return true
+	}
+	for _, key := range wantedKeys {
+		if bytes.Equal(key.Bytes(), actualKey.Bytes()) {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *Filterer) filterStorage(streamFilters config.Subscription, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) error {
+	if !streamFilters.StorageFilter.Off && checkRange(streamFilters.StartingBlock.Int64(), streamFilters.EndingBlock.Int64(), payload.BlockNumber.Int64()) {
+		response.StorageNodesRlp = make(map[common.Hash]map[common.Hash][]byte)
+		stateKeyFilters := make([]common.Hash, 0, len(streamFilters.StorageFilter.Addresses))
+		for _, addr := range streamFilters.StorageFilter.Addresses {
+			keyFilter := ipfs.AddressToKey(common.HexToAddress(addr))
+			stateKeyFilters = append(stateKeyFilters, keyFilter)
+		}
+		storageKeyFilters := make([]common.Hash, 0, len(streamFilters.StorageFilter.StorageKeys))
+		for _, store := range streamFilters.StorageFilter.StorageKeys {
+			keyFilter := ipfs.HexToKey(store)
+			storageKeyFilters = append(storageKeyFilters, keyFilter)
+		}
+		for stateKey, storageNodes := range payload.StorageNodes {
+			if checkNodeKeys(stateKeyFilters, stateKey) {
+				response.StorageNodesRlp[stateKey] = make(map[common.Hash][]byte)
+				for _, storageNode := range storageNodes {
+					if checkNodeKeys(storageKeyFilters, storageNode.Key) {
+						response.StorageNodesRlp[stateKey][storageNode.Key] = storageNode.Value
+					}
+				}
+			}
+		}
+	}
+	return nil
+}

pkg/super_node/filterer_test.go

@@ -0,0 +1,153 @@
+// 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 super_node_test
+
+import (
+	"bytes"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node"
+)
+
+var (
+	filterer                  super_node.ResponseFilterer
+	expectedRctForStorageRLP1 []byte
+	expectedRctForStorageRLP2 []byte
+)
+
+var _ = Describe("Filterer", func() {
+	Describe("FilterResponse", func() {
+		BeforeEach(func() {
+			filterer = super_node.NewResponseFilterer()
+			expectedRctForStorageRLP1 = getReceiptForStorageRLP(mocks.MockReceipts, 0)
+			expectedRctForStorageRLP2 = getReceiptForStorageRLP(mocks.MockReceipts, 1)
+		})
+
+		It("Transcribes all the data from the IPLDPayload into the SuperNodePayload if given an open filter", func() {
+			superNodePayload, err := filterer.FilterResponse(openFilter, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(superNodePayload.HeadersRlp).To(Equal(mocks.MockSeeNodePayload.HeadersRlp))
+			Expect(superNodePayload.UnclesRlp).To(Equal(mocks.MockSeeNodePayload.UnclesRlp))
+			Expect(len(superNodePayload.TransactionsRlp)).To(Equal(2))
+			Expect(super_node.ListContainsBytes(superNodePayload.TransactionsRlp, mocks.MockTransactions.GetRlp(0))).To(BeTrue())
+			Expect(super_node.ListContainsBytes(superNodePayload.TransactionsRlp, mocks.MockTransactions.GetRlp(1))).To(BeTrue())
+			Expect(len(superNodePayload.ReceiptsRlp)).To(Equal(2))
+			Expect(super_node.ListContainsBytes(superNodePayload.ReceiptsRlp, expectedRctForStorageRLP1)).To(BeTrue())
+			Expect(super_node.ListContainsBytes(superNodePayload.ReceiptsRlp, expectedRctForStorageRLP2)).To(BeTrue())
+			Expect(len(superNodePayload.StateNodesRlp)).To(Equal(2))
+			Expect(superNodePayload.StateNodesRlp[mocks.ContractLeafKey]).To(Equal(mocks.ValueBytes))
+			Expect(superNodePayload.StateNodesRlp[mocks.AnotherContractLeafKey]).To(Equal(mocks.AnotherValueBytes))
+			Expect(superNodePayload.StorageNodesRlp).To(Equal(mocks.MockSeeNodePayload.StorageNodesRlp))
+		})
+
+		It("Applies filters from the provided config.Subscription", func() {
+			superNodePayload1, err := filterer.FilterResponse(rctContractFilter, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload1.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload1.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload1.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload1.TransactionsRlp)).To(Equal(0))
+			Expect(len(superNodePayload1.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload1.StateNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload1.ReceiptsRlp)).To(Equal(1))
+			Expect(superNodePayload1.ReceiptsRlp[0]).To(Equal(expectedRctForStorageRLP2))
+
+			superNodePayload2, err := filterer.FilterResponse(rctTopicsFilter, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload2.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload2.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload2.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload2.TransactionsRlp)).To(Equal(0))
+			Expect(len(superNodePayload2.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload2.StateNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload2.ReceiptsRlp)).To(Equal(1))
+			Expect(superNodePayload2.ReceiptsRlp[0]).To(Equal(expectedRctForStorageRLP1))
+
+			superNodePayload3, err := filterer.FilterResponse(rctTopicsAndContractFilter, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload3.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload3.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload3.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload3.TransactionsRlp)).To(Equal(0))
+			Expect(len(superNodePayload3.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload3.StateNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload3.ReceiptsRlp)).To(Equal(1))
+			Expect(superNodePayload3.ReceiptsRlp[0]).To(Equal(expectedRctForStorageRLP1))
+
+			superNodePayload4, err := filterer.FilterResponse(rctContractsAndTopicFilter, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload4.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload4.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload4.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload4.TransactionsRlp)).To(Equal(0))
+			Expect(len(superNodePayload4.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload4.StateNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload4.ReceiptsRlp)).To(Equal(1))
+			Expect(superNodePayload4.ReceiptsRlp[0]).To(Equal(expectedRctForStorageRLP2))
+
+			superNodePayload5, err := filterer.FilterResponse(rctsForAllCollectedTrxs, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload5.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload5.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload5.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload5.TransactionsRlp)).To(Equal(2))
+			Expect(super_node.ListContainsBytes(superNodePayload5.TransactionsRlp, mocks.MockTransactions.GetRlp(0))).To(BeTrue())
+			Expect(super_node.ListContainsBytes(superNodePayload5.TransactionsRlp, mocks.MockTransactions.GetRlp(1))).To(BeTrue())
+			Expect(len(superNodePayload5.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload5.StateNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload5.ReceiptsRlp)).To(Equal(2))
+			Expect(super_node.ListContainsBytes(superNodePayload5.ReceiptsRlp, expectedRctForStorageRLP1)).To(BeTrue())
+			Expect(super_node.ListContainsBytes(superNodePayload5.ReceiptsRlp, expectedRctForStorageRLP2)).To(BeTrue())
+
+			superNodePayload6, err := filterer.FilterResponse(rctsForSelectCollectedTrxs, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload6.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload6.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload6.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload6.TransactionsRlp)).To(Equal(1))
+			Expect(super_node.ListContainsBytes(superNodePayload5.TransactionsRlp, mocks.MockTransactions.GetRlp(1))).To(BeTrue())
+			Expect(len(superNodePayload6.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload6.StateNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload6.ReceiptsRlp)).To(Equal(1))
+			Expect(superNodePayload4.ReceiptsRlp[0]).To(Equal(expectedRctForStorageRLP2))
+
+			superNodePayload7, err := filterer.FilterResponse(stateFilter, *mocks.MockIPLDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(superNodePayload7.BlockNumber.Int64()).To(Equal(mocks.MockSeeNodePayload.BlockNumber.Int64()))
+			Expect(len(superNodePayload7.HeadersRlp)).To(Equal(0))
+			Expect(len(superNodePayload7.UnclesRlp)).To(Equal(0))
+			Expect(len(superNodePayload7.TransactionsRlp)).To(Equal(0))
+			Expect(len(superNodePayload7.StorageNodesRlp)).To(Equal(0))
+			Expect(len(superNodePayload7.ReceiptsRlp)).To(Equal(0))
+			Expect(len(superNodePayload7.StateNodesRlp)).To(Equal(1))
+			Expect(superNodePayload7.StateNodesRlp[mocks.ContractLeafKey]).To(Equal(mocks.ValueBytes))
+		})
+	})
+})
+
+func getReceiptForStorageRLP(receipts types.Receipts, i int) []byte {
+	receiptForStorage := (*types.ReceiptForStorage)(receipts[i])
+	receiptBuffer := new(bytes.Buffer)
+	err := receiptForStorage.EncodeRLP(receiptBuffer)
+	Expect(err).ToNot(HaveOccurred())
+	return receiptBuffer.Bytes()
+}

pkg/super_node/mocks/repository.go

@@ -0,0 +1,31 @@
+// 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/vulcanizedb/pkg/ipfs"
+
+// CIDRepository is the underlying struct for the Repository interface
+type CIDRepository struct {
+	PassedCIDPayload []*ipfs.CIDPayload
+	ReturnErr        error
+}
+
+// Index indexes a cidPayload in Postgres
+func (repo *CIDRepository) Index(cidPayload *ipfs.CIDPayload) error {
+	repo.PassedCIDPayload = append(repo.PassedCIDPayload, cidPayload)
+	return repo.ReturnErr
+}

pkg/super_node/mocks/retriever.go

@@ -0,0 +1,44 @@
+package mocks
+
+import (
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// MockCIDRetriever is a mock CID retriever for use in tests
+type MockCIDRetriever struct {
+	GapsToRetrieve              [][2]int64
+	GapsToRetrieveErr           error
+	CalledTimes                 int
+	FirstBlockNumberToReturn    int64
+	RetrieveFirstBlockNumberErr error
+}
+
+// RetrieveCIDs mock method
+func (*MockCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, blockNumber int64) (*ipfs.CIDWrapper, error) {
+	panic("implement me")
+}
+
+// RetrieveLastBlockNumber mock method
+func (*MockCIDRetriever) RetrieveLastBlockNumber() (int64, error) {
+	panic("implement me")
+}
+
+// RetrieveFirstBlockNumber mock method
+func (mcr *MockCIDRetriever) RetrieveFirstBlockNumber() (int64, error) {
+	return mcr.FirstBlockNumberToReturn, mcr.RetrieveFirstBlockNumberErr
+}
+
+// RetrieveGapsInData mock method
+func (mcr *MockCIDRetriever) RetrieveGapsInData() ([][2]int64, error) {
+	mcr.CalledTimes++
+	return mcr.GapsToRetrieve, mcr.GapsToRetrieveErr
+}
+
+// SetGapsToRetrieve mock method
+func (mcr *MockCIDRetriever) SetGapsToRetrieve(gaps [][2]int64) {
+	if mcr.GapsToRetrieve == nil {
+		mcr.GapsToRetrieve = make([][2]int64, 0)
+	}
+	mcr.GapsToRetrieve = append(mcr.GapsToRetrieve, gaps...)
+}

pkg/super_node/repository.go

@@ -0,0 +1,156 @@
+// 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 super_node
+
+import (
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// CIDRepository is an interface for indexing ipfs.CIDPayloads
+type CIDRepository interface {
+	Index(cidPayload *ipfs.CIDPayload) error
+}
+
+// Repository is the underlying struct for the CIDRepository interface
+type Repository struct {
+	db *postgres.DB
+}
+
+// NewCIDRepository creates a new pointer to a Repository which satisfies the CIDRepository interface
+func NewCIDRepository(db *postgres.DB) *Repository {
+	return &Repository{
+		db: db,
+	}
+}
+
+// Index indexes a cidPayload in Postgres
+func (repo *Repository) Index(cidPayload *ipfs.CIDPayload) error {
+	tx, err := repo.db.Beginx()
+	if err != nil {
+		return err
+	}
+	headerID, err := repo.indexHeaderCID(tx, cidPayload.HeaderCID, cidPayload.BlockNumber, cidPayload.BlockHash.Hex())
+	if err != nil {
+		rollbackErr := tx.Rollback()
+		if rollbackErr != nil {
+			log.Error(rollbackErr)
+		}
+		return err
+	}
+	for uncleHash, cid := range cidPayload.UncleCIDs {
+		err = repo.indexUncleCID(tx, cid, cidPayload.BlockNumber, uncleHash.Hex())
+		if err != nil {
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
+			return err
+		}
+	}
+	err = repo.indexTransactionAndReceiptCIDs(tx, cidPayload, headerID)
+	if err != nil {
+		rollbackErr := tx.Rollback()
+		if rollbackErr != nil {
+			log.Error(rollbackErr)
+		}
+		return err
+	}
+	err = repo.indexStateAndStorageCIDs(tx, cidPayload, headerID)
+	if err != nil {
+		rollbackErr := tx.Rollback()
+		if rollbackErr != nil {
+			log.Error(rollbackErr)
+		}
+		return err
+	}
+	return tx.Commit()
+}
+
+func (repo *Repository) indexHeaderCID(tx *sqlx.Tx, cid, blockNumber, hash string) (int64, error) {
+	var headerID int64
+	err := tx.QueryRowx(`INSERT INTO public.header_cids (block_number, block_hash, cid, final) VALUES ($1, $2, $3, $4)
+								ON CONFLICT (block_number, block_hash) DO UPDATE SET (cid, final) = ($3, $4)
+								RETURNING id`,
+		blockNumber, hash, cid, true).Scan(&headerID)
+	return headerID, err
+}
+
+func (repo *Repository) indexUncleCID(tx *sqlx.Tx, cid, blockNumber, hash string) error {
+	_, err := tx.Exec(`INSERT INTO public.header_cids (block_number, block_hash, cid, final) VALUES ($1, $2, $3, $4)
+								ON CONFLICT (block_number, block_hash) DO UPDATE SET (cid, final) = ($3, $4)`,
+		blockNumber, hash, cid, false)
+	return err
+}
+
+func (repo *Repository) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *ipfs.CIDPayload, headerID int64) error {
+	for hash, trxCidMeta := range payload.TransactionCIDs {
+		var txID int64
+		err := tx.QueryRowx(`INSERT INTO public.transaction_cids (header_id, tx_hash, cid, dst, src) VALUES ($1, $2, $3, $4, $5) 
+									ON CONFLICT (header_id, tx_hash) DO UPDATE SET (cid, dst, src) = ($3, $4, $5)
+									RETURNING id`,
+			headerID, hash.Hex(), trxCidMeta.CID, trxCidMeta.Dst, trxCidMeta.Src).Scan(&txID)
+		if err != nil {
+			return err
+		}
+		receiptCidMeta, ok := payload.ReceiptCIDs[hash]
+		if ok {
+			err = repo.indexReceiptCID(tx, receiptCidMeta, txID)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (repo *Repository) indexReceiptCID(tx *sqlx.Tx, cidMeta *ipfs.ReceiptMetaData, txID int64) error {
+	_, err := tx.Exec(`INSERT INTO public.receipt_cids (tx_id, cid, contract, topic0s) VALUES ($1, $2, $3, $4)`,
+		txID, cidMeta.CID, cidMeta.ContractAddress, pq.Array(cidMeta.Topic0s))
+	return err
+}
+
+func (repo *Repository) indexStateAndStorageCIDs(tx *sqlx.Tx, payload *ipfs.CIDPayload, headerID int64) error {
+	for accountKey, stateCID := range payload.StateNodeCIDs {
+		var stateID int64
+		err := tx.QueryRowx(`INSERT INTO public.state_cids (header_id, state_key, cid, leaf) VALUES ($1, $2, $3, $4)
+									ON CONFLICT (header_id, state_key) DO UPDATE SET (cid, leaf) = ($3, $4)
+									RETURNING id`,
+			headerID, accountKey.Hex(), stateCID.CID, stateCID.Leaf).Scan(&stateID)
+		if err != nil {
+			return err
+		}
+		for _, storageCID := range payload.StorageNodeCIDs[accountKey] {
+			err = repo.indexStorageCID(tx, storageCID, stateID)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (repo *Repository) indexStorageCID(tx *sqlx.Tx, storageCID ipfs.StorageNodeCID, stateID int64) error {
+	_, err := tx.Exec(`INSERT INTO public.storage_cids (state_id, storage_key, cid, leaf) VALUES ($1, $2, $3, $4) 
+								   ON CONFLICT (state_id, storage_key) DO UPDATE SET (cid, leaf) = ($3, $4)`,
+		stateID, storageCID.Key, storageCID.CID, storageCID.Leaf)
+	return err
+}

pkg/super_node/repository_test.go

@@ -0,0 +1,111 @@
+// 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 super_node_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node"
+)
+
+var (
+	db   *postgres.DB
+	err  error
+	repo super_node.CIDRepository
+)
+
+var _ = Describe("Repository", func() {
+	BeforeEach(func() {
+		db, err = super_node.SetupDB()
+		Expect(err).ToNot(HaveOccurred())
+		repo = super_node.NewCIDRepository(db)
+	})
+	AfterEach(func() {
+		super_node.TearDownDB(db)
+	})
+
+	Describe("Index", func() {
+		It("Indexes CIDs and related metadata into vulcanizedb", func() {
+			err = repo.Index(mocks.MockCIDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			pgStr := `SELECT cid FROM header_cids
+				WHERE block_number = $1 AND final IS TRUE`
+			// check header was properly indexed
+			headers := make([]string, 0)
+			err = db.Select(&headers, pgStr, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(headers)).To(Equal(1))
+			Expect(headers[0]).To(Equal("mockHeaderCID"))
+			// check trxs were properly indexed
+			trxs := make([]string, 0)
+			pgStr = `SELECT transaction_cids.cid FROM transaction_cids INNER JOIN header_cids ON (transaction_cids.header_id = header_cids.id)
+				WHERE header_cids.block_number = $1`
+			err = db.Select(&trxs, pgStr, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(trxs)).To(Equal(2))
+			Expect(super_node.ListContainsString(trxs, "mockTrxCID1")).To(BeTrue())
+			Expect(super_node.ListContainsString(trxs, "mockTrxCID2")).To(BeTrue())
+			// check receipts were properly indexed
+			rcts := make([]string, 0)
+			pgStr = `SELECT receipt_cids.cid FROM receipt_cids, transaction_cids, header_cids
+				WHERE receipt_cids.tx_id = transaction_cids.id 
+				AND transaction_cids.header_id = header_cids.id
+				AND header_cids.block_number = $1`
+			err = db.Select(&rcts, pgStr, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(rcts)).To(Equal(2))
+			Expect(super_node.ListContainsString(rcts, "mockRctCID1")).To(BeTrue())
+			Expect(super_node.ListContainsString(rcts, "mockRctCID2")).To(BeTrue())
+			// check that state nodes were properly indexed
+			stateNodes := make([]ipfs.StateNodeCID, 0)
+			pgStr = `SELECT state_cids.cid, state_cids.state_key, state_cids.leaf FROM state_cids INNER JOIN header_cids ON (state_cids.header_id = header_cids.id)
+				WHERE header_cids.block_number = $1`
+			err = db.Select(&stateNodes, pgStr, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(stateNodes)).To(Equal(2))
+			for _, stateNode := range stateNodes {
+				if stateNode.CID == "mockStateCID1" {
+					Expect(stateNode.Leaf).To(Equal(true))
+					Expect(stateNode.Key).To(Equal(mocks.ContractLeafKey.Hex()))
+				}
+				if stateNode.CID == "mockStateCID2" {
+					Expect(stateNode.Leaf).To(Equal(true))
+					Expect(stateNode.Key).To(Equal(mocks.AnotherContractLeafKey.Hex()))
+				}
+			}
+			// check that storage nodes were properly indexed
+			storageNodes := make([]ipfs.StorageNodeCID, 0)
+			pgStr = `SELECT storage_cids.cid, state_cids.state_key, storage_cids.storage_key, storage_cids.leaf FROM storage_cids, state_cids, header_cids
+				WHERE storage_cids.state_id = state_cids.id 
+				AND state_cids.header_id = header_cids.id
+				AND header_cids.block_number = $1`
+			err = db.Select(&storageNodes, pgStr, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(storageNodes)).To(Equal(1))
+			Expect(storageNodes[0]).To(Equal(ipfs.StorageNodeCID{
+				CID:      "mockStorageCID",
+				Leaf:     true,
+				Key:      "0x0000000000000000000000000000000000000000000000000000000000000001",
+				StateKey: mocks.ContractLeafKey.Hex(),
+			}))
+		})
+	})
+})

pkg/super_node/retriever.go

@@ -0,0 +1,329 @@
+// 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 super_node
+
+import (
+	"math/big"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+// CIDRetriever is the interface for retrieving CIDs from the Postgres cache
+type CIDRetriever interface {
+	RetrieveCIDs(streamFilters config.Subscription, blockNumber int64) (*ipfs.CIDWrapper, error)
+	RetrieveLastBlockNumber() (int64, error)
+	RetrieveFirstBlockNumber() (int64, error)
+	RetrieveGapsInData() ([][2]int64, error)
+}
+
+// EthCIDRetriever is the underlying struct supporting the CIDRetriever interface
+type EthCIDRetriever struct {
+	db *postgres.DB
+}
+
+// NewCIDRetriever returns a pointer to a new EthCIDRetriever which supports the CIDRetriever interface
+func NewCIDRetriever(db *postgres.DB) *EthCIDRetriever {
+	return &EthCIDRetriever{
+		db: db,
+	}
+}
+
+func (ecr *EthCIDRetriever) RetrieveFirstBlockNumber() (int64, error) {
+	var blockNumber int64
+	err := ecr.db.Get(&blockNumber, "SELECT block_number FROM header_cids ORDER BY block_number ASC LIMIT 1")
+	return blockNumber, err
+}
+
+// GetLastBlockNumber is used to retrieve the latest block number in the cache
+func (ecr *EthCIDRetriever) RetrieveLastBlockNumber() (int64, error) {
+	var blockNumber int64
+	err := ecr.db.Get(&blockNumber, "SELECT block_number FROM header_cids ORDER BY block_number DESC LIMIT 1 ")
+	return blockNumber, err
+}
+
+// RetrieveCIDs is used to retrieve all of the CIDs which conform to the passed StreamFilters
+func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, blockNumber int64) (*ipfs.CIDWrapper, error) {
+	log.Debug("retrieving cids")
+	var err error
+	tx, err := ecr.db.Beginx()
+	if err != nil {
+		return nil, err
+	}
+	// THIS IS SUPER EXPENSIVE HAVING TO CYCLE THROUGH EACH BLOCK, NEED BETTER WAY TO FETCH CIDS
+	// WHILE STILL MAINTAINING RELATION INFO ABOUT WHAT BLOCK THE CIDS BELONG TO
+	cw := new(ipfs.CIDWrapper)
+	cw.BlockNumber = big.NewInt(blockNumber)
+
+	// Retrieve cached header CIDs
+	if !streamFilters.HeaderFilter.Off {
+		cw.Headers, err = ecr.retrieveHeaderCIDs(tx, streamFilters, blockNumber)
+		if err != nil {
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
+			log.Error("header cid retrieval error")
+			return nil, err
+		}
+		if !streamFilters.HeaderFilter.FinalOnly {
+			cw.Uncles, err = ecr.retrieveUncleCIDs(tx, streamFilters, blockNumber)
+			if err != nil {
+				rollbackErr := tx.Rollback()
+				if rollbackErr != nil {
+					log.Error(rollbackErr)
+				}
+				log.Error("uncle cid retrieval error")
+				return nil, err
+			}
+		}
+	}
+
+	// Retrieve cached trx CIDs
+	var trxIds []int64
+	if !streamFilters.TrxFilter.Off {
+		cw.Transactions, trxIds, err = ecr.retrieveTrxCIDs(tx, streamFilters, blockNumber)
+		if err != nil {
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
+			log.Error("transaction cid retrieval error")
+			return nil, err
+		}
+	}
+
+	// Retrieve cached receipt CIDs
+	if !streamFilters.ReceiptFilter.Off {
+		cw.Receipts, err = ecr.retrieveRctCIDs(tx, streamFilters, blockNumber, trxIds)
+		if err != nil {
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
+			log.Error("receipt cid retrieval error")
+			return nil, err
+		}
+	}
+
+	// Retrieve cached state CIDs
+	if !streamFilters.StateFilter.Off {
+		cw.StateNodes, err = ecr.retrieveStateCIDs(tx, streamFilters, blockNumber)
+		if err != nil {
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
+			log.Error("state cid retrieval error")
+			return nil, err
+		}
+	}
+
+	// Retrieve cached storage CIDs
+	if !streamFilters.StorageFilter.Off {
+		cw.StorageNodes, err = ecr.retrieveStorageCIDs(tx, streamFilters, blockNumber)
+		if err != nil {
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
+			log.Error("storage cid retrieval error")
+			return nil, err
+		}
+	}
+
+	return cw, tx.Commit()
+}
+
+func (ecr *EthCIDRetriever) retrieveHeaderCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, error) {
+	log.Debug("retrieving header cids for block ", blockNumber)
+	headers := make([]string, 0)
+	pgStr := `SELECT cid FROM header_cids
+				WHERE block_number = $1 AND final IS TRUE`
+	err := tx.Select(&headers, pgStr, blockNumber)
+	return headers, err
+}
+
+func (ecr *EthCIDRetriever) retrieveUncleCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, error) {
+	log.Debug("retrieving header cids for block ", blockNumber)
+	headers := make([]string, 0)
+	pgStr := `SELECT cid FROM header_cids
+				WHERE block_number = $1 AND final IS FALSE`
+	err := tx.Select(&headers, pgStr, blockNumber)
+	return headers, err
+}
+
+func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, []int64, error) {
+	log.Debug("retrieving transaction cids for block ", blockNumber)
+	args := make([]interface{}, 0, 3)
+	type result struct {
+		ID  int64  `db:"id"`
+		Cid string `db:"cid"`
+	}
+	results := make([]result, 0)
+	pgStr := `SELECT transaction_cids.id, transaction_cids.cid FROM transaction_cids INNER JOIN header_cids ON (transaction_cids.header_id = header_cids.id)
+			WHERE header_cids.block_number = $1`
+	args = append(args, blockNumber)
+	if len(streamFilters.TrxFilter.Dst) > 0 {
+		pgStr += ` AND transaction_cids.dst = ANY($2::VARCHAR(66)[])`
+		args = append(args, pq.Array(streamFilters.TrxFilter.Dst))
+	}
+	if len(streamFilters.TrxFilter.Src) > 0 {
+		pgStr += ` AND transaction_cids.src = ANY($3::VARCHAR(66)[])`
+		args = append(args, pq.Array(streamFilters.TrxFilter.Src))
+	}
+	err := tx.Select(&results, pgStr, args...)
+	if err != nil {
+		return nil, nil, err
+	}
+	ids := make([]int64, 0, len(results))
+	cids := make([]string, 0, len(results))
+	for _, res := range results {
+		cids = append(cids, res.Cid)
+		ids = append(ids, res.ID)
+	}
+	return cids, ids, nil
+}
+
+func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64, trxIds []int64) ([]string, error) {
+	log.Debug("retrieving receipt cids for block ", blockNumber)
+	args := make([]interface{}, 0, 4)
+	pgStr := `SELECT receipt_cids.cid FROM receipt_cids, transaction_cids, header_cids
+			WHERE receipt_cids.tx_id = transaction_cids.id 
+			AND transaction_cids.header_id = header_cids.id
+			AND header_cids.block_number = $1`
+	args = append(args, blockNumber)
+	if len(streamFilters.ReceiptFilter.Topic0s) > 0 {
+		pgStr += ` AND ((receipt_cids.topic0s && $2::VARCHAR(66)[]`
+		args = append(args, pq.Array(streamFilters.ReceiptFilter.Topic0s))
+		if len(streamFilters.ReceiptFilter.Contracts) > 0 {
+			pgStr += ` AND receipt_cids.contract = ANY($3::VARCHAR(66)[]))`
+			args = append(args, pq.Array(streamFilters.ReceiptFilter.Contracts))
+			if len(trxIds) > 0 {
+				pgStr += ` OR receipt_cids.tx_id = ANY($4::INTEGER[]))`
+				args = append(args, pq.Array(trxIds))
+			} else {
+				pgStr += `)`
+			}
+		} else {
+			pgStr += `)`
+			if len(trxIds) > 0 {
+				pgStr += ` OR receipt_cids.tx_id = ANY($3::INTEGER[]))`
+				args = append(args, pq.Array(trxIds))
+			} else {
+				pgStr += `)`
+			}
+		}
+	} else {
+		if len(streamFilters.ReceiptFilter.Contracts) > 0 {
+			pgStr += ` AND (receipt_cids.contract = ANY($2::VARCHAR(66)[])`
+			args = append(args, pq.Array(streamFilters.ReceiptFilter.Contracts))
+			if len(trxIds) > 0 {
+				pgStr += ` OR receipt_cids.tx_id = ANY($3::INTEGER[]))`
+				args = append(args, pq.Array(trxIds))
+			} else {
+				pgStr += `)`
+			}
+		} else if len(trxIds) > 0 {
+			pgStr += ` AND receipt_cids.tx_id = ANY($2::INTEGER[])`
+			args = append(args, pq.Array(trxIds))
+		}
+	}
+	receiptCids := make([]string, 0)
+	err := tx.Select(&receiptCids, pgStr, args...)
+	return receiptCids, err
+}
+
+func (ecr *EthCIDRetriever) retrieveStateCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]ipfs.StateNodeCID, error) {
+	log.Debug("retrieving state cids for block ", blockNumber)
+	args := make([]interface{}, 0, 2)
+	pgStr := `SELECT state_cids.cid, state_cids.state_key, state_cids.leaf FROM state_cids INNER JOIN header_cids ON (state_cids.header_id = header_cids.id)
+			WHERE header_cids.block_number = $1`
+	args = append(args, blockNumber)
+	addrLen := len(streamFilters.StateFilter.Addresses)
+	if addrLen > 0 {
+		keys := make([]string, 0, addrLen)
+		for _, addr := range streamFilters.StateFilter.Addresses {
+			keys = append(keys, ipfs.HexToKey(addr).Hex())
+		}
+		pgStr += ` AND state_cids.state_key = ANY($2::VARCHAR(66)[])`
+		args = append(args, pq.Array(keys))
+	}
+	if !streamFilters.StorageFilter.IntermediateNodes {
+		pgStr += ` AND state_cids.leaf = TRUE`
+	}
+	stateNodeCIDs := make([]ipfs.StateNodeCID, 0)
+	err := tx.Select(&stateNodeCIDs, pgStr, args...)
+	return stateNodeCIDs, err
+}
+
+func (ecr *EthCIDRetriever) retrieveStorageCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]ipfs.StorageNodeCID, error) {
+	log.Debug("retrieving storage cids for block ", blockNumber)
+	args := make([]interface{}, 0, 3)
+	pgStr := `SELECT storage_cids.cid, state_cids.state_key, storage_cids.storage_key, storage_cids.leaf FROM storage_cids, state_cids, header_cids
+			WHERE storage_cids.state_id = state_cids.id 
+			AND state_cids.header_id = header_cids.id
+			AND header_cids.block_number = $1`
+	args = append(args, blockNumber)
+	addrLen := len(streamFilters.StorageFilter.Addresses)
+	if addrLen > 0 {
+		keys := make([]string, 0, addrLen)
+		for _, addr := range streamFilters.StorageFilter.Addresses {
+			keys = append(keys, ipfs.HexToKey(addr).Hex())
+		}
+		pgStr += ` AND state_cids.state_key = ANY($2::VARCHAR(66)[])`
+		args = append(args, pq.Array(keys))
+	}
+	if len(streamFilters.StorageFilter.StorageKeys) > 0 {
+		pgStr += ` AND storage_cids.storage_key = ANY($3::VARCHAR(66)[])`
+		args = append(args, pq.Array(streamFilters.StorageFilter.StorageKeys))
+	}
+	if !streamFilters.StorageFilter.IntermediateNodes {
+		pgStr += ` AND storage_cids.leaf = TRUE`
+	}
+	storageNodeCIDs := make([]ipfs.StorageNodeCID, 0)
+	err := tx.Select(&storageNodeCIDs, pgStr, args...)
+	return storageNodeCIDs, err
+}
+
+type gap struct {
+	Start int64 `db:"start"`
+	Stop  int64 `db:"stop"`
+}
+
+func (ecr *EthCIDRetriever) RetrieveGapsInData() ([][2]int64, error) {
+	pgStr := `SELECT header_cids.block_number + 1 AS start, min(fr.block_number) - 1 AS stop FROM header_cids
+				LEFT JOIN header_cids r on header_cids.block_number = r.block_number - 1
+				LEFT JOIN header_cids fr on header_cids.block_number < fr.block_number
+				WHERE r.block_number is NULL and fr.block_number IS NOT NULL
+				GROUP BY header_cids.block_number, r.block_number`
+	gaps := make([]gap, 0)
+	err := ecr.db.Select(&gaps, pgStr)
+	if err != nil {
+		return nil, err
+	}
+	gapRanges := make([][2]int64, 0)
+	for _, gap := range gaps {
+		gapRanges = append(gapRanges, [2]int64{gap.Start, gap.Stop})
+	}
+	return gapRanges, nil
+}

pkg/super_node/retriever_test.go

@@ -0,0 +1,318 @@
+// 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 super_node_test
+
+import (
+	"math/big"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node"
+)
+
+var (
+	retriever  super_node.CIDRetriever
+	openFilter = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter:  config.HeaderFilter{},
+		TrxFilter:     config.TrxFilter{},
+		ReceiptFilter: config.ReceiptFilter{},
+		StateFilter:   config.StateFilter{},
+		StorageFilter: config.StorageFilter{},
+	}
+	rctContractFilter = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{
+			Off: true,
+		},
+		ReceiptFilter: config.ReceiptFilter{
+			Contracts: []string{"0x0000000000000000000000000000000000000001"},
+		},
+		StateFilter: config.StateFilter{
+			Off: true,
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+	rctTopicsFilter = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{
+			Off: true,
+		},
+		ReceiptFilter: config.ReceiptFilter{
+			Topic0s: []string{"0x0000000000000000000000000000000000000000000000000000000000000004"},
+		},
+		StateFilter: config.StateFilter{
+			Off: true,
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+	rctTopicsAndContractFilter = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{
+			Off: true,
+		},
+		ReceiptFilter: config.ReceiptFilter{
+			Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000004", "0x0000000000000000000000000000000000000000000000000000000000000005"},
+			Contracts: []string{"0x0000000000000000000000000000000000000000"},
+		},
+		StateFilter: config.StateFilter{
+			Off: true,
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+	rctContractsAndTopicFilter = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{
+			Off: true,
+		},
+		ReceiptFilter: config.ReceiptFilter{
+			Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000005"},
+			Contracts: []string{"0x0000000000000000000000000000000000000000", "0x0000000000000000000000000000000000000001"},
+		},
+		StateFilter: config.StateFilter{
+			Off: true,
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+	rctsForAllCollectedTrxs = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{}, // Trx filter open so we will collect all trxs, therefore we will also collect all corresponding rcts despite rct filter
+		ReceiptFilter: config.ReceiptFilter{
+			Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000006"}, // Topic isn't one of the topics we have
+			Contracts: []string{"0x0000000000000000000000000000000000000002"},                         // Contract isn't one of the contracts we have
+		},
+		StateFilter: config.StateFilter{
+			Off: true,
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+	rctsForSelectCollectedTrxs = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{
+			Dst: []string{"0x0000000000000000000000000000000000000001"}, // We only filter for one of the trxs so we will only get the one corresponding receipt
+		},
+		ReceiptFilter: config.ReceiptFilter{
+			Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000006"}, // Topic isn't one of the topics we have
+			Contracts: []string{"0x0000000000000000000000000000000000000002"},                         // Contract isn't one of the contracts we have
+		},
+		StateFilter: config.StateFilter{
+			Off: true,
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+	stateFilter = config.Subscription{
+		StartingBlock: big.NewInt(0),
+		EndingBlock:   big.NewInt(1),
+		HeaderFilter: config.HeaderFilter{
+			Off: true,
+		},
+		TrxFilter: config.TrxFilter{
+			Off: true,
+		},
+		ReceiptFilter: config.ReceiptFilter{
+			Off: true,
+		},
+		StateFilter: config.StateFilter{
+			Addresses: []string{mocks.Address.Hex()},
+		},
+		StorageFilter: config.StorageFilter{
+			Off: true,
+		},
+	}
+)
+
+var _ = Describe("Retriever", func() {
+	BeforeEach(func() {
+		db, err = super_node.SetupDB()
+		Expect(err).ToNot(HaveOccurred())
+		repo = super_node.NewCIDRepository(db)
+		err = repo.Index(mocks.MockCIDPayload)
+		Expect(err).ToNot(HaveOccurred())
+		retriever = super_node.NewCIDRetriever(db)
+	})
+	AfterEach(func() {
+		super_node.TearDownDB(db)
+	})
+
+	Describe("RetrieveCIDs", func() {
+		It("Retrieves all CIDs for the given blocknumber when provided an open filter", func() {
+			cidWrapper, err := retriever.RetrieveCIDs(openFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper.Headers)).To(Equal(1))
+			Expect(cidWrapper.Headers).To(Equal(mocks.MockCIDWrapper.Headers))
+			Expect(len(cidWrapper.Transactions)).To(Equal(2))
+			Expect(super_node.ListContainsString(cidWrapper.Transactions, mocks.MockCIDWrapper.Transactions[0])).To(BeTrue())
+			Expect(super_node.ListContainsString(cidWrapper.Transactions, mocks.MockCIDWrapper.Transactions[1])).To(BeTrue())
+			Expect(len(cidWrapper.Receipts)).To(Equal(2))
+			Expect(super_node.ListContainsString(cidWrapper.Receipts, mocks.MockCIDWrapper.Receipts[0])).To(BeTrue())
+			Expect(super_node.ListContainsString(cidWrapper.Receipts, mocks.MockCIDWrapper.Receipts[1])).To(BeTrue())
+			Expect(len(cidWrapper.StateNodes)).To(Equal(2))
+			for _, stateNode := range cidWrapper.StateNodes {
+				if stateNode.CID == "mockStateCID1" {
+					Expect(stateNode.Key).To(Equal(mocks.ContractLeafKey.Hex()))
+					Expect(stateNode.Leaf).To(Equal(true))
+				}
+				if stateNode.CID == "mockStateCID2" {
+					Expect(stateNode.Key).To(Equal(mocks.AnotherContractLeafKey.Hex()))
+					Expect(stateNode.Leaf).To(Equal(true))
+				}
+			}
+			Expect(len(cidWrapper.StorageNodes)).To(Equal(1))
+			Expect(cidWrapper.StorageNodes).To(Equal(mocks.MockCIDWrapper.StorageNodes))
+		})
+	})
+
+	Describe("RetrieveCIDs", func() {
+		It("Applies filters from the provided config.Subscription", func() {
+			cidWrapper1, err := retriever.RetrieveCIDs(rctContractFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper1.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper1.Headers)).To(Equal(0))
+			Expect(len(cidWrapper1.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper1.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper1.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper1.Receipts)).To(Equal(1))
+			Expect(cidWrapper1.Receipts[0]).To(Equal("mockRctCID2"))
+
+			cidWrapper2, err := retriever.RetrieveCIDs(rctTopicsFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper2.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper2.Headers)).To(Equal(0))
+			Expect(len(cidWrapper2.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper2.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper2.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper2.Receipts)).To(Equal(1))
+			Expect(cidWrapper2.Receipts[0]).To(Equal("mockRctCID1"))
+
+			cidWrapper3, err := retriever.RetrieveCIDs(rctTopicsAndContractFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper3.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper3.Headers)).To(Equal(0))
+			Expect(len(cidWrapper3.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper3.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper3.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper3.Receipts)).To(Equal(1))
+			Expect(cidWrapper3.Receipts[0]).To(Equal("mockRctCID1"))
+
+			cidWrapper4, err := retriever.RetrieveCIDs(rctContractsAndTopicFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper4.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper4.Headers)).To(Equal(0))
+			Expect(len(cidWrapper4.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper4.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper4.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper4.Receipts)).To(Equal(1))
+			Expect(cidWrapper4.Receipts[0]).To(Equal("mockRctCID2"))
+
+			cidWrapper5, err := retriever.RetrieveCIDs(rctsForAllCollectedTrxs, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper5.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper5.Headers)).To(Equal(0))
+			Expect(len(cidWrapper5.Transactions)).To(Equal(2))
+			Expect(super_node.ListContainsString(cidWrapper5.Transactions, "mockTrxCID1")).To(BeTrue())
+			Expect(super_node.ListContainsString(cidWrapper5.Transactions, "mockTrxCID2")).To(BeTrue())
+			Expect(len(cidWrapper5.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper5.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper5.Receipts)).To(Equal(2))
+			Expect(super_node.ListContainsString(cidWrapper5.Receipts, "mockRctCID1")).To(BeTrue())
+			Expect(super_node.ListContainsString(cidWrapper5.Receipts, "mockRctCID2")).To(BeTrue())
+
+			cidWrapper6, err := retriever.RetrieveCIDs(rctsForSelectCollectedTrxs, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper6.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper6.Headers)).To(Equal(0))
+			Expect(len(cidWrapper6.Transactions)).To(Equal(1))
+			Expect(cidWrapper6.Transactions[0]).To(Equal("mockTrxCID2"))
+			Expect(len(cidWrapper6.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper6.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper6.Receipts)).To(Equal(1))
+			Expect(cidWrapper6.Receipts[0]).To(Equal("mockRctCID2"))
+
+			cidWrapper7, err := retriever.RetrieveCIDs(stateFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper7.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper7.Headers)).To(Equal(0))
+			Expect(len(cidWrapper7.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper7.Receipts)).To(Equal(0))
+			Expect(len(cidWrapper7.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper7.StateNodes)).To(Equal(1))
+			Expect(cidWrapper7.StateNodes[0]).To(Equal(ipfs.StateNodeCID{
+				Leaf: true,
+				Key:  mocks.ContractLeafKey.Hex(),
+				CID:  "mockStateCID1",
+			}))
+		})
+	})
+
+	Describe("RetrieveFirstBlockNumber", func() {
+		It("Gets the number of the first block that has data in the database", func() {
+			num, err := retriever.RetrieveFirstBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(num).To(Equal(int64(1)))
+		})
+	})
+
+	Describe("RetrieveLastBlockNumber", func() {
+		It("Gets the number of the latest block that has data in the database", func() {
+			num, err := retriever.RetrieveLastBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(num).To(Equal(int64(1)))
+		})
+	})
+})

pkg/super_node/seed_node_suite_test.go

@@ -0,0 +1,35 @@
+// 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 super_node_test
+
+import (
+	"io/ioutil"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/sirupsen/logrus"
+)
+
+func TestSuperNode(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Seed Node Suite Test")
+}
+
+var _ = BeforeSuite(func() {
+	logrus.SetOutput(ioutil.Discard)
+})

pkg/super_node/service.go

@@ -0,0 +1,432 @@
+// 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 super_node
+
+import (
+	"sync"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/params"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+)
+
+const (
+	payloadChanBufferSize = 20000 // the max eth sub buffer size
+)
+
+// NodeInterface is the top level interface for streaming, converting to IPLDs, publishing,
+// and indexing all Ethereum data; screening this data; and serving it up to subscribed clients
+// This service is compatible with the Ethereum service interface (node.Service)
+type NodeInterface interface {
+	// APIs(), Protocols(), Start() and Stop()
+	node.Service
+	// Main event loop for syncAndPublish processes
+	SyncAndPublish(wg *sync.WaitGroup, forwardPayloadChan chan<- ipfs.IPLDPayload, forwardQuitchan chan<- bool) error
+	// Main event loop for handling client pub-sub
+	ScreenAndServe(screenAndServePayload <-chan ipfs.IPLDPayload, screenAndServeQuit <-chan bool)
+	// Method to subscribe to receive state diff processing output
+	Subscribe(id rpc.ID, sub chan<- streamer.SuperNodePayload, quitChan chan<- bool, streamFilters config.Subscription)
+	// Method to unsubscribe from state diff processing
+	Unsubscribe(id rpc.ID)
+	// Method to access the Geth node info for this service
+	Node() core.Node
+	// Method used to retrieve the underlying IPFS publisher for this service, so that is can be used for backfilling
+	// This is needed because it's not possible to initialize two ipfs nodes at the same path
+	GetPublisher() ipfs.IPLDPublisher
+}
+
+// Service is the underlying struct for the super node
+type Service struct {
+	// Used to sync access to the Subscriptions
+	sync.Mutex
+	// Interface for streaming statediff payloads over a geth rpc subscription
+	Streamer streamer.IStateDiffStreamer
+	// Interface for converting statediff payloads into ETH-IPLD object payloads
+	Converter ipfs.PayloadConverter
+	// Interface for publishing the ETH-IPLD payloads to IPFS
+	Publisher ipfs.IPLDPublisher
+	// Interface for indexing the CIDs of the published ETH-IPLDs in Postgres
+	Repository CIDRepository
+	// Interface for filtering and serving data according to subscribed clients according to their specification
+	Filterer ResponseFilterer
+	// Interface for fetching ETH-IPLD objects from IPFS
+	IPLDFetcher ipfs.IPLDFetcher
+	// Interface for searching and retrieving CIDs from Postgres index
+	Retriever CIDRetriever
+	// Interface for resolving ipfs blocks to their data types
+	Resolver ipfs.IPLDResolver
+	// Chan the processor uses to subscribe to state diff payloads from the Streamer
+	PayloadChan chan statediff.Payload
+	// Used to signal shutdown of the service
+	QuitChan chan bool
+	// A mapping of rpc.IDs to their subscription channels, mapped to their subscription type (hash of the StreamFilters)
+	Subscriptions map[common.Hash]map[rpc.ID]Subscription
+	// A mapping of subscription hash type to the corresponding StreamFilters
+	SubscriptionTypes map[common.Hash]config.Subscription
+	// Number of workers
+	WorkerPoolSize int
+	// Info for the Geth node that this super node is working with
+	gethNode core.Node
+}
+
+// NewSuperNode creates a new super_node.Interface using an underlying super_node.Service struct
+func NewSuperNode(ipfsPath string, db *postgres.DB, rpcClient core.RpcClient, qc chan bool, workers int, node core.Node) (NodeInterface, error) {
+	publisher, err := ipfs.NewIPLDPublisher(ipfsPath)
+	if err != nil {
+		return nil, err
+	}
+	ipldFetcher, err := ipfs.NewIPLDFetcher(ipfsPath)
+	if err != nil {
+		return nil, err
+	}
+	return &Service{
+		Streamer:          streamer.NewStateDiffStreamer(rpcClient),
+		Repository:        NewCIDRepository(db),
+		Converter:         ipfs.NewPayloadConverter(params.MainnetChainConfig),
+		Publisher:         publisher,
+		Filterer:          NewResponseFilterer(),
+		IPLDFetcher:       ipldFetcher,
+		Retriever:         NewCIDRetriever(db),
+		Resolver:          ipfs.NewIPLDResolver(),
+		PayloadChan:       make(chan statediff.Payload, payloadChanBufferSize),
+		QuitChan:          qc,
+		Subscriptions:     make(map[common.Hash]map[rpc.ID]Subscription),
+		SubscriptionTypes: make(map[common.Hash]config.Subscription),
+		WorkerPoolSize:    workers,
+		gethNode:          node,
+	}, nil
+}
+
+// Protocols exports the services p2p protocols, this service has none
+func (sap *Service) Protocols() []p2p.Protocol {
+	return []p2p.Protocol{}
+}
+
+// APIs returns the RPC descriptors the super node service offers
+func (sap *Service) APIs() []rpc.API {
+	return []rpc.API{
+		{
+			Namespace: APIName,
+			Version:   APIVersion,
+			Service:   NewPublicSuperNodeAPI(sap),
+			Public:    true,
+		},
+	}
+}
+
+// SyncAndPublish is the backend processing loop which streams data from geth, converts it to iplds, publishes them to ipfs, and indexes their cids
+// This continues on no matter if or how many subscribers there are, it then forwards the data to the ScreenAndServe() loop
+// which filters and sends relevant data to client subscriptions, if there are any
+func (sap *Service) SyncAndPublish(wg *sync.WaitGroup, screenAndServePayload chan<- ipfs.IPLDPayload, screenAndServeQuit chan<- bool) error {
+	sub, err := sap.Streamer.Stream(sap.PayloadChan)
+	if err != nil {
+		return err
+	}
+	wg.Add(1)
+
+	// Channels for forwarding data to the publishAndIndex workers
+	publishAndIndexPayload := make(chan ipfs.IPLDPayload, payloadChanBufferSize)
+	publishAndIndexQuit := make(chan bool, sap.WorkerPoolSize)
+	// publishAndIndex worker pool to handle publishing and indexing concurrently, while
+	// limiting the number of Postgres connections we can possibly open so as to prevent error
+	for i := 0; i < sap.WorkerPoolSize; i++ {
+		sap.publishAndIndex(i, publishAndIndexPayload, publishAndIndexQuit)
+	}
+
+	go func() {
+		for {
+			select {
+			case payload := <-sap.PayloadChan:
+				ipldPayload, err := sap.Converter.Convert(payload)
+				if err != nil {
+					log.Error(err)
+					continue
+				}
+				// If we have a ScreenAndServe process running, forward the payload to it
+				select {
+				case screenAndServePayload <- *ipldPayload:
+				default:
+				}
+				// Forward the payload to the publishAndIndex workers
+				select {
+				case publishAndIndexPayload <- *ipldPayload:
+				default:
+				}
+			case err = <-sub.Err():
+				log.Error(err)
+			case <-sap.QuitChan:
+				// If we have a ScreenAndServe process running, forward the quit signal to it
+				select {
+				case screenAndServeQuit <- true:
+				default:
+				}
+				// Also forward a quit signal for each of the workers
+				for i := 0; i < sap.WorkerPoolSize; i++ {
+					select {
+					case publishAndIndexQuit <- true:
+					default:
+					}
+				}
+				log.Info("quiting SyncAndPublish process")
+				wg.Done()
+				return
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (sap *Service) publishAndIndex(id int, publishAndIndexPayload <-chan ipfs.IPLDPayload, publishAndIndexQuit <-chan bool) {
+	go func() {
+		for {
+			select {
+			case payload := <-publishAndIndexPayload:
+				cidPayload, err := sap.Publisher.Publish(&payload)
+				if err != nil {
+					log.Errorf("worker %d error: %v", id, err)
+					continue
+				}
+				err = sap.Repository.Index(cidPayload)
+				if err != nil {
+					log.Errorf("worker %d error: %v", id, err)
+				}
+			case <-publishAndIndexQuit:
+				log.Infof("quiting publishAndIndex worker %d", id)
+				return
+			}
+		}
+	}()
+}
+
+// ScreenAndServe is the loop used to screen data streamed from the state diffing eth node
+// and send the appropriate portions of it to a requesting client subscription, according to their subscription configuration
+func (sap *Service) ScreenAndServe(screenAndServePayload <-chan ipfs.IPLDPayload, screenAndServeQuit <-chan bool) {
+	go func() {
+		for {
+			select {
+			case payload := <-screenAndServePayload:
+				err := sap.sendResponse(payload)
+				if err != nil {
+					log.Error(err)
+				}
+			case <-screenAndServeQuit:
+				log.Info("quiting ScreenAndServe process")
+				return
+			}
+		}
+	}()
+}
+
+func (sap *Service) sendResponse(payload ipfs.IPLDPayload) error {
+	sap.Lock()
+	for ty, subs := range sap.Subscriptions {
+		// Retrieve the subscription parameters for this subscription type
+		subConfig, ok := sap.SubscriptionTypes[ty]
+		if !ok {
+			log.Errorf("subscription configuration for subscription type %s not available", ty.Hex())
+			continue
+		}
+		response, err := sap.Filterer.FilterResponse(subConfig, payload)
+		if err != nil {
+			log.Error(err)
+			continue
+		}
+		for id, sub := range subs {
+			select {
+			case sub.PayloadChan <- response:
+				log.Infof("sending super node payload to subscription %s", id)
+			default:
+				log.Infof("unable to send payload to subscription %s; channel has no receiver", id)
+			}
+		}
+	}
+	sap.Unlock()
+	return nil
+}
+
+// Subscribe is used by the API to subscribe to the service loop
+func (sap *Service) Subscribe(id rpc.ID, sub chan<- streamer.SuperNodePayload, quitChan chan<- bool, streamFilters config.Subscription) {
+	log.Info("Subscribing to the super node service")
+	// Subscription type is defined as the hash of its content
+	// Group subscriptions by type and screen payloads once for subs of the same type
+	by, err := rlp.EncodeToBytes(streamFilters)
+	if err != nil {
+		log.Error(err)
+	}
+	subscriptionHash := crypto.Keccak256(by)
+	subscriptionType := common.BytesToHash(subscriptionHash)
+	subscription := Subscription{
+		PayloadChan: sub,
+		QuitChan:    quitChan,
+	}
+	// If the subscription requests a backfill, use the Postgres index to lookup and retrieve historical data
+	// Otherwise we only filter new data as it is streamed in from the state diffing geth node
+	if streamFilters.BackFill || streamFilters.BackFillOnly {
+		sap.backFill(subscription, id, streamFilters)
+	}
+	if !streamFilters.BackFillOnly {
+		sap.Lock()
+		if sap.Subscriptions[subscriptionType] == nil {
+			sap.Subscriptions[subscriptionType] = make(map[rpc.ID]Subscription)
+		}
+		sap.Subscriptions[subscriptionType][id] = subscription
+		sap.SubscriptionTypes[subscriptionType] = streamFilters
+		sap.Unlock()
+	}
+}
+
+func (sap *Service) backFill(sub Subscription, id rpc.ID, con config.Subscription) {
+	log.Debug("back-filling data for id", id)
+	// Retrieve cached CIDs relevant to this subscriber
+	var endingBlock int64
+	var startingBlock int64
+	var err error
+	startingBlock, err = sap.Retriever.RetrieveFirstBlockNumber()
+	if err != nil {
+		sub.PayloadChan <- streamer.SuperNodePayload{
+			ErrMsg: "unable to set block range start; error: " + err.Error(),
+		}
+	}
+	if startingBlock < con.StartingBlock.Int64() {
+		startingBlock = con.StartingBlock.Int64()
+	}
+	endingBlock, err = sap.Retriever.RetrieveLastBlockNumber()
+	if err != nil {
+		sub.PayloadChan <- streamer.SuperNodePayload{
+			ErrMsg: "unable to set block range end; error: " + err.Error(),
+		}
+	}
+	if endingBlock > con.EndingBlock.Int64() && con.EndingBlock.Int64() > 0 && con.EndingBlock.Int64() > startingBlock {
+		endingBlock = con.EndingBlock.Int64()
+	}
+	log.Debug("backfill starting block:", con.StartingBlock)
+	log.Debug("backfill ending block:", endingBlock)
+	// Backfilled payloads are sent concurrently to the streamed payloads, so the receiver needs to pay attention to
+	// the blocknumbers in the payloads they receive to keep things in order
+	// TODO: separate backfill into a different rpc subscription method altogether?
+	go func() {
+		for i := startingBlock; i <= endingBlock; i++ {
+			cidWrapper, err := sap.Retriever.RetrieveCIDs(con, i)
+			if err != nil {
+				sub.PayloadChan <- streamer.SuperNodePayload{
+					ErrMsg: "CID retrieval error: " + err.Error(),
+				}
+				continue
+			}
+			if ipfs.EmptyCIDWrapper(*cidWrapper) {
+				continue
+			}
+			blocksWrapper, err := sap.IPLDFetcher.FetchCIDs(*cidWrapper)
+			if err != nil {
+				log.Error(err)
+				sub.PayloadChan <- streamer.SuperNodePayload{
+					ErrMsg: "IPLD fetching error: " + err.Error(),
+				}
+				continue
+			}
+			backFillIplds, err := sap.Resolver.ResolveIPLDs(*blocksWrapper)
+			if err != nil {
+				log.Error(err)
+				sub.PayloadChan <- streamer.SuperNodePayload{
+					ErrMsg: "IPLD resolving error: " + err.Error(),
+				}
+				continue
+			}
+			select {
+			case sub.PayloadChan <- backFillIplds:
+				log.Infof("sending super node back-fill payload to subscription %s", id)
+			default:
+				log.Infof("unable to send back-fill payload to subscription %s; channel has no receiver", id)
+			}
+		}
+	}()
+}
+
+// Unsubscribe is used to unsubscribe to the StateDiffingService loop
+func (sap *Service) Unsubscribe(id rpc.ID) {
+	log.Info("Unsubscribing from the super node service")
+	sap.Lock()
+	for ty := range sap.Subscriptions {
+		delete(sap.Subscriptions[ty], id)
+		if len(sap.Subscriptions[ty]) == 0 {
+			// If we removed the last subscription of this type, remove the subscription type outright
+			delete(sap.Subscriptions, ty)
+			delete(sap.SubscriptionTypes, ty)
+		}
+	}
+	sap.Unlock()
+}
+
+// Start is used to begin the service
+func (sap *Service) Start(*p2p.Server) error {
+	log.Info("Starting super node service")
+	wg := new(sync.WaitGroup)
+	payloadChan := make(chan ipfs.IPLDPayload, payloadChanBufferSize)
+	quitChan := make(chan bool, 1)
+	if err := sap.SyncAndPublish(wg, payloadChan, quitChan); err != nil {
+		return err
+	}
+	sap.ScreenAndServe(payloadChan, quitChan)
+	return nil
+}
+
+// Stop is used to close down the service
+func (sap *Service) Stop() error {
+	log.Info("Stopping super node service")
+	close(sap.QuitChan)
+	return nil
+}
+
+// Node returns the Geth node info for this service
+func (sap *Service) Node() core.Node {
+	return sap.gethNode
+}
+
+// close is used to close all listening subscriptions
+func (sap *Service) close() {
+	sap.Lock()
+	for ty, subs := range sap.Subscriptions {
+		for id, sub := range subs {
+			select {
+			case sub.QuitChan <- true:
+				log.Infof("closing subscription %s", id)
+			default:
+				log.Infof("unable to close subscription %s; channel has no receiver", id)
+			}
+		}
+		delete(sap.Subscriptions, ty)
+		delete(sap.SubscriptionTypes, ty)
+	}
+	sap.Unlock()
+}
+
+func (sap *Service) GetPublisher() ipfs.IPLDPublisher {
+	return sap.Publisher
+}

pkg/super_node/service_test.go

@@ -0,0 +1,79 @@
+// 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 super_node_test
+
+import (
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	mocks2 "github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/super_node"
+	mocks3 "github.com/vulcanize/vulcanizedb/pkg/super_node/mocks"
+)
+
+var _ = Describe("Service", func() {
+	Describe("SyncAndPublish", func() {
+		It("Streams statediff.Payloads, converts them to IPLDPayloads, publishes IPLDPayloads, and indexes CIDPayloads", func() {
+			wg := new(sync.WaitGroup)
+			payloadChan := make(chan statediff.Payload, 1)
+			quitChan := make(chan bool, 1)
+			mockCidRepo := &mocks3.CIDRepository{
+				ReturnErr: nil,
+			}
+			mockPublisher := &mocks.IPLDPublisher{
+				ReturnCIDPayload: mocks.MockCIDPayload,
+				ReturnErr:        nil,
+			}
+			mockStreamer := &mocks2.StateDiffStreamer{
+				ReturnSub: &rpc.ClientSubscription{},
+				StreamPayloads: []statediff.Payload{
+					mocks.MockStateDiffPayload,
+				},
+				ReturnErr: nil,
+			}
+			mockConverter := &mocks.PayloadConverter{
+				ReturnIPLDPayload: mocks.MockIPLDPayload,
+				ReturnErr:         nil,
+			}
+			processor := &super_node.Service{
+				Repository:     mockCidRepo,
+				Publisher:      mockPublisher,
+				Streamer:       mockStreamer,
+				Converter:      mockConverter,
+				PayloadChan:    payloadChan,
+				QuitChan:       quitChan,
+				WorkerPoolSize: 1,
+			}
+			err := processor.SyncAndPublish(wg, nil, nil)
+			Expect(err).ToNot(HaveOccurred())
+			time.Sleep(2 * time.Second)
+			quitChan <- true
+			wg.Wait()
+			Expect(mockConverter.PassedStatediffPayload).To(Equal(mocks.MockStateDiffPayload))
+			Expect(len(mockCidRepo.PassedCIDPayload)).To(Equal(1))
+			Expect(mockCidRepo.PassedCIDPayload[0]).To(Equal(mocks.MockCIDPayload))
+			Expect(mockPublisher.PassedIPLDPayload).To(Equal(mocks.MockIPLDPayload))
+			Expect(mockStreamer.PassedPayloadChan).To(Equal(payloadChan))
+		})
+	})
+})

pkg/super_node/subscription.go

@@ -0,0 +1,27 @@
+// 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 super_node
+
+import (
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+)
+
+// Subscription holds the information for an individual client subscription to the super node
+type Subscription struct {
+	PayloadChan chan<- streamer.SuperNodePayload
+	QuitChan    chan<- bool
+}

pkg/super_node/test_helpers.go

@@ -0,0 +1,74 @@
+// 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 super_node
+
+import (
+	"bytes"
+
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+func SetupDB() (*postgres.DB, error) {
+	return postgres.NewDB(config.Database{
+		Hostname: "localhost",
+		Name:     "vulcanize_private",
+		Port:     5432,
+	}, core.Node{})
+}
+
+func TearDownDB(db *postgres.DB) {
+	tx, err := db.Beginx()
+	Expect(err).NotTo(HaveOccurred())
+
+	_, err = tx.Exec(`DELETE FROM header_cids`)
+	Expect(err).NotTo(HaveOccurred())
+	_, err = tx.Exec(`DELETE FROM transaction_cids`)
+	Expect(err).NotTo(HaveOccurred())
+	_, err = tx.Exec(`DELETE FROM receipt_cids`)
+	Expect(err).NotTo(HaveOccurred())
+	_, err = tx.Exec(`DELETE FROM state_cids`)
+	Expect(err).NotTo(HaveOccurred())
+	_, err = tx.Exec(`DELETE FROM storage_cids`)
+	Expect(err).NotTo(HaveOccurred())
+	_, err = tx.Exec(`DELETE FROM blocks`)
+	Expect(err).NotTo(HaveOccurred())
+
+	err = tx.Commit()
+	Expect(err).NotTo(HaveOccurred())
+}
+
+func ListContainsString(sss []string, s string) bool {
+	for _, str := range sss {
+		if s == str {
+			return true
+		}
+	}
+	return false
+}
+
+func ListContainsBytes(bbb [][]byte, b []byte) bool {
+	for _, by := range bbb {
+		if bytes.Equal(by, b) {
+			return true
+		}
+	}
+	return false
+}