fix: updating block explorer fetching limit and changing the validati…

…ons of streak to combine function (cherry picked from commit 31780a3)
Constellation-Labs · Jan 6, 2025 · a741bd9 · a741bd9
1 parent 9b9baf0
commit a741bd9
Show file tree

Hide file tree

Showing 6 changed files with 79 additions and 98 deletions.
diff --git a/...hared_data/src/main/scala/org/elpaca_metagraph/shared_data/LifecycleSharedFunctions.scala b/...hared_data/src/main/scala/org/elpaca_metagraph/shared_data/LifecycleSharedFunctions.scala
@@ -12,10 +12,10 @@ import org.elpaca_metagraph.shared_data.combiners.OutflowTransactionsCombiner.cl
 import org.elpaca_metagraph.shared_data.combiners.WalletCreationHoldingDAGCombiner.cleanWalletCreationHoldingDAGAlreadyRewardedWallets
 import org.elpaca_metagraph.shared_data.combiners.XCombiner.updateRewardsOlderThanOneDay
 import org.elpaca_metagraph.shared_data.combiners.YouTubeCombiner.updateYoutubeRewardsOlderThanOneDay
-import org.elpaca_metagraph.shared_data.types.DataUpdates.{ElpacaUpdate, IntegrationnetNodeOperatorUpdate, StreakUpdate}
+import org.elpaca_metagraph.shared_data.types.DataUpdates.{ElpacaUpdate, IntegrationnetNodeOperatorUpdate}
 import org.elpaca_metagraph.shared_data.types.States.{ElpacaCalculatedState, ElpacaOnChainState}
 import org.elpaca_metagraph.shared_data.validations.Errors.valid
-import org.elpaca_metagraph.shared_data.validations.Validations.{integrationnetNodeOperatorsValidationsL1, streakValidationsL0}
+import org.elpaca_metagraph.shared_data.validations.Validations.integrationnetNodeOperatorsValidationsL1
 import org.tessellation.currency.dataApplication.dataApplication.DataApplicationValidationErrorOr
 import org.tessellation.currency.dataApplication.{DataState, L0NodeContext}
 import org.tessellation.security.signature.Signed
