Merge pull request #402 from trails-cool/stigi/import-batches-sweep

Add cron sweep for stale import batches

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");
+    }
+  },
+};

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);