PR 211 rebased on master

modules/explorer-api/src/main/scala/org/ergoplatform/explorer/http/api/v1/routes/AddressesRoutes.scala

@@ -39,7 +39,7 @@ final class AddressesRoutes[F[_]: Concurrent: ContextShift: Timer: AdaptThrowabl
 
   private def getTotalBalanceR =
     interpreter.toRoutes(defs.getTotalBalanceDef) { addr =>
-      addresses.totalBalanceOf(addr).adaptThrowable.value
+      addresses.totalBalanceWithConsiderationOfMempoolFor(addr).adaptThrowable.value
     }
 
   private def getBatchAddressInfo =

modules/explorer-api/src/main/scala/org/ergoplatform/explorer/http/api/v1/services/Addresses.scala

@@ -26,6 +26,10 @@ trait Addresses[F[_]] {
   def totalBalanceOf(address: Address): F[TotalBalance]
 
   def addressInfoOf(batch: List[Address], minConfirmations: Int = 0): F[Map[Address, AddressInfo]]
+
+  /** Get TotalBalance of address with consideration of mempool data
+    */
+  def totalBalanceWithConsiderationOfMempoolFor(address: Address): F[TotalBalance]
 }
 
 object Addresses {
@@ -73,6 +77,12 @@ object Addresses {
       )) ||> trans.xa
     }
 
+    def totalBalanceWithConsiderationOfMempoolFor(address: Address): F[TotalBalance] =
+      for {
+        confirmed   <- confirmedBalanceOf(address, 0)
+        unconfirmed <- memprops.getUnconfirmedBalanceByAddress(address, confirmed)
+      } yield TotalBalance(confirmed, unconfirmed)
+
     private def hasBeenUsedByErgoTree(ergoTree: HexString): F[Boolean] =
       outputRepo.countAllByErgoTree(ergoTree).map(_ > 0) ||> trans.xa
 

modules/explorer-api/src/main/scala/org/ergoplatform/explorer/http/api/v1/shared/MempoolProps.scala

