From 034c73128f00d42e12a5226948929d22c8bca072 Mon Sep 17 00:00:00 2001 From: Randall Stillwell Date: Tue, 7 Apr 2026 13:44:04 -0500 Subject: [PATCH] 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 --- .../api/integrations/monday/sync-all/route.ts | 32 ++++++++++---- src/app/api/upload/route.ts | 13 +++++- src/lib/email-watcher.ts | 12 ++++- src/lib/processing-queue.ts | 44 +++++++++++++++++++ src/lib/watcher.ts | 11 ++++- 5 files changed, 97 insertions(+), 15 deletions(-) create mode 100644 src/lib/processing-queue.ts diff --git a/src/app/api/integrations/monday/sync-all/route.ts b/src/app/api/integrations/monday/sync-all/route.ts index d0c7fb3..f19cc78 100644 --- a/src/app/api/integrations/monday/sync-all/route.ts +++ b/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); - synced++; - } catch (err) { - 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 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++; + } else { + failed++; + 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)); } } diff --git a/src/app/api/upload/route.ts b/src/app/api/upload/route.ts index 18265be..eb5da0e 100644 --- a/src/app/api/upload/route.ts +++ b/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); }); } diff --git a/src/lib/email-watcher.ts b/src/lib/email-watcher.ts index cfd7489..0a6910c 100644 --- a/src/lib/email-watcher.ts +++ b/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); }); diff --git a/src/lib/processing-queue.ts b/src/lib/processing-queue.ts new file mode 100644 index 0000000..7a48eb1 --- /dev/null +++ b/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 { + 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 }; +} diff --git a/src/lib/watcher.ts b/src/lib/watcher.ts index 46e9516..eb20bc6 100644 --- a/src/lib/watcher.ts +++ b/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(); +const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50 MB export async function startWatching(watchDir: string): Promise { if (watcher) { @@ -40,6 +41,12 @@ export async function startWatching(watchDir: string): Promise { 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 { }, }); - processFile(job.id, fileName, buffer, isPdf).catch((err) => { + enqueueProcessing(job.id, fileName, buffer, isPdf).catch((err) => { console.error(`[watcher] Processing failed for ${fileName}:`, err); });