NODE-2617 Return asset distribution route (#3893)

Author: ivan-mashonskiy

node-it/src/test/scala/com/wavesplatform/it/api/AsyncHttpApi.scala

@@ -23,7 +23,7 @@ import com.wavesplatform.lang.v1.FunctionHeader
 import com.wavesplatform.lang.v1.compiler.Terms
 import com.wavesplatform.lang.v1.compiler.Terms.FUNCTION_CALL
 import com.wavesplatform.state.DataEntry.Format
-import com.wavesplatform.state.{AssetDistributionPage, DataEntry, EmptyDataEntry, LeaseBalance, Portfolio}
+import com.wavesplatform.state.{AssetDistribution, AssetDistributionPage, DataEntry, EmptyDataEntry, LeaseBalance, Portfolio}
 import com.wavesplatform.transaction.Asset.{IssuedAsset, Waves}
 import com.wavesplatform.transaction.assets.*
 import com.wavesplatform.transaction.assets.exchange.{Order, ExchangeTransaction as ExchangeTx}
@@ -338,6 +338,11 @@ object AsyncHttpApi extends Assertions {
       get(url, amountsAsStrings).as[AssetDistributionPage](amountsAsStrings)
     }
 
+    def assetDistribution(asset: String, amountsAsStrings: Boolean = false): Future[AssetDistribution] = {
+      val req = s"/assets/$asset/distribution"
+      get(req, amountsAsStrings).as[AssetDistribution](amountsAsStrings)
+    }
+
     def effectiveBalance(address: String, confirmations: Option[Int] = None, amountsAsStrings: Boolean = false): Future[Balance] = {
       val maybeConfirmations = confirmations.fold("")(a => s"/$a")
       get(s"/addresses/effectiveBalance/$address$maybeConfirmations", amountsAsStrings).as[Balance](amountsAsStrings)

node-it/src/test/scala/com/wavesplatform/it/api/SyncHttpApi.scala

@@ -14,7 +14,7 @@ import com.wavesplatform.it.Node
 import com.wavesplatform.it.sync.*
 import com.wavesplatform.lang.script.v1.ExprScript
 import com.wavesplatform.lang.v1.compiler.Terms
-import com.wavesplatform.state.{AssetDistributionPage, DataEntry}
+import com.wavesplatform.state.{AssetDistribution, AssetDistributionPage, DataEntry}
 import com.wavesplatform.transaction.assets.exchange.Order
 import com.wavesplatform.transaction.lease.{LeaseCancelTransaction, LeaseTransaction}
 import com.wavesplatform.transaction.smart.InvokeScriptTransaction
@@ -261,6 +261,9 @@ object SyncHttpApi extends Assertions with matchers.should.Matchers {
     ): AssetDistributionPage =
       sync(async(n).assetDistributionAtHeight(asset, height, limit, maybeAfter, amountsAsStrings))
 
+    def assetDistribution(asset: String): AssetDistribution =
+      sync(async(n).assetDistribution(asset))
+
     def broadcastIssue(
         source: KeyPair,
         name: String,

node-it/src/test/scala/com/wavesplatform/it/asset/IssueReissueBurnAssetSuite.scala

@@ -250,15 +250,15 @@ class IssueReissueBurnAssetSuite extends BaseFreeSpec {
       val acc   = createDapp(script(simpleReissuableAsset))
       val asset = issueValidated(acc, simpleReissuableAsset)
       invokeScript(acc, "transferAndBurn", assetId = asset, count = 100)
-      val height1 = nodes.waitForHeightArise()
-      sender.assetDistributionAtHeight(asset, height1 - 1, 10).items.map { case (a, v) => a.toString -> v } shouldBe Map(
+      nodes.waitForHeightArise()
+      sender.assetDistribution(asset).map { case (a, v) => a.toString -> v } shouldBe Map(
         miner.address          -> 100L,
         acc.toAddress.toString -> (simpleReissuableAsset.quantity - 200)
       )
       reissue(acc, CallableMethod, asset, 400, reissuable = false)
       invokeScript(acc, "transferAndBurn", assetId = asset, count = 100)
-      val height2 = nodes.waitForHeightArise()
-      sender.assetDistributionAtHeight(asset, height2 - 1, 10).items.map { case (a, v) => a.toString -> v } shouldBe Map(
+      nodes.waitForHeightArise()
+      sender.assetDistribution(asset).map { case (a, v) => a.toString -> v } shouldBe Map(
         miner.address          -> 200L,
         acc.toAddress.toString -> simpleReissuableAsset.quantity
       )

node-it/src/test/scala/com/wavesplatform/it/sync/AssetDistributionSuite.scala

@@ -8,6 +8,8 @@ import com.wavesplatform.state.AssetDistributionPage
 import com.wavesplatform.transaction.transfer.MassTransferTransaction
 import org.scalatest.CancelAfterFailure
 
+import scala.concurrent.duration.*
+
 class AssetDistributionSuite extends BaseTransactionSuite with CancelAfterFailure {
 
   lazy val node: Node = nodes.head
@@ -23,7 +25,7 @@ class AssetDistributionSuite extends BaseTransactionSuite with CancelAfterFailur
 
     nodes.waitForHeightArise()
 
-    val issueTx = node.issue(issuer, "TestCoin", "no description", issueAmount, 8, false, issueFee, waitForTx = true).id
+    val issueTx = node.issue(issuer, "TestCoin", "no description", issueAmount, 8, reissuable = false, issueFee, waitForTx = true).id
 
     node.massTransfer(
       issuer,
@@ -47,6 +49,8 @@ class AssetDistributionSuite extends BaseTransactionSuite with CancelAfterFailur
 
     val issuerAssetDis = assetDis.view.filterKeys(_ == issuer.toAddress).values
 
+    assetDis should be equals node.assetDistribution(issueTx)
+
     issuerAssetDis.size shouldBe 1
     issuerAssetDis.head shouldBe (issueAmount - addresses.length * transferAmount)
 
@@ -68,10 +72,34 @@ class AssetDistributionSuite extends BaseTransactionSuite with CancelAfterFailur
     )
   }
 
+  test("'Asset distribution' works properly") {
+    val receivers = for (i <- 0 until 10) yield KeyPair(s"receiver#$i".getBytes("UTF-8"))
+
+    val issueTx = node.issue(issuer, "TestCoin#2", "no description", issueAmount, 8, reissuable = false, issueFee, waitForTx = true).id
+
+    node
+      .massTransfer(
+        issuer,
+        receivers.map(rc => MassTransferTransaction.Transfer(rc.toAddress.toString, 10)).toList,
+        minFee + minFee * receivers.length,
+        assetId = Some(issueTx),
+        waitForTx = true
+      )
+
+    nodes.waitForHeightArise()
+
+    val distribution = node.assetDistribution(issueTx)
+
+    distribution.size shouldBe (receivers.size + 1)
+    distribution(issuer.toAddress) shouldBe (issueAmount - 10 * receivers.length)
+
+    assert(receivers.forall(rc => distribution(rc.toAddress) == 10), "Distribution correct")
+  }
+
   test("Correct last page and entry count") {
     val receivers = for (i <- 0 until 50) yield KeyPair(s"receiver#$i".getBytes("UTF-8"))
 
-    val issueTx = node.issue(issuer, "TestCoin#2", "no description", issueAmount, 8, false, issueFee, waitForTx = true).id
+    val issueTx = node.issue(issuer, "TestCoin#2", "no description", issueAmount, 8, reissuable = false, issueFee, waitForTx = true).id
 
     node
       .massTransfer(
@@ -96,6 +124,24 @@ class AssetDistributionSuite extends BaseTransactionSuite with CancelAfterFailur
     assert(pages.map(_.items.size).sum == 51)
   }
 
+  test("Unlimited list") {
+    val assetId = node.issue(issuer, "TestCoin#2", "no description", issueAmount, 8, reissuable = false, issueFee, waitForTx = true).id
+
+    val receivers = for (i <- 0 until 2000) yield KeyPair(s"receiver#$i".getBytes("UTF-8"))
+
+    val transfers = receivers.map { r => MassTransferTransaction.Transfer(r.toAddress.toString, 10L) }.toList
+
+    transfers.grouped(100).foreach { t =>
+      node.massTransfer(issuer, t, minFee + t.length * minFee, assetId = Some(assetId))
+    }
+
+    node.waitFor("empty utx")(_.utxSize, (_: Int) == 0, 1 second)
+    nodes.waitForHeightArise()
+
+    val list = node.assetDistribution(assetId)
+    list should have size 2001
+  }
+
   def distributionPages(asset: String, height: Int, limit: Int): List[AssetDistributionPage] = {
     def _load(acc: List[AssetDistributionPage], maybeAfter: Option[String]): List[AssetDistributionPage] = {
       val page = node.assetDistributionAtHeight(asset, height, limit, maybeAfter)

node/src/main/resources/swagger-ui/openapi.yaml

@@ -2440,6 +2440,38 @@ paths:
                     type: string
                   balance:
                     type: string
+  '/assets/{assetId}/distribution':
+    get:
+      tags:
+        - assets
+      summary: Asset balance distribution
+      description: Get asset balance distribution by addresses
+      operationId: getAssetDistributionOld
+      parameters:
+        - $ref: '#/components/parameters/assetId'
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties:
+                  type: integer
+                  format: int64
+                description: map of assetId <-> balance
+                example:
+                  2eEUvypDSivnzPiLrbYEW39SM8yMZ1aq4eJuiKfs4sEY: 15
+                  3PPqZ623dAfbmxmnpTjwV6yD5GA5s3PJiUG: 25
+            application/json;large-significand-format=string:
+              schema:
+                type: object
+                additionalProperties:
+                  type: string
+                description: map of assetId <-> balance
+                example:
+                  2eEUvypDSivnzPiLrbYEW39SM8yMZ1aq4eJuiKfs4sEY: "15"
+                  3PPqZ623dAfbmxmnpTjwV6yD5GA5s3PJiUG: "25"
   '/assets/{assetId}/distribution/{height}/limit/{limit}':
     get:
       tags:

node/src/main/scala/com/wavesplatform/Application.scala

@@ -374,9 +374,8 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
         else heavyRequestExecutor
       )
 
-      val routeTimeout = new RouteTimeout(
-        FiniteDuration(settings.config.getDuration("akka.http.server.request-timeout").getSeconds, TimeUnit.SECONDS)
-      )(heavyRequestScheduler)
+      val serverRequestTimeout = FiniteDuration(settings.config.getDuration("akka.http.server.request-timeout").getSeconds, TimeUnit.SECONDS)
+      val routeTimeout         = new RouteTimeout(serverRequestTimeout)(heavyRequestScheduler)
 
       val apiRoutes = Seq(
         new EthRpcRoute(blockchainUpdater, extensionContext.transactionsApi, time),
@@ -440,6 +439,7 @@ class Application(val actorSystem: ActorSystem, val settings: WavesSettings, con
         ),
         AssetsApiRoute(
           settings.restAPISettings,
+          serverRequestTimeout,
           wallet,
           transactionPublisher,
           blockchainUpdater,

node/src/main/scala/com/wavesplatform/api/http/assets/AssetsApiRoute.scala

@@ -18,7 +18,13 @@ import com.wavesplatform.api.common.{CommonAccountsApi, CommonAssetsApi}
 import com.wavesplatform.api.http.*
 import com.wavesplatform.api.http.ApiError.*
 import com.wavesplatform.api.http.StreamSerializerUtils.*
-import com.wavesplatform.api.http.assets.AssetsApiRoute.{AssetDetails, AssetInfo, DistributionParams, assetDetailsSerializer}
+import com.wavesplatform.api.http.assets.AssetsApiRoute.{
+  AssetDetails,
+  AssetInfo,
+  DistributionParams,
+  assetDetailsSerializer,
+  assetDistributionSerializer
+}
 import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.lang.ValidationError
 import com.wavesplatform.network.TransactionPublisher
@@ -39,9 +45,11 @@ import play.api.libs.json.*
 
 import java.util.concurrent.*
 import scala.concurrent.Future
+import scala.concurrent.duration.FiniteDuration
 
 case class AssetsApiRoute(
     settings: RestAPISettings,
+    serverRequestTimeout: FiniteDuration,
     wallet: Wallet,
     transactionPublisher: TransactionPublisher,
     blockchain: Blockchain,
@@ -65,6 +73,8 @@ case class AssetsApiRoute(
     )
   )
 
+  private val assetDistRouteTimeout = new RouteTimeout(serverRequestTimeout)(distributionTaskScheduler)
+
   override lazy val route: Route =
     pathPrefix("assets") {
       pathPrefix("balance" / AddrSegment) { address =>
@@ -97,9 +107,10 @@ case class AssetsApiRoute(
         (path("nft" / AddrSegment / "limit" / IntNumber) & parameter("after".as[String].?)) { (address, limit, maybeAfter) =>
           nft(address, limit, maybeAfter)
         } ~ pathPrefix(AssetId / "distribution") { assetId =>
-          (path(IntNumber / "limit" / IntNumber) & parameter("after".?)) { (height, limit, maybeAfter) =>
-            balanceDistributionAtHeight(assetId, height, limit, maybeAfter)
-          }
+          pathEndOrSingleSlash(balanceDistribution(assetId)) ~
+            (path(IntNumber / "limit" / IntNumber) & parameter("after".?)) { (height, limit, maybeAfter) =>
+              balanceDistributionAtHeight(assetId, height, limit, maybeAfter)
+            }
         }
       }
     }
@@ -180,6 +191,16 @@ case class AssetsApiRoute(
       }
     }
 
+  def balanceDistribution(assetId: IssuedAsset): Route = {
+    implicit val jsonStreamingSupport: ToResponseMarshaller[Source[(Address, Long), NotUsed]] =
+      jacksonStreamMarshaller(prefix = "{", suffix = "}")(assetDistributionSerializer)
+
+    assetDistRouteTimeout.executeFromObservable(
+      commonAssetsApi
+        .assetDistribution(assetId, blockchain.height, None)
+    )
+  }
+
   def balanceDistributionAtHeight(assetId: IssuedAsset, heightParam: Int, limitParam: Int, afterParam: Option[String]): Route =
     optionalHeaderValueByType(Accept) { accept =>
       val paramsEi: Either[ValidationError, DistributionParams] =
@@ -511,4 +532,16 @@ object AssetsApiRoute {
           gen.writeEndObject()
       }
     }
+
+  def assetDistributionSerializer(numbersAsString: Boolean): JsonSerializer[(Address, Long)] =
+    (value: (Address, Long), gen: JsonGenerator, _: SerializerProvider) => {
+      val (address, balance) = value
+      if (numbersAsString) {
+        gen.writeRaw(s"\"${address.toString}\":")
+        gen.writeString(balance.toString)
+      } else {
+        gen.writeRaw(s"\"${address.toString}\":")
+        gen.writeNumber(balance)
+      }
+    }
 }

node/src/test/scala/com/wavesplatform/api/http/CustomJsonMarshallerSpec.scala

@@ -115,6 +115,7 @@ class CustomJsonMarshallerSpec
 
   private val assetsRoute = AssetsApiRoute(
     restAPISettings,
+    60.seconds,
     testWallet,
     publisher,
     blockchain,

node/src/test/scala/com/wavesplatform/http/AssetsRouteSpec.scala

@@ -57,6 +57,7 @@ class AssetsRouteSpec
         seal(
           AssetsApiRoute(
             restAPISettings,
+            60.seconds,
             testWallet,
             DummyTransactionPublisher.accepting,
             d.blockchain,