Fix get logs unit test.

pkg/eth/api.go

@@ -40,7 +40,6 @@ import (
 	"github.com/ethereum/go-ethereum/statediff"
 	"github.com/sirupsen/logrus"
 
-	"github.com/vulcanize/ipld-eth-indexer/pkg/eth"
 	"github.com/vulcanize/ipld-eth-server/pkg/shared"
 )
 
@@ -550,9 +549,11 @@ func (pea *PublicEthAPI) localGetLogs(crit filters.FilterCriteria) ([]*types.Log
 	for i, addr := range crit.Addresses {
 		addrStrs[i] = addr.String()
 	}
-	topicStrSets := make([][]string, 4)
+
+	topicStrSets := make([][]string, len(crit.Topics))
 	for i, topicSet := range crit.Topics {
 		if i > 3 {
+			topicStrSets = topicStrSets[:4]
 			// don't allow more than 4 topics
 			break
 		}
@@ -598,8 +599,9 @@ func (pea *PublicEthAPI) localGetLogs(crit filters.FilterCriteria) ([]*types.Log
 		if err != nil {
 			return nil, err
 		}
-		return extractLogsOfInterest(pea.B.Config.ChainConfig, *crit.BlockHash, block.NumberU64(), block.Transactions(), rctIPLDs, filter.Topics)
+		return extractLogsOfInterest(pea.B.Config.ChainConfig, *crit.BlockHash, block.NumberU64(), block.Transactions(), rctIPLDs, filter)
 	}
+
 	// Otherwise, create block range from criteria
 	// nil values are filled in; to request a single block have both ToBlock and FromBlock equal that number
 	startingBlock := crit.FromBlock
@@ -607,6 +609,7 @@ func (pea *PublicEthAPI) localGetLogs(crit filters.FilterCriteria) ([]*types.Log
 	if startingBlock == nil {
 		startingBlock = common.Big0
 	}
+
 	if endingBlock == nil {
 		endingBlockInt, err := pea.B.Retriever.RetrieveLastBlockNumber()
 		if err != nil {
@@ -614,25 +617,38 @@ func (pea *PublicEthAPI) localGetLogs(crit filters.FilterCriteria) ([]*types.Log
 		}
 		endingBlock = big.NewInt(endingBlockInt)
 	}
+
 	start := startingBlock.Int64()
 	end := endingBlock.Int64()
-	allRctCIDs := make([]eth.ReceiptModel, 0)
+	var logs []*types.Log
 	for i := start; i <= end; i++ {
 		rctCIDs, err := pea.B.Retriever.RetrieveRctCIDs(tx, filter, i, nil, nil)
 		if err != nil {
 			return nil, err
 		}
-		allRctCIDs = append(allRctCIDs, rctCIDs...)
+
-	}
+		block, err := pea.B.BlockByNumber(context.Background(), rpc.BlockNumber(i))
-	rctIPLDs, err := pea.B.Fetcher.FetchRcts(tx, allRctCIDs)
 		if err != nil {
 			return nil, err
 		}
+
+		rctIPLDs, err := pea.B.Fetcher.FetchRcts(tx, rctCIDs)
+		if err != nil {
+			return nil, err
+		}
+
+		log, err := extractLogsOfInterest(pea.B.Config.ChainConfig, block.Hash(), uint64(i), block.Transactions(), rctIPLDs, filter)
+		if err != nil {
+			return nil, err
+		}
+
+		logs = append(logs, log...)
+	}
+
 	if err := tx.Commit(); err != nil {
 		return nil, err
 	}
-	// @TODO refactor this and pass actual block hash and block number
+
-	logs, err := extractLogsOfInterest(pea.B.Config.ChainConfig, common.Hash{}, 0, types.Transactions{}, rctIPLDs, filter.Topics)
 	return logs, err // need to return err variable so that we return the err = tx.Commit() assignment in the defer
 }
 

pkg/eth/backend.go

@@ -51,7 +51,6 @@ import (
 var (
 	errPendingBlockNumber  = errors.New("pending block number not supported")
 	errNegativeBlockNumber = errors.New("negative block number not supported")
-	errInvalidBlockNumber  = errors.New("invalid block number")
 )
 
 const (

pkg/eth/backend_utils.go

@@ -20,9 +20,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/ethereum/go-ethereum/params"
 	"math/big"
 
+	"github.com/ethereum/go-ethereum/params"
+
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/rpc"
 
@@ -245,8 +246,8 @@ func newRPCTransactionFromBlockIndex(b *types.Block, index uint64) *RPCTransacti
 }
 
 // extractLogsOfInterest returns logs from the receipt IPLD
-func extractLogsOfInterest(config *params.ChainConfig, blockHash common.Hash, blockNumber uint64, txs types.Transactions, rctIPLDs []ipfs.BlockModel, wantedTopics [][]string) ([]*types.Log, error) {
+func extractLogsOfInterest(config *params.ChainConfig, blockHash common.Hash, blockNumber uint64,
-	var logs []*types.Log
+	txs types.Transactions, rctIPLDs []ipfs.BlockModel, filter ReceiptFilter) ([]*types.Log, error) {
 	receipts := make(types.Receipts, len(rctIPLDs))
 
 	for i, rctBytes := range rctIPLDs {
@@ -262,17 +263,74 @@ func extractLogsOfInterest(config *params.ChainConfig, blockHash common.Hash, bl
 		return nil, err
 	}
 
+	var unfilteredLogs []*types.Log
 	for _, receipt := range receipts {
-		for _, log := range receipt.Logs {
+		unfilteredLogs = append(unfilteredLogs, receipt.Logs...)
-			if wanted := wantedLog(wantedTopics, log.Topics); wanted == true {
-				logs = append(logs, log)
 	}
+
+	adders := make([]common.Address, len(filter.LogAddresses))
+	for i, addr := range filter.LogAddresses {
+		adders[i] = common.HexToAddress(addr)
+	}
+
+	topics := make([][]common.Hash, len(filter.Topics))
+	for i, v := range filter.Topics {
+		topics[i] = make([]common.Hash, len(v))
+		for j, topic := range v {
+			topics[i][j] = common.HexToHash(topic)
 		}
 	}
 
+	logs := filterLogs(unfilteredLogs, nil, nil, adders, topics)
 	return logs, nil
 }
 
+func includes(addresses []common.Address, a common.Address) bool {
+	for _, addr := range addresses {
+		if addr == a {
+			return true
+		}
+	}
+
+	return false
+}
+
+// filterLogs creates a slice of logs matching the given criteria.
+func filterLogs(logs []*types.Log, fromBlock, toBlock *big.Int, addresses []common.Address, topics [][]common.Hash) []*types.Log {
+	var ret []*types.Log
+Logs:
+	for _, log := range logs {
+		if fromBlock != nil && fromBlock.Int64() >= 0 && fromBlock.Uint64() > log.BlockNumber {
+			continue
+		}
+		if toBlock != nil && toBlock.Int64() >= 0 && toBlock.Uint64() < log.BlockNumber {
+			continue
+		}
+
+		if len(addresses) > 0 && !includes(addresses, log.Address) {
+			continue
+		}
+		// If the to filtered topics is greater than the amount of topics in logs, skip.
+		if len(topics) > len(log.Topics) {
+			continue Logs
+		}
+		for i, sub := range topics {
+			match := len(sub) == 0 // empty rule set == wildcard
+			for _, topic := range sub {
+				if log.Topics[i] == topic {
+					match = true
+					break
+				}
+			}
+			if !match {
+				continue Logs
+			}
+		}
+		ret = append(ret, log)
+	}
+	return ret
+}
+
 // returns true if the log matches on the filter
 func wantedLog(wantedTopics [][]string, actualTopics []common.Hash) bool {
 	// actualTopics will always have length <= 4

pkg/eth/cid_retriever.go

@@ -310,6 +310,14 @@ func (ecr *CIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter ReceiptFilter, b
 		args = append(args, blockHash.String())
 		id++
 	}
+
+	// TODO: Add the below filters when we have log index in DB.
+	if true {
+		pgStr += ` ORDER BY transaction_cids.index`
+		receiptCids := make([]eth.ReceiptModel, 0)
+		return receiptCids, tx.Select(&receiptCids, pgStr, args...)
+	}
+
 	if len(rctFilter.LogAddresses) > 0 {
 		// Filter on log contract addresses if there are any
 		pgStr += fmt.Sprintf(` AND ((receipt_cids.log_contracts && $%d::VARCHAR(66)[]`, id)
@@ -371,6 +379,7 @@ func (ecr *CIDRetriever) RetrieveRctCIDs(tx *sqlx.Tx, rctFilter ReceiptFilter, b
 			args = append(args, pq.Array(trxIds))
 		}
 	}
+
 	pgStr += ` ORDER BY transaction_cids.index`
 	receiptCids := make([]eth.ReceiptModel, 0)
 	return receiptCids, tx.Select(&receiptCids, pgStr, args...)