Cleanup nodes that are not reachable anymore

modules/joex/src/main/resources/reference.conf

@@ -170,6 +170,14 @@ docspell.joex {
       # whether more or less memory should be used.
       delete-batch = "100"
     }
+
+    # Removes node entries that are not reachable anymore.
+    check-nodes {
+      # Whether this task is enabled
+      enabled = true
+      # How often the node must be unreachable, before it is removed.
+      min-not-found = 2
+    }
   }
 
   # Configuration of text extraction

modules/joex/src/main/scala/docspell/joex/hk/CheckNodesTask.scala

@@ -0,0 +1,67 @@
+package docspell.joex.hk
+
+import cats.effect._
+import cats.implicits._
+
+import docspell.common._
+import docspell.joex.scheduler.{Context, Task}
+import docspell.store.records._
+
+import org.http4s.client.Client
+import org.http4s.client.blaze.BlazeClientBuilder
+
+object CheckNodesTask {
+
+  def apply[F[_]: ConcurrentEffect](
+      cfg: HouseKeepingConfig.CheckNodes
+  ): Task[F, Unit, Unit] =
+    Task { ctx =>
+      if (cfg.enabled)
+        for {
+          _ <- ctx.logger.info("Check nodes reachability")
+          _ <- BlazeClientBuilder[F](ctx.blocker.blockingContext).resource.use { client =>
+            checkNodes(ctx, client)
+          }
+          _ <- ctx.logger.info(
+            s"Remove nodes not found more than ${cfg.minNotFound} times"
+          )
+          n <- removeNodes(ctx, cfg)
+          _ <- ctx.logger.info(s"Removed $n nodes")
+        } yield ()
+      else
+        ctx.logger.info("CheckNodes task is disabled in the configuration")
+    }
+
+  def checkNodes[F[_]: Sync](ctx: Context[F, _], client: Client[F]): F[Unit] =
+    ctx.store
+      .transact(RNode.streamAll)
+      .evalMap(node =>
+        checkNode(ctx.logger, client)(node.url)
+          .flatMap(seen =>
+            if (seen) ctx.store.transact(RNode.resetNotFound(node.id))
+            else ctx.store.transact(RNode.incrementNotFound(node.id))
+          )
+      )
+      .compile
+      .drain
+
+  def checkNode[F[_]: Sync](logger: Logger[F], client: Client[F])(
+      url: LenientUri
+  ): F[Boolean] = {
+    val apiVersion = url / "api" / "info" / "version"
+    for {
+      res <- client.expect[String](apiVersion.asString).attempt
+      _ <- res.fold(
+        ex => logger.info(s"Node ${url.asString} not found: ${ex.getMessage}"),
+        _ => logger.info(s"Node ${url.asString} is reachable")
+      )
+    } yield res.isRight
+  }
+
+  def removeNodes[F[_]: Sync](
+      ctx: Context[F, _],
+      cfg: HouseKeepingConfig.CheckNodes
+  ): F[Int] =
+    ctx.store.transact(RNode.deleteNotFound(cfg.minNotFound))
+
+}

modules/joex/src/main/scala/docspell/joex/hk/HouseKeepingConfig.scala

