feat: renku-core client

build.sbt

@@ -52,6 +52,7 @@ lazy val root = project
     entitiesViewingsCollector,
     projectAuth,
     triplesGenerator,
+    renkuCoreClient,
     knowledgeGraph
   )
 
@@ -215,6 +216,13 @@ lazy val tokenRepository = project
     AutomateHeaderPlugin
   )
 
+lazy val renkuCoreClient = project
+  .in(file("renku-core-client"))
+  .withId("renku-core-client")
+  .settings(commonSettings)
+  .dependsOn(graphCommons % "compile->compile; test->test")
+  .enablePlugins(AutomateHeaderPlugin)
+
 lazy val knowledgeGraph = project
   .in(file("knowledge-graph"))
   .withId("knowledge-graph")
@@ -231,6 +239,7 @@ lazy val knowledgeGraph = project
     graphCommons        % "compile->compile; test->test",
     entitiesSearch      % "compile->compile; test->test",
     triplesGeneratorApi % "compile->compile; test->test",
+    renkuCoreClient     % "compile->compile; test->test",
     entitiesViewingsCollector
   )
   .enablePlugins(

generators/src/test/scala/io/renku/generators/Generators.scala

@@ -194,15 +194,18 @@ object Generators {
 
   val httpPorts: Gen[Port] = choose(2000, 10000).map(Port.fromInt).map(_.getOrElse(sys.error("Invalid generated port")))
 
-  def httpUrls(hostGenerator: Gen[String] = nonEmptyStrings(),
-               pathGenerator: Gen[String] = relativePaths(minSegments = 0, maxSegments = 2)
+  def httpUrls(protocolGenerator: Gen[String] = Gen.oneOf("http", "https"),
+               hostGenerator:     Gen[String] = nonEmptyStrings(),
+               portGenerator:     Gen[Option[Port]] = Gen.some(httpPorts),
+               pathGenerator:     Gen[String] = relativePaths(minSegments = 0, maxSegments = 2)
   ): Gen[String] = for {
-    protocol <- Gen.oneOf("http", "https")
-    port     <- httpPorts
-    host     <- hostGenerator
-    path     <- pathGenerator
+    protocol  <- protocolGenerator
+    host      <- hostGenerator
+    maybePort <- portGenerator
+    path      <- pathGenerator
+    portValidated = maybePort.map(p => s":$p").getOrElse("")
     pathValidated = if (path.isEmpty) "" else s"/$path"
-  } yield s"$protocol://$host:$port$pathValidated"
+  } yield s"$protocol://$host$portValidated$pathValidated"
 
   val localHttpUrls: Gen[String] = for {
     protocol <- Gen.oneOf("http", "https")

graph-commons/src/main/scala/io/renku/http/client/GitLabClient.scala

@@ -35,7 +35,8 @@ import io.renku.http.rest.paging.model.{Page, Total}
 import io.renku.metrics.{GitLabApiCallRecorder, MetricsRegistry}
 import org.http4s.Method.{DELETE, GET, HEAD, POST, PUT}
 import org.http4s.circe.{jsonEncoder, jsonEncoderOf}
-import org.http4s.{EntityEncoder, Method, Response, Uri, UrlForm}
+import org.http4s.multipart.Multipart
+import org.http4s.{EntityEncoder, Method, Response, Uri}
 import org.typelevel.ci._
 import org.typelevel.log4cats.Logger
 
@@ -55,7 +56,7 @@ trait GitLabClient[F[_]] {
       mapResponse: ResponseMappingF[F, ResultType]
   )(implicit maybeAccessToken: Option[AccessToken]): F[ResultType]
 
-  def put[ResultType](path: Uri, endpointName: String Refined NonEmpty, payload: UrlForm)(
+  def put[ResultType](path: Uri, endpointName: String Refined NonEmpty, payload: Multipart[F])(
       mapResponse: ResponseMappingF[F, ResultType]
   )(implicit maybeAccessToken: Option[AccessToken]): F[ResultType]
 
@@ -104,13 +105,17 @@ final class GitLabClientImpl[F[_]: Async: Logger](
     result  <- super.send(request)(mapResponse)
   } yield result
 
-  override def put[ResultType](path: Uri, endpointName: String Refined NonEmpty, payload: UrlForm)(
+  override def put[ResultType](path: Uri, endpointName: String Refined NonEmpty, payload: Multipart[F])(
       mapResponse: ResponseMappingF[F, ResultType]
-  )(implicit maybeAccessToken: Option[AccessToken]): F[ResultType] = for {
-    uri     <- validateUri(show"$gitLabApiUrl/$path")
-    request <- secureNamedRequest(PUT, uri, endpointName, payload)
-    result  <- super.send(request)(mapResponse)
-  } yield result
+  )(implicit maybeAccessToken: Option[AccessToken]): F[ResultType] =
+    validateUri(show"$gitLabApiUrl/$path")
+      .flatMap(
+        secureNamedRequest(PUT, _, endpointName)
+          .map(req => req.copy(request = req.request.withEntity(payload).putHeaders(payload.headers)))
+      )
+      .flatMap(
+        send(_: NamedRequest[F])(mapResponse)
+      )
 
   override def delete[ResultType](path: Uri, endpointName: Refined[String, NonEmpty])(
       mapResponse: ResponseMappingF[F, ResultType]

graph-commons/src/main/scala/io/renku/http/client/RestClient.scala

@@ -251,6 +251,7 @@ abstract class RestClient[F[_]: Async: Logger, ThrottlingTarget](
       new MultipartBuilder(request, parts).build()
 
     class MultipartBuilder private[RequestOps] (request: Request[F], parts: Vector[Part[F]] = Vector.empty[Part[F]]) {
+
       def addPart[PartType](name: String, value: PartType)(implicit
           encoder: PartEncoder[PartType]
       ): MultipartBuilder =

graph-commons/src/test/scala/io/renku/events/consumers/subscriptions/SubscriberUrlSpec.scala

@@ -23,7 +23,6 @@ import io.renku.generators.CommonGraphGenerators.microserviceBaseUrls
 import io.renku.generators.Generators.Implicits._
 import io.renku.generators.Generators.{httpUrls, nonBlankStrings, relativePaths}
 import io.renku.microservices.MicroserviceBaseUrl
-import org.scalacheck.Gen
 import org.scalatest.matchers.should
 import org.scalatest.wordspec.AnyWordSpec
 import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks
@@ -44,7 +43,7 @@ class SubscriberUrlSpec extends AnyWordSpec with should.Matchers with ScalaCheck
   "to MicroserviceBaseUrl conversion" should {
 
     "successfully convert MicroserviceBaseUrl if well defined" in {
-      forAll(httpUrls(pathGenerator = Gen.const("")), relativePaths(minSegments = 0, maxSegments = 2)) { (url, path) =>
+      forAll(httpUrls(pathGenerator = ""), relativePaths(minSegments = 0, maxSegments = 2)) { (url, path) =>
         val pathValidated = if (path.isEmpty) "" else s"/$path"
         SubscriberUrl(s"$url$pathValidated").toUnsafe[MicroserviceBaseUrl] shouldBe MicroserviceBaseUrl(url)
       }

graph-commons/src/test/scala/io/renku/http/client/GitLabClientSpec.scala

@@ -36,10 +36,13 @@ import io.renku.logging.TestExecutionTimeRecorder
 import io.renku.metrics.GitLabApiCallRecorder
 import io.renku.stubbing.ExternalServiceStubbing
 import io.renku.testtools.IOSpec
-import org.http4s.Method.{GET, _}
+import org.http4s.MediaType.multipart
+import org.http4s.Method._
 import org.http4s.Status.{Accepted, NotFound, Ok, Unauthorized}
 import org.http4s.circe.CirceEntityCodec.circeEntityDecoder
-import org.http4s.{Header, Method, Request, Response, Status, Uri, UrlForm}
+import org.http4s.multipart.{Multiparts, Part}
+import org.http4s.util.Renderer
+import org.http4s.{Header, Method, Request, Response, Status, Uri}
 import org.scalacheck.Gen
 import org.scalamock.scalatest.MockFactory
 import org.scalatest.TryValues
@@ -226,46 +229,40 @@ class GitLabClientSpec
 
   "put" should {
 
-    val mapPutResponse: PartialFunction[(Status, Request[IO], Response[IO]), IO[Unit]] = {
-      case (Accepted, _, _)     => ().pure[IO]
-      case (Unauthorized, _, _) => UnauthorizedException.raiseError[IO, Unit]
-    }
-
     forAll(tokenScenarios) { (tokenType, accessToken: AccessToken) =>
-      s"send form data with the $tokenType to the endpoint" in new TestCase {
+      s"send the given multipart request with the $tokenType to the endpoint" in new TestCase {
+
+        implicit val mat: Option[AccessToken] = accessToken.some
+        val partName  = nonEmptyStrings().generateOne
+        val partValue = nonEmptyStrings().generateOne
 
-        val propName  = nonEmptyStrings().generateOne
-        val propValue = nonEmptyStrings().generateOne
+        val multipartPayload = Multiparts
+          .forSync[IO]
+          .flatMap(_.multipart(Vector(Part.formData[IO](partName, partValue))))
+          .unsafeRunSync()
 
         stubFor {
           put(s"/api/v4/$path")
             .withAccessToken(accessToken.some)
-            .withRequestBody(equalTo(s"$propName=$propValue"))
+            .withMultipartRequestBody(
+              aMultipart(partName).withBody(equalTo(partValue))
+            )
+            .withHeader(
+              "Content-Type",
+              containing(Renderer.renderString(multipart.`form-data`))
+            )
+            .withHeader(
+              "Content-Type",
+              containing(s"""boundary="${multipartPayload.boundary.value}"""")
+            )
             .willReturn(aResponse().withStatus(Accepted.code))
         }
 
         client
-          .put(path, endpointName, UrlForm(propName -> propValue))(mapPutResponse)(accessToken.some)
+          .put(path, endpointName, multipartPayload) { case (Accepted, _, _) => ().pure[IO] }
           .unsafeRunSync() shouldBe ()
       }
     }
-
-    "return an UnauthorizedException if remote client responds with UNAUTHORIZED" in new TestCase {
-
-      val propName  = nonEmptyStrings().generateOne
-      val propValue = nonEmptyStrings().generateOne
-
-      stubFor {
-        put(s"/api/v4/$path")
-          .withAccessToken(maybeAccessToken)
-          .withRequestBody(equalTo(s"$propName=$propValue"))
-          .willReturn(unauthorized())
-      }
-
-      intercept[Exception] {
-        client.put(path, endpointName, UrlForm(propName -> propValue))(mapPutResponse).unsafeRunSync()
-      } shouldBe UnauthorizedException
-    }
   }
 
   "delete" should {

graph-commons/src/test/scala/io/renku/interpreters/TestLogger.scala

@@ -19,23 +19,25 @@
 package io.renku.interpreters
 
 import cats.Show
-import cats.effect.kernel.Sync
+import cats.effect.Async
 import cats.syntax.all._
+import fs2.Stream
 import io.renku.interpreters.TestLogger.LogMessage._
 import org.scalatest.matchers.should
-import org.scalatest.{Assertion, Succeeded}
+import org.scalatest.{Assertion, Failed, Outcome, Succeeded}
 import org.typelevel.log4cats.Logger
 
 import java.util.concurrent.ConcurrentLinkedQueue
+import scala.concurrent.duration._
 import scala.jdk.CollectionConverters._
 
-class TestLogger[F[_]: Sync] extends Logger[F] with should.Matchers {
+class TestLogger[F[_]: Async] extends Logger[F] with should.Matchers {
 
   import TestLogger.Level._
   import TestLogger._
   import LogMessage._
 
-  private[this] val F           = Sync[F]
+  private[this] val F           = Async[F]
   private[this] val invocations = new ConcurrentLinkedQueue[LogEntry]()
 
   def getMessages(severity: Level): List[LogMessage] =
@@ -70,6 +72,27 @@ class TestLogger[F[_]: Sync] extends Logger[F] with should.Matchers {
 
   def reset(): Unit = invocations.clear()
 
+  def waitFor(expected: LogEntry*): F[Assertion] = {
+
+    val interval = 100 millis
+    val attempts = 20
+
+    val logCheck =
+      F.delay(loggedOnly(expected.toList))
+        .as(Succeeded)
+        .widen[Outcome]
+        .handleError(Failed(_))
+
+    val stream = Stream.eval(logCheck) ++
+      Stream
+        .awakeDelay[F](interval)
+        .void
+        .evalMap(_ => logCheck)
+        .take(attempts)
+        .takeThrough(_ != Succeeded)
+    stream.covary[F].compile.lastOrError.map(_ shouldBe Succeeded)
+  }
+
   private def add(entry: LogEntry): F[Unit] = F.delay {
     invocations.add(entry)
     ()
@@ -119,7 +142,7 @@ class TestLogger[F[_]: Sync] extends Logger[F] with should.Matchers {
 
 object TestLogger {
 
-  def apply[F[_]: Sync](): TestLogger[F] = new TestLogger[F]
+  def apply[F[_]: Async](): TestLogger[F] = new TestLogger[F]
 
   private[TestLogger] case class LogEntry(level: Level, message: LogMessage) {
     override lazy val toString = show"\n\t$level: $message"

graph-commons/src/test/scala/io/renku/stubbing/ExternalServiceStubbing.scala

@@ -18,6 +18,7 @@
 
 package io.renku.stubbing
 
+import cats.effect.{IO, Resource}
 import com.github.tomakehurst.wiremock.WireMockServer
 import com.github.tomakehurst.wiremock.client.WireMock.equalTo
 import com.github.tomakehurst.wiremock.client.{MappingBuilder, WireMock}
@@ -26,6 +27,7 @@ import eu.timepit.refined.api.Refined
 import eu.timepit.refined.numeric.Positive
 import io.renku.http.client.AccessToken
 import io.renku.http.client.AccessToken._
+import org.http4s.Uri
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Suite}
 
 trait ExternalServiceStubbing extends BeforeAndAfterEach with BeforeAndAfterAll {
@@ -46,6 +48,7 @@ trait ExternalServiceStubbing extends BeforeAndAfterEach with BeforeAndAfterAll
   }
 
   lazy val externalServiceBaseUrl: String = s"http://localhost:${server.port()}"
+  lazy val externalServiceBaseUri: Uri    = Uri.unsafeFromString(externalServiceBaseUrl)
 
   override def beforeEach(): Unit =
     server.resetAll()
@@ -64,4 +67,30 @@ trait ExternalServiceStubbing extends BeforeAndAfterEach with BeforeAndAfterAll
       case None                              => mappingBuilder
     }
   }
+
+  protected def otherWireMockResource =
+    Resource.make[IO, WireMockServer] {
+      IO {
+        val config = WireMockConfiguration.wireMockConfig().dynamicPort()
+        val server = new WireMockServer(config)
+
+        server.start()
+
+        server
+      }
+    }(server =>
+      IO {
+        server.shutdownServer()
+      }
+    )
+
+  protected implicit class WireMockOps(server: WireMockServer) {
+
+    lazy val baseUri: Uri = Uri.unsafeFromString(server.baseUrl())
+
+    def stubFor(mappingBuilder: MappingBuilder): Unit =
+      new WireMock(server.port()).register {
+        WireMock.stubFor(mappingBuilder)
+      }
+  }
 }

graph-commons/src/test/scala/io/renku/testtools/GitLabClientTools.scala

@@ -27,7 +27,8 @@ import io.renku.generators.Generators.Implicits._
 import io.renku.http.client.RestClient.ResponseMappingF
 import io.renku.http.client.{AccessToken, GitLabClient}
 import org.http4s.Method.{DELETE, GET, HEAD, POST, PUT}
-import org.http4s.{Method, Uri, UrlForm}
+import org.http4s.multipart.Multipart
+import org.http4s.{Method, Uri}
 import org.scalacheck.Gen
 import org.scalamock.clazz.Mock
 import org.scalamock.function.MockFunctions
@@ -73,7 +74,7 @@ trait GitLabClientTools[F[_]] {
           .repeat(expectedNumberOfCalls)
       case PUT =>
         (gitLabClient
-          .put(_: Uri, _: String Refined NonEmpty, _: UrlForm)(_: ResponseMappingF[F, ResultType])(
+          .put(_: Uri, _: String Refined NonEmpty, _: Multipart[F])(_: ResponseMappingF[F, ResultType])(
             _: Option[AccessToken]
           ))
           .expects(*, maybeEndpointName.map(new MockParameter(_)).getOrElse(*), *, capture(responseMapping), *)

helm-chart/renku-graph/templates/_helpers.tpl

@@ -90,6 +90,25 @@ If release name contains chart name it will be used as a full name.
 {{- printf "%s-commit-event-service" .Release.Name | trunc 63 | trimSuffix "-" -}}
 {{- end -}}
 {{- end -}}
+
+{{- define "renkuCoreLatest.fullname" -}}
+{{- $coreBaseName := printf "%s-core" .Release.Name -}}
+{{- printf "%s-%s" $coreBaseName (get $.Values.global.core.versions "latest").name -}}
+{{- end -}}
+
+{{/*
+Comma separated list of renku-core service names
+*/}}
+{{- define "renkuCore.serviceUrls" -}}
+{{- $serviceUrls := list -}}
+{{- $coreBaseName := printf "%s-core" .Release.Name -}}
+{{- range $i, $k := (keys .Values.global.core.versions | sortAlpha) -}}
+{{- $serviceUrl := printf "http://%s-%s" $coreBaseName (get $.Values.global.core.versions $k).name -}}
+{{- $serviceUrls = mustAppend $serviceUrls $serviceUrl -}}
+{{- end -}}
+{{- join "," $serviceUrls | quote -}}
+{{- end -}}
+
 {{/*
 Create chart name and version as used by the chart label.
 */}}

helm-chart/renku-graph/templates/knowledge-graph-deployment.yaml

@@ -55,6 +55,10 @@ spec:
               value: "http://{{ template "tokenRepository.fullname" . }}:{{ .Values.tokenRepository.service.port }}"
             - name: EVENT_LOG_BASE_URL
               value: "http://{{ template "eventLog.fullname" . }}:{{ .Values.eventLog.service.port }}"
+            - name: RENKU_CORE_LATEST_URL
+              value: "http://{{ template "renkuCoreLatest.fullname" . }}"
+            - name: RENKU_CORE_SERVICE_URLS
+              value: {{ template "renkuCore.serviceUrls" . }}
             - name: GITLAB_BASE_URL
               value: {{ .Values.gitlab.url }}
             - name: GITLAB_RATE_LIMIT