[EVM] ETH Log Filter Endpoints (#1116)

* [EVM] add subscription utility

* finished new filter

* finished tests for new filter

* finished uninstall filter

* query tendermint for events

* in progress

* got first couple of tests done

* finished tests for get logs

* get filter logs and get filter changes

* refactor

* get filter logs accepts multiple addresses

* use multiple addresses and use cursor map

* remove bad test

* remove prints and todos

* minor fix

* fix lint errors and failing test

* poc to pass filtercriteria into JSON RPC

* made topics 2d array and adjusted regex

* GetLogs tests work

* Finished other functions

* remove unncessary inputs from NewFilterAPI

* add expiration/timeouts for filters

* removed error from UninstallFilter

* minor fix

* appease linter

---------

Co-authored-by: codchen <codchen03@gmail.com>

evmrpc/filter.go

@@ -0,0 +1,268 @@
+package evmrpc
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	ethtypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/eth/filters"
+	abci "github.com/tendermint/tendermint/abci/types"
+	rpcclient "github.com/tendermint/tendermint/rpc/client"
+	"github.com/tendermint/tendermint/rpc/coretypes"
+)
+
+type filter struct {
+	fc       filters.FilterCriteria
+	deadline *time.Timer
+	cursors  map[common.Address]string
+}
+
+type FilterAPI struct {
+	tmClient     rpcclient.Client
+	nextFilterID uint64
+	filtersMu    sync.Mutex
+	filters      map[uint64]filter
+	filterConfig *FilterConfig
+}
+
+type FilterConfig struct {
+	timeout time.Duration
+}
+
+func NewFilterAPI(tmClient rpcclient.Client, filterConfig *FilterConfig) *FilterAPI {
+	filters := make(map[uint64]filter)
+	api := &FilterAPI{
+		tmClient:     tmClient,
+		nextFilterID: 1,
+		filtersMu:    sync.Mutex{},
+		filters:      filters,
+		filterConfig: filterConfig,
+	}
+
+	go api.timeoutLoop(filterConfig.timeout)
+
+	return api
+}
+
+func (a *FilterAPI) timeoutLoop(timeout time.Duration) {
+	ticker := time.NewTicker(timeout)
+	defer ticker.Stop()
+	for {
+		<-ticker.C
+		a.filtersMu.Lock()
+		for id, filter := range a.filters {
+			select {
+			case <-filter.deadline.C:
+				delete(a.filters, id)
+			default:
+				continue
+			}
+		}
+		a.filtersMu.Unlock()
+	}
+}
+
+func (a *FilterAPI) NewFilter(
+	_ context.Context,
+	crit filters.FilterCriteria,
+) (*uint64, error) {
+	a.filtersMu.Lock()
+	defer a.filtersMu.Unlock()
+	curFilterID := a.nextFilterID
+	a.nextFilterID++
+	a.filters[curFilterID] = filter{
+		fc:       crit,
+		deadline: time.NewTimer(a.filterConfig.timeout),
+		cursors:  make(map[common.Address]string),
+	}
+	return &curFilterID, nil
+}
+
+func (a *FilterAPI) GetFilterChanges(
+	ctx context.Context,
+	filterID uint64,
+) ([]*ethtypes.Log, error) {
+	a.filtersMu.Lock()
+	filter, ok := a.filters[filterID]
+	a.filtersMu.Unlock()
+	if !ok {
+		return nil, errors.New("filter does not exist")
+	}
+
+	if !filter.deadline.Stop() {
+		// timer expired but filter is not yet removed in timeout loop
+		// receive timer value and reset timer
+		<-filter.deadline.C
+	}
+	filter.deadline.Reset(a.filterConfig.timeout)
+
+	res, cursors, err := a.getLogsOverAddresses(ctx, filter.fc, filter.cursors)
+	if err != nil {
+		return nil, err
+	}
+	a.filtersMu.Lock()
+	updatedFilter := a.filters[filterID]
+	updatedFilter.cursors = cursors
+	a.filters[filterID] = updatedFilter
+	a.filtersMu.Unlock()
+	return res, nil
+}
+
+func (a *FilterAPI) GetFilterLogs(
+	ctx context.Context,
+	filterID uint64,
+) ([]*ethtypes.Log, error) {
+	a.filtersMu.Lock()
+	filter, ok := a.filters[filterID]
+	a.filtersMu.Unlock()
+	if !ok {
+		return nil, errors.New("filter does not exist")
+	}
+
+	if !filter.deadline.Stop() {
+		// timer expired but filter is not yet removed in timeout loop
+		// receive timer value and reset timer
+		<-filter.deadline.C
+	}
+	filter.deadline.Reset(a.filterConfig.timeout)
+
+	noCursors := make(map[common.Address]string)
+	res, cursors, err := a.getLogsOverAddresses(ctx, filter.fc, noCursors)
+	if err != nil {
+		return nil, err
+	}
+	a.filtersMu.Lock()
+	updatedFilter := a.filters[filterID]
+	updatedFilter.cursors = cursors
+	a.filters[filterID] = updatedFilter
+	a.filtersMu.Unlock()
+	return res, nil
+}
+
+func (a *FilterAPI) GetLogs(
+	ctx context.Context,
+	crit filters.FilterCriteria,
+) ([]*ethtypes.Log, error) {
+	logs, _, err := a.getLogsOverAddresses(
+		ctx,
+		crit,
+		make(map[common.Address]string),
+	)
+	return logs, err
+}
+
+// pulls logs from tendermint client over multiple addresses.
+func (a *FilterAPI) getLogsOverAddresses(
+	ctx context.Context,
+	crit filters.FilterCriteria,
+	cursors map[common.Address]string,
+) ([]*ethtypes.Log, map[common.Address]string, error) {
+	res := make([]*ethtypes.Log, 0)
+	if len(crit.Addresses) == 0 {
+		crit.Addresses = append(crit.Addresses, common.Address{})
+	}
+	updatedAddrToCursor := make(map[common.Address]string)
+	for _, address := range crit.Addresses {
+		var cursor string
+		if _, ok := cursors[address]; !ok {
+			cursor = ""
+		} else {
+			cursor = cursors[address]
+		}
+		resAddr, cursor, err := a.getLogs(
+			ctx,
+			crit.BlockHash,
+			crit.FromBlock,
+			crit.ToBlock,
+			address,
+			crit.Topics,
+			cursor,
+		)
+		if err != nil {
+			return nil, nil, err
+		}
+		res = append(res, resAddr...)
+		updatedAddrToCursor[address] = cursor
+	}
+	return res, updatedAddrToCursor, nil
+}
+
+// pulls logs from tendermint client for a single address.
+func (a *FilterAPI) getLogs(
+	ctx context.Context,
+	blockHash *common.Hash,
+	fromBlock *big.Int,
+	toBlock *big.Int,
+	address common.Address,
+	topics [][]common.Hash,
+	cursor string,
+) ([]*ethtypes.Log, string, error) {
+	q := NewQueryBuilder()
+	if blockHash != nil {
+		q = q.FilterBlockHash(blockHash.Hex())
+	}
+	if fromBlock != nil {
+		q = q.FilterBlockNumberStart(fromBlock.Int64())
+	}
+	if toBlock != nil {
+		q = q.FilterBlockNumberEnd(toBlock.Int64())
+	}
+	if (address != common.Address{}) {
+		q = q.FilterContractAddress(address.Hex())
+	}
+	if len(topics) > 0 {
+		topicsStrs := make([][]string, len(topics))
+		for i, topic := range topics {
+			topicsStrs[i] = make([]string, len(topic))
+			for j, t := range topic {
+				topicsStrs[i][j] = t.Hex()
+			}
+		}
+		q = q.FilterTopics(topicsStrs)
+	}
+	hasMore := true
+	logs := []*ethtypes.Log{}
+	for hasMore {
+		res, err := a.tmClient.Events(ctx, &coretypes.RequestEvents{
+			Filter: &coretypes.EventFilter{Query: q.Build()},
+			After:  cursor,
+		})
+		if err != nil {
+			return nil, "", err
+		}
+		hasMore = res.More
+		cursor = res.Newest
+		for _, log := range res.Items {
+			abciEvent := abci.Event{}
+			err := json.Unmarshal(log.Data, &abciEvent)
+			if err != nil {
+				return nil, "", err
+			}
+			ethLog, err := encodeEventToLog(abciEvent)
+			if err != nil {
+				return nil, "", err
+			}
+			logs = append(logs, ethLog)
+		}
+	}
+	return logs, cursor, nil
+}
+
+func (a *FilterAPI) UninstallFilter(
+	_ context.Context,
+	filterID uint64,
+) bool {
+	a.filtersMu.Lock()
+	defer a.filtersMu.Unlock()
+	_, found := a.filters[filterID]
+	if !found {
+		return false
+	}
+	delete(a.filters, filterID)
+	return true
+}

