support querying next snapshot after a timestamp (#3552)

Signed-off-by: Itai Segall <itai.segall@digitalasset.com>

Author: isegall-da

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala

@@ -394,6 +394,16 @@ abstract class ScanAppReference(
       )
     }
 
+  def getDateOfFirstSnapshotAfter(after: CantonTimestamp, migrationId: Long) =
+    consoleEnvironment.run {
+      httpCommand(
+        HttpScanAppClient.GetDateOfFirstSnapshotAfter(
+          after.toInstant.atOffset(java.time.ZoneOffset.UTC),
+          migrationId,
+        )
+      )
+    }
+
   def getAcsSnapshotAt(
       at: CantonTimestamp,
       migrationId: Long,

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

@@ -446,11 +446,24 @@ class ScanTimeBasedIntegrationTest
       }
     }
 
-    val snapshotBefore = sv1ScanBackend.getDateOfMostRecentSnapshotBefore(
-      getLedgerTime,
-      migrationId,
+    val startTime = getLedgerTime
+
+    advanceTime(
+      java.time.Duration
+        .ofHours(sv1ScanBackend.config.acsSnapshotPeriodHours.toLong)
+        .plusSeconds(1L)
     )
 
+    val snapshot1 = eventually() {
+      val snapshot1 = sv1ScanBackend.getDateOfMostRecentSnapshotBefore(
+        getLedgerTime,
+        migrationId,
+      )
+      snapshot1 should not be None
+      snapshot1.value.toInstant shouldBe >(startTime.toInstant)
+      snapshot1
+    }
+
     createAnsEntry(
       aliceAnsExternalClient,
       perTestCaseName("snapshot"),
@@ -469,10 +482,15 @@ class ScanTimeBasedIntegrationTest
         getLedgerTime,
         migrationId,
       )
-      snapshotBefore should not(be(snapshotAfter))
+      snapshot1 should not(be(snapshotAfter))
       snapshotAfter
     }
 
