feat(crawler): support real-time page persistence, partial export and prevent duplicate rerun

docker-compose.yml

 version: "3.8"
 
 services:
-  postgres:
-    image: postgres:16-alpine
-    container_name: crawl_data_postgres
-    environment:
-      POSTGRES_USER: ${DB_USER:-postgres}
-      POSTGRES_PASSWORD: ${DB_PASSWORD:-postgres}
-      POSTGRES_DB: ${DB_NAME:-crawl_data_db}
-    ports:
-      - "${DB_PORT:-5432}:5432"
-    volumes:
-      - postgres_data:/var/lib/postgresql/data
+  # postgres:
+  #   image: postgres:16-alpine
+  #   container_name: crawl_data_postgres
+  #   environment:
+  #     POSTGRES_USER: ${DB_USER:-postgres}
+  #     POSTGRES_PASSWORD: ${DB_PASSWORD:-postgres}
+  #     POSTGRES_DB: ${DB_NAME:-crawl_data_db}
+  #   ports:
+  #     - "${DB_PORT:-5432}:5432"
+  #   volumes:
+  #     - postgres_data:/var/lib/postgresql/data
 
   redis:
     image: redis:7-alpine

src/modules/crawl-exports/__tests__/crawl-export.flow.test.ts

@@ -94,7 +94,7 @@ describe("CrawlExportService", () => {
       expect(result).toEqual(exportRecord);
     });
 
-    it("throws 400 when job is not COMPLETED", async () => {
+    it("throws 400 when job is still RUNNING", async () => {
       (prisma.crawlJob.findUnique as jest.Mock).mockResolvedValue(
         makeJob({ status: "RUNNING" }),
       );
@@ -104,6 +104,58 @@ describe("CrawlExportService", () => {
       ).rejects.toMatchObject({ statusCode: 400 });
     });
 
+    it("allows exporting a CANCELED job when successPages > 0", async () => {
+      (prisma.crawlJob.findUnique as jest.Mock).mockResolvedValue(
+        makeJob({ status: "CANCELED", successPages: 5 }),
+      );
+      const exportRecord = makeExport();
+      mockExportService.generate.mockResolvedValue(exportRecord as any);
+
+      const result = await service.createExport(
+        "user-1",
+        "CRAWLER_USER",
+        "job-1",
+        "JSON",
+      );
+
+      expect(mockExportService.generate).toHaveBeenCalledWith(
+        expect.objectContaining({ id: "job-1" }),
+        "JSON",
+      );
+      expect(result).toEqual(exportRecord);
+    });
+
+    it("throws 400 when CANCELED job has 0 successPages", async () => {
+      (prisma.crawlJob.findUnique as jest.Mock).mockResolvedValue(
+        makeJob({ status: "CANCELED", successPages: 0 }),
+      );
+
+      await expect(
+        service.createExport("user-1", "CRAWLER_USER", "job-1", "JSON"),
+      ).rejects.toMatchObject({ statusCode: 400 });
+    });
+
+    it("allows exporting a FAILED job when successPages > 0", async () => {
+      (prisma.crawlJob.findUnique as jest.Mock).mockResolvedValue(
+        makeJob({ status: "FAILED", successPages: 10 }),
+      );
+      const exportRecord = makeExport();
+      mockExportService.generate.mockResolvedValue(exportRecord as any);
+
+      const result = await service.createExport(
+        "user-1",
+        "CRAWLER_USER",
+        "job-1",
+        "CSV",
+      );
+
+      expect(mockExportService.generate).toHaveBeenCalledWith(
+        expect.objectContaining({ id: "job-1" }),
+        "CSV",
+      );
+      expect(result).toEqual(exportRecord);
+    });
+
     it("throws 404 when job does not exist", async () => {
       (prisma.crawlJob.findUnique as jest.Mock).mockResolvedValue(null);
 

src/modules/crawl-exports/crawl-export.service.ts

@@ -61,11 +61,28 @@ export class CrawlExportService {
       );
     }
 
-    if (job.status !== JOB_STATUS.COMPLETED) {
+    const isExportable =
+      job.status === JOB_STATUS.COMPLETED ||
+      (job.status === JOB_STATUS.CANCELED && (job.successPages ?? 0) > 0) ||
+      (job.status === JOB_STATUS.FAILED && (job.successPages ?? 0) > 0);
+
+    if (!isExportable) {
+      if (
+        job.status === JOB_STATUS.RUNNING ||
+        job.status === JOB_STATUS.PENDING ||
+        job.status === JOB_STATUS.QUEUED ||
+        job.status === JOB_STATUS.PROCESSING_EXPORT
+      ) {
+        throw new AppError(
+          "Crawl job is still in progress",
+          400,
+          ERROR_CODE.CRAWL_JOB_NOT_COMPLETED,
+        );
+      }
       throw new AppError(
-        "Crawl job is not completed yet",
+        "No successfully crawled pages available to export",
         400,
-        ERROR_CODE.CRAWL_JOB_NOT_COMPLETED,
+        ERROR_CODE.VALIDATION_ERROR,
       );
     }
 

src/modules/crawl-jobs/__tests__/crawl-job.events.test.ts

@@ -14,7 +14,7 @@ describe("CrawlJobController - SSE Events", () => {
       findById: jest.fn(),
     } as any;
 
-    (CrawlJobService as jest.Mock).mockReturnValue(mockService);
+    (CrawlJobService as unknown as jest.Mock).mockReturnValue(mockService);
     controller = new CrawlJobController();
   });
 

src/modules/crawl-jobs/crawl-job.repository.ts

@@ -389,4 +389,19 @@ export class CrawlJobRepository {
     ]);
     return { items, total, page: safePage, limit: safeLimit };
   }
+
+  findRecentActiveJob(userId: string, startUrl: string, windowMs = 5000) {
+    const since = new Date(Date.now() - windowMs);
+    return prisma.crawlJob.findFirst({
+      where: {
+        userId,
+        startUrl,
+        status: {
+          in: [JOB_STATUS.PENDING, JOB_STATUS.QUEUED, JOB_STATUS.RUNNING],
+        },
+        createdAt: { gte: since },
+      },
+      orderBy: { createdAt: "desc" },
+    });
+  }
 }

src/modules/crawl-jobs/crawl-job.service.ts

@@ -165,7 +165,7 @@ export class CrawlJobService {
       );
     }
 
