From a181dfe6b8952ea6660db11b45c2b6d29fe57743 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ullrich=20Sch=C3=A4fer?= <hi+github@ullrich.is>
Date: Sat, 23 May 2026 19:08:35 +0200
Subject: [PATCH] Add cron job to sweep stale import batches every minute

Batches stuck in pending or running for more than 10 minutes (server
restart mid-import, pg-boss job dropped) are marked failed with a
user-visible message. Runs every minute via pg-boss cron with a 55s
expiry so overlapping runs are dropped.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 apps/journal/app/jobs/import-batches-sweep.ts | 38 +++++++++++++++++++
 apps/journal/server.ts                        |  3 +-
 2 files changed, 40 insertions(+), 1 deletion(-)
 create mode 100644 apps/journal/app/jobs/import-batches-sweep.ts

diff --git a/apps/journal/app/jobs/import-batches-sweep.ts b/apps/journal/app/jobs/import-batches-sweep.ts
new file mode 100644
index 0000000..c94c63e
--- /dev/null
+++ b/apps/journal/app/jobs/import-batches-sweep.ts
@@ -0,0 +1,38 @@
+import type { JobDefinition } from "@trails-cool/jobs";
+import { and, lt, inArray } from "drizzle-orm";
+import { getDb } from "../lib/db.ts";
+import { importBatches } from "@trails-cool/db/schema/journal";
+import { logger } from "../lib/logger.server.ts";
+
+// Batches stuck longer than this without completing are considered stale.
+const STALE_MS = 10 * 60 * 1000;
+
+export const importBatchesSweepJob: JobDefinition = {
+  name: "import-batches-sweep",
+  cron: "* * * * *",
+  retryLimit: 0,
+  expireInSeconds: 55,
+  async handler() {
+    const db = getDb();
+    const cutoff = new Date(Date.now() - STALE_MS);
+
+    const result = await db
+      .update(importBatches)
+      .set({
+        status: "failed",
+        errorMessage: "Import timed out — the server may have restarted mid-import. Click 'Run again' to retry.",
+        completedAt: new Date(),
+      })
+      .where(
+        and(
+          inArray(importBatches.status, ["pending", "running"]),
+          lt(importBatches.startedAt, cutoff),
+        ),
+      )
+      .returning({ id: importBatches.id });
+
+    if (result.length > 0) {
+      logger.info({ count: result.length }, "import-batches-sweep: marked stale batches as failed");
+    }
+  },
+};
diff --git a/apps/journal/server.ts b/apps/journal/server.ts
index d77a5c9..d4f48ec 100644
--- a/apps/journal/server.ts
+++ b/apps/journal/server.ts
@@ -144,8 +144,9 @@ server.listen(port, async () => {
   const { notificationsFanoutJob } = await import("./app/jobs/notifications-fanout.ts");
   const { notificationsPurgeJob } = await import("./app/jobs/notifications-purge.ts");
   const { komootBulkImportJob } = await import("./app/jobs/komoot-bulk-import.ts");
+  const { importBatchesSweepJob } = await import("./app/jobs/import-batches-sweep.ts");
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any);
+  jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any, importBatchesSweepJob);
 
   const boss = createBoss(process.env.DATABASE_URL ?? "postgres://trails:trails@localhost:5432/trails");
   await startWorker(boss, jobs);