evmrpc/filter_test.go

@@ -0,0 +1,304 @@
+package evmrpc
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFilterNew(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		fromBlock string
+		toBlock   string
+		blockHash common.Hash
+		addrs     []common.Address
+		topics    [][]common.Hash
+		wantErr   bool
+	}{
+		{
+			name:      "happy path",
+			fromBlock: "0x1",
+			toBlock:   "0x2",
+			addrs:     []common.Address{common.HexToAddress("0x123")},
+			topics:    [][]common.Hash{{common.HexToHash("0x456")}},
+			wantErr:   false,
+		},
+		{
+			name:      "error: block hash and block range both given",
+			fromBlock: "0x1",
+			toBlock:   "0x2",
+			blockHash: common.HexToHash("0xabc"),
+			wantErr:   true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			filterCriteria := map[string]interface{}{
+				"fromBlock": tt.fromBlock,
+				"toBlock":   tt.toBlock,
+				"address":   tt.addrs,
+				"topics":    tt.topics,
+			}
+			if tt.blockHash != (common.Hash{}) {
+				filterCriteria["blockHash"] = tt.blockHash.Hex()
+			}
+			if len(tt.fromBlock) > 0 || len(tt.toBlock) > 0 {
+				filterCriteria["fromBlock"] = tt.fromBlock
+				filterCriteria["toBlock"] = tt.toBlock
+			}
+			resObj := sendRequestGood(t, "newFilter", filterCriteria)
+			if tt.wantErr {
+				_, ok := resObj["error"]
+				require.True(t, ok)
+			} else {
+				got := resObj["result"].(float64)
+				// make sure next filter id is not equal to this one
+				resObj := sendRequestGood(t, "newFilter", filterCriteria)
+				got2 := resObj["result"].(float64)
+				require.NotEqual(t, got, got2)
+			}
+		})
+	}
+}
+
+func TestFilterUninstall(t *testing.T) {
+	t.Parallel()
+	// uninstall existing filter
+	filterCriteria := map[string]interface{}{
+		"fromBlock": "0x1",
+		"toBlock":   "0xa",
+	}
+	resObj := sendRequestGood(t, "newFilter", filterCriteria)
+	filterId := int(resObj["result"].(float64))
+	require.GreaterOrEqual(t, filterId, 1)
+
+	resObj = sendRequest(t, TestPort, "uninstallFilter", filterId)
+	uninstallSuccess := resObj["result"].(bool)
+	require.True(t, uninstallSuccess)
+
+	// uninstall non-existing filter
+	nonExistingFilterId := 100
+	resObj = sendRequest(t, TestPort, "uninstallFilter", nonExistingFilterId)
+	uninstallSuccess = resObj["result"].(bool)
+	require.False(t, uninstallSuccess)
+}
+
+func TestFilterGetLogs(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		blockHash common.Hash
+		fromBlock string
+		toBlock   string
+		addrs     []common.Address
+		topics    [][]common.Hash
+		wantErr   bool
+		wantLen   int
+		check     func(t *testing.T, log map[string]interface{})
+	}{
+		{
+			name:      "filter by single address",
+			fromBlock: "0x2",
+			toBlock:   "0x2",
+			addrs:     []common.Address{common.HexToAddress("0x1111111111111111111111111111111111111112")},
+			wantErr:   false,
+			check: func(t *testing.T, log map[string]interface{}) {
+				require.Equal(t, "0x1111111111111111111111111111111111111112", log["address"].(string))
+			},
+			wantLen: 1,
+		},
+		{
+			name:      "filter by single topic",
+			blockHash: common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000"),
+			fromBlock: "0x3",
+			toBlock:   "0x3",
+			topics:    [][]common.Hash{{common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000123")}},
+			wantErr:   false,
+			check: func(t *testing.T, log map[string]interface{}) {
+				require.Equal(t, "0x0000000000000000000000000000000000000000000000000000000000000123", log["topics"].([]interface{})[0].(string))
+			},
+			wantLen: 1,
+		},
+		{
+			name:      "multiple addresses, multiple topics",
+			blockHash: common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000"),
+			fromBlock: "0x2",
+			toBlock:   "0x2",
+			addrs: []common.Address{
+				common.HexToAddress("0x1111111111111111111111111111111111111112"),
+				common.HexToAddress("0x1111111111111111111111111111111111111113"),
+			},
+			topics: [][]common.Hash{
+				{common.Hash(common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000123"))},
+				{common.Hash(common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000456"))},
+			},
+			wantErr: false,
+			check: func(t *testing.T, log map[string]interface{}) {
+				if log["address"].(string) != "0x1111111111111111111111111111111111111112" && log["address"].(string) != "0x1111111111111111111111111111111111111113" {
+					t.Fatalf("address %s not in expected list", log["address"].(string))
+				}
+				firstTopic := log["topics"].([]interface{})[0].(string)
+				secondTopic := log["topics"].([]interface{})[1].(string)
+				require.Equal(t, "0x0000000000000000000000000000000000000000000000000000000000000123", firstTopic)
+				require.Equal(t, "0x0000000000000000000000000000000000000000000000000000000000000456", secondTopic)
+			},
+			wantLen: 2,
+		},
+		{
+			name:      "wildcard first topic",
+			blockHash: common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000"),
+			fromBlock: "0x2",
+			toBlock:   "0x2",
+			topics: [][]common.Hash{
+				{},
+				{common.Hash(common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000456"))},
+			},
+			wantErr: false,
+			check: func(t *testing.T, log map[string]interface{}) {
+				secondTopic := log["topics"].([]interface{})[1].(string)
+				require.Equal(t, "0x0000000000000000000000000000000000000000000000000000000000000456", secondTopic)
+			},
+			wantLen: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			filterCriteria := map[string]interface{}{
+				"fromBlock": tt.fromBlock,
+				"toBlock":   tt.toBlock,
+				"address":   tt.addrs,
+				"topics":    tt.topics,
+			}
+			if tt.blockHash != (common.Hash{}) {
+				filterCriteria["blockHash"] = tt.blockHash.Hex()
+			}
+			if len(tt.fromBlock) > 0 || len(tt.toBlock) > 0 {
+				filterCriteria["fromBlock"] = tt.fromBlock
+				filterCriteria["toBlock"] = tt.toBlock
+			}
+			resObj := sendRequestGood(t, "getLogs", filterCriteria)
+			if tt.wantErr {
+				_, ok := resObj["error"]
+				require.True(t, ok)
+			} else {
+				got := resObj["result"].([]interface{})
+				for _, log := range got {
+					logObj := log.(map[string]interface{})
+					tt.check(t, logObj)
+				}
+				require.Equal(t, len(got), tt.wantLen)
+			}
+		})
+	}
+}
+
+func TestFilterGetFilterLogs(t *testing.T) {
+	t.Parallel()
+	filterCriteria := map[string]interface{}{
+		"fromBlock": "0x4",
+		"toBlock":   "0x4",
+	}
+	resObj := sendRequestGood(t, "newFilter", filterCriteria)
+	filterId := int(resObj["result"].(float64))
+
+	resObj = sendRequest(t, TestPort, "getFilterLogs", filterId)
+	logs := resObj["result"].([]interface{})
+	require.Equal(t, 1, len(logs))
+	for _, log := range logs {
+		logObj := log.(map[string]interface{})
+		require.Equal(t, "0x4", logObj["blockNumber"].(string))
+	}
+
+	// error: filter id does not exist
+	nonexistentFilterId := 1000
+	resObj = sendRequest(t, TestPort, "getFilterLogs", nonexistentFilterId)
+	_, ok := resObj["error"]
+	require.True(t, ok)
+}
+
+func TestFilterGetFilterChanges(t *testing.T) {
+	t.Parallel()
+	filterCriteria := map[string]interface{}{
+		"fromBlock": "0x5",
+	}
+	resObj := sendRequest(t, TestPort, "newFilter", filterCriteria)
+	filterId := int(resObj["result"].(float64))
+
+	resObj = sendRequest(t, TestPort, "getFilterChanges", filterId)
+	logs := resObj["result"].([]interface{})
+	require.Equal(t, 1, len(logs))
+	logObj := logs[0].(map[string]interface{})
+	require.Equal(t, "0x5", logObj["blockNumber"].(string))
+
+	// another query
+	resObj = sendRequest(t, TestPort, "getFilterChanges", filterId)
+	logs = resObj["result"].([]interface{})
+	require.Equal(t, 1, len(logs))
+	logObj = logs[0].(map[string]interface{})
+	require.Equal(t, "0x6", logObj["blockNumber"].(string))
+
+	// error: filter id does not exist
+	nonExistingFilterId := 1000
+	resObj = sendRequest(t, TestPort, "getFilterChanges", nonExistingFilterId)
+	_, ok := resObj["error"]
+	require.True(t, ok)
+}
+
+func TestFilterExpiration(t *testing.T) {
+	t.Parallel()
+	filterCriteria := map[string]interface{}{
+		"fromBlock": "0x1",
+		"toBlock":   "0xa",
+	}
+	resObj := sendRequestGood(t, "newFilter", filterCriteria)
+	filterId := int(resObj["result"].(float64))
+
+	// wait for filter to expire
+	time.Sleep(2 * filterTimeoutDuration)
+
+	resObj = sendRequest(t, TestPort, "getFilterLogs", filterId)
+	_, ok := resObj["error"]
+	require.True(t, ok)
+}
+
+func TestFilterGetFilterLogsKeepsFilterAlive(t *testing.T) {
+	t.Parallel()
+	filterCriteria := map[string]interface{}{
+		"fromBlock": "0x1",
+		"toBlock":   "0xa",
+	}
+	resObj := sendRequestGood(t, "newFilter", filterCriteria)
+	filterId := int(resObj["result"].(float64))
+
+	for i := 0; i < 5; i++ {
+		// should keep filter alive
+		resObj = sendRequestGood(t, "getFilterLogs", filterId)
+		_, ok := resObj["error"]
+		require.False(t, ok)
+		time.Sleep(filterTimeoutDuration / 2)
+	}
+}
+
+func TestFilterGetFilterChangesKeepsFilterAlive(t *testing.T) {
+	t.Parallel()
+	filterCriteria := map[string]interface{}{
+		"fromBlock": "0x1",
+		"toBlock":   "0xa",
+	}
+	resObj := sendRequestGood(t, "newFilter", filterCriteria)
+	filterId := int(resObj["result"].(float64))
+
+	for i := 0; i < 5; i++ {
+		// should keep filter alive
+		resObj = sendRequestGood(t, "getFilterChanges", filterId)
+		_, ok := resObj["error"]
+		require.False(t, ok)
+		time.Sleep(filterTimeoutDuration / 2)
+	}
+}

evmrpc/server.go

@@ -23,6 +23,7 @@ func NewEVMHTTPServer(
 	ctxProvider func(int64) sdk.Context,
 	txConfig client.TxConfig,
 	simulationConfig *SimulateConfig,
+	filterConfig *FilterConfig,
 ) (EVMServer, error) {
 	httpServer := newHTTPServer(logger, timeouts)
 	if err := httpServer.setListenAddr(addr, port); err != nil {
@@ -57,6 +58,10 @@ func NewEVMHTTPServer(
 			Namespace: "eth",
 			Service:   NewSimulationAPI(ctxProvider, k, tmClient, simulationConfig),
 		},
+		{
+			Namespace: "eth",
+			Service:   NewFilterAPI(tmClient, filterConfig),
+		},
 	}
 	if err := httpServer.enableRPC(apis, httpConfig{
 		// TODO: add CORS configs and virtual host configs

evmrpc/subscribe.go

@@ -2,9 +2,11 @@ package evmrpc
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
+	"github.com/sei-protocol/sei-chain/utils"
 	"github.com/sei-protocol/sei-chain/x/evm/types"
 	rpcclient "github.com/tendermint/tendermint/rpc/client"
 	"github.com/tendermint/tendermint/rpc/coretypes"
@@ -68,11 +70,48 @@ func (q *QueryBuilder) FilterTopic(topic string) *QueryBuilder {
 	return q
 }
 
+func (q *QueryBuilder) FilterTopics(topics [][]string) *QueryBuilder {
+	if len(topics) == 0 {
+		return q
+	}
+	pattern, err := getTopicsRegex(topics)
+	if err != nil {
+		panic(err)
+	}
+	q.conditions = append(q.conditions, fmt.Sprintf("%s.%s = MATCHES '%s'", types.EventTypeEVMLog, types.AttributeTypeTopics, pattern))
+	return q
+}
+
+func getTopicsRegex(topics [][]string) (string, error) {
+	if len(topics) == 0 {
+		return "", errors.New("topics array must be at least length 1")
+	}
+
+	topicRegex := func(topic []string) string {
+		if len(topic) == 0 {
+			return ""
+		}
+		return fmt.Sprintf("(%s)", strings.Join(topic, "|"))
+	}
+
+	return fmt.Sprintf("\\[%s.*\\]", strings.Join(utils.Map(topics, topicRegex), "[^\\,]*,")), nil
+}
+
 func (q *QueryBuilder) FilterBlockNumber(blockNumber int64) *QueryBuilder {
 	q.conditions = append(q.conditions, fmt.Sprintf("%s.%s = '%d'", types.EventTypeEVMLog, types.AttributeTypeBlockNumber, blockNumber))
 	return q
 }
 
+func (q *QueryBuilder) FilterBlockNumberStart(blockNumber int64) *QueryBuilder {
+	q.conditions = append(q.conditions, fmt.Sprintf("%s.%s >= '%d'", types.EventTypeEVMLog, types.AttributeTypeBlockNumber, blockNumber))
+	return q
+}
+
+func (q *QueryBuilder) FilterBlockNumberEnd(blockNumber int64) *QueryBuilder {
+	q.conditions = append(q.conditions, fmt.Sprintf("%s.%s <= '%d'", types.EventTypeEVMLog, types.AttributeTypeBlockNumber, blockNumber))
+	return q
+}
+
 func (q *QueryBuilder) FilterTxIndex(txIndex int64) *QueryBuilder {
 	q.conditions = append(q.conditions, fmt.Sprintf("%s.%s = '%d'", types.EventTypeEVMLog, types.AttributeTypeTxIndex, txIndex))
 	return q

evmrpc/subscribe_test.go

@@ -2,6 +2,7 @@ package evmrpc
 
 import (
 	"context"
+	"regexp"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -40,3 +41,65 @@ func mockQueryBuilder() *QueryBuilder {
 	q.FilterTopic("topic b")
 	return q
 }
+
+func TestGetTopicsRegex(t *testing.T) {
+	tests := []struct {
+		name         string
+		topics       [][]string
+		wantErr      bool
+		wantMatch    []string
+		wantNotMatch []string
+	}{
+		{
+			name:    "error: topics length 0",
+			topics:  [][]string{},
+			wantErr: true,
+		},
+		{
+			name:         "match first topic",
+			topics:       [][]string{{"a"}},
+			wantErr:      false,
+			wantMatch:    []string{"[a]", "[a,b]", "[a,a,a,a]"},
+			wantNotMatch: []string{"b", "[b]", "[b,a]", "[a,b"},
+		},
+		{
+			name:         "match first topic with OR",
+			topics:       [][]string{{"a", "b"}}, // first topic can be a or b
+			wantErr:      false,
+			wantMatch:    []string{"[a]", "[a,b]", "[a,c,c,c]", "[b]", "[b,c]", "[b,c,c,c]"},
+			wantNotMatch: []string{"b", "[c]", "[c,a]", "[c,b"},
+		},
+		{
+			name:         "match second topic",
+			topics:       [][]string{{}, {"a"}},
+			wantErr:      false,
+			wantMatch:    []string{"[b,a]", "[c,a]", "[a,a,a]"},
+			wantNotMatch: []string{"b,a]", "[a,b,a]"},
+		},
+		{
+			name:         "match second and fourth topic",
+			topics:       [][]string{{""}, {"a", "c"}, {""}, {"b", "d"}},
+			wantErr:      false,
+			wantMatch:    []string{"[d,a,c,b]", "[c,a,c,d,c]", "[a,c,b,d]"},
+			wantNotMatch: []string{"[a,a,a,a]", "[a,b]", "[c,a,b,c]"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := getTopicsRegex(tt.topics)
+			regex := regexp.MustCompile(got)
+			if tt.wantErr {
+				require.NotNil(t, err)
+				return
+			}
+			require.Nil(t, err)
+			for _, toMatch := range tt.wantMatch {
+				require.True(t, regex.MatchString(toMatch))
+			}
+			for _, toNotMatch := range tt.wantNotMatch {
+				require.False(t, regex.MatchString(toNotMatch))
+			}
+		})
+	}
+}

evmrpc/testutils.go

@@ -9,6 +9,8 @@ import (
 	"io"
 	"math/big"
 	"net/http"
+	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -48,6 +50,9 @@ var Tx sdk.Tx
 
 var SConfig = SimulateConfig{GasCap: 10000000}
 
+var filterTimeoutDuration = 500 * time.Millisecond
+var FConfig = FilterConfig{timeout: filterTimeoutDuration}
+
 type MockClient struct {
 	mock.Client
 }
@@ -93,6 +98,7 @@ func (c *MockClient) BlockByHash(context.Context, bytes.HexBytes) (*coretypes.Re
 }
 
 func (c *MockClient) BlockResults(context.Context, *int64) (*coretypes.ResultBlockResults, error) {
+	abciEvent := NewABCIEventBuilder().SetBlockNum(8).Build()
 	return &coretypes.ResultBlockResults{
 		TxsResults: []*abci.ExecTxResult{
 			{
@@ -102,37 +108,9 @@ func (c *MockClient) BlockResults(context.Context, *int64) (*coretypes.ResultBlo
 				}(),
 				GasWanted: 10,
 				GasUsed:   5,
-				Events: []abci.Event{{
-					Type: types.EventTypeEVMLog,
-					Attributes: []abci.EventAttribute{{
-						Key:   []byte(types.AttributeTypeContractAddress),
-						Value: []byte("0x1111111111111111111111111111111111111111111111111111111111111111"),
-					}, {
-						Key:   []byte(types.AttributeTypeBlockHash),
-						Value: []byte("0x1111111111111111111111111111111111111111111111111111111111111111"),
-					}, {
-						Key:   []byte(types.AttributeTypeBlockNumber),
-						Value: []byte("8"),
-					}, {
-						Key:   []byte(types.AttributeTypeData),
-						Value: []byte("xyz"),
-					}, {
-						Key:   []byte(types.AttributeTypeIndex),
-						Value: []byte("1"),
-					}, {
-						Key:   []byte(types.AttributeTypeTxIndex),
-						Value: []byte("2"),
-					}, {
-						Key:   []byte(types.AttributeTypeRemoved),
-						Value: []byte("true"),
-					}, {
-						Key:   []byte(types.AttributeTypeTopics),
-						Value: []byte("0x1111111111111111111111111111111111111111111111111111111111111111,0x1111111111111111111111111111111111111111111111111111111111111112"),
-					}, {
-						Key:   []byte(types.AttributeTypeTxHash),
-						Value: []byte("0x1111111111111111111111111111111111111111111111111111111111111113"),
-					}},
-				}},
+				Events: []abci.Event{
+					abciEvent,
+				},
 			},
 		},
 	}, nil
@@ -142,6 +120,81 @@ func (c *MockClient) Subscribe(context.Context, string, string, ...int) (<-chan
 	return make(chan coretypes.ResultEvent, 1), nil
 }
 
+func (c *MockClient) Events(_ context.Context, req *coretypes.RequestEvents) (*coretypes.ResultEvents, error) {
+	eb := NewABCIEventBuilder()
+
+	// assume cursor is block number for testing purposes
+	var cursor string
+	if req.After != "" {
+		cursorAfter, err := strconv.Atoi(req.After)
+		if err != nil {
+			panic("invalid cursor")
+		}
+		nextBlockNum := cursorAfter + 1
+		eb = eb.SetBlockNum(nextBlockNum)
+		cursor = strconv.FormatInt(int64(nextBlockNum), 10)
+	} else {
+		var startBlock int
+		re := regexp.MustCompile(`evm_log.block_number >= '(\d+?)'`)
+		matches := re.FindStringSubmatch(req.Filter.Query)
+		if len(matches) == 2 {
+			var err error
+			startBlock, err = strconv.Atoi(matches[1])
+			if err != nil {
+				return nil, err
+			}
+			eb = eb.SetBlockNum(startBlock)
+			cursor = strconv.FormatInt(int64(startBlock), 10)
+		}
+	}
+
+	var blockHash string
+	re := regexp.MustCompile(`evm_log.block_hash = '(.+?)'`)
+	matches := re.FindStringSubmatch(req.Filter.Query)
+	if len(matches) == 2 {
+		blockHash = matches[1]
+		eb = eb.SetBlockHash(blockHash)
+	}
+
+	var contractAddress string
+	re = regexp.MustCompile(`evm_log.contract_address = '(.+?)'.*`)
+	matches = re.FindStringSubmatch(req.Filter.Query)
+	if len(matches) == 2 {
+		contractAddress = matches[1]
+		eb = eb.SetContractAddress(contractAddress)
+	}
+
+	// hardcode topic matches to match up with tests since doing the regex is too complicated
+	if strings.Contains(req.Filter.Query, "evm_log.topics = MATCHES '\\[(0x0000000000000000000000000000000000000000000000000000000000000123).*\\]'") {
+		eb = eb.SetTopics([]string{"0x0000000000000000000000000000000000000000000000000000000000000123"})
+	} else if strings.Contains(req.Filter.Query, "evm_log.topics = MATCHES '\\[(0x0000000000000000000000000000000000000000000000000000000000000123)[^\\,]*,(0x0000000000000000000000000000000000000000000000000000000000000456).*\\]'") {
+		eb = eb.SetTopics([]string{"0x0000000000000000000000000000000000000000000000000000000000000123", "0x0000000000000000000000000000000000000000000000000000000000000456"})
+	} else if strings.Contains(req.Filter.Query, "evm_log.topics = MATCHES '\\[[^\\,]*,(0x0000000000000000000000000000000000000000000000000000000000000456).*\\]'") {
+		eb = eb.SetTopics([]string{"0x0000000000000000000000000000000000000000000000000000000000000123", "0x0000000000000000000000000000000000000000000000000000000000000456"})
+	}
+
+	return buildSingleResultEvent(eb.Build(), false, cursor, "event"), nil
+}
+
+func buildSingleResultEvent(abciEvent abci.Event, more bool, cursor string, event string) *coretypes.ResultEvents {
+	eventData, err := json.Marshal(abciEvent)
+	if err != nil {
+		panic(err)
+	}
+	return &coretypes.ResultEvents{
+		Items: []*coretypes.EventItem{
+			{
+				Cursor: cursor,
+				Event:  event,
+				Data:   eventData,
+			},
+		},
+		More:   more,
+		Oldest: cursor,
+		Newest: cursor,
+	}
+}
+
 func (c *MockClient) BroadcastTx(context.Context, tmtypes.Tx) (*coretypes.ResultBroadcastTx, error) {
 	return &coretypes.ResultBroadcastTx{Code: 0}, nil
 }
@@ -176,14 +229,14 @@ var Ctx sdk.Context
 func init() {
 	types.RegisterInterfaces(EncodingConfig.InterfaceRegistry)
 	EVMKeeper, _, Ctx = keeper.MockEVMKeeper()
-	httpServer, err := NewEVMHTTPServer(log.NewNopLogger(), TestAddr, TestPort, rpc.DefaultHTTPTimeouts, &MockClient{}, EVMKeeper, func(int64) sdk.Context { return Ctx }, TxConfig, &SConfig)
+	httpServer, err := NewEVMHTTPServer(log.NewNopLogger(), TestAddr, TestPort, rpc.DefaultHTTPTimeouts, &MockClient{}, EVMKeeper, func(int64) sdk.Context { return Ctx }, TxConfig, &SConfig, &FConfig)
 	if err != nil {
 		panic(err)
 	}
 	if err := httpServer.Start(); err != nil {
 		panic(err)
 	}
-	badHTTPServer, err := NewEVMHTTPServer(log.NewNopLogger(), TestAddr, TestBadPort, rpc.DefaultHTTPTimeouts, &MockBadClient{}, EVMKeeper, func(int64) sdk.Context { return Ctx }, TxConfig, &SConfig)
+	badHTTPServer, err := NewEVMHTTPServer(log.NewNopLogger(), TestAddr, TestBadPort, rpc.DefaultHTTPTimeouts, &MockBadClient{}, EVMKeeper, func(int64) sdk.Context { return Ctx }, TxConfig, &SConfig, &FConfig)
 	if err != nil {
 		panic(err)
 	}
@@ -306,6 +359,27 @@ func formatParam(p interface{}) string {
 		return fmt.Sprintf("%f", v)
 	case string:
 		return fmt.Sprintf("\"%s\"", v)
+	case common.Address:
+		return fmt.Sprintf("\"%s\"", v)
+	case []common.Address:
+		wrapper := func(i common.Address) string {
+			return formatParam(i)
+		}
+		return fmt.Sprintf("[%s]", strings.Join(utils.Map(v, wrapper), ","))
+	case common.Hash:
+		return fmt.Sprintf("\"%s\"", v)
+	case []common.Hash:
+		wrapper := func(i common.Hash) string {
+			return formatParam(i)
+		}
+		return fmt.Sprintf("[%s]", strings.Join(utils.Map(v, wrapper), ","))
+	case [][]common.Hash:
+		wrapper := func(i []common.Hash) string {
+			return formatParam(i)
+		}
+		return fmt.Sprintf("[%s]", strings.Join(utils.Map(v, wrapper), ","))
+	case []string:
+		return fmt.Sprintf("[%s]", strings.Join(v, ","))
 	case []interface{}:
 		return fmt.Sprintf("[%s]", strings.Join(utils.Map(v, formatParam), ","))
 	case map[string]interface{}:
@@ -315,6 +389,111 @@ func formatParam(p interface{}) string {
 		}
 		return fmt.Sprintf("{%s}", strings.Join(kvs, ","))
 	default:
-		return fmt.Sprintf("%s", p)
+		panic("did not match on type")
+	}
+}
+
+type ABCIEventBuilder struct {
+	contractAddress string
+	blockHash       string
+	blockNum        int
+	data            string
+	index           int
+	txIndex         int
+	removed         bool
+	topics          []string
+	txHash          common.Hash
+}
+
+func NewABCIEventBuilder() *ABCIEventBuilder {
+	return &ABCIEventBuilder{
+		contractAddress: "0x1111111111111111111111111111111111111111111111111111111111111111",
+		blockHash:       "0x1111111111111111111111111111111111111111111111111111111111111111",
+		blockNum:        8,
+		data:            "xyz",
+		index:           1,
+		txIndex:         2,
+		removed:         true,
+		topics:          []string{"0x1111111111111111111111111111111111111111111111111111111111111111,0x1111111111111111111111111111111111111111111111111111111111111112"},
+		txHash:          common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111113"),
+	}
+}
+
+func (b *ABCIEventBuilder) SetContractAddress(contractAddress string) *ABCIEventBuilder {
+	b.contractAddress = contractAddress
+	return b
+}
+
+func (b *ABCIEventBuilder) SetBlockHash(blockHash string) *ABCIEventBuilder {
+	b.blockHash = blockHash
+	return b
+}
+
+func (b *ABCIEventBuilder) SetBlockNum(blockNum int) *ABCIEventBuilder {
+	b.blockNum = blockNum
+	return b
+}
+
+func (b *ABCIEventBuilder) SetData(data string) *ABCIEventBuilder {
+	b.data = data
+	return b
+}
+
+func (b *ABCIEventBuilder) SetIndex(index int) *ABCIEventBuilder {
+	b.index = index
+	return b
+}
+
+func (b *ABCIEventBuilder) SetTxIndex(txIndex int) *ABCIEventBuilder {
+	b.txIndex = txIndex
+	return b
+}
+
+func (b *ABCIEventBuilder) SetRemoved(removed bool) *ABCIEventBuilder {
+	b.removed = removed
+	return b
+}
+
+func (b *ABCIEventBuilder) SetTopics(topics []string) *ABCIEventBuilder {
+	b.topics = topics
+	return b
+}
+
+func (b *ABCIEventBuilder) SetTxHash(txHash common.Hash) *ABCIEventBuilder {
+	b.txHash = txHash
+	return b
+}
+
+func (b *ABCIEventBuilder) Build() abci.Event {
+	return abci.Event{
+		Type: types.EventTypeEVMLog,
+		Attributes: []abci.EventAttribute{{
+			Key:   []byte(types.AttributeTypeContractAddress),
+			Value: []byte(b.contractAddress),
+		}, {
+			Key:   []byte(types.AttributeTypeBlockHash),
+			Value: []byte(b.blockHash),
+		}, {
+			Key:   []byte(types.AttributeTypeBlockNumber),
+			Value: []byte(fmt.Sprintf("%d", b.blockNum)),
+		}, {
+			Key:   []byte(types.AttributeTypeData),
+			Value: []byte(b.data),
+		}, {
+			Key:   []byte(types.AttributeTypeIndex),
+			Value: []byte(fmt.Sprintf("%d", b.index)),
+		}, {
+			Key:   []byte(types.AttributeTypeTxIndex),
+			Value: []byte(fmt.Sprintf("%d", b.txIndex)),
+		}, {
+			Key:   []byte(types.AttributeTypeRemoved),
+			Value: []byte(fmt.Sprintf("%t", b.removed)),
+		}, {
+			Key:   []byte(types.AttributeTypeTopics),
+			Value: []byte(strings.Join(b.topics, ",")),
+		}, {
+			Key:   []byte(types.AttributeTypeTxHash),
+			Value: []byte(b.txHash.Hex()),
+		}},
 	}
 }