Add endpoint to return latest event record time

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

@@ -604,6 +604,11 @@ abstract class ScanAppReference(
     }
   }
 
+  def getLatestEventRecordTime(): Option[definitions.EventLatestRecordTimeResponse] =
+    consoleEnvironment.run {
+      httpCommand(HttpScanAppClient.GetLatestEventRecordTime())
+    }
+
   def getEventById(
       updateId: String,
       damlValueEncoding: Option[definitions.DamlValueEncoding],

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

@@ -1577,6 +1577,28 @@ paths:
           $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/400"
         "500":
           $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/500"
+
+  /v0/events/latest-record-time:
+    get:
+      tags: [external, scan]
+      x-jvm-package: scan
+      operationId: "getLatestEventRecordTime"
+      description: |
+        Returns the latest record time for which /v0/events will be able to return events.
+      responses:
+        "200":
+          description: ok
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/EventLatestRecordTimeResponse"
+        "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/events/{update_id}:
     get:
       tags: [external, scan]
@@ -3845,6 +3867,16 @@ components:
         app_activity_records:
           $ref: "#/components/schemas/EventHistoryAppActivityRecords"
           nullable: true
+    EventLatestRecordTimeResponse:
+      type: object
+      required:
+        - record_time
+      properties:
+        record_time:
+          description: |
+            The record_time of the latest event.
+          type: string
+          format: date-time
     EventHistoryVerdict:
       type: object
       required:

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

@@ -1478,6 +1478,25 @@ object HttpScanAppClient {
     }
   }
 
+  case class GetLatestEventRecordTime()
+      extends InternalBaseCommand[
+        http.GetLatestEventRecordTimeResponse,
+        Option[definitions.EventLatestRecordTimeResponse],
+      ] {
+    override def submitRequest(
+        client: http.ScanClient,
+        headers: List[HttpHeader],
+    ): EitherT[Future, Either[Throwable, HttpResponse], http.GetLatestEventRecordTimeResponse] =
+      client.getLatestEventRecordTime()
+
+    override def handleOk()(implicit decoder: TemplateJsonDecoder) = {
+      case http.GetLatestEventRecordTimeResponse.OK(response) =>
+        Right(Some(response))
+      case http.GetLatestEventRecordTimeResponse.NotFound(_) =>
+        Right(None)
+    }
+  }
+
   case class GetEventById(
       updateId: String,
       damlValueEncoding: Option[definitions.DamlValueEncoding],

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

@@ -957,6 +957,24 @@ class HttpScanHandler(
     }
   }
 
+  override def getLatestEventRecordTime(
+      respond: ScanResource.GetLatestEventRecordTimeResponse.type
+  )()(extracted: TraceContext): Future[ScanResource.GetLatestEventRecordTimeResponse] = {
+    implicit val tc = extracted
+    withSpan(s"$workflowId.getLatestEventRecordTime") { _ => _ =>
+      eventStore.getLatestEventRecordTime(updateHistory.domainMigrationId).map {
+        case Some(timestamp) =>
+          ScanResource.GetLatestEventRecordTimeResponse.OK(
+            definitions.EventLatestRecordTimeResponse(Codec.encode(timestamp))
+          )
+        case None =>
+          ScanResource.GetLatestEventRecordTimeResponse.NotFound(
+            definitions.ErrorResponse("No events found")
+          )
+      }
+    }
+  }
+
   private def toUpdateV2WithHash(update: UpdateHistoryItem): UpdateHistoryItemV2WithHash =
     update match {
       case UpdateHistoryItem.members.UpdateHistoryReassignment(r) =>

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

@@ -122,6 +122,15 @@ class ScanEventStore(
     }
   }
 
+  def getLatestEventRecordTime(
+      currentMigrationId: Long
+  )(implicit tc: TraceContext): Future[Option[CantonTimestamp]] =
+    resolveCurrentMigrationCap(
+      verdictStore.lastIngestedRecordTime,
+      updateHistory.lastIngestedRecordTime,
+      currentMigrationId,
+    ).map(ts => if (ts == CantonTimestamp.MinValue) None else Some(ts))
+
   def getAppActivityRecords(verdictRowIds: Seq[Long])(implicit
       tc: TraceContext
   ): Future[Map[Long, AppActivityRecordT]] =

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStoreTest.scala

@@ -864,6 +864,57 @@ class ScanEventStoreTest extends StoreTestBase with HasExecutionContext with Spl
         allow(mig2, recordTs3) shouldBe false // > after and > cap
       }
     }
+
+    "getLatestEventRecordTime returns the record time of the last event getEvents returns" in {
+      for {
+        ctx <- newEventStore()
+        // update + verdict at ts1
+        ts1 = CantonTimestamp.now()
+        tx1 <- insertUpdate(ctx.updateHistory, ts1, "update1")
+        _ <- insertVerdict(ctx.verdictStore, tx1.getUpdateId, ts1)
+
+        // update + verdict at ts2
+        ts2 = ts1.plusSeconds(1)
+        tx2 <- insertUpdate(ctx.updateHistory, ts2, "update2")
+        _ <- insertVerdict(ctx.verdictStore, tx2.getUpdateId, ts2)
+
+        // loose update at ts3 must be filtered out of getEvents
+        ts3 = ts2.plusSeconds(1)
+        _ <- insertUpdate(ctx.updateHistory, ts3, "update-loose")
+
+        events <- fetchEvents(ctx.eventStore, None, domainMigrationId, pageLimit)
+        latest <- ctx.eventStore.getLatestEventRecordTime(domainMigrationId)(traceContext)
+      } yield {
+        // getEvents caps at ts2, the ts3 loose update is excluded
+        events.nonEmpty shouldBe true
+        val lastReturnedRt = events.last._1
+          .map(_._1.recordTime)
+          .orElse(events.last._2.map(_.update.update.recordTime))
+          .value
+        lastReturnedRt shouldBe ts2
+
+        latest shouldBe Some(lastReturnedRt)
+      }
+    }
+
+    "getLatestEventRecordTime returns None when there are no events" in {
+      for {
+        ctx <- newEventStore()
+        latest <- ctx.eventStore.getLatestEventRecordTime(domainMigrationId)(traceContext)
+      } yield {
+        latest shouldBe None
+      }
+    }
+
+    "getLatestEventRecordTime returns None when there is only a loose update" in {
+      for {
+        ctx <- newEventStore()
+        _ <- insertUpdate(ctx.updateHistory, CantonTimestamp.now(), "update1")
+        latest <- ctx.eventStore.getLatestEventRecordTime(domainMigrationId)(traceContext)
+      } yield {
+        latest shouldBe None
+      }
+    }
   }
 
   private def newUpdateHistory(