Merge pull request #557 from captableinc/feat/queue-refactor

feat: supporting both long running process and serverless, and 100% test coverage

Author: dahal

apps/captable/app/api/cron/process-jobs/route.ts

@@ -1,5 +1,5 @@
 import { logger } from "@captable/logger";
-import { processJobs } from "@captable/queue";
+import { processJobsServerless } from "@captable/queue";
 import { type NextRequest, NextResponse } from "next/server";
 import "@/jobs"; // Import to register all jobs
 
@@ -16,45 +16,27 @@ export async function GET(request: NextRequest) {
   }
 
   try {
-    const startTime = Date.now();
-
-    // Process jobs in batches
-    let totalProcessed = 0;
-    let batchCount = 0;
-    const maxBatches = 10; // Prevent infinite loops
-
-    while (batchCount < maxBatches) {
-      const processed = await processJobs(20); // Process 20 jobs per batch
-      totalProcessed += processed;
-      batchCount++;
-
-      if (processed === 0) {
-        break; // No more jobs to process
-      }
-
-      // Small delay between batches
-      await new Promise((resolve) => setTimeout(resolve, 100));
-    }
-
-    const duration = Date.now() - startTime;
+    // Use serverless-optimized processing with built-in timeouts and batch management
+    const result = await processJobsServerless({
+      maxJobs: 200, // Maximum jobs to process in this run
+      maxBatches: 10, // Maximum batches to prevent infinite loops
+      batchSize: 20, // Jobs per batch
+      timeout: 25000, // 25 second timeout (safe for Vercel)
+      batchDelay: 100, // 100ms delay between batches
+    });
 
-    log.info(
-      {
-        totalProcessed,
-        batches: batchCount,
-        duration,
-      },
-      "Cron job processing completed",
-    );
+    log.info(result, "Serverless cron job processing completed");
 
     return NextResponse.json({
       success: true,
-      processed: totalProcessed,
-      batches: batchCount,
-      duration,
+      processed: result.processed,
+      batches: result.batches,
+      duration: result.duration,
+      timeoutReached: result.timeoutReached,
+      errors: result.errors,
     });
   } catch (error) {
-    log.error({ error }, "Cron job processing failed");
+    log.error({ error }, "Serverless cron job processing failed");
     return NextResponse.json(
       { error: "Internal server error" },
       { status: 500 },

bun.lock

@@ -13,6 +13,8 @@
     "db:migrate": "dotenv -- turbo run db:migrate",
     "db:studio": "dotenv -- turbo run db:studio",
     "email:dev": "bun run --cwd packages/email dev --port 3001",
+    "test": "dotenv -- turbo run test:run",
+    "test:watch": "dotenv -- turbo run test",
 
     "// Parallel execution scripts": "",
     "dx": "dotenv -- turbo run dev db:studio email:dev jobs:dev --parallel",

package.json

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Captable
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.