Merge pull request #25 from Constellation-Labs/addapting-to-2.0.1

Constellation-Labs · Nov 27, 2023 · 82be142 · 82be142
2 parents c7b0fcd + b767e7e
commit 82be142
Show file tree

Hide file tree

Showing 11 changed files with 230 additions and 110 deletions.
diff --git a/metagraph/modules/data_l1/src/main/scala/com/my/dor_metagraph/data_l1/Main.scala b/metagraph/modules/data_l1/src/main/scala/com/my/dor_metagraph/data_l1/Main.scala
@@ -1,10 +1,10 @@
 package com.my.dor_metagraph.data_l1
 
 import cats.data.NonEmptyList
-import cats.effect.IO
-import cats.implicits.catsSyntaxValidatedIdBinCompat0
+import cats.effect.{IO, Resource}
+import cats.implicits.{catsSyntaxOptionId, catsSyntaxValidatedIdBinCompat0}
 import com.my.dor_metagraph.shared_data.Data
-import com.my.dor_metagraph.shared_data.calculated_state.CalculatedState
+import com.my.dor_metagraph.shared_data.calculated_state.CalculatedStateService
 import com.my.dor_metagraph.shared_data.decoders.Decoders
 import com.my.dor_metagraph.shared_data.deserializers.Deserializers
 import com.my.dor_metagraph.shared_data.serializers.Serializers
@@ -18,6 +18,7 @@ import org.tessellation.currency.l1.CurrencyL1App
 import org.tessellation.schema.cluster.ClusterId
 import org.tessellation.security.signature.Signed
 import org.http4s.EntityDecoder
+import org.tessellation.ext.cats.effect.ResourceIO
 import org.tessellation.schema.SnapshotOrdinal
 import org.tessellation.security.hash.Hash
 
