[Access] Enable Event streaming on REST API

cmd/access/node_builder/access_node_builder.go

@@ -34,6 +34,7 @@ import (
 	"github.com/onflow/flow-go/consensus/hotstuff/verification"
 	recovery "github.com/onflow/flow-go/consensus/recovery/protocol"
 	"github.com/onflow/flow-go/crypto"
+	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/access/ingestion"
 	pingeng "github.com/onflow/flow-go/engine/access/ping"
 	"github.com/onflow/flow-go/engine/access/rest"
@@ -265,6 +266,8 @@ type FlowAccessNodeBuilder struct {
 	secureGrpcServer      *grpcserver.GrpcServer
 	unsecureGrpcServer    *grpcserver.GrpcServer
 	stateStreamGrpcServer *grpcserver.GrpcServer
+
+	stateStreamBackend *state_stream.StateStreamBackend
 }
 
 func (builder *FlowAccessNodeBuilder) buildFollowerState() *FlowAccessNodeBuilder {
@@ -445,7 +448,7 @@ func (builder *FlowAccessNodeBuilder) BuildConsensusFollower() *FlowAccessNodeBu
 	return builder
 }
 
-func (builder *FlowAccessNodeBuilder) BuildExecutionDataRequester() *FlowAccessNodeBuilder {
+func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccessNodeBuilder {
 	var ds *badger.Datastore
 	var bs network.BlobService
 	var processedBlockHeight storage.ConsumerProgress
@@ -625,20 +628,32 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionDataRequester() *FlowAccessN
 			if err != nil {
 				return nil, fmt.Errorf("could not get highest consecutive height: %w", err)
 			}
+			broadcaster := engine.NewBroadcaster()
 
-			stateStreamEng, err := state_stream.NewEng(
-				node.Logger,
+			builder.stateStreamBackend, err = state_stream.New(node.Logger,
 				builder.stateStreamConf,
-				builder.ExecutionDataStore,
-				executionDataCache,
 				node.State,
 				node.Storage.Headers,
 				node.Storage.Seals,
 				node.Storage.Results,
-				node.RootChainID,
+				builder.ExecutionDataStore,
+				executionDataCache,
+				broadcaster,
 				builder.executionDataConfig.InitialBlockHeight,
-				highestAvailableHeight,
+				highestAvailableHeight)
+			if err != nil {
+				return nil, fmt.Errorf("could not create state stream backend: %w", err)
+			}
+
+			stateStreamEng, err := state_stream.NewEng(
+				node.Logger,
+				builder.stateStreamConf,
+				executionDataCache,
+				node.Storage.Headers,
+				node.RootChainID,
 				builder.stateStreamGrpcServer,
+				builder.stateStreamBackend,
+				broadcaster,
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create state stream engine: %w", err)
@@ -898,6 +913,10 @@ func (builder *FlowAccessNodeBuilder) enqueueRelayNetwork() {
 }
 
 func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
+	if builder.executionDataSyncEnabled {
+		builder.BuildExecutionSyncComponents()
+	}
+
 	builder.
 		BuildConsensusFollower().
 		Module("collection node client", func(node *cmd.NodeConfig) error {
@@ -1103,6 +1122,9 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				nodeBackend,
 				builder.secureGrpcServer,
 				builder.unsecureGrpcServer,
+				builder.stateStreamBackend,
+				builder.stateStreamConf.EventFilterConfig,
+				builder.stateStreamConf.MaxGlobalStreams,
 			)
 			if err != nil {
 				return nil, err
@@ -1188,10 +1210,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		})
 	}
 
-	if builder.executionDataSyncEnabled {
-		builder.BuildExecutionDataRequester()
-	}
-
 	builder.Component("secure grpc server", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 		return builder.secureGrpcServer, nil
 	})

cmd/observer/node_builder/observer_builder.go

@@ -34,6 +34,7 @@ import (
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	rpcConnection "github.com/onflow/flow-go/engine/access/rpc/connection"
+	"github.com/onflow/flow-go/engine/access/state_stream"
 	"github.com/onflow/flow-go/engine/common/follower"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/protocol"
@@ -970,6 +971,9 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			restHandler,
 			builder.secureGrpcServer,
 			builder.unsecureGrpcServer,
+			nil, // state streaming is not supported
+			state_stream.DefaultEventFilterConfig,
+			0,
 		)
 		if err != nil {
 			return nil, err

engine/access/integration_unsecure_grpc_server_test.go

@@ -194,6 +194,9 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		bnd,
 		suite.secureGrpcServer,
 		suite.unsecureGrpcServer,
+		nil,
+		state_stream.DefaultEventFilterConfig,
+		0,
 	)
 	assert.NoError(suite.T(), err)
 	suite.rpcEng, err = rpcEngBuilder.WithLegacy().Build()
@@ -220,20 +223,31 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		ClientSendBufferSize: state_stream.DefaultSendBufferSize,
 	}
 
-	// create state stream engine
-	suite.stateStreamEng, err = state_stream.NewEng(
+	stateStreamBackend, err := state_stream.New(
 		suite.log,
 		conf,
-		nil,
-		suite.execDataCache,
 		suite.state,
 		suite.headers,
 		suite.seals,
 		suite.results,
-		suite.chainID,
+		nil,
+		suite.execDataCache,
+		nil,
 		rootBlock.Header.Height,
 		rootBlock.Header.Height,
+	)
+	assert.NoError(suite.T(), err)
+
+	// create state stream engine
+	suite.stateStreamEng, err = state_stream.NewEng(
+		suite.log,
+		conf,
+		suite.execDataCache,
+		suite.headers,
+		suite.chainID,
 		suite.unsecureGrpcServer,
+		stateStreamBackend,
+		nil,
 	)
 	assert.NoError(suite.T(), err)
 

engine/access/rest/middleware/logging.go

@@ -1,6 +1,9 @@
 package middleware
 
 import (
+	"bufio"
+	"fmt"
+	"net"
 	"net/http"
 	"time"
 
@@ -40,6 +43,9 @@ type responseWriter struct {
 	statusCode int
 }
 
+// http.Hijacker necessary for using middleware with gorilla websocket connections.
+var _ http.Hijacker = (*responseWriter)(nil)
+
 func newResponseWriter(w http.ResponseWriter) *responseWriter {
 	return &responseWriter{w, http.StatusOK}
 }
@@ -48,3 +54,11 @@ func (rw *responseWriter) WriteHeader(code int) {
 	rw.statusCode = code
 	rw.ResponseWriter.WriteHeader(code)
 }
+
+func (rw *responseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	hijacker, ok := rw.ResponseWriter.(http.Hijacker)
+	if !ok {
+		return nil, nil, fmt.Errorf("hijacking not supported")
+	}
+	return hijacker.Hijack()
+}

engine/access/rest/request/event_type.go

@@ -0,0 +1,54 @@
+package request
+
+import (
+	"fmt"
+	"regexp"
+)
+
+type EventType string
+
+var basicEventRe = regexp.MustCompile(`[A-Z]\.[a-f0-9]{16}\.[\w+]*\.[\w+]*`)
+var flowEventRe = regexp.MustCompile(`flow\.[\w]*`)
+
+func (e *EventType) Parse(raw string) error {
+	if !basicEventRe.MatchString(raw) && !flowEventRe.MatchString(raw) {
+		return fmt.Errorf("invalid event type format")
+	}
+	*e = EventType(raw)
+	return nil
+}
+
+func (e EventType) Flow() string {
+	return string(e)
+}
+
+type EventTypes []EventType
+
+func (e *EventTypes) Parse(raw []string) error {
+	// make a map to have only unique values as keys
+	eventTypes := make(EventTypes, 0)
+	uniqueTypes := make(map[string]bool)
+	for i, r := range raw {
+		var eType EventType
+		err := eType.Parse(r)
+		if err != nil {
+			return fmt.Errorf("error at index %d: %w", i, err)
+		}
+
+		if !uniqueTypes[eType.Flow()] {
+			uniqueTypes[eType.Flow()] = true
+			eventTypes = append(eventTypes, eType)
+		}
+	}
+
+	*e = eventTypes
+	return nil
+}
+
+func (e EventTypes) Flow() []string {
+	eventTypes := make([]string, len(e))
+	for j, eType := range e {
+		eventTypes[j] = eType.Flow()
+	}
+	return eventTypes
+}

engine/access/rest/request/get_events.go

@@ -2,7 +2,6 @@ package request
 
 import (
 	"fmt"
-	"regexp"
 
 	"github.com/onflow/flow-go/model/flow"
 )
@@ -57,19 +56,15 @@ func (g *GetEvents) Parse(rawType string, rawStart string, rawEnd string, rawBlo
 		return fmt.Errorf("must provide either block IDs or start and end height range")
 	}
 
-	g.Type = rawType
-	if g.Type == "" {
+	if rawType == "" {
 		return fmt.Errorf("event type must be provided")
 	}
-
-	// match basic format A.address.contract.event (ignore err since regex will always compile)
-	basic, _ := regexp.MatchString(`[A-Z]\.[a-f0-9]{16}\.[\w+]*\.[\w+]*`, g.Type)
-	// match core events flow.event
-	core, _ := regexp.MatchString(`flow\.[\w]*`, g.Type)
-
-	if !core && !basic {
-		return fmt.Errorf("invalid event type format")
+	var eventType EventType
+	err = eventType.Parse(rawType)
+	if err != nil {
+		return err
 	}
+	g.Type = eventType.Flow()
 
 	// validate start end height option
 	if g.StartHeight != EmptyHeight && g.EndHeight != EmptyHeight {

engine/access/rest/request/request.go

@@ -96,6 +96,12 @@ func (rd *Request) CreateTransactionRequest() (CreateTransaction, error) {
 	return req, err
 }
 
+func (rd *Request) SubscribeEventsRequest() (SubscribeEvents, error) {
+	var req SubscribeEvents
+	err := req.Build(rd)
+	return req, err
+}
+
 func (rd *Request) Expands(field string) bool {
 	return rd.ExpandFields[field]
 }

engine/access/rest/request/subscribe_events.go

@@ -0,0 +1,69 @@
+package request
+
+import (
+	"fmt"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+const startBlockIdQuery = "start_block_id"
+const eventTypesQuery = "event_types"
+const addressesQuery = "addresses"
+const contractsQuery = "contracts"
+
+type SubscribeEvents struct {
+	StartBlockID flow.Identifier
+	StartHeight  uint64
+
+	EventTypes []string
+	Addresses  []string
+	Contracts  []string
+}
+
+func (g *SubscribeEvents) Build(r *Request) error {
+	return g.Parse(
+		r.GetQueryParam(startBlockIdQuery),
+		r.GetQueryParam(startHeightQuery),
+		r.GetQueryParams(eventTypesQuery),
+		r.GetQueryParams(addressesQuery),
+		r.GetQueryParams(contractsQuery),
+	)
+}
+
+func (g *SubscribeEvents) Parse(rawStartBlockID string, rawStartHeight string, rawTypes []string, rawAddresses []string, rawContracts []string) error {
+	var startBlockID ID
+	err := startBlockID.Parse(rawStartBlockID)
+	if err != nil {
+		return err
+	}
+	g.StartBlockID = startBlockID.Flow()
+
+	var height Height
+	err = height.Parse(rawStartHeight)
+	if err != nil {
+		return fmt.Errorf("invalid start height: %w", err)
+	}
+	g.StartHeight = height.Flow()
+
+	// if both start_block_id and start_height are provided
+	if g.StartBlockID != flow.ZeroID && g.StartHeight != EmptyHeight {
+		return fmt.Errorf("can only provide either block ID or start height")
+	}
+
+	// default to root block
+	if g.StartHeight == EmptyHeight {
+		g.StartHeight = 0
+	}
+
+	var eventTypes EventTypes
+	err = eventTypes.Parse(rawTypes)
+	if err != nil {
+		return err
+	}
+
+	g.EventTypes = eventTypes.Flow()
+	g.Addresses = rawAddresses
+	g.Contracts = rawContracts
+
+	return nil
+}