+    sv1ScanBackend.getDateOfFirstSnapshotAfter(startTime, 0).value shouldBe snapshot1.value
+    sv1ScanBackend
+      .getDateOfFirstSnapshotAfter(CantonTimestamp.tryFromInstant(snapshot1.value.toInstant), 0)
+      .value shouldBe snapshotAfter.value
+
     val snapshotAfterData = sv1ScanBackend.getAcsSnapshotAt(
       CantonTimestamp.assertFromInstant(snapshotAfter.value.toInstant),
       migrationId,

apps/scan/src/main/openapi/scan.yaml

@@ -445,6 +445,45 @@ paths:
         "500":
           $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/500"
 
+  /v0/state/acs/snapshot-timestamp-after:
+    get:
+      tags: [external, scan]
+      x-jvm-package: scan
+      operationId: "getDateOfFirstSnapshotAfter"
+      description: |
+        Returns the timestamp of the first snapshot after the given date, for the given migration_id or larger.
+      parameters:
+        - name: "after"
+          in: "query"
+          required: true
+          schema:
+            type: string
+            format: date-time
+            description: |
+              The endpoint will return the record time of the first snapshot after this parameter.
+        - name: "migration_id"
+          in: "query"
+          required: true
+          schema:
+            type: integer
+            format: int64
+            description: |
+              The endpoint will return the record time of the first snapshot for this migration id or larger.
+      responses:
+        "200":
+          description: ok
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AcsSnapshotTimestampResponse"
+        "400":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/400"
+        "404":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/404"
+        "500":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/500"
+
+
   /v0/state/acs:
     post:
       tags: [external, scan]
@@ -1180,12 +1219,12 @@ paths:
       tags: [deprecated, scan]
       x-jvm-package: scan
       operationId: "getTotalAmuletBalance"
-      description: | 
-        **Deprecated**, use /registry/metadata/v1/instruments/Amulet token standard metadata API endpoint, see 
+      description: |
+        **Deprecated**, use /registry/metadata/v1/instruments/Amulet token standard metadata API endpoint, see
         https://docs.sync.global/app_dev/token_standard/openapi/token_metadata.html.
-        
-        **This endpoint will be removed in a future release**        
-        
+
+        **This endpoint will be removed in a future release**
+
         Get the total balance of Amulet in the network.
       parameters:
         - in: query
@@ -1211,10 +1250,10 @@ paths:
       x-jvm-package: scan
       operationId: "getWalletBalance"
       description: |
-        **Deprecated**, use /v0/holdings/summary with /v0/state/acs/snapshot-timestamp instead. 
-        
+        **Deprecated**, use /v0/holdings/summary with /v0/state/acs/snapshot-timestamp instead.
+
         **This endpoint will be removed in a future release**
-        
+
         Get the Amulet balance for a specific party at the end of a closed round.
       parameters:
         - in: query
@@ -2629,7 +2668,7 @@ components:
           type: string
           format: date-time
           description: |
-            The timestamp at which the contract set was active. 
+            The timestamp at which the contract set was active.
             This needs to be an exact timestamp, i.e.,
             needs to correspond to a timestamp reported by `/v0/state/acs/snapshot-timestamp` if `record_time_match` is set to `exact` (which is the default).
             If `record_time_match` is set to `at_or_before`, this can be any timestamp, and the most recent snapshot at or before the given `record_time` will be returned.
@@ -2684,7 +2723,7 @@ components:
           type: string
           format: date-time
           description: |
-            The timestamp at which the contract set was active. 
+            The timestamp at which the contract set was active.
             This needs to be an exact timestamp, i.e.,
             needs to correspond to a timestamp reported by `/v0/state/acs/snapshot-timestamp` if `record_time_match` is set to `exact` (which is the default).
             If `record_time_match` is set to `at_or_before`, this can be any timestamp, and the most recent snapshot at or before the given `record_time` will be returned.
@@ -2731,7 +2770,7 @@ components:
           type: string
           format: date-time
           description: |
-            The timestamp at which the contract set was active. 
+            The timestamp at which the contract set was active.
             This needs to be an exact timestamp, i.e.,
             needs to correspond to a timestamp reported by `/v0/state/acs/snapshot-timestamp` if `record_time_match` is set to `exact` (which is the default).
             If `record_time_match` is set to `at_or_before`, this can be any timestamp, and the most recent snapshot at or before the given `record_time` will be returned.

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala

@@ -38,6 +38,7 @@ import org.lfdecentralizedtrust.tokenstandard.{
 import org.lfdecentralizedtrust.splice.http.v0.scan.{
   ForceAcsSnapshotNowResponse,
   GetDateOfMostRecentSnapshotBeforeResponse,
+  GetDateOfFirstSnapshotAfterResponse,
 }
 import org.lfdecentralizedtrust.splice.scan.admin.http.{
   CompactJsonScanHttpEncodings,
@@ -1016,6 +1017,32 @@ object HttpScanAppClient {
     }
   }
 
+  case class GetDateOfFirstSnapshotAfter(
+      after: java.time.OffsetDateTime,
+      migrationId: Long,
+  ) extends InternalBaseCommand[
+        http.GetDateOfFirstSnapshotAfterResponse,
+        Option[java.time.OffsetDateTime],
+      ] {
+    override def submitRequest(
+        client: ScanClient,
+        headers: List[HttpHeader],
+    ): EitherT[Future, Either[Throwable, HttpResponse], GetDateOfFirstSnapshotAfterResponse] =
+      client.getDateOfFirstSnapshotAfter(after, migrationId, headers)
+
+    override protected def handleOk()(implicit
+        decoder: TemplateJsonDecoder
+    ): PartialFunction[GetDateOfFirstSnapshotAfterResponse, Either[
+      String,
+      Option[java.time.OffsetDateTime],
+    ]] = {
+      case http.GetDateOfFirstSnapshotAfterResponse.OK(value) =>
+        Right(Some(value.recordTime))
+      case http.GetDateOfFirstSnapshotAfterResponse.NotFound(_) =>
+        Right(None)
+    }
+  }
+
   case class GetAcsSnapshotAt(
       at: java.time.OffsetDateTime,
       migrationId: Long,

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

@@ -1428,6 +1428,31 @@ class HttpScanHandler(
     }
   }
 
+  override def getDateOfFirstSnapshotAfter(
+      respond: ScanResource.GetDateOfFirstSnapshotAfterResponse.type
+  )(after: OffsetDateTime, migrationId: Long)(
+      extracted: TraceContext
+  ): Future[ScanResource.GetDateOfFirstSnapshotAfterResponse] = {
+    implicit val tc: TraceContext = extracted
+    withSpan(s"$workflowId.getDateOfFirstSnapshotAfter") { _ => _ =>
+      snapshotStore
+        .lookupSnapshotAfter(migrationId, Codec.tryDecode(Codec.OffsetDateTime)(after))
+        .map {
+          case Some(snapshot) =>
+            ScanResource.GetDateOfFirstSnapshotAfterResponseOK(
+              definitions
+                .AcsSnapshotTimestampResponse(
+                  Codec.encode(snapshot.snapshotRecordTime)
+                )
+            )
+          case None =>
+            ScanResource.GetDateOfFirstSnapshotAfterResponseNotFound(
+              definitions.ErrorResponse(s"No snapshots found after $after")
+            )
+        }
+    }
+  }
+
   override def forceAcsSnapshotNow(
       respond: ScanResource.ForceAcsSnapshotNowResponse.type
   )()(extracted: TraceContext): Future[ScanResource.ForceAcsSnapshotNowResponse] = {

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

@@ -68,6 +68,34 @@ class AcsSnapshotStore(
       .value
   }
 
+  def lookupSnapshotAfter(
+      migrationId: Long,
+      after: CantonTimestamp,
+  )(implicit tc: TraceContext): Future[Option[AcsSnapshot]] = {
+
+    val select =
+      sql"select snapshot_record_time, migration_id, history_id, first_row_id, last_row_id, unlocked_amulet_balance, locked_amulet_balance "
+    val orderLimit = sql" order by snapshot_record_time asc limit 1 "
+    val sameMig = select ++ sql""" from acs_snapshot
+            where snapshot_record_time > $after
+              and migration_id = $migrationId
+              and history_id = $historyId """ ++ orderLimit
+    val largerMig = select ++ sql""" from acs_snapshot
+            where migration_id > $migrationId
+              and history_id = $historyId """ ++ orderLimit
+
+    val query =
+      sql"select * from ((" ++ sameMig ++ sql") union all (" ++ largerMig ++ sql")) all_queries order by snapshot_record_time asc limit 1"
+
+    storage
+      .querySingle(
+        query.toActionBuilder.as[AcsSnapshot].headOption,
+        "lookupSnapshotAfter",
+      )
+      .value
+
+  }
+
   def insertNewSnapshot(
       lastSnapshot: Option[AcsSnapshot],
       migrationId: Long,

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala

@@ -58,8 +58,15 @@ class AcsSnapshotStoreTest
         for {
           updateHistory <- mkUpdateHistory()
           store = mkStore(updateHistory)
-          result <- store.lookupSnapshotAtOrBefore(DefaultMigrationId, CantonTimestamp.MaxValue)
-        } yield result should be(None)
+          resultBefore <- store.lookupSnapshotAtOrBefore(
+            DefaultMigrationId,
+            CantonTimestamp.MaxValue,
+          )
+          resultAfter <- store.lookupSnapshotAfter(DefaultMigrationId, CantonTimestamp.MinValue)
+        } yield {
+          resultBefore should be(None)
+          resultAfter should be(None)
+        }
       }
 
       "only return the last snapshot of the passed migration id" in {
@@ -95,7 +102,7 @@ class AcsSnapshotStoreTest
         } yield result should be(None)
       }
 
-      "return the latest snapshot before the given timestamp" in {
+      "return correct snapshots before and after given timestamps" in {
         for {
           updateHistory <- mkUpdateHistory()
           store = mkStore(updateHistory)
@@ -109,8 +116,17 @@ class AcsSnapshotStoreTest
               snapshot <- store.insertNewSnapshot(None, DefaultMigrationId, timestamp)
             } yield snapshot
           }
-          result <- store.lookupSnapshotAtOrBefore(DefaultMigrationId, timestamp4)
-        } yield result.map(_.snapshotRecordTime) should be(Some(timestamp3))
+          resultBefore4 <- store.lookupSnapshotAtOrBefore(DefaultMigrationId, timestamp4)
+          firstResult <- store.lookupSnapshotAfter(DefaultMigrationId, CantonTimestamp.MinValue)
+          secondResult <- store.lookupSnapshotAfter(
+            DefaultMigrationId,
+            firstResult.value.snapshotRecordTime,
+          )
+        } yield {
+          resultBefore4.map(_.snapshotRecordTime) should be(Some(timestamp3))
+          firstResult.map(_.snapshotRecordTime) should be(Some(timestamp1))
+          secondResult.map(_.snapshotRecordTime) should be(Some(timestamp2))
+        }
       }
 
     }