Repository: BlueBrain/nexus (scala)

Issue:
Remove Nexus metadata from payloads during the import
**Motivation**
Some resource still contain Nexus metadata in their original payloads before fields starting with `_` were forbidden.
The import tool can take care of cleaning up those fields

**Acceptance criteria**
1. During an import, all fields starting by `_` at the root level are removed from the original payload before the resource is evaluated

Pull request:
Some resource still contain Nexus metadata in their original payloads before fields starting with _ were forbidden. The import tool can take care of cleaning up those fields

Fixes #5035

Patch:
diff --git a/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/EventProcessor.scala b/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/EventProcessor.scala
index 5bf6fd6ffb..d4cb791543 100644
--- a/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/EventProcessor.scala
+++ b/ship/src/main/scala/ch/epfl/bluebrain/nexus/ship/EventProcessor.scala
@@ -26,7 +26,9 @@ trait EventProcessor[Event <: ScopedEvent] {
   def evaluate(event: RowEvent)(implicit iriPatcher: IriPatcher): IO[ImportStatus] = {
     val value                            = event.value
     def patchEventId(idAsString: String) = Iri(idAsString).map(iriPatcher(_).toString).getOrElse(idAsString)
-    val patchedValue                     = root.id.string.modify { patchEventId }(value)
+    def updateIdToPatchedVersion         = root.id.string.modify { patchEventId }
+    def removeSourceMetadata             = root.source.obj.modify(_.filterKeys(!_.startsWith("_")))
+    val patchedValue                     = updateIdToPatchedVersion.andThen(removeSourceMetadata)(value)
     IO.fromEither(decoder.decodeJson(patchedValue))
       .onError(err => logger.error(err)(s"Error while attempting to decode $resourceType at offset ${event.ordering}"))
       .flatMap(evaluate)


Test patch:
diff --git a/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/EventProcessorSuite.scala b/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/EventProcessorSuite.scala
index 5e932cd173..cc1087b1d2 100644
--- a/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/EventProcessorSuite.scala
+++ b/ship/src/test/scala/ch/epfl/bluebrain/nexus/ship/EventProcessorSuite.scala
@@ -3,22 +3,38 @@ package ch.epfl.bluebrain.nexus.ship
 import cats.effect.{IO, Ref}
 import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri
 import ch.epfl.bluebrain.nexus.delta.rdf.syntax.iriStringContextSyntax
+import ch.epfl.bluebrain.nexus.delta.sdk.resources.ResourceInstanceFixture
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestEvent
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestEvent.PullRequestCreated
+import ch.epfl.bluebrain.nexus.delta.sourcing.event.Event.ScopedEvent
 import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.RowEvent
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Anonymous
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, ProjectRef}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, Identity, ProjectRef}
 import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
 import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
-import io.circe.syntax.EncoderOps
-import io.circe.{Codec, Decoder}
+import io.circe.generic.extras.Configuration
+import io.circe.generic.extras.semiauto.deriveConfiguredCodec
+import io.circe.syntax.{EncoderOps, KeyOps}
+import io.circe.{Codec, Decoder, Json}
 
 import java.time.Instant
 
-class EventProcessorSuite extends NexusSuite {
+final case class ExampleEvent(
+    id: Iri,
+    project: ProjectRef,
+    rev: Int,
+    instant: Instant,
+    subject: Identity.Subject,
+    source: Json
+) extends ScopedEvent
 
-  implicit private val eventDecoder: Codec[PullRequestEvent] = PullRequest.PullRequestEvent.serializer.codec
+class EventProcessorSuite extends NexusSuite with ResourceInstanceFixture {
+
+  import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Database._
+  implicit private val configuration: Configuration                    = Configuration.default.withDiscriminator("@type")
+  implicit private val eventDecoder: Codec[PullRequestEvent]           = PullRequest.PullRequestEvent.serializer.codec
+  implicit private val exampleEventCodec: Codec.AsObject[ExampleEvent] = deriveConfiguredCodec[ExampleEvent]
 
   private def createEventProcessor(ref: Ref[IO, Option[Iri]]) = new EventProcessor[PullRequestEvent] {
     override def resourceType: EntityType = PullRequest.entityType
@@ -28,6 +44,12 @@ class EventProcessorSuite extends NexusSuite {
     override def evaluate(event: PullRequestEvent): IO[ImportStatus] = ref.set(Some(event.id)).as(ImportStatus.Success)
   }
 
+  private def exampleEventProcessor(ref: Ref[IO, Option[ExampleEvent]]) = new EventProcessor[ExampleEvent] {
+    override def resourceType: EntityType                        = EntityType("example")
+    override def decoder: Decoder[ExampleEvent]                  = exampleEventCodec
+    override def evaluate(event: ExampleEvent): IO[ImportStatus] = ref.set(Some(event)).as(ImportStatus.Success)
+  }
+
   test("Replace the original id in the value and pass it down") {
     implicit val iriPatcher     = IriPatcher(iri"https://bbp.epfl.ch/", iri"https://openbrainplatform.com/", Map.empty)
     val project                 = ProjectRef.unsafe("org", "proj")
@@ -52,4 +74,34 @@ class EventProcessorSuite extends NexusSuite {
     } yield ()
   }
 
+  test("Replace the original id in the value and pass it down 2") {
+    implicit val iriPatcher = IriPatcher.noop
+    val project             = ProjectRef.unsafe("org", "proj")
+    val id                  = iri"https://bbp.epfl.ch/resource1"
+    val event               = ExampleEvent(
+      id,
+      project,
+      1,
+      Instant.EPOCH,
+      Anonymous,
+      Json.obj("_instant" := Instant.now(), "normalField" := "value")
+    )
+    val rowEvent            = RowEvent(
+      Offset.At(42L),
+      PullRequest.entityType,
+      project.organization,
+      project.project,
+      id,
+      1,
+      event.asJson.deepMerge(Json.obj()),
+      Instant.EPOCH
+    )
+    for {
+      ref           <- Ref.of[IO, Option[ExampleEvent]](None)
+      eventProcessor = exampleEventProcessor(ref)
+      _             <- eventProcessor.evaluate(rowEvent)
+      _             <- ref.get.assert(_.exists(_.source == Json.obj("normalField" := "value")))
+    } yield ()
+  }
+
 }