✨ feat(api): enhance job management in ScrapeController and SearchController; implement per-page result handling and logging for job status updates in SearchService.

QThans · QThans · commit 975d22c8185a · 2025-08-24T23:23:20.000+08:00
diff --git a/apps/api/src/controllers/v1/ScrapeController.ts b/apps/api/src/controllers/v1/ScrapeController.ts
@@ -11,27 +11,28 @@ export class ScrapeController {
         try {
             // Validate request body and transform it to the job payload structure
             const jobPayload = scrapeSchema.parse(req.body);
+            // Keep engine name available for error paths (e.g., timeout) before awaiting
+            engineName = jobPayload.engine;
 
-            jobId = await QueueManager.getInstance().addJob(`scrape-${jobPayload.engine}`, jobPayload);
+            jobId = await QueueManager.getInstance().addJob(`scrape-${engineName}`, jobPayload);
             await createJob({
                 job_id: jobId,
                 job_type: 'scrape',
-                job_queue_name: `scrape-${jobPayload.engine}`,
+                job_queue_name: `scrape-${engineName}`,
                 url: jobPayload.url,
                 req,
                 status: STATUS.PENDING,
             });
             // Propagate jobId for downstream middlewares (e.g., credits logging)
             req.jobId = jobId;
             // waiting job done
-            const job = await QueueManager.getInstance().waitJobDone(`scrape-${jobPayload.engine}`, jobId, jobPayload.options.timeout || 60_000);
+            const job = await QueueManager.getInstance().waitJobDone(`scrape-${engineName}`, jobId, jobPayload.options.timeout || 60_000);
             const { uniqueKey, queueName, options, engine, ...jobData } = job;
             // for failed job to cancel the job in the queue
-            engineName = engine;
             // Check if job failed
             if (job.status === 'failed' || job.error) {
                 const message = job.message || "The scraping task could not be completed";
-                await QueueManager.getInstance().cancelJob(`scrape-${jobPayload.engine}`, jobId);
+                await QueueManager.getInstance().cancelJob(`scrape-${engineName}`, jobId);
                 await failedJob(jobId, message, false, { total: 1, completed: 0, failed: 1 });
                 res.status(200).json({
                     success: false,
@@ -90,8 +91,15 @@ export class ScrapeController {
             } else {
                 const message = error instanceof Error ? error.message : "Unknown error occurred";
                 if (jobId) {
-                    await QueueManager.getInstance().cancelJob(`scrape-${engineName}`, jobId);
-                    await failedJob(jobId, message, false, { total: 1, completed: 0, failed: 1 });
+                    // Best-effort cancel; do not block failed marking if cancel throws
+                    try {
+                        if (engineName) {
+                            await QueueManager.getInstance().cancelJob(`scrape-${engineName}`, jobId);
+                        }
+                    } catch { /* ignore cancel errors */ }
+                    try {
+                        await failedJob(jobId, message, false, { total: 1, completed: 0, failed: 1 });
+                    } catch { /* ignore DB errors to still return response */ }
                 }
                 res.status(500).json({
                     success: false,
diff --git a/apps/api/src/controllers/v1/SearchController.ts b/apps/api/src/controllers/v1/SearchController.ts
@@ -4,6 +4,8 @@ import { SearchService } from "@anycrawl/search/SearchService";
 import { log } from "@anycrawl/libs/log";
 import { searchSchema } from "../../types/SearchSchema.js";
 import { RequestWithAuth } from "../../types/Types.js";
+import { randomUUID } from "crypto";
+import { STATUS, createJob, insertJobResult, completedJob, failedJob, updateJobCounts, JOB_RESULT_STATUS } from "@anycrawl/db";
 export class SearchController {
     private searchService: SearchService;
 
@@ -26,21 +28,86 @@ export class SearchController {
     }
 
     public handle = async (req: RequestWithAuth, res: Response): Promise<void> => {
+        let jobId: string | null = null;
+        let engineName: string | null = null;
         try {
             // Validate request body against searchSchema
             const validatedData = searchSchema.parse(req.body);
 
             // Execute search and wait for results
-            const results = await this.searchService.search(validatedData.engine ?? "google", {
+            engineName = validatedData.engine ?? "google";
+
+            // Create job for search request (pending)
+            jobId = randomUUID();
+            await createJob({
+                job_id: jobId,
+                job_type: "search",
+                job_queue_name: `search-${engineName}`,
+                url: `search:${validatedData.query}`,
+                req,
+                status: STATUS.PENDING,
+            });
+            req.jobId = jobId;
+
+            const expectedPages = validatedData.pages || 1;
+            let pagesProcessed = 0;
+            let failedPages = 0;
+            let successPages = 0;
+
+            const results = await this.searchService.search(engineName, {
                 query: validatedData.query,
                 limit: validatedData.limit || 10,
                 offset: validatedData.offset || 0,
-                pages: validatedData.pages || 1,
+                pages: expectedPages,
                 lang: validatedData.lang,
                 // country: validatedData.country,
+            }, async (page, pageResults, _uniqueKey, success) => {
+                try {
+                    pagesProcessed += 1;
+                    if (!success) {
+                        failedPages += 1;
+                        // Record a failed page entry (single record per page)
+                        await insertJobResult(
+                            jobId!,
+                            `search:${engineName}:${validatedData.query}:page:${page}`,
+                            { page, query: validatedData.query, results: [] },
+                            JOB_RESULT_STATUS.FAILED
+                        );
+                    } else {
+                        successPages += 1;
+                        // Insert a single record for this page with aggregated results
+                        await insertJobResult(
+                            jobId!,
+                            `search:${engineName}:${validatedData.query}:page:${page}`,
+                            { page, query: validatedData.query, results: pageResults },
+                            JOB_RESULT_STATUS.SUCCESS
+                        );
+                    }
+
+                    // Update job counts based on pages for progress
+                    await updateJobCounts(jobId!, { total: expectedPages, completed: successPages, failed: failedPages });
+                } catch (e) {
+                    log.error(`Per-page handler error for job_id=${jobId}: ${e instanceof Error ? e.message : String(e)}`);
+                }
             });
             // credits used is the number of pages.
-            req.creditsUsed = validatedData.pages;
+            req.creditsUsed = validatedData.pages ?? 1;
+
+            // Mark job status based on page results
+            try {
+                if (failedPages >= expectedPages) {
+                    await failedJob(
+                        jobId,
+                        `All pages failed (${failedPages}/${expectedPages})`,
+                        false,
+                        { total: expectedPages, completed: successPages, failed: failedPages }
+                    );
+                } else {
+                    await completedJob(jobId, true, { total: expectedPages, completed: successPages, failed: failedPages });
+                }
+            } catch (e) {
+                log.error(`Failed to mark job final status for job_id=${jobId}: ${e instanceof Error ? e.message : String(e)}`);
+            }
             res.json({
                 success: true,
                 data: results,
@@ -62,6 +129,13 @@ export class SearchController {
                     },
                 });
             } else {
+                if (jobId) {
+                    try {
+                        await failedJob(jobId, error instanceof Error ? error.message : "Unknown error", false, { total: 0, completed: 0, failed: 0 });
+                    } catch (e) {
+                        log.error(`Failed to mark job failed for job_id=${jobId}: ${e instanceof Error ? e.message : String(e)}`);
+                    }
+                }
                 res.status(500).json({
                     success: false,
                     error: "Internal server error",
diff --git a/apps/api/src/types/BaseSchema.ts b/apps/api/src/types/BaseSchema.ts
@@ -57,7 +57,7 @@ export const baseSchema = z.object({
     /**
      * The timeout to be used
      */
-    timeout: z.number().min(1000).max(600_000).default(60_000),
+    timeout: z.number().min(1000).max(600_000).default(300_000),
 
     /**
      * The wait for to be used
diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts
@@ -1,6 +1,6 @@
 import { eq, and, gt, gte, sql } from "drizzle-orm";
 import { getDB, schemas } from "./db/index.js";
-import { STATUS } from "./map.js";
+import { STATUS, JOB_RESULT_STATUS } from "./map.js";
 import { Job, CreateJobParams } from "./model/Job.js";
 
 // Backward compatibility functions
@@ -14,6 +14,7 @@ export const insertJobResult = Job.insertJobResult;
 export const getJobResults = Job.getJobResults;
 export const getJobResultsPaginated = Job.getJobResultsPaginated;
 export const getJobResultsCount = Job.getJobResultsCount;
+export const updateJobCounts = Job.updateCounts;
 
-export { eq, and, gt, gte, sql, getDB, schemas, STATUS, Job };
+export { eq, and, gt, gte, sql, getDB, schemas, STATUS, JOB_RESULT_STATUS, Job };
 export type { CreateJobParams };
diff --git a/packages/db/src/model/Job.ts b/packages/db/src/model/Job.ts
@@ -1,6 +1,7 @@
 import { asc } from "drizzle-orm";
 import { getDB, schemas, eq, STATUS, sql } from "../index.js";
 import { JOB_RESULT_STATUS, JobResultStatus } from "../map.js";
+import { log } from "@anycrawl/libs/log";
 
 export interface CreateJobParams {
     job_id: string;
@@ -72,6 +73,10 @@ export class Job {
             createdAt: new Date(),
             updatedAt: new Date(),
         });
+
+        try {
+            log.info(`[DB][Job] Created job_id=${job_id} type=${job_type} queue=${job_queue_name} origin=${origin} apiKey=${api_key_id ?? "none"}`);
+        } catch { }
     }
 
     /**
@@ -95,8 +100,14 @@ export class Job {
         const job = await Job.get(job_id);
         if (job) {
             await db.update(schemas.jobs).set({ status: STATUS.CANCELLED }).where(eq(schemas.jobs.jobId, job_id));
+            try {
+                log.info(`[DB][Job] Cancelled job_id=${job_id} status=${STATUS.CANCELLED}`);
+            } catch { }
             return job;
         }
+        try {
+            log.warning(`[DB][Job] Cancel attempt for nonexistent job_id=${job_id}`);
+        } catch { }
         return null;
     }
 
@@ -112,6 +123,9 @@ export class Job {
         } else {
             await db.update(schemas.jobs).set({ status: status }).where(eq(schemas.jobs.jobId, job_id));
         }
+        try {
+            log.info(`[DB][Job] Updated status job_id=${job_id} status=${status}${isSuccess !== null ? ` isSuccess=${isSuccess}` : ""}`);
+        } catch { }
     }
 
     /**
@@ -132,6 +146,11 @@ export class Job {
             ...(counts?.failed !== undefined ? { failed: counts.failed } : {}),
             updatedAt: new Date(),
         }).where(eq(schemas.jobs.jobId, jobId));
+
+        try {
+            const countsInfo = `total=${counts?.total ?? "-"} completed=${counts?.completed ?? "-"} failed=${counts?.failed ?? "-"}`;
+            log.info(`[DB][Job] Marked completed job_id=${jobId} isSuccess=${isSuccess} ${countsInfo}`);
+        } catch { }
     }
 
     /**
@@ -162,6 +181,11 @@ export class Job {
             ...(counts?.failed !== undefined ? { failed: counts.failed } : {}),
             updatedAt: new Date(),
         }).where(eq(schemas.jobs.jobId, jobId));
+
+        try {
+            const countsInfo = `total=${counts?.total ?? "-"} completed=${counts?.completed ?? "-"} failed=${counts?.failed ?? "-"}`;
+            log.error(`[DB][Job] Marked failed job_id=${jobId} message="${errorMessage}" ${countsInfo}`);
+        } catch { }
     }
 
     /**
@@ -188,6 +212,11 @@ export class Job {
             createdAt: new Date(),
             updatedAt: new Date(),
         });
+
+        try {
+            const dataType = data === null || data === undefined ? "null" : Array.isArray(data) ? `array(len=${data.length})` : typeof data;
+            log.info(`[DB][JobResult] Inserted result job_id=${jobId} url=${url} status=${status} dataType=${dataType}`);
+        } catch { }
     }
 
     /**
@@ -248,4 +277,22 @@ export class Job {
         const count = rows?.[0]?.count as unknown as number;
         return Number(count || 0);
     }
+
+    /**
+     * Update job counts without changing status
+     */
+    public static async updateCounts(jobId: string, counts: { total?: number; completed?: number; failed?: number }) {
+        const db = await getDB();
+        await db.update(schemas.jobs).set({
+            ...(counts.total !== undefined ? { total: counts.total } : {}),
+            ...(counts.completed !== undefined ? { completed: counts.completed } : {}),
+            ...(counts.failed !== undefined ? { failed: counts.failed } : {}),
+            updatedAt: new Date(),
+        }).where(eq(schemas.jobs.jobId, jobId));
+
+        try {
+            const countsInfo = `total=${counts.total ?? "-"} completed=${counts.completed ?? "-"} failed=${counts.failed ?? "-"}`;
+            log.info(`[DB][Job] Updated counts job_id=${jobId} ${countsInfo}`);
+        } catch { }
+    }
 }
diff --git a/packages/search/src/SearchService.ts b/packages/search/src/SearchService.ts
@@ -11,12 +11,14 @@ export class SearchService {
     private pendingRequests: Map<string, number>;
     private crawler: Engine | null = null;
     private searchQueue: any | null = null;
+    private requestsToOnPage: Map<string, (page: number, results: SearchResult[], uniqueKey: string, success: boolean) => void>;
 
     constructor() {
         this.engines = new Map();
         this.requestsToResponses = new Map();
         this.partialResults = new Map();
         this.pendingRequests = new Map();
+        this.requestsToOnPage = new Map();
     }
 
     private createEngine(name: string): SearchEngine {
@@ -65,6 +67,12 @@ export class SearchService {
                             const results = await engine.parse(html, request);
                             log.info(`Parsed results: ${results.length}`);
 
+                            const pageNumber = request.userData.page ?? 1;
+
+                            // Per-page callback
+                            const onPageCb = this.requestsToOnPage.get(uniqueKey);
+                            if (onPageCb) onPageCb(pageNumber, results, uniqueKey, true);
+
                             // Accumulate results
                             const currentResults = this.partialResults.get(uniqueKey) || [];
                             this.partialResults.set(uniqueKey, [...currentResults, ...results]);
@@ -83,6 +91,7 @@ export class SearchService {
                                     this.requestsToResponses.delete(uniqueKey);
                                     this.partialResults.delete(uniqueKey);
                                     this.pendingRequests.delete(uniqueKey);
+                                    this.requestsToOnPage.delete(uniqueKey);
                                 }
                             }
                         } catch (error) {
@@ -97,6 +106,12 @@ export class SearchService {
                             const uniqueKey = request.userData.uniqueKey;
                             log.error(`Failed to process ${request.url}:`, error);
 
+                            const pageNumber = request.userData.page ?? 1;
+
+                            // Per-page callback (failure)
+                            const onPageCb = this.requestsToOnPage.get(uniqueKey);
+                            if (onPageCb) onPageCb(pageNumber, [], uniqueKey, false);
+
                             // Decrement pending requests even for failed requests
                             const pendingCount = this.pendingRequests.get(uniqueKey) || 0;
                             const newPendingCount = pendingCount - 1;
@@ -113,6 +128,7 @@ export class SearchService {
                                     this.requestsToResponses.delete(uniqueKey);
                                     this.partialResults.delete(uniqueKey);
                                     this.pendingRequests.delete(uniqueKey);
+                                    this.requestsToOnPage.delete(uniqueKey);
                                 }
                             }
                         } catch (error) {
@@ -128,14 +144,19 @@ export class SearchService {
         }
     }
 
-    async search(engineName: string, options: SearchOptions): Promise<SearchResult[]> {
+    async search(
+        engineName: string,
+        options: SearchOptions,
+        onPage?: (page: number, results: SearchResult[], uniqueKey: string, success: boolean) => void,
+    ): Promise<SearchResult[]> {
         log.info("Search called with options:", options);
         return new Promise(async (resolve) => {
             const uniqueKey = randomUUID();
             log.info(`Created uniqueKey for search: ${uniqueKey}`);
             this.requestsToResponses.set(uniqueKey, resolve);
             this.partialResults.set(uniqueKey, []);
             this.pendingRequests.set(uniqueKey, options.pages ?? 1);
+            if (onPage) this.requestsToOnPage.set(uniqueKey, onPage);
 
             await this.executeSearch(engineName, options, uniqueKey);
         });
@@ -205,4 +226,5 @@ export class SearchService {
             }
         }
     }
+
 }