@@ -30,42 +31,60 @@ object Main
     ClusterId(UUID.fromString("517c3a05-9219-471b-a54c-21b7d72f4ae5")),
     version = BuildInfo.version
   ) {
-  override def dataApplication: Option[BaseDataApplicationL1Service[IO]] = Option(BaseDataApplicationL1Service(new DataApplicationL1Service[IO, CheckInUpdate, CheckInStateOnChain, CheckInDataCalculatedState] {
+  private def makeBaseDataApplicationL1Service(
+    calculatedStateService: CalculatedStateService[IO]
+  ): BaseDataApplicationL1Service[IO] =
+    BaseDataApplicationL1Service(new DataApplicationL1Service[IO, CheckInUpdate, CheckInStateOnChain, CheckInDataCalculatedState] {
 
-    override def validateData(state: DataState[CheckInStateOnChain, CheckInDataCalculatedState], updates: NonEmptyList[Signed[CheckInUpdate]])(implicit context: L1NodeContext[IO]): IO[DataApplicationValidationErrorOr[Unit]] = IO.pure(().validNec)
+      override def validateData(state: DataState[CheckInStateOnChain, CheckInDataCalculatedState], updates: NonEmptyList[Signed[CheckInUpdate]])(implicit context: L1NodeContext[IO]): IO[DataApplicationValidationErrorOr[Unit]] = IO.pure(().validNec)
 
-    override def validateUpdate(update: CheckInUpdate)(implicit context: L1NodeContext[IO]): IO[DataApplicationValidationErrorOr[Unit]] = Data.validateUpdate(update)
+      override def validateUpdate(update: CheckInUpdate)(implicit context: L1NodeContext[IO]): IO[DataApplicationValidationErrorOr[Unit]] = Data.validateUpdate(update)
 
-    override def combine(state: DataState[CheckInStateOnChain, CheckInDataCalculatedState], updates: List[Signed[CheckInUpdate]])(implicit context: L1NodeContext[IO]): IO[DataState[CheckInStateOnChain, CheckInDataCalculatedState]] = IO.pure(state)
+      override def combine(state: DataState[CheckInStateOnChain, CheckInDataCalculatedState], updates: List[Signed[CheckInUpdate]])(implicit context: L1NodeContext[IO]): IO[DataState[CheckInStateOnChain, CheckInDataCalculatedState]] = IO.pure(state)
 
-    override def routes(implicit context: L1NodeContext[IO]): HttpRoutes[IO] = HttpRoutes.empty
+      override def routes(implicit context: L1NodeContext[IO]): HttpRoutes[IO] = HttpRoutes.empty
 
-    override def dataEncoder: Encoder[CheckInUpdate] = implicitly[Encoder[CheckInUpdate]]
+      override def dataEncoder: Encoder[CheckInUpdate] = implicitly[Encoder[CheckInUpdate]]
 
-    override def dataDecoder: Decoder[CheckInUpdate] = implicitly[Decoder[CheckInUpdate]]
+      override def dataDecoder: Decoder[CheckInUpdate] = implicitly[Decoder[CheckInUpdate]]
 
-    override def calculatedStateEncoder: Encoder[CheckInDataCalculatedState] = implicitly[Encoder[CheckInDataCalculatedState]]
+      override def calculatedStateEncoder: Encoder[CheckInDataCalculatedState] = implicitly[Encoder[CheckInDataCalculatedState]]
 
-    override def calculatedStateDecoder: Decoder[CheckInDataCalculatedState] = implicitly[Decoder[CheckInDataCalculatedState]]
+      override def calculatedStateDecoder: Decoder[CheckInDataCalculatedState] = implicitly[Decoder[CheckInDataCalculatedState]]
 
-    override def signedDataEntityDecoder: EntityDecoder[IO, Signed[CheckInUpdate]] = Decoders.signedDataEntityDecoder
+      override def signedDataEntityDecoder: EntityDecoder[IO, Signed[CheckInUpdate]] = Decoders.signedDataEntityDecoder
 
-    override def serializeBlock(block: Signed[DataApplicationBlock]): IO[Array[Byte]] = IO(Serializers.serializeBlock(block)(dataEncoder.asInstanceOf[Encoder[DataUpdate]]))
+      override def serializeBlock(block: Signed[DataApplicationBlock]): IO[Array[Byte]] = IO(Serializers.serializeBlock(block)(dataEncoder.asInstanceOf[Encoder[DataUpdate]]))
 
-    override def deserializeBlock(bytes: Array[Byte]): IO[Either[Throwable, Signed[DataApplicationBlock]]] = IO(Deserializers.deserializeBlock(bytes)(dataDecoder.asInstanceOf[Decoder[DataUpdate]]))
+      override def deserializeBlock(bytes: Array[Byte]): IO[Either[Throwable, Signed[DataApplicationBlock]]] = IO(Deserializers.deserializeBlock(bytes)(dataDecoder.asInstanceOf[Decoder[DataUpdate]]))
 
-    override def serializeState(state: CheckInStateOnChain): IO[Array[Byte]] = IO(Serializers.serializeState(state))
+      override def serializeState(state: CheckInStateOnChain): IO[Array[Byte]] = IO(Serializers.serializeState(state))
 
-    override def deserializeState(bytes: Array[Byte]): IO[Either[Throwable, CheckInStateOnChain]] = IO(Deserializers.deserializeState(bytes))
+      override def deserializeState(bytes: Array[Byte]): IO[Either[Throwable, CheckInStateOnChain]] = IO(Deserializers.deserializeState(bytes))
 
-    override def serializeUpdate(update: CheckInUpdate): IO[Array[Byte]] = IO(Serializers.serializeUpdate(update))
+      override def serializeUpdate(update: CheckInUpdate): IO[Array[Byte]] = IO(Serializers.serializeUpdate(update))
 
-    override def deserializeUpdate(bytes: Array[Byte]): IO[Either[Throwable, CheckInUpdate]] = IO(Deserializers.deserializeUpdate(bytes))
+      override def deserializeUpdate(bytes: Array[Byte]): IO[Either[Throwable, CheckInUpdate]] = IO(Deserializers.deserializeUpdate(bytes))
 
-    override def getCalculatedState(implicit context: L1NodeContext[IO]): IO[(SnapshotOrdinal, CheckInDataCalculatedState)] = CalculatedState.getCalculatedState
+      override def getCalculatedState(implicit context: L1NodeContext[IO]): IO[(SnapshotOrdinal, CheckInDataCalculatedState)] = calculatedStateService.getCalculatedState.map(calculatedState => (calculatedState.ordinal, calculatedState.state))
 
-    override def setCalculatedState(ordinal: SnapshotOrdinal, state: CheckInDataCalculatedState)(implicit context: L1NodeContext[IO]): IO[Boolean] = CalculatedState.setCalculatedState(ordinal, state)
+      override def setCalculatedState(ordinal: SnapshotOrdinal, state: CheckInDataCalculatedState)(implicit context: L1NodeContext[IO]): IO[Boolean] = calculatedStateService.setCalculatedState(ordinal, state)
+
+      override def hashCalculatedState(state: CheckInDataCalculatedState)(implicit context: L1NodeContext[IO]): IO[Hash] = calculatedStateService.hashCalculatedState(state)
+
+      override def serializeCalculatedState(state: CheckInDataCalculatedState): IO[Array[Byte]] = IO(Serializers.serializeCalculatedState(state))
+
+      override def deserializeCalculatedState(bytes: Array[Byte]): IO[Either[Throwable, CheckInDataCalculatedState]] = IO(Deserializers.deserializeCalculatedState(bytes))
+    })
+
+  private def makeL1Service: IO[BaseDataApplicationL1Service[IO]] = {
+    for {
+      calculatedStateService <- CalculatedStateService.make[IO]
+      dataApplicationL1Service = makeBaseDataApplicationL1Service(calculatedStateService)
+    } yield dataApplicationL1Service
+  }
+
+  override def dataApplication: Option[Resource[IO, BaseDataApplicationL1Service[IO]]] =
+    makeL1Service.asResource.some
 
-    override def hashCalculatedState(state: CheckInDataCalculatedState)(implicit context: L1NodeContext[IO]): IO[Hash] = CalculatedState.hashCalculatedState(state)
-  }))
 }
