refactor: provisioning routes refactor

modules/config-values/src/main/scala/io/renku/search/config/ConfigValues.scala

@@ -47,8 +47,8 @@ object ConfigValues extends ConfigDecoders:
     (host, port, sentinel, maybeDB, maybePass, maybeMasterSet).mapN(RedisConfig.apply)
   }
 
-  val eventsQueueName: ConfigValue[Effect, QueueName] =
-    renv("REDIS_QUEUE_NAME").default("events").as[QueueName]
+  def eventQueue(eventType: String): ConfigValue[Effect, QueueName] =
+    renv(s"REDIS_QUEUE_$eventType").default("events").as[QueueName]
 
   val retryOnErrorDelay: ConfigValue[Effect, FiniteDuration] =
     renv("RETRY_ON_ERROR_DELAY").default("2 seconds").as[FiniteDuration]

modules/redis-client/src/test/scala/io/renku/redis/client/util/RedisSpec.scala

@@ -39,6 +39,7 @@ trait RedisSpec:
   abstract class RedisFixture extends Fixture[Resource[IO, RedisClient]]("redis"):
     def asRedisCommands(): Resource[IO, RedisCommands[IO, String, String]]
     def asRedisQueueClient(): Resource[IO, RedisQueueClient[IO]]
+    def redisConfig: RedisConfig
 
   val withRedisClient: RedisFixture = new RedisFixture:
 
@@ -54,11 +55,12 @@ trait RedisSpec:
       apply().flatMap(createRedisCommands)
 
     override def asRedisQueueClient(): Resource[IO, RedisQueueClient[IO]] =
