Adopt backend to collective-id

modules/scheduler/api/src/main/scala/docspell/scheduler/FindJobOwner.scala

@@ -1,6 +1,7 @@
 package docspell.scheduler
 
 import cats.Applicative
+import cats.data.{Kleisli, OptionT}
 import docspell.common.AccountInfo
 
 /** Strategy to find the user that submitted the job. This is used to emit events about
@@ -8,12 +9,18 @@ import docspell.common.AccountInfo
   *
   * If an account cannot be determined, no events can be send.
   */
-trait FindJobOwner[F[_]] {
+trait FindJobOwner[F[_]] { self =>
   def apply(job: Job[_]): F[Option[AccountInfo]]
+
+  final def kleisli: Kleisli[OptionT[F, *], Job[_], AccountInfo] =
+    Kleisli(job => OptionT(self(job)))
 }
 
 object FindJobOwner {
 
   def none[F[_]: Applicative]: FindJobOwner[F] =
     (_: Job[_]) => Applicative[F].pure(None)
+
+  def of[F[_]](f: Job[_] => F[Option[AccountInfo]]): FindJobOwner[F] =
+    (job: Job[_]) => f(job)
 }

modules/scheduler/api/src/main/scala/docspell/scheduler/Job.scala

@@ -8,9 +8,8 @@ package docspell.scheduler
 
 import cats.effect.Sync
 import cats.syntax.functor._
-
 import docspell.common._
-
+import docspell.scheduler.usertask.UserTaskScope
 import io.circe.Encoder
 
 final case class Job[A](
@@ -31,14 +30,14 @@ final case class Job[A](
 object Job {
   def createNew[F[_]: Sync, A](
       task: Ident,
-      group: Ident,
+      submitter: UserTaskScope,
       args: A,
       subject: String,
-      submitter: Ident,
       priority: Priority,
       tracker: Option[Ident]
   ): F[Job[A]] =
     Ident.randomId[F].map { id =>
-      Job(id, task, group, args, subject, submitter, priority, tracker)
+      val accId = submitter.toAccountId
+      Job(id, task, accId.collective, args, subject, accId.user, priority, tracker)
     }
 }

modules/scheduler/api/src/main/scala/docspell/scheduler/usertask/UserTaskScope.scala

@@ -7,52 +7,69 @@
 package docspell.scheduler.usertask
 
 import docspell.common._
+import docspell.scheduler.usertask.UserTaskScope._
 
 sealed trait UserTaskScope { self: Product =>
 
   def name: String =
     productPrefix.toLowerCase
 
-  def collective: Ident
+  def collectiveId: Option[CollectiveId]
 
-  def fold[A](fa: AccountId => A, fb: Ident => A): A
+  def fold[A](fa: Account => A, fb: CollectiveId => A, fc: => A): A
 
   /** Maps to the account or uses the collective for both parts if the scope is collective
     * wide.
     */
-  private[scheduler] def toAccountId: AccountId =
-    AccountId(collective, fold(_.user, identity))
+  protected[scheduler] def toAccountId: AccountId
 }
 
 object UserTaskScope {
 
-  final case class Account(account: AccountId) extends UserTaskScope {
-    val collective = account.collective
+  final case class Account(collective: CollectiveId, userId: Ident)
+      extends UserTaskScope {
+    val collectiveId = Some(collective)
 
-    def fold[A](fa: AccountId => A, fb: Ident => A): A =
-      fa(account)
+    def fold[A](fa: Account => A, fb: CollectiveId => A, fc: => A): A =
+      fa(this)
+
+    protected[scheduler] val toAccountId: AccountId =
+      AccountId(collective.valueAsIdent, userId)
   }
 
-  final case class Collective(collective: Ident) extends UserTaskScope {
-    def fold[A](fa: AccountId => A, fb: Ident => A): A =
+  final case class Collective(collective: CollectiveId) extends UserTaskScope {
+    val collectiveId = Some(collective)
+    def fold[A](fa: Account => A, fb: CollectiveId => A, fc: => A): A =
       fb(collective)
+
+    protected[scheduler] val toAccountId: AccountId = {
+      val c = collective.valueAsIdent
+      AccountId(c, c)
+    }
   }
 
-  def collective(id: Ident): UserTaskScope =
+  case object System extends UserTaskScope {
+    val collectiveId = None
+
+    def fold[A](fa: Account => A, fb: CollectiveId => A, fc: => A): A =
+      fc
+
+    protected[scheduler] val toAccountId: AccountId =
+      DocspellSystem.account
+  }
+
+  def collective(id: CollectiveId): UserTaskScope =
     Collective(id)
 
-  def account(accountId: AccountId): UserTaskScope =
-    Account(accountId)
+  def account(collectiveId: CollectiveId, userId: Ident): UserTaskScope =
+    Account(collectiveId, userId)
 
-  def apply(accountId: AccountId): UserTaskScope =
-    UserTaskScope.account(accountId)
+  def apply(collectiveId: CollectiveId, userId: Option[Ident]): UserTaskScope =
+    userId.map(Account(collectiveId, _)).getOrElse(collective(collectiveId))
 
-  def apply(collective: Ident): UserTaskScope =
-    UserTaskScope.collective(collective)
-
-  def apply(collective: Ident, login: Option[Ident]): UserTaskScope =
-    login.map(AccountId(collective, _)).map(account).getOrElse(apply(collective))
+  def apply(info: AccountInfo): UserTaskScope =
+    account(info.collectiveId, info.userId)
 
   def system: UserTaskScope =
-    collective(DocspellSystem.taskGroup)
+    UserTaskScope.System
 }

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

@@ -39,14 +39,14 @@ object QUserTask {
     ): F[RPeriodicTask] =
       QUserTask
         .create[F](
-          ut.enabled,
-          scope,
-          ut.name,
-          ut.args,
-          subject.getOrElse(s"${scope.fold(_.user.id, _.id)}: ${ut.name.id}"),
-          Priority.Low,
-          ut.timer,
-          ut.summary
+          enabled = ut.enabled,
+          scope = scope,
+          task = ut.name,
+          args = ut.args,
+          subject = subject.getOrElse(s"${scope.toAccountId.asString}: ${ut.name.id}"),
+          priority = Priority.Low,
+          timer = ut.timer,
+          summary = ut.summary
         )
         .map(r => r.copy(id = ut.id))
   }
@@ -151,24 +151,24 @@ object QUserTask {
           .current[F]
           .map { now =>
             RPeriodicTask(
-              id,
-              enabled,
-              task,
-              scope.collective,
-              args,
-              subject,
-              scope.fold(_.user, identity),
-              priority,
-              None,
-              None,
-              timer,
-              timer
+              id = id,
+              enabled = enabled,
+              task = task,
+              group = scope.toAccountId.collective,
+              args = args,
+              subject = subject,
+              submitter = scope.toAccountId.user,
+              priority = priority,
+              worker = None,
+              marked = None,
+              timer = timer,
+              nextrun = timer
                 .nextElapse(now.atZone(Timestamp.UTC))
                 .map(_.toInstant)
                 .map(Timestamp.apply)
                 .getOrElse(Timestamp.Epoch),
-              now,
-              summary
+              created = now,
+              summary = summary
             )
           }
       )

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

@@ -39,6 +39,7 @@ case class SchedulerModuleBuilder[F[_]: Async] private (
       schedulerR <- schedulerBuilder
         .withPubSub(jobStoreModule.pubSubT)
         .withEventSink(jobStoreModule.eventSink)
+        .withFindJobOwner(jobStoreModule.findJobOwner)
         .withQueue(queue)
         .resource