review fixes

pkg/seed_node/api.go

@@ -0,0 +1,85 @@
+// 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 seed_node
+
+import (
+	"context"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+)
+
+// APIName is the namespace used for the state diffing service API
+const APIName = "vulcanizedb"
+
+// APIVersion is the version of the state diffing service API
+const APIVersion = "0.0.1"
+
+// PublicSeedNodeAPI is the public api for the seed node
+type PublicSeedNodeAPI struct {
+	snp Processor
+}
+
+// NewPublicSeedNodeAPI creates a new PublicSeedNodeAPI with the provided underlying SyncPublishScreenAndServe process
+func NewPublicSeedNodeAPI(seedNodeProcessor Processor) *PublicSeedNodeAPI {
+	return &PublicSeedNodeAPI{
+		snp: seedNodeProcessor,
+	}
+}
+
+// Stream is the public method to setup a subscription that fires off SyncPublishScreenAndServe payloads as they are created
+func (api *PublicSeedNodeAPI) 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.SeedNodePayload, payloadChanBufferSize)
+		quitChan := make(chan bool, 1)
+		go api.snp.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.snp.Unsubscribe(rpcSub.ID)
+					return
+				}
+			case <-rpcSub.Err():
+				api.snp.Unsubscribe(rpcSub.ID)
+				return
+			case <-quitChan:
+				// don't need to unsubscribe, SyncPublishScreenAndServe service does so before sending the quit signal
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}

pkg/seed_node/filterer.go

@@ -0,0 +1,244 @@
+// 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 seed_node
+
+import (
+	"bytes"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/rlp"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+)
+
+// 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.SeedNodePayload, error)
+}
+
+// Filterer is the underlying struct for the ReponseFilterer interface
+type Filterer struct{}
+
+// NewResponseFilterer creates a new Filterer satisfyign the ReponseFilterer 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.SeedNodePayload, error) {
+	response := new(streamer.SeedNodePayload)
+	err := s.filterHeaders(streamFilters, response, payload)
+	if err != nil {
+		return nil, err
+	}
+	txHashes, err := s.filterTransactions(streamFilters, response, payload)
+	if err != nil {
+		return nil, err
+	}
+	err = s.filerReceipts(streamFilters, response, payload, txHashes)
+	if err != nil {
+		return nil, err
+	}
+	err = s.filterState(streamFilters, response, payload)
+	if err != nil {
+		return nil, err
+	}
+	err = s.filterStorage(streamFilters, response, payload)
+	if err != nil {
+		return nil, err
+	}
+	response.BlockNumber = payload.BlockNumber
+	return response, nil
+}
+
+func (s *Filterer) filterHeaders(streamFilters config.Subscription, response *streamer.SeedNodePayload, 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.SeedNodePayload, 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.SeedNodePayload, 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 only keep receipts with logs of interest if we are interested in that contract
+		for _, wantedTopic := range wantedTopics {
+			for _, actualTopic := range actualTopics {
+				if wantedTopic == actualTopic {
+					for _, wantedContract := range wantedContracts {
+						if wantedContract == actualContract {
+							return true
+						}
+					}
+				}
+			}
+		}
+	}
+
+	return false
+}
+
+func (s *Filterer) filterState(streamFilters config.Subscription, response *streamer.SeedNodePayload, 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.SeedNodePayload, 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/seed_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 = cidPayload
+	return repo.ReturnErr
+}

pkg/seed_node/repository.go

