fix: TG not to write to TS until all exclusive migrations are run

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/collector/datasets/EventHandler.scala

@@ -22,12 +22,13 @@ import cats.effect.{Async, MonadCancelThrow}
 import cats.syntax.all._
 import eu.timepit.refined.auto._
 import io.renku.events.{CategoryName, consumers}
-import io.renku.events.consumers.ProcessExecutor
+import io.renku.events.consumers.{EventSchedulingResult, ProcessExecutor}
 import io.renku.triplesgenerator.api.events.DatasetViewedEvent
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
 private class EventHandler[F[_]: MonadCancelThrow: Logger](
+    precondition:              F[Option[EventSchedulingResult]],
     eventUploader:             EventUploader[F],
     processExecutor:           ProcessExecutor[F],
     override val categoryName: CategoryName = categoryName
@@ -38,7 +39,8 @@ private class EventHandler[F[_]: MonadCancelThrow: Logger](
   override def createHandlingDefinition(): EventHandlingDefinition =
     EventHandlingDefinition(
       _.event.as[DatasetViewedEvent],
-      process
+      process,
+      precondition
     )
 
   private def process(event: Event) =
@@ -48,8 +50,9 @@ private class EventHandler[F[_]: MonadCancelThrow: Logger](
 
 private object EventHandler {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder](
-      connConfig: ProjectsConnectionConfig
+      precondition: F[Option[EventSchedulingResult]],
+      connConfig:   ProjectsConnectionConfig
   ): F[consumers.EventHandler[F]] =
     (EventUploader[F](connConfig), ProcessExecutor.concurrent(processesCount = 100))
-      .mapN(new EventHandler[F](_, _))
+      .mapN(new EventHandler[F](precondition, _, _))
 }

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/collector/datasets/SubscriptionFactory.scala

@@ -21,13 +21,15 @@ package io.renku.entities.viewings.collector.datasets
 import cats.effect.Async
 import cats.syntax.all._
 import io.renku.events.consumers
+import io.renku.events.consumers.EventSchedulingResult
 import io.renku.events.consumers.subscriptions.SubscriptionMechanism
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
 object SubscriptionFactory {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder](
-      connConfig: ProjectsConnectionConfig
+      precondition: F[Option[EventSchedulingResult]],
+      connConfig:   ProjectsConnectionConfig
   ): F[(consumers.EventHandler[F], SubscriptionMechanism[F])] =
-    EventHandler[F](connConfig).map(_ -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName))
+    EventHandler[F](precondition, connConfig).map(_ -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName))
 }

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/collector/projects/activated/EventHandler.scala

@@ -23,12 +23,13 @@ import cats.syntax.all._
 import eu.timepit.refined.auto._
 import io.renku.events.{CategoryName, EventRequestContent, consumers}
 import io.renku.events.consumers.EventDecodingTools.JsonOps
