Merge pull request #7359 from onflow/peter/7337-jsoncdc-encode-websockets-v0.40

[Access] Update websockets events and account status reponses to use json-cdc

Author: peterargue

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

@@ -83,12 +83,25 @@ func NewAccountStatusesDataProvider(
 func (p *AccountStatusesDataProvider) Run() error {
 	return run(
 		p.createAndStartSubscription(p.ctx, p.arguments),
-		func(response *backend.AccountStatusesResponse) error {
-			return p.sendResponse(response)
-		},
+		p.handleResponse,
 	)
 }
 
+// handleResponse processes the response from the subscription and sends it to the client's channel.
+// As part of the processing, it converts the event payloads from CCF to JSON-CDC format.
+// This function is not expected to be called concurrently.
+//
+// No errors expected during normal operations.
+func (p *AccountStatusesDataProvider) handleResponse(response *backend.AccountStatusesResponse) error {
+	// convert events to JSON-CDC format
+	convertedResponse, err := convertAccountStatusesResponse(response)
+	if err != nil {
+		return fmt.Errorf("failed to convert account status events to JSON-CDC format: %w", err)
+	}
+
+	return p.sendResponse(convertedResponse)
+}
+
 // sendResponse processes an account statuses message and sends it to data provider's channel.
 // This function is not safe to call concurrently.
 //
@@ -133,6 +146,27 @@ func (p *AccountStatusesDataProvider) createAndStartSubscription(
 	return p.stateStreamApi.SubscribeAccountStatusesFromLatestBlock(ctx, args.Filter)
 }
 
+// convertAccountStatusesResponse converts events in the provided AccountStatusesResponse from CCF
+// to JSON-CDC format.
+//
+// No errors expected during normal operations.
+func convertAccountStatusesResponse(resp *backend.AccountStatusesResponse) (*backend.AccountStatusesResponse, error) {
+	jsoncdcEvents := make(map[string]flow.EventsList, len(resp.AccountEvents))
+	for eventType, events := range resp.AccountEvents {
+		convertedEvents, err := convertEvents(events)
+		if err != nil {
+			return nil, fmt.Errorf("failed to convert %s events to JSON-CDC: %w", eventType, err)
+		}
+		jsoncdcEvents[eventType] = convertedEvents
+	}
+
+	return &backend.AccountStatusesResponse{
+		BlockID:       resp.BlockID,
+		Height:        resp.Height,
+		AccountEvents: jsoncdcEvents,
+	}, nil
+}
+
 // parseAccountStatusesArguments validates and initializes the account statuses arguments.
 func parseAccountStatusesArguments(
 	arguments wsmodels.Arguments,

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

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/onflow/flow/protobuf/go/flow/entities"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -18,8 +19,10 @@ import (
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
 	ssmock "github.com/onflow/flow-go/engine/access/state_stream/mock"
 	"github.com/onflow/flow-go/engine/access/subscription"
+	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/utils/unittest"
+	"github.com/onflow/flow-go/utils/unittest/generator"
 )
 
 // AccountStatusesProviderSuite is a test suite for testing the account statuses providers functionality.
@@ -66,11 +69,16 @@ func (s *AccountStatusesProviderSuite) SetupTest() {
 // validates that events are correctly streamed to the channel and ensures
 // no unexpected errors occur.
 func (s *AccountStatusesProviderSuite) TestAccountStatusesDataProvider_HappyPath() {
+	eventGenerator := generator.EventGenerator(generator.WithEncoding(entities.EventEncodingVersion_CCF_V0))
 	events := []flow.Event{
-		unittest.EventFixture(state_stream.CoreEventAccountCreated, 0, 0, unittest.IdentifierFixture(), 0),
-		unittest.EventFixture(state_stream.CoreEventAccountKeyAdded, 0, 0, unittest.IdentifierFixture(), 0),
+		eventGenerator.New(),
+		eventGenerator.New(),
 	}
 
+	// use account status events
+	events[0].Type = state_stream.CoreEventAccountCreated
+	events[1].Type = state_stream.CoreEventAccountKeyAdded
+
 	backendResponses := s.backendAccountStatusesResponses(events)
 
 	testHappyPath(
@@ -190,7 +198,25 @@ func (s *AccountStatusesProviderSuite) expectedAccountStatusesResponses(backendR
 	expectedResponses := make([]interface{}, len(backendResponses))
 
 	for i, resp := range backendResponses {
-		expectedResponsePayload := models.NewAccountStatusesResponse(resp, uint64(i))
+		// avoid updating the original response
+		expected := &backend.AccountStatusesResponse{
+			Height:        resp.Height,
+			BlockID:       resp.BlockID,
+			AccountEvents: make(map[string]flow.EventsList, len(resp.AccountEvents)),
+		}
+
+		// events are provided in CCF format, but we expect all event payloads in JSON-CDC format
+		for eventType, events := range resp.AccountEvents {
+			convertedEvents := make([]flow.Event, len(events))
+			for j, event := range events {
+				converted, err := convert.CcfEventToJsonEvent(event)
+				s.Require().NoError(err)
+				convertedEvents[j] = *converted
+			}
+			expected.AccountEvents[eventType] = convertedEvents
+		}
+
+		expectedResponsePayload := models.NewAccountStatusesResponse(expected, uint64(i))
 		expectedResponses[i] = &models.BaseDataProvidersResponse{
 			Topic:   AccountStatusesTopic,
 			Payload: expectedResponsePayload,

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

@@ -12,6 +12,7 @@ import (
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
 	"github.com/onflow/flow-go/engine/access/subscription"
+	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/counters"
 )
@@ -84,10 +85,25 @@ func NewEventsDataProvider(
 func (p *EventsDataProvider) Run() error {
 	return run(
 		p.createAndStartSubscription(p.ctx, p.arguments),
-		p.sendResponse,
+		p.handleResponse,
 	)
 }
 
+// handleResponse processes the response from the subscription and sends it to the client's channel.
+// As part of the processing, it converts the event payloads from CCF to JSON-CDC format.
+// This function is not expected to be called concurrently.
+//
+// No errors expected during normal operations.
+func (p *EventsDataProvider) handleResponse(response *backend.EventsResponse) error {
+	// convert events to JSON-CDC format
+	convertedResponse, err := convertEventsResponse(response)
+	if err != nil {
+		return fmt.Errorf("failed to convert events to JSON-CDC format: %w", err)
+	}
+
+	return p.sendResponse(convertedResponse)
+}
+
 // sendResponse processes an event message and sends it to client's channel.
 // This function is not expected to be called concurrently.
 //
@@ -129,6 +145,43 @@ func (p *EventsDataProvider) createAndStartSubscription(ctx context.Context, arg
 	return p.stateStreamApi.SubscribeEventsFromLatest(ctx, args.Filter)
 }
 
+// convertEventsResponse converts events in the provided EventsResponse from CCF to JSON-CDC format.
+//
+// No errors expected during normal operations.
+func convertEventsResponse(resp *backend.EventsResponse) (*backend.EventsResponse, error) {
+	jsoncdcEvents, err := convertEvents(resp.Events)
+	if err != nil {
+		return nil, fmt.Errorf("failed to convert events to JSON-CDC: %w", err)
+	}
+
+	return &backend.EventsResponse{
+		BlockID:        resp.BlockID,
+		Height:         resp.Height,
+		BlockTimestamp: resp.BlockTimestamp,
+		Events:         jsoncdcEvents,
+	}, nil
+}
+
+// convertEvents converts a slice events with CCF encoded payloads into a slice of new events who's
+// payloads are encoded in JSON-CDC format.
+//
+// Note: this function creates a copy of the original events before converting the payload. This
+// is important to ensure the original data structure is not modified, which could impact data held
+// in caches.
+//
+// No errors expected during normal operations.
+func convertEvents(ccfEvents []flow.Event) ([]flow.Event, error) {
+	jsoncdcEvents := make([]flow.Event, len(ccfEvents))
+	for i, ccfEvent := range ccfEvents {
+		converted, err := convert.CcfEventToJsonEvent(ccfEvent)
+		if err != nil {
+			return nil, fmt.Errorf("failed to convert event %d: %w", i, err)
+		}
+		jsoncdcEvents[i] = *converted
+	}
+	return jsoncdcEvents, nil
+}
+
 // parseEventsArguments validates and initializes the events arguments.
 func parseEventsArguments(
 	arguments wsmodels.Arguments,

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

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/onflow/flow/protobuf/go/flow/entities"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
@@ -17,8 +18,10 @@ import (
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
 	ssmock "github.com/onflow/flow-go/engine/access/state_stream/mock"
 	"github.com/onflow/flow-go/engine/access/subscription"
+	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/utils/unittest"
+	"github.com/onflow/flow-go/utils/unittest/generator"
 )
 
 // EventsProviderSuite is a test suite for testing the events providers functionality.
@@ -65,9 +68,10 @@ func (s *EventsProviderSuite) SetupTest() {
 // validates that events are correctly streamed to the channel and ensures
 // no unexpected errors occur.
 func (s *EventsProviderSuite) TestEventsDataProvider_HappyPath() {
+	eventGenerator := generator.EventGenerator(generator.WithEncoding(entities.EventEncodingVersion_CCF_V0))
 	events := []flow.Event{
-		unittest.EventFixture(flow.EventAccountCreated, 0, 0, unittest.IdentifierFixture(), 0),
-		unittest.EventFixture(flow.EventAccountUpdated, 0, 0, unittest.IdentifierFixture(), 0),
+		eventGenerator.New(),
+		eventGenerator.New(),
 	}
 
 	backendResponses := s.backendEventsResponses(events)
@@ -182,7 +186,23 @@ func (s *EventsProviderSuite) expectedEventsResponses(
 	expectedResponses := make([]interface{}, len(backendResponses))
 
 	for i, resp := range backendResponses {
-		expectedResponsePayload := models.NewEventResponse(resp, uint64(i))
+		// avoid updating the original response
+		expected := &backend.EventsResponse{
+			Height:         resp.Height,
+			BlockID:        resp.BlockID,
+			BlockTimestamp: resp.BlockTimestamp,
+			Events:         make(flow.EventsList, len(resp.Events)),
+		}
+
+		// events are provided in CCF format, but we expect all event payloads in JSON-CDC format
+		for i, event := range resp.Events {
+			converted, err := convert.CcfEventToJsonEvent(event)
+			s.Require().NoError(err)
+
+			expected.Events[i] = *converted
+		}
+
+		expectedResponsePayload := models.NewEventResponse(expected, uint64(i))
 		expectedResponses[i] = &models.BaseDataProvidersResponse{
 			Topic:   EventsTopic,
 			Payload: expectedResponsePayload,