[ci] add support for key-value store in Scan

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

Author: isegall-da

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

@@ -0,0 +1,53 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.scan.store
+
+import com.digitalasset.canton.data.CantonTimestamp
+import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
+import io.circe.Codec
+import io.circe.generic.semiauto.deriveCodec
+import org.lfdecentralizedtrust.splice.store.KeyValueStore
+import cats.data.OptionT
+import cats.implicits.toBifunctorOps
+import com.digitalasset.canton.tracing.TraceContext
+import org.lfdecentralizedtrust.splice.scan.store.ScanKeyValueProvider.*
+
+import scala.concurrent.{ExecutionContext, Future}
+
+class ScanKeyValueProvider(val store: KeyValueStore, val loggerFactory: NamedLoggerFactory)
+    extends NamedLogging {
+
+  private val latestAcsSnapshotInBulkStorageKey = "latest_acs_snapshot_in_bulk_storage"
+
+  final def setLatestAcsSnapshotsInBulkStorage(
+      timestamp: CantonTimestamp,
+      migrationId: Long,
+  )(implicit tc: TraceContext): Future[Unit] = store.setValue(
+    latestAcsSnapshotInBulkStorageKey,
+    AcsSnapshotTimestampMigration(timestamp, migrationId),
+  )
+
+  final def getLatestAcsSnapshotInBulkStorage()(implicit
+      tc: TraceContext,
+      ec: ExecutionContext,
+  ): OptionT[Future, (CantonTimestamp, Long)] = {
+    val result: OptionT[Future, AcsSnapshotTimestampMigration] = store.readValueAndLogOnDecodingFailure(latestAcsSnapshotInBulkStorageKey)
+    result.map(result => (result.timestamp, result.migrationId))
+  }
+}
+
+object ScanKeyValueProvider {
+  final case class AcsSnapshotTimestampMigration(
+      timestamp: CantonTimestamp,
+      migrationId: Long,
+  )
+  implicit val timestampCodec: Codec[CantonTimestamp] =
+    Codec
+      .from[Long](implicitly, implicitly)
+      .iemap(timestamp => CantonTimestamp.fromProtoPrimitive(timestamp).leftMap(_.message))(
+        _.toProtoPrimitive
+      )
+  implicit val acsSnapshotTimestampMigrationCodec: Codec[AcsSnapshotTimestampMigration] =
+    deriveCodec[AcsSnapshotTimestampMigration]
+}

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

@@ -0,0 +1,39 @@
+package org.lfdecentralizedtrust.splice.scan.store
+
+import com.digitalasset.canton.lifecycle.CloseContext
+import com.digitalasset.canton.logging.{ErrorLoggingContext, NamedLoggerFactory}
+import com.digitalasset.canton.resource.DbStorage
+import com.digitalasset.canton.topology.{ParticipantId, PartyId}
+import com.digitalasset.canton.tracing.TraceContext
+import org.lfdecentralizedtrust.splice.store.KeyValueStore
+import org.lfdecentralizedtrust.splice.store.db.StoreDescriptor
+
+import scala.concurrent.{ExecutionContext, Future}
+
+object ScanKeyValueStore {
+  def apply(
+      dsoParty: PartyId,
+      participantId: ParticipantId,
+      storage: DbStorage,
+      loggerFactory: NamedLoggerFactory,
+  )(implicit
+      ec: ExecutionContext,
+      lc: ErrorLoggingContext,
+      cc: CloseContext,
+      tc: TraceContext,
+  ): Future[KeyValueStore] = {
+    KeyValueStore(
+      StoreDescriptor(
+        version = 1,
+        name = "ScanKeyValueStore",
+        party = dsoParty,
+        participant = participantId,
+        key = Map(
+          "dsoParty" -> dsoParty.toProtoPrimitive
+        ),
+      ),
+      storage,
+      loggerFactory,
+    )
+  }
+}

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

@@ -0,0 +1,48 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.scan.store
+
+import com.digitalasset.canton.HasExecutionContext
+import com.digitalasset.canton.data.CantonTimestamp
+import com.digitalasset.canton.lifecycle.FutureUnlessShutdown
+import com.digitalasset.canton.resource.DbStorage
+import com.digitalasset.canton.tracing.TraceContext
+import org.lfdecentralizedtrust.splice.store.StoreTest
+import org.lfdecentralizedtrust.splice.store.db.SplicePostgresTest
+import org.scalatest.matchers.should.Matchers
+
+import scala.concurrent.Future
+
+class ScanKeyValueProviderTest
+    extends StoreTest
+    with Matchers
+    with HasExecutionContext
+    with SplicePostgresTest {
+  "ScanKeyValueProvider" should {
+    "set and get acs snapshots timestamps" in {
+      val ts = CantonTimestamp.now()
+      val migrationId = 7L
+      for {
+        provider <- mkProvider
+        _ <- provider.setLatestAcsSnapshotsInBulkStorage(ts, migrationId)
+        readBack <- provider.getLatestAcsSnapshotInBulkStorage().value
+      } yield {
+        readBack.value shouldBe (ts, migrationId)
+      }
+    }
+  }
+
+  def mkProvider: Future[ScanKeyValueProvider] = {
+    ScanKeyValueStore(
+      dsoParty = dsoParty,
+      participantId = mkParticipantId("participant"),
+      storage,
+      loggerFactory,
+    ).map(new ScanKeyValueProvider(_, loggerFactory))
+  }
+
+  override protected def cleanDb(
+      storage: DbStorage
+  )(implicit traceContext: TraceContext): FutureUnlessShutdown[?] = resetAllAppTables(storage)
+}

test-full-class-names-non-integration.log

@@ -12,6 +12,7 @@ org.lfdecentralizedtrust.splice.scan.admin.api.client.BftScanConnectionTest
 org.lfdecentralizedtrust.splice.scan.admin.http.ScanHttpEncodingsTest
 org.lfdecentralizedtrust.splice.scan.automation.AcsSnapshotTriggerTest
 org.lfdecentralizedtrust.splice.scan.store.ScanEventStoreTest
+org.lfdecentralizedtrust.splice.scan.store.ScanKeyValueProviderTest
 org.lfdecentralizedtrust.splice.store.DomainTimeStoreTest
 org.lfdecentralizedtrust.splice.store.InMemorySynchronizerStoreTest
 org.lfdecentralizedtrust.splice.store.ScanHistoryBackfillingTest