DEX-1694 dynamic asset fee config

dex-it-common/src/main/scala/com/wavesplatform/dex/it/api/dex/AsyncEnrichedDexApi.scala

@@ -444,6 +444,28 @@ class AsyncEnrichedDexApi(apiKey: String, host: => InetSocketAddress)(implicit e
       .headers(headers)
   }
 
+  override def addCustomFeeAssets(assets: Set[Asset], headers: Map[String, String]): AsyncEnrichedDexApi.R[HttpMessage] = mk {
+    basicRequest
+      .post(uri"$apiUri/matcher/settings/custom-fee-assets")
+      .followRedirects(false)
+      .headers(headers)
+      .body(Json.stringify(Json.toJson(HttpCustomFeeAssets(assets))))
+  }
+
+  override def addCustomFeeAssets(assets: Set[Asset]): AsyncEnrichedDexApi.R[HttpMessage] =
+    addCustomFeeAssets(assets, apiKeyHeaders)
+
+  override def deleteCustomFeeAssets(assets: Set[Asset], headers: Map[String, String]): AsyncEnrichedDexApi.R[HttpMessage] = mk {
+    basicRequest
+      .delete(uri"$apiUri/matcher/settings/custom-fee-assets")
+      .followRedirects(false)
+      .headers(headers)
+      .body(Json.stringify(Json.toJson(HttpCustomFeeAssets(assets))))
+  }
+
+  override def deleteCustomFeeAssets(assets: Set[Asset]): AsyncEnrichedDexApi.R[HttpMessage] =
+    deleteCustomFeeAssets(assets, apiKeyHeaders)
+
   override def deleteOrderBookWithKey(amountAsset: String, priceAsset: String, headers: Map[String, String]): R[HttpMessage] = mk {
     basicRequest
       .delete(uri"$apiUri/matcher/orderbook/$amountAsset/$priceAsset")

dex-it-common/src/main/scala/com/wavesplatform/dex/it/api/dex/DexApi.scala

@@ -199,6 +199,11 @@ trait DexApi[F[_]] {
   def cancelAllInOrderBookWithKey(amountAsset: String, priceAsset: String, headers: Map[String, String]): F[HttpMessage]
   def cancelAllInOrderBookWithKey(assetPair: AssetPair): F[HttpMessage]
 
+  def addCustomFeeAssets(assets: Set[Asset], headers: Map[String, String]): F[HttpMessage]
+  def addCustomFeeAssets(assets: Set[Asset]): F[HttpMessage]
+  def deleteCustomFeeAssets(assets: Set[Asset], headers: Map[String, String]): F[HttpMessage]
+  def deleteCustomFeeAssets(assets: Set[Asset]): F[HttpMessage]
+
   def upsertAssetRate(assetId: String, rate: Double, headers: Map[String, String] = Map.empty): F[HttpMessage]
   def upsertAssetRate(asset: Asset, rate: Double): F[HttpMessage]
   def upsertAssetRate(asset: Asset, rate: String): F[HttpMessage]

dex-it/src/test/scala/com/wavesplatform/it/sync/orders/OrderCompositeFeeTestSuite.scala

@@ -1,8 +1,10 @@
 package com.wavesplatform.it.sync.orders
 
 import com.typesafe.config.{Config, ConfigFactory}
+import com.wavesplatform.dex.api.http.entities.HttpCalculatedFeeResponse.CalculatedFee
 import com.wavesplatform.dex.api.http.entities.HttpOrderStatus
 import com.wavesplatform.dex.api.http.entities.HttpOrderStatus.Status
+import com.wavesplatform.dex.domain.asset.Asset
 import com.wavesplatform.dex.domain.order.OrderType
 import com.wavesplatform.dex.error.FeeNotEnough
 
@@ -81,6 +83,67 @@ final class OrderCompositeFeeTestSuite extends OrderFeeBaseTestSuite {
       aliceBalance2 - aliceBalance1 shouldBe 5.waves - baseFee / 2
       bobBalance2 - bobBalance1 shouldBe -5.waves - baseFee
     }
+
+    "add custom fee assets, but not use it until offset" in {
+      dex1.api.addCustomFeeAssets(Set(Asset.Waves, btc, usd))
+
+      dex1.api.calculateFee(wavesUsdPair, OrderType.SELL, 10.waves, 25.usd)
+        .base.value shouldBe CalculatedFee(Asset.Waves, baseFee)
+
+      dex1.api.calculateFee(wavesBtcPair, OrderType.SELL, 10.waves, 1.btc)
+        .base.value shouldBe CalculatedFee(Asset.Waves, 0.2.waves)
+
+      dex1.api.place(mkOrder(alice, wavesUsdnPair, OrderType.SELL, 20.waves, 5.usdn, matcherFee = baseFee)) // just for offset to be reached
+      dex1.api.place(mkOrder(alice, wavesUsdnPair, OrderType.SELL, 20.waves, 6.usdn, matcherFee = baseFee)) // just for offset to be reached
+
+      dex1.api.place(mkOrder(alice, wavesBtcPair, OrderType.SELL, 10.waves, 1.btc, matcherFee = 0.04.waves))
+      dex1.api.place(mkOrder(alice, wavesUsdPair, OrderType.SELL, 10.waves, 1.usd, matcherFee = 0.04.waves))
+
+      dex1.api.getOrderHistoryByPKWithSig(alice, activeOnly = Some(true)).foreach { item =>
+        dex1.api.cancelOneOrderWithKey(item.id, Some(alice.publicKey))
+      }
+
+      dex1.api.calculateFee(wavesBtcPair, OrderType.SELL, 10.waves, 1.btc)
+        .base.value shouldBe CalculatedFee(Asset.Waves, 400000L)
+
+      dex1.api.calculateFee(wavesUsdPair, OrderType.SELL, 10.waves, 1.usd)
+        .base.value shouldBe CalculatedFee(Asset.Waves, 400000L)
+
+      dex1.api.deleteCustomFeeAssets(Set(usd))
+
+      dex1.api.place(mkOrder(alice, wavesBtcPair, OrderType.SELL, 10.waves, 1.btc, matcherFee = 0.04.waves))
+      dex1.api.place(mkOrder(alice, wavesUsdPair, OrderType.SELL, 10.waves, 1.usd, matcherFee = baseFee))
+
+      dex1.api.calculateFee(wavesBtcPair, OrderType.SELL, 10.waves, 1.btc)
+        .base.value shouldBe CalculatedFee(Asset.Waves, 400000L)
+
+      dex1.api.calculateFee(wavesUsdPair, OrderType.SELL, 10.waves, 1.usd)
+        .base.value shouldBe CalculatedFee(Asset.Waves, baseFee)
+
+    }
+
+    "read all custom assets action from levelDB at start" in {
+      (1 to 20).foreach { i =>
+        val order = mkOrder(alice, wavesUsdnPair, OrderType.SELL, 1.waves, (5 + i).usdn, matcherFee = baseFee)
+        dex1.api.place(order)
+        dex1.api.cancelOrderById(order)
+      }
+
+      dex1.api.saveSnapshots
+
+      dex1.restart()
+
+      dex1.api.place(mkOrder(alice, wavesBtcPair, OrderType.SELL, 10.waves, 1.btc, matcherFee = 0.04.waves))
+      dex1.api.place(mkOrder(alice, wavesUsdPair, OrderType.SELL, 10.waves, 1.usd, matcherFee = baseFee))
+
+      dex1.api.calculateFee(wavesBtcPair, OrderType.SELL, 10.waves, 1.btc)
+        .base.value shouldBe CalculatedFee(Asset.Waves, 400000L)
+
+      dex1.api.calculateFee(wavesUsdPair, OrderType.SELL, 10.waves, 1.usd)
+        .base.value shouldBe CalculatedFee(Asset.Waves, baseFee)
+
+    }
+
   }
 
   private def multiplyAmountByDouble(a: Long, d: Double): Long =
@@ -89,12 +152,12 @@ final class OrderCompositeFeeTestSuite extends OrderFeeBaseTestSuite {
   override protected def dexInitialSuiteConfig: Config = ConfigFactory.parseString(
     s"""
        |waves.dex {
-       |  price-assets = [ "$UsdId", "$BtcId", "WAVES" ]
+       |  price-assets = [ "$UsdId", "$UsdnId", "$BtcId", "WAVES" ]
        |  allowed-order-versions = [1, 2, 3]
        |  order-fee.-1 {
        |    mode = composite
        |    composite {
-       |      custom {
+       |      custom-pairs {
        |        "WAVES-$BtcId": {
        |          mode = percent
        |          percent {
@@ -111,17 +174,50 @@ final class OrderCompositeFeeTestSuite extends OrderFeeBaseTestSuite {
        |          base-taker-fee = $baseFee
        |        }
        |      }
+       |      discount {
+       |        asset = "$WctId"
+       |        value = 0.1
+       |      }
        |    }
        |  }
+       |  order-fee.11 {
+       |    mode = composite
+       |    composite {
+       |      dynamic-custom-assets = true
+       |      custom-pairs {}
+       |      custom-assets {
+       |        assets = []
+       |        settings {
+       |           mode = "fixed"
+       |           fixed {
+       |              asset = "WAVES"
+       |              min-fee = 400000
+       |            }
+       |        }
+       |      }
+       |      default {
+       |        mode = dynamic
+       |        dynamic {
+       |          base-maker-fee = $baseFee
+       |          base-taker-fee = $baseFee
+       |        }
+       |      }
+       |      discount {
+       |        asset = "$WctId"
+       |        value = 0.1
+       |      }
+       |	  }
+       |  }
        |}
        """.stripMargin
   )
 
   override protected def beforeAll(): Unit = {
     wavesNode1.start()
-    broadcastAndAwait(IssueUsdTx, IssueBtcTx)
+    broadcastAndAwait(IssueUsdTx, IssueBtcTx, IssueUsdnTx, IssueWctTx)
     broadcastAndAwait(mkTransfer(bob, alice, 100.btc, btc))
     dex1.start()
+    upsertAssetRate((usdn, 1.0), (wct, 1.0))
   }
 
 }

dex-pb/src/main/protobuf/dex_order.proto

@@ -67,6 +67,14 @@ message CancelOrder {
     bytes owner = 3;
 };
 
+message AddFeeCustomAssets {
+    repeated bytes asset_id = 1;
+};
+
+message DeleteFeeCustomAssets {
+    repeated bytes asset_id = 1;
+};
+
 message ValidatedCommand {
     AssetPair asset_pair = 1;
     bytes kamon_ctx = 2;
@@ -76,5 +84,7 @@ message ValidatedCommand {
         CancelOrder cancel_order = 5;
         google.protobuf.Empty cancel_all_orders = 6;
         google.protobuf.Empty delete_order_book = 7;
+        AddFeeCustomAssets add_fee_custom_assets = 8;
+        DeleteFeeCustomAssets delete_fee_custom_assets = 9;
     }
 };

dex/src/main/resources/application.conf

@@ -273,7 +273,9 @@ waves.dex {
       }
 
       composite {
-        custom {
+        dynamic-custom-assets = false # use dynamic assets from kafka and levelDB for custom-assets section
+
+        custom-pairs {
           DWgwcZTMhSvnyYCoWLRUXXSH1RSkzThXLJhww9gwkqdn-25FEqEjRkqK6yCkiT7Lz6SAYz7gUFCtxfCChnrVFD5AT {
             mode = "percent"
             percent {
@@ -296,7 +298,7 @@ waves.dex {
             # WAVES or Issued asset ID in base 58
             assets = []
 
-            # will be applied to all valid pair combinations from the 'assets' field
+            # will be applied to all valid pair combinations from the 'assets' field or from kafka and levelDB, if dynamic-custom-assets set to true
             settings {
                 mode = "percent"
                 percent {

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

@@ -33,6 +33,7 @@ import com.wavesplatform.dex.api.routes.ApiRoute
 import com.wavesplatform.dex.api.ws.actors.{WsExternalClientDirectoryActor, WsInternalBroadcastActor}
 import com.wavesplatform.dex.api.ws.routes.MatcherWebSocketRoute
 import com.wavesplatform.dex.app._
+import com.wavesplatform.dex.caches.OrderFeeSettingsCache.{AssetsActionForOffset, CustomAssetFeeState}
 import com.wavesplatform.dex.caches.{MatchingRulesCache, OrderFeeSettingsCache, RateCache}
 import com.wavesplatform.dex.db._
 import com.wavesplatform.dex.db.leveldb.{openDb, LevelDb}
@@ -141,6 +142,7 @@ class Application(settings: MatcherSettings, config: Config)(implicit val actorS
   private val assetPairsDb = AssetPairsDb.levelDb(commonLevelDb)
   private val orderBookSnapshotDb = OrderBookSnapshotDb.levelDb(snapshotsLevelDb)
   private val orderDb = OrderDb.levelDb(settings.orderDb, db, levelDbEcMap)
+  private val customFeeAssetsDb = CustomFeeAssetsDb.levelDb(commonLevelDb)
 
   private val assetsCache = AssetsCache.from(AssetsDb.levelDb(commonLevelDb))
 
@@ -211,8 +213,14 @@ class Application(settings: MatcherSettings, config: Config)(implicit val actorS
 
   private val pairBuilder = new AssetPairBuilder(settings, getAndCacheDescription, settings.blacklistedAssets)
 
+  private val allFeeAssetsActions = Await.result(customFeeAssetsDb.all(), 1.minute)
+
+  private val cfaState = allFeeAssetsActions.foldLeft(CustomAssetFeeState.empty) {
+    case (state, action) => state.applyAssetsActionForOffset(action)
+  }
+
   private val orderFeeSettingsCache =
-    new OrderFeeSettingsCache(settings.orderFee.view.mapValues(_(pairBuilder.quickValidateAssetPair(_).isRight)).toMap)
+    new OrderFeeSettingsCache(settings.orderFee.view.mapValues(_(pairBuilder.quickValidateAssetPair(_).isRight)).toMap, cfaState)
 
   private val exchangeTxStorage = ExchangeTxStorage.levelDB(commonLevelDb)
 
@@ -506,8 +514,21 @@ class Application(settings: MatcherSettings, config: Config)(implicit val actorS
     currentOffset = () => lastProcessedOffset
   )
 
+  private val customFeeAssetsRoute = new CustomAssetsFeeRoute(apiKeyHashes, orderFeeSettingsCache, storeCommand)
+
   private val v0HttpRoute =
-    Seq(infoRoute, ratesRoute, debugRoute, marketsRoute, historyRoute, placeRoute, cancelRoute, balancesRoute, transactionsRoute)
+    Seq(
+      infoRoute,
+      ratesRoute,
+      debugRoute,
+      marketsRoute,
+      historyRoute,
+      placeRoute,
+      cancelRoute,
+      balancesRoute,
+      transactionsRoute,
+      customFeeAssetsRoute
+    )
 
   private val v1HttpRoute = Seq(OrderBookRoute(
     assetPairBuilder = pairBuilder,
@@ -666,37 +687,77 @@ class Application(settings: MatcherSettings, config: Config)(implicit val actorS
       val loadAssets = eventAssets.traverse(getAndCacheDescription).value
 
       loadAssets.flatMap { _ =>
-        val assetPairs: Set[AssetPair] = xs
-          .map { validatedCommandWithMeta =>
-            lazy val handleValidatedCommandWithMeta = {
-              log.debug(s"Consumed $validatedCommandWithMeta")
-              orderBookDirectoryActorRef ! validatedCommandWithMeta
-              lastProcessedOffset = validatedCommandWithMeta.offset
-              validatedCommandWithMeta.command.assetPair
+        val (pairs, future) = xs.foldLeft((Set.empty[AssetPair], Future.unit)) {
+          case ((pairs, future), validatedCommand) =>
+            log.debug(s"Consumed $validatedCommand")
+            validatedCommand.command match {
+              case cmd: OrderBookValidatedCommand =>
+                lazy val handleCommand = {
+                  orderBookDirectoryActorRef ! OrderBookDirectoryActor.ApplyValidatedCommandWithPair(
+                    validatedCommand.offset,
+                    validatedCommand.timestamp,
+                    cmd
+                  )
+                  lastProcessedOffset = validatedCommand.offset
+                  cmd.assetPair
+                }
+
+                val newFuture = future.map { _ =>
+                  cmd.maybeCtx.fold(handleCommand) { ctx =>
+                    if (status == MatcherStatus.Working) {
+                      val parentSpan = ctx.get(kamon.trace.Span.Key)
+                      val span =
+                        Kamon.spanBuilder(s"consumedValidatedCommandWithMeta")
+                          .asChildOf(parentSpan)
+                          .traceId(parentSpan.trace.id)
+                          .tag(ctx.tags)
+                          .samplingDecision(KamonTraceUtils.Sample)
+                          .doNotTrackMetrics()
+                          .start()
+                      Kamon.runWithSpan[AssetPair](span)(handleCommand)
+                    } else
+                      handleCommand
+                  }
+                  ()
+                }
+                (pairs + cmd.assetPair, newFuture)
+              case cmd: ValidatedCommand.AddCustomAssetToFee =>
+                val action = AssetsActionForOffset(assets = cmd.assets, offset = validatedCommand.offset, isAdded = true)
+                val newFuture = customFeeAssetsDb.save(action).tap(_.onComplete {
+                  case Failure(ex) =>
+                    log.error("Error while event processing occurred: ", ex)
+                    forceStopApplication(EventProcessingError)
+                  case Success(_) =>
+                    orderFeeSettingsCache.applyAssetsActionForOffset(action)
+                    lastProcessedOffset = action.offset
+                })
+                (pairs, newFuture)
+
+              case cmd: ValidatedCommand.DeleteCustomAssetToFee =>
+                val action = AssetsActionForOffset(assets = cmd.assets, offset = validatedCommand.offset, isAdded = false)
+                val newFuture = customFeeAssetsDb.save(action).tap(_.onComplete {
+                  case Failure(ex) =>
+                    log.error("Error while event processing occurred: ", ex)
+                    forceStopApplication(EventProcessingError)
+                  case Success(_) =>
+                    orderFeeSettingsCache.applyAssetsActionForOffset(action)
+                    lastProcessedOffset = action.offset
+                })
+                (pairs, newFuture)
             }
+        }
 
-            validatedCommandWithMeta.command.maybeCtx.fold(handleValidatedCommandWithMeta) { ctx =>
-              if (status == MatcherStatus.Working) {
-                val parentSpan = ctx.get(kamon.trace.Span.Key)
-                val span =
-                  Kamon.spanBuilder(s"consumedValidatedCommandWithMeta")
-                    .asChildOf(parentSpan)
-                    .traceId(parentSpan.trace.id)
-                    .tag(ctx.tags)
-                    .samplingDecision(KamonTraceUtils.Sample)
-                    .doNotTrackMetrics()
-                    .start()
-                Kamon.runWithSpan[AssetPair](span)(handleValidatedCommandWithMeta)
-              } else
-                handleValidatedCommandWithMeta
-            }
+        if (pairs.nonEmpty)
+          future.flatMap { _ =>
+            orderBookDirectoryActorRef
+              .ask(OrderBookDirectoryActor.PingAll(pairs))(processConsumedTimeout)
+              .recover {
+                case NonFatal(e) =>
+                  log.error("PingAll is timed out!", e)
+              }.map(_ => ())
           }
-          .to(Set)
+        else future
 
-        orderBookDirectoryActorRef
-          .ask(OrderBookDirectoryActor.PingAll(assetPairs))(processConsumedTimeout)
-          .recover { case NonFatal(e) => log.error("PingAll is timed out!", e) }
-          .map(_ => ())
       } andThen {
         case Failure(ex) =>
           log.error("Error while event processing occurred: ", ex)