diff --git a/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala b/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala index fd9c5c4e3b..0f36ebe923 100644 --- a/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala +++ b/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], diff --git a/apps/scan/src/main/openapi/scan.yaml b/apps/scan/src/main/openapi/scan.yaml index f5513b5d87..d7298cbc19 100644 --- a/apps/scan/src/main/openapi/scan.yaml +++ b/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: diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala index 11f3149683..b4716be7b7 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala +++ b/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], diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala index ee2bc1c9b9..923308dd5c 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala +++ b/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) => diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStore.scala index d7d378c32b..82bc2dadd0 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStore.scala +++ b/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]] = diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStoreTest.scala index f066410b23..41ef3da099 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/ScanEventStoreTest.scala +++ b/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(