-    await crawlQueue.add("crawl-job", { jobId: job.id });
+    await crawlQueue.add("crawl-job", { jobId: job.id }, { jobId: job.id });
 
     return job;
   }
@@ -216,6 +216,25 @@ export class CrawlJobService {
       JOB_STATUS.CANCELED,
     );
 
+    // Remove from BullMQ queue if still waiting/delayed
+    if (crawlQueue) {
+      try {
+        const bullJob = await crawlQueue.getJob(jobId);
+        if (bullJob) {
+          await bullJob.remove();
+        } else {
+          const waitingJobs = await crawlQueue.getJobs(["waiting", "delayed", "prioritized"]);
+          for (const wj of waitingJobs) {
+            if (wj.data?.jobId === jobId) {
+              await wj.remove();
+            }
+          }
+        }
+      } catch {
+        // Ignored
+      }
+    }
+
     // For CRAWL mode: also cancel at the Firecrawl provider level to stop
     // quota consumption. firecrawlJobId is saved by the worker as soon as
     // asyncCrawlUrl() returns, so it may be null if the job was canceled
@@ -288,20 +307,68 @@ export class CrawlJobService {
       await storage.deleteFile(job.diffReportPath).catch(() => {});
     }
 
+    if (crawlQueue) {
+      try {
+        const bullJob = await crawlQueue.getJob(jobId);
+        if (bullJob) {
+          await bullJob.remove();
+        } else {
+          const waitingJobs = await crawlQueue.getJobs(["waiting", "delayed", "prioritized"]);
+          for (const wj of waitingJobs) {
+            if (wj.data?.jobId === jobId) {
+              await wj.remove();
+            }
+          }
+        }
+      } catch {
+        // Ignored
+      }
+    }
+
     await this.repository.delete(jobId);
     return { success: true, message: "Crawl job deleted successfully" };
   }
 
