func (s GlobalStateService) AccountDataStream(in *pb.AccountDataStreamRequest, stream pb.GlobalStateService_AccountDataStreamServer) error {
	log.Info("GRPC GlobalStateService.AccountDataStream")
	if in.Filter == nil {
		return status.Errorf(codes.InvalidArgument, "`Filter` must be provided")
	}
	if in.Filter.AccountId == nil {
		return status.Errorf(codes.InvalidArgument, "`Filter.AccountId` must be provided")
	}
	if in.Filter.AccountDataFlags == uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_UNSPECIFIED) {
		return status.Errorf(codes.InvalidArgument, "`Filter.AccountDataFlags` must set at least one bitfield")
	}
	addr := types.BytesToAddress(in.Filter.AccountId.Address)
	filterAccount := in.Filter.AccountDataFlags&uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_ACCOUNT) != 0
	filterReward := in.Filter.AccountDataFlags&uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_REWARD) != 0
	filterReceipt := in.Filter.AccountDataFlags&uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_TRANSACTION_RECEIPT) != 0
	var (
		channelAccount chan types.Address
		channelReward  chan events.Reward
		channelReceipt chan events.TxReceipt
	)
	if filterAccount {
		channelAccount = events.GetAccountChannel()
	}
	if filterReward {
		channelReward = events.GetRewardChannel()
	}
	if filterReceipt {
		channelReceipt = events.GetReceiptChannel()
	}
	for {
		select {
		case updatedAccount, ok := <-channelAccount:
			if !ok {
				log.Info("account channel closed, shutting down")
				return nil
			}
			if updatedAccount == addr {
				acct, err := s.getAccount(addr)
				if err != nil {
					log.With().Error("unable to fetch projected account state", log.Err(err))
					return status.Errorf(codes.Internal, "error fetching projected account data")
				}
				resp := &pb.AccountDataStreamResponse{Datum: &pb.AccountData{Datum: &pb.AccountData_AccountWrapper{
					AccountWrapper: acct,
				}}}
				if err := stream.Send(resp); err != nil {
					return fmt.Errorf("send to stream: %w", err)
				}
			}
		case reward, ok := <-channelReward:
			if !ok {
				log.Info("reward channel closed, shutting down")
				return nil
			}
			if reward.Coinbase == addr {
				resp := &pb.AccountDataStreamResponse{Datum: &pb.AccountData{Datum: &pb.AccountData_Reward{
					Reward: &pb.Reward{
						Layer:       &pb.LayerNumber{Number: reward.Layer.Uint32()},
						Total:       &pb.Amount{Value: reward.Total},
						LayerReward: &pb.Amount{Value: reward.LayerReward},
						Coinbase: &pb.AccountId{Address: addr.Bytes()},
						Smesher:  &pb.SmesherId{Id: reward.Smesher.ToBytes()},
					},
				}}}
				if err := stream.Send(resp); err != nil {
					return fmt.Errorf("send to stream: %w", err)
				}
			}
		case receipt, ok := <-channelReceipt:
			if !ok {
				log.Info("receipt channel closed, shutting down")
				return nil
			}
			if receipt.Address == addr {
				resp := &pb.AccountDataStreamResponse{Datum: &pb.AccountData{Datum: &pb.AccountData_Receipt{
					Receipt: &pb.TransactionReceipt{
						Id: &pb.TransactionId{Id: receipt.ID.Bytes()},
						GasUsed: receipt.GasUsed,
						Fee:     &pb.Amount{Value: receipt.Fee},
						Layer:   &pb.LayerNumber{Number: receipt.Layer.Uint32()},
						Index:   receipt.Index,
					},
				}}}
				if err := stream.Send(resp); err != nil {
					return fmt.Errorf("send to stream: %w", err)
				}
			}
		case <-stream.Context().Done():
			log.Info("AccountDataStream closing stream, client disconnected")
			return nil
		}
	}
}
STREAMS
AccountDataStream exposes a stream of account-related data.
package grpcserver
import (
	"bytes"
	"context"
	"fmt"
	pb "github.com/spacemeshos/api/release/go/spacemesh/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"github.com/spacemeshos/go-spacemesh/api"
	"github.com/spacemeshos/go-spacemesh/common/types"
	"github.com/spacemeshos/go-spacemesh/events"
	"github.com/spacemeshos/go-spacemesh/log"
)
type GlobalStateService struct {
	Mesh    api.TxAPI
	Mempool api.MempoolAPI
}
func (s GlobalStateService) RegisterService(server *Server) {
	pb.RegisterGlobalStateServiceServer(server.GrpcServer, s)
}
func NewGlobalStateService(tx api.TxAPI, mempool api.MempoolAPI) *GlobalStateService {
	return &GlobalStateService{
		Mesh:    tx,
		Mempool: mempool,
	}
}
func (s GlobalStateService) GlobalStateHash(context.Context, *pb.GlobalStateHashRequest) (*pb.GlobalStateHashResponse, error) {
	log.Info("GRPC GlobalStateService.GlobalStateHash")
	return &pb.GlobalStateHashResponse{Response: &pb.GlobalStateHash{
		RootHash: s.Mesh.GetStateRoot().Bytes(),
		Layer:    &pb.LayerNumber{Number: s.Mesh.LatestLayerInState().Uint32()},
	}}, nil
}
func (s GlobalStateService) getProjection(curCounter, curBalance uint64, addr types.Address) (counter, balance uint64, err error) {
	counter, balance, err = s.Mesh.GetProjection(addr, curCounter, curBalance)
	if err != nil {
		return 0, 0, fmt.Errorf("get mesh projection: %w", err)
	}
	counter, balance = s.Mempool.GetProjection(addr, counter, balance)
	return counter, balance, nil
}
func (s GlobalStateService) getAccount(addr types.Address) (acct *pb.Account, err error) {
	balanceActual := s.Mesh.GetBalance(addr)
	counterActual := s.Mesh.GetNonce(addr)
	counterProjected, balanceProjected, err := s.getProjection(counterActual, balanceActual, addr)
	if err != nil {
		return nil, err
	}
	return &pb.Account{
		AccountId: &pb.AccountId{Address: addr.Bytes()},
		StateCurrent: &pb.AccountState{
			Counter: counterActual,
			Balance: &pb.Amount{Value: balanceActual},
		},
		StateProjected: &pb.AccountState{
			Counter: counterProjected,
			Balance: &pb.Amount{Value: balanceProjected},
		},
	}, nil
}
func (s GlobalStateService) Account(_ context.Context, in *pb.AccountRequest) (*pb.AccountResponse, error) {
	log.Info("GRPC GlobalStateService.Account")
	if in.AccountId == nil {
		return nil, status.Errorf(codes.InvalidArgument, "`AccountId` must be provided")
	}
	addr := types.BytesToAddress(in.AccountId.Address)
	acct, err := s.getAccount(addr)
	if err != nil {
		log.With().Error("unable to fetch projected account state", log.Err(err))
		return nil, status.Errorf(codes.Internal, "error fetching projected account data")
	}
	log.With().Debug("GRPC GlobalStateService.Account",
		addr,
		log.Uint64("balance", acct.StateCurrent.Balance.Value),
		log.Uint64("counter", acct.StateCurrent.Counter),
		log.Uint64("balance projected", acct.StateProjected.Balance.Value),
		log.Uint64("counter projected", acct.StateProjected.Counter))
	return &pb.AccountResponse{AccountWrapper: acct}, nil
}
func (s GlobalStateService) AccountDataQuery(_ context.Context, in *pb.AccountDataQueryRequest) (*pb.AccountDataQueryResponse, error) {
	log.Info("GRPC GlobalStateService.AccountDataQuery")
	if in.Filter == nil {
		return nil, status.Errorf(codes.InvalidArgument, "`Filter` must be provided")
	}
	if in.Filter.AccountId == nil {
		return nil, status.Errorf(codes.InvalidArgument, "`Filter.AccountId` must be provided")
	}
	if in.Filter.AccountDataFlags == uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_UNSPECIFIED) {
		return nil, status.Errorf(codes.InvalidArgument, "`Filter.AccountMeshDataFlags` must set at least one bitfield")
	}
	_ = in.Filter.AccountDataFlags&uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_TRANSACTION_RECEIPT) != 0
	filterReward := in.Filter.AccountDataFlags&uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_REWARD) != 0
	filterAccount := in.Filter.AccountDataFlags&uint32(pb.AccountDataFlag_ACCOUNT_DATA_FLAG_ACCOUNT) != 0
	addr := types.BytesToAddress(in.Filter.AccountId.Address)
	res := &pb.AccountDataQueryResponse{}
	if filterReward {
		dbRewards, err := s.Mesh.GetRewards(addr)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "error getting rewards data")
		}
		for _, r := range dbRewards {
			res.AccountItem = append(res.AccountItem, &pb.AccountData{Datum: &pb.AccountData_Reward{
				Reward: &pb.Reward{
					Layer:       &pb.LayerNumber{Number: r.Layer.Uint32()},
					Total:       &pb.Amount{Value: r.TotalReward},
					LayerReward: &pb.Amount{Value: r.LayerRewardEstimate},
					Coinbase: &pb.AccountId{Address: addr.Bytes()},
					Smesher:  &pb.SmesherId{Id: r.SmesherID.ToBytes()},
				},
			}})
		}
	}
	if filterAccount {
		acct, err := s.getAccount(addr)
		if err != nil {
			log.With().Error("unable to fetch projected account state", log.Err(err))
			return nil, status.Errorf(codes.Internal, "error fetching projected account data")
		}
		res.AccountItem = append(res.AccountItem, &pb.AccountData{Datum: &pb.AccountData_AccountWrapper{
			AccountWrapper: acct,
		}})
	}
	res.TotalResults = uint32(len(res.AccountItem))
	offset := in.Offset
	if offset > uint32(len(res.AccountItem)) {
		return &pb.AccountDataQueryResponse{}, nil
	}
	maxResults := in.MaxResults
	if maxResults == 0 || offset+maxResults > uint32(len(res.AccountItem)) {
		maxResults = uint32(len(res.AccountItem)) - offset
	}
	res.AccountItem = res.AccountItem[offset : offset+maxResults]
	return res, nil
}
func (s GlobalStateService) SmesherDataQuery(_ context.Context, in *pb.SmesherDataQueryRequest) (*pb.SmesherDataQueryResponse, error) {
	log.Info("GRPC GlobalStateService.SmesherDataQuery")
	if in.SmesherId == nil {
		return nil, status.Errorf(codes.InvalidArgument, "`Id` must be provided")
	}
	if in.SmesherId.Id == nil {
		return nil, status.Errorf(codes.InvalidArgument, "`Id.Id` must be provided")
	}
	smesherIDBytes := in.SmesherId.Id
	smesherID, err := types.BytesToNodeID(smesherIDBytes)
	if err != nil {
		log.With().Error("unable to convert bytes to nodeID", log.Err(err))
		return nil, status.Errorf(codes.Internal, "error deserializing NodeID")
	}
	dbRewards, err := s.Mesh.GetRewardsBySmesherID(*smesherID)
	if err != nil {
		log.With().Error("unable to fetch projected reward state for smesher",
			log.FieldNamed("smesher_id", smesherID),
			log.Err(err))
		return nil, status.Errorf(codes.Internal, "error getting rewards data")
	}
	res := &pb.SmesherDataQueryResponse{}
	for _, r := range dbRewards {
		res.Rewards = append(res.Rewards, &pb.Reward{
			Layer:       &pb.LayerNumber{Number: r.Layer.Uint32()},
			Total:       &pb.Amount{Value: r.TotalReward},
			LayerReward: &pb.Amount{Value: r.LayerRewardEstimate},
			Coinbase: &pb.AccountId{Address: r.Coinbase.Bytes()},
			Smesher:  &pb.SmesherId{Id: r.SmesherID.ToBytes()},
		})
	}
	res.TotalResults = uint32(len(res.Rewards))
	offset := in.Offset
	if offset > uint32(len(res.Rewards)) {
		return &pb.SmesherDataQueryResponse{}, nil
	}
	maxResults := in.MaxResults
	if maxResults == 0 || offset+maxResults > uint32(len(res.Rewards)) {
		maxResults = uint32(len(res.Rewards)) - offset
	}
	res.Rewards = res.Rewards[offset : offset+maxResults]
	return res, nil
}