diff --git a/metagraph/modules/l0/src/main/scala/com/my/dor_metagraph/l0/Main.scala b/metagraph/modules/l0/src/main/scala/com/my/dor_metagraph/l0/Main.scala
@@ -1,12 +1,12 @@
 package com.my.dor_metagraph.l0
 
 import cats.data.NonEmptyList
-import cats.effect.IO
-import cats.implicits.catsSyntaxValidatedIdBinCompat0
+import cats.effect.{IO, Resource}
+import cats.implicits.{catsSyntaxOptionId, catsSyntaxValidatedIdBinCompat0}
 import com.my.dor_metagraph.l0.custom_routes.CustomRoutes.getLatestCalculatedState
 import com.my.dor_metagraph.l0.rewards.MainRewards
 import com.my.dor_metagraph.shared_data.Data
-import com.my.dor_metagraph.shared_data.calculated_state.CalculatedState
+import com.my.dor_metagraph.shared_data.calculated_state.CalculatedStateService
 import com.my.dor_metagraph.shared_data.decoders.Decoders
 import com.my.dor_metagraph.shared_data.deserializers.Deserializers
 import com.my.dor_metagraph.shared_data.serializers.Serializers
@@ -25,8 +25,8 @@ import org.tessellation.sdk.domain.rewards.Rewards
 import org.tessellation.security.SecurityProvider
 import org.tessellation.security.hash.Hash
 import org.tessellation.security.signature.Signed
-
 import org.http4s.dsl.io._
+import org.tessellation.ext.cats.effect.ResourceIO
 
 import java.util.UUID
 
