Add filter manager logic for the JSON-RPC server

.github/golangci.yaml

@@ -7,6 +7,10 @@ run:
  modules-download-mode: readonly
  allow-parallel-runners: false
  go: ""
+ build-tags:
+ - testmocks
+ skip-dirs:
+ - serve/handlers/subs/filters/mocks
 
 output:
  uniq-by-line: false

Makefile

@@ -18,3 +18,8 @@ gofumpt:
 fixalign:
  go install golang.org/x/tools/go/analysis/passes/fieldalignment/cmd/fieldalignment@latest
  fieldalignment -fix $(filter-out $@,$(MAKECMDGOALS)) # the full package name (not path!)
+
+.PHONY: test
+test:
+ go clean -testcache
+ go test -v -tags "testmocks" ./...

serve/handler_test.go

@@ -0,0 +1,182 @@
+package serve
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "testing"
+
+ "github.com/gnolang/tx-indexer/serve/metadata"
+ "github.com/gnolang/tx-indexer/serve/spec"
+ "github.com/go-chi/chi/v5"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "go.uber.org/zap"
+)
+
+func decodeResponse[T spec.BaseJSONResponse | spec.BaseJSONResponses](t *testing.T, responseBody []byte) *T {
+ t.Helper()
+
+ var response *T
+
+ require.NoError(t, json.NewDecoder(bytes.NewReader(responseBody)).Decode(&response))
+
+ return response
+}
+
+// setupTestWebServer is a helper function for common setup logic
+func setupTestWebServer(t *testing.T, callback func(s *JSONRPC)) *testWebServer {
+ t.Helper()
+
+ s := newWebServer(t, callback)
+ s.start()
+
+ return s
+}
+
+// TestHTTP_Handle_BatchRequest verifies that the JSON-RPC server:
+// - can handle a single HTTP request to a dummy endpoint
+// - can handle a batch HTTP request to a dummy endpoint
+func TestHTTP_Handle(t *testing.T) {
+ t.Parallel()
+
+ var (
+ commonResponse = "This is a common response!"
+ method = "dummy"
+ )
+
+ singleRequest, err := json.Marshal(
+ spec.NewJSONRequest(1, method, nil),
+ )
+ require.NoError(t, err)
+
+ requests := spec.BaseJSONRequests{
+ spec.NewJSONRequest(1, method, nil),
+ spec.NewJSONRequest(2, method, nil),
+ spec.NewJSONRequest(3, method, nil),
+ }
+
+ batchRequest, err := json.Marshal(requests)
+ require.NoError(t, err)
+
+ testTable := []struct {
+ verifyResponse func(response []byte) error
+ name string
+ request []byte
+ }{
+ {
+ func(resp []byte) error {
+ response := decodeResponse[spec.BaseJSONResponse](t, resp)
+
+ assert.Equal(t, spec.NewJSONResponse(1, commonResponse, nil), response)
+
+ return nil
+ },
+ "single HTTP request",
+ singleRequest,
+ },
+ {
+ func(resp []byte) error {
+ responses := decodeResponse[spec.BaseJSONResponses](t, resp)
+
+ for index, response := range *responses {
+ assert.Equal(
+ t,
+ spec.NewJSONResponse(uint(index+1), commonResponse, nil),
+ response,
+ )
+ }
+
+ return nil
+ },
+ "batch HTTP request",
+ batchRequest,
+ },
+ }
+
+ for _, testCase := range testTable {
+ testCase := testCase
+
+ t.Run(testCase.name, func(t *testing.T) {
+ t.Parallel()
+
+ // Create a new JSON-RPC server
+ webServer := setupTestWebServer(t, func(s *JSONRPC) {
+ s.handlers = make(handlers)
+
+ s.handlers.addHandler(method, func(_ *metadata.Metadata, _ []any) (any, *spec.BaseJSONError) {
+ return commonResponse, nil
+ })
+ })
+
+ defer webServer.stop()
+
+ respRaw, err := http.Post(
+ webServer.address(),
+ jsonMimeType,
+ bytes.NewBuffer(testCase.request),
+ )
+ if err != nil {
+ t.Fatalf("unexpected HTTP error, %v", err)
+ }
+
+ resp, err := io.ReadAll(respRaw.Body)
+ if err != nil {
+ t.Fatalf("unable to read response body, %v", err)
+ }
+
+ if err := testCase.verifyResponse(resp); err != nil {
+ t.Fatalf("unable to verify response, %v", err)
+ }
+ })
+ }
+}
+
+type testWebServer struct {
+ mux *chi.Mux
+ listener net.Listener
+}
+
+func newWebServer(t *testing.T, callbacks ...func(s *JSONRPC)) *testWebServer {
+ t.Helper()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatalf("unable to start listen, %v", err)
+ }
+
+ mux := chi.NewMux()
+ webServer := &testWebServer{
+ mux: mux,
+ listener: listener,
+ }
+
+ s := NewJSONRPC(WithLogger(zap.NewNop()))
+
+ for _, callback := range callbacks {
+ callback(s)
+ }
+
+ // Hook up the JSON-RPC server to the mux
+ mux.Mount("/", s.SetupRouter())
+
+ return webServer
+}
+
+func (ms *testWebServer) start() {
+ go func() {
+ //nolint:errcheck // No need to check error
+ _ = http.Serve(ms.listener, ms.mux)
+ }()
+}
+
+func (ms *testWebServer) stop() {
+ _ = ms.listener.Close()
+}
+
+func (ms *testWebServer) address() string {
+ return fmt.Sprintf("http://%s", ms.listener.Addr().String())
+}