-import io.renku.events.consumers.ProcessExecutor
+import io.renku.events.consumers.{EventSchedulingResult, ProcessExecutor}
 import io.renku.triplesgenerator.api.events.ProjectActivated
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
 private class EventHandler[F[_]: MonadCancelThrow: Logger](
+    precondition:              F[Option[EventSchedulingResult]],
     eventPersister:            EventPersister[F],
     processExecutor:           ProcessExecutor[F],
     override val categoryName: CategoryName = categoryName
@@ -39,7 +40,8 @@ private class EventHandler[F[_]: MonadCancelThrow: Logger](
   override def createHandlingDefinition(): EventHandlingDefinition =
     EventHandlingDefinition(
       decode,
-      process
+      process,
+      precondition
     )
 
   private lazy val decode: EventRequestContent => Either[Exception, ProjectActivated] = { req =>
@@ -55,8 +57,9 @@ private class EventHandler[F[_]: MonadCancelThrow: Logger](
 
 private object EventHandler {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder](
-      connConfig: ProjectsConnectionConfig
+      precondition: F[Option[EventSchedulingResult]],
+      connConfig:   ProjectsConnectionConfig
   ): F[consumers.EventHandler[F]] =
     (EventPersister[F](connConfig), ProcessExecutor.concurrent(processesCount = 100))
-      .mapN(new EventHandler[F](_, _))
+      .mapN(new EventHandler[F](precondition, _, _))
 }

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/collector/projects/activated/SubscriptionFactory.scala

@@ -21,13 +21,15 @@ package io.renku.entities.viewings.collector.projects.activated
 import cats.effect.Async
 import cats.syntax.all._
 import io.renku.events.consumers
+import io.renku.events.consumers.EventSchedulingResult
 import io.renku.events.consumers.subscriptions.SubscriptionMechanism
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
 object SubscriptionFactory {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder](
-      connConfig: ProjectsConnectionConfig
+      precondition: F[Option[EventSchedulingResult]],
+      connConfig:   ProjectsConnectionConfig
   ): F[(consumers.EventHandler[F], SubscriptionMechanism[F])] =
-    EventHandler[F](connConfig).map(_ -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName))
+    EventHandler[F](precondition, connConfig).map(_ -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName))
 }

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/collector/projects/viewed/EventsDequeueingInitiator.scala

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2023 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.entities.viewings.collector.projects.viewed
+
+import cats.effect.{Async, Temporal}
+import cats.syntax.all._
+import io.renku.eventsqueue.EventsQueue
+import org.typelevel.log4cats.Logger
+
+import scala.concurrent.duration._
+
+private object EventsDequeueingInitiator {
+  def apply[F[_]: Async: Logger](tsReadyCheck:     F[Boolean],
+                                 eventsQueue:      EventsQueue[F],
+                                 processor:        EventProcessor[F],
+                                 onErrorWait:      Duration = 2 seconds,
+                                 onTSNotReadyWait: Duration = 10 seconds
+  ): EventsDequeueingInitiator[F] =
+    new EventsDequeueingInitiator[F](tsReadyCheck, eventsQueue, processor, onErrorWait, onTSNotReadyWait)
+}
+
+private class EventsDequeueingInitiator[F[_]: Async: Logger](tsReadyCheck: F[Boolean],
+                                                             eventsQueue:      EventsQueue[F],
+                                                             processor:        EventProcessor[F],
+                                                             onErrorWait:      Duration = 2 seconds,
+                                                             onTSNotReadyWait: Duration = 10 seconds
+) {
+
+  def initDequeueing: F[Unit] =
+    Async[F].start(initIfTSReady).void
+
+  private def initIfTSReady: F[Unit] =
+    tsReadyCheck
+      .flatMap {
+        case true =>
+          Logger[F].info(show"Starting events dequeueing for $categoryName") >>
+            hookToTheEventsStream
+        case false =>
+          Logger[F].info(show"TS not ready for writing; dequeueing for $categoryName on hold") >>
+            Temporal[F].delayBy(initIfTSReady, onTSNotReadyWait)
+      }
+      .handleErrorWith(waitAndRetry)
+
+  private lazy val waitAndRetry: Throwable => F[Unit] =
+    Logger[F].error(_)(show"An error in the $categoryName processing pipe; restarting") >>
+      Temporal[F].delayBy(initIfTSReady, onErrorWait)
+
+  private def hookToTheEventsStream =
+    eventsQueue.acquireEventsStream(categoryName).through(processor).compile.drain
+}

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/collector/projects/viewed/SubscriptionFactory.scala

@@ -18,7 +18,7 @@
 
 package io.renku.entities.viewings.collector.projects.viewed
 
-import cats.effect.{Async, Temporal}
+import cats.effect.Async
 import cats.syntax.all._
 import io.renku.db.SessionResource
 import io.renku.events.consumers.subscriptions.SubscriptionMechanism