@@ -9,7 +9,8 @@ case class HouseKeepingConfig(
     schedule: CalEvent,
     cleanupInvites: CleanupInvites,
     cleanupJobs: CleanupJobs,
-    cleanupRememberMe: CleanupRememberMe
+    cleanupRememberMe: CleanupRememberMe,
+    checkNodes: CheckNodes
 )
 
 object HouseKeepingConfig {
@@ -20,4 +21,6 @@ object HouseKeepingConfig {
 
   case class CleanupRememberMe(enabled: Boolean, olderThan: Duration)
 
+  case class CheckNodes(enabled: Boolean, minNotFound: Int)
+
 }

modules/joex/src/main/scala/docspell/joex/hk/HouseKeepingTask.scala

@@ -15,12 +15,13 @@ object HouseKeepingTask {
 
   val taskName: Ident = Ident.unsafe("housekeeping")
 
-  def apply[F[_]: Sync](cfg: Config): Task[F, Unit, Unit] =
+  def apply[F[_]: ConcurrentEffect](cfg: Config): Task[F, Unit, Unit] =
     Task
       .log[F, Unit](_.info(s"Running house-keeping task now"))
       .flatMap(_ => CleanupInvitesTask(cfg.houseKeeping.cleanupInvites))
       .flatMap(_ => CleanupRememberMeTask(cfg.houseKeeping.cleanupRememberMe))
       .flatMap(_ => CleanupJobsTask(cfg.houseKeeping.cleanupJobs))
+      .flatMap(_ => CheckNodesTask(cfg.houseKeeping.checkNodes))
 
   def onCancel[F[_]: Sync]: Task[F, Unit, Unit] =
     Task.log[F, Unit](_.warn("Cancelling house-keeping task"))

modules/store/src/main/resources/db/migration/h2/V1.20.3__node_seen.sql

@@ -0,0 +1,2 @@
+ALTER TABLE "node"
+ADD COLUMN "not_found" int not null default 0;

modules/store/src/main/resources/db/migration/mariadb/V1.20.3__node_seen.sql

@@ -0,0 +1,2 @@
+ALTER TABLE `node`
+ADD COLUMN `not_found` int not null default 0;

modules/store/src/main/resources/db/migration/postgresql/V1.20.3__node_seen.sql

@@ -0,0 +1,2 @@
+ALTER TABLE "node"
+ADD COLUMN "not_found" int not null default 0;

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

@@ -1,8 +1,8 @@
 package docspell.store.records
-
 import cats.data.NonEmptyList
 import cats.effect.Sync
 import cats.implicits._
+import fs2.Stream
 
 import docspell.common._
 import docspell.store.qb.DSL._
@@ -16,13 +16,14 @@ case class RNode(
     nodeType: NodeType,
     url: LenientUri,
     updated: Timestamp,
-    created: Timestamp
+    created: Timestamp,
+    notFound: Int
 ) {}
 
 object RNode {
 
   def apply[F[_]: Sync](id: Ident, nodeType: NodeType, uri: LenientUri): F[RNode] =
-    Timestamp.current[F].map(now => RNode(id, nodeType, uri, now, now))
+    Timestamp.current[F].map(now => RNode(id, nodeType, uri, now, now, 0))
 
   final case class Table(alias: Option[String]) extends TableDef {
     val tableName = "node"
@@ -32,18 +33,20 @@ object RNode {
     val url      = Column[LenientUri]("url", this)
     val updated  = Column[Timestamp]("updated", this)
     val created  = Column[Timestamp]("created", this)
-    val all      = NonEmptyList.of[Column[_]](id, nodeType, url, updated, created)
+    val notFound = Column[Int]("not_found", this)
+    val all      = NonEmptyList.of[Column[_]](id, nodeType, url, updated, created, notFound)
   }
 
   def as(alias: String): Table =
     Table(Some(alias))
+  val T = Table(None)
 
   def insert(v: RNode): ConnectionIO[Int] = {
     val t = Table(None)
     DML.insert(
       t,
       t.all,
-      fr"${v.id},${v.nodeType},${v.url},${v.updated},${v.created}"
+      fr"${v.id},${v.nodeType},${v.url},${v.updated},${v.created},${v.notFound}"
     )
   }
 
@@ -61,6 +64,22 @@ object RNode {
       )
   }
 
+  def incrementNotFound(nid: Ident): ConnectionIO[Int] =
+    Timestamp
+      .current[ConnectionIO]
+      .flatMap(now =>
+        DML
+          .update(T, T.id === nid, DML.set(T.notFound.increment(1), T.updated.setTo(now)))
+      )
+
+  def resetNotFound(id: Ident): ConnectionIO[Int] =
+    Timestamp
+      .current[ConnectionIO]
+      .flatMap(now =>
+        DML
+          .update(T, T.id === id, DML.set(T.notFound.setTo(0), T.updated.setTo(now)))
+      )
+
   def set(v: RNode): ConnectionIO[Int] =
     for {
       n <- update(v)
@@ -81,4 +100,10 @@ object RNode {
     val t = Table(None)
     run(select(t.all), from(t), t.id === nodeId).query[RNode].option
   }
+
+  def streamAll: Stream[ConnectionIO, RNode] =
+    run(select(T.all), from(T)).query[RNode].streamWithChunkSize(50)
+
+  def deleteNotFound(min: Int): ConnectionIO[Int] =
+    DML.delete(T, T.notFound >= min)
 }

nix/module-joex.nix

@@ -58,6 +58,10 @@ let
         enabled = true;
         older-than = "30 days";
       };
+      check-nodes = {
+        enabled = true;
+        min-not-found = 2;
+      };
     };
     extraction = {
       pdf = {
@@ -540,6 +544,24 @@ in {
               default = defaults.house-keeping.cleanup-remember-me;
               description = "Settings for cleaning up remember me tokens.";
             };
+            check-nodes = mkOption {
+              type = types.submodule({
+                options = {
+                  enabled = mkOption {
+                    type = types.bool;
+                    default = defaults.house-keeping.check-nodes.enabled;
+                    description = "Whether this task is enabled.";
+                  };
+                  min-not-found = mkOption {
+                    type = types.int;
+                    default = defaults.house-keeping.check-nodes.min-not-found;
+                    description = "How often the node must be unreachable, before it is removed.";
+                  };
+                };
+              });
+              default = defaults.house-keeping.cleanup-nodes;
+              description = "Removes node entries that are not reachable anymore.";
+            };
           };
         });
         default = defaults.house-keeping;