feat: rpc block fetcher, move filters to internal

* move filters to internal folder
* rpc block fetcher
* add benchmarks:

goos: linux
goarch: amd64
pkg: github.com/grassrootseconomics/cic-chain-events/pkg/fetch
cpu: AMD EPYC Processor
Benchmark_RPC
Benchmark_RPC/RPC_Block_Fetcher_Benchmark
Benchmark_RPC/RPC_Block_Fetcher_Benchmark-4                   25          46000646 ns/op          221697 B/op        844 allocs/op
Benchmark_GraphQL
Benchmark_GraphQL/GraphQL_Block_Fetcher_Benchmark
Benchmark_GraphQL/GraphQL_Block_Fetcher_Benchmark-4           56          21219962 ns/op           56686 B/op         94 allocs/op
PASS
ok      github.com/grassrootseconomics/cic-chain-events/pkg/fetch       2.920s

Makefile

@@ -16,6 +16,9 @@ mod:
 test-pkg:
 	TEST_GRAPHQL_ENDPOINT=https://rpc.alfajores.celo.grassecon.net/graphql go test -v -covermode atomic -coverprofile=covprofile ./pkg/...
 
+bench-pkg:
+	TEST_RPC_ENDPOINT=https://rpc.alfajores.celo.grassecon.net TEST_GRAPHQL_ENDPOINT=https://rpc.alfajores.celo.grassecon.net/graphql go test -v -bench=. -run=^Benchmark ./pkg/...
+
 migrate:
 	tern migrate -c migrations/tern.conf
 

cmd/filters.go