@@ -28,10 +28,9 @@ import io.renku.lock.Lock
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
-import scala.concurrent.duration._
-
 object SubscriptionFactory {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder, DB](
+      tsReadyCheck:    F[Boolean],
       categoryLock:    Lock[F, CategoryName],
       sessionResource: SessionResource[F, DB],
       connConfig:      ProjectsConnectionConfig
@@ -41,26 +40,7 @@ object SubscriptionFactory {
       eventsQueue    <- EventsQueue[F, DB](categoryLock).pure[F]
       handler        <- EventHandler[F](eventsQueue)
       eventProcessor <- EventProcessor[F](connConfig, eventsQueue)
-      _              <- kickOffEventsDequeueing[F](eventsQueue, eventProcessor)
+      _              <- EventsDequeueingInitiator(tsReadyCheck, eventsQueue, eventProcessor).initDequeueing
     } yield handler -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName)
   }
-
-  private[viewed] def kickOffEventsDequeueing[F[_]: Async: Logger](eventsQueue: EventsQueue[F],
-                                                                   processor:   EventProcessor[F]
-  ) = Async[F].start {
-    startDequeueingEvents(eventsQueue, processor)
-  }.void
-
-  private def startDequeueingEvents[F[_]: Async: Logger](eventsQueue: EventsQueue[F],
-                                                         processor:   EventProcessor[F]
-  ): F[Unit] =
-    Logger[F].info(show"Starting events dequeueing for $categoryName") >>
-      hookToTheEventsStream(eventsQueue, processor)
-        .handleErrorWith {
-          Logger[F].error(_)(show"An error in the $categoryName processing pipe; restarting") >>
-            Temporal[F].delayBy(startDequeueingEvents(eventsQueue, processor), 2 seconds)
-        }
-
-  private def hookToTheEventsStream[F[_]: Async](eventsQueue: EventsQueue[F], processor: EventProcessor[F]) =
-    eventsQueue.acquireEventsStream(categoryName).through(processor).compile.drain
 }

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/deletion/projects/EventHandler.scala

@@ -23,12 +23,13 @@ import cats.syntax.all._
 import eu.timepit.refined.auto._
 import io.renku.events.{CategoryName, consumers}
 import io.renku.events.consumers.EventDecodingTools.JsonOps
-import io.renku.events.consumers.ProcessExecutor
+import io.renku.events.consumers.{EventSchedulingResult, ProcessExecutor}
 import io.renku.triplesgenerator.api.events.ProjectViewingDeletion
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
 private class EventHandler[F[_]: MonadCancelThrow: Logger](
+    precondition:              F[Option[EventSchedulingResult]],
     viewingRemover:            ViewingRemover[F],
     processExecutor:           ProcessExecutor[F],
     override val categoryName: CategoryName = categoryName
@@ -39,7 +40,8 @@ private class EventHandler[F[_]: MonadCancelThrow: Logger](
   override def createHandlingDefinition(): EventHandlingDefinition =
     EventHandlingDefinition(
       decode = _.event.getProjectSlug.map(ProjectViewingDeletion.apply),
-      process
+      process,
+      precondition
     )
 
   private def process(event: Event) =
@@ -49,8 +51,9 @@ private class EventHandler[F[_]: MonadCancelThrow: Logger](
 
 private object EventHandler {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder](
-      connConfig: ProjectsConnectionConfig
+      precondition: F[Option[EventSchedulingResult]],
+      connConfig:   ProjectsConnectionConfig
   ): F[consumers.EventHandler[F]] =
     (ViewingRemover[F](connConfig), ProcessExecutor.concurrent(processesCount = 100))
-      .mapN(new EventHandler[F](_, _))
+      .mapN(new EventHandler[F](precondition, _, _))
 }

entities-viewings-collector/src/main/scala/io/renku/entities/viewings/deletion/projects/SubscriptionFactory.scala

@@ -21,13 +21,15 @@ package io.renku.entities.viewings.deletion.projects
 import cats.effect.Async
 import cats.syntax.all._
 import io.renku.events.consumers
+import io.renku.events.consumers.EventSchedulingResult
 import io.renku.events.consumers.subscriptions.SubscriptionMechanism
 import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
 import org.typelevel.log4cats.Logger
 
 object SubscriptionFactory {
   def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder](
-      connConfig: ProjectsConnectionConfig
+      precondition: F[Option[EventSchedulingResult]],
+      connConfig:   ProjectsConnectionConfig
   ): F[(consumers.EventHandler[F], SubscriptionMechanism[F])] =
-    EventHandler[F](connConfig).map(_ -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName))
+    EventHandler[F](precondition, connConfig).map(_ -> SubscriptionMechanism.noOpSubscriptionMechanism(categoryName))
 }