@@ -40,18 +40,8 @@ object LifecycleSharedFunctions {
     updates  : NonEmptyList[Signed[ElpacaUpdate]],
     oldState : DataState[ElpacaOnChainState, ElpacaCalculatedState],
     appConfig: ApplicationConfig
-  )(implicit context: L0NodeContext[F]): F[DataApplicationValidationErrorOr[Unit]] = {
-    updates.traverse { update =>
-      update.value match {
-        case streakUpdate: StreakUpdate =>
-          for {
-            epochProgress <- getCurrentEpochProgress
-          } yield streakValidationsL0(Signed(streakUpdate, update.proofs), oldState, appConfig, epochProgress)
-        case _ => valid.pure[F]
-      }
-    }.map(_.reduce)
-
-  }
+  )(implicit context: L0NodeContext[F]): F[DataApplicationValidationErrorOr[Unit]] =
+    valid.pure[F]
 
   def combine[F[_] : Async](
     oldState : DataState[ElpacaOnChainState, ElpacaCalculatedState],

diff --git a/modules/shared_data/src/main/scala/org/elpaca_metagraph/shared_data/combiners/Combiner.scala b/modules/shared_data/src/main/scala/org/elpaca_metagraph/shared_data/combiners/Combiner.scala
@@ -112,8 +112,7 @@ object Combiner {
         val updatedStreakDataSource = updateStateStreak(
           currentDataSources,
           currentEpochProgress,
-          streakUpdate,
-          update.proofs,
+          Signed(streakUpdate, update.proofs),
           applicationConfig
         ).map(_.asInstanceOf[DataSource])
 

diff --git a/...hared_data/src/main/scala/org/elpaca_metagraph/shared_data/combiners/StreakCombiner.scala b/...hared_data/src/main/scala/org/elpaca_metagraph/shared_data/combiners/StreakCombiner.scala
@@ -1,6 +1,6 @@
 package org.elpaca_metagraph.shared_data.combiners
 
-import cats.data.NonEmptySet
+import cats.data.Validated
 import cats.effect.Async
 import cats.syntax.all._
 import eu.timepit.refined.types.numeric.NonNegLong
@@ -10,9 +10,10 @@ import org.elpaca_metagraph.shared_data.app.ApplicationConfig
 import org.elpaca_metagraph.shared_data.types.DataUpdates._
 import org.elpaca_metagraph.shared_data.types.States._
 import org.elpaca_metagraph.shared_data.types.Streak.StreakDataSourceAddress
+import org.elpaca_metagraph.shared_data.validations.Validations.streakValidationsL0
 import org.tessellation.schema.balance.Amount
 import org.tessellation.schema.epoch.EpochProgress
-import org.tessellation.security.signature.signature.SignatureProof
+import org.tessellation.security.signature.Signed
 import org.typelevel.log4cats.Logger
 
 object StreakCombiner {
@@ -35,88 +36,86 @@ object StreakCombiner {
   def updateStateStreak[F[_] : Async : Logger](
     currentCalculatedState: Map[DataSourceType, DataSource],
     currentEpochProgress  : EpochProgress,
-    streakUpdate          : StreakUpdate,
-    proofs                : NonEmptySet[SignatureProof],
+    signedStreakUpdate    : Signed[StreakUpdate],
     applicationConfig     : ApplicationConfig
   ): F[StreakDataSource] = {
     val streakDataSource = getCurrentStreakDataSource(currentCalculatedState)
-    val isSignedByStargazer = walletSignedTheMessage(applicationConfig.streak.stargazerPublicKey, proofs)
-    if (!isSignedByStargazer) {
-      Logger[F].warn(s"The update $streakUpdate was not signed by stargazer, ignoring").as(
-        streakDataSource
-      )
-    } else {
-      case class StreakInfo(streakDays: NonNegLong, rewardAmount: Amount)
+    streakValidationsL0(signedStreakUpdate, streakDataSource, applicationConfig, currentEpochProgress) match {
+      case Validated.Invalid(errors) =>
+        Logger[F].warn(s"Could not update streak, reasons: $errors").as(streakDataSource)
+      case Validated.Valid(_) =>
+        case class StreakInfo(streakDays: NonNegLong, rewardAmount: Amount)
 
-      val streakDataSourceAddress = streakDataSource.existingWallets
-        .get(streakUpdate.address) match {
-        case Some(streakDataSourceAddress) => streakDataSourceAddress
-        case None => StreakDataSourceAddress.empty
-      }
+        val streakUpdate = signedStreakUpdate.value
+        val streakDataSourceAddress =
+          streakDataSource.existingWallets.get(streakUpdate.address) match {
+            case Some(streakDataSourceAddress) => streakDataSourceAddress
+            case None => StreakDataSourceAddress.empty
+          }
 
-      if (!isNewDay(streakDataSourceAddress.epochProgressToReward, currentEpochProgress)) {
-        Logger[F].info(s"This address already claimed reward of the day, ignoring").as(
-          streakDataSource
-        )
-      } else {
-        def shouldResetStreak: Boolean =
-          currentEpochProgress.value.value - streakDataSourceAddress.epochProgressToReward.value.value > (epochProgressOneDay * 2)
+        if (!isNewDay(streakDataSourceAddress.epochProgressToReward, currentEpochProgress)) {
+          Logger[F].warn(s"This address already claimed reward of the day, ignoring").as(
+            streakDataSource
+          )
+        } else {
+          def shouldResetStreak: Boolean =
+            currentEpochProgress.value.value - streakDataSourceAddress.epochProgressToReward.value.value > (epochProgressOneDay * 2)
 
-        def streakBetween(min: Long, max: Long, updatedStreakDays: NonNegLong): Boolean =
-          updatedStreakDays.value >= min && updatedStreakDays.value <= max
+          def streakBetween(min: Long, max: Long, updatedStreakDays: NonNegLong): Boolean =
+            updatedStreakDays.value >= min && updatedStreakDays.value <= max
 
-        def calculateRewardAmount(streakDays: NonNegLong): Amount = {
-          if (streakBetween(1L, 4L, streakDays)) Amount(level1Streak)
-          else if (streakBetween(5L, 10L, streakDays)) Amount(level2Streak)
-          else Amount(level3Streak)
-        }
+          def calculateRewardAmount(streakDays: NonNegLong): Amount = {
+            if (streakBetween(1L, 4L, streakDays)) Amount(level1Streak)
+            else if (streakBetween(5L, 10L, streakDays)) Amount(level2Streak)
+            else Amount(level3Streak)
+          }
 
-        def getUpdateStreakInfo(streakDays: NonNegLong, ignoreReset: Boolean): StreakInfo = {
-          if (shouldResetStreak && !ignoreReset) {
-            StreakInfo(level1Streak, toTokenAmountFormat(Amount(level1Streak)))
-          } else {
-            val updatedStreakDays = NonNegLong.unsafeFrom(streakDays.value + 1L)
-            val rewardAmount = toTokenAmountFormat(calculateRewardAmount(updatedStreakDays))
-            StreakInfo(updatedStreakDays, rewardAmount)
+          def getUpdateStreakInfo(streakDays: NonNegLong, ignoreReset: Boolean): StreakInfo = {
+            if (shouldResetStreak && !ignoreReset) {
+              StreakInfo(level1Streak, toTokenAmountFormat(Amount(level1Streak)))
+            } else {
+              val updatedStreakDays = NonNegLong.unsafeFrom(streakDays.value + 1L)
+              val rewardAmount = toTokenAmountFormat(calculateRewardAmount(updatedStreakDays))
+              StreakInfo(updatedStreakDays, rewardAmount)
+            }
           }
-        }
 
-        def updateStreakDataSource(streakInfo: StreakInfo, nextStreakInfo: StreakInfo): F[StreakDataSourceAddress] = {
-          val totalEarned = Amount(
-            NonNegLong.unsafeFrom(streakDataSourceAddress.totalEarned.value.value + streakInfo.rewardAmount.value.value)
-          )
+          def updateStreakDataSource(streakInfo: StreakInfo, nextStreakInfo: StreakInfo): F[StreakDataSourceAddress] = {
+            val totalEarned = Amount(
+              NonNegLong.unsafeFrom(streakDataSourceAddress.totalEarned.value.value + streakInfo.rewardAmount.value.value)
+            )
 
-          randomString(
-            s"${streakUpdate.address.value.value}-${currentEpochProgress.value.value.toString}",
-            randomStringLength
-          ).map { token =>
-            streakDataSourceAddress
-              .focus(_.dailyEpochProgress)
-              .replace(currentEpochProgress)
-              .focus(_.epochProgressToReward)
-              .replace(currentEpochProgress)
-              .focus(_.amountToReward)
-              .replace(streakInfo.rewardAmount)
-              .focus(_.totalEarned)
-              .replace(totalEarned)
-              .focus(_.nextClaimReward)
-              .replace(nextStreakInfo.rewardAmount)
-              .focus(_.streakDays)
-              .replace(streakInfo.streakDays)
-              .focus(_.nextToken)
-              .replace(token.some)
+            randomString(
+              s"${streakUpdate.address.value.value}-${currentEpochProgress.value.value.toString}",
+              randomStringLength
+            ).map { token =>
+              streakDataSourceAddress
+                .focus(_.dailyEpochProgress)
+                .replace(currentEpochProgress)
+                .focus(_.epochProgressToReward)
+                .replace(currentEpochProgress)
+                .focus(_.amountToReward)
+                .replace(streakInfo.rewardAmount)
+                .focus(_.totalEarned)
+                .replace(totalEarned)
+                .focus(_.nextClaimReward)
+                .replace(nextStreakInfo.rewardAmount)
+                .focus(_.streakDays)
+                .replace(streakInfo.streakDays)
+                .focus(_.nextToken)
+                .replace(token.some)
+            }
           }
-        }
 
-        val currentStreakInfo = getUpdateStreakInfo(streakDataSourceAddress.streakDays, ignoreReset = false)
-        val nextStreakInfo = getUpdateStreakInfo(currentStreakInfo.streakDays, ignoreReset = true)
-        updateStreakDataSource(currentStreakInfo, nextStreakInfo).flatMap { updatedDataSourceAddress =>
-          Logger[F].info(s"Claiming reward of the address ${streakUpdate.address}. Streak: ${currentStreakInfo.streakDays}").as(
-            streakDataSource
-              .focus(_.existingWallets)
-              .modify(_.updated(streakUpdate.address, updatedDataSourceAddress)))
+          val currentStreakInfo = getUpdateStreakInfo(streakDataSourceAddress.streakDays, ignoreReset = false)
+          val nextStreakInfo = getUpdateStreakInfo(currentStreakInfo.streakDays, ignoreReset = true)
+          updateStreakDataSource(currentStreakInfo, nextStreakInfo).flatMap { updatedDataSourceAddress =>
+            Logger[F].info(s"Claiming reward of the address ${streakUpdate.address}. Streak: ${currentStreakInfo.streakDays}").as(
+              streakDataSource
+                .focus(_.existingWallets)
+                .modify(_.updated(streakUpdate.address, updatedDataSourceAddress)))
+          }
         }
-      }
     }
   }
 }
diff --git a/...in/scala/org/elpaca_metagraph/shared_data/daemons/fetcher/InflowTransactionsFetcher.scala b/...in/scala/org/elpaca_metagraph/shared_data/daemons/fetcher/InflowTransactionsFetcher.scala
@@ -80,7 +80,7 @@ object InflowTransactionsFetcher {
         val inflowConfig = applicationConfig.inflowTransactionsDaemon
 
         val walletsUpdates = inflowConfig.walletsInfo.traverse { walletInfo =>
-          val url = s"${inflowConfig.apiUrl.get}/addresses/${walletInfo.address}/transactions/received?limit=10000"
+          val url = s"${inflowConfig.apiUrl.get}/addresses/${walletInfo.address}/transactions/received?limit=5000"
 
           for {
             _ <- logger.info(s"Inflow - Fetching from block explorer using URL: $url")

diff --git a/...n/scala/org/elpaca_metagraph/shared_data/daemons/fetcher/OutflowTransactionsFetcher.scala b/...n/scala/org/elpaca_metagraph/shared_data/daemons/fetcher/OutflowTransactionsFetcher.scala
@@ -80,7 +80,7 @@ object OutflowTransactionsFetcher {
         val outflowConfig = applicationConfig.outflowTransactionsDaemon
 
         val walletsUpdates = outflowConfig.walletsInfo.traverse { walletInfo =>
-          val url = s"${outflowConfig.apiUrl.get}/addresses/${walletInfo.address}/transactions/sent?limit=10000"
+          val url = s"${outflowConfig.apiUrl.get}/addresses/${walletInfo.address}/transactions/sent?limit=5000"
 
           for {
             _ <- logger.info(s"Outflow - Fetching from block explorer using URL: $url")

diff --git a/...shared_data/src/main/scala/org/elpaca_metagraph/shared_data/validations/Validations.scala b/...shared_data/src/main/scala/org/elpaca_metagraph/shared_data/validations/Validations.scala
@@ -3,10 +3,8 @@ package org.elpaca_metagraph.shared_data.validations
 import cats.syntax.all._
 import org.elpaca_metagraph.shared_data.app.ApplicationConfig
 import org.elpaca_metagraph.shared_data.types.DataUpdates.{IntegrationnetNodeOperatorUpdate, StreakUpdate}
-import org.elpaca_metagraph.shared_data.types.States.DataSourceType.Streak
-import org.elpaca_metagraph.shared_data.types.States.{ElpacaCalculatedState, ElpacaOnChainState, StreakDataSource}
+import org.elpaca_metagraph.shared_data.types.States.StreakDataSource
 import org.elpaca_metagraph.shared_data.validations.TypeValidators._
-import org.tessellation.currency.dataApplication.DataState
 import org.tessellation.currency.dataApplication.dataApplication.DataApplicationValidationErrorOr
 import org.tessellation.schema.epoch.EpochProgress
 import org.tessellation.security.signature.Signed
@@ -19,16 +17,11 @@ object Validations {
 
   def streakValidationsL0(
     streakUpdate        : Signed[StreakUpdate],
-    oldState            : DataState[ElpacaOnChainState, ElpacaCalculatedState],
+    streakDataSource    : StreakDataSource,
     appConfig           : ApplicationConfig,
     currentEpochProgress: EpochProgress
   ): DataApplicationValidationErrorOr[Unit] = {
-    val streakDataSource = oldState.calculated.dataSources
-      .get(Streak)
-      .collect { case ds: StreakDataSource => ds }
-      .getOrElse(StreakDataSource(Map.empty))
-
-      validateIfUpdateWasSignedByStargazer(streakUpdate.proofs, appConfig)
+    validateIfUpdateWasSignedByStargazer(streakUpdate.proofs, appConfig)
       .productR(validateIfAddressAlreadyRewardedInCurrentDay(streakUpdate.value, streakDataSource, currentEpochProgress))
       .productR(validateIfTokenIsValid(streakUpdate.value, streakDataSource))