Make project in Elem mandatory (#5170)

Co-authored-by: Simon Dumas <simon.dumas@epfl.ch>

delta/app/src/main/resources/app.conf

@@ -299,15 +299,8 @@ app {
 
   # SSE configuration
   sse {
-    # the schemas event-log configuration
+    # the SSE query configuration
     query = ${app.defaults.query}
-    # the sse cache to get projects uuids
-    cache {
-      # The max number of tokens in the projects uuids cache
-      max-size = 500
-      # The duration after which the cache expires
-      expire-after = 15 minutes
-    }
   }
 
   # projection configuration

delta/app/src/test/scala/ch/epfl/bluebrain/nexus/delta/routes/SchemaJobRoutesSpec.scala

@@ -67,9 +67,8 @@ class SchemaJobRoutesSpec extends BaseRouteSpec {
     val projectionMetadata = SchemaValidationCoordinator.projectionMetadata(project.ref)
 
     val reason = FailureReason("ValidationFail", json"""{ "details":  "..." }""")
-    val fail1  = FailedElem(EntityType("ACL"), resourceId, Some(project.ref), Instant.EPOCH, Offset.At(42L), reason, rev)
-    val fail2  =
-      FailedElem(EntityType("Schema"), resourceId, Some(project.ref), Instant.EPOCH, Offset.At(42L), reason, rev)
+    val fail1  = FailedElem(EntityType("ACL"), resourceId, project.ref, Instant.EPOCH, Offset.At(42L), reason, rev)
+    val fail2  = FailedElem(EntityType("Schema"), resourceId, project.ref, Instant.EPOCH, Offset.At(42L), reason, rev)
 
     (
       projections.save(projectionMetadata, progress) >>

delta/plugins/blazegraph/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/BlazegraphViews.scala

@@ -299,9 +299,7 @@ final class BlazegraphViews(
     }
 
   private def toIndexViewDef(elem: Elem.SuccessElem[BlazegraphViewState]) =
-    elem.withProject(elem.value.project).traverse { v =>
-      IndexingViewDef(v, prefix)
-    }
+    elem.traverse { v => IndexingViewDef(v, prefix) }
 
   private def eval(cmd: BlazegraphViewCommand): IO[ViewResource] =
     log.evaluate(cmd.project, cmd.id, cmd).map(_._2.toResource)

delta/plugins/blazegraph/src/test/resources/routes/list-indexing-errors.json

@@ -23,11 +23,12 @@
       }
     },
     {
-      "id": "https://bluebrain.github.io/nexus/vocabulary/myid",
+      "id": "https://bluebrain.github.io/nexus/vocabulary/myid2",
       "offset": {
         "@type": "At",
-        "value": 42
+        "value": 43
       },
+      "project": "org/proj",
       "_rev": 1,
       "reason": {
         "type": "UnexpectedError",

delta/plugins/blazegraph/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/BlazegraphIndexingActionSuite.scala

@@ -80,7 +80,7 @@ class BlazegraphIndexingActionSuite extends NexusSuite with Fixtures {
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view1.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(1L),
         value = view1,
@@ -89,7 +89,7 @@ class BlazegraphIndexingActionSuite extends NexusSuite with Fixtures {
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view2.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(2L),
         value = view2,
@@ -98,7 +98,7 @@ class BlazegraphIndexingActionSuite extends NexusSuite with Fixtures {
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view3.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(3L),
         value = view3,
@@ -107,7 +107,7 @@ class BlazegraphIndexingActionSuite extends NexusSuite with Fixtures {
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view4.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(4L),
         value = view4,
@@ -141,7 +141,7 @@ class BlazegraphIndexingActionSuite extends NexusSuite with Fixtures {
   private val elem = SuccessElem(
     tpe = PullRequest.entityType,
     id = pr.id,
-    project = Some(project),
+    project = project,
     instant = pr.updatedAt,
     offset = Offset.at(1L),
     value = PullRequestState.toGraphResource(pr, base),
@@ -171,7 +171,7 @@ class BlazegraphIndexingActionSuite extends NexusSuite with Fixtures {
     val failed = FailedElem(
       tpe = PullRequest.entityType,
       id = pr.id,
-      project = Some(project),
+      project = project,
       instant = pr.updatedAt,
       offset = Offset.at(1L),
       new IllegalStateException("Boom"),

delta/plugins/blazegraph/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/indexing/BlazegraphCoordinatorSuite.scala

@@ -92,7 +92,7 @@ class BlazegraphCoordinatorSuite extends NexusSuite with SupervisorSetup.Fixture
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view1.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(1L),
         value = view1,
@@ -101,7 +101,7 @@ class BlazegraphCoordinatorSuite extends NexusSuite with SupervisorSetup.Fixture
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view2.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(2L),
         value = view2,
@@ -110,7 +110,7 @@ class BlazegraphCoordinatorSuite extends NexusSuite with SupervisorSetup.Fixture
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = view3.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(3L),
         value = view3,
@@ -120,7 +120,7 @@ class BlazegraphCoordinatorSuite extends NexusSuite with SupervisorSetup.Fixture
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = deprecatedView1.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(4L),
         value = deprecatedView1,
@@ -129,7 +129,7 @@ class BlazegraphCoordinatorSuite extends NexusSuite with SupervisorSetup.Fixture
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = updatedView2.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(5L),
         value = updatedView2,
@@ -139,7 +139,7 @@ class BlazegraphCoordinatorSuite extends NexusSuite with SupervisorSetup.Fixture
       SuccessElem(
         tpe = BlazegraphViews.entityType,
         id = updatedView2.ref.viewId,
-        project = Some(project),
+        project = project,
         instant = Instant.EPOCH,
         offset = Offset.at(6L),
         value = updatedView2,

delta/plugins/blazegraph/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/indexing/BlazegraphSinkSuite.scala

@@ -9,7 +9,7 @@ import ch.epfl.bluebrain.nexus.delta.rdf.graph.{Graph, NTriples}
 import ch.epfl.bluebrain.nexus.delta.rdf.query.SparqlQuery.SparqlConstructQuery
 import ch.epfl.bluebrain.nexus.delta.sdk.implicits._
 import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, Label}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, Label, ProjectRef}
 import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.Elem.{DroppedElem, SuccessElem}
 import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
@@ -32,6 +32,8 @@ class BlazegraphSinkSuite extends NexusSuite with BlazegraphClientSetup.Fixture
 
   private lazy val sink = createSink(namespace)
 
+  private val project = ProjectRef.unsafe("org", "project")
+
   private val resource1Id              = iri"https://bbp.epfl.ch/resource1"
   private val resource1Ntriples        = NTriples(contentOf("sparql/resource1.ntriples"), resource1Id)
   private val resource1NtriplesUpdated = NTriples(contentOf("sparql/resource1_updated.ntriples"), resource1Id)
@@ -54,14 +56,14 @@ class BlazegraphSinkSuite extends NexusSuite with BlazegraphClientSetup.Fixture
 
   private def asElems(chunk: Chunk[(Iri, NTriples)]) =
     chunk.zipWithIndex.map { case ((id, ntriples), index) =>
-      SuccessElem(entityType, id, None, Instant.EPOCH, Offset.at(index.toLong + 1), ntriples, 1)
+      SuccessElem(entityType, id, project, Instant.EPOCH, Offset.at(index.toLong + 1), ntriples, 1)
     }
 
   private def createGraph(chunk: Chunk[(Iri, NTriples)]) = chunk.foldLeft(Graph.empty) { case (acc, (_, ntriples)) =>
     acc ++ Graph(ntriples).getOrElse(Graph.empty)
   }
 
-  private def dropped(id: Iri, offset: Offset) = DroppedElem(entityType, id, None, Instant.EPOCH, offset, 1)
+  private def dropped(id: Iri, offset: Offset) = DroppedElem(entityType, id, project, Instant.EPOCH, offset, 1)
 
   private def query(namespace: String) =
     client
@@ -96,7 +98,7 @@ class BlazegraphSinkSuite extends NexusSuite with BlazegraphClientSetup.Fixture
 
   test("Report errors when the id is not a valid absolute iri") {
     val chunk    = Chunk(
-      SuccessElem(entityType, nxv + "é-wrong", None, Instant.EPOCH, Offset.at(5L), resource1Ntriples, 1)
+      SuccessElem(entityType, nxv + "é-wrong", project, Instant.EPOCH, Offset.at(5L), resource1Ntriples, 1)
     )
     val expected = createGraph(Chunk(resource1Id -> resource1Ntriples, resource3Id -> resource3Ntriples))
 

delta/plugins/blazegraph/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/routes/BlazegraphViewsIndexingRoutesSpec.scala

@@ -30,6 +30,7 @@ class BlazegraphViewsIndexingRoutesSpec extends BlazegraphViewRoutesFixtures {
   private lazy val projectionErrors = ProjectionErrors(xas, queryConfig, clock)
 
   private val myId         = nxv + "myid"
+  private val myId2        = nxv + "myid2"
   private val indexingView = ActiveViewDef(
     ViewRef(projectRef, myId),
     "projection",
@@ -65,8 +66,8 @@ class BlazegraphViewsIndexingRoutesSpec extends BlazegraphViewRoutesFixtures {
     super.beforeAll()
     val error = new Exception("boom")
     val rev   = 1
-    val fail1 = FailedElem(EntityType("ACL"), myId, Some(projectRef), Instant.EPOCH, Offset.At(42L), error, rev)
-    val fail2 = FailedElem(EntityType("Schema"), myId, None, Instant.EPOCH, Offset.At(42L), error, rev)
+    val fail1 = FailedElem(EntityType("ACL"), myId, projectRef, Instant.EPOCH, Offset.At(42L), error, rev)
+    val fail2 = FailedElem(EntityType("Schema"), myId2, projectRef, Instant.EPOCH, Offset.At(43L), error, rev)
     val save  = for {
       _ <- projections.save(indexingView.projectionMetadata, progress)
       _ <- projectionErrors.saveFailedElems(indexingView.projectionMetadata, List(fail1, fail2))

delta/plugins/composite-views/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/CompositeViews.scala

@@ -303,13 +303,11 @@ final class CompositeViews private (
   /**
     * Return the indexing views in a non-ending stream
     */
-  def views(start: Offset): ElemStream[CompositeViewDef] =
+  def views(start: Offset): SuccessElemStream[CompositeViewDef] =
     log.states(Scope.Root, start).map(toCompositeViewDef)
 
   private def toCompositeViewDef(elem: Elem.SuccessElem[CompositeViewState]) =
-    elem.withProject(elem.value.project).mapValue { v =>
-      CompositeViewDef(v)
-    }
+    elem.mapValue { v => CompositeViewDef(v) }
 
   private def eval(cmd: CompositeViewCommand): IO[ViewResource] =
     log.evaluate(cmd.project, cmd.id, cmd).map(_._2.toResource)

delta/plugins/composite-views/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/store/CompositeRestartStore.scala

@@ -81,7 +81,7 @@ final class CompositeRestartStore(xas: Transactors) {
          |LIMIT 1""".stripMargin
       .query[(Offset, ProjectRef, Iri, Json, Instant)]
       .map { case (offset, project, id, json, instant) =>
-        Elem.fromEither(entityType, id, Some(project), instant, offset, json.as[CompositeRestart], 1)
+        Elem.fromEither(entityType, id, project, instant, offset, json.as[CompositeRestart], 1)
       }
       .option
       .transact(xas.read)

delta/plugins/composite-views/src/test/resources/routes/list-indexing-errors.json

@@ -23,11 +23,12 @@
       }
     },
     {
-      "id": "https://bluebrain.github.io/nexus/vocabulary/myid",
+      "id": "https://bluebrain.github.io/nexus/vocabulary/myid2",
       "offset": {
         "@type": "At",
-        "value": 42
+        "value": 43
       },
+      "project": "myorg/myproj",
       "_rev": 1,
       "reason": {
         "type": "UnexpectedError",

delta/plugins/composite-views/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/client/DeltaClientSpec.scala

@@ -73,15 +73,7 @@ class DeltaClientSpec
   private val invalidTag = Some(UserTag.unsafe("unknowntag"))
 
   private def elem(i: Int): Elem[Unit] =
-    SuccessElem(
-      EntityType("test"),
-      iri"https://bbp.epfl.ch/$i",
-      Some(project),
-      Instant.EPOCH,
-      Offset.at(i.toLong),
-      (),
-      1
-    )
+    SuccessElem(EntityType("test"), iri"https://bbp.epfl.ch/$i", project, Instant.EPOCH, Offset.at(i.toLong), (), 1)
 
   override def beforeAll(): Unit = {
     super.beforeAll()

delta/plugins/composite-views/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeIndexingSuite.scala

@@ -166,7 +166,7 @@ abstract class CompositeIndexingSuite(sinkConfig: SinkConfig, query: SparqlConst
     } yield SuccessElem(
       entityType,
       value.id,
-      Some(project),
+      project,
       Instant.EPOCH,
       Offset.at(offset),
       resource,

delta/plugins/composite-views/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/routes/CompositeViewsIndexingRoutesSpec.scala

@@ -38,6 +38,7 @@ class CompositeViewsIndexingRoutesSpec extends CompositeViewsRoutesFixtures {
   private val nowPlus5 = now.plusSeconds(5)
 
   private val myId         = nxv + "myid"
+  private val myId2        = nxv + "myid2"
   private val view         = CompositeViewsGen.resourceFor(projectRef, myId, uuid, viewValue, source = Json.obj())
   private val indexingView = ActiveViewDef(
     ViewRef(view.value.project, view.id),
@@ -93,8 +94,8 @@ class CompositeViewsIndexingRoutesSpec extends CompositeViewsRoutesFixtures {
     super.beforeAll()
     val error           = new Exception("boom")
     val rev             = 1
-    val fail1           = FailedElem(EntityType("ACL"), myId, Some(projectRef), Instant.EPOCH, Offset.At(42L), error, rev)
-    val fail2           = FailedElem(EntityType("Schema"), myId, None, Instant.EPOCH, Offset.At(42L), error, rev)
+    val fail1           = FailedElem(EntityType("ACL"), myId, projectRef, Instant.EPOCH, Offset.At(42L), error, rev)
+    val fail2           = FailedElem(EntityType("Schema"), myId2, projectRef, Instant.EPOCH, Offset.At(43L), error, rev)
     val saveFailedElems = projectionErrors.saveFailedElems(indexingView.metadata, List(fail1, fail2))
 
     saveFailedElems.accepted

delta/plugins/composite-views/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/store/CompositeRestartStoreSuite.scala

@@ -37,7 +37,7 @@ class CompositeRestartStoreSuite extends NexusSuite with Doobie.Fixture with Doo
   private val cr3        = PartialRebuild(viewRef, projection, Instant.EPOCH.plusSeconds(5L), Anonymous)
 
   private def toElem(offset: Offset, restart: CompositeRestart) =
-    SuccessElem(entityType, restart.id, Some(restart.project), restart.instant, offset, restart, 1)
+    SuccessElem(entityType, restart.id, restart.project, restart.instant, offset, restart, 1)
 
   test("Save composite restarts") {
     for {

delta/plugins/composite-views/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/stream/RemoteGraphStreamSuite.scala

@@ -36,7 +36,7 @@ class RemoteGraphStreamSuite extends NexusSuite {
   private val id = iri"https://example.com/testresource"
 
   private val project = ProjectRef.unsafe("org", "proj")
-  private val elem    = SuccessElem(EntityType("test"), id, Some(project), Instant.EPOCH, Offset.Start, (), 1)
+  private val elem    = SuccessElem(EntityType("test"), id, project, Instant.EPOCH, Offset.Start, (), 1)
   private val nQuads  = NQuads(contentOf("remote/resource.nq"), id)
 
   test("Metadata should be filtered correctly") {

delta/plugins/elasticsearch/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViews.scala

@@ -364,7 +364,7 @@ final class ElasticSearchViews private (
       }
 
   private def toIndexViewDef(elem: Elem.SuccessElem[ElasticSearchViewState]) =
-    elem.withProject(elem.value.project).traverse { v =>
+    elem.traverse { v =>
       IndexingViewDef(v, defaultElasticsearchMapping, defaultElasticsearchSettings, prefix)
     }
 

delta/plugins/elasticsearch/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/indexing/ElasticSearchSink.scala

@@ -77,11 +77,8 @@ object ElasticSearchSink {
     * @return
     *   a function that maps an elem to a documentId based on the project (if available), the elem id, and the revision
     */
-  private val eventDocumentId: Elem[_] => String = elem =>
-    elem.project match {
-      case Some(project) => s"$project/${elem.id}:${elem.rev}"
-      case None          => s"${elem.id}/${elem.rev}"
-    }
+  private val eventDocumentId: Elem[_] => String =
+    elem => s"${elem.project}/${elem.id}:${elem.rev}"
 
   /**
     * Mark and update the elements according to the elasticsearch response

delta/plugins/elasticsearch/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/metrics/EventMetricsProjection.scala

@@ -100,9 +100,7 @@ object EventMetricsProjection {
       init: IO[Unit]
   ): IO[EventMetricsProjection] = {
 
-    val source = Source { (offset: Offset) =>
-      metrics(offset).map(e => e.withProject(e.value.project))
-    }
+    val source = Source { (offset: Offset) => metrics(offset) }
 
     val compiledProjection =
       CompiledProjection.compile(

delta/plugins/elasticsearch/src/test/resources/routes/list-indexing-errors.json

@@ -23,11 +23,12 @@
       }
     },
     {
-      "id": "https://bluebrain.github.io/nexus/vocabulary/myid",
+      "id": "https://bluebrain.github.io/nexus/vocabulary/myid2",
       "offset": {
         "@type": "At",
-        "value": 42
+        "value": 43
       },
+      "project": "myorg/myproject",
       "_rev": 1,
       "reason": {
         "type": "UnexpectedError",