+  private static readonly rerunLocks = new Set<string>();
+
   async rerun(userId: string, role: string, jobId: string) {
     const existing = await this.findById(userId, role, jobId);
 
-    return this.create(userId, {
-      startUrl: existing.startUrl,
-      mode: existing.mode,
-      maxPages: existing.maxPages,
-      maxDepth: existing.maxDepth,
-      urls: existing.urls,
-    });
+    const lockKey = `${userId}:${jobId}`;
+    if (CrawlJobService.rerunLocks.has(lockKey)) {
+      if (this.repository.findRecentActiveJob) {
+        const recent = await this.repository.findRecentActiveJob(
+          userId,
+          existing.startUrl,
+          10000,
+        );
+        if (recent) return recent;
+      }
+    }
+
+    if (this.repository.findRecentActiveJob) {
+      const recent = await this.repository.findRecentActiveJob(
+        userId,
+        existing.startUrl,
+        5000,
+      );
+      if (recent) {
+        return recent;
+      }
+    }
+
+    CrawlJobService.rerunLocks.add(lockKey);
+    try {
+      return await this.create(userId, {
+        startUrl: existing.startUrl,
+        mode: existing.mode,
+        maxPages: existing.maxPages,
+        maxDepth: existing.maxDepth,
+        urls: existing.urls,
+      });
+    } finally {
+      setTimeout(() => CrawlJobService.rerunLocks.delete(lockKey), 3000);
+    }
   }
 
   async getLogs(

src/modules/firecrawl/firecrawl.service.ts

@@ -111,7 +111,11 @@ export class FirecrawlService {
     url: string,
     maxPages: number,
     maxDepth: number,
-    onProgress?: (completed: number, total: number) => void | Promise<void>,
+    onProgress?: (
+      completed: number,
+      total: number,
+      currentPages?: FirecrawlPageResult[],
+    ) => void | Promise<void>,
     shouldCancel?: () => Promise<boolean>,
   ): Promise<CrawlStatusResult> {
     const client = getFirecrawlClient();
@@ -171,7 +175,16 @@ export class FirecrawlService {
         };
       }
 
-      await onProgress?.(status.completed, status.total);
+      let currentPages: FirecrawlPageResult[] | undefined;
+      if (status.data && status.data.length > 0) {
+        currentPages = status.data.map((doc) => normalizePage(doc, url));
+      }
+
+      if (currentPages) {
+        await onProgress?.(status.completed, status.total, currentPages);
+      } else {
+        await onProgress?.(status.completed, status.total);
+      }
 
       if (
         status.status === "completed" ||
@@ -292,7 +305,11 @@ export class FirecrawlService {
   async batchScrapePages(
     urls: string[],
     maxPages: number,
-    onProgress?: (completed: number, total: number) => void | Promise<void>,
+    onProgress?: (
+      completed: number,
+      total: number,
+      currentPages?: FirecrawlPageResult[],
+    ) => void | Promise<void>,
     shouldCancel?: () => Promise<boolean>,
   ): Promise<CrawlStatusResult> {
     const client = getFirecrawlClient();
@@ -361,7 +378,18 @@ export class FirecrawlService {
         };
       }
 
-      await onProgress?.(status.completed, status.total);
+      let currentPages: FirecrawlPageResult[] | undefined;
+      if (status.data && status.data.length > 0) {
+        currentPages = status.data.map((doc) =>
+          normalizePage(doc as FirecrawlDocument, urls[0]),
+        );
+      }
+
+      if (currentPages) {
+        await onProgress?.(status.completed, status.total, currentPages);
+      } else {
+        await onProgress?.(status.completed, status.total);
+      }
 
       if (
         status.status === "completed" ||

src/queues/__tests__/crawl.worker.test.ts

@@ -488,3 +488,95 @@ describe("processCrawlJob — URL_LIST mode", () => {
     expect(mockPageRepo.upsert).toHaveBeenCalledTimes(1);
   });
 });
+
+// ── Real-time persistence & Cancel preservation ────────────────────────────
+
+describe("processCrawlJob — Real-time persistence & Cancel preservation", () => {
+  it("persists pages incrementally during live crawl onProgress calls", async () => {
+    mockJobRepo.findById = jest.fn().mockResolvedValue(makeJob({ mode: "CRAWL" }));
+    mockFirecrawl.crawlSite = jest
+      .fn()
+      .mockImplementation(async (_url, _maxPages, _maxDepth, onProgress) => {
+        // First progress event with 1 page
+        await onProgress?.(1, 2, [
+          {
+            url: "https://example.com/p1",
+            success: true,
+            markdown: "# P1",
+            metadata: { statusCode: 200 },
+          },
+        ]);
+        return {
+          success: true,
+          status: "completed",
+          pages: [
+            {
+              url: "https://example.com/p1",
+              success: true,
+              markdown: "# P1",
+              metadata: { statusCode: 200 },
+            },
+            {
+              url: "https://example.com/p2",
+              success: true,
+              markdown: "# P2",
+              metadata: { statusCode: 200 },
+            },
+          ],
+          failedUrls: [],
+          robotsBlockedUrls: [],
+          total: 2,
+        };
+      });
+
+    await processCrawlJob(makeBullJob("job-1"));
+
+    // Upsert called twice for 2 pages
+    expect(mockPageRepo.upsert).toHaveBeenCalledTimes(2);
+    expect(mockJobRepo.updateStatus).toHaveBeenCalledWith(
+      "job-1",
+      "COMPLETED",
+      expect.objectContaining({
+        successPages: 2,
+        totalPages: 2,
+      }),
+    );
+  });
+
+  it("preserves scraped pages and maintains CANCELED status when cancelled during crawl", async () => {
+    mockJobRepo.findById = jest.fn().mockResolvedValue(makeJob({ mode: "CRAWL" }));
+    mockFirecrawl.crawlSite = jest
+      .fn()
+      .mockImplementation(async (_url, _maxPages, _maxDepth, onProgress) => {
+        await onProgress?.(1, 10, [
+          {
+            url: "https://example.com/p1",
+            success: true,
+            markdown: "# P1",
+            metadata: { statusCode: 200 },
+          },
+        ]);
+        return {
+          success: false,
+          status: "cancelled",
+          error: "Cancelled locally",
+          pages: [],
+          total: 10,
+        };
+      });
+
+    await processCrawlJob(makeBullJob("job-1"));
+
+    // Page p1 was persisted during onProgress
+    expect(mockPageRepo.upsert).toHaveBeenCalledTimes(1);
+    // Preserved CANCELED status instead of overwriting to FAILED
+    expect(mockJobRepo.updateStatus).toHaveBeenCalledWith(
+      "job-1",
+      "CANCELED",
+      expect.objectContaining({
+        successPages: 1,
+      }),
+    );
+  });
+});
+

src/queues/crawl.worker.processor.ts

@@ -127,10 +127,69 @@ export async function scanAndFlagPage(
  *
  * Updates crawl job progress counters every 10 pages saved.
  */
+/**
+ * Persists a single scraped page (upsert, assets, sensitive scan, extraction template).
+ */
+export async function persistSinglePage(
+  jobId: string,
+  item: FirecrawlPageResult,
+  userId?: string,
+): Promise<{ success: boolean; saved: boolean }> {
+  try {
+    const normalized = getPageProcessor().normalize(item, jobId);
+    const page = await getPageRepository().upsert(normalized);
+    await savePageAssets(jobId, page.id, item);
+    await scanAndFlagPage(
+      page.id,
+      normalized.markdownContent,
+      normalized.title,
+      normalized.description,
+    );
+    await runExtractionIfTemplate(jobId, page.id, item.url, item, userId);
+    return { success: item.success, saved: true };
+  } catch (err: unknown) {
+    console.error(
+      `[Worker] Failed to save page ${item.url}: ${getErrorMessage(err)}`,
+    );
+    return { success: false, saved: false };
+  }
+}
+
+/**
+ * Persists incremental scraped pages, skipping already persisted URLs.
+ */
+export async function persistIncrementalPages(
+  jobId: string,
+  pages: FirecrawlPageResult[],
+  persistedUrls: Set<string>,
+  userId?: string,
+): Promise<{ newSuccess: number; newFailed: number; newErrors: number }> {
+  let newSuccess = 0;
+  let newFailed = 0;
+  let newErrors = 0;
+
+  for (const item of pages) {
+    if (persistedUrls.has(item.url)) continue;
+    persistedUrls.add(item.url);
+
+    const res = await persistSinglePage(jobId, item, userId);
+    if (!res.saved) {
+      newErrors++;
+    } else if (res.success) {
+      newSuccess++;
+    } else {
+      newFailed++;
+    }
+  }
+
+  return { newSuccess, newFailed, newErrors };
+}
+
 export async function persistBatchResults(
   jobId: string,
   result: CrawlStatusResult,
   userId?: string,
+  persistedUrls = new Set<string>(),
 ): Promise<{
   successCount: number;
   failedCount: number;
@@ -140,30 +199,21 @@ export async function persistBatchResults(
   let successCount = 0;
   let failedCount = 0;
   let saveErrors = 0;
-  const seenUrls = new Set<string>();
 
   for (const item of result.pages) {
-    if (seenUrls.has(item.url)) continue;
-    seenUrls.add(item.url);
-    try {
-      const normalized = getPageProcessor().normalize(item, jobId);
-      // upsert on (jobId, url) — idempotent on BullMQ retries
-      const page = await getPageRepository().upsert(normalized);
-      await savePageAssets(jobId, page.id, item);
-      await scanAndFlagPage(
-        page.id,
-        normalized.markdownContent,
-        normalized.title,
-        normalized.description,
-      );
-      await runExtractionIfTemplate(jobId, page.id, item.url, item, userId);
+    if (persistedUrls.has(item.url)) {
       if (item.success) successCount++;
       else failedCount++;
-    } catch (err: unknown) {
-      console.error(
-        `[Worker] Failed to save page ${item.url}: ${getErrorMessage(err)}`,
-      );
+      continue;
+    }
+    persistedUrls.add(item.url);
+    const res = await persistSinglePage(jobId, item, userId);
+    if (!res.saved) {
       saveErrors++;
+    } else if (res.success) {
+      successCount++;
+    } else {
+      failedCount++;
     }
 
     // Update DB progress counters every 10 pages — avoids N DB writes for large batches
@@ -172,14 +222,14 @@ export async function persistBatchResults(
       void getJobRepository().updateProgress(jobId, {
         successPages: successCount,
         failedPages: failedCount + saveErrors,
-        totalPages: seenUrls.size,
+        totalPages: persistedUrls.size,
       });
     }
   }
 
   for (const failed of result.failedUrls ?? []) {
-    if (seenUrls.has(failed.url)) continue;
-    seenUrls.add(failed.url);
+    if (persistedUrls.has(failed.url)) continue;
+    persistedUrls.add(failed.url);
     try {
       const normalized = getPageProcessor().normalizeFailedPage(failed, jobId);
       await getPageRepository().upsert(normalized);
@@ -193,8 +243,8 @@ export async function persistBatchResults(
   }
 
   for (const blockedUrl of result.robotsBlockedUrls ?? []) {
-    if (seenUrls.has(blockedUrl)) continue;
-    seenUrls.add(blockedUrl);
+    if (persistedUrls.has(blockedUrl)) continue;
+    persistedUrls.add(blockedUrl);
     try {
       const normalized = getPageProcessor().normalizeFailedPage(
         { url: blockedUrl, error: "Blocked by robots.txt" },
@@ -211,7 +261,7 @@ export async function persistBatchResults(
     }
   }
 
-  return { successCount, failedCount, saveErrors, totalPages: seenUrls.size };
+  return { successCount, failedCount, saveErrors, totalPages: persistedUrls.size };
 }
 
 async function logStep(
@@ -369,17 +419,41 @@ export async function processCrawlJob(job: Job): Promise<void> {
           `[Worker] Job ${jobId} sitemap parsed: ${sitemapUrls.length} URLs`,
         );
 
+        const persistedUrls = new Set<string>();
+        let currentSuccessCount = 0;
+        let currentFailedCount = 0;
         let pollCount = 0;
+
         const result = await getFirecrawlService().batchScrapePages(
           sitemapUrls,
           crawlJob.maxPages,
-          async (completed, total) => {
+          async (completed, total, currentPages) => {
+            if (currentPages && currentPages.length > 0) {
+              const { newSuccess, newFailed } = await persistIncrementalPages(
+                jobId,
+                currentPages,
+                persistedUrls,
+                crawlJob.userId,
+              );
+              currentSuccessCount += newSuccess;
+              currentFailedCount += newFailed;
+            }
+
+            const effectiveSuccess = Math.max(completed, currentSuccessCount);
+            const progressData: {
+              successPages: number;
+              totalPages: number;
+              failedPages?: number;
+            } = {
+              successPages: effectiveSuccess,
+              totalPages: total,
+            };
+            if (currentFailedCount > 0) {
+              progressData.failedPages = currentFailedCount;
+            }
             await Promise.all([
-              job.updateProgress({ completed, total }),
-              getJobRepository().updateProgress(jobId, {
-                successPages: completed,
-                totalPages: total,
-              }),
+              job.updateProgress({ completed: effectiveSuccess, total }),
+              getJobRepository().updateProgress(jobId, progressData),
             ]);
           },
           async () => {
@@ -391,12 +465,24 @@ export async function processCrawlJob(job: Job): Promise<void> {
         );
 
         if (!result.success) {
+          if (result.status === "cancelled") {
+            console.log(
+              `[Worker] Job ${jobId} sitemap scrape was cancelled. Preserved ${persistedUrls.size} crawled pages.`,
+            );
+            await getJobRepository().updateStatus(jobId, JOB_STATUS.CANCELED, {
+              finishedAt: new Date(),
+              successPages: currentSuccessCount,
+              totalPages: Math.max(result.total, persistedUrls.size),
+            });
+            return;
+          }
+
           await getJobRepository().updateStatus(jobId, JOB_STATUS.FAILED, {
             finishedAt: new Date(),
             errorMessage: mapCrawlError(result.error ?? "Batch scrape failed"),
-            totalPages: result.total,
-            successPages: 0,
-            failedPages: result.total || 1,
+            totalPages: Math.max(result.total, persistedUrls.size),
+            successPages: currentSuccessCount,
+            failedPages: Math.max(1, result.total - currentSuccessCount),
           });
           console.log(
             `[Worker] Job ${jobId} batch scrape failed: ${result.error ?? "Batch scrape failed"}`,
@@ -405,7 +491,7 @@ export async function processCrawlJob(job: Job): Promise<void> {
         }
 
         const { successCount, failedCount, saveErrors, totalPages } =
-          await persistBatchResults(jobId, result, crawlJob.userId);
+          await persistBatchResults(jobId, result, crawlJob.userId, persistedUrls);
         console.log(
           `[Worker] Job ${jobId} completed: ${successCount} success, ${failedCount} failed, ${saveErrors} save errors, ${totalPages} total`,
         );
@@ -431,17 +517,41 @@ export async function processCrawlJob(job: Job): Promise<void> {
           return;
         }
 
+        const persistedUrls = new Set<string>();
+        let currentSuccessCount = 0;
+        let currentFailedCount = 0;
         let pollCount = 0;
+
         const result = await getFirecrawlService().batchScrapePages(
           crawlJob.urls,
           crawlJob.maxPages,
-          async (completed, total) => {
+          async (completed, total, currentPages) => {
+            if (currentPages && currentPages.length > 0) {
+              const { newSuccess, newFailed } = await persistIncrementalPages(
+                jobId,
+                currentPages,
+                persistedUrls,
+                crawlJob.userId,
+              );
+              currentSuccessCount += newSuccess;
+              currentFailedCount += newFailed;
+            }
+
+            const effectiveSuccess = Math.max(completed, currentSuccessCount);
+            const progressData: {
+              successPages: number;
+              totalPages: number;
+              failedPages?: number;
+            } = {
+              successPages: effectiveSuccess,
+              totalPages: total,
+            };
+            if (currentFailedCount > 0) {
+              progressData.failedPages = currentFailedCount;
+            }
             await Promise.all([
-              job.updateProgress({ completed, total }),
-              getJobRepository().updateProgress(jobId, {
-                successPages: completed,
-                totalPages: total,
-              }),
+              job.updateProgress({ completed: effectiveSuccess, total }),
+              getJobRepository().updateProgress(jobId, progressData),
             ]);
           },
           async () => {
@@ -453,18 +563,30 @@ export async function processCrawlJob(job: Job): Promise<void> {
         );
 
         if (!result.success) {
+          if (result.status === "cancelled") {
+            console.log(
+              `[Worker] Job ${jobId} URL_LIST scrape was cancelled. Preserved ${persistedUrls.size} crawled pages.`,
+            );
+            await getJobRepository().updateStatus(jobId, JOB_STATUS.CANCELED, {
+              finishedAt: new Date(),
+              successPages: currentSuccessCount,
+              totalPages: Math.max(result.total, persistedUrls.size),
+            });
+            return;
+          }
+
           await getJobRepository().updateStatus(jobId, JOB_STATUS.FAILED, {
             finishedAt: new Date(),
             errorMessage: mapCrawlError(result.error ?? "Batch scrape failed"),
-            totalPages: result.total,
-            successPages: 0,
-            failedPages: result.total || 1,
+            totalPages: Math.max(result.total, persistedUrls.size),
+            successPages: currentSuccessCount,
+            failedPages: Math.max(1, result.total - currentSuccessCount),
           });
           return;
         }
 
         const { successCount, failedCount, saveErrors, totalPages } =
-          await persistBatchResults(jobId, result, crawlJob.userId);
+          await persistBatchResults(jobId, result, crawlJob.userId, persistedUrls);
         console.log(
           `[Worker] Job ${jobId} completed: ${successCount} success, ${failedCount} failed, ${saveErrors} save errors, ${totalPages} total`,
         );
@@ -480,6 +602,9 @@ export async function processCrawlJob(job: Job): Promise<void> {
           `[Worker] Job ${jobId} started, mode=${crawlJob.mode}, url=${crawlJob.startUrl}`,
         );
 
+        const persistedUrls = new Set<string>();
+        let currentSuccessCount = 0;
+        let currentFailedCount = 0;
         let pollCount = 0;
         let firecrawlJobIdSaved = false;
 
@@ -487,13 +612,33 @@ export async function processCrawlJob(job: Job): Promise<void> {
           crawlJob.startUrl,
           crawlJob.maxPages,
           crawlJob.maxDepth,
-          async (completed, total) => {
+          async (completed, total, currentPages) => {
+            if (currentPages && currentPages.length > 0) {
+              const { newSuccess, newFailed } = await persistIncrementalPages(
+                jobId,
+                currentPages,
+                persistedUrls,
+                crawlJob.userId,
+              );
+              currentSuccessCount += newSuccess;
+              currentFailedCount += newFailed;
+            }
+
+            const effectiveSuccess = Math.max(completed, currentSuccessCount);
+            const progressData: {
+              successPages: number;
+              totalPages: number;
+              failedPages?: number;
+            } = {
+              successPages: effectiveSuccess,
+              totalPages: total,
+            };
+            if (currentFailedCount > 0) {
+              progressData.failedPages = currentFailedCount;
+            }
             await Promise.all([
-              job.updateProgress({ completed, total }),
-              getJobRepository().updateProgress(jobId, {
-                successPages: completed,
-                totalPages: total,
-              }),
+              job.updateProgress({ completed: effectiveSuccess, total }),
+              getJobRepository().updateProgress(jobId, progressData),
             ]);
           },
           async () => {
@@ -521,12 +666,24 @@ export async function processCrawlJob(job: Job): Promise<void> {
         }
 
         if (!result.success) {
+          if (result.status === "cancelled") {
+            console.log(
+              `[Worker] Job ${jobId} crawl was cancelled. Preserved ${persistedUrls.size} crawled pages.`,
+            );
+            await getJobRepository().updateStatus(jobId, JOB_STATUS.CANCELED, {
+              finishedAt: new Date(),
+              successPages: currentSuccessCount,
+              totalPages: Math.max(result.total, persistedUrls.size),
+            });
+            return;
+          }
+
           await getJobRepository().updateStatus(jobId, JOB_STATUS.FAILED, {
             finishedAt: new Date(),
             errorMessage: mapCrawlError(result.error ?? "Crawl failed"),
-            totalPages: result.total,
-            successPages: 0,
-            failedPages: result.total || 1,
+            totalPages: Math.max(result.total, persistedUrls.size),
+            successPages: currentSuccessCount,
+            failedPages: Math.max(1, result.total - currentSuccessCount),
           });
           console.log(
             `[Worker] Job ${jobId} crawl failed: ${result.error ?? "Crawl failed"}`,
@@ -535,7 +692,7 @@ export async function processCrawlJob(job: Job): Promise<void> {
         }
 
         const { successCount, failedCount, saveErrors, totalPages } =
-          await persistBatchResults(jobId, result, crawlJob.userId);
+          await persistBatchResults(jobId, result, crawlJob.userId, persistedUrls);
         console.log(
           `[Worker] Job ${jobId} completed: ${successCount} success, ${failedCount} failed, ${saveErrors} save errors, ${totalPages} total`,
         );