Add runtime safeguards: OCR concurrency limiter, upload size cap, and batched Monday sync

- Processing queue limits OCR to 2 concurrent jobs across all sources (upload, email, folder watcher)
- 50 MB file size cap enforced on uploads, email attachments, and folder watcher
- Monday.com bulk sync now processes in batches of 5 with 1s delay between batches

Made-with: Cursor

src/app/api/integrations/monday/sync-all/route.ts

@@ -31,15 +31,29 @@ export async function POST() {
     let failed = 0;
     const errors: string[] = [];
 
-    for (const card of cards) {
-      try {
-        await pushCardToMonday(card.id, settings);
+    const BATCH_SIZE = 5;
+    const BATCH_DELAY_MS = 1000;
+
+    for (let i = 0; i < cards.length; i += BATCH_SIZE) {
+      const batch = cards.slice(i, i + BATCH_SIZE);
+      const results = await Promise.allSettled(
+        batch.map((card) => pushCardToMonday(card.id, settings))
+      );
+
+      for (let j = 0; j < results.length; j++) {
+        const result = results[j];
+        if (result.status === "fulfilled") {
           synced++;
-      } catch (err) {
+        } else {
           failed++;
-        const msg = err instanceof Error ? err.message : "Unknown error";
-        errors.push(`${card.name || card.id}: ${msg}`);
-        console.error(`[sync-all] Failed to push card ${card.id}:`, err);
+          const msg = result.reason instanceof Error ? result.reason.message : "Unknown error";
+          errors.push(`${batch[j].name || batch[j].id}: ${msg}`);
+          console.error(`[sync-all] Failed to push card ${batch[j].id}:`, result.reason);
+        }
+      }
+
+      if (i + BATCH_SIZE < cards.length) {
+        await new Promise((r) => setTimeout(r, BATCH_DELAY_MS));
       }
     }
 

src/app/api/upload/route.ts

@@ -1,6 +1,8 @@
 import { NextRequest, NextResponse } from "next/server";
 import { prisma } from "@/lib/db";
-import { processFile } from "@/lib/ocr";
+import { enqueueProcessing } from "@/lib/processing-queue";
+
+const MAX_UPLOAD_SIZE = 50 * 1024 * 1024; // 50 MB
 
 const ALLOWED_TYPES = [
   "application/pdf",
@@ -38,6 +40,13 @@ export async function POST(request: NextRequest) {
         );
       }
 
+      if (file.size > MAX_UPLOAD_SIZE) {
+        return NextResponse.json(
+          { error: `File "${file.name}" exceeds the 50 MB limit` },
+          { status: 400 }
+        );
+      }
+
       const buffer = Buffer.from(await file.arrayBuffer());
       const fileName = file.name || `upload-${Date.now()}`;
       const isPdf = contentType === "application/pdf";
@@ -52,7 +61,7 @@ export async function POST(request: NextRequest) {
 
       jobIds.push(job.id);
 
-      processFile(job.id, fileName, buffer, isPdf).catch((err) => {
+      enqueueProcessing(job.id, fileName, buffer, isPdf).catch((err) => {
         console.error(`[upload] Background processing failed for job ${job.id}:`, err);
       });
     }

src/lib/email-watcher.ts

@@ -1,7 +1,7 @@
 import { ImapFlow } from "imapflow";
 import { simpleParser } from "mailparser";
 import { prisma } from "./db";
-import { processFile } from "./ocr";
+import { enqueueProcessing } from "./processing-queue";
 
 const ALLOWED_CONTENT_TYPES = [
   "application/pdf",
@@ -11,6 +11,8 @@ const ALLOWED_CONTENT_TYPES = [
   "image/webp",
 ];
 
+const MAX_ATTACHMENT_SIZE = 50 * 1024 * 1024; // 50 MB
+
 type EmailConfig = {
   host: string;
   port: number;
@@ -60,6 +62,12 @@ async function handleMessage(client: ImapFlow, uid: number, config: EmailConfig)
   for (const att of attachments) {
     const fileName = att.filename || `email-attachment-${Date.now()}`;
     const buffer = att.content;
+
+    if (buffer.length > MAX_ATTACHMENT_SIZE) {
+      log(`Skipping attachment "${fileName}" — exceeds 50 MB size limit (${(buffer.length / 1024 / 1024).toFixed(1)} MB)`);
+      continue;
+    }
+
     const isPdf = att.contentType?.toLowerCase() === "application/pdf";
 
     try {
@@ -71,7 +79,7 @@ async function handleMessage(client: ImapFlow, uid: number, config: EmailConfig)
         },
       });
 
-      processFile(job.id, fileName, buffer, isPdf).catch((err) => {
+      enqueueProcessing(job.id, fileName, buffer, isPdf).catch((err) => {
         logError(`Processing failed for ${fileName}:`, err);
       });
 

src/lib/processing-queue.ts

@@ -0,0 +1,44 @@
+import { processFile } from "./ocr";
+
+const MAX_CONCURRENT = 2;
+const queue: QueueItem[] = [];
+let running = 0;
+
+type QueueItem = {
+  jobId: string;
+  fileName: string;
+  fileBuffer: Buffer;
+  isPdf: boolean;
+  resolve: (ids: string[]) => void;
+  reject: (err: unknown) => void;
+};
+
+function drain() {
+  while (running < MAX_CONCURRENT && queue.length > 0) {
+    const item = queue.shift()!;
+    running++;
+    processFile(item.jobId, item.fileName, item.fileBuffer, item.isPdf)
+      .then(item.resolve)
+      .catch(item.reject)
+      .finally(() => {
+        running--;
+        drain();
+      });
+  }
+}
+
+export function enqueueProcessing(
+  jobId: string,
+  fileName: string,
+  fileBuffer: Buffer,
+  isPdf: boolean
+): Promise<string[]> {
+  return new Promise((resolve, reject) => {
+    queue.push({ jobId, fileName, fileBuffer, isPdf, resolve, reject });
+    drain();
+  });
+}
+
+export function getQueueStatus() {
+  return { running, queued: queue.length, maxConcurrent: MAX_CONCURRENT };
+}

src/lib/watcher.ts

@@ -2,11 +2,12 @@ import chokidar, { type FSWatcher } from "chokidar";
 import fs from "fs/promises";
 import path from "path";
 import { prisma } from "./db";
-import { processFile } from "./ocr";
+import { enqueueProcessing } from "./processing-queue";
 
 let watcher: FSWatcher | null = null;
 
 const processedFiles = new Set<string>();
+const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50 MB
 
 export async function startWatching(watchDir: string): Promise<void> {
   if (watcher) {
@@ -40,6 +41,12 @@ export async function startWatching(watchDir: string): Promise<void> {
     const isPdf = ext === ".pdf";
 
     try {
+      const stat = await fs.stat(filePath);
+      if (stat.size > MAX_FILE_SIZE) {
+        console.log(`[watcher] Skipping ${fileName} — exceeds 50 MB size limit (${(stat.size / 1024 / 1024).toFixed(1)} MB)`);
+        return;
+      }
+
       const buffer = await fs.readFile(filePath);
       const job = await prisma.processingJob.create({
         data: {
@@ -49,7 +56,7 @@ export async function startWatching(watchDir: string): Promise<void> {
         },
       });
 
-      processFile(job.id, fileName, buffer, isPdf).catch((err) => {
+      enqueueProcessing(job.id, fileName, buffer, isPdf).catch((err) => {
         console.error(`[watcher] Processing failed for ${fileName}:`, err);
       });