-      RedisQueueClient.make[IO](
-        RedisConfig(
-          RedisHost(server.host),
-          RedisPort(server.port)
-        )
+      RedisQueueClient.make[IO](redisConfig)
+
+    override lazy val redisConfig: RedisConfig =
+      RedisConfig(
+        RedisHost(server.host),
+        RedisPort(server.port)
       )
 
     override def beforeAll(): Unit =

modules/search-api/src/test/scala/io/renku/search/api/SearchApiSpec.scala

@@ -26,6 +26,7 @@ import io.renku.search.query.Query
 import io.renku.search.solr.client.SearchSolrClientGenerators.*
 import io.renku.search.solr.client.SearchSolrSpec
 import io.renku.search.solr.documents.Project as SolrProject
+import io.renku.search.solr.documents.Project.given
 import munit.CatsEffectSuite
 import scribe.Scribe
 
@@ -39,7 +40,7 @@ class SearchApiSpec extends CatsEffectSuite with SearchSolrSpec:
       val project2 = projectDocumentGen("disparate", "disparate description").generateOne
       val searchApi = new SearchApiImpl[IO](client)
       for {
-        _ <- client.insertProjects(project1 :: project2 :: Nil)
+        _ <- client.insert(project1 :: project2 :: Nil)
         results <- searchApi
           .query(mkQuery("matching"))
           .map(_.fold(err => fail(s"Calling Search API failed with $err"), identity))

modules/search-provision/src/main/scala/io/renku/search/provision/Microservice.scala

@@ -19,9 +19,10 @@
 package io.renku.search.provision
 
 import cats.effect.{ExitCode, IO, IOApp, Temporal}
+import io.renku.logging.LoggingSetup
+import io.renku.search.provision.project.ProjectCreatedProvisioning
 import io.renku.search.solr.schema.Migrations
 import io.renku.solr.client.migration.SchemaMigrator
-import io.renku.logging.LoggingSetup
 import scribe.Scribe
 import scribe.cats.*
 
@@ -39,9 +40,9 @@ object Microservice extends IOApp:
     } yield ExitCode.Success
 
   private def startProvisioning(cfg: SearchProvisionConfig): IO[Unit] =
-    SearchProvisioner
-      .make[IO](cfg.queueName, cfg.redisConfig, cfg.solrConfig)
-      .evalMap(_.provisionSolr.start)
+    ProjectCreatedProvisioning
+      .make[IO](cfg.queuesConfig.projectCreated, cfg.redisConfig, cfg.solrConfig)
+      .evalMap(_.provisioningProcess.start)
       .use(_ => IO.never)
       .handleErrorWith { err =>
         Scribe[IO].error("Starting provisioning failure, retrying", err) >>

modules/search-provision/src/main/scala/io/renku/search/provision/QueuesConfig.scala

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2024 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.renku.search.provision
+
+import ciris.{ConfigValue, Effect}
+import io.renku.redis.client.QueueName
+import io.renku.search.config.ConfigValues
+
+final case class QueuesConfig(
+    projectCreated: QueueName
+)
+
+object QueuesConfig:
+  val config: ConfigValue[Effect, QueuesConfig] =
+    ConfigValues
+      .eventQueue("projectCreated")
+      .map(QueuesConfig.apply)

modules/search-provision/src/main/scala/io/renku/search/provision/SearchProvisionConfig.scala

@@ -20,27 +20,27 @@ package io.renku.search.provision
 
 import cats.syntax.all.*
 import ciris.{ConfigValue, Effect}
-import io.renku.redis.client.{QueueName, RedisConfig}
+import io.renku.redis.client.RedisConfig
 import io.renku.search.config.ConfigValues
 import io.renku.solr.client.SolrConfig
 
 import scala.concurrent.duration.FiniteDuration
 
 final case class SearchProvisionConfig(
     redisConfig: RedisConfig,
-    queueName: QueueName,
     solrConfig: SolrConfig,
     retryOnErrorDelay: FiniteDuration,
-    verbosity: Int
+    verbosity: Int,
+    queuesConfig: QueuesConfig
 )
 
 object SearchProvisionConfig:
 
   val config: ConfigValue[Effect, SearchProvisionConfig] =
     (
       ConfigValues.redisConfig,
-      ConfigValues.eventsQueueName,
       ConfigValues.solrConfig,
       ConfigValues.retryOnErrorDelay,
-      ConfigValues.logLevel
+      ConfigValues.logLevel,
+      QueuesConfig.config
     ).mapN(SearchProvisionConfig.apply)

modules/search-provision/src/main/scala/io/renku/search/provision/SearchProvisioner.scala → modules/search-provision/src/main/scala/io/renku/search/provision/SolrProvisioningProcess.scala

@@ -18,57 +18,60 @@
 
 package io.renku.search.provision
 
-import cats.MonadThrow
 import cats.effect.{Async, Resource, Temporal}
 import cats.syntax.all.*
+import cats.{MonadThrow, Show}
 import fs2.Chunk
 import fs2.io.net.Network
+import io.bullet.borer.Encoder
 import io.github.arainko.ducktape.*
-import io.renku.avro.codec.AvroReader
-import io.renku.avro.codec.decoders.all.given
-import io.renku.events.v1
-import io.renku.events.v1.{ProjectCreated, Visibility}
-import io.renku.queue.client.*
+import io.renku.avro.codec.{AvroDecoder, AvroReader}
+import io.renku.queue.client.{DataContentType, QueueClient, QueueMessage}
 import io.renku.redis.client.{ClientId, QueueName, RedisConfig}
-import io.renku.search.model.*
 import io.renku.search.solr.client.SearchSolrClient
-import io.renku.search.solr.documents.*
 import io.renku.solr.client.SolrConfig
+import org.apache.avro.Schema
 import scribe.Scribe
 
 import scala.concurrent.duration.*
 
-trait SearchProvisioner[F[_]]:
-  def provisionSolr: F[Unit]
-
-object SearchProvisioner:
+trait SolrProvisioningProcess[F[_]]:
+  def provisioningProcess: F[Unit]
 
+object SolrProvisioningProcess:
   private val clientId: ClientId = ClientId("search-provisioner")
 
-  def make[F[_]: Async: Network](
+  def make[F[_]: Async: Network: Scribe, In, Out](
       queueName: QueueName,
+      inSchema: Schema,
       redisConfig: RedisConfig,
       solrConfig: SolrConfig
-  ): Resource[F, SearchProvisioner[F]] =
+  )(using
+      Show[In],
+      Transformer[In, Out],
+      AvroDecoder[In],
+      Encoder[Out]
+  ): Resource[F, SolrProvisioningProcess[F]] =
     SearchSolrClient.make[F](solrConfig).map {
-      new SearchProvisionerImpl[F](
-        clientId,
+      new SolrProvisioningProcessImpl[F, In, Out](
         queueName,
+        inSchema,
+        clientId,
         QueueClient.make[F](redisConfig),
         _
       )
     }
 
-private class SearchProvisionerImpl[F[_]: Async](
-    clientId: ClientId,
+private class SolrProvisioningProcessImpl[F[_]: Async: Scribe, In, Out](
     queueName: QueueName,
+    inSchema: Schema,
+    clientId: ClientId,
     queueClientResource: Resource[F, QueueClient[F]],
     solrClient: SearchSolrClient[F]
-) extends SearchProvisioner[F]:
-
-  private given Scribe[F] = scribe.cats[F]
+)(using Show[In], Transformer[In, Out], AvroDecoder[In], Encoder[Out])
+    extends SolrProvisioningProcess[F]:
 
-  override def provisionSolr: F[Unit] =
+  override def provisioningProcess: F[Unit] =
     queueClientResource
       .use { queueClient =>
         findLastProcessed(queueClient) >>= { maybeLastProcessed =>
@@ -88,26 +91,29 @@ private class SearchProvisionerImpl[F[_]: Async](
   private def findLastProcessed(queueClient: QueueClient[F]) =
     queueClient.findLastProcessed(clientId, queueName)
 
-  private lazy val logInfo: ((QueueMessage, Seq[ProjectCreated])) => F[Unit] = {
-    case (m, v) =>
-      Scribe[F].info(
-        s"Received messageId: ${m.id} for projects: ${v.map(_.slug).mkString(", ")}"
-      )
+  private lazy val logInfo: ((QueueMessage, Seq[In])) => F[Unit] = { case (m, v) =>
+    Scribe[F].info(
+      "Received message " +
+        s"id: ${m.id}, " +
+        s"source: ${m.header.source}, " +
+        s"type: ${m.header.`type`} " +
+        s"for: ${v.mkString_(", ")}"
+    )
   }
 
-  private val avro = AvroReader(ProjectCreated.SCHEMA$)
+  private val avro = AvroReader(inSchema)
 
   private def decodeMessage(queueClient: QueueClient[F])(
       message: QueueMessage
-  ): F[(QueueMessage, Seq[ProjectCreated])] =
+  ): F[(QueueMessage, Seq[In])] =
     MonadThrow[F]
       .fromEither(DataContentType.from(message.header.dataContentType))
       .flatMap { ct =>
         MonadThrow[F]
           .catchNonFatal {
             ct match {
-              case DataContentType.Binary => avro.read[ProjectCreated](message.payload)
-              case DataContentType.Json => avro.readJson[ProjectCreated](message.payload)
+              case DataContentType.Binary => avro.read[In](message.payload)
+              case DataContentType.Json   => avro.readJson[In](message.payload)
             }
           }
           .map(message -> _)
@@ -116,23 +122,20 @@ private class SearchProvisionerImpl[F[_]: Async](
 
   private def pushToSolr(
       queueClient: QueueClient[F]
-  )(chunk: Chunk[(QueueMessage, Seq[ProjectCreated])]): F[Unit] =
+  )(chunk: Chunk[(QueueMessage, Seq[In])]): F[Unit] =
     chunk.toList match {
       case Nil => ().pure[F]
       case tuples =>
-        val allSolrDocs = toSolrDocuments(tuples.flatMap(_._2))
+        val docs = toSolrDocuments(tuples.flatMap(_._2))
         val (lastMessage, _) = tuples.last
         solrClient
-          .insertProjects(allSolrDocs)
+          .insert(docs)
           .flatMap(_ => markProcessed(lastMessage, queueClient))
           .onError(markProcessedOnFailure(lastMessage, queueClient))
     }
 
-  private given Transformer[v1.Visibility, projects.Visibility] =
-    (from: v1.Visibility) => projects.Visibility.unsafeFromString(from.name())
-
-  private lazy val toSolrDocuments: Seq[ProjectCreated] => Seq[Project] =
-    _.map(_.into[Project].transform(Field.default(_.score)))
+  private lazy val toSolrDocuments: Seq[In] => Seq[Out] =
+    _.map(_.to[Out])
 
   private def markProcessedOnFailure(
       message: QueueMessage,
@@ -145,5 +148,5 @@ private class SearchProvisionerImpl[F[_]: Async](
     queueClient.markProcessed(clientId, queueName, message.id)
 
   private def logAndRestart: Throwable => F[Unit] = err =>
-    Scribe[F].error("Failure in the provisioning process", err) >>
-      Temporal[F].delayBy(provisionSolr, 30 seconds)
+    Scribe[F].error(s"Failure in the provisioning process for '$queueName'", err) >>
+      Temporal[F].delayBy(provisioningProcess, 30 seconds)

modules/search-provision/src/main/scala/io/renku/search/provision/TypeTransformers.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2024 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.renku.search.provision
+
+import io.github.arainko.ducktape.Transformer
+import io.renku.events.v1
+import io.renku.search.model.projects
+
+object TypeTransformers extends TypeTransformers
+
+trait TypeTransformers:
+
+  given Transformer[v1.Visibility, projects.Visibility] =
+    (from: v1.Visibility) => projects.Visibility.unsafeFromString(from.name())

modules/search-provision/src/main/scala/io/renku/search/provision/project/ProjectCreatedProvisioning.scala

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2024 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.renku.search.provision.project
+
+import cats.Show
+import cats.effect.{Async, Resource}
+import cats.syntax.all.*
+import fs2.io.net.Network
+import io.github.arainko.ducktape.*
+import io.renku.avro.codec.decoders.all.given
+import io.renku.events.v1
+import io.renku.events.v1.{ProjectCreated, Visibility}
+import io.renku.redis.client.{QueueName, RedisConfig}
+import io.renku.search.model.*
+import io.renku.search.provision.SolrProvisioningProcess
+import io.renku.search.provision.TypeTransformers.given
+import io.renku.search.solr.documents
+import io.renku.solr.client.SolrConfig
+import scribe.Scribe
+
+trait ProjectCreatedProvisioning[F[_]] extends SolrProvisioningProcess[F]
+
+object ProjectCreatedProvisioning:
+
+  def make[F[_]: Async: Network](
+      queueName: QueueName,
+      redisConfig: RedisConfig,
+      solrConfig: SolrConfig
+  ): Resource[F, SolrProvisioningProcess[F]] =
+    given Scribe[F] = scribe.cats[F]
+    SolrProvisioningProcess.make[F, ProjectCreated, documents.Project](
+      queueName,
+      ProjectCreated.SCHEMA$,
+      redisConfig,
+      solrConfig
+    )
+
+  private given Show[ProjectCreated] =
+    Show.show[ProjectCreated](pc => show"slug '${pc.slug}'")
+
+  private given Transformer[ProjectCreated, documents.Project] =
+    _.into[documents.Project].transform(Field.default(_.score))