@@ -37,8 +37,10 @@ object Main
     ClusterId(UUID.fromString("517c3a05-9219-471b-a54c-21b7d72f4ae5")),
     version = BuildInfo.version
   ) {
-  def dataApplication: Option[BaseDataApplicationL0Service[IO]] =
-    Option(BaseDataApplicationL0Service(new DataApplicationL0Service[IO, CheckInUpdate, CheckInStateOnChain, CheckInDataCalculatedState] {
+  private def makeBaseDataApplicationL0Service(
+    calculatedStateService: CalculatedStateService[IO]
+  ): BaseDataApplicationL0Service[IO] =
+    BaseDataApplicationL0Service(new DataApplicationL0Service[IO, CheckInUpdate, CheckInStateOnChain, CheckInDataCalculatedState] {
       override def genesis: DataState[CheckInStateOnChain, CheckInDataCalculatedState] = DataState(CheckInStateOnChain(List.empty), CheckInDataCalculatedState(Map.empty))
 
       override def validateUpdate(update: CheckInUpdate)(implicit context: L0NodeContext[IO]): IO[DataApplicationValidationErrorOr[Unit]] = IO.pure(().validNec)
@@ -69,18 +71,31 @@ object Main
 
       override def deserializeUpdate(bytes: Array[Byte]): IO[Either[Throwable, CheckInUpdate]] = IO(Deserializers.deserializeUpdate(bytes))
 
-      override def getCalculatedState(implicit context: L0NodeContext[IO]): IO[(SnapshotOrdinal, CheckInDataCalculatedState)] = CalculatedState.getCalculatedState
+      override def getCalculatedState(implicit context: L0NodeContext[IO]): IO[(SnapshotOrdinal, CheckInDataCalculatedState)] = calculatedStateService.getCalculatedState.map(calculatedState => (calculatedState.ordinal, calculatedState.state))
 
-      override def setCalculatedState(ordinal: SnapshotOrdinal, state: CheckInDataCalculatedState)(implicit context: L0NodeContext[IO]): IO[Boolean] = CalculatedState.setCalculatedState(ordinal, state)
+      override def setCalculatedState(ordinal: SnapshotOrdinal, state: CheckInDataCalculatedState)(implicit context: L0NodeContext[IO]): IO[Boolean] = calculatedStateService.setCalculatedState(ordinal, state)
 
-      override def hashCalculatedState(state: CheckInDataCalculatedState)(implicit context: L0NodeContext[IO]): IO[Hash] = CalculatedState.hashCalculatedState(state)
+      override def hashCalculatedState(state: CheckInDataCalculatedState)(implicit context: L0NodeContext[IO]): IO[Hash] = calculatedStateService.hashCalculatedState(state)
 
       override def routes(implicit context: L0NodeContext[IO]): HttpRoutes[IO] = HttpRoutes.of {
-        case GET -> Root / "calculated-state" / "latest" => getLatestCalculatedState
+        case GET -> Root / "calculated-state" / "latest" => getLatestCalculatedState(calculatedStateService)
       }
-    }))
 
-  def rewards(implicit sp: SecurityProvider[IO]): Option[Rewards[IO, CurrencySnapshotStateProof, CurrencyIncrementalSnapshot, CurrencySnapshotEvent]] = Some(
-    MainRewards.make[IO]
-  )
+      override def serializeCalculatedState(state: CheckInDataCalculatedState): IO[Array[Byte]] = IO(Serializers.serializeCalculatedState(state))
+
+      override def deserializeCalculatedState(bytes: Array[Byte]): IO[Either[Throwable, CheckInDataCalculatedState]] = IO(Deserializers.deserializeCalculatedState(bytes))
+    })
+
+  private def makeL0Service: IO[BaseDataApplicationL0Service[IO]] = {
+    for {
+      calculatedStateService <- CalculatedStateService.make[IO]
+      dataApplicationL0Service = makeBaseDataApplicationL0Service(calculatedStateService)
+    } yield dataApplicationL0Service
+  }
+
+  override def dataApplication: Option[Resource[IO, BaseDataApplicationL0Service[IO]]] =
+    makeL0Service.asResource.some
+
+  override def rewards(implicit sp: SecurityProvider[IO]): Option[Rewards[IO, CurrencySnapshotStateProof, CurrencyIncrementalSnapshot, CurrencySnapshotEvent]] =
+    MainRewards.make[IO].some
 }
diff --git a/metagraph/modules/l0/src/main/scala/com/my/dor_metagraph/l0/custom_routes/CustomRoutes.scala b/metagraph/modules/l0/src/main/scala/com/my/dor_metagraph/l0/custom_routes/CustomRoutes.scala
@@ -2,7 +2,7 @@ package com.my.dor_metagraph.l0.custom_routes
 
 import cats.effect.IO
 import cats.effect.unsafe.implicits.global
-import com.my.dor_metagraph.shared_data.calculated_state.CalculatedState.getCalculatedState
+import com.my.dor_metagraph.shared_data.calculated_state.CalculatedStateService
 import com.my.dor_metagraph.shared_data.types.Types.CheckInDataCalculatedState
 import derevo.circe.magnolia.{decoder, encoder}
 import derevo.derive
