Integrate learn-classifier task into the app

modules/backend/src/main/scala/docspell/backend/BackendApp.scala

@@ -52,12 +52,12 @@ object BackendApp {
       queue          <- JobQueue(store)
       loginImpl      <- Login[F](store)
       signupImpl     <- OSignup[F](store)
-      collImpl       <- OCollective[F](store)
+      joexImpl       <- OJoex(JoexClient(httpClient), store)
+      collImpl       <- OCollective[F](store, utStore, joexImpl)
       sourceImpl     <- OSource[F](store)
       tagImpl        <- OTag[F](store)
       equipImpl      <- OEquipment[F](store)
       orgImpl        <- OOrganization(store)
-      joexImpl       <- OJoex(JoexClient(httpClient), store)
       uploadImpl     <- OUpload(store, queue, cfg.files, joexImpl)
       nodeImpl       <- ONode(store)
       jobImpl        <- OJob(store, joexImpl)

modules/backend/src/main/scala/docspell/backend/ops/OCollective.scala

@@ -9,8 +9,12 @@ import docspell.backend.ops.OCollective._
 import docspell.common._
 import docspell.store.queries.QCollective
 import docspell.store.records._
+import docspell.store.usertask.UserTask
+import docspell.store.usertask.UserTaskStore
 import docspell.store.{AddResult, Store}
 
+import com.github.eikek.calev.CalEvent
+
 trait OCollective[F[_]] {
 
   def find(name: Ident): F[Option[RCollective]]
@@ -95,7 +99,11 @@ object OCollective {
     }
   }
 
-  def apply[F[_]: Effect](store: Store[F]): Resource[F, OCollective[F]] =
+  def apply[F[_]: Effect](
+      store: Store[F],
+      uts: UserTaskStore[F],
+      joex: OJoex[F]
+  ): Resource[F, OCollective[F]] =
     Resource.pure[F, OCollective[F]](new OCollective[F] {
       def find(name: Ident): F[Option[RCollective]] =
         store.transact(RCollective.findById(name))
@@ -105,6 +113,23 @@ object OCollective {
           .transact(RCollective.updateSettings(collective, sett))
           .attempt
           .map(AddResult.fromUpdate)
+          .flatMap(res => updateLearnClassifierTask(collective, sett) *> res.pure[F])
+
+      def updateLearnClassifierTask(coll: Ident, sett: Settings) =
+        for {
+          id <- Ident.randomId[F]
+          on    = sett.classifier.map(_.enabled).getOrElse(false)
+          timer = sett.classifier.map(_.schedule).getOrElse(CalEvent.unsafe(""))
+          ut = UserTask(
+            id,
+            LearnClassifierArgs.taskName,
+            on,
+            timer,
+            LearnClassifierArgs(coll)
+          )
+          _ <- uts.updateOneTask(AccountId(coll, LearnClassifierArgs.taskName), ut)
+          _ <- joex.notifyAllNodes
+        } yield ()
 
       def findSettings(collective: Ident): F[Option[OCollective.Settings]] =
         store.transact(RCollective.getSettings(collective))

modules/joex/src/main/scala/docspell/joex/JoexAppImpl.scala

@@ -14,6 +14,7 @@ import docspell.ftssolr.SolrFtsClient
 import docspell.joex.analysis.RegexNerFile
 import docspell.joex.fts.{MigrationTask, ReIndexTask}
 import docspell.joex.hk._
+import docspell.joex.learn.LearnClassifierTask
 import docspell.joex.notify._
 import docspell.joex.pdfconv.ConvertAllPdfTask
 import docspell.joex.pdfconv.PdfConvTask
@@ -159,6 +160,13 @@ object JoexAppImpl {
             ConvertAllPdfTask.onCancel[F]
           )
         )
+        .withTask(
+          JobTask.json(
+            LearnClassifierArgs.taskName,
+            LearnClassifierTask[F](cfg.textAnalysis, blocker, analyser),
+            LearnClassifierTask.onCancel[F]
+          )
+        )
         .resource
       psch <- PeriodicScheduler.create(
         cfg.periodicScheduler,

modules/joex/src/main/scala/docspell/joex/learn/LearnClassifierTask.scala

@@ -12,11 +12,15 @@ import docspell.backend.ops.OCollective
 import docspell.common._
 import docspell.joex.Config
 import docspell.joex.scheduler._
+import docspell.store.records.RClassifierSetting
 
 object LearnClassifierTask {
 
   type Args = LearnClassifierArgs
 
+  def onCancel[F[_]: Sync]: Task[F, Args, Unit] =
+    Task.log(_.warn("Cancelling learn-classifier task"))
+
   def apply[F[_]: Sync: ContextShift](
       cfg: Config.TextAnalysis,
       blocker: Blocker,
@@ -24,7 +28,7 @@ object LearnClassifierTask {
   ): Task[F, Args, Unit] =
     Task { ctx =>
       (for {
-        sett <- findActiveSettings[F](ctx.args.collective, cfg)
+        sett <- findActiveSettings[F](ctx, cfg)
         data = selectItems(
           ctx,
           math.min(cfg.classification.itemCount, sett.itemCount),
@@ -52,10 +56,16 @@ object LearnClassifierTask {
     ???
 
   private def findActiveSettings[F[_]: Sync](
-      coll: Ident,
+      ctx: Context[F, Args],
       cfg: Config.TextAnalysis
   ): OptionT[F, OCollective.Classifier] =
-    ???
+    if (cfg.classification.enabled)
+      OptionT(ctx.store.transact(RClassifierSetting.findById(ctx.args.collective)))
+        .filter(_.enabled)
+        .filter(_.category.nonEmpty)
+        .map(OCollective.Classifier.fromRecord)
+    else
+      OptionT.none
 
   private def logInactiveWarning[F[_]: Sync](logger: Logger[F]): F[Unit] =
     logger.warn(

modules/store/src/main/scala/docspell/store/records/RClassifierSetting.scala

@@ -102,5 +102,9 @@ object RClassifierSetting {
         created
       )
   }
+  object Classifier {
+    def fromRecord(r: RClassifierSetting): Classifier =
+      Classifier(r.enabled, r.schedule, r.itemCount, r.category.some)
+  }
 
 }