[ci] pre-calculating total amulet balance (#2407)

---------

Signed-off-by: Raymond Roestenburg <raymond.roestenburg@digitalasset.com>

Author: ray-roestenburg-da

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTimeBasedIntegrationTest.scala

@@ -364,11 +364,11 @@ class ScanTimeBasedIntegrationTest
       val total0 =
         sv1ScanBackend
           .getTotalAmuletBalance(firstRound + 0)
-          .valueOrFail("Amulet balance not yet computed")
+          .getOrElse(BigDecimal(0))
       val total1 =
         sv1ScanBackend
           .getTotalAmuletBalance(firstRound + 1)
-          .valueOrFail("Amulet balance not yet computed")
+          .getOrElse(BigDecimal(0))
       val total2 =
         sv1ScanBackend
           .getTotalAmuletBalance(firstRound + 2)

apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V046__scan_round_total_amulet_balance.sql

@@ -0,0 +1,19 @@
+-- aggregation result table for total amulet balance changes per store and closed round
+-- this table only contains rows for rounds where the sums changed
+create table round_total_amulet_balance
+(
+    -- the store id of the scan store for which the totals are calculated
+    store_id             int not null,
+    -- the closed round for which the sums are calculated.
+    closed_round         bigint not null,
+    -- the sum of cumulative change_to_initial_amount_as_of_round_zero from round zero up to and including the round
+    sum_cumulative_change_to_initial_amount_as_of_round_zero numeric not null,
+    -- the sum of cumulative change_to_holding_fees_rate from round zero up to and including the round
+    sum_cumulative_change_to_holding_fees_rate numeric not null,
+    primary key (store_id, closed_round)
+);
+
+-- round_total_amulet_balance does not have a sums for rounds when nothing changed.
+-- this index is for efficiently finding an earlier round and calculating the total balance as of a given round
+create index idx_round_total_amulet_balance_sid_cr_desc
+    on round_total_amulet_balance (store_id, closed_round desc);

apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/db/SpliceDbTest.scala

@@ -62,6 +62,7 @@ trait SpliceDbTest extends DbTest with BeforeAndAfterAll { this: Suite =>
         for {
           _ <- debugPrintPgActivity()
           _ <- sql"""TRUNCATE
+                active_parties,
                 user_wallet_acs_store,
                 user_wallet_acs_interface_views,
                 user_wallet_txlog_store,
@@ -78,6 +79,7 @@ trait SpliceDbTest extends DbTest with BeforeAndAfterAll { this: Suite =>
                 store_last_ingested_offsets,
                 round_totals,
                 round_party_totals,
+                round_total_amulet_balance,
                 update_history_descriptors,
                 update_history_last_ingested_offsets,
                 update_history_transactions,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala

@@ -33,13 +33,14 @@ import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   AcsRequest,
   BatchListVotesByVoteRequestsRequest,
   DamlValueEncoding,
+  ErrorResponse,
+  EventHistoryRequest,
   HoldingsStateRequest,
   HoldingsSummaryRequest,
   ListVoteResultsRequest,
   MaybeCachedContractWithState,
   UpdateHistoryItemV2,
   UpdateHistoryRequestV2,
-  EventHistoryRequest,
 }
 import org.lfdecentralizedtrust.splice.http.v0.scan.ScanResource
 import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as v0}
@@ -49,7 +50,6 @@ import org.lfdecentralizedtrust.splice.scan.store.{
   ScanStore,
   TxLogEntry,
 }
-
 import org.lfdecentralizedtrust.splice.util.{
   Codec,
   Contract,
@@ -413,8 +413,15 @@ class HttpScanHandler(
             HttpErrorHandler.onGrpcNotFound(s"Data for round ${asOfEndOfRound} not yet computed")
           )
       } yield {
-        definitions.GetTotalAmuletBalanceResponse(
-          Codec.encode(total)
+        total.fold(
+          v0.ScanResource.GetTotalAmuletBalanceResponse
+            .NotFound(ErrorResponse(s"No total amulet balance found for round $asOfEndOfRound"))
+        )(total =>
+          v0.ScanResource.GetTotalAmuletBalanceResponse.OK(
+            definitions.GetTotalAmuletBalanceResponse(
+              Codec.encode(total)
+            )
+          )
         )
       }
     }
@@ -447,7 +454,7 @@ class HttpScanHandler(
     withSpan(s"$workflowId.getAmuletConfigForRound") { _ => _ =>
       store
         .getAmuletConfigForRound(round)
-        .map(cfg => {
+        .map { cfg =>
           val transferFee = cfg.transferFee.getOrElse(throw new RuntimeException("No transfer fee"))
           v0.ScanResource.GetAmuletConfigForRoundResponse.OK(
             definitions.GetAmuletConfigForRoundResponse(
@@ -464,7 +471,7 @@ class HttpScanHandler(
               ),
             )
           )
-        })
+        }
         .transform(HttpErrorHandler.onGrpcNotFound(s"Round ${round} not found"))
     }
   }

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpTokenStandardMetadataHandler.scala

@@ -70,7 +70,7 @@ class HttpTokenStandardMetadataHandler(
   private def lookupTotalSupplyByLatestRound()(implicit ec: ExecutionContext, tc: TraceContext) =
     for {
       (latestRoundNr, effectiveAt) <- OptionT(store.lookupRoundOfLatestData())
-      totalSupply <- OptionT.liftF(store.getTotalAmuletBalance(latestRoundNr))
+      totalSupply <- OptionT(store.getTotalAmuletBalance(latestRoundNr))
     } yield TotalSupply(amount = totalSupply, asOfTimestamp = effectiveAt)
 
   private def lookupTotalSupplyByLatestAcsSnapshot()(implicit tc: TraceContext) = {

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/CachingScanStore.scala

@@ -70,7 +70,7 @@ class CachingScanStore(
   // TODO(#800) remove when amulet expiry works again
   override def getTotalAmuletBalance(
       asOfEndOfRound: Long
-  )(implicit tc: TraceContext): Future[BigDecimal] = {
+  )(implicit tc: TraceContext): Future[Option[BigDecimal]] = {
     getCache(
       "totalAmuletBalance",
       cacheConfig.totalAmuletBalance,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala

@@ -154,7 +154,9 @@ trait ScanStore
       tc: TraceContext
   ): Future[Option[ContractWithState[splice.ans.AnsRules.ContractId, splice.ans.AnsRules]]]
 
-  def getTotalAmuletBalance(asOfEndOfRound: Long)(implicit tc: TraceContext): Future[BigDecimal]
+  def getTotalAmuletBalance(asOfEndOfRound: Long)(implicit
+      tc: TraceContext
+  ): Future[Option[BigDecimal]]
 
   def getTotalRewardsCollectedEver()(implicit tc: TraceContext): Future[BigDecimal]
   def getRewardsCollectedInRound(round: Long)(implicit tc: TraceContext): Future[BigDecimal]

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala

@@ -593,51 +593,30 @@ class DbScanStore(
 
   override def getTotalAmuletBalance(asOfEndOfRound: Long)(implicit
       tc: TraceContext
-  ): Future[BigDecimal] =
+  ): Future[Option[BigDecimal]] =
     waitUntilAcsIngested {
       for {
-        result <- ensureAggregated(asOfEndOfRound) { lastAggregatedRound =>
-          // if the asOfEndOrRound is the latest aggregated round, we can use the active_parties for a faster query.
-          if (lastAggregatedRound == asOfEndOfRound) {
-            // using greatest(0, ...) to handle negative balances caused by amulets never expiring.
-            storage.query(
-              // TODO(#800) change to query from round_totals when amulet expiry works again
-              sql"""
-              select sum(greatest(0, rpt.cumulative_change_to_initial_amount_as_of_round_zero - rpt.cumulative_change_to_holding_fees_rate * ($asOfEndOfRound + 1)))
-              from    round_party_totals rpt
-              join    active_parties ap
-              on      rpt.store_id = ap.store_id
-              and     rpt.party = ap.party
-              and     rpt.closed_round = ap.closed_round
-              where   rpt.store_id = $roundTotalsStoreId;
-              """.as[Option[BigDecimal]].headOption,
-              "getTotalAmuletBalanceForLastAggregatedRound",
-            )
-          } else {
-            // fall back if the requested round is not the latest aggregated round.
-            storage.query(
-              // TODO(#800) change to query from round_totals when amulet expiry works again
-              sql"""
-              with most_recent as (
-                select   max(closed_round) as closed_round,
-                party
-                  from     round_party_totals
-                  where    store_id = $roundTotalsStoreId
-                  and      closed_round <= $asOfEndOfRound
-                  group by party
+        result <- ensureAggregated(asOfEndOfRound) { _ =>
+          storage.query(
+            // TODO(#800) change to query from round_totals when amulet expiry works again
+            // the round_total_amulet_balance is sparse and might not have an entry for the requested round
+            // which is why the first entry found <= asOfEndOfRound is used
+            sql"""
+              select greatest(
+                0,
+                sum_cumulative_change_to_initial_amount_as_of_round_zero -
+                sum_cumulative_change_to_holding_fees_rate * ($asOfEndOfRound + 1)
               )
-              select sum(greatest(0, rpt.cumulative_change_to_initial_amount_as_of_round_zero - rpt.cumulative_change_to_holding_fees_rate * ($asOfEndOfRound + 1)))
-              from   round_party_totals rpt,
-              most_recent mr
-              where  rpt.store_id = $roundTotalsStoreId
-              and    rpt.party = mr.party
-              and    rpt.closed_round = mr.closed_round;
-              """.as[Option[BigDecimal]].headOption,
-              "getTotalAmuletBalanceForEarlierRound",
-            )
-          }
+              from    round_total_amulet_balance
+              where   store_id = $roundTotalsStoreId
+              and     closed_round <= $asOfEndOfRound
+              order by closed_round desc
+              limit 1;
+              """.as[BigDecimal].headOption,
+            "getTotalAmuletBalance",
+          )
         }
-      } yield result.flatten.getOrElse(0)
+      } yield result
     }
 
   override def getTotalRewardsCollectedEver()(implicit tc: TraceContext): Future[BigDecimal] =