@@ -12,7 +12,8 @@ import org.ergoplatform.explorer.db.models.aggregates.{ExtendedAsset, ExtendedUA
 import org.ergoplatform.explorer.db.models.{UOutput, UTransaction}
 import org.ergoplatform.explorer.db.repositories.bundles.UtxRepoBundle
 import org.ergoplatform.explorer.http.api.streaming.CompileStream
-import org.ergoplatform.explorer.http.api.v1.models.{UInputInfo, UOutputInfo, UTransactionInfo}
+import org.ergoplatform.explorer.http.api.v1.models.{Balance, UInputInfo, UOutputInfo, UTransactionInfo}
+import org.ergoplatform.explorer.http.api.v1.utils.BuildUnconfirmedBalance
 import org.ergoplatform.explorer.settings.{ServiceSettings, UtxCacheSettings}
 import org.ergoplatform.explorer.{Address, BoxId, ErgoTree, TxId}
 import org.ergoplatform.{explorer, ErgoAddressEncoder}
@@ -27,6 +28,7 @@ trait MempoolProps[F[_], D[_]] {
   def hasUnconfirmedBalance(ergoTree: ErgoTree): F[Boolean]
   def mkUnspentOutputInfo: Pipe[D, Chunk[UOutput], UOutputInfo]
   def mkTransaction: Pipe[D, Chunk[UTransaction], UTransactionInfo]
+  def getUnconfirmedBalanceByAddress(address: Address, confirmedBalance: Balance): F[Balance]
 }
 
 object MempoolProps {
@@ -47,6 +49,28 @@ object MempoolProps {
 
     import repo._
 
+    def getUnconfirmedBalanceByAddress(
+      address: Address,
+      confirmedBalance: Balance
+    ): F[Balance] = {
+      val ergoTree = addressToErgoTreeNewtype(address)
+      txs
+        .countByErgoTree(ergoTree.value)
+        .flatMap {
+          case 0 => Balance.empty.pure[D]
+          case _ =>
+            txs
+              .streamRelatedToErgoTree(ergoTree, 0, Int.MaxValue)
+              .chunkN(settings.chunkSize)
+              .through(mkTransaction)
+              .to[List]
+              .map { poolItems =>
+                BuildUnconfirmedBalance(poolItems, confirmedBalance, ergoTree, ergoTree.value)
+              }
+        }
+        .thrushK(trans.xa)
+    }
+
     def hasUnconfirmedBalance(ergoTree: ErgoTree): F[Boolean] =
       txs
         .countByErgoTree(ergoTree.value)

modules/explorer-api/src/main/scala/org/ergoplatform/explorer/http/api/v1/utils/BuildUnconfirmedBalance.scala

@@ -0,0 +1,88 @@
+package org.ergoplatform.explorer.http.api.v1.utils
+
+import org.ergoplatform.explorer.http.api.models.AssetInstanceInfo
+import org.ergoplatform.explorer.http.api.v1.models._
+import org.ergoplatform.explorer.{ErgoTree, HexString, TokenId}
+
+object BuildUnconfirmedBalance {
+
+  /** Build unconfirmed balance considering MemPool Data
+    * by reducing a `List[`[[org.ergoplatform.explorer.http.api.v1.models.UTransactionInfo]]`]`
+    * into a [[org.ergoplatform.explorer.http.api.v1.models.Balance]]
+    *
+    * Unconfirmed Balance arithmetic is completed in three steps:
+    * <li> determine if a [[org.ergoplatform.explorer.http.api.v1.models.UTransactionInfo]] is a credit or debit by
+    *  matching its [[org.ergoplatform.explorer.http.api.v1.models.UInputInfo]] to wallets </li>
+    *  <li> reducing similar [[org.ergoplatform.explorer.http.api.v1.models.UOutputInfo]]
+    *  sums (credits/debits) into a single value: `debitSum/creditSum` </li>
+    *  <li> subtracting or adding `debitSum/creditSum` into iterations current balance </li>
+    *
+    * @param items  unsigned transactions from the MemPool
+    * @param confirmedBalance last signed & unspent balance
+    * @param ergoTree  for transaction type identification
+    * @param hexString for NSum evaluation
+    * @return
+    */
+  def apply(
+    items: List[UTransactionInfo],
+    confirmedBalance: Balance,
+    ergoTree: ErgoTree,
+    hexString: HexString
+  ): Balance =
+    items.foldLeft(confirmedBalance) { case (balance, transactionInfo) =>
+      transactionInfo.inputs.head.ergoTree match {
+        case ieT if ieT == ergoTree =>
+          val debitSum = transactionInfo.outputs.foldLeft(0L) { case (sum, outputInfo) =>
+            if (outputInfo.ergoTree != hexString) sum + outputInfo.value
+            else sum
+          }
+
+          val debitSumTokenGroups = transactionInfo.outputs
+            .foldLeft(Map[TokenId, AssetInstanceInfo]()) { case (groupedTokens, outputInfo) =>
+              if (outputInfo.ergoTree != hexString)
+                outputInfo.assets.foldLeft(groupedTokens) { case (gT, asset) =>
+                  val gTAsset = gT.getOrElse(asset.tokenId, asset.copy(amount = 0L))
+                  gT + (asset.tokenId -> gTAsset.copy(amount = gTAsset.amount + asset.amount))
+                }
+              else groupedTokens
+            }
+
+          val newTokensBalance = balance.tokens.map { token =>
+            debitSumTokenGroups.get(token.tokenId).map { assetInfo =>
+              token.copy(amount = token.amount - assetInfo.amount)
+            } match {
+              case Some(value) => value
+              case None        => token
+            }
+          }
+
+          Balance(balance.nanoErgs - debitSum, newTokensBalance)
+        case _ =>
+          val creditSum = transactionInfo.outputs.foldLeft(0L) { case (sum, outputInfo) =>
+            if (outputInfo.ergoTree == hexString) sum + outputInfo.value
+            else sum
+          }
+
+          val creditSumTokenGroups = transactionInfo.outputs
+            .foldLeft(Map[TokenId, AssetInstanceInfo]()) { case (groupedTokens, outputInfo) =>
+              if (outputInfo.ergoTree == hexString)
+                outputInfo.assets.foldLeft(groupedTokens) { case (gT, asset) =>
+                  val gTAsset = gT.getOrElse(asset.tokenId, asset.copy(amount = 0L))
+                  gT + (asset.tokenId -> gTAsset.copy(amount = gTAsset.amount + asset.amount))
+                }
+              else groupedTokens
+            }
+
+          val newTokensBalance = balance.tokens.map { token =>
+            creditSumTokenGroups.get(token.tokenId).map { assetInfo =>
+              token.copy(amount = token.amount + assetInfo.amount)
+            } match {
+              case Some(value) => value
+              case None        => token
+            }
+          }
+
+          Balance(balance.nanoErgs + creditSum, newTokensBalance)
+      }
+    }
+}

modules/explorer-api/src/test/scala/org/ergoplatform/explorer/v1/services/AddressesSpec.scala

@@ -1,45 +1,38 @@
 package org.ergoplatform.explorer.v1.services
 
-import cats.{Monad, Parallel}
-import cats.syntax.option._
 import cats.effect.{Concurrent, ContextShift, IO}
+import cats.syntax.option._
+import cats.{Monad, Parallel}
 import dev.profunktor.redis4cats.RedisCommands
 import doobie.free.connection.ConnectionIO
 import eu.timepit.refined.api.Refined
 import eu.timepit.refined.auto._
 import eu.timepit.refined.string.ValidByte
 import org.ergoplatform.ErgoAddressEncoder
-import org.ergoplatform.explorer.{Address, ErgoTree}
 import org.ergoplatform.explorer.cache.Redis
 import org.ergoplatform.explorer.commonGenerators.forSingleInstance
-import org.ergoplatform.explorer.db.{repositories, RealDbTest, Trans}
 import org.ergoplatform.explorer.db.algebra.LiftConnectionIO
-import org.ergoplatform.explorer.testSyntax.runConnectionIO._
-import org.ergoplatform.explorer.db.repositories.{
-  AssetRepo,
-  HeaderRepo,
-  InputRepo,
-  OutputRepo,
-  TokenRepo,
-  TransactionRepo
-}
+import org.ergoplatform.explorer.db.models.aggregates.AggregatedAsset
+import org.ergoplatform.explorer.db.models.generators._
+import org.ergoplatform.explorer.db.repositories._
+import org.ergoplatform.explorer.db.{repositories, RealDbTest, Trans}
 import org.ergoplatform.explorer.http.api.streaming.CompileStream
+import org.ergoplatform.explorer.http.api.v1.models.{AddressInfo, TokenAmount}
 import org.ergoplatform.explorer.http.api.v1.services.{Addresses, Mempool}
+import org.ergoplatform.explorer.http.api.v1.shared.MempoolProps
+import org.ergoplatform.explorer.protocol.sigma
 import org.ergoplatform.explorer.settings.{RedisSettings, ServiceSettings, UtxCacheSettings}
 import org.ergoplatform.explorer.testContainers.RedisTest
-import org.scalatest.{PrivateMethodTester, TryValues}
+import org.ergoplatform.explorer.testSyntax.runConnectionIO._
+import org.ergoplatform.explorer.v1.services.AddressesSpec._
+import org.ergoplatform.explorer.v1.services.constants.{ReceiverAddressString, SenderAddressString}
+import org.ergoplatform.explorer.{Address, ErgoTree}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should
+import org.scalatest.{PrivateMethodTester, TryValues}
 import tofu.syntax.monadic._
-
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
 import scala.util.Try
-import org.ergoplatform.explorer.http.api.v1.shared.MempoolProps
-import org.ergoplatform.explorer.v1.services.AddressesSpec._
-import org.ergoplatform.explorer.db.models.generators._
-import org.ergoplatform.explorer.http.api.v1.models.AddressInfo
-import org.ergoplatform.explorer.protocol.sigma
-import org.ergoplatform.explorer.v1.services.constants.{ReceiverAddressString, SenderAddressString}
 
 trait AddressesSpec
   extends AnyFlatSpec
@@ -55,7 +48,81 @@ class AS_A extends AddressesSpec {
   implicit val addressEncoder: ErgoAddressEncoder =
     ErgoAddressEncoder(networkPrefix.value.toByte)
 
-  "Address Service" should "" in {}
+  "Address Service" should "get confirmed Balance (nanoErgs) of address" in {
+    implicit val trans: Trans[ConnectionIO, IO] = Trans.fromDoobie(xa)
+    import tofu.fs2Instances._
+    withResources[IO](container.mappedPort(redisTestPort))
+      .use { case (settings, utxCache, redis) =>
+        withServices[IO, ConnectionIO](settings, utxCache, redis) { (addr, _, _) =>
+          val addressT = Address.fromString[Try](SenderAddressString)
+          addressT.isSuccess should be(true)
+          val addressTree = sigma.addressToErgoTreeHex(addressT.get)
+          val boxValues   = List((100.toNanoErgo, 1), (200.toNanoErgo, 1))
+          withLiveRepos[ConnectionIO] { (headerRepo, txRepo, oRepo, _, _, _) =>
+            forSingleInstance(
+              balanceOfAddressGen(
+                mainChain = true,
+                address   = addressT.get,
+                addressTree,
+                values = boxValues
+              )
+            ) { infoTupleList =>
+              infoTupleList.foreach { case (header, out, tx) =>
+                headerRepo.insert(header).runWithIO()
+                oRepo.insert(out).runWithIO()
+                txRepo.insert(tx).runWithIO()
+              }
+              val balance = addr.confirmedBalanceOf(addressT.get, 0).unsafeRunSync().nanoErgs
+              balance should be(300.toNanoErgo)
+            }
+          }
+
+        }
+      }
+      .unsafeRunSync()
+  }
+
+}
+
+class AS_B extends AddressesSpec {
+
+  val networkPrefix: String Refined ValidByte = "16" // strictly run test-suite with testnet network prefix
+  implicit val addressEncoder: ErgoAddressEncoder =
+    ErgoAddressEncoder(networkPrefix.value.toByte)
+
+  "Address Service" should "get confirmed balance (tokens) of address" in {
+    implicit val trans: Trans[ConnectionIO, IO] = Trans.fromDoobie(xa)
+    import tofu.fs2Instances._
+    withResources[IO](container.mappedPort(redisTestPort))
+      .use { case (settings, utxCache, redis) =>
+        withServices[IO, ConnectionIO](settings, utxCache, redis) { (addr, _, _) =>
+          val addressT = Address.fromString[Try](SenderAddressString)
+          addressT.isSuccess should be(true)
+          val addressTree = sigma.addressToErgoTreeHex(addressT.get)
+          withLiveRepos[ConnectionIO] { (headerRepo, txRepo, oRepo, _, tokenRepo, assetRepo) =>
+            forSingleInstance(
+              balanceOfAddressWithTokenGen(mainChain = true, address = addressT.get, addressTree, 1, 5)
+            ) { infoTupleList =>
+              infoTupleList.foreach { case (header, out, tx, _, token, asset) =>
+                headerRepo.insert(header).runWithIO()
+                oRepo.insert(out).runWithIO()
+                txRepo.insert(tx).runWithIO()
+                tokenRepo.insert(token).runWithIO()
+                assetRepo.insert(asset).runWithIO()
+              }
+              val tokeAmount = infoTupleList.map { x =>
+                val tk  = x._5
+                val ase = x._6
+                TokenAmount(AggregatedAsset(tk.id, ase.amount, tk.name, tk.decimals, tk.`type`))
+              }
+              val tokenBalance = addr.confirmedBalanceOf(addressT.get, 0).unsafeRunSync().tokens
+              tokenBalance should contain theSameElementsAs tokeAmount
+            }
+          }
+        }
+      }
+      .unsafeRunSync()
+  }
 }
 
 class AS_C extends AddressesSpec {
@@ -194,8 +261,9 @@ class AS_D extends AddressesSpec {
 object AddressesSpec {
 
   import cats.effect.Sync
-  import scala.concurrent.duration._
   import cats.syntax.traverse._
+
+  import scala.concurrent.duration._
   implicit val ec: ExecutionContextExecutor = ExecutionContext.global
 
   def withResources[F[_]: Sync: Monad: Parallel: Concurrent: ContextShift](port: Int) = for {