fix: obtaining redis client from Redis Sentinel

modules/redis-client/src/main/scala/io/renku/queue/client/QueueClient.scala

@@ -43,5 +43,5 @@ trait QueueClient[F[_]] {
 }
 
 object QueueClient:
-  def apply[F[_]: Async](redisConfig: RedisConfig): Resource[F, QueueClient[F]] =
-    RedisQueueClient[F](redisConfig)
+  def make[F[_]: Async](redisConfig: RedisConfig): Resource[F, QueueClient[F]] =
+    RedisQueueClient.make[F](redisConfig)

modules/redis-client/src/main/scala/io/renku/redis/client/ClientCreator.scala

@@ -0,0 +1,97 @@
+/*
+ * 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.redis.client
+
+import cats.MonadThrow
+import cats.effect.{Async, Resource}
+import cats.syntax.all.*
+import dev.profunktor.redis4cats.connection.{RedisClient, RedisURI}
+import dev.profunktor.redis4cats.effect.Log
+import io.lettuce.core.RedisURI as JRedisURI
+import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection
+
+import scala.collection.mutable
+import scala.jdk.CollectionConverters.given
+
+sealed private trait ClientCreator[F[_]]:
+  def makeClient: Resource[F, RedisClient]
+
+private object ClientCreator:
+
+  def apply[F[_]: Async: Log](cfg: RedisConfig): ClientCreator[F] =
+    if cfg.sentinel then new SentinelClientCreator(cfg)
+    else new SingleNodeClientCreator(cfg)
+
+private class SingleNodeClientCreator[F[_]: Async: Log](cfg: RedisConfig)
+    extends ClientCreator[F]:
+
+  override def makeClient: Resource[F, RedisClient] =
+    RedisClient[F].fromUri(cfg.asRedisUri)
+
+private class SentinelClientCreator[F[_]: Async: Log](cfg: RedisConfig)
+    extends ClientCreator[F]:
+
+  override def makeClient: Resource[F, RedisClient] =
+    makeSentinelClient >>= makeMasterNodeClient
+
+  private def makeSentinelClient =
+    RedisClient[F].fromUri(cfg.asRedisUri)
+
+  private def makeMasterNodeClient(sentinelRedisClient: RedisClient) =
+    Resource
+      .eval(connectSentinel(sentinelRedisClient) >>= findMasterNodeUri)
+      .flatMap(RedisClient[F].fromUri(_))
+
+  private def connectSentinel(
+      client: RedisClient
+  ): F[StatefulRedisSentinelConnection[String, String]] =
+    MonadThrow[F].catchNonFatal(
+      client.underlying.connectSentinel()
+    )
+
+  private def findMasterNodeUri(conn: StatefulRedisSentinelConnection[String, String]) =
+    findMasterSet >>= (findMasterNodeInfo(conn, _)) >>= findNodeUri
+
+  private def findMasterSet: F[RedisMasterSet] =
+    MonadThrow[F].fromOption(
+      cfg.maybeMasterSet,
+      new Exception("No Redis MasterSet configured")
+    )
+
+  private def findMasterNodeInfo(
+      conn: StatefulRedisSentinelConnection[String, String],
+      masterSet: RedisMasterSet
+  ): F[mutable.Map[String, String]] =
+    MonadThrow[F].catchNonFatal(
+      conn.sync().master(masterSet.value).asScala
+    )
+
+  private def findNodeUri(
+      nodeInfo: mutable.Map[String, String]
+  ): F[RedisURI] =
+    MonadThrow[F].fromOption(
+      (nodeInfo.get("ip") -> nodeInfo.get("port"))
+        .mapN { (host, port) =>
+          val builder = JRedisURI.Builder.redis(host, port.toInt)
+          cfg.maybePassword.map(_.value.toCharArray).fold(builder)(builder.withPassword)
+          cfg.maybeDB.map(_.value).fold(builder)(builder.withDatabase)
+          RedisURI.fromUnderlying(builder.build())
+        },
+      new Exception("No Redis Master node URI found")
+    )

modules/redis-client/src/main/scala/io/renku/redis/client/RedisConfig.scala

@@ -18,14 +18,22 @@
 
 package io.renku.redis.client
 
+import dev.profunktor.redis4cats.connection.RedisURI
+import io.lettuce.core.RedisURI as JRedisURI
+
 final case class RedisConfig(
     host: RedisHost,
     port: RedisPort,
     sentinel: Boolean = false,
     maybeDB: Option[RedisDB] = None,
     maybePassword: Option[RedisPassword] = None,
     maybeMasterSet: Option[RedisMasterSet] = None
-)
+):
+  lazy val asRedisUri: RedisURI =
+    val builder = JRedisURI.Builder.redis(host.value, port.value)
+    maybePassword.map(_.value.toCharArray).fold(builder)(builder.withPassword)
+    maybeDB.map(_.value).fold(builder)(builder.withDatabase)
+    RedisURI.fromUnderlying(builder.build())
 
 opaque type RedisHost = String
 object RedisHost {

modules/redis-client/src/main/scala/io/renku/redis/client/RedisQueueClient.scala

@@ -20,22 +20,25 @@ package io.renku.redis.client
 
 import cats.effect.{Async, Resource}
 import cats.syntax.all.*
+import dev.profunktor.redis4cats.connection.RedisClient
+import dev.profunktor.redis4cats.data.RedisCodec
 import dev.profunktor.redis4cats.effect.Log
 import dev.profunktor.redis4cats.streams.data.{StreamingOffset, XAddMessage, XReadMessage}
+import dev.profunktor.redis4cats.streams.{RedisStream, Streaming}
+import dev.profunktor.redis4cats.{Redis, RedisCommands}
 import fs2.Stream
 import io.renku.queue.client.*
 import scodec.bits.ByteVector
 import scribe.Scribe
 
 object RedisQueueClient:
 
-  def apply[F[_]: Async](redisConfig: RedisConfig): Resource[F, QueueClient[F]] =
+  def make[F[_]: Async](redisConfig: RedisConfig): Resource[F, QueueClient[F]] =
     given Scribe[F] = scribe.cats[F]
     given Log[F] = RedisLogger[F]
-    ConnectionCreator.create[F](redisConfig).map(new RedisQueueClient(_))
+    ClientCreator[F](redisConfig).makeClient.map(new RedisQueueClient(_))
 
-class RedisQueueClient[F[_]: Async: Log](cc: ConnectionCreator[F])
-    extends QueueClient[F] {
+class RedisQueueClient[F[_]: Async: Log](client: RedisClient) extends QueueClient[F] {
 
   private val payloadKey = "payload"
   private val encodingKey = "encoding"
@@ -52,7 +55,7 @@ class RedisQueueClient[F[_]: Async: Log](cc: ConnectionCreator[F])
           Map(payloadKey -> message, encodingKey -> encodeEncoding(encoding))
         )
       )
-    cc.createStreamingConnection
+    createStreamingConnection
       .flatMap(_.append(m))
       .map(id => MessageId(id.value))
       .compile
@@ -78,7 +81,7 @@ class RedisQueueClient[F[_]: Async: Log](cc: ConnectionCreator[F])
         .map(id => StreamingOffset.Custom[String](_, id.value))
         .getOrElse(StreamingOffset.All[String])
 
-    cc.createStreamingConnection >>= {
+    createStreamingConnection >>= {
       _.read(Set(queueName.name), chunkSize, initialOffset)
         .map(toMessage)
         .collect { case Some(m) => m }
@@ -95,18 +98,26 @@ class RedisQueueClient[F[_]: Async: Log](cc: ConnectionCreator[F])
       queueName: QueueName,
       messageId: MessageId
   ): F[Unit] =
-    cc.createStringCommands.use {
+    createStringCommands.use {
       _.set(formProcessedKey(clientId, queueName), messageId.value)
     }
 
   override def findLastProcessed(
       clientId: ClientId,
       queueName: QueueName
   ): F[Option[MessageId]] =
-    cc.createStringCommands.use {
+    createStringCommands.use {
       _.get(formProcessedKey(clientId, queueName)).map(_.map(MessageId.apply))
     }
 
   private def formProcessedKey(clientId: ClientId, queueName: QueueName) =
     s"$queueName.$clientId"
+
+  private def createStreamingConnection
+      : Stream[F, Streaming[[A] =>> Stream[F, A], String, ByteVector]] =
+    RedisStream
+      .mkStreamingConnection[F, String, ByteVector](client, StringBytesCodec.instance)
+
+  private def createStringCommands: Resource[F, RedisCommands[F, String, String]] =
+    Redis[F].fromClient(client, RedisCodec.Utf8)
 }

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

@@ -55,7 +55,7 @@ trait RedisSpec:
       apply().flatMap(createRedisCommands)
 
     override def asQueueClient(): Resource[IO, QueueClient[IO]] =
-      RedisQueueClient[IO](
+      RedisQueueClient.make[IO](
         RedisConfig(
           RedisHost(server.host),
           RedisPort(server.port)

modules/search-api/src/main/scala/io/renku/search/api/SearchApi.scala

@@ -30,4 +30,4 @@ object SearchApi:
   def apply[F[_]: Async: Network](
       solrConfig: SolrConfig
   ): Resource[F, SearchApi[F]] =
-    SearchSolrClient[F](solrConfig).map(new SearchApiImpl[F](_))
+    SearchSolrClient.make[F](solrConfig).map(new SearchApiImpl[F](_))

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

@@ -37,7 +37,8 @@ object Microservice extends IOApp:
     } yield ExitCode.Success
 
   private def startProvisioning(cfg: SearchProvisionConfig): IO[Unit] =
-    SearchProvisioner[IO](cfg.queueName, cfg.redisConfig, cfg.solrConfig)
+    SearchProvisioner
+      .make[IO](cfg.queueName, cfg.redisConfig, cfg.solrConfig)
       .evalMap(_.provisionSolr.start)
       .use(_ => IO.never)
       .handleErrorWith { err =>