Rename some classes, split incoming/outgoing from queries from clients (#5314)

Co-authored-by: Simon Dumas <simon.dumas@senscience.ai>

Author: imsdu

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

@@ -8,8 +8,10 @@ import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.SparqlClient
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.config.BlazegraphViewsConfig
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.indexing.BlazegraphCoordinator
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.model.{contexts, BlazegraphViewEvent}
+import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.query.IncomingOutgoingLinks
+import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.query.IncomingOutgoingLinks.Queries
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.routes.{BlazegraphSupervisionRoutes, BlazegraphViewsIndexingRoutes, BlazegraphViewsRoutes, BlazegraphViewsRoutesHandler}
-import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries.{BlazegraphSlowQueryDeleter, BlazegraphSlowQueryLogger, BlazegraphSlowQueryStore}
+import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries.{SparqlSlowQueryDeleter, SparqlSlowQueryLogger, SparqlSlowQueryStore}
 import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.{ContextValue, RemoteContextResolution}
 import ch.epfl.bluebrain.nexus.delta.rdf.utils.JsonKeyOrdering
 import ch.epfl.bluebrain.nexus.delta.sdk._
@@ -40,11 +42,11 @@ class BlazegraphPluginModule(priority: Int) extends ModuleDef {
 
   make[BlazegraphViewsConfig].from { BlazegraphViewsConfig.load(_) }
 
-  make[BlazegraphSlowQueryStore].from { (xas: Transactors) => BlazegraphSlowQueryStore(xas) }
+  make[SparqlSlowQueryStore].from { (xas: Transactors) => SparqlSlowQueryStore(xas) }
 
-  make[BlazegraphSlowQueryDeleter].fromEffect {
-    (supervisor: Supervisor, store: BlazegraphSlowQueryStore, cfg: BlazegraphViewsConfig, clock: Clock[IO]) =>
-      BlazegraphSlowQueryDeleter.start(
+  make[SparqlSlowQueryDeleter].fromEffect {
+    (supervisor: Supervisor, store: SparqlSlowQueryStore, cfg: BlazegraphViewsConfig, clock: Clock[IO]) =>
+      SparqlSlowQueryDeleter.start(
         supervisor,
         store,
         cfg.slowQueries.logTtl,
@@ -53,9 +55,8 @@ class BlazegraphPluginModule(priority: Int) extends ModuleDef {
       )
   }
 
-  make[BlazegraphSlowQueryLogger].from {
-    (cfg: BlazegraphViewsConfig, store: BlazegraphSlowQueryStore, clock: Clock[IO]) =>
-      BlazegraphSlowQueryLogger(store, cfg.slowQueries.slowQueryThreshold, clock)
+  make[SparqlSlowQueryLogger].from { (cfg: BlazegraphViewsConfig, store: SparqlSlowQueryStore, clock: Clock[IO]) =>
+    SparqlSlowQueryLogger(store, cfg.slowQueries.slowQueryThreshold, clock)
   }
 
   make[SparqlClient].named("sparql-indexing-client").from { (cfg: BlazegraphViewsConfig, as: ActorSystem) =>
@@ -123,13 +124,13 @@ class BlazegraphPluginModule(priority: Int) extends ModuleDef {
       )(baseUri)
   }
 
-  make[BlazegraphViewsQuery].fromEffect {
+  make[BlazegraphViewsQuery].from {
     (
         aclCheck: AclCheck,
         fetchContext: FetchContext,
         views: BlazegraphViews,
         client: SparqlClient @Id("sparql-query-client"),
-        slowQueryLogger: BlazegraphSlowQueryLogger,
+        slowQueryLogger: SparqlSlowQueryLogger,
         cfg: BlazegraphViewsConfig,
         xas: Transactors
     ) =>
@@ -144,12 +145,25 @@ class BlazegraphPluginModule(priority: Int) extends ModuleDef {
       )
   }
 
+  make[IncomingOutgoingLinks].fromEffect {
+    (
+        fetchContext: FetchContext,
+        views: BlazegraphViews,
+        client: SparqlClient @Id("sparql-query-client"),
+        base: BaseUri
+    ) =>
+      Queries.load.map { queries =>
+        IncomingOutgoingLinks(fetchContext, views, client, queries)(base)
+      }
+  }
+
   make[BlazegraphViewsRoutes].from {
     (
         identities: Identities,
         aclCheck: AclCheck,
         views: BlazegraphViews,
         viewsQuery: BlazegraphViewsQuery,
+        incomingOutgoingLinks: IncomingOutgoingLinks,
         baseUri: BaseUri,
         cfg: BlazegraphViewsConfig,
         cr: RemoteContextResolution @Id("aggregate"),
@@ -159,6 +173,7 @@ class BlazegraphPluginModule(priority: Int) extends ModuleDef {
       new BlazegraphViewsRoutes(
         views,
         viewsQuery,
+        incomingOutgoingLinks,
         identities,
         aclCheck
       )(

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

@@ -1,76 +1,30 @@
 package ch.epfl.bluebrain.nexus.delta.plugins.blazegraph
 
 import cats.effect.IO
-import ch.epfl.bluebrain.nexus.delta.kernel.syntax.kamonSyntax
 import ch.epfl.bluebrain.nexus.delta.kernel.kamon.KamonMetricComponent
-import ch.epfl.bluebrain.nexus.delta.kernel.search.Pagination.FromPagination
-import ch.epfl.bluebrain.nexus.delta.kernel.utils.ClasspathResourceLoader
+import ch.epfl.bluebrain.nexus.delta.kernel.syntax.kamonSyntax
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.BlazegraphViews.entityType
-import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.SparqlQueryResponseType.{Aux, SparqlResultsJson}
+import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.SparqlQueryResponseType.Aux
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client._
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.model.BlazegraphViewRejection._
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.model.BlazegraphViewValue.{AggregateBlazegraphViewValue, IndexingBlazegraphViewValue}
-import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.model.SparqlLink.{SparqlExternalLink, SparqlResourceLink}
 import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.model._
-import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries.BlazegraphSlowQueryLogger
-import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri
+import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries.SparqlSlowQueryLogger
 import ch.epfl.bluebrain.nexus.delta.rdf.query.SparqlQuery
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress.{Project => ProjectAcl}
 import ch.epfl.bluebrain.nexus.delta.sdk.error.ServiceError.AuthorizationFailed
 import ch.epfl.bluebrain.nexus.delta.sdk.identities.model.Caller
 import ch.epfl.bluebrain.nexus.delta.sdk.jsonld.ExpandIri
-import ch.epfl.bluebrain.nexus.delta.sdk.model.IdSegment.IriSegment
-import ch.epfl.bluebrain.nexus.delta.sdk.model.search.ResultEntry.UnscoredResultEntry
-import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults
-import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults.UnscoredSearchResults
-import ch.epfl.bluebrain.nexus.delta.sdk.model.{BaseUri, IdSegment}
+import ch.epfl.bluebrain.nexus.delta.sdk.model.IdSegment
 import ch.epfl.bluebrain.nexus.delta.sdk.projects.FetchContext
 import ch.epfl.bluebrain.nexus.delta.sdk.views.View.{AggregateView, IndexingView}
-import ch.epfl.bluebrain.nexus.delta.sdk.views.{ViewRef, ViewsStore}
+import ch.epfl.bluebrain.nexus.delta.sdk.views.{View, ViewRef, ViewsStore}
 import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Subject
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef
 
-import java.util.regex.Pattern.quote
-
 trait BlazegraphViewsQuery {
 
-  /**
-    * List incoming links for a given resource.
-    *
-    * @param id
-    *   the resource identifier
-    * @param projectRef
-    *   the project of the resource
-    * @param pagination
-    *   the pagination config
-    */
-  def incoming(
-      id: IdSegment,
-      projectRef: ProjectRef,
-      pagination: FromPagination
-  )(implicit caller: Caller, base: BaseUri): IO[SearchResults[SparqlLink]]
-
-  /**
-    * List outgoing links for a given resource.
-    *
-    * @param id
-    *   the resource identifier
-    * @param projectRef
-    *   the project of the resource
-    * @param pagination
-    *   the pagination config
-    * @param includeExternalLinks
-    *   whether to include links to resources not managed by Delta
-    */
-  def outgoing(
-      id: IdSegment,
-      projectRef: ProjectRef,
-      pagination: FromPagination,
-      includeExternalLinks: Boolean
-  )(implicit caller: Caller, base: BaseUri): IO[SearchResults[SparqlLink]]
-
   /**
     * Queries the blazegraph namespace (or namespaces) managed by the view with the passed ''id''. We check for the
     * caller to have the necessary query permissions on the view before performing the query.
@@ -94,130 +48,78 @@ trait BlazegraphViewsQuery {
 
 object BlazegraphViewsQuery {
 
-  private val loader = ClasspathResourceLoader.withContext(getClass)
-
   implicit private val kamonComponent: KamonMetricComponent = KamonMetricComponent(entityType.value)
 
   final def apply(
       aclCheck: AclCheck,
       fetchContext: FetchContext,
       views: BlazegraphViews,
       client: SparqlQueryClient,
-      logSlowQueries: BlazegraphSlowQueryLogger,
+      logSlowQueries: SparqlSlowQueryLogger,
       prefix: String,
       xas: Transactors
-  ): IO[BlazegraphViewsQuery] = {
-    for {
-      incomingQuery             <- loader.contentOf("blazegraph/incoming.txt")
-      outgoingWithExternalQuery <- loader.contentOf("blazegraph/outgoing_include_external.txt")
-      outgoingScopedQuery       <- loader.contentOf("blazegraph/outgoing_scoped.txt")
-      viewsStore                 = ViewsStore[BlazegraphViewRejection, BlazegraphViewState](
-                                     BlazegraphViewState.serializer,
-                                     views.fetchState(_, _),
-                                     view =>
-                                       IO.raiseWhen(view.deprecated)(ViewIsDeprecated(view.id))
-                                         .as {
-                                           view.value match {
-                                             case _: AggregateBlazegraphViewValue =>
-                                               Left(view.id)
-                                             case i: IndexingBlazegraphViewValue  =>
-                                               Right(
-                                                 IndexingView(
-                                                   ViewRef(view.project, view.id),
-                                                   BlazegraphViews.namespace(view.uuid, view.indexingRev, prefix),
-                                                   i.permission
-                                                 )
-                                               )
-                                           }
-                                         },
-                                     xas
-                                   )
-    } yield new BlazegraphViewsQuery {
+  ): BlazegraphViewsQuery = {
+    val viewsStore = ViewsStore[BlazegraphViewRejection, BlazegraphViewState](
+      BlazegraphViewState.serializer,
+      views.fetchState,
+      view =>
+        IO.raiseWhen(view.deprecated)(ViewIsDeprecated(view.id))
+          .as {
+            view.value match {
+              case _: AggregateBlazegraphViewValue =>
+                Left(view.id)
+              case i: IndexingBlazegraphViewValue  =>
+                Right(
+                  IndexingView(
+                    ViewRef(view.project, view.id),
+                    BlazegraphViews.namespace(view.uuid, view.indexingRev, prefix),
+                    i.permission
+                  )
+                )
+            }
+          },
+      xas
+    )
+    new BlazegraphViewsQuery {
 
       private val expandIri: ExpandIri[BlazegraphViewRejection] = new ExpandIri(InvalidResourceId.apply)
 
-      private def replace(query: String, id: Iri, pagination: FromPagination): String =
-        query
-          .replaceAll(quote("{id}"), id.toString)
-          .replaceAll(quote("{offset}"), pagination.from.toString)
-          .replaceAll(quote("{size}"), pagination.size.toString)
-
-      override def incoming(id: IdSegment, projectRef: ProjectRef, pagination: FromPagination)(implicit
-          caller: Caller,
-          base: BaseUri
-      ): IO[SearchResults[SparqlLink]] =
-        for {
-          p        <- fetchContext.onRead(projectRef)
-          iri      <- expandIri(id, p)
-          q         = SparqlQuery(replace(incomingQuery, iri, pagination))
-          bindings <- query(IriSegment(defaultViewId), projectRef, q, SparqlResultsJson)
-          links     = toSparqlLinks(bindings.value)
-        } yield links
-
-      override def outgoing(
-          id: IdSegment,
-          projectRef: ProjectRef,
-          pagination: FromPagination,
-          includeExternalLinks: Boolean
-      )(implicit caller: Caller, base: BaseUri): IO[SearchResults[SparqlLink]] =
-        for {
-          p            <- fetchContext.onRead(projectRef)
-          iri          <- expandIri(id, p)
-          queryTemplate = if (includeExternalLinks) outgoingWithExternalQuery else outgoingScopedQuery
-          q             = SparqlQuery(replace(queryTemplate, iri, pagination))
-          bindings     <- query(IriSegment(defaultViewId), projectRef, q, SparqlResultsJson)
-          links         = toSparqlLinks(bindings.value)
-        } yield links
-
       override def query[R <: SparqlQueryResponse](
           id: IdSegment,
           project: ProjectRef,
-          query: SparqlQuery,
+          sparqlQuery: SparqlQuery,
           responseType: Aux[R]
-      )(implicit caller: Caller): IO[R] =
+      )(implicit caller: Caller): IO[R] = {
         for {
-          view    <- viewsStore.fetch(id, project)
-          p       <- fetchContext.onRead(project)
-          iri     <- expandIri(id, p)
-          indices <- view match {
-                       case i: IndexingView  =>
-                         aclCheck
-                           .authorizeForOr(i.ref.project, i.permission)(
-                             AuthorizationFailed(i.ref.project, i.permission)
-                           )
-                           .as(Set(i.index))
-                       case a: AggregateView =>
-                         aclCheck
-                           .mapFilter[IndexingView, String](
-                             a.views,
-                             v => ProjectAcl(v.ref.project) -> v.permission,
-                             _.index
-                           )
-                     }
-          qr      <- logSlowQueries(
-                       BlazegraphQueryContext(ViewRef.apply(project, iri), query, caller.subject),
-                       client.query(indices, query, responseType).adaptError { case e: SparqlClientError =>
-                         WrappedBlazegraphClientError(e)
-                       }
-                     ).span("blazegraphQuery")
+          view       <- viewsStore.fetch(id, project)
+          p          <- fetchContext.onRead(project)
+          iri        <- expandIri(id, p)
+          namespaces <- viewToNamespaces(view)
+          queryIO     = client.query(namespaces, sparqlQuery, responseType).adaptError { case e: SparqlClientError =>
+                          WrappedBlazegraphClientError(e)
+                        }
+          qr         <- logSlowQueries(ViewRef(project, iri), sparqlQuery, caller.subject, queryIO)
         } yield qr
+      }.span("blazegraphUserQuery")
+
+      // Translate a view to the set of underlying namespaces according to the current caller acls
+      private def viewToNamespaces(view: View)(implicit caller: Caller) =
+        view match {
+          case i: IndexingView  =>
+            aclCheck
+              .authorizeForOr(i.ref.project, i.permission)(
+                AuthorizationFailed(i.ref.project, i.permission)
+              )
+              .as(Set(i.index))
+          case a: AggregateView =>
+            aclCheck
+              .mapFilter[IndexingView, String](
+                a.views,
+                v => ProjectAcl(v.ref.project) -> v.permission,
+                _.index
+              )
+        }
 
-      private def toSparqlLinks(sparqlResults: SparqlResults)(implicit
-          base: BaseUri
-      ): SearchResults[SparqlLink] = {
-        val (count, results) =
-          sparqlResults.results.bindings
-            .foldLeft((0L, List.empty[SparqlLink])) { case ((total, acc), bindings) =>
-              val newTotal = bindings.get("total").flatMap(v => v.value.toLongOption).getOrElse(total)
-              val res      = (SparqlResourceLink(bindings) orElse SparqlExternalLink(bindings))
-                .map(_ :: acc)
-                .getOrElse(acc)
-              (newTotal, res)
-            }
-        UnscoredSearchResults(count, results.map(UnscoredResultEntry(_)))
-      }
     }
   }
-
-  final case class BlazegraphQueryContext(view: ViewRef, query: SparqlQuery, subject: Subject)
 }