@@ -0,0 +1,143 @@
+// 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 seed_node
+
+import (
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+// 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 {
+		tx.Rollback()
+		return err
+	}
+	for uncleHash, cid := range cidPayload.UncleCIDS {
+		err = repo.indexUncleCID(tx, cid, cidPayload.BlockNumber, uncleHash.Hex())
+		if err != nil {
+			tx.Rollback()
+			return err
+		}
+	}
+	err = repo.indexTransactionAndReceiptCIDs(tx, cidPayload, headerID)
+	if err != nil {
+		tx.Rollback()
+		return err
+	}
+	err = repo.indexStateAndStorageCIDs(tx, cidPayload, headerID)
+	if err != nil {
+		tx.Rollback()
+		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/seed_node/retreiver.go

@@ -0,0 +1,267 @@
+// 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 seed_node
+
+import (
+	"math/big"
+
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+
+	"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"
+)
+
+// 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)
+}
+
+// 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 {
+			tx.Rollback()
+			log.Error("header cid retrieval error")
+			return nil, err
+		}
+		if !streamFilters.HeaderFilter.FinalOnly {
+			cw.Uncles, err = ecr.retrieveUncleCIDs(tx, streamFilters, blockNumber)
+			if err != nil {
+				tx.Rollback()
+				log.Error("header 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 {
+			tx.Rollback()
+			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 {
+			tx.Rollback()
+			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 {
+			tx.Rollback()
+			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 {
+			tx.Rollback()
+			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, 2)
+	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)[])`
+	} else {
+		pgStr += `)`
+	}
+	if len(trxIds) > 0 {
+		pgStr += ` OR receipt_cids.tx_id = ANY($4::INTEGER[]))`
+		args = append(args, pq.Array(trxIds))
+	} else {
+		pgStr += `)`
+	}
+	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 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 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
+}

pkg/seed_node/seed_node_suite_test.go

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

pkg/seed_node/service.go

@@ -0,0 +1,414 @@
+// 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 seed_node
+
+import (
+	"sync"
+
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+
+	"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/rlp"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/statediff"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/core"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
+)
+
+const payloadChanBufferSize = 20000 // the max eth sub buffer size
+const workerPoolSize = 1
+
+// Processor 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 Processor 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.SeedNodePayload, quitChan chan<- bool, streamFilters config.Subscription)
+	// Method to unsubscribe from state diff processing
+	Unsubscribe(id rpc.ID)
+}
+
+// Service is the underlying struct for the SyncAndPublish interface
+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
+	Fetcher 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
+}
+
+// NewProcessor creates a new Processor interface using an underlying Service struct
+func NewProcessor(ipfsPath string, db *postgres.DB, ethClient core.EthClient, rpcClient core.RpcClient, qc chan bool) (Processor, error) {
+	publisher, err := ipfs.NewIPLDPublisher(ipfsPath)
+	if err != nil {
+		return nil, err
+	}
+	fetcher, err := ipfs.NewIPLDFetcher(ipfsPath)
+	if err != nil {
+		return nil, err
+	}
+	return &Service{
+		Streamer:          streamer.NewStateDiffStreamer(rpcClient),
+		Repository:        NewCIDRepository(db),
+		Converter:         ipfs.NewPayloadConverter(ethClient),
+		Publisher:         publisher,
+		Filterer:          NewResponseFilterer(),
+		Fetcher:           fetcher,
+		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),
+	}, 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 StateDiffingService offers
+func (sap *Service) APIs() []rpc.API {
+	return []rpc.API{
+		{
+			Namespace: APIName,
+			Version:   APIVersion,
+			Service:   NewPublicSeedNodeAPI(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, 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 < workerPoolSize; i++ {
+		sap.publishAndIndex(i, publishAndIndexPayload, publishAndIndexQuit)
+	}
+
+	go func() {
+		for {
+			select {
+			case payload := <-sap.PayloadChan:
+				if payload.Err != nil {
+					log.Error(err)
+					continue
+				}
+				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 < 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 worked %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 seed 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.SeedNodePayload, quitChan chan<- bool, streamFilters config.Subscription) {
+	log.Info("Subscribing to the seed 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.SeedNodePayload{
+			ErrMsg: "unable to set block range; error: " + err.Error(),
+		}
+	}
+	if startingBlock < con.StartingBlock.Int64() {
+		startingBlock = con.StartingBlock.Int64()
+	}
+	if con.EndingBlock.Int64() <= 0 || con.EndingBlock.Int64() <= startingBlock {
+		endingBlock, err = sap.Retriever.RetrieveLastBlockNumber()
+		if err != nil {
+			sub.PayloadChan <- streamer.SeedNodePayload{
+				ErrMsg: "unable to set block range; error: " + err.Error(),
+			}
+		}
+	}
+	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 := con.StartingBlock.Int64(); i <= endingBlock; i++ {
+			cidWrapper, err := sap.Retriever.RetrieveCIDs(con, i)
+			if err != nil {
+				sub.PayloadChan <- streamer.SeedNodePayload{
+					ErrMsg: "CID retrieval error: " + err.Error(),
+				}
+				continue
+			}
+			if ipfs.EmptyCIDWrapper(*cidWrapper) {
+				continue
+			}
+			blocksWrapper, err := sap.Fetcher.FetchCIDs(*cidWrapper)
+			if err != nil {
+				log.Error(err)
+				sub.PayloadChan <- streamer.SeedNodePayload{
+					ErrMsg: "IPLD fetching error: " + err.Error(),
+				}
+				continue
+			}
+			backFillIplds, err := sap.Resolver.ResolveIPLDs(*blocksWrapper)
+			if err != nil {
+				log.Error(err)
+				sub.PayloadChan <- streamer.SeedNodePayload{
+					ErrMsg: "IPLD resolving error: " + err.Error(),
+				}
+				continue
+			}
+			select {
+			case sub.PayloadChan <- *backFillIplds:
+				log.Infof("sending seed 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 seed 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 seed 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 seed node service")
+	close(sap.QuitChan)
+	return nil
+}
+
+// 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()
+}

pkg/seed_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 seed_node_test
+
+import (
+	"sync"
+	"time"
+
+	"github.com/vulcanize/vulcanizedb/pkg/seed_node"
+
+	"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"
+	mocks3 "github.com/vulcanize/vulcanizedb/pkg/seed_node/mocks"
+)
+
+var _ = Describe("Service", func() {
+
+	Describe("Loop", func() {
+		It("Streams StatediffPayloads, 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 := &seed_node.Service{
+				Repository:  mockCidRepo,
+				Publisher:   mockPublisher,
+				Streamer:    mockStreamer,
+				Converter:   mockConverter,
+				PayloadChan: payloadChan,
+				QuitChan:    quitChan,
+			}
+			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(mockCidRepo.PassedCIDPayload).To(Equal(&mocks.MockCIDPayload))
+			Expect(mockPublisher.PassedIPLDPayload).To(Equal(&mocks.MockIPLDPayload))
+			Expect(mockStreamer.PassedPayloadChan).To(Equal(payloadChan))
+		})
+	})
+})

pkg/seed_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 seed_node
+
+import (
+	"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
+)
+
+// Subscription holds the information for an individual client subscription to the seed node
+type Subscription struct {
+	PayloadChan chan<- streamer.SeedNodePayload
+	QuitChan    chan<- bool
+}