Merge pull request #6636 from The-K-R-O-K/UlyanaAndrukhiv/6585-block-data-provider

[Access] Add implementation BlockProvider

Author: peterargue

Makefile

@@ -203,7 +203,8 @@ generate-mocks: install-mock-generators
 	mockery --name 'API' --dir="./engine/protocol" --case=underscore --output="./engine/protocol/mock" --outpkg="mock"
 	mockery --name '.*' --dir="./engine/access/state_stream" --case=underscore --output="./engine/access/state_stream/mock" --outpkg="mock"
 	mockery --name 'BlockTracker' --dir="./engine/access/subscription" --case=underscore --output="./engine/access/subscription/mock"  --outpkg="mock"
-	mockery --name 'DataProvider' --dir="./engine/access/rest/websockets/data_provider" --case=underscore --output="./engine/access/rest/websockets/data_provider/mock"  --outpkg="mock"
+	mockery --name 'DataProvider' --dir="./engine/access/rest/websockets/data_providers" --case=underscore --output="./engine/access/rest/websockets/data_providers/mock"  --outpkg="mock"
+	mockery --name 'DataProviderFactory' --dir="./engine/access/rest/websockets/data_providers" --case=underscore --output="./engine/access/rest/websockets/data_providers/mock"  --outpkg="mock"
 	mockery --name 'ExecutionDataTracker' --dir="./engine/access/subscription" --case=underscore --output="./engine/access/subscription/mock"  --outpkg="mock"
 	mockery --name 'ConnectionFactory' --dir="./engine/access/rpc/connection" --case=underscore --output="./engine/access/rpc/connection/mock" --outpkg="mock"
 	mockery --name 'Communicator' --dir="./engine/access/rpc/backend" --case=underscore --output="./engine/access/rpc/backend/mock" --outpkg="mock"

access/handler.go

@@ -1066,7 +1066,7 @@ func (h *Handler) SubscribeBlocksFromStartBlockID(request *access.SubscribeBlock
 	}
 
 	sub := h.api.SubscribeBlocksFromStartBlockID(stream.Context(), startBlockID, blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlocksResponse(stream.Send, request.GetFullBlockResponse(), blockStatus))
+	return subscription.HandleRPCSubscription(sub, h.handleBlocksResponse(stream.Send, request.GetFullBlockResponse(), blockStatus))
 }
 
 // SubscribeBlocksFromStartHeight handles subscription requests for blocks started from block height.
@@ -1093,7 +1093,7 @@ func (h *Handler) SubscribeBlocksFromStartHeight(request *access.SubscribeBlocks
 	}
 
 	sub := h.api.SubscribeBlocksFromStartHeight(stream.Context(), request.GetStartBlockHeight(), blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlocksResponse(stream.Send, request.GetFullBlockResponse(), blockStatus))
+	return subscription.HandleRPCSubscription(sub, h.handleBlocksResponse(stream.Send, request.GetFullBlockResponse(), blockStatus))
 }
 
 // SubscribeBlocksFromLatest handles subscription requests for blocks started from latest sealed block.
@@ -1120,7 +1120,7 @@ func (h *Handler) SubscribeBlocksFromLatest(request *access.SubscribeBlocksFromL
 	}
 
 	sub := h.api.SubscribeBlocksFromLatest(stream.Context(), blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlocksResponse(stream.Send, request.GetFullBlockResponse(), blockStatus))
+	return subscription.HandleRPCSubscription(sub, h.handleBlocksResponse(stream.Send, request.GetFullBlockResponse(), blockStatus))
 }
 
 // handleBlocksResponse handles the subscription to block updates and sends
@@ -1179,7 +1179,7 @@ func (h *Handler) SubscribeBlockHeadersFromStartBlockID(request *access.Subscrib
 	}
 
 	sub := h.api.SubscribeBlockHeadersFromStartBlockID(stream.Context(), startBlockID, blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlockHeadersResponse(stream.Send))
+	return subscription.HandleRPCSubscription(sub, h.handleBlockHeadersResponse(stream.Send))
 }
 
 // SubscribeBlockHeadersFromStartHeight handles subscription requests for block headers started from block height.
@@ -1206,7 +1206,7 @@ func (h *Handler) SubscribeBlockHeadersFromStartHeight(request *access.Subscribe
 	}
 
 	sub := h.api.SubscribeBlockHeadersFromStartHeight(stream.Context(), request.GetStartBlockHeight(), blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlockHeadersResponse(stream.Send))
+	return subscription.HandleRPCSubscription(sub, h.handleBlockHeadersResponse(stream.Send))
 }
 
 // SubscribeBlockHeadersFromLatest handles subscription requests for block headers started from latest sealed block.
@@ -1233,7 +1233,7 @@ func (h *Handler) SubscribeBlockHeadersFromLatest(request *access.SubscribeBlock
 	}
 
 	sub := h.api.SubscribeBlockHeadersFromLatest(stream.Context(), blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlockHeadersResponse(stream.Send))
+	return subscription.HandleRPCSubscription(sub, h.handleBlockHeadersResponse(stream.Send))
 }
 
 // handleBlockHeadersResponse handles the subscription to block updates and sends
@@ -1293,7 +1293,7 @@ func (h *Handler) SubscribeBlockDigestsFromStartBlockID(request *access.Subscrib
 	}
 
 	sub := h.api.SubscribeBlockDigestsFromStartBlockID(stream.Context(), startBlockID, blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlockDigestsResponse(stream.Send))
+	return subscription.HandleRPCSubscription(sub, h.handleBlockDigestsResponse(stream.Send))
 }
 
 // SubscribeBlockDigestsFromStartHeight handles subscription requests for lightweight blocks started from block height.
