Rename stream to data_service.all_events

The stream is not only for search, rename it according to the producer. See comments in https://github.com/SwissDataScienceCenter/renku-data-services/pull/377/files/acba8f289afa1023a54c40d6836139e44a964c98#r1743224983
SwissDataScienceCenter · Sep 4, 2024 · 753f07c · 753f07c
1 parent b8ffdb9
commit 753f07c
Show file tree

Hide file tree

Showing 19 changed files with 23 additions and 23 deletions.
diff --git a/flake.nix b/flake.nix
@@ -66,7 +66,7 @@
         groupMemberAdded = "groupMemberAdded";
         groupMemberUpdated = "groupMemberUpdated";
         groupMemberRemoved = "groupMemberRemoved";
-        searchSync = "search.sync";
+        dataServiceAllEvents = "data_service.all_events";
       };
 
       queueNameConfig = with nixpkgs.lib; mapAttrs' (key: qn: nameValuePair "RS_REDIS_QUEUE_${key}" qn) queueNames;

diff --git a/modules/config-values/src/main/scala/io/renku/search/config/QueuesConfig.scala b/modules/config-values/src/main/scala/io/renku/search/config/QueuesConfig.scala
@@ -38,7 +38,7 @@ final case class QueuesConfig(
     groupMemberAdded: QueueName,
     groupMemberUpdated: QueueName,
     groupMemberRemoved: QueueName,
-    searchSync: QueueName
+    dataServiceAllEvents: QueueName
 ):
   lazy val all: Set[QueueName] = Set(
     projectCreated,
@@ -56,7 +56,7 @@ final case class QueuesConfig(
     groupMemberAdded,
     groupMemberUpdated,
     groupMemberRemoved,
-    searchSync
+    dataServiceAllEvents
   )
 
 object QueuesConfig:
@@ -77,5 +77,5 @@ object QueuesConfig:
       ConfigValues.eventQueue("groupMemberAdded"),
       ConfigValues.eventQueue("groupMemberUpdated"),
       ConfigValues.eventQueue("groupMemberRemoved"),
-      ConfigValues.eventQueue("searchSync")
+      ConfigValues.eventQueue("dataServiceAllEvents")
     ).mapN(QueuesConfig.apply)
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/groups/AddCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/groups/AddCmd.scala
@@ -40,6 +40,6 @@ object AddCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/groups/MemberAddCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/groups/MemberAddCmd.scala
@@ -40,6 +40,6 @@ object MemberAddCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/groups/MemberRemoveCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/groups/MemberRemoveCmd.scala
@@ -40,6 +40,6 @@ object MemberRemoveCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/groups/MemberUpdateCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/groups/MemberUpdateCmd.scala
@@ -40,6 +40,6 @@ object MemberUpdateCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/groups/RemoveCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/groups/RemoveCmd.scala
@@ -40,6 +40,6 @@ object RemoveCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/groups/UpdateCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/groups/UpdateCmd.scala
@@ -40,6 +40,6 @@ object UpdateCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/projects/CreateCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/projects/CreateCmd.scala
@@ -73,6 +73,6 @@ object CreateCmd extends CommonOpts:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/projects/MemberAddCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/projects/MemberAddCmd.scala
@@ -41,6 +41,6 @@ object MemberAddCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/projects/MemberRemoveCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/projects/MemberRemoveCmd.scala
@@ -40,6 +40,6 @@ object MemberRemoveCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/projects/MemberUpdateCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/projects/MemberUpdateCmd.scala
@@ -42,6 +42,6 @@ object MemberUpdateCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/projects/RemoveCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/projects/RemoveCmd.scala
@@ -40,6 +40,6 @@ object RemoveCmd extends CommonOpts:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/projects/UpdateCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/projects/UpdateCmd.scala
@@ -67,6 +67,6 @@ object UpdateCmd extends CommonOpts:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/users/AddCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/users/AddCmd.scala
@@ -52,6 +52,6 @@ object AddCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/users/RemoveCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/users/RemoveCmd.scala
@@ -40,6 +40,6 @@ object RemoveCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-cli/src/main/scala/io/renku/search/cli/users/UpdateCmd.scala b/modules/search-cli/src/main/scala/io/renku/search/cli/users/UpdateCmd.scala
@@ -52,6 +52,6 @@ object UpdateCmd:
       for
         queuesCfg <- QueuesConfig.config.load[IO]
         msg <- Services.createMessage(cfg.asPayload)
-        _ <- queue.enqueue(queuesCfg.searchSync, msg)
+        _ <- queue.enqueue(queuesCfg.dataServiceAllEvents, msg)
       yield ExitCode.Success
     }
diff --git a/modules/search-provision/src/main/scala/io/renku/search/provision/MessageHandlers.scala b/modules/search-provision/src/main/scala/io/renku/search/provision/MessageHandlers.scala
@@ -49,9 +49,9 @@ final class MessageHandlers[F[_]: Async](
 
   def getAll: Map[String, F[Unit]] = tasks
 
-  val searchSync = add(
-    cfg.searchSync,
-    SyncMessageHandler(steps(cfg.searchSync), maxConflictRetries).create
+  val allEvents = add(
+    cfg.dataServiceAllEvents,
+    SyncMessageHandler(steps(cfg.dataServiceAllEvents), maxConflictRetries).create
   )
 
   val projectCreated: Stream[F, Unit] =

diff --git a/modules/search-provision/src/test/scala/io/renku/search/provision/ProvisioningSuite.scala b/modules/search-provision/src/test/scala/io/renku/search/provision/ProvisioningSuite.scala
@@ -52,7 +52,7 @@ trait ProvisioningSuite extends CatsEffectSuite with SearchSolrSuite with QueueS
     groupMemberAdded = QueueName("groupMemberAdded"),
     groupMemberUpdated = QueueName("groupMemberUpdated"),
     groupMemberRemoved = QueueName("groupMemberRemoved"),
-    searchSync = QueueName("searchSync")
+    dataServiceAllEvents = QueueName("dataServiceAllEvents")
   )
 
   override def munitIOTimeout: Duration = Duration(1, "min")