entities-viewings-collector/src/test/scala/io/renku/entities/viewings/collector/datasets/EventHandlerSpec.scala

@@ -22,7 +22,8 @@ import cats.effect.IO
 import cats.syntax.all._
 import io.circe.syntax._
 import io.renku.events.EventRequestContent
-import io.renku.events.consumers.ProcessExecutor
+import io.renku.events.consumers.ConsumersModelGenerators.eventSchedulingResults
+import io.renku.events.consumers.{EventSchedulingResult, ProcessExecutor}
 import io.renku.generators.Generators.Implicits._
 import io.renku.interpreters.TestLogger
 import io.renku.testtools.IOSpec
@@ -54,20 +55,27 @@ class EventHandlerSpec extends AnyWordSpec with should.Matchers with IOSpec with
     }
   }
 
-  "handlingDefinition.precondition and onRelease" should {
+  "handlingDefinition.precondition" should {
+
+    "be the given precondition" in new TestCase {
+      handler.createHandlingDefinition().precondition shouldBe precondition
+    }
+  }
+
+  "handlingDefinition.onRelease" should {
 
     "be not defined" in new TestCase {
-      handler.createHandlingDefinition().precondition.unsafeRunSync() shouldBe None
-      handler.createHandlingDefinition().onRelease                    shouldBe None
+      handler.createHandlingDefinition().onRelease shouldBe None
     }
   }
 
   private trait TestCase {
 
     val event = datasetViewedEvents.generateOne
 
-    implicit val logger: TestLogger[IO] = TestLogger[IO]()
+    implicit val logger: TestLogger[IO]                    = TestLogger[IO]()
+    val precondition:    IO[Option[EventSchedulingResult]] = eventSchedulingResults.generateSome.pure[IO]
     val eventUploader = mock[EventUploader[IO]]
-    val handler       = new EventHandler[IO](eventUploader, mock[ProcessExecutor[IO]])
+    val handler       = new EventHandler[IO](precondition, eventUploader, mock[ProcessExecutor[IO]])
   }
 }

entities-viewings-collector/src/test/scala/io/renku/entities/viewings/collector/projects/activated/EventHandlerSpec.scala

@@ -22,7 +22,8 @@ import cats.effect.IO
 import cats.syntax.all._
 import io.circe.syntax._
 import io.renku.events.EventRequestContent
-import io.renku.events.consumers.ProcessExecutor
+import io.renku.events.consumers.ConsumersModelGenerators.eventSchedulingResults
+import io.renku.events.consumers.{EventSchedulingResult, ProcessExecutor}
 import io.renku.generators.Generators.Implicits._
 import io.renku.interpreters.TestLogger
 import io.renku.testtools.IOSpec
@@ -56,8 +57,8 @@ class EventHandlerSpec extends AnyWordSpec with should.Matchers with IOSpec with
 
   "handlingDefinition.precondition" should {
 
-    "be not defined" in new TestCase {
-      handler.createHandlingDefinition().precondition.unsafeRunSync() shouldBe None
+    "be the given precondition" in new TestCase {
+      handler.createHandlingDefinition().precondition shouldBe precondition
     }
   }
 
@@ -72,8 +73,9 @@ class EventHandlerSpec extends AnyWordSpec with should.Matchers with IOSpec with
 
     val event = projectActivatedEvents.generateOne
 
-    implicit val logger: TestLogger[IO] = TestLogger[IO]()
+    implicit val logger: TestLogger[IO]                    = TestLogger[IO]()
+    val precondition:    IO[Option[EventSchedulingResult]] = eventSchedulingResults.generateSome.pure[IO]
     val tsPersister = mock[EventPersister[IO]]
-    val handler     = new EventHandler[IO](tsPersister, mock[ProcessExecutor[IO]])
+    val handler     = new EventHandler[IO](precondition, tsPersister, mock[ProcessExecutor[IO]])
   }
 }