@@ -4,7 +4,7 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/grassrootseconomics/cic-chain-events/pkg/filter"
+	"github.com/grassrootseconomics/cic-chain-events/internal/filter"
 )
 
 func initAddressFilter() filter.Filter {

cmd/main.go

@@ -13,7 +13,7 @@ import (
 	"github.com/grassrootseconomics/cic-chain-events/internal/api"
 	"github.com/grassrootseconomics/cic-chain-events/internal/pipeline"
 	"github.com/grassrootseconomics/cic-chain-events/internal/syncer"
-	"github.com/grassrootseconomics/cic-chain-events/pkg/filter"
+	"github.com/grassrootseconomics/cic-chain-events/internal/filter"
 	"github.com/knadh/goyesql/v2"
 	"github.com/knadh/koanf"
 	"github.com/zerodha/logf"

internal/pipeline/pipeline.go

@@ -3,9 +3,9 @@ package pipeline
 import (
 	"context"
 
+	"github.com/grassrootseconomics/cic-chain-events/internal/filter"
 	"github.com/grassrootseconomics/cic-chain-events/internal/store"
 	"github.com/grassrootseconomics/cic-chain-events/pkg/fetch"
-	"github.com/grassrootseconomics/cic-chain-events/pkg/filter"
 	"github.com/jackc/pgx/v5"
 	"github.com/zerodha/logf"
 )

pkg/fetch/benchmark_test.go

@@ -0,0 +1,50 @@
+package fetch
+
+import (
+	"context"
+	"testing"
+
+	celo "github.com/grassrootseconomics/cic-celo-sdk"
+)
+
+func Benchmark_RPC(b *testing.B) {
+	celoProvider, err := celo.NewProvider(celo.ProviderOpts{
+		ChainId:     celo.MainnetChainId,
+		RpcEndpoint: rpcEndpoint,
+	})
+
+	rpc := NewRPCFetcher(RPCOpts{
+		RPCProvider: celoProvider,
+	})
+
+	if err != nil {
+		return
+	}
+
+	b.Run("RPC_Block_Fetcher_Benchmark", func(b *testing.B) {
+		for n := 0; n < b.N; n++ {
+			_, err := rpc.Block(context.Background(), 14974600)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+		b.ReportAllocs()
+	})
+}
+
+func Benchmark_GraphQL(b *testing.B) {
+	graphql := NewGraphqlFetcher(GraphqlOpts{
+		GraphqlEndpoint: graphqlEndpoint,
+	})
+
+	b.Run("GraphQL_Block_Fetcher_Benchmark", func(b *testing.B) {
+		for n := 0; n < b.N; n++ {
+			_, err := graphql.Block(context.Background(), 14974600)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+		b.ReportAllocs()
+	})
+
+}

pkg/fetch/fetch.go

@@ -10,7 +10,7 @@ type Fetch interface {
 // Transaction reprsents a JSON object of all important mined transaction information
 type Transaction struct {
 	Block struct {
-		Number    uint   `json:"number"`
+		Number    uint64 `json:"number"`
 		Timestamp string `json:"timestamp"`
 	} `json:"block"`
 	Hash  string `json:"hash"`
@@ -23,8 +23,8 @@ type Transaction struct {
 	} `json:"to"`
 	Value     string `json:"value"`
 	InputData string `json:"inputData"`
-	Status    uint   `json:"status"`
-	GasUsed   uint   `json:"gasUsed"`
+	Status    uint64 `json:"status"`
+	GasUsed   uint64 `json:"gasUsed"`
 }
 
 // BlockResponse represents a full fetch JSON response

pkg/fetch/rpc.go

@@ -3,7 +3,9 @@ package fetch
 import (
 	"context"
 	"math/big"
+	"strings"
 
+	"github.com/celo-org/celo-blockchain/common/hexutil"
 	"github.com/celo-org/celo-blockchain/core/types"
 	celo "github.com/grassrootseconomics/cic-celo-sdk"
 	"github.com/grassrootseconomics/w3-celo-patch/module/eth"
@@ -37,7 +39,7 @@ func (f *RPC) Block(ctx context.Context, blockNumber uint64) (FetchResponse, err
 		ctx,
 		eth.BlockByNumber(big.NewInt(int64(blockNumber))).Returns(&block),
 	); err != nil {
-		return FetchResponse{}, nil
+		return FetchResponse{}, err
 	}
 
 	txCount := len(block.Transactions())
@@ -58,7 +60,31 @@ func (f *RPC) Block(ctx context.Context, blockNumber uint64) (FetchResponse, err
 		return FetchResponse{}, nil
 	}
 
-	// TODO: Create FetchResponse
+	for i := 0; i < txCount; i++ {
+		tx := Transaction{}
+
+		tx.Block.Number = block.NumberU64()
+		tx.Block.Timestamp = hexutil.EncodeUint64(block.Time())
+		tx.Hash = txsReceipt[i].TxHash.Hex()
+		tx.Index = txsReceipt[i].TransactionIndex
+
+		from, err := types.Sender(types.LatestSignerForChainID(txs[i].ChainId()), &txs[i])
+		if err != nil {
+			return FetchResponse{}, err
+		}
+
+		tx.From.Address = strings.ToLower(from.Hex())
+		tx.To.Address = strings.ToLower(txs[i].To().Hex())
+		tx.Value = hexutil.EncodeBig(txs[i].Value())
+		tx.InputData = hexutil.Encode(txs[i].Data())
+		tx.Status = txsReceipt[i].Status
+		tx.GasUsed = txsReceipt[i].GasUsed
+
+		fetchResponse.Data.Block.Transactions = append(
+			fetchResponse.Data.Block.Transactions,
+			tx,
+		)
+	}
 
 	return fetchResponse, nil
 }

pkg/fetch/rpc_test.go

@@ -0,0 +1,49 @@
+package fetch
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	celo "github.com/grassrootseconomics/cic-celo-sdk"
+	"github.com/stretchr/testify/suite"
+)
+
+var (
+	rpcEndpoint = os.Getenv("TEST_RPC_ENDPOINT")
+)
+
+type RPCTestSuite struct {
+	suite.Suite
+	fetch Fetch
+}
+
+func (s *RPCTestSuite) SetupSuite() {
+	celoProvider, err := celo.NewProvider(celo.ProviderOpts{
+		ChainId:     celo.MainnetChainId,
+		RpcEndpoint: rpcEndpoint,
+	})
+
+	if err != nil {
+		return
+	}
+
+	s.fetch = NewRPCFetcher(RPCOpts{
+		RPCProvider: celoProvider,
+	})
+}
+
+func (s *RPCTestSuite) Test_E2E_Fetch_Existing_Block() {
+	resp, err := s.fetch.Block(context.Background(), 14974600)
+	s.NoError(err)
+	s.Len(resp.Data.Block.Transactions, 3)
+}
+
+func (s *RPCTestSuite) Test_E2E_Fetch_Non_Existing_Block() {
+	_, err := s.fetch.Block(context.Background(), 14974600000)
+	s.Error(err)
+}
+
+func TestRPCSuite(t *testing.T) {
+	suite.Run(t, new(RPCTestSuite))
+}