[Access] Make response for data providers consistent.

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

@@ -116,11 +116,15 @@ func (p *AccountStatusesDataProvider) handleResponse() func(accountStatusesRespo
 			return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
 		}
 
-		p.send <- &models.AccountStatusesResponse{
-			BlockID:       accountStatusesResponse.BlockID.String(),
-			Height:        strconv.FormatUint(accountStatusesResponse.Height, 10),
-			AccountEvents: accountStatusesResponse.AccountEvents,
-			MessageIndex:  index,
+		p.send <- &models.BaseDataProvidersResponse{
+			SubscriptionID: p.ID().String(),
+			Topic:          p.Topic(),
+			Payload: &models.AccountStatusesResponse{
+				BlockID:       accountStatusesResponse.BlockID.String(),
+				Height:        strconv.FormatUint(accountStatusesResponse.Height, 10),
+				AccountEvents: accountStatusesResponse.AccountEvents,
+				MessageIndex:  index,
+			},
 		}
 
 		return nil

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

@@ -146,14 +146,17 @@ func (s *AccountStatusesProviderSuite) requireAccountStatuses(
 	expectedAccountStatusesResponse, ok := expectedResponse.(backend.AccountStatusesResponse)
 	require.True(s.T(), ok, "unexpected type: %T", expectedResponse)
 
-	actualResponse, ok := v.(*models.AccountStatusesResponse)
-	require.True(s.T(), ok, "Expected *models.AccountStatusesResponse, got %T", v)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
+	require.True(s.T(), ok, "Expected *models.BaseDataProvidersResponse, got %T", v)
 
-	require.Equal(s.T(), expectedAccountStatusesResponse.BlockID.String(), actualResponse.BlockID)
-	require.Equal(s.T(), len(expectedAccountStatusesResponse.AccountEvents), len(actualResponse.AccountEvents))
+	actualResponseData, ok := actualResponse.Payload.(*models.AccountStatusesResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
+
+	require.Equal(s.T(), expectedAccountStatusesResponse.BlockID.String(), actualResponseData.BlockID)
+	require.Equal(s.T(), len(expectedAccountStatusesResponse.AccountEvents), len(actualResponseData.AccountEvents))
 
 	for key, expectedEvents := range expectedAccountStatusesResponse.AccountEvents {
-		actualEvents, ok := actualResponse.AccountEvents[key]
+		actualEvents, ok := actualResponseData.AccountEvents[key]
 		require.True(s.T(), ok, "Missing key in actual AccountEvents: %s", key)
 
 		s.Require().Equal(expectedEvents, actualEvents, "Mismatch for key: %s", key)
@@ -252,9 +255,13 @@ func (s *AccountStatusesProviderSuite) TestMessageIndexAccountStatusesProviderRe
 	var responses []*models.AccountStatusesResponse
 	for i := 0; i < accountStatusesCount; i++ {
 		res := <-send
-		accountStatusesRes, ok := res.(*models.AccountStatusesResponse)
+		accountStatusesRes, ok := res.(*models.BaseDataProvidersResponse)
+		s.Require().True(ok, "Expected *models.BaseDataProvidersResponse, got %T", res)
+
+		accountStatusesResData, ok := accountStatusesRes.Payload.(*models.AccountStatusesResponse)
 		s.Require().True(ok, "Expected *models.AccountStatusesResponse, got %T", res)
-		responses = append(responses, accountStatusesRes)
+
+		responses = append(responses, accountStatusesResData)
 	}
 
 	// Verifying that indices are starting from 0

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

@@ -61,9 +61,14 @@ func (p *BlockDigestsDataProvider) Run() error {
 	return subscription.HandleSubscription(
 		p.subscription,
 		subscription.HandleResponse(p.send, func(block *flow.BlockDigest) (interface{}, error) {
-			return &models.BlockDigestMessageResponse{
-				Block: block,
+			return &models.BaseDataProvidersResponse{
+				SubscriptionID: p.ID().String(),
+				Topic:          p.Topic(),
+				Payload: &models.BlockDigestMessageResponse{
+					Block: block,
+				},
 			}, nil
+
 		}),
 	)
 }

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

@@ -126,10 +126,13 @@ func (s *BlocksProviderSuite) requireBlockDigest(v interface{}, expected interfa
 	expectedBlock, ok := expected.(*flow.Block)
 	require.True(s.T(), ok, "unexpected type: %T", v)
 
-	actualResponse, ok := v.(*models.BlockDigestMessageResponse)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
 	require.True(s.T(), ok, "unexpected response type: %T", v)
 
-	s.Require().Equal(expectedBlock.Header.ID(), actualResponse.Block.ID())
-	s.Require().Equal(expectedBlock.Header.Height, actualResponse.Block.Height)
-	s.Require().Equal(expectedBlock.Header.Timestamp, actualResponse.Block.Timestamp)
+	actualResponseData, ok := actualResponse.Payload.(*models.BlockDigestMessageResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
+
+	s.Require().Equal(expectedBlock.Header.ID(), actualResponseData.Block.ID())
+	s.Require().Equal(expectedBlock.Header.Height, actualResponseData.Block.Height)
+	s.Require().Equal(expectedBlock.Header.Timestamp, actualResponseData.Block.Timestamp)
 }

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

@@ -61,8 +61,12 @@ func (p *BlockHeadersDataProvider) Run() error {
 	return subscription.HandleSubscription(
 		p.subscription,
 		subscription.HandleResponse(p.send, func(header *flow.Header) (interface{}, error) {
-			return &models.BlockHeaderMessageResponse{
-				Header: header,
+			return &models.BaseDataProvidersResponse{
+				SubscriptionID: p.ID().String(),
+				Topic:          p.Topic(),
+				Payload: &models.BlockHeaderMessageResponse{
+					Header: header,
+				},
 			}, nil
 		}),
 	)

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

@@ -126,8 +126,11 @@ func (s *BlockHeadersProviderSuite) requireBlockHeader(v interface{}, expected i
 	expectedBlock, ok := expected.(*flow.Block)
 	require.True(s.T(), ok, "unexpected type: %T", v)
 
-	actualResponse, ok := v.(*models.BlockHeaderMessageResponse)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
 	require.True(s.T(), ok, "unexpected response type: %T", v)
 
-	s.Require().Equal(expectedBlock.Header, actualResponse.Header)
+	actualResponseData, ok := actualResponse.Payload.(*models.BlockHeaderMessageResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
+
+	s.Require().Equal(expectedBlock.Header, actualResponseData.Header)
 }

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

@@ -70,8 +70,12 @@ func (p *BlocksDataProvider) Run() error {
 	return subscription.HandleSubscription(
 		p.subscription,
 		subscription.HandleResponse(p.send, func(block *flow.Block) (interface{}, error) {
-			return &models.BlockMessageResponse{
-				Block: block,
+			return &models.BaseDataProvidersResponse{
+				SubscriptionID: p.ID().String(),
+				Topic:          p.Topic(),
+				Payload: &models.BlockMessageResponse{
+					Block: block,
+				},
 			}, nil
 		}),
 	)

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

@@ -203,8 +203,11 @@ func (s *BlocksProviderSuite) requireBlock(v interface{}, expected interface{})
 	expectedBlock, ok := expected.(*flow.Block)
 	require.True(s.T(), ok, "unexpected type: %T", v)
 
-	actualResponse, ok := v.(*models.BlockMessageResponse)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
 	require.True(s.T(), ok, "unexpected response type: %T", v)
 
-	s.Require().Equal(expectedBlock, actualResponse.Block)
+	actualResponseData, ok := actualResponse.Payload.(*models.BlockMessageResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
+
+	s.Require().Equal(expectedBlock, actualResponseData.Block)
 }

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

@@ -102,12 +102,16 @@ func (p *EventsDataProvider) handleResponse() func(eventsResponse *backend.Event
 			return fmt.Errorf("message index already incremented to: %d", messageIndex.Value())
 		}
 
-		p.send <- &models.EventResponse{
-			BlockId:        eventsResponse.BlockID.String(),
-			BlockHeight:    strconv.FormatUint(eventsResponse.Height, 10),
-			BlockTimestamp: eventsResponse.BlockTimestamp,
-			Events:         eventsResponse.Events,
-			MessageIndex:   index,
+		p.send <- &models.BaseDataProvidersResponse{
+			SubscriptionID: p.ID().String(),
+			Topic:          p.Topic(),
+			Payload: &models.EventResponse{
+				BlockId:        eventsResponse.BlockID.String(),
+				BlockHeight:    strconv.FormatUint(eventsResponse.Height, 10),
+				BlockTimestamp: eventsResponse.BlockTimestamp,
+				Events:         eventsResponse.Events,
+				MessageIndex:   index,
+			},
 		}
 
 		return nil

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

@@ -144,10 +144,13 @@ func (s *EventsProviderSuite) requireEvents(v interface{}, expectedResponse inte
 	expectedEventsResponse, ok := expectedResponse.(backend.EventsResponse)
 	require.True(s.T(), ok, "unexpected type: %T", expectedResponse)
 
-	actualResponse, ok := v.(*models.EventResponse)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
 	require.True(s.T(), ok, "Expected *models.EventResponse, got %T", v)
 
-	s.Require().ElementsMatch(expectedEventsResponse.Events, actualResponse.Events)
+	actualResponseData, ok := actualResponse.Payload.(*models.EventResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
+
+	s.Require().ElementsMatch(expectedEventsResponse.Events, actualResponseData.Events)
 }
 
 // invalidArgumentsTestCases returns a list of test cases with invalid argument combinations
@@ -279,9 +282,13 @@ func (s *EventsProviderSuite) TestMessageIndexEventProviderResponse_HappyPath()
 	var responses []*models.EventResponse
 	for i := 0; i < eventsCount; i++ {
 		res := <-send
-		eventRes, ok := res.(*models.EventResponse)
+		eventRes, ok := res.(*models.BaseDataProvidersResponse)
+		s.Require().True(ok, "Expected *models.BaseDataProvidersResponse, got %T", res)
+
+		eventResData, ok := eventRes.Payload.(*models.EventResponse)
 		s.Require().True(ok, "Expected *models.EventResponse, got %T", res)
-		responses = append(responses, eventRes)
+
+		responses = append(responses, eventResData)
 	}
 
 	// Verifying that indices are starting from 1

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

@@ -93,9 +93,13 @@ func (p *SendAndGetTransactionStatusesDataProvider) handleResponse() func(txResu
 				return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
 			}
 
-			p.send <- &models.TransactionStatusesResponse{
-				TransactionResult: txResults[i],
-				MessageIndex:      index,
+			p.send <- &models.BaseDataProvidersResponse{
+				SubscriptionID: p.ID().String(),
+				Topic:          p.Topic(),
+				Payload: &models.TransactionStatusesResponse{
+					TransactionResult: txResults[i],
+					MessageIndex:      index,
+				},
 			}
 		}
 

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

@@ -104,12 +104,14 @@ func (s *SendTransactionStatusesProviderSuite) requireTransactionStatuses(
 	expectedTxStatusesResponse, ok := expectedResponse.(*access.TransactionResult)
 	require.True(s.T(), ok, "unexpected type: %T", expectedResponse)
 
-	actualResponse, ok := v.(*models.TransactionStatusesResponse)
-	require.True(s.T(), ok, "Expected *models.TransactionStatusesResponse, got %T", v)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
+	require.True(s.T(), ok, "Expected *models.BaseDataProvidersResponse, got %T", v)
 
-	require.Equal(s.T(), expectedTxStatusesResponse.BlockID, actualResponse.TransactionResult.BlockID)
-	require.Equal(s.T(), expectedTxStatusesResponse.BlockHeight, actualResponse.TransactionResult.BlockHeight)
+	actualResponseData, ok := actualResponse.Payload.(*models.TransactionStatusesResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
 
+	require.Equal(s.T(), expectedTxStatusesResponse.BlockID, actualResponseData.TransactionResult.BlockID)
+	require.Equal(s.T(), expectedTxStatusesResponse.BlockHeight, actualResponseData.TransactionResult.BlockHeight)
 }
 
 // TestSendTransactionStatusesDataProvider_InvalidArguments tests the behavior of the send transaction statuses data provider

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

@@ -104,9 +104,13 @@ func (p *TransactionStatusesDataProvider) handleResponse() func(txResults []*acc
 				return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
 			}
 
-			p.send <- &models.TransactionStatusesResponse{
-				TransactionResult: txResults[i],
-				MessageIndex:      index,
+			p.send <- &models.BaseDataProvidersResponse{
+				SubscriptionID: p.ID().String(),
+				Topic:          p.Topic(),
+				Payload: &models.TransactionStatusesResponse{
+					TransactionResult: txResults[i],
+					MessageIndex:      index,
+				},
 			}
 		}
 

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

@@ -134,11 +134,14 @@ func (s *TransactionStatusesProviderSuite) requireTransactionStatuses(
 	expectedTxStatusesResponse, ok := expectedResponse.(*access.TransactionResult)
 	require.True(s.T(), ok, "unexpected type: %T", expectedResponse)
 
-	actualResponse, ok := v.(*models.TransactionStatusesResponse)
-	require.True(s.T(), ok, "Expected *models.TransactionStatusesResponse, got %T", v)
+	actualResponse, ok := v.(*models.BaseDataProvidersResponse)
+	require.True(s.T(), ok, "Expected *models.BaseDataProvidersResponse, got %T", v)
 
-	require.Equal(s.T(), expectedTxStatusesResponse.BlockID, actualResponse.TransactionResult.BlockID)
-	require.Equal(s.T(), expectedTxStatusesResponse.BlockHeight, actualResponse.TransactionResult.BlockHeight)
+	actualResponseData, ok := actualResponse.Payload.(*models.TransactionStatusesResponse)
+	require.True(s.T(), ok, "unexpected response data type: %T", v)
+
+	require.Equal(s.T(), expectedTxStatusesResponse.BlockID, actualResponseData.TransactionResult.BlockID)
+	require.Equal(s.T(), expectedTxStatusesResponse.BlockHeight, actualResponseData.TransactionResult.BlockHeight)
 }
 
 // TestTransactionStatusesDataProvider_InvalidArguments tests the behavior of the transaction statuses data provider
@@ -277,9 +280,14 @@ func (s *TransactionStatusesProviderSuite) TestMessageIndexTransactionStatusesPr
 	var responses []*models.TransactionStatusesResponse
 	for i := 0; i < txStatusesCount; i++ {
 		res := <-send
-		txStatusesRes, ok := res.(*models.TransactionStatusesResponse)
+
+		txStatusesRes, ok := res.(*models.BaseDataProvidersResponse)
+		s.Require().True(ok, "Expected *models.BaseDataProvidersResponse, got %T", res)
+
+		txStatusesResData, ok := txStatusesRes.Payload.(*models.TransactionStatusesResponse)
 		s.Require().True(ok, "Expected *models.TransactionStatusesResponse, got %T", res)
-		responses = append(responses, txStatusesRes)
+
+		responses = append(responses, txStatusesResData)
 	}
 
 	// Verifying that indices are starting from 0

engine/access/rest/websockets/models/base_message.go

@@ -19,3 +19,10 @@ type BaseMessageResponse struct {
 	Success         bool         `json:"success"`
 	Error           ErrorMessage `json:"error,omitempty"`
 }
+
+// BaseDataProvidersResponse represents a base structure for responses from subscriptions.
+type BaseDataProvidersResponse struct {
+	SubscriptionID string      `json:"subscription_id"` // Unique subscriptionID
+	Topic          string      `json:"topic"`           // Topic of the subscription
+	Payload        interface{} `json:"payload"`         // Payload that's being returned within a subscription.
+}