[Access] Add implementation for transaction statuses data providers

access/api.go

@@ -203,10 +203,50 @@ type API interface {
 	//
 	// If invalid parameters will be supplied SubscribeBlockDigestsFromLatest will return a failed subscription.
 	SubscribeBlockDigestsFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription
-	// SubscribeTransactionStatuses streams transaction statuses starting from the reference block saved in the
-	// transaction itself until the block containing the transaction becomes sealed or expired. When the transaction
-	// status becomes TransactionStatusSealed or TransactionStatusExpired, the subscription will automatically shut down.
-	SubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
+	// SubscribeTransactionStatusesFromStartBlockID subscribes to transaction status updates for a given transaction ID.
+	// Monitoring begins from the specified block ID. The subscription streams status updates until the transaction
+	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
+	// these final statuses, the subscription will automatically terminate.
+	//
+	// Parameters:
+	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
+	//   - txID: The identifier of the transaction to monitor.
+	//   - startBlockID: The block ID from which to start monitoring.
+	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
+	SubscribeTransactionStatusesFromStartBlockID(ctx context.Context, txID flow.Identifier, startBlockID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
+	// SubscribeTransactionStatusesFromStartHeight subscribes to transaction status updates for a given transaction ID.
+	// Monitoring begins from the specified block height. The subscription streams status updates until the transaction
+	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
+	// these final statuses, the subscription will automatically terminate.
+	//
+	// Parameters:
+	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
+	//   - txID: The unique identifier of the transaction to monitor.
+	//   - startHeight: The block height from which to start monitoring.
+	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
+	SubscribeTransactionStatusesFromStartHeight(ctx context.Context, txID flow.Identifier, startHeight uint64, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
+	// SubscribeTransactionStatusesFromLatest subscribes to transaction status updates for a given transaction ID.
+	// Monitoring begins from the latest block. The subscription streams status updates until the transaction
+	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
+	// these final statuses, the subscription will automatically terminate.
+	//
+	// Parameters:
+	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
+	//   - txID: The unique identifier of the transaction to monitor.
+	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
+	SubscribeTransactionStatusesFromLatest(ctx context.Context, txID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
+	// SendAndSubscribeTransactionStatuses sends a transaction to the execution node and subscribes to its status updates.
+	// Monitoring begins from the reference block saved in the transaction itself and streams status updates until the transaction
+	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). Once a final status is reached, the subscription
+	// automatically terminates.
+	//
+	// Parameters:
+	//   - ctx: The context to manage the transaction sending and subscription lifecycle, including cancellation.
+	//   - tx: The transaction body to be sent and monitored.
+	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
+	//
+	// If the transaction cannot be sent, the subscription will fail and return a failed subscription.
+	SendAndSubscribeTransactionStatuses(ctx context.Context, tx *flow.TransactionBody, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
 }
 
 // TODO: Combine this with flow.TransactionResult?

access/handler.go

@@ -1425,12 +1425,7 @@ func (h *Handler) SendAndSubscribeTransactionStatuses(
 		return status.Error(codes.InvalidArgument, err.Error())
 	}
 
-	err = h.api.SendTransaction(ctx, &tx)
-	if err != nil {
-		return err
-	}
-
-	sub := h.api.SubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion())
+	sub := h.api.SendAndSubscribeTransactionStatuses(ctx, &tx, request.GetEventEncodingVersion())
 
 	messageIndex := counters.NewMonotonousCounter(0)
 	return subscription.HandleRPCSubscription(sub, func(txResults []*TransactionResult) error {

cmd/util/cmd/run-script/cmd.go

@@ -532,7 +532,32 @@ func (*api) SubscribeBlockDigestsFromLatest(
 	return nil
 }
 
-func (*api) SubscribeTransactionStatuses(
+func (a *api) SubscribeTransactionStatusesFromStartBlockID(
+	_ context.Context,
+	_ flow.Identifier,
+	_ flow.Identifier,
+	_ entities.EventEncodingVersion,
+) subscription.Subscription {
+	return nil
+}
+
+func (a *api) SubscribeTransactionStatusesFromStartHeight(
+	_ context.Context,
+	_ flow.Identifier,
+	_ uint64,
+	_ entities.EventEncodingVersion,
+) subscription.Subscription {
+	return nil
+}
+
+func (a *api) SubscribeTransactionStatusesFromLatest(
+	_ context.Context,
+	_ flow.Identifier,
+	_ entities.EventEncodingVersion,
+) subscription.Subscription {
+	return nil
+}
+func (a *api) SendAndSubscribeTransactionStatuses(
 	_ context.Context,
 	_ *flow.TransactionBody,
 	_ entities.EventEncodingVersion,

engine/access/rest/http/request/event_type.go → engine/access/rest/common/parser/event_type.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/get_events.go

@@ -71,7 +71,7 @@ func (g *GetEvents) Parse(rawType string, rawStart string, rawEnd string, rawBlo
 	if rawType == "" {
 		return fmt.Errorf("event type must be provided")
 	}
-	var eventType EventType
+	var eventType parser.EventType
 	err = eventType.Parse(rawType)
 	if err != nil {
 		return err

engine/access/rest/server.go

@@ -51,7 +51,13 @@ 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,
+		stateStreamConfig.HeartbeatInterval)
 	builder.AddWebsocketsRoute(chain, wsConfig, config.MaxRequestSize, dataProviderFactory)
 
 	c := cors.New(cors.Options{