[Access] Add implementation for events data providers

engine/access/rest/server.go

@@ -51,7 +51,7 @@ func NewServer(serverAPI access.API,
 		builder.AddLegacyWebsocketsRoutes(stateStreamApi, chain, stateStreamConfig, config.MaxRequestSize)
 	}
 
-	dataProviderFactory := dp.NewDataProviderFactory(logger, stateStreamApi, serverAPI)
+	dataProviderFactory := dp.NewDataProviderFactory(logger, stateStreamApi, serverAPI, chain, stateStreamConfig.EventFilterConfig)
 	builder.AddWebsocketsRoute(chain, wsConfig, config.MaxRequestSize, dataProviderFactory)
 
 	c := cors.New(cors.Options{

engine/access/rest/websockets/data_providers/events_provider.go

@@ -0,0 +1,185 @@
+package data_providers
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/rs/zerolog"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
+	"github.com/onflow/flow-go/engine/access/rest/http/request"
+	"github.com/onflow/flow-go/engine/access/rest/util"
+	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
+	"github.com/onflow/flow-go/engine/access/state_stream"
+	"github.com/onflow/flow-go/engine/access/subscription"
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/counters"
+)
+
+// EventsArguments contains the arguments required for subscribing to events
+type EventsArguments struct {
+	StartBlockID     flow.Identifier          // ID of the block to start subscription from
+	StartBlockHeight uint64                   // Height of the block to start subscription from
+	Filter           state_stream.EventFilter // Filter applied to events for a given subscription
+}
+
+// EventsDataProvider is responsible for providing events
+type EventsDataProvider struct {
+	*BaseDataProviderImpl
+
+	logger         zerolog.Logger
+	args           EventsArguments
+	stateStreamApi state_stream.API
+}
+
+var _ DataProvider = (*EventsDataProvider)(nil)
+
+// NewEventsDataProvider creates a new instance of EventsDataProvider.
+func NewEventsDataProvider(
+	ctx context.Context,
+	logger zerolog.Logger,
+	stateStreamApi state_stream.API,
+	chain flow.Chain,
+	eventFilterConfig state_stream.EventFilterConfig,
+	topic string,
+	arguments map[string]string,
+	send chan<- interface{},
+) (*EventsDataProvider, error) {
+	p := &EventsDataProvider{
+		logger:         logger.With().Str("component", "events-data-provider").Logger(),
+		stateStreamApi: stateStreamApi,
+	}
+
+	// Initialize arguments passed to the provider.
+	var err error
+	p.args, err = ParseEventsArguments(arguments, chain, eventFilterConfig)
+	if err != nil {
+		return nil, fmt.Errorf("invalid arguments for events data provider: %w", err)
+	}
+
+	subCtx, cancel := context.WithCancel(ctx)
+
+	// Set up a subscription to events based on arguments.
+	sub := p.createSubscription(subCtx)
+
+	p.BaseDataProviderImpl = NewBaseDataProviderImpl(
+		topic,
+		cancel,
+		send,
+		sub,
+	)
+
+	return p, nil
+}
+
+// Run starts processing the subscription for events and handles responses.
+//
+// No errors are expected during normal operations.
+func (p *EventsDataProvider) Run() error {
+	return subscription.HandleSubscription(p.subscription, p.handleResponse(p.send))
+}
+
+// createSubscription creates a new subscription using the specified input arguments.
+func (p *EventsDataProvider) createSubscription(ctx context.Context) subscription.Subscription {
+	if p.args.StartBlockID != flow.ZeroID {
+		return p.stateStreamApi.SubscribeEventsFromStartBlockID(ctx, p.args.StartBlockID, p.args.Filter)
+	}
+
+	if p.args.StartBlockHeight != request.EmptyHeight {
+		return p.stateStreamApi.SubscribeEventsFromStartHeight(ctx, p.args.StartBlockHeight, p.args.Filter)
+	}
+
+	return p.stateStreamApi.SubscribeEventsFromLatest(ctx, p.args.Filter)
+}
+
+// handleResponse processes an event and sends the formatted response.
+//
+// No errors are expected during normal operations.
+func (p *EventsDataProvider) handleResponse(send chan<- interface{}) func(*flow.Event) error {
+	messageIndex := counters.NewMonotonousCounter(0)
+
+	return func(event *flow.Event) error {
+		if ok := messageIndex.Set(messageIndex.Value() + 1); !ok {
+			return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
+		}
+		index := messageIndex.Value()
+
+		send <- &models.EventResponse{
+			Event:        event,
+			MessageIndex: strconv.FormatUint(index, 10),
+		}
+
+		return nil
+	}
+}
+
+// ParseEventsArguments validates and initializes the events arguments.
+func ParseEventsArguments(
+	arguments map[string]string,
+	chain flow.Chain,
+	eventFilterConfig state_stream.EventFilterConfig,
+) (EventsArguments, error) {
+	var args EventsArguments
+
+	// Check for mutual exclusivity of start_block_id and start_block_height early
+	_, hasStartBlockID := arguments["start_block_id"]
+	_, hasStartBlockHeight := arguments["start_block_height"]
+
+	if hasStartBlockID && hasStartBlockHeight {
+		return args, fmt.Errorf("can only provide either 'start_block_id' or 'start_block_height'")
+	}
+
+	// Parse 'start_block_id' if provided
+	if hasStartBlockID {
+		var startBlockID parser.ID
+		err := startBlockID.Parse(arguments["start_block_id"])
+		if err != nil {
+			return args, fmt.Errorf("invalid 'start_block_id': %w", err)
+		}
+		args.StartBlockID = startBlockID.Flow()
+	} else {
+		args.StartBlockID = flow.ZeroID
+	}
+
+	// Parse 'start_block_height' if provided
+	if hasStartBlockHeight {
+		var err error
+		args.StartBlockHeight, err = util.ToUint64(arguments["start_block_height"])
+		if err != nil {
+			return args, fmt.Errorf("invalid 'start_block_height': %w", err)
+		}
+	} else {
+		args.StartBlockHeight = request.EmptyHeight
+	}
+
+	// Parse 'event_types' as []string{}
+	var eventTypes []string
+	if eventTypesIn, ok := arguments["event_types"]; ok && eventTypesIn != "" {
+		eventTypes = strings.Split(eventTypesIn, ",")
+	}
+
+	// Parse 'addresses' as []string{}
+	var addresses []string
+	if addressesIn, ok := arguments["addresses"]; ok && addressesIn != "" {
+		addresses = strings.Split(addressesIn, ",")
+	}
+
+	// Parse 'contracts' as []string{}
+	var contracts []string
+	if contractsIn, ok := arguments["contracts"]; ok && contractsIn != "" {
+		contracts = strings.Split(contractsIn, ",")
+	}
+
+	// Initialize the event filter with the parsed arguments
+	filter, err := state_stream.NewEventFilter(eventFilterConfig, chain, eventTypes, addresses, contracts)
+	if err != nil {
+		return args, fmt.Errorf("failed to create event filter: %w", err)
+	}
+	args.Filter = filter
+
+	return args, nil
+}

engine/access/rest/websockets/data_providers/events_provider_test.go

@@ -0,0 +1,181 @@
+package data_providers
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
+	"github.com/onflow/flow-go/engine/access/state_stream"
+	ssmock "github.com/onflow/flow-go/engine/access/state_stream/mock"
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// EventsProviderSuite is a test suite for testing the events providers functionality.
+type EventsProviderSuite struct {
+	suite.Suite
+
+	log zerolog.Logger
+	api *ssmock.API
+
+	chain          flow.Chain
+	rootBlock      flow.Block
+	finalizedBlock *flow.Header
+}
+
+func TestEventsProviderSuite(t *testing.T) {
+	suite.Run(t, new(EventsProviderSuite))
+}
+
+func (s *EventsProviderSuite) SetupTest() {
+	s.log = unittest.Logger()
+	s.api = ssmock.NewAPI(s.T())
+
+	s.chain = flow.Testnet.Chain()
+
+	s.rootBlock = unittest.BlockFixture()
+	s.rootBlock.Header.Height = 0
+}
+
+// invalidArgumentsTestCases returns a list of test cases with invalid argument combinations
+// for testing the behavior of events data providers. Each test case includes a name,
+// a set of input arguments, and the expected error message that should be returned.
+//
+// The test cases cover scenarios such as:
+// 1. Supplying both 'start_block_id' and 'start_block_height' simultaneously, which is not allowed.
+// 2. Providing invalid 'start_block_id' value.
+// 3. Providing invalid 'start_block_height' value.
+func (s *EventsProviderSuite) invalidArgumentsTestCases() []testErrType {
+	return []testErrType{
+		{
+			name: "provide both 'start_block_id' and 'start_block_height' arguments",
+			arguments: map[string]string{
+				"start_block_id":     s.rootBlock.ID().String(),
+				"start_block_height": fmt.Sprintf("%d", s.rootBlock.Header.Height),
+			},
+			expectedErrorMsg: "can only provide either 'start_block_id' or 'start_block_height'",
+		},
+		{
+			name: "invalid 'start_block_id' argument",
+			arguments: map[string]string{
+				"start_block_id": "invalid_block_id",
+			},
+			expectedErrorMsg: "invalid ID format",
+		},
+		{
+			name: "invalid 'start_block_height' argument",
+			arguments: map[string]string{
+				"start_block_height": "-1",
+			},
+			expectedErrorMsg: "value must be an unsigned 64 bit integer",
+		},
+	}
+}
+
+// TestEventsDataProvider_InvalidArguments tests the behavior of the event data provider
+// when invalid arguments are provided. It verifies that appropriate errors are returned
+// for missing or conflicting arguments.
+// This test covers the test cases:
+// 1. Providing both 'start_block_id' and 'start_block_height' simultaneously.
+// 2. Invalid 'start_block_id' argument.
+// 3. Invalid 'start_block_height' argument.
+func (s *EventsProviderSuite) TestEventsDataProvider_InvalidArguments() {
+	ctx := context.Background()
+	send := make(chan interface{})
+
+	topic := EventsTopic
+
+	for _, test := range s.invalidArgumentsTestCases() {
+		s.Run(test.name, func() {
+			provider, err := NewEventsDataProvider(
+				ctx,
+				s.log,
+				s.api,
+				s.chain,
+				state_stream.DefaultEventFilterConfig,
+				topic,
+				test.arguments,
+				send)
+			s.Require().Nil(provider)
+			s.Require().Error(err)
+			s.Require().Contains(err.Error(), test.expectedErrorMsg)
+		})
+	}
+}
+
+func (s *EventsProviderSuite) TestMessageIndexEventProviderResponse_HappyPath() {
+	ctx := context.Background()
+	send := make(chan interface{}, 10)
+	topic := EventsTopic
+	eventsCount := 4
+
+	// Create a channel to simulate the subscription's event channel
+	eventChan := make(chan interface{})
+
+	// Create a mock subscription and mock the channel
+	sub := ssmock.NewSubscription(s.T())
+	sub.On("Channel").Return((<-chan interface{})(eventChan))
+	sub.On("Err").Return(nil)
+
+	s.api.On("SubscribeEventsFromStartBlockID", mock.Anything, mock.Anything, mock.Anything).Return(sub)
+
+	arguments :=
+		map[string]string{
+			"start_block_id": s.rootBlock.ID().String(),
+		}
+
+	// Create the EventsDataProvider instance
+	provider, err := NewEventsDataProvider(
+		ctx,
+		s.log,
+		s.api,
+		s.chain,
+		state_stream.DefaultEventFilterConfig,
+		topic,
+		arguments,
+		send)
+	s.Require().NotNil(provider)
+	s.Require().NoError(err)
+
+	// Run the provider in a separate goroutine to simulate subscription processing
+	go func() {
+		err = provider.Run()
+		s.Require().NoError(err)
+	}()
+
+	// Simulate emitting events to the event channel
+	go func() {
+		defer close(eventChan) // Close the channel when done
+
+		for i := 0; i < eventsCount; i++ {
+			eventChan <- &flow.Event{
+				Type: "flow.AccountCreated",
+			}
+		}
+	}()
+
+	// Collect responses
+	var responses []*models.EventResponse
+	for i := 0; i < eventsCount; i++ {
+		res := <-send
+		eventRes, ok := res.(*models.EventResponse)
+		s.Require().True(ok, "Expected *models.EventResponse, got %T", res)
+		responses = append(responses, eventRes)
+	}
+
+	// Verifying that indices are strictly increasing
+	for i := 1; i < len(responses); i++ {
+		prevIndex, _ := strconv.Atoi(responses[i-1].MessageIndex)
+		currentIndex, _ := strconv.Atoi(responses[i].MessageIndex)
+		s.Require().Equal(prevIndex+1, currentIndex, "Expected MessageIndex to increment by 1")
+	}
+
+	// Ensure the provider is properly closed after the test
+	provider.Close()
+}