@@ -1320,7 +1320,7 @@ func (h *Handler) SubscribeBlockDigestsFromStartHeight(request *access.Subscribe
 	}
 
 	sub := h.api.SubscribeBlockDigestsFromStartHeight(stream.Context(), request.GetStartBlockHeight(), blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlockDigestsResponse(stream.Send))
+	return subscription.HandleRPCSubscription(sub, h.handleBlockDigestsResponse(stream.Send))
 }
 
 // SubscribeBlockDigestsFromLatest handles subscription requests for lightweight block started from latest sealed block.
@@ -1347,7 +1347,7 @@ func (h *Handler) SubscribeBlockDigestsFromLatest(request *access.SubscribeBlock
 	}
 
 	sub := h.api.SubscribeBlockDigestsFromLatest(stream.Context(), blockStatus)
-	return subscription.HandleSubscription(sub, h.handleBlockDigestsResponse(stream.Send))
+	return subscription.HandleRPCSubscription(sub, h.handleBlockDigestsResponse(stream.Send))
 }
 
 // handleBlockDigestsResponse handles the subscription to block updates and sends
@@ -1433,7 +1433,7 @@ func (h *Handler) SendAndSubscribeTransactionStatuses(
 	sub := h.api.SubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion())
 
 	messageIndex := counters.NewMonotonousCounter(0)
-	return subscription.HandleSubscription(sub, func(txResults []*TransactionResult) error {
+	return subscription.HandleRPCSubscription(sub, func(txResults []*TransactionResult) error {
 		for i := range txResults {
 			index := messageIndex.Value()
 			if ok := messageIndex.Set(index + 1); !ok {

engine/access/rest/common/parser/block_status.go

@@ -0,0 +1,24 @@
+package parser
+
+import (
+	"fmt"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// Finalized and Sealed represents the status of a block.
+// It is used in rest arguments to provide block status.
+const (
+	Finalized = "finalized"
+	Sealed    = "sealed"
+)
+
+func ParseBlockStatus(blockStatus string) (flow.BlockStatus, error) {
+	switch blockStatus {
+	case Finalized:
+		return flow.BlockStatusFinalized, nil
+	case Sealed:
+		return flow.BlockStatusSealed, nil
+	}
+	return flow.BlockStatusUnknown, fmt.Errorf("invalid 'block_status', must be '%s' or '%s'", Finalized, Sealed)
+}

engine/access/rest/common/parser/block_status_test.go

@@ -0,0 +1,39 @@
+package parser
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// TestParseBlockStatus_Invalid tests the ParseBlockStatus function with invalid inputs.
+// It verifies that for each invalid block status string, the function returns an error
+// matching the expected error message format.
+func TestParseBlockStatus_Invalid(t *testing.T) {
+	tests := []string{"unknown", "pending", ""}
+	expectedErr := fmt.Sprintf("invalid 'block_status', must be '%s' or '%s'", Finalized, Sealed)
+
+	for _, input := range tests {
+		_, err := ParseBlockStatus(input)
+		assert.EqualError(t, err, expectedErr)
+	}
+}
+
+// TestParseBlockStatus_Valid tests the ParseBlockStatus function with valid inputs.
+// It ensures that the function returns the correct flow.BlockStatus for valid status
+// strings "finalized" and "sealed" without errors.
+func TestParseBlockStatus_Valid(t *testing.T) {
+	tests := map[string]flow.BlockStatus{
+		Finalized: flow.BlockStatusFinalized,
+		Sealed:    flow.BlockStatusSealed,
+	}
+
+	for input, expectedStatus := range tests {
+		status, err := ParseBlockStatus(input)
+		assert.NoError(t, err)
+		assert.Equal(t, expectedStatus, status)
+	}
+}

engine/access/rest/http/request/id.go engine/access/rest/common/parser/id.goengine/access/rest/http/request/id.go renamed to engine/access/rest/common/parser/id.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"errors"

…gine/access/rest/http/request/id_test.go …ine/access/rest/common/parser/id_test.goengine/access/rest/http/request/id_test.go renamed to engine/access/rest/common/parser/id_test.go engine/access/rest/http/request/id_test.go renamed to engine/access/rest/common/parser/id_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"testing"

engine/access/rest/http/request/get_block.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -122,7 +123,7 @@ func (g *GetBlockByIDs) Build(r *common.Request) error {
 }
 
 func (g *GetBlockByIDs) Parse(rawIds []string) error {
-	var ids IDs
+	var ids parser.IDs
 	err := ids.Parse(rawIds)
 	if err != nil {
 		return err

engine/access/rest/http/request/get_events.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -50,7 +51,7 @@ func (g *GetEvents) Parse(rawType string, rawStart string, rawEnd string, rawBlo
 	}
 	g.EndHeight = height.Flow()
 
-	var blockIDs IDs
+	var blockIDs parser.IDs
 	err = blockIDs.Parse(rawBlockIDs)
 	if err != nil {
 		return err

engine/access/rest/http/request/get_execution_result.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -30,7 +31,7 @@ func (g *GetExecutionResultByBlockIDs) Build(r *common.Request) error {
 }
 
 func (g *GetExecutionResultByBlockIDs) Parse(rawIDs []string) error {
-	var ids IDs
+	var ids parser.IDs
 	err := ids.Parse(rawIDs)
 	if err != nil {
 		return err

engine/access/rest/http/request/get_script.go

@@ -5,6 +5,7 @@ import (
 	"io"
 
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -42,7 +43,7 @@ func (g *GetScript) Parse(rawHeight string, rawID string, rawScript io.Reader) e
 	}
 	g.BlockHeight = height.Flow()
 
-	var id ID
+	var id parser.ID
 	err = id.Parse(rawID)
 	if err != nil {
 		return err