Feature: bulk request of block data (#136)

* Feature: bulk request of block data

* comments fix

go.mod

@@ -6,6 +6,7 @@ require (
 	cosmossdk.io/math v1.1.2
 	github.com/celestiaorg/celestia-app v1.0.0
 	github.com/cosmos/cosmos-sdk v0.46.14
+	github.com/cosmos/ibc-go/v6 v6.2.0
 	github.com/dipdup-io/workerpool v0.0.4
 	github.com/dipdup-net/go-lib v0.3.5
 	github.com/dipdup-net/indexer-sdk v0.0.3
@@ -75,7 +76,6 @@ require (
 	github.com/cosmos/gogoproto v1.4.11 // indirect
 	github.com/cosmos/gorocksdb v1.2.0 // indirect
 	github.com/cosmos/iavl v0.19.6 // indirect
-	github.com/cosmos/ibc-go/v6 v6.2.0 // indirect
 	github.com/cosmos/ledger-cosmos-go v0.13.1 // indirect
 	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/crate-crypto/go-kzg-4844 v0.3.0 // indirect

pkg/indexer/receiver/sync_test.go

@@ -5,14 +5,15 @@ package receiver
 
 import (
 	"context"
+	"sort"
+	"time"
+
 	ic "github.com/dipdup-io/celestia-indexer/pkg/indexer/config"
 	nodeTypes "github.com/dipdup-io/celestia-indexer/pkg/node/types"
 	"github.com/dipdup-io/celestia-indexer/pkg/types"
 	"github.com/dipdup-net/indexer-sdk/pkg/modules/stopper"
 	"github.com/pkg/errors"
 	"go.uber.org/mock/gomock"
-	"sort"
-	"time"
 )
 
 func (s *ModuleTestSuite) TestModule_SyncGracefullyStops() {
@@ -82,14 +83,13 @@ func (s *ModuleTestSuite) TestModule_SyncReadsBlocks() {
 
 		for i := types.Level(1); i <= blockCount; i++ {
 			s.api.EXPECT().
-				Block(gomock.Any(), i).
-				Return(getResultBlock(i), nil).
-				MaxTimes(1)
-
-			s.api.EXPECT().
-				BlockResults(gomock.Any(), i).
-				Return(getResultBlockResults(i), nil).
-				MaxTimes(1)
+				BlockData(gomock.Any(), i).
+				Return(types.BlockData{
+					ResultBlock:        getResultBlock(i),
+					ResultBlockResults: getResultBlockResults(i),
+				}, nil).
+				MaxTimes(1).
+				MinTimes(1)
 		}
 	})
 

pkg/indexer/receiver/worker.go

@@ -24,7 +24,7 @@ func (r *Module) worker(ctx context.Context, level types.Level) {
 		default:
 		}
 
-		block, err := r.blockData(ctx, level)
+		block, err := r.api.BlockData(ctx, level)
 		if err != nil {
 			if errors.Is(err, context.Canceled) {
 				return
@@ -47,17 +47,3 @@ func (r *Module) worker(ctx context.Context, level types.Level) {
 		Msg("received block")
 	r.blocks <- result
 }
-
-func (r *Module) blockData(ctx context.Context, level types.Level) (types.BlockData, error) {
-	block, err := r.api.Block(ctx, level)
-	if err != nil {
-		return types.BlockData{}, err
-	}
-
-	blockResults, err := r.api.BlockResults(ctx, level)
-	if err != nil {
-		return types.BlockData{}, err
-	}
-
-	return types.BlockData{ResultBlock: block, ResultBlockResults: blockResults}, nil
-}

pkg/node/api.go

@@ -18,6 +18,7 @@ type Api interface {
 	Block(ctx context.Context, level pkgTypes.Level) (pkgTypes.ResultBlock, error)
 	BlockResults(ctx context.Context, level pkgTypes.Level) (pkgTypes.ResultBlockResults, error)
 	Genesis(ctx context.Context) (types.Genesis, error)
+	BlockData(ctx context.Context, level pkgTypes.Level) (pkgTypes.BlockData, error)
 }
 
 //go:generate mockgen -source=$GOFILE -destination=mock/$GOFILE -package=mock -typed

pkg/node/rpc/api.go

@@ -4,12 +4,14 @@
 package rpc
 
 import (
+	"bytes"
 	"context"
 	"io"
 	"net/http"
 	"net/url"
 	"time"
 
+	"github.com/dipdup-io/celestia-indexer/pkg/node/types"
 	"github.com/goccy/go-json"
 
 	"github.com/pkg/errors"
@@ -50,7 +52,6 @@ func NewAPI(cfg config.DataSource) API {
 	}
 }
 
-// get -
 func (api *API) get(ctx context.Context, path string, args map[string]string, output any) error {
 	u, err := url.Parse(api.cfg.URL)
 	if err != nil {
@@ -99,6 +100,49 @@ func (api *API) get(ctx context.Context, path string, args map[string]string, ou
 	return err
 }
 
+func (api *API) post(ctx context.Context, requests []types.Request, output any) error {
+	u, err := url.Parse(api.cfg.URL)
+	if err != nil {
+		return err
+	}
+
+	body := new(bytes.Buffer)
+	if err := json.NewEncoder(body).Encode(requests); err != nil {
+		return errors.Wrap(err, "invalid bulk post request")
+	}
+
+	if api.rateLimit != nil {
+		if err := api.rateLimit.Wait(ctx); err != nil {
+			return err
+		}
+	}
+
+	start := time.Now()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), body)
+	if err != nil {
+		return err
+	}
+
+	response, err := api.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer closeWithLogError(response.Body, api.log)
+
+	api.log.Trace().
+		Int64("ms", time.Since(start).Milliseconds()).
+		Str("url", u.String()).
+		Msg("post request")
+
+	if response.StatusCode != http.StatusOK {
+		return errors.Errorf("invalid status: %d", response.StatusCode)
+	}
+
+	err = json.NewDecoder(response.Body).DecodeWithOption(output)
+	return err
+}
+
 func closeWithLogError(stream io.ReadCloser, log zerolog.Logger) {
 	if _, err := io.Copy(io.Discard, stream); err != nil {
 		log.Err(err).Msg("api copy GET body response to discard")