rpc/server: add new getalltransfertx API

It unifies UTXO and NEP5 transfers for a given address and presents it with
transaction-level grouping (and additional metadata).

pkg/rpc/client/rpc.go

@@ -28,6 +28,20 @@ func (c *Client) GetAccountState(address string) (*result.AccountState, error) {
 	return resp, nil
 }
 
+// GetAllTransferTx returns all transfer transactions for a given account within
+// specified timestamps (by block time) with specified output limits and page. It
+// only works with neo-go 0.78.0+ servers.
+func (c *Client) GetAllTransferTx(acc util.Uint160, start, end uint32, limit, page int) ([]result.TransferTx, error) {
+	var (
+		params = request.NewRawParams(acc.StringLE(), start, end, limit, page)
+		resp   = new([]result.TransferTx)
+	)
+	if err := c.performRequest("getalltransfertx", params, resp); err != nil {
+		return nil, err
+	}
+	return *resp, nil
+}
+
 // GetApplicationLog returns the contract log based on the specified txid.
 func (c *Client) GetApplicationLog(hash util.Uint256) (*result.ApplicationLog, error) {
 	var (

pkg/rpc/response/result/nep5.go

@@ -69,3 +69,24 @@ func (b *NEP5Balance) UnmarshalJSON(data []byte) error {
 	b.LastUpdated = s.LastUpdated
 	return nil
 }
+
+// TransferTx is a type used to represent and element of `getalltransfertx`
+// result. It combines transaction's inputs/outputs with NEP5 events.
+type TransferTx struct {
+	TxID       util.Uint256      `json:"txid"`
+	Timestamp  uint32            `json:"timestamp"`
+	Index      uint32            `json:"block_index"`
+	SystemFee  int64             `json:"sys_fee"`
+	NetworkFee int64             `json:"net_fee"`
+	Elements   []TransferTxEvent `json:"elements,omitempty"`
+	Events     []TransferTxEvent `json:"events,omitempty"`
+}
+
+// TransferTxEvent is an event used for elements or events of TransferTx, it's
+// either a single input/output, or a nep5 transfer.
+type TransferTxEvent struct {
+	Address string `json:"address"`
+	Type    string `json:"type"`
+	Value   string `json:"value"`
+	Asset   string `json:"asset"`
+}

pkg/rpc/server/server.go

@@ -81,6 +81,7 @@ const (
 
 var rpcHandlers = map[string]func(*Server, request.Params) (interface{}, *response.Error){
 	"getaccountstate":      (*Server).getAccountState,
+	"getalltransfertx":     (*Server).getAllTransferTx,
 	"getapplicationlog":    (*Server).getApplicationLog,
 	"getassetstate":        (*Server).getAssetState,
 	"getbestblockhash":     (*Server).getBestBlockHash,
@@ -828,6 +829,194 @@ func (s *Server) getNEP5Transfers(ps request.Params) (interface{}, *response.Err
 	return bs, nil
 }
 
+func (s *Server) getAllTransferTx(ps request.Params) (interface{}, *response.Error) {
+	var respErr *response.Error
+
+	u, err := ps.Value(0).GetUint160FromAddressOrHex()
+	if err != nil {
+		return nil, response.ErrInvalidParams
+	}
+
+	start, end, limit, page, err := getTimestampsAndLimit(ps, 1)
+	if err != nil {
+		return nil, response.NewInvalidParamsError("", err)
+	}
+
+	var (
+		utxoCont = make(chan bool)
+		nep5Cont = make(chan bool)
+		utxoTrs  = make(chan state.Transfer)
+		nep5Trs  = make(chan state.NEP5Transfer)
+	)
+
+	go func() {
+		tr := new(state.Transfer)
+		_ = s.chain.ForEachTransfer(u, tr, func() (bool, error) {
+			var cont bool
+
+			// Iterating from newest to oldest, not yet reached required
+			// time frame, continue looping.
+			if tr.Timestamp > end {
+				return true, nil
+			}
+			// Iterating from newest to oldest, moved past required
+			// time frame, stop looping.
+			if tr.Timestamp < start {
+				return false, nil
+			}
+			utxoTrs <- *tr
+			cont = <-utxoCont
+			return cont, nil
+		})
+		close(utxoTrs)
+	}()
+
+	go func() {
+		tr := new(state.NEP5Transfer)
+		_ = s.chain.ForEachNEP5Transfer(u, tr, func() (bool, error) {
+			var cont bool
+
+			// Iterating from newest to oldest, not yet reached required
+			// time frame, continue looping.
+			if tr.Timestamp > end {
+				return true, nil
+			}
+			// Iterating from newest to oldest, moved past required
+			// time frame, stop looping.
+			if tr.Timestamp < start {
+				return false, nil
+			}
+			nep5Trs <- *tr
+			cont = <-nep5Cont
+			return cont, nil
+		})
+		close(nep5Trs)
+	}()
+
+	var (
+		res                = make([]result.TransferTx, 0, limit)
+		frameCount         int
+		utxoLast           state.Transfer
+		nep5Last           state.NEP5Transfer
+		haveUtxo, haveNep5 bool
+	)
+
+	utxoLast, haveUtxo = <-utxoTrs
+	if haveUtxo {
+		utxoCont <- true
+	}
+	nep5Last, haveNep5 = <-nep5Trs
+	if haveNep5 {
+		nep5Cont <- true
+	}
+	for len(res) < limit {
+		if !haveUtxo && !haveNep5 {
+			break
+		}
+		var isNep5 = haveNep5 && (!haveUtxo || (nep5Last.Timestamp > utxoLast.Timestamp))
+		var transfer result.TransferTx
+		if isNep5 {
+			transfer.TxID = nep5Last.Tx
+			transfer.Timestamp = nep5Last.Timestamp
+			transfer.Index = nep5Last.Block
+		} else {
+			transfer.TxID = utxoLast.Tx
+			transfer.Timestamp = utxoLast.Timestamp
+			transfer.Index = utxoLast.Block
+		}
+		frameCount++
+		// Using limits, not yet reached required page. But still need
+		// to drain inputs for this tx.
+		skipTx := page*limit >= frameCount
+
+		if !skipTx {
+			tx, _, err := s.chain.GetTransaction(transfer.TxID)
+			if err != nil {
+				respErr = response.NewInternalServerError("invalid NEP5 transfer log", err)
+				break
+			}
+			transfer.NetworkFee = int64(s.chain.NetworkFee(tx))
+			transfer.SystemFee = int64(s.chain.SystemFee(tx))
+
+			inouts, err := s.chain.References(tx)
+			if err != nil {
+				respErr = response.NewInternalServerError("invalid tx", err)
+				break
+			}
+			for _, inout := range inouts {
+				var event result.TransferTxEvent
+
+				event.Address = address.Uint160ToString(inout.Out.ScriptHash)
+				event.Type = "input"
+				event.Value = inout.Out.Amount.String()
+				event.Asset = inout.Out.AssetID.StringLE()
+				transfer.Elements = append(transfer.Elements, event)
+			}
+			for _, out := range tx.Outputs {
+				var event result.TransferTxEvent
+
+				event.Address = address.Uint160ToString(out.ScriptHash)
+				event.Type = "output"
+				event.Value = out.Amount.String()
+				event.Asset = out.AssetID.StringLE()
+				transfer.Elements = append(transfer.Elements, event)
+			}
+		}
+		// Pick all NEP5 events for this transaction, if there are any.
+		for haveNep5 && nep5Last.Tx.Equals(transfer.TxID) {
+			if !skipTx {
+				var event result.TransferTxEvent
+				event.Asset = nep5Last.Asset.StringLE()
+				if nep5Last.Amount > 0 { // token was received
+					event.Value = strconv.FormatInt(nep5Last.Amount, 10)
+					event.Type = "receive"
+					if !nep5Last.From.Equals(util.Uint160{}) {
+						event.Address = address.Uint160ToString(nep5Last.From)
+					}
+				} else {
+					event.Value = strconv.FormatInt(-nep5Last.Amount, 10)
+					event.Type = "send"
+					if !nep5Last.To.Equals(util.Uint160{}) {
+						event.Address = address.Uint160ToString(nep5Last.To)
+					}
+				}
+				transfer.Events = append(transfer.Events, event)
+			}
+			nep5Last, haveNep5 = <-nep5Trs
+			if haveNep5 {
+				nep5Cont <- true
+			}
+		}
+
+		// Skip UTXO events, we've already got them from inputs and outputs.
+		for haveUtxo && utxoLast.Tx.Equals(transfer.TxID) {
+			utxoLast, haveUtxo = <-utxoTrs
+			if haveUtxo {
+				utxoCont <- true
+			}
+		}
+		if !skipTx {
+			res = append(res, transfer)
+		}
+	}
+	if haveUtxo {
+		_, ok := <-utxoTrs
+		if ok {
+			utxoCont <- false
+		}
+	}
+	if haveNep5 {
+		_, ok := <-nep5Trs
+		if ok {
+			nep5Cont <- false
+		}
+	}
+	if respErr != nil {
+		return nil, respErr
+	}
+	return res, nil
+}
+
 func (s *Server) getMinimumNetworkFee(ps request.Params) (interface{}, *response.Error) {
 	return s.chain.GetConfig().MinimumNetworkFee, nil
 }

pkg/rpc/server/server_test.go

@@ -1223,6 +1223,108 @@ func testRPCProtocol(t *testing.T, doRPCCall func(string, string, *testing.T) []
 		require.NoError(t, json.Unmarshal(res, actual))
 		checkNep5TransfersAux(t, e, actual, true)
 	})
+
+	t.Run("getalltransfertx", func(t *testing.T) {
+		testGetTxs := func(t *testing.T, asset string, start, stop, limit, page int, present []util.Uint256) {
+			ps := []string{`"AKkkumHbBipZ46UMZJoFynJMXzSRnBvKcs"`}
+			ps = append(ps, strconv.Itoa(start))
+			ps = append(ps, strconv.Itoa(stop))
+			if limit != 0 {
+				ps = append(ps, strconv.Itoa(limit))
+			}
+			if page != 0 {
+				ps = append(ps, strconv.Itoa(page))
+			}
+			p := strings.Join(ps, ", ")
+			rpc := fmt.Sprintf(`{"jsonrpc": "2.0", "id": 1, "method": "getalltransfertx", "params": [%s]}`, p)
+			body := doRPCCall(rpc, httpSrv.URL, t)
+			res := checkErrGetResult(t, body, false)
+			actualp := new([]result.TransferTx)
+			require.NoError(t, json.Unmarshal(res, actualp))
+			actual := *actualp
+			require.Equal(t, len(present), len(actual))
+			for _, id := range present {
+				var isThere bool
+				var ttx result.TransferTx
+				for i := range actual {
+					if id.Equals(actual[i].TxID) {
+						ttx = actual[i]
+						isThere = true
+						break
+					}
+				}
+				require.True(t, isThere)
+				tx, h, err := e.chain.GetTransaction(id)
+				require.NoError(t, err)
+				require.Equal(t, h, ttx.Index)
+				require.Equal(t, int64(e.chain.SystemFee(tx)), ttx.SystemFee)
+				require.Equal(t, int64(e.chain.NetworkFee(tx)), ttx.NetworkFee)
+				require.Equal(t, len(tx.Inputs)+len(tx.Outputs), len(ttx.Elements))
+			}
+		}
+		b, err := e.chain.GetBlock(e.chain.GetHeaderHash(1))
+		require.NoError(t, err)
+		txMoveNeo := b.Transactions[1].Hash()
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(202))
+		require.NoError(t, err)
+		txNeoRT := b.Transactions[1].Hash()
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(203))
+		require.NoError(t, err)
+		txGasClaim := b.Transactions[1].Hash()
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(204))
+		require.NoError(t, err)
+		txDeploy := b.Transactions[1].Hash()
+		ts204 := int(b.Timestamp)
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(206))
+		require.NoError(t, err)
+		txNeoTo1 := b.Transactions[1].Hash()
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(207))
+		require.NoError(t, err)
+		txNep5Tr := b.Transactions[2].Hash()
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(208))
+		require.NoError(t, err)
+		txNep5To1 := b.Transactions[1].Hash()
+		ts208 := int(b.Timestamp)
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(209))
+		require.NoError(t, err)
+		txMigrate := b.Transactions[1].Hash()
+		b, err = e.chain.GetBlock(e.chain.GetHeaderHash(210))
+		require.NoError(t, err)
+		txNep5To0 := b.Transactions[1].Hash()
+		lastTs := int(b.Timestamp)
+		t.Run("All", func(t *testing.T) {
+			testGetTxs(t, "", 0, lastTs, 0, 0, []util.Uint256{
+				txMoveNeo, txNeoRT, txGasClaim, txDeploy, txNeoTo1, txNep5Tr,
+				txNep5To1, txMigrate, txNep5To0,
+			})
+		})
+		t.Run("last 3", func(t *testing.T) {
+			testGetTxs(t, "", 0, lastTs, 3, 0, []util.Uint256{
+				txNep5To1, txMigrate, txNep5To0,
+			})
+		})
+		t.Run("3, page 1", func(t *testing.T) {
+			testGetTxs(t, "", 0, lastTs, 3, 1, []util.Uint256{
+				txDeploy, txNeoTo1, txNep5Tr,
+			})
+		})
+		t.Run("3, page 2", func(t *testing.T) {
+			testGetTxs(t, "", 0, lastTs, 3, 2, []util.Uint256{
+				txMoveNeo, txNeoRT, txGasClaim,
+			})
+		})
+		t.Run("3, page 3", func(t *testing.T) {
+			testGetTxs(t, "", 0, lastTs, 3, 3, []util.Uint256{})
+		})
+		t.Run("no dates", func(t *testing.T) {
+			testGetTxs(t, "", 0, 1000000, 0, 0, []util.Uint256{})
+		})
+		t.Run("204-208", func(t *testing.T) {
+			testGetTxs(t, "", ts204, ts208, 0, 0, []util.Uint256{
+				txDeploy, txNeoTo1, txNep5Tr, txNep5To1,
+			})
+		})
+	})
 }
 
 func (tc rpcTestCase) getResultPair(e *executor) (expected interface{}, res interface{}) {