expand retriever, fetcher, and resolver interfaces in prep for eth api mimick

pkg/super_node/filterer.go

@@ -43,35 +43,35 @@ func NewResponseFilterer() *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)
-	headersErr := s.filterHeaders(streamFilters, response, payload)
-	if headersErr != nil {
-		return streamer.SuperNodePayload{}, headersErr
+	if checkRange(streamFilters.StartingBlock.Int64(), streamFilters.EndingBlock.Int64(), payload.BlockNumber.Int64()) {
+		response := new(streamer.SuperNodePayload)
+		if err := s.filterHeaders(streamFilters.HeaderFilter, response, payload); err != nil {
+			return streamer.SuperNodePayload{}, err
+		}
+		txHashes, err := s.filterTransactions(streamFilters.TrxFilter, response, payload)
+		if err != nil {
+			return streamer.SuperNodePayload{}, err
+		}
+		if err := s.filerReceipts(streamFilters.ReceiptFilter, response, payload, txHashes); err != nil {
+			return streamer.SuperNodePayload{}, err
+		}
+		if err := s.filterState(streamFilters.StateFilter, response, payload); err != nil {
+			return streamer.SuperNodePayload{}, err
+		}
+		if err := s.filterStorage(streamFilters.StorageFilter, response, payload); err != nil {
+			return streamer.SuperNodePayload{}, err
+		}
+		response.BlockNumber = payload.BlockNumber
+		return *response, nil
 	}
-	txHashes, trxsErr := s.filterTransactions(streamFilters, response, payload)
-	if trxsErr != nil {
-		return streamer.SuperNodePayload{}, trxsErr
-	}
-	rctsErr := s.filerReceipts(streamFilters, response, payload, txHashes)
-	if rctsErr != nil {
-		return streamer.SuperNodePayload{}, rctsErr
-	}
-	stateErr := s.filterState(streamFilters, response, payload)
-	if stateErr != nil {
-		return streamer.SuperNodePayload{}, stateErr
-	}
-	storageErr := s.filterStorage(streamFilters, response, payload)
-	if storageErr != nil {
-		return streamer.SuperNodePayload{}, storageErr
-	}
-	response.BlockNumber = payload.BlockNumber
-	return *response, nil
+	return streamer.SuperNodePayload{}, 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()) {
+func (s *Filterer) filterHeaders(headerFilter config.HeaderFilter, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) error {
+	if !headerFilter.Off {
 		response.HeadersRlp = append(response.HeadersRlp, payload.HeaderRLP)
-		if streamFilters.HeaderFilter.Uncles {
+		if headerFilter.Uncles {
+			response.UnclesRlp = make([][]byte, 0, len(payload.BlockBody.Uncles))
 			for _, uncle := range payload.BlockBody.Uncles {
 				uncleRlp, err := rlp.EncodeToBytes(uncle)
 				if err != nil {
@@ -91,11 +91,11 @@ func checkRange(start, end, actual int64) bool {
 	return false
 }
 
-func (s *Filterer) filterTransactions(streamFilters config.Subscription, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) ([]common.Hash, error) {
+func (s *Filterer) filterTransactions(trxFilter config.TrxFilter, 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()) {
+	if !trxFilter.Off {
 		for i, trx := range payload.BlockBody.Transactions {
-			if checkTransactions(streamFilters.TrxFilter.Src, streamFilters.TrxFilter.Dst, payload.TrxMetaData[i].Src, payload.TrxMetaData[i].Dst) {
+			if checkTransactions(trxFilter.Src, trxFilter.Dst, payload.TrxMetaData[i].Src, payload.TrxMetaData[i].Dst) {
 				trxBuffer := new(bytes.Buffer)
 				err := trx.EncodeRLP(trxBuffer)
 				if err != nil {
@@ -127,10 +127,10 @@ func checkTransactions(wantedSrc, wantedDst []string, actualSrc, actualDst strin
 	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()) {
+func (s *Filterer) filerReceipts(receiptFilter config.ReceiptFilter, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload, trxHashes []common.Hash) error {
+	if !receiptFilter.Off {
 		for i, receipt := range payload.Receipts {
-			if checkReceipts(receipt, streamFilters.ReceiptFilter.Topic0s, payload.ReceiptMetaData[i].Topic0s, streamFilters.ReceiptFilter.Contracts, payload.ReceiptMetaData[i].ContractAddress, trxHashes) {
+			if checkReceipts(receipt, receiptFilter.Topic0s, payload.ReceiptMetaData[i].Topic0s, receiptFilter.Contracts, payload.ReceiptMetaData[i].ContractAddress, trxHashes, receiptFilter.MatchTxs) {
 				receiptForStorage := (*types.ReceiptForStorage)(receipt)
 				receiptBuffer := new(bytes.Buffer)
 				err := receiptForStorage.EncodeRLP(receiptBuffer)
@@ -144,15 +144,17 @@ func (s *Filterer) filerReceipts(streamFilters config.Subscription, response *st
 	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 {
+func checkReceipts(rct *types.Receipt, wantedTopics, actualTopics, wantedContracts []string, actualContract string, wantedTrxHashes []common.Hash, matchTxs bool) bool {
+	// If we aren't filtering for any topics, contracts, or corresponding trxs then all receipts are a go
+	if len(wantedTopics) == 0 && len(wantedContracts) == 0 && (len(wantedTrxHashes) == 0 || !matchTxs) {
 		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
+	// No matter what filters we have, we keep receipts for specific trxs we are interested in
+	if matchTxs {
+		for _, wantedTrxHash := range wantedTrxHashes {
+			if bytes.Equal(wantedTrxHash.Bytes(), rct.TxHash.Bytes()) {
+				return true
+			}
 		}
 	}
 
@@ -165,7 +167,7 @@ func checkReceipts(rct *types.Receipt, wantedTopics, actualTopics, wantedContrac
 				}
 			}
 		}
-	} else { // We keep receipts that belong to one of the specified contracts and have logs with topics if we aren't filtering on topics
+	} else { // We keep all receipts that belong to one of the specified contracts if we aren't filtering on topics
 		for _, wantedContract := range wantedContracts {
 			if wantedContract == actualContract {
 				if len(wantedTopics) == 0 {
@@ -186,17 +188,17 @@ func checkReceipts(rct *types.Receipt, wantedTopics, actualTopics, wantedContrac
 	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()) {
+func (s *Filterer) filterState(stateFilter config.StateFilter, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) error {
+	if !stateFilter.Off {
 		response.StateNodesRlp = make(map[common.Hash][]byte)
-		keyFilters := make([]common.Hash, 0, len(streamFilters.StateFilter.Addresses))
-		for _, addr := range streamFilters.StateFilter.Addresses {
+		keyFilters := make([]common.Hash, 0, len(stateFilter.Addresses))
+		for _, addr := range 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 {
+				if stateNode.Leaf || stateFilter.IntermediateNodes {
 					response.StateNodesRlp[key] = stateNode.Value
 				}
 			}
@@ -218,16 +220,16 @@ func checkNodeKeys(wantedKeys []common.Hash, actualKey common.Hash) bool {
 	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()) {
+func (s *Filterer) filterStorage(storageFilter config.StorageFilter, response *streamer.SuperNodePayload, payload ipfs.IPLDPayload) error {
+	if !storageFilter.Off {
 		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 {
+		stateKeyFilters := make([]common.Hash, 0, len(storageFilter.Addresses))
+		for _, addr := range 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 {
+		storageKeyFilters := make([]common.Hash, 0, len(storageFilter.StorageKeys))
+		for _, store := range storageFilter.StorageKeys {
 			keyFilter := ipfs.HexToKey(store)
 			storageKeyFilters = append(storageKeyFilters, keyFilter)
 		}

pkg/super_node/filterer_test.go

@@ -46,7 +46,8 @@ var _ = Describe("Filterer", func() {
 			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))
+			var unclesRlp [][]byte
+			Expect(superNodePayload.UnclesRlp).To(Equal(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())

pkg/super_node/mocks/retriever.go

@@ -1,7 +1,9 @@
 package mocks
 
 import (
+	"github.com/jmoiron/sqlx"
 	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
 )
 
@@ -19,6 +21,36 @@ func (*MockCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, blockNu
 	panic("implement me")
 }
 
+// RetrieveHeaderCIDs mock method
+func (*MockCIDRetriever) RetrieveHeaderCIDs(tx *sqlx.Tx, blockNumber int64) ([]string, error) {
+	panic("implement me")
+
+}
+
+// RetrieveUncleCIDs mock method
+func (*MockCIDRetriever) RetrieveUncleCIDs(tx *sqlx.Tx, blockNumber int64) ([]string, error) {
+	panic("implement me")
+
+}
+
+// RetrieveTrxCIDs mock method
+func (*MockCIDRetriever) RetrieveTrxCIDs(tx *sqlx.Tx, txFilter config.TrxFilter, blockNumber int64) ([]string, []int64, error) {
+	panic("implement me")
+
+}
+
+// RetrieveRctCIDs mock method
+func (*MockCIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter config.ReceiptFilter, blockNumber int64, trxIds []int64) ([]string, error) {
+	panic("implement me")
+
+}
+
+// RetrieveStateCIDs mock method
+func (*MockCIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter config.StateFilter, blockNumber int64) ([]ipfs.StateNodeCID, error) {
+	panic("implement me")
+
+}
+
 // RetrieveLastBlockNumber mock method
 func (*MockCIDRetriever) RetrieveLastBlockNumber() (int64, error) {
 	panic("implement me")
@@ -42,3 +74,7 @@ func (mcr *MockCIDRetriever) SetGapsToRetrieve(gaps [][2]uint64) {
 	}
 	mcr.GapsToRetrieve = append(mcr.GapsToRetrieve, gaps...)
 }
+
+func (mcr *MockCIDRetriever) Database() *postgres.DB {
+	panic("implement me")
+}

pkg/super_node/retriever.go

@@ -34,6 +34,12 @@ type CIDRetriever interface {
 	RetrieveLastBlockNumber() (int64, error)
 	RetrieveFirstBlockNumber() (int64, error)
 	RetrieveGapsInData() ([][2]uint64, error)
+	RetrieveHeaderCIDs(tx *sqlx.Tx, blockNumber int64) ([]string, error)
+	RetrieveUncleCIDs(tx *sqlx.Tx, blockNumber int64) ([]string, error)
+	RetrieveTrxCIDs(tx *sqlx.Tx, txFilter config.TrxFilter, blockNumber int64) ([]string, []int64, error)
+	RetrieveRctCIDs(tx *sqlx.Tx, rctFilter config.ReceiptFilter, blockNumber int64, trxIds []int64) ([]string, error)
+	RetrieveStateCIDs(tx *sqlx.Tx, stateFilter config.StateFilter, blockNumber int64) ([]ipfs.StateNodeCID, error)
+	Database() *postgres.DB
 }
 
 // EthCIDRetriever is the underlying struct supporting the CIDRetriever interface
@@ -65,37 +71,32 @@ func (ecr *EthCIDRetriever) RetrieveLastBlockNumber() (int64, error) {
 // 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")
-	tx, beginErr := ecr.db.Beginx()
-	if beginErr != nil {
-		return nil, beginErr
+	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 {
-		var headersErr error
-		cw.Headers, headersErr = ecr.retrieveHeaderCIDs(tx, streamFilters, blockNumber)
-		if headersErr != nil {
-			rollbackErr := tx.Rollback()
-			if rollbackErr != nil {
-				log.Error(rollbackErr)
+		cw.Headers, err = ecr.RetrieveHeaderCIDs(tx, blockNumber)
+		if err != nil {
+			if err := tx.Rollback(); err != nil {
+				log.Error(err)
 			}
 			log.Error("header cid retrieval error")
-			return nil, headersErr
+			return nil, err
 		}
 		if streamFilters.HeaderFilter.Uncles {
-			var unclesErr error
-			cw.Uncles, unclesErr = ecr.retrieveUncleCIDs(tx, streamFilters, blockNumber)
-			if unclesErr != nil {
-				rollbackErr := tx.Rollback()
-				if rollbackErr != nil {
-					log.Error(rollbackErr)
+			cw.Uncles, err = ecr.RetrieveUncleCIDs(tx, blockNumber)
+			if err != nil {
+				if err := tx.Rollback(); err != nil {
+					log.Error(err)
 				}
 				log.Error("uncle cid retrieval error")
-				return nil, unclesErr
+				return nil, err
 			}
 		}
 	}
@@ -103,64 +104,58 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, bloc
 	// Retrieve cached trx CIDs
 	var trxIds []int64
 	if !streamFilters.TrxFilter.Off {
-		var trxsErr error
-		cw.Transactions, trxIds, trxsErr = ecr.retrieveTrxCIDs(tx, streamFilters, blockNumber)
-		if trxsErr != nil {
-			rollbackErr := tx.Rollback()
-			if rollbackErr != nil {
-				log.Error(rollbackErr)
+		cw.Transactions, trxIds, err = ecr.RetrieveTrxCIDs(tx, streamFilters.TrxFilter, blockNumber)
+		if err != nil {
+			err := tx.Rollback()
+			if err != nil {
+				log.Error(err)
 			}
 			log.Error("transaction cid retrieval error")
-			return nil, trxsErr
+			return nil, err
 		}
 	}
 
 	// Retrieve cached receipt CIDs
 	if !streamFilters.ReceiptFilter.Off {
-		var rctsErr error
-		cw.Receipts, rctsErr = ecr.retrieveRctCIDs(tx, streamFilters, blockNumber, trxIds)
-		if rctsErr != nil {
-			rollbackErr := tx.Rollback()
-			if rollbackErr != nil {
-				log.Error(rollbackErr)
+		cw.Receipts, err = ecr.RetrieveRctCIDs(tx, streamFilters.ReceiptFilter, blockNumber, trxIds)
+		if err != nil {
+			if err := tx.Rollback(); err != nil {
+				log.Error(err)
 			}
 			log.Error("receipt cid retrieval error")
-			return nil, rctsErr
+			return nil, err
 		}
 	}
 
 	// Retrieve cached state CIDs
 	if !streamFilters.StateFilter.Off {
-		var stateErr error
-		cw.StateNodes, stateErr = ecr.retrieveStateCIDs(tx, streamFilters, blockNumber)
-		if stateErr != nil {
-			rollbackErr := tx.Rollback()
-			if rollbackErr != nil {
-				log.Error(rollbackErr)
+		cw.StateNodes, err = ecr.RetrieveStateCIDs(tx, streamFilters.StateFilter, blockNumber)
+		if err != nil {
+			if err := tx.Rollback(); err != nil {
+				log.Error(err)
 			}
 			log.Error("state cid retrieval error")
-			return nil, stateErr
+			return nil, err
 		}
 	}
 
 	// Retrieve cached storage CIDs
 	if !streamFilters.StorageFilter.Off {
-		var storageErr error
-		cw.StorageNodes, storageErr = ecr.retrieveStorageCIDs(tx, streamFilters, blockNumber)
-		if storageErr != nil {
-			rollbackErr := tx.Rollback()
-			if rollbackErr != nil {
-				log.Error(rollbackErr)
+		cw.StorageNodes, err = ecr.RetrieveStorageCIDs(tx, streamFilters.StorageFilter, blockNumber)
+		if err != nil {
+			if err := tx.Rollback(); err != nil {
+				log.Error(err)
 			}
 			log.Error("storage cid retrieval error")
-			return nil, storageErr
+			return nil, err
 		}
 	}
 
 	return cw, tx.Commit()
 }
 
-func (ecr *EthCIDRetriever) retrieveHeaderCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, error) {
+// RetrieveHeaderCIDs retrieves and returns all of the header cids at the provided blockheight
+func (ecr *EthCIDRetriever) RetrieveHeaderCIDs(tx *sqlx.Tx, blockNumber int64) ([]string, error) {
 	log.Debug("retrieving header cids for block ", blockNumber)
 	headers := make([]string, 0)
 	pgStr := `SELECT cid FROM header_cids
@@ -169,7 +164,8 @@ func (ecr *EthCIDRetriever) retrieveHeaderCIDs(tx *sqlx.Tx, streamFilters config
 	return headers, err
 }
 
-func (ecr *EthCIDRetriever) retrieveUncleCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, error) {
+// RetrieveUncleCIDs retrieves and returns all of the uncle cids at the provided blockheight
+func (ecr *EthCIDRetriever) RetrieveUncleCIDs(tx *sqlx.Tx, blockNumber int64) ([]string, error) {
 	log.Debug("retrieving header cids for block ", blockNumber)
 	headers := make([]string, 0)
 	pgStr := `SELECT cid FROM header_cids
@@ -178,7 +174,9 @@ func (ecr *EthCIDRetriever) retrieveUncleCIDs(tx *sqlx.Tx, streamFilters config.
 	return headers, err
 }
 
-func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, []int64, error) {
+// RetrieveTrxCIDs retrieves and returns all of the trx cids at the provided blockheight that conform to the provided filter parameters
+// also returns the ids for the returned transaction cids
+func (ecr *EthCIDRetriever) RetrieveTrxCIDs(tx *sqlx.Tx, txFilter config.TrxFilter, blockNumber int64) ([]string, []int64, error) {
 	log.Debug("retrieving transaction cids for block ", blockNumber)
 	args := make([]interface{}, 0, 3)
 	type result struct {
@@ -189,13 +187,13 @@ func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Su
 	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 {
+	if len(txFilter.Dst) > 0 {
 		pgStr += ` AND transaction_cids.dst = ANY($2::VARCHAR(66)[])`
-		args = append(args, pq.Array(streamFilters.TrxFilter.Dst))
+		args = append(args, pq.Array(txFilter.Dst))
 	}
-	if len(streamFilters.TrxFilter.Src) > 0 {
+	if len(txFilter.Src) > 0 {
 		pgStr += ` AND transaction_cids.src = ANY($3::VARCHAR(66)[])`
-		args = append(args, pq.Array(streamFilters.TrxFilter.Src))
+		args = append(args, pq.Array(txFilter.Src))
 	}
 	err := tx.Select(&results, pgStr, args...)
 	if err != nil {
@@ -210,7 +208,9 @@ func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Su
 	return cids, ids, nil
 }
 
-func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64, trxIds []int64) ([]string, error) {
+// RetrieveRctCIDs retrieves and returns all of the rct cids at the provided blockheight that conform to the provided
+// filter parameters and correspond to the provided tx ids
+func (ecr *EthCIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter config.ReceiptFilter, 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
@@ -218,13 +218,13 @@ func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Su
 			AND transaction_cids.header_id = header_cids.id
 			AND header_cids.block_number = $1`
 	args = append(args, blockNumber)
-	if len(streamFilters.ReceiptFilter.Topic0s) > 0 {
+	if len(rctFilter.Topic0s) > 0 {
 		pgStr += ` AND ((receipt_cids.topic0s && $2::VARCHAR(66)[]`
-		args = append(args, pq.Array(streamFilters.ReceiptFilter.Topic0s))
-		if len(streamFilters.ReceiptFilter.Contracts) > 0 {
+		args = append(args, pq.Array(rctFilter.Topic0s))
+		if len(rctFilter.Contracts) > 0 {
 			pgStr += ` AND receipt_cids.contract = ANY($3::VARCHAR(66)[]))`
-			args = append(args, pq.Array(streamFilters.ReceiptFilter.Contracts))
-			if len(trxIds) > 0 {
+			args = append(args, pq.Array(rctFilter.Contracts))
+			if rctFilter.MatchTxs && len(trxIds) > 0 {
 				pgStr += ` OR receipt_cids.tx_id = ANY($4::INTEGER[]))`
 				args = append(args, pq.Array(trxIds))
 			} else {
@@ -232,7 +232,7 @@ func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Su
 			}
 		} else {
 			pgStr += `)`
-			if len(trxIds) > 0 {
+			if rctFilter.MatchTxs && len(trxIds) > 0 {
 				pgStr += ` OR receipt_cids.tx_id = ANY($3::INTEGER[]))`
 				args = append(args, pq.Array(trxIds))
 			} else {
@@ -240,16 +240,16 @@ func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Su
 			}
 		}
 	} else {
-		if len(streamFilters.ReceiptFilter.Contracts) > 0 {
+		if len(rctFilter.Contracts) > 0 {
 			pgStr += ` AND (receipt_cids.contract = ANY($2::VARCHAR(66)[])`
-			args = append(args, pq.Array(streamFilters.ReceiptFilter.Contracts))
-			if len(trxIds) > 0 {
+			args = append(args, pq.Array(rctFilter.Contracts))
+			if rctFilter.MatchTxs && 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 {
+		} else if rctFilter.MatchTxs && len(trxIds) > 0 {
 			pgStr += ` AND receipt_cids.tx_id = ANY($2::INTEGER[])`
 			args = append(args, pq.Array(trxIds))
 		}
@@ -259,22 +259,23 @@ func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Su
 	return receiptCids, err
 }
 
-func (ecr *EthCIDRetriever) retrieveStateCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]ipfs.StateNodeCID, error) {
+// RetrieveStateCIDs retrieves and returns all of the state node cids at the provided blockheight that conform to the provided filter parameters
+func (ecr *EthCIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter config.StateFilter, 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)
+	addrLen := len(stateFilter.Addresses)
 	if addrLen > 0 {
 		keys := make([]string, 0, addrLen)
-		for _, addr := range streamFilters.StateFilter.Addresses {
+		for _, addr := range 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 {
+	if !stateFilter.IntermediateNodes {
 		pgStr += ` AND state_cids.leaf = TRUE`
 	}
 	stateNodeCIDs := make([]ipfs.StateNodeCID, 0)
@@ -282,7 +283,8 @@ func (ecr *EthCIDRetriever) retrieveStateCIDs(tx *sqlx.Tx, streamFilters config.
 	return stateNodeCIDs, err
 }
 
-func (ecr *EthCIDRetriever) retrieveStorageCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]ipfs.StorageNodeCID, error) {
+// RetrieveStorageCIDs retrieves and returns all of the storage node cids at the provided blockheight that conform to the provided filter parameters
+func (ecr *EthCIDRetriever) RetrieveStorageCIDs(tx *sqlx.Tx, storageFilter config.StorageFilter, 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
@@ -290,20 +292,23 @@ func (ecr *EthCIDRetriever) retrieveStorageCIDs(tx *sqlx.Tx, streamFilters confi
 			AND state_cids.header_id = header_cids.id
 			AND header_cids.block_number = $1`
 	args = append(args, blockNumber)
-	addrLen := len(streamFilters.StorageFilter.Addresses)
+	addrLen := len(storageFilter.Addresses)
 	if addrLen > 0 {
 		keys := make([]string, 0, addrLen)
-		for _, addr := range streamFilters.StorageFilter.Addresses {
+		for _, addr := range 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(storageFilter.StorageKeys) > 0 {
+			pgStr += ` AND storage_cids.storage_key = ANY($3::VARCHAR(66)[])`
+			args = append(args, pq.Array(storageFilter.StorageKeys))
+		}
+	} else if len(storageFilter.StorageKeys) > 0 {
+		pgStr += ` AND storage_cids.storage_key = ANY($2::VARCHAR(66)[])`
+		args = append(args, pq.Array(storageFilter.StorageKeys))
 	}
-	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 {
+	if !storageFilter.IntermediateNodes {
 		pgStr += ` AND storage_cids.leaf = TRUE`
 	}
 	storageNodeCIDs := make([]ipfs.StorageNodeCID, 0)
@@ -334,3 +339,7 @@ func (ecr *EthCIDRetriever) RetrieveGapsInData() ([][2]uint64, error) {
 	}
 	return gapRanges, nil
 }
+
+func (ecr *EthCIDRetriever) Database() *postgres.DB {
+	return ecr.db
+}

pkg/super_node/retriever_test.go

@@ -126,6 +126,7 @@ var (
 		},
 		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{
+			MatchTxs:  true,
 			Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000006"}, // Topic isn't one of the topics we have
 			Contracts: []string{"0x0000000000000000000000000000000000000002"},                         // Contract isn't one of the contracts we have
 		},
@@ -146,6 +147,7 @@ var (
 			Dst: []string{"0x0000000000000000000000000000000000000001"}, // We only filter for one of the trxs so we will only get the one corresponding receipt
 		},
 		ReceiptFilter: config.ReceiptFilter{
+			MatchTxs:  true,
 			Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000006"}, // Topic isn't one of the topics we have
 			Contracts: []string{"0x0000000000000000000000000000000000000002"},                         // Contract isn't one of the contracts we have
 		},
@@ -195,8 +197,8 @@ var _ = Describe("Retriever", func() {
 
 	Describe("RetrieveCIDs", func() {
 		BeforeEach(func() {
-			indexErr := repo.Index(mocks.MockCIDPayload)
-			Expect(indexErr).ToNot(HaveOccurred())
+			err := repo.Index(mocks.MockCIDPayload)
+			Expect(err).ToNot(HaveOccurred())
 		})
 		It("Retrieves all CIDs for the given blocknumber when provided an open filter", func() {
 			cidWrapper, err := retriever.RetrieveCIDs(openFilter, 1)
@@ -308,20 +310,20 @@ var _ = Describe("Retriever", func() {
 
 	Describe("RetrieveFirstBlockNumber", func() {
 		It("Gets the number of the first block that has data in the database", func() {
-			indexErr := repo.Index(mocks.MockCIDPayload)
-			Expect(indexErr).ToNot(HaveOccurred())
-			num, retrieveErr := retriever.RetrieveFirstBlockNumber()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(mocks.MockCIDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			num, err := retriever.RetrieveFirstBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(num).To(Equal(int64(1)))
 		})
 
 		It("Gets the number of the first block that has data in the database", func() {
 			payload := *mocks.MockCIDPayload
 			payload.BlockNumber = "1010101"
-			indexErr := repo.Index(&payload)
-			Expect(indexErr).ToNot(HaveOccurred())
-			num, retrieveErr := retriever.RetrieveFirstBlockNumber()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload)
+			Expect(err).ToNot(HaveOccurred())
+			num, err := retriever.RetrieveFirstBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(num).To(Equal(int64(1010101)))
 		})
 
@@ -330,32 +332,32 @@ var _ = Describe("Retriever", func() {
 			payload1.BlockNumber = "1010101"
 			payload2 := payload1
 			payload2.BlockNumber = "5"
-			indexErr := repo.Index(&payload1)
-			Expect(indexErr).ToNot(HaveOccurred())
-			indexErr2 := repo.Index(&payload2)
-			Expect(indexErr2).ToNot(HaveOccurred())
-			num, retrieveErr := retriever.RetrieveFirstBlockNumber()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload1)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload2)
+			Expect(err).ToNot(HaveOccurred())
+			num, err := retriever.RetrieveFirstBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(num).To(Equal(int64(5)))
 		})
 	})
 
 	Describe("RetrieveLastBlockNumber", func() {
 		It("Gets the number of the latest block that has data in the database", func() {
-			indexErr := repo.Index(mocks.MockCIDPayload)
-			Expect(indexErr).ToNot(HaveOccurred())
-			num, retrieveErr := retriever.RetrieveLastBlockNumber()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(mocks.MockCIDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			num, err := retriever.RetrieveLastBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(num).To(Equal(int64(1)))
 		})
 
 		It("Gets the number of the latest block that has data in the database", func() {
 			payload := *mocks.MockCIDPayload
 			payload.BlockNumber = "1010101"
-			indexErr := repo.Index(&payload)
-			Expect(indexErr).ToNot(HaveOccurred())
-			num, retrieveErr := retriever.RetrieveLastBlockNumber()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload)
+			Expect(err).ToNot(HaveOccurred())
+			num, err := retriever.RetrieveLastBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(num).To(Equal(int64(1010101)))
 		})
 
@@ -364,12 +366,12 @@ var _ = Describe("Retriever", func() {
 			payload1.BlockNumber = "1010101"
 			payload2 := payload1
 			payload2.BlockNumber = "5"
-			indexErr := repo.Index(&payload1)
-			Expect(indexErr).ToNot(HaveOccurred())
-			indexErr2 := repo.Index(&payload2)
-			Expect(indexErr2).ToNot(HaveOccurred())
-			num, retrieveErr := retriever.RetrieveLastBlockNumber()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload1)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload2)
+			Expect(err).ToNot(HaveOccurred())
+			num, err := retriever.RetrieveLastBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(num).To(Equal(int64(1010101)))
 		})
 	})
@@ -380,24 +382,24 @@ var _ = Describe("Retriever", func() {
 			payload1.BlockNumber = "2"
 			payload2 := payload1
 			payload2.BlockNumber = "3"
-			indexErr1 := repo.Index(mocks.MockCIDPayload)
-			Expect(indexErr1).ToNot(HaveOccurred())
-			indexErr2 := repo.Index(&payload1)
-			Expect(indexErr2).ToNot(HaveOccurred())
-			indexErr3 := repo.Index(&payload2)
-			Expect(indexErr3).ToNot(HaveOccurred())
-			gaps, retrieveErr := retriever.RetrieveGapsInData()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(mocks.MockCIDPayload)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload1)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload2)
+			Expect(err).ToNot(HaveOccurred())
+			gaps, err := retriever.RetrieveGapsInData()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(len(gaps)).To(Equal(0))
 		})
 
 		It("Doesn't return the gap from 0 to the earliest block", func() {
 			payload := *mocks.MockCIDPayload
 			payload.BlockNumber = "5"
-			indexErr := repo.Index(&payload)
-			Expect(indexErr).ToNot(HaveOccurred())
-			gaps, retrieveErr := retriever.RetrieveGapsInData()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload)
+			Expect(err).ToNot(HaveOccurred())
+			gaps, err := retriever.RetrieveGapsInData()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(len(gaps)).To(Equal(0))
 		})
 
@@ -406,12 +408,12 @@ var _ = Describe("Retriever", func() {
 			payload1.BlockNumber = "1010101"
 			payload2 := payload1
 			payload2.BlockNumber = "5"
-			indexErr := repo.Index(&payload1)
-			Expect(indexErr).ToNot(HaveOccurred())
-			indexErr2 := repo.Index(&payload2)
-			Expect(indexErr2).ToNot(HaveOccurred())
-			gaps, retrieveErr := retriever.RetrieveGapsInData()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload1)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload2)
+			Expect(err).ToNot(HaveOccurred())
+			gaps, err := retriever.RetrieveGapsInData()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(len(gaps)).To(Equal(1))
 			Expect(gaps[0][0]).To(Equal(uint64(6)))
 			Expect(gaps[0][1]).To(Equal(uint64(1010100)))
@@ -430,20 +432,20 @@ var _ = Describe("Retriever", func() {
 			payload5.BlockNumber = "102"
 			payload6 := payload5
 			payload6.BlockNumber = "1000"
-			indexErr := repo.Index(&payload1)
-			Expect(indexErr).ToNot(HaveOccurred())
-			indexErr2 := repo.Index(&payload2)
-			Expect(indexErr2).ToNot(HaveOccurred())
-			indexErr3 := repo.Index(&payload3)
-			Expect(indexErr3).ToNot(HaveOccurred())
-			indexErr4 := repo.Index(&payload4)
-			Expect(indexErr4).ToNot(HaveOccurred())
-			indexErr5 := repo.Index(&payload5)
-			Expect(indexErr5).ToNot(HaveOccurred())
-			indexErr6 := repo.Index(&payload6)
-			Expect(indexErr6).ToNot(HaveOccurred())
-			gaps, retrieveErr := retriever.RetrieveGapsInData()
-			Expect(retrieveErr).ToNot(HaveOccurred())
+			err := repo.Index(&payload1)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload2)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload3)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload4)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload5)
+			Expect(err).ToNot(HaveOccurred())
+			err = repo.Index(&payload6)
+			Expect(err).ToNot(HaveOccurred())
+			gaps, err := retriever.RetrieveGapsInData()
+			Expect(err).ToNot(HaveOccurred())
 			Expect(len(gaps)).To(Equal(3))
 			Expect(super_node.ListContainsRange(gaps, [2]uint64{6, 99})).To(BeTrue())
 			Expect(super_node.ListContainsRange(gaps, [2]uint64{103, 999})).To(BeTrue())