Add subscribe account statuses endpoint

access/grpc/client.go

@@ -360,3 +360,26 @@ func convertSubscribeOptions(opts ...access.SubscribeOption) *SubscribeConfig {
  }
  return subsConf
 }
+
+func (c *Client) SubscribeAccountStatusesFromStartHeight(
+ ctx context.Context,
+ startBlockHeight uint64,
+ filter flow.AccountStatusFilter,
+) (<-chan flow.AccountStatus, <-chan error, error) {
+ return c.grpc.SubscribeAccountStatusesFromStartHeight(ctx, startBlockHeight, filter)
+}
+
+func (c *Client) SubscribeAccountStatusesFromStartBlockID(
+ ctx context.Context,
+ startBlockID flow.Identifier,
+ filter flow.AccountStatusFilter,
+) (<-chan flow.AccountStatus, <-chan error, error) {
+ return c.grpc.SubscribeAccountStatusesFromStartBlockID(ctx, startBlockID, filter)
+}
+
+func (c *Client) SubscribeAccountStatusesFromLatestBlock(
+ ctx context.Context,
+ filter flow.AccountStatusFilter,
+) (<-chan flow.AccountStatus, <-chan error, error) {
+ return c.grpc.SubscribeAccountStatusesFromLatestBlock(ctx, filter)
+}

access/grpc/convert/convert.go

@@ -23,6 +23,7 @@ import (
  "fmt"
  "time"
 
+ "github.com/onflow/flow/protobuf/go/flow/executiondata"
  "google.golang.org/protobuf/types/known/timestamppb"
 
  "github.com/onflow/cadence"
@@ -76,6 +77,43 @@ func MessageToAccount(m *entities.Account) (flow.Account, error) {
  }, nil
 }
 
