Merge pull request #1987 from eikek/notify-jobs

Fix notifications for collective scoped tasks

modules/backend/src/main/scala/docspell/backend/joex/FindJobOwnerAccount.scala

@@ -6,6 +6,8 @@
 
 package docspell.backend.joex
 
+import cats.syntax.all._
+
 import docspell.common.AccountId
 import docspell.scheduler.FindJobOwner
 import docspell.store.Store
@@ -15,9 +17,11 @@ import docspell.store.queries.QLogin
   * login.
   */
 object FindJobOwnerAccount {
-  def apply[F[_]](store: Store[F]): FindJobOwner[F] =
+  def apply[F[_]: cats.effect.Sync](store: Store[F]): FindJobOwner[F] =
     FindJobOwner.of { job =>
+      val logger = docspell.logging.getLogger[F]
       val accountId = AccountId(job.group, job.submitter)
-      store.transact(QLogin.findAccount(accountId))
+      logger.debug(s"Searching for account of ids: $accountId ") *>
+        store.transact(QLogin.findAccount(accountId))
     }
 }

modules/notification/impl/src/main/scala/docspell/notification/impl/EventNotify.scala

@@ -30,6 +30,7 @@ object EventNotify {
     Kleisli { event =>
       (for {
         hooks <- OptionT.liftF(store.transact(QNotification.findChannelsForEvent(event)))
+        _ <- OptionT.liftF(logger.trace(s"Found hooks: $hooks for event: $event"))
         evctx <- DbEventContext.apply.run(event).mapK(store.transform)
         channels = hooks
           .filter(hc =>

modules/scheduler/impl/src/main/scala/docspell/scheduler/impl/SchedulerImpl.scala

@@ -136,15 +136,14 @@ final class SchedulerImpl[F[_]: Async](
   def mainLoop: Stream[F, Nothing] = {
     val body: F[Boolean] =
       for {
-        _ <- permits.available.flatMap(a =>
+        _ <- permits.available.flatTap(a =>
           logger.debug(s"Try to acquire permit ($a free)")
         )
         _ <- permits.acquire
         _ <- logger.debug("New permit acquired")
         down <- state.get.map(_.shutdownRequest)
         rjob <-
-          if (down)
-            logger.info("") *> permits.release *> (None: Option[RJob]).pure[F]
+          if (down) permits.release.as(Option.empty[RJob])
           else
             queue.nextJob(
               group => state.modify(_.nextPrio(group, config.countingScheme)),
@@ -220,7 +219,7 @@ final class SchedulerImpl[F[_]: Async](
     }
   }
 
-  def onFinish(job: RJob, result: JobTaskResult, finishState: JobState): F[Unit] =
+  private def onFinish(job: RJob, result: JobTaskResult, finishState: JobState): F[Unit] =
     for {
       _ <- logger.debug(s"Job ${job.info} done $finishState. Releasing resources.")
       _ <- permits.release *> permits.available.flatMap(a =>
@@ -229,13 +228,13 @@ final class SchedulerImpl[F[_]: Async](
       _ <- state.modify(_.removeRunning(job))
       _ <- QJob.setFinalState(job.id, finishState, store)
       _ <- Sync[F].whenA(JobState.isDone(finishState))(
-        pubSub.publish1IgnoreErrors(
+        logger.trace("Publishing JobDone event") *> pubSub.publish1IgnoreErrors(
           JobDone.topic,
           JobDone(job.id, job.group, job.task, job.args, finishState, result.json)
         )
       )
       _ <- Sync[F].whenA(JobState.isDone(finishState))(
-        makeJobDoneEvent(job, result)
+        logger.trace("Sending JobDone to event sink") *> makeJobDoneEvent(job, result)
           .semiflatMap(eventSink.offer)
           .value
       )
@@ -243,7 +242,9 @@ final class SchedulerImpl[F[_]: Async](
 
   private def makeJobDoneEvent(job: RJob, result: JobTaskResult) =
     for {
-      acc <- OptionT(findJobOwner(convertJob(job)))
+      acc <- OptionT(findJobOwner(convertJob(job))).flatTransform(acc =>
+        logger.debug(s"Found job owner $acc for job $job").as(acc)
+      )
       ev = Event.JobDone(
         acc,
         job.id,

modules/store/src/main/scala/docspell/store/queries/QLogin.scala

@@ -61,13 +61,25 @@ object QLogin {
   /** Finds the account given a combination of login/user-id and coll-id/coll-name pair.
     */
   def findAccount(acc: AccountId): ConnectionIO[Option[AccountInfo]] = {
+    // collective may be given as id or name and it is possible to have two collective
+    // ids given. In that (not so nice) case, we need to lookup the collective name and
+    // match it against the user. This kind-of edge case is currently used when a user task
+    // of scope UserTaskScope.collective is run.
+
     val collIdOpt = acc.collective.id.toLongOption.map(CollectiveId(_))
-    findUser0((ut, ct) =>
-      (ut.login === acc.user || ut.uid === acc.user) && collIdOpt
-        .map(id => ct.id === id)
-        .getOrElse(ct.name === acc.collective)
-    )
-      .map(_.map(_.account))
+    collIdOpt match {
+      case Some(cid) if acc.user == acc.collective =>
+        findUser0((ut, ct) => ct.id === cid && ut.login === ct.name)
+          .map(_.map(_.account))
+
+      case _ =>
+        findUser0((ut, ct) =>
+          (ut.login === acc.user || ut.uid === acc.user) && collIdOpt
+            .map(id => ct.id === id)
+            .getOrElse(ct.name === acc.collective)
+        )
+          .map(_.map(_.account))
+    }
   }
 
   def findByRememberMe(