serve/handlers/subs/filters/filter/base.go

@@ -0,0 +1,44 @@
+package filter
+
+import (
+ "sync"
+ "time"
+)
+
+// baseFilter defines the common properties
+// for all filter types
+type baseFilter struct {
+ lastUsed time.Time
+ filterType Type
+
+ sync.RWMutex
+}
+
+func newBaseFilter(filterType Type) *baseFilter {
+ return &baseFilter{
+ filterType: filterType,
+ lastUsed: time.Now(),
+ }
+}
+
+func (b *baseFilter) GetType() Type {
+ return b.filterType
+}
+
+func (b *baseFilter) GetLastUsed() time.Time {
+ b.RLock()
+ defer b.RUnlock()
+
+ return b.lastUsed
+}
+
+func (b *baseFilter) UpdateLastUsed() {
+ b.Lock()
+ defer b.Unlock()
+
+ b.lastUsed = time.Now()
+}
+
+func (b *baseFilter) GetChanges() any {
+ return nil
+}

serve/handlers/subs/filters/filter/base_test.go

@@ -0,0 +1,47 @@
+package filter
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestBaseFilter_GetType(t *testing.T) {
+ t.Parallel()
+
+ testTable := []struct {
+ name string
+ filterType Type
+ }{
+ {
+ "Block filter",
+ BlockFilterType,
+ },
+ }
+
+ for _, testCase := range testTable {
+ testCase := testCase
+
+ t.Run(testCase.name, func(t *testing.T) {
+ t.Parallel()
+
+ f := newBaseFilter(testCase.filterType)
+
+ assert.Equal(t, testCase.filterType, f.GetType())
+ assert.Nil(t, f.GetChanges())
+ })
+ }
+}
+
+func TestBaseFilter_LastUsed(t *testing.T) {
+ t.Parallel()
+
+ f := newBaseFilter(BlockFilterType)
+ lastUsed := f.GetLastUsed()
+
+ // Update the time it was last used
+ f.UpdateLastUsed()
+
+ // Make sure the last used time changed
+ assert.True(t, lastUsed.Before(f.GetLastUsed()))
+}

serve/handlers/subs/filters/filter/block.go

@@ -0,0 +1,43 @@
+package filter
+
+import "github.com/gnolang/tx-indexer/types"
+
+// BlockFilter type of filter for querying blocks
+type BlockFilter struct {
+ *baseFilter
+
+ blockHashes [][]byte // TODO keep as strings in 0x format?
+}
+
+// NewBlockFilter creates new block filter object
+func NewBlockFilter() *BlockFilter {
+ return &BlockFilter{
+ baseFilter: newBaseFilter(BlockFilterType),
+ blockHashes: make([][]byte, 0),
+ }
+}
+
+// GetChanges returns all new blocks from last query
+func (b *BlockFilter) GetChanges() any {
+ b.RLock()
+ defer b.RUnlock()
+
+ // Get hashes
+ hashes := b.blockHashes
+
+ // Empty hashes
+ b.blockHashes = b.blockHashes[:0]
+
+ return hashes
+}
+
+func (b *BlockFilter) UpdateWithBlock(block types.Block) {
+ b.Lock()
+ defer b.Unlock()
+
+ // Fetch block hash
+ hash := block.Hash()
+
+ // Add hash into block hash array
+ b.blockHashes = append(b.blockHashes, hash)
+}

serve/handlers/subs/filters/filter/block_test.go

@@ -0,0 +1,46 @@
+package filter
+
+import (
+ "testing"
+
+ "github.com/gnolang/tx-indexer/serve/handlers/subs/filters/mocks"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestBlockFilter_GetChanges(t *testing.T) {
+ t.Parallel()
+
+ // Generate dummy hashes
+ hashes := [][]byte{
+ []byte("hash 1"),
+ []byte("hash 2"),
+ []byte("hash 3"),
+ }
+
+ // Create a new block filter
+ f := NewBlockFilter()
+
+ // Make sure the filter is of a correct type
+ assert.Equal(t, BlockFilterType, f.GetType())
+
+ // Update the block filter with dummy blocks
+ for _, hash := range hashes {
+ hash := hash
+
+ f.UpdateWithBlock(&mocks.MockBlock{
+ HashFn: func() []byte {
+ return hash
+ },
+ })
+ }
+
+ // Get changes
+ changesRaw := f.GetChanges()
+
+ changes, ok := changesRaw.([][]byte)
+ require.True(t, ok)
+
+ // Make sure the hashes match
+ assert.Equal(t, hashes, changes)
+}

serve/handlers/subs/filters/filter/types.go

@@ -0,0 +1,7 @@
+package filter
+
+type Type string
+
+const (
+ BlockFilterType Type = "BlockFilter"
+)