+func MessageToAccountStatus(m *executiondata.SubscribeAccountStatusesResponse) (flow.AccountStatus, error) {
+ if m == nil {
+ return flow.AccountStatus{}, ErrEmptyMessage
+ }
+
+ results, err := MessageToAccountStatusResults(m.GetResults())
+ if err != nil {
+ return flow.AccountStatus{}, fmt.Errorf("error converting results: %w", err)
+ }
+
+ return flow.AccountStatus{
+ BlockID: MessageToIdentifier(m.GetBlockId()),
+ BlockHeight: m.GetBlockHeight(),
+ MessageIndex: m.GetMessageIndex(),
+ Results: results,
+ }, nil
+}
+
+func MessageToAccountStatusResults(m []*executiondata.SubscribeAccountStatusesResponse_Result) ([]*flow.AccountStatusResult, error) {
+ results := make([]*flow.AccountStatusResult, len(m))
+ var emptyOptions []jsoncdc.Option
+
+ for i, r := range m {
+ events, err := MessagesToEvents(r.GetEvents(), emptyOptions)
+ if err != nil {
+ return nil, fmt.Errorf("error converting events: %w", err)
+ }
+
+ results[i] = &flow.AccountStatusResult{
+ Address: MessageToIdentifier(r.GetAddress()),
+ Events: events,
+ }
+ }
+
+ return results, nil
+}
+
 func AccountKeyToMessage(a *flow.AccountKey) *entities.AccountKey {
  return &entities.AccountKey{
  Index: uint32(a.Index),

access/grpc/grpc.go

@@ -26,6 +26,7 @@ import (
  "errors"
  "fmt"
  "io"
+ "sync/atomic"
 
  "github.com/onflow/flow/protobuf/go/flow/entities"
  "google.golang.org/grpc"
@@ -84,6 +85,7 @@ type BaseClient struct {
  close func() error
  jsonOptions []json.Option
  eventEncoding flow.EventEncodingVersion
+ messageIndex atomic.Uint64
 }
 
 // NewBaseClient creates a new gRPC handler for network communication.
@@ -1296,3 +1298,159 @@ func receiveBlocksFromClient[Client interface {
  }
  }
 }
+
+func (c *BaseClient) SubscribeAccountStatusesFromStartHeight(
+ ctx context.Context,
+ startHeight uint64,
+ filter flow.AccountStatusFilter,
+ opts ...grpc.CallOption,
+) (<-chan flow.AccountStatus, <-chan error, error) {
+ request := &executiondata.SubscribeAccountStatusesFromStartHeightRequest{
+ StartBlockHeight: startHeight,
+ EventEncodingVersion: c.eventEncoding,
+ }
+ request.Filter = &executiondata.StatusFilter{
+ EventType: filter.EventTypes,
+ Address: filter.Addresses,
+ }
+
+ subscribeClient, err := c.executionDataClient.SubscribeAccountStatusesFromStartHeight(ctx, request, opts...)
+ if err != nil {
+ return nil, nil, newRPCError(err)
+ }
+
+ accountStatutesChan := make(chan flow.AccountStatus)
+ errChan := make(chan error)
+
+ go func() {
+ defer close(accountStatutesChan)
+ defer close(errChan)
+ receiveAccountStatusesFromClient(ctx, subscribeClient, accountStatutesChan, errChan, &c.messageIndex)
+ }()
+
+ return accountStatutesChan, errChan, nil
+}
+
+func (c *BaseClient) SubscribeAccountStatusesFromStartBlockID(
+ ctx context.Context,
+ startBlockID flow.Identifier,
+ filter flow.AccountStatusFilter,
+ opts ...grpc.CallOption,
+) (<-chan flow.AccountStatus, <-chan error, error) {
+ request := &executiondata.SubscribeAccountStatusesFromStartBlockIDRequest{
+ StartBlockId: startBlockID.Bytes(),
+ EventEncodingVersion: c.eventEncoding,
+ }
+ request.Filter = &executiondata.StatusFilter{
+ EventType: filter.EventTypes,
+ Address: filter.Addresses,
+ }
+
+ subscribeClient, err := c.executionDataClient.SubscribeAccountStatusesFromStartBlockID(ctx, request, opts...)
+ if err != nil {
+ return nil, nil, newRPCError(err)
+ }
+
+ accountStatutesChan := make(chan flow.AccountStatus)
+ errChan := make(chan error)
+
+ go func() {
+ defer close(accountStatutesChan)
+ defer close(errChan)
+ receiveAccountStatusesFromClient(ctx, subscribeClient, accountStatutesChan, errChan, &c.messageIndex)
+ }()
+
+ return accountStatutesChan, errChan, nil
+}
+
+func (c *BaseClient) SubscribeAccountStatusesFromLatestBlock(
+ ctx context.Context,
+ filter flow.AccountStatusFilter,
+ opts ...grpc.CallOption,
+) (<-chan flow.AccountStatus, <-chan error, error) {
+ request := &executiondata.SubscribeAccountStatusesFromLatestBlockRequest{
+ EventEncodingVersion: c.eventEncoding,
+ }
+ request.Filter = &executiondata.StatusFilter{
+ EventType: filter.EventTypes,
+ Address: filter.Addresses,
+ }
+
+ subscribeClient, err := c.executionDataClient.SubscribeAccountStatusesFromLatestBlock(ctx, request, opts...)
+ if err != nil {
+ return nil, nil, newRPCError(err)
+ }
+
+ accountStatutesChan := make(chan flow.AccountStatus)
+ errChan := make(chan error)
+
+ go func() {
+ defer close(accountStatutesChan)
+ defer close(errChan)
+ receiveAccountStatusesFromClient(ctx, subscribeClient, accountStatutesChan, errChan, &c.messageIndex)
+ }()
+
+ return accountStatutesChan, errChan, nil
+}
+
+func receiveAccountStatusesFromClient[Client interface {
+ Recv() (*executiondata.SubscribeAccountStatusesResponse, error)
+}](
+ ctx context.Context,
+ client Client,
+ accountStatutesChan chan<- flow.AccountStatus,
+ errChan chan<- error,
+ previousMsgIndex *atomic.Uint64,
+) {
+ sendErr := func(err error) {
+ select {
+ case <-ctx.Done():
+ case errChan <- err:
+ }
+ }
+
+ for {
+ accountStatusResponse, err := client.Recv()
+ if err != nil {
+ if err == io.EOF {
+ // End of stream, return gracefully
+ return
+ }
+
+ sendErr(fmt.Errorf("error receiving account status: %w", err))
+ return
+ }
+
+ accountStatus, err := convert.MessageToAccountStatus(accountStatusResponse)
+ if err != nil {
+ sendErr(fmt.Errorf("error converting message to account status: %w", err))
+ return
+ }
+
+ if err = checkAndIncrementMessageIndex(previousMsgIndex, accountStatus.MessageIndex); err != nil {
+ sendErr(fmt.Errorf("error checking message index. messages are not ordered: %w", err))
+ return
+ }
+
+ select {
+ case <-ctx.Done():
+ return
+ case accountStatutesChan <- accountStatus:
+ }
+ }
+}
+
+func checkAndIncrementMessageIndex(previousMessageIndex *atomic.Uint64, currentMessageIndex uint64) error {
+ local := previousMessageIndex.Load()
+
+ if local == 0 && currentMessageIndex == 0 {
+ return nil
+ }
+
+ if currentMessageIndex != local+1 {
+ return errors.New("current message index is not exactly one larger than the previous one")
+ }
+
+ previousMessageIndex.Add(1)
+ return nil
+}