Rollback changes to rdf, sdk and elasticsearch

BlueBrain · Oct 17, 2023 · 9e62293 · 9e62293
1 parent 0e7f7b0
commit 9e62293
Show file tree

Hide file tree

Showing 23 changed files with 111 additions and 134 deletions.
diff --git a/...el/src/main/scala/ch/epfl/bluebrain/nexus/delta/kernel/utils/ClasspathResourceUtils.scala b/...el/src/main/scala/ch/epfl/bluebrain/nexus/delta/kernel/utils/ClasspathResourceUtils.scala
@@ -5,19 +5,18 @@ import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceError.{Invali
 import com.github.jknack.handlebars.{EscapingStrategy, Handlebars}
 import io.circe.parser.parse
 import io.circe.{Json, JsonObject}
-import cats.syntax.all._
+import monix.bio.IO
 
 import java.io.InputStream
 import java.util.Properties
 import scala.io.{Codec, Source}
 import scala.jdk.CollectionConverters._
-import cats.effect.IO
-import io.circe.ParsingFailure
 
 trait ClasspathResourceUtils {
 
-  final def absolutePath(resourcePath: String)(implicit classLoader: ClassLoader): IO[String] =
-    IO.fromOption(Option(getClass.getResource(resourcePath)) orElse Option(classLoader.getResource(resourcePath)))(
+  final def absolutePath(resourcePath: String)(implicit classLoader: ClassLoader): IO[ResourcePathNotFound, String] =
+    IO.fromOption(
+      Option(getClass.getResource(resourcePath)) orElse Option(classLoader.getResource(resourcePath)),
       ResourcePathNotFound(resourcePath)
     ).map(_.getPath)
 
@@ -30,11 +29,11 @@ trait ClasspathResourceUtils {
     *   the content of the referenced resource as an [[InputStream]] or a [[ClasspathResourceError]] when the resource
     *   is not found
     */
-  def ioStreamOf(resourcePath: String)(implicit classLoader: ClassLoader): IO[InputStream] =
-    IO.defer {
+  def ioStreamOf(resourcePath: String)(implicit classLoader: ClassLoader): IO[ClasspathResourceError, InputStream] =
+    IO.deferAction { _ =>
       lazy val fromClass  = Option(getClass.getResourceAsStream(resourcePath))
       val fromClassLoader = Option(classLoader.getResourceAsStream(resourcePath))
-      IO.fromOption(fromClass orElse fromClassLoader)(ResourcePathNotFound(resourcePath))
+      IO.fromOption(fromClass orElse fromClassLoader, ResourcePathNotFound(resourcePath))
     }
 
   /**
@@ -50,7 +49,7 @@ trait ClasspathResourceUtils {
   final def ioContentOf(
       resourcePath: String,
       attributes: (String, Any)*
-  )(implicit classLoader: ClassLoader): IO[String] =
+  )(implicit classLoader: ClassLoader): IO[ClasspathResourceError, String] =
     resourceAsTextFrom(resourcePath).map {
       case text if attributes.isEmpty => text
       case text                       => handleBars.compileInline(text).apply(attributes.toMap.asJava)
@@ -68,7 +67,7 @@ trait ClasspathResourceUtils {
     */
   final def ioPropertiesOf(resourcePath: String)(implicit
       classLoader: ClassLoader
-  ): IO[Map[String, String]] =
+  ): IO[ClasspathResourceError, Map[String, String]] =
     ioStreamOf(resourcePath).map { is =>
       val props = new Properties()
       props.load(is)
@@ -88,10 +87,10 @@ trait ClasspathResourceUtils {
   final def ioJsonContentOf(
       resourcePath: String,
       attributes: (String, Any)*
-  )(implicit classLoader: ClassLoader): IO[Json] =
+  )(implicit classLoader: ClassLoader): IO[ClasspathResourceError, Json] =
     for {
       text <- ioContentOf(resourcePath, attributes: _*)
-      json <- IO.fromEither(parse(text)).adaptError { case e: ParsingFailure => InvalidJson(resourcePath, text, e) }
+      json <- IO.fromEither(parse(text)).mapError(InvalidJson(resourcePath, text, _))
     } yield json
 
   /**
@@ -106,15 +105,15 @@ trait ClasspathResourceUtils {
     */
   final def ioJsonObjectContentOf(resourcePath: String, attributes: (String, Any)*)(implicit
       classLoader: ClassLoader
-  ): IO[JsonObject] =
+  ): IO[ClasspathResourceError, JsonObject] =
     for {
       json    <- ioJsonContentOf(resourcePath, attributes: _*)
-      jsonObj <- IO.fromOption(json.asObject)(InvalidJsonObject(resourcePath))
+      jsonObj <- IO.fromOption(json.asObject, InvalidJsonObject(resourcePath))
     } yield jsonObj
 
   private def resourceAsTextFrom(resourcePath: String)(implicit
       classLoader: ClassLoader
-  ): IO[String] =
+  ): IO[ClasspathResourceError, String] =
     ioStreamOf(resourcePath).map(is => Source.fromInputStream(is)(Codec.UTF8).mkString)
 }
 

diff --git a/...rc/test/scala/ch/epfl/bluebrain/nexus/delta/kernel/utils/ClasspathResourceUtilsSpec.scala b/...rc/test/scala/ch/epfl/bluebrain/nexus/delta/kernel/utils/ClasspathResourceUtilsSpec.scala
@@ -1,26 +1,26 @@
 package ch.epfl.bluebrain.nexus.delta.kernel.utils
 
-import cats.effect.IO
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceError.{InvalidJson, InvalidJsonObject, ResourcePathNotFound}
 import io.circe.syntax._
 import io.circe.{Json, JsonObject}
+import monix.bio.IO
+import monix.execution.Scheduler
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpecLike
 
-import ClasspathResourceUtils._
-
-class ClasspathResourceUtilsSpec extends AnyWordSpecLike with Matchers with ScalaFutures {
+class ClasspathResourceUtilsSpec extends AnyWordSpecLike with Matchers with ClasspathResourceUtils with ScalaFutures {
+  implicit private val sc: Scheduler            = Scheduler.global
   implicit private val classLoader: ClassLoader = getClass.getClassLoader
 
-  private def accept[A](io: IO[A]): A =
-    io.attempt.unsafeRunSync() match {
+  private def accept[E, A](io: IO[E, A]): A =
+    io.attempt.runSyncUnsafe() match {
       case Left(value)  => fail(s"Expected Right, found Left with value '$value'")
       case Right(value) => value
     }
 
-  private def reject[A](io: IO[A]): Throwable =
-    io.attempt.unsafeRunSync() match {
+  private def reject[E, A](io: IO[E, A]): E =
+    io.attempt.runSyncUnsafe() match {
       case Left(value)  => value
       case Right(value) => fail(s"Expected Left, found Right with value '$value'")
     }

diff --git a/...rc/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/BlazegraphViewsQuery.scala b/...rc/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/BlazegraphViewsQuery.scala
@@ -1,6 +1,5 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.blazegraph
 
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.kernel.search.Pagination.FromPagination
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceUtils.ioContentOf
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.SparqlQueryResponseType.{Aux, SparqlResultsJson}

diff --git a/...main/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/client/BlazegraphClient.scala b/...main/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/client/BlazegraphClient.scala
@@ -7,7 +7,6 @@ import akka.http.scaladsl.model.headers.{BasicHttpCredentials, HttpCredentials,
 import akka.http.scaladsl.model.{HttpEntity, HttpHeader, Uri}
 import akka.http.scaladsl.unmarshalling.FromEntityUnmarshaller
 import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers.stringUnmarshaller
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceUtils
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.BlazegraphClient.timeoutHeader
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.SparqlClientError.{InvalidCountRequest, WrappedHttpClientError}
@@ -41,7 +40,7 @@ class BlazegraphClient(
 
   // TODODODODODO Sort out memoization - maybe inject in files values?
   private val defaultProperties =
-    ClasspathResourceUtils.ioPropertiesOf("blazegraph/index.properties")
+    ClasspathResourceUtils.ioPropertiesOf("blazegraph/index.properties").hideErrors.memoizeOnSuccess
 
   override def query[R <: SparqlQueryResponse](
       indices: Iterable[String],

diff --git a/...l/bluebrain/nexus/delta/plugins/compositeviews/migration/MigrateCompositeViewsSuite.scala b/...l/bluebrain/nexus/delta/plugins/compositeviews/migration/MigrateCompositeViewsSuite.scala
@@ -1,7 +1,6 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.migration
 
 import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceUtils
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.CompositeViews
 import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.migration.MigrateCompositeViews.{eventsToMigrate, statesToMigrate}
@@ -21,7 +20,7 @@ import doobie.Get
 import doobie.implicits._
 import io.circe.JsonObject
 import io.circe.syntax.EncoderOps
-import monix.bio.{IO, UIO}
+import monix.bio.IO
 import munit.{AnyFixture, Location}
 
 import java.time.Instant
@@ -112,7 +111,7 @@ object MigrateCompositeViewsSuite extends ClasspathResourceUtils {
     } yield (project, id, rev)
   }
 
-  def loadEvent(jsonPath: String)(implicit xas: Transactors, classLoader: ClassLoader): IO[Unit, Unit] = {
+  def loadEvent(jsonPath: String)(implicit xas: Transactors, classLoader: ClassLoader) = {
     def insert(project: ProjectRef, id: String, rev: Int, json: JsonObject) =
       sql"""
            | INSERT INTO scoped_events (
@@ -132,17 +131,17 @@ object MigrateCompositeViewsSuite extends ClasspathResourceUtils {
            |  $rev,
            |  ${json.asJson},
            |  ${Instant.EPOCH}
-           | )""".stripMargin.update.run.void.transact(xas.write).hideErrors
+           | )""".stripMargin.update.run.void.transact(xas.write)
 
     for {
-      json               <- ioJsonObjectContentOf(jsonPath): UIO[JsonObject]
+      json               <- ioJsonObjectContentOf(jsonPath)
       (project, id, rev) <- extractIdentifiers(json)
       _                  <- insert(project, id, rev, json)
     } yield ()
   }
 
   def loadState(tag: Tag, jsonPath: String)(implicit xas: Transactors, classLoader: ClassLoader) = {
-    def insert(project: ProjectRef, id: String, rev: Int, json: JsonObject): UIO[Unit] =
+    def insert(project: ProjectRef, id: String, rev: Int, json: JsonObject) =
       sql"""
            | INSERT INTO scoped_states (
            |  type,
@@ -165,10 +164,10 @@ object MigrateCompositeViewsSuite extends ClasspathResourceUtils {
            |  ${json.asJson},
            |  ${false},
            |  ${Instant.EPOCH}
-           | )""".stripMargin.update.run.void.transact(xas.write).hideErrors
+           | )""".stripMargin.update.run.void.transact(xas.write)
 
     for {
-      json               <- ioJsonObjectContentOf(jsonPath): UIO[JsonObject]
+      json               <- ioJsonObjectContentOf(jsonPath)
       (project, id, rev) <- extractIdentifiers(json)
       _                  <- insert(project, id, rev, json)
     } yield ()

diff --git a/...scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchPluginModule.scala b/...scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchPluginModule.scala
@@ -2,7 +2,6 @@ package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
 import akka.actor.typed.ActorSystem
 import cats.effect.Clock
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration.toMonixBIO
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UUIDF
 import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient

diff --git a/...c/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViews.scala b/...c/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViews.scala
@@ -2,7 +2,6 @@ package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
 import cats.effect.Clock
 import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.kernel.kamon.KamonMetricComponent
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.{IOUtils, UUIDF}
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.ElasticSearchViews._

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
@@ -1,7 +1,6 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
 import cats.data.NonEmptyChain
-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
@@ -1,7 +1,6 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
 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}
@@ -15,6 +14,7 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.{PipeChain, ProjectionErr}
 import io.circe.JsonObject
 import monix.bio.{IO, UIO}
+import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 
 import java.util.UUID
 

diff --git a/...cala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/client/ElasticSearchClient.scala b/...cala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/client/ElasticSearchClient.scala
@@ -9,7 +9,6 @@ import akka.http.scaladsl.model.headers.BasicHttpCredentials
 import cats.syntax.all._
 import ch.epfl.bluebrain.nexus.delta.kernel.RetryStrategy
 import ch.epfl.bluebrain.nexus.delta.kernel.RetryStrategy.logError
-import ch.epfl.bluebrain.nexus.delta.kernel.effect.migration._
 import ch.epfl.bluebrain.nexus.delta.kernel.search.Pagination
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UrlUtils
 import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient.BulkResponse.MixedOutcomes.Outcome

diff --git a/...ch/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/model/package.scala b/...ch/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/model/package.scala
@@ -1,6 +1,5 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch
 
-import cats.effect.IO
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceUtils
 import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.{contexts => nxvContexts, nxv, schemas}
 import ch.epfl.bluebrain.nexus.delta.sdk.model.ResourceF
@@ -12,6 +11,7 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.model.ResourceRef.Latest
 import com.typesafe.scalalogging.Logger
 import io.circe.syntax._
 import io.circe.{Json, JsonObject}
+import monix.bio.UIO
 
 package object model {
 
@@ -52,33 +52,37 @@ package object model {
 
   implicit private val logger: Logger = Logger("ElasticSearchPlugin")
 
-  // TODODODODODO Sort out memoization
   /**
     * Default elasticsearch mapping for a view
     */
-  val defaultElasticsearchMapping: IO[JsonObject] = ClasspathResourceUtils
+  val defaultElasticsearchMapping: UIO[JsonObject] = ClasspathResourceUtils
     .ioJsonObjectContentOf("defaults/default-mapping.json")
     .logAndDiscardErrors("loading default elasticsearch mapping")
+    .memoize
 
   /**
     * Default elasticsearch settings for a view
     */
-  val defaultElasticsearchSettings: IO[JsonObject] = ClasspathResourceUtils
+  val defaultElasticsearchSettings: UIO[JsonObject] = ClasspathResourceUtils
     .ioJsonObjectContentOf("defaults/default-settings.json")
     .logAndDiscardErrors("loading default elasticsearch settings")
+    .memoize
 
-  val emptyResults: IO[Json] = ClasspathResourceUtils
+  val emptyResults: UIO[Json] = ClasspathResourceUtils
     .ioJsonObjectContentOf("defaults/empty-results.json")
     .logAndDiscardErrors("loading empty elasticsearch results")
     .map(_.asJson)
+    .memoize
 
   /** Mapping for the event metrics index */
-  val metricsMapping: IO[JsonObject] = ClasspathResourceUtils
+  val metricsMapping: UIO[JsonObject] = ClasspathResourceUtils
     .ioJsonObjectContentOf("metrics/metrics-mapping.json")
     .logAndDiscardErrors("loading metrics mapping")
+    .memoize
 
   /** Settings for the event metrics index */
-  val metricsSettings: IO[JsonObject] = ClasspathResourceUtils
+  val metricsSettings: UIO[JsonObject] = ClasspathResourceUtils
     .ioJsonObjectContentOf("metrics/metrics-settings.json")
     .logAndDiscardErrors("loading metrics settings")
+    .memoize
 }
diff --git a/...st/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViewsSpec.scala b/...st/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/ElasticSearchViewsSpec.scala
@@ -25,7 +25,6 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef}
 import ch.epfl.bluebrain.nexus.delta.sourcing.postgres.DoobieScalaTestFixture
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.PipeChain
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.pipes.{FilterBySchema, FilterByType, FilterDeprecated}
-import ch.epfl.bluebrain.nexus.testkit.ce.CatsIOValues
 import ch.epfl.bluebrain.nexus.testkit.{EitherValuable, IOFixedClock}
 import io.circe.Json
 import io.circe.literal._
@@ -44,8 +43,7 @@ class ElasticSearchViewsSpec
     with EitherValuable
     with IOFixedClock
     with ConfigFixtures
-    with Fixtures
-    with CatsIOValues {
+    with Fixtures {
 
   private val realm                  = Label.unsafe("myrealm")
   implicit private val alice: Caller = Caller(User("Alice", realm), Set(User("Alice", realm), Group("users", realm)))

diff --git a/...a/ch/epfl/bluebrain/nexus/delta/plugins/graph/analytics/indexing/GraphAnalyticsView.scala b/...a/ch/epfl/bluebrain/nexus/delta/plugins/graph/analytics/indexing/GraphAnalyticsView.scala
@@ -1,7 +1,7 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.graph.analytics.indexing
 
 import cats.effect.IO
-import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceUtils.ioJsonObjectContentOf
+import ch.epfl.bluebrain.nexus.delta.kernel.utils.CatsEffectsClasspathResourceUtils.ioJsonObjectContentOf
 import ch.epfl.bluebrain.nexus.delta.sdk.syntax._
 import com.typesafe.scalalogging.Logger
 import io.circe.JsonObject

diff --git a/...c/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/graph/analytics/indexing/package.scala b/...c/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/graph/analytics/indexing/package.scala
@@ -1,7 +1,7 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.graph.analytics
 
 import cats.effect.IO
-import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceUtils.{ioContentOf, ioJsonObjectContentOf}
+import ch.epfl.bluebrain.nexus.delta.kernel.utils.CatsEffectsClasspathResourceUtils.{ioContentOf, ioJsonObjectContentOf}
 import ch.epfl.bluebrain.nexus.delta.plugins.graph.analytics.config.GraphAnalyticsConfig.TermAggregationsConfig
 import ch.epfl.bluebrain.nexus.delta.sdk.syntax._
 import com.typesafe.scalalogging.Logger