Merge branch 'master' into resolved-resource-source

BlueBrain · Nov 7, 2023 · 8c4c0ca · 8c4c0ca
2 parents f6c98a7 + 1f42c1b
commit 8c4c0ca
Show file tree

Hide file tree

Showing 35 changed files with 196 additions and 217 deletions.
diff --git a/.../main/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/ElasticSearchQuery.scala b/.../main/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/ElasticSearchQuery.scala
@@ -4,15 +4,14 @@ import akka.http.scaladsl.model.Uri
 import akka.http.scaladsl.model.Uri.Query
 import cats.effect.IO
 import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.CompositeViewDef.ActiveViewDef
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.projectionIndex
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewProjection.ElasticSearchProjection
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewRejection.WrappedElasticSearchClientError
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck
 import ch.epfl.bluebrain.nexus.delta.sdk.error.ServiceError.AuthorizationFailed
-import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClient.HttpResult
+import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClientError
 import ch.epfl.bluebrain.nexus.delta.sdk.identities.model.Caller
 import ch.epfl.bluebrain.nexus.delta.sdk.model.IdSegment
 import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SortList
@@ -69,7 +68,7 @@ trait ElasticSearchQuery {
 object ElasticSearchQuery {
 
   private[compositeviews] type ElasticSearchClientQuery =
-    (JsonObject, Set[String], Query) => HttpResult[Json]
+    (JsonObject, Set[String], Query) => IO[Json]
 
   final def apply(
       aclCheck: AclCheck,
@@ -101,7 +100,9 @@ object ElasticSearchQuery {
           _          <-
             aclCheck.authorizeForOr(project, projection.permission)(AuthorizationFailed(project, projection.permission))
           index       = projectionIndex(projection, view.uuid, prefix).value
-          search     <- elasticSearchQuery(query, Set(index), qp).mapError(WrappedElasticSearchClientError)
+          search     <- elasticSearchQuery(query, Set(index), qp).adaptError { case e: HttpClientError =>
+                          WrappedElasticSearchClientError(e)
+                        }
         } yield search
 
       override def queryProjections(
@@ -113,7 +114,9 @@ object ElasticSearchQuery {
         for {
           view    <- fetchView(id, project)
           indices <- allowedProjections(view, project)
-          search  <- elasticSearchQuery(query, indices, qp).mapError(WrappedElasticSearchClientError)
+          search  <- elasticSearchQuery(query, indices, qp).adaptError { case e: HttpClientError =>
+                       WrappedElasticSearchClientError(e)
+                     }
         } yield search
 
       private def fetchProjection(view: ActiveViewDef, projectionId: IdSegment) =

diff --git a/...in/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/ValidateCompositeView.scala b/...in/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/ValidateCompositeView.scala
@@ -2,7 +2,6 @@ package ch.epfl.bluebrain.nexus.delta.plugins.compositeviews
 
 import cats.effect.IO
 import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.client.DeltaClient
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.projectionIndex
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewProjection.{ElasticSearchProjection, SparqlProjection}
@@ -20,6 +19,8 @@ import ch.epfl.bluebrain.nexus.delta.sdk.projects.Projects
 
 import java.util.UUID
 
+import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration.toCatsIOOps
+
 /**
   * Validate an [[CompositeViewValue]] during command evaluation
   */
@@ -52,17 +53,16 @@ object ValidateCompositeView {
         IO.raiseWhen(!perms.contains(permission))(PermissionIsNotDefined(permission))
       }
 
-    def validateIndex(es: ElasticSearchProjection, index: IndexLabel) = toCatsIO {
+    def validateIndex(es: ElasticSearchProjection, index: IndexLabel) =
       client
         .createIndex(index, Some(es.mapping), es.settings)
-        .mapError {
+        .adaptError {
           case err: HttpClientStatusError => InvalidElasticSearchProjectionPayload(err.jsonBody)
-          case err                        => WrappedElasticSearchClientError(err)
+          case err: HttpClientError       => WrappedElasticSearchClientError(err)
         }
         .void
-    }
 
-    val checkRemoteEvent: RemoteProjectSource => IO[Unit] = deltaClient.checkElems
+    val checkRemoteEvent: RemoteProjectSource => IO[Unit] = deltaClient.checkElems(_).toCatsIO
 
     val validateSource: CompositeViewSource => IO[Unit] = {
       case _: ProjectSource             => IO.unit

diff --git a/...scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeSpaces.scala b/...scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeSpaces.scala
@@ -47,7 +47,7 @@ object CompositeSpaces {
       val result       = view.value.projections.foldLeft[IO[Unit]](createCommon) {
         case (acc, e: ElasticSearchProjection) =>
           val index = projectionIndex(e, view.uuid, prefix)
-          acc >> esClient.createIndex(index, Some(e.mapping), e.settings).toCatsIO.void
+          acc >> esClient.createIndex(index, Some(e.mapping), e.settings).void
         case (acc, s: SparqlProjection)        =>
           val namespace = projectionNamespace(s, view.uuid, prefix)
           acc >> blazeClient.createNamespace(namespace).toCatsIO.void
@@ -69,7 +69,7 @@ object CompositeSpaces {
         projection match {
           case e: ElasticSearchProjection =>
             val index = projectionIndex(e, view.uuid, prefix)
-            esClient.deleteIndex(index).toCatsIO.void
+            esClient.deleteIndex(index).void
           case s: SparqlProjection        =>
             val namespace = projectionNamespace(s, view.uuid, prefix)
             blazeClient.deleteNamespace(namespace).toCatsIO.void

diff --git a/...t/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/ElasticSearchQuerySpec.scala b/...t/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/ElasticSearchQuerySpec.scala
@@ -3,6 +3,7 @@ package ch.epfl.bluebrain.nexus.delta.plugins.compositeviews
 import akka.http.scaladsl.model.HttpRequest
 import akka.http.scaladsl.model.Uri.Query
 import cats.data.NonEmptyList
+import cats.effect.IO
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.model.permissions
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.CompositeViewDef.ActiveViewDef
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.projectionIndex
@@ -19,7 +20,6 @@ import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclSimpleCheck
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress
 import ch.epfl.bluebrain.nexus.delta.sdk.error.ServiceError.AuthorizationFailed
 import ch.epfl.bluebrain.nexus.delta.sdk.generators.ProjectGen
-import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClient.HttpResult
 import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClientError.HttpUnexpectedError
 import ch.epfl.bluebrain.nexus.delta.sdk.identities.model.Caller
 import ch.epfl.bluebrain.nexus.delta.sdk.implicits._
@@ -32,7 +32,6 @@ import ch.epfl.bluebrain.nexus.testkit._
 import ch.epfl.bluebrain.nexus.testkit.scalatest.ce.CatsEffectSpec
 import io.circe.syntax._
 import io.circe.{Json, JsonObject}
-import monix.bio.IO
 import org.scalatest.CancelAfterFailure
 
 import java.util.UUID
@@ -126,7 +125,7 @@ class ElasticSearchQuerySpec extends CatsEffectSpec with CirceLiteral with Cance
       q: JsonObject,
       indices: Set[String],
       qp: Query
-  ): HttpResult[Json] =
+  ): IO[Json] =
     if (q == query)
       IO.pure(Json.arr(indices.foldLeft(Seq.empty[Json])((acc, idx) => acc :+ indexResults(idx).asJson): _*))
     else IO.raiseError(HttpUnexpectedError(HttpRequest(), ""))

diff --git a/...h/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeIndexingSuite.scala b/...h/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeIndexingSuite.scala
@@ -87,7 +87,11 @@ trait CompositeIndexingFixture extends CatsEffectSuite with BioRunContext {
     )
 
   private def resource(sinkConfig: SinkConfig): Resource[IO, Setup] = {
-    (Doobie.resource(), ElasticSearchClientSetup.resource(), BlazegraphClientSetup.resource())
+    (
+      Doobie.resource().mapK(taskToIoK),
+      ElasticSearchClientSetup.resource(),
+      BlazegraphClientSetup.resource().mapK(taskToIoK)
+    )
       .parMapN { case (xas, esClient, bgClient) =>
         val compositeRestartStore = new CompositeRestartStore(xas)
         val projections           =
@@ -96,7 +100,6 @@ trait CompositeIndexingFixture extends CatsEffectSuite with BioRunContext {
         val sinks                 = CompositeSinks(prefix, esClient, bgClient, compositeConfig.copy(sinkConfig = sinkConfig))
         Setup(esClient, bgClient, projections, spaces, sinks)
       }
-      .mapK(taskToIoK)
   }
 
   def suiteLocalFixture(name: String, sinkConfig: SinkConfig): ResourceFixture.IOFixture[Setup] =
@@ -336,12 +339,12 @@ abstract class CompositeIndexingSuite(sinkConfig: SinkConfig, query: SparqlConst
       _ <- spaces.init(view)
       _ <- bgClient.existsNamespace(commonNs).toCatsIO.assertEquals(true)
       _ <- bgClient.existsNamespace(sparqlNamespace).toCatsIO.assertEquals(true)
-      _ <- esClient.existsIndex(elasticIndex).toCatsIO.assertEquals(true)
+      _ <- esClient.existsIndex(elasticIndex).assertEquals(true)
       // Delete them on destroy
       _ <- spaces.destroyAll(view)
       _ <- bgClient.existsNamespace(commonNs).toCatsIO.assertEquals(false)
       _ <- bgClient.existsNamespace(sparqlNamespace).toCatsIO.assertEquals(false)
-      _ <- esClient.existsIndex(elasticIndex).toCatsIO.assertEquals(false)
+      _ <- esClient.existsIndex(elasticIndex).assertEquals(false)
     } yield ()
   }
 
@@ -361,17 +364,17 @@ abstract class CompositeIndexingSuite(sinkConfig: SinkConfig, query: SparqlConst
       _ <- spaces.init(view)
       _ <- bgClient.existsNamespace(commonNs).toCatsIO.assertEquals(true)
       _ <- bgClient.existsNamespace(sparqlNamespace).toCatsIO.assertEquals(true)
-      _ <- esClient.existsIndex(elasticIndex).toCatsIO.assertEquals(true)
+      _ <- esClient.existsIndex(elasticIndex).assertEquals(true)
       // Delete the blazegraph projection
       _ <- spaces.destroyProjection(view, blazegraphProjection)
       _ <- bgClient.existsNamespace(commonNs).toCatsIO.assertEquals(true)
       _ <- bgClient.existsNamespace(sparqlNamespace).toCatsIO.assertEquals(false)
-      _ <- esClient.existsIndex(elasticIndex).toCatsIO.assertEquals(true)
+      _ <- esClient.existsIndex(elasticIndex).assertEquals(true)
       // Delete the elasticsearch projection
       _ <- spaces.destroyProjection(view, elasticSearchProjection)
       _ <- bgClient.existsNamespace(commonNs).toCatsIO.assertEquals(true)
       _ <- bgClient.existsNamespace(sparqlNamespace).toCatsIO.assertEquals(false)
-      _ <- esClient.existsIndex(elasticIndex).toCatsIO.assertEquals(false)
+      _ <- esClient.existsIndex(elasticIndex).assertEquals(false)
     } yield ()
   }
 
@@ -574,14 +577,13 @@ abstract class CompositeIndexingSuite(sinkConfig: SinkConfig, query: SparqlConst
   private def checkElasticSearchDocuments(index: IndexLabel, expected: Json*): IO[Unit] = {
     val page = FromPagination(0, 5000)
     for {
-      _       <- esClient.refresh(index).toCatsIO
+      _       <- esClient.refresh(index)
       results <- esClient
                    .search(
                      QueryBuilder.empty.withSort(SortList(List(Sort("@id")))).withPage(page),
                      Set(index.value),
                      Query.Empty
                    )
-                   .toCatsIO
       _        = assertEquals(results.sources.size, expected.size)
       _        = results.sources.zip(expected).foreach { case (obtained, expected) =>
                    obtained.asJson.equalsIgnoreArrayOrder(expected)

diff --git a/.../ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchServiceDependency.scala b/.../ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchServiceDependency.scala
@@ -4,7 +4,6 @@ import cats.effect.IO
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient
 import ch.epfl.bluebrain.nexus.delta.sdk.ServiceDependency
 import ch.epfl.bluebrain.nexus.delta.sdk.model.ComponentDescription.ServiceDescription
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 
 /**
   * Describes the remote storage [[ServiceDependency]] providing a way to extract the [[ServiceDescription]] from a

diff --git a/...n/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViewsQuery.scala b/...n/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViewsQuery.scala
@@ -94,7 +94,9 @@ final class ElasticSearchViewsQueryImpl private[elasticsearch] (
     for {
       view    <- viewStore.fetch(id, project).toCatsIO
       indices <- extractIndices(view)
-      search  <- client.search(query, indices, qp)(SortList.empty).mapError(WrappedElasticSearchClientError)
+      search  <- client.search(query, indices, qp)(SortList.empty).adaptError { case e: HttpClientError =>
+                   WrappedElasticSearchClientError(e)
+                 }
     } yield search
   }
 
@@ -119,7 +121,7 @@ final class ElasticSearchViewsQueryImpl private[elasticsearch] (
       _      <- aclCheck.authorizeForOr(project, permissions.write)(AuthorizationFailed(project, permissions.write))
       view   <- viewStore.fetch(id, project).toCatsIO
       idx    <- indexOrError(view, id)
-      search <- client.mapping(IndexLabel.unsafe(idx)).toCatsIO.adaptError { case e: HttpClientError =>
+      search <- client.mapping(IndexLabel.unsafe(idx)).adaptError { case e: HttpClientError =>
                   WrappedElasticSearchClientError(e)
                 }
     } yield search

diff --git a/...in/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/EventMetricsProjection.scala b/...in/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/EventMetricsProjection.scala
@@ -2,7 +2,6 @@ package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
 import cats.data.NonEmptyChain
 import cats.effect.{ContextShift, IO, Timer}
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient.Refresh
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.{ElasticSearchClient, IndexLabel}
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.indexing.ElasticSearchSink

diff --git a/...scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ValidateElasticSearchView.scala b/...scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ValidateElasticSearchView.scala
@@ -2,12 +2,10 @@ package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
 import cats.effect.IO
 import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.{ElasticSearchClient, IndexLabel}
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.model.ElasticSearchViewRejection.{InvalidElasticSearchIndexPayload, InvalidPipeline, InvalidViewReferences, PermissionIsNotDefined, TooManyViewReferences, WrappedElasticSearchClientError}
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.model.ElasticSearchViewValue.{AggregateElasticSearchViewValue, IndexingElasticSearchViewValue}
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.model.{DefaultMapping, DefaultSettings, ElasticSearchViewValue}
-import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClient.HttpResult
 import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClientError
 import ch.epfl.bluebrain.nexus.delta.sdk.http.HttpClientError.HttpClientStatusError
 import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions
@@ -55,7 +53,7 @@ object ValidateElasticSearchView {
   def apply(
       validatePipeChain: PipeChain => Either[ProjectionErr, Unit],
       fetchPermissionSet: IO[Set[Permission]],
-      createIndex: (IndexLabel, Option[JsonObject], Option[JsonObject]) => HttpResult[Unit],
+      createIndex: (IndexLabel, Option[JsonObject], Option[JsonObject]) => IO[Unit],
       prefix: String,
       maxViewRefs: Int,
       xas: Transactors,
@@ -81,7 +79,7 @@ object ValidateElasticSearchView {
                IndexLabel.fromView(prefix, uuid, indexingRev),
                value.mapping.orElse(Some(defaultMapping.value)),
                value.settings.orElse(Some(defaultSettings.value))
-             ).toCatsIO
+             )
                .adaptError {
                  case err: HttpClientStatusError => InvalidElasticSearchIndexPayload(err.jsonBody)
                  case err: HttpClientError       => WrappedElasticSearchClientError(err)