[Access] Add arguments getter for data providers

engine/access/rest/websockets/controller.go

@@ -462,6 +462,7 @@ func (c *Controller) handleListSubscriptions(ctx context.Context, _ models.ListS
 		subs = append(subs, &models.SubscriptionEntry{
 			SubscriptionID: id.String(),
 			Topic:          provider.Topic(),
+			Arguments:      provider.Arguments(),
 		})
 		return nil
 	})

engine/access/rest/websockets/controller_test.go

@@ -8,6 +8,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
+
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 	"github.com/rs/zerolog"
@@ -298,7 +300,6 @@ func (s *WsControllerSuite) TestUnsubscribeRequest() {
 				require.True(t, ok)
 				require.Empty(t, response.Error)
 				require.Equal(t, request.SubscriptionID, response.SubscriptionID)
-				require.Equal(t, request.SubscriptionID, response.SubscriptionID)
 
 				return websocket.ErrCloseSent
 			}).
@@ -470,7 +471,12 @@ func (s *WsControllerSuite) TestListSubscriptions() {
 		done := make(chan struct{})
 
 		topic := dp.BlocksTopic
+		arguments := models.Arguments{
+			"start_block_id": unittest.IdentifierFixture().String(),
+			"block_status":   parser.Finalized,
+		}
 		dataProvider.On("Topic").Return(topic)
+		dataProvider.On("Arguments").Return(arguments)
 		// data provider might finish on its own or controller will close it via Close()
 		dataProvider.On("Close").Return(nil).Maybe()
 		dataProvider.
@@ -514,6 +520,7 @@ func (s *WsControllerSuite) TestListSubscriptions() {
 				require.Equal(t, 1, len(response.Subscriptions))
 				require.Equal(t, subscriptionID, response.Subscriptions[0].SubscriptionID)
 				require.Equal(t, topic, response.Subscriptions[0].Topic)
+				require.Equal(t, arguments, response.Subscriptions[0].Arguments)
 				require.Equal(t, models.ListSubscriptionsAction, response.Action)
 
 				return websocket.ErrCloseSent

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

@@ -66,6 +66,7 @@ func NewAccountStatusesDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, accountStatusesArgs), // Set up a subscription to account statuses based on arguments.

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

@@ -3,13 +3,15 @@ package data_providers
 import (
 	"context"
 
+	"github.com/onflow/flow-go/engine/access/rest/websockets/models"
 	"github.com/onflow/flow-go/engine/access/subscription"
 )
 
 // baseDataProvider holds common objects for the provider
 type baseDataProvider struct {
 	subscriptionID string
 	topic          string
+	arguments      models.Arguments
 	cancel         context.CancelFunc
 	send           chan<- interface{}
 	subscription   subscription.Subscription
@@ -19,13 +21,15 @@ type baseDataProvider struct {
 func newBaseDataProvider(
 	subscriptionID string,
 	topic string,
+	arguments models.Arguments,
 	cancel context.CancelFunc,
 	send chan<- interface{},
 	subscription subscription.Subscription,
 ) *baseDataProvider {
 	return &baseDataProvider{
 		subscriptionID: subscriptionID,
 		topic:          topic,
+		arguments:      arguments,
 		cancel:         cancel,
 		send:           send,
 		subscription:   subscription,
@@ -42,6 +46,11 @@ func (b *baseDataProvider) Topic() string {
 	return b.topic
 }
 
+// Arguments returns the arguments associated with the data provider.
+func (b *baseDataProvider) Arguments() models.Arguments {
+	return b.arguments
+}
+
 // Close terminates the data provider.
 //
 // No errors are expected during normal operations.

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

@@ -48,6 +48,7 @@ func NewBlockDigestsDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, blockArgs), // Set up a subscription to block digests based on arguments.

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

@@ -49,6 +49,7 @@ func NewBlockHeadersDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, blockArgs), // Set up a subscription to block headers based on arguments.

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

@@ -63,6 +63,7 @@ func NewBlocksDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, p.arguments), // Set up a subscription to blocks based on arguments.

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

@@ -1,12 +1,16 @@
 package data_providers
 
+import "github.com/onflow/flow-go/engine/access/rest/websockets/models"
+
 // The DataProvider is the interface abstracts of the actual data provider used by the WebSocketCollector.
 // It provides methods for retrieving the provider's unique SubscriptionID, topic, and a methods to close and run the provider.
 type DataProvider interface {
 	// ID returns the unique identifier of the data provider.
 	ID() string
 	// Topic returns the topic associated with the data provider.
 	Topic() string
+	// Arguments returns the arguments associated with the data provider.
+	Arguments() models.Arguments
 	// Close terminates the data provider.
 	//
 	// No errors are expected during normal operations.

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

@@ -65,6 +65,7 @@ func NewEventsDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, eventArgs), // Set up a subscription to events based on arguments.

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

@@ -164,6 +164,7 @@ func (s *DataProviderFactorySuite) TestSupportedTopics() {
 			s.Require().NotNil(provider, "Expected provider for topic %s", test.topic)
 			s.Require().NoError(err, "Expected no error for topic %s", test.topic)
 			s.Require().Equal(test.topic, provider.Topic())
+			s.Require().Equal(test.arguments, provider.Arguments())
 
 			test.assertExpectations()
 		})

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

@@ -62,6 +62,7 @@ func NewSendAndGetTransactionStatusesDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, sendTxStatusesArgs), // Set up a subscription to tx statuses based on arguments.

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

@@ -65,6 +65,7 @@ func NewTransactionStatusesDataProvider(
 	p.baseDataProvider = newBaseDataProvider(
 		subscriptionID,
 		topic,
+		arguments,
 		cancel,
 		send,
 		p.createSubscription(subCtx, txStatusesArgs), // Set up a subscription to tx statuses based on arguments.

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

@@ -6,7 +6,7 @@ type Arguments map[string]interface{}
 type SubscribeMessageRequest struct {
 	BaseMessageRequest
 	Topic     string    `json:"topic"`     // Topic to subscribe to
-	Arguments Arguments `json:"arguments"` // Arguments are the arguments for the subscribed topic
+	Arguments Arguments `json:"arguments"` // Additional arguments for subscription
 }
 
 // SubscribeMessageResponse represents the response to a subscription request.