@@ -13,9 +13,9 @@ import org.http4s.dsl.io._
 object CustomRoutes {
   @derive(encoder, decoder)
   case class CalculatedStateResponse(ordinal: Long, calculatedState:CheckInDataCalculatedState)
-  def getLatestCalculatedState: IO[Response[IO]] = {
-    val calculatedState = getCalculatedState
-    val response = calculatedState.map(state => CalculatedStateResponse(state._1.value.value, state._2))
+  def getLatestCalculatedState(calculatedStateService: CalculatedStateService[IO]): IO[Response[IO]] = {
+    val calculatedState = calculatedStateService.getCalculatedState
+    val response = calculatedState.map(state => CalculatedStateResponse(state.ordinal.value.value, state.state))
     Ok(response.unsafeRunSync())
   }
 }
diff --git a/metagraph/modules/l0/src/main/scala/com/my/dor_metagraph/l0/rewards/MainRewards.scala b/metagraph/modules/l0/src/main/scala/com/my/dor_metagraph/l0/rewards/MainRewards.scala
@@ -6,20 +6,22 @@ import com.my.dor_metagraph.l0.rewards.BountyRewards.getBountyRewardsTransaction
 import com.my.dor_metagraph.l0.rewards.ValidatorNodesRewards.getValidatorNodesTransactions
 import com.my.dor_metagraph.shared_data.combiners.ValidatorNodes.getValidatorNodes
 import com.my.dor_metagraph.shared_data.types.Types.{CheckInDataCalculatedState, EPOCH_PROGRESS_1_DAY}
+import eu.timepit.refined.types.all.PosLong
 import org.slf4j.LoggerFactory
 import org.tessellation.currency.dataApplication.DataCalculatedState
 import org.tessellation.currency.l0.snapshot.CurrencySnapshotEvent
 import org.tessellation.currency.schema.currency.{CurrencyIncrementalSnapshot, CurrencySnapshotStateProof}
 import org.tessellation.schema.address.Address
 import org.tessellation.schema.balance.Balance
 import org.tessellation.schema.transaction
-import org.tessellation.schema.transaction.{RewardTransaction, Transaction}
+import org.tessellation.schema.transaction.{RewardTransaction, Transaction, TransactionAmount}
 import org.tessellation.sdk.domain.rewards.Rewards
 import org.tessellation.sdk.infrastructure.consensus.trigger
 import org.tessellation.sdk.infrastructure.consensus.trigger.ConsensusTrigger
 import org.tessellation.security.SecurityProvider
 import org.tessellation.security.signature.Signed
 
+import scala.collection.MapView
 import scala.collection.immutable.{SortedMap, SortedSet}
 
 object MainRewards {
@@ -64,11 +66,25 @@ object MainRewards {
     buildRewardsTransactionsSortedSet(bountyTransactions, validatorNodesTransactions).pure
   }
 
-  private def buildRewardsTransactionsSortedSet(bountyTransactions: List[RewardTransaction], validatorNodesTransactions: List[RewardTransaction]): SortedSet[RewardTransaction] = {
+  private def buildRewardsTransactionsSortedSet(
+    bountyTransactions        : List[RewardTransaction],
+    validatorNodesTransactions: List[RewardTransaction]
+  ): SortedSet[RewardTransaction] = {
     val allTransactions = bountyTransactions ::: validatorNodesTransactions
-    val allTransactionsFiltered = allTransactions.filter(_.amount.value.value > 0)
+    val groupedTransactions: MapView[Address, Long] =
+      allTransactions
+        .filter(_.amount.value.value > 0)
+        .groupBy(_.destination)
+        .view
+        .mapValues(_.map(_.amount.value.value).sum)
 
-    SortedSet(allTransactionsFiltered: _*)
+    val summedTransactions: List[RewardTransaction] =
+      groupedTransactions.map {
+        case (address, totalAmount) =>
+          RewardTransaction(address, TransactionAmount(PosLong.unsafeFrom(totalAmount)))
+      }.toList
+
+    SortedSet(summedTransactions: _*)
   }
 
   private def logInitialRewardDistribution(epochProgressModulus: Long): Unit = {