Merge pull request #14 from Klimatbyran/rerun-and-retrigger-saving-too

rerunning all or a limited number of scope workers

Author: eli6

src/routes/readQueues.ts

@@ -275,8 +275,8 @@ export async function readQueuesRoute(app: FastifyInstance) {
     '/rerun-by-worker',
     {
       schema: {
-        summary: 'Re-run all jobs that match a given worker name',
-        description: 'Re-runs all jobs across one or more queues whose data.runOnly[] contains the specified worker name. Defaults to completed and failed jobs.',
+        summary: 'Re-run all jobs that match a given worker name (using rerun-and-save)',
+        description: 'Re-runs all jobs across one or more queues whose data.runOnly[] contains the specified worker name using the rerun-and-save approach (finds original EXTRACT_EMISSIONS job and creates new one with specified scopes). Defaults to completed and failed jobs.',
         tags: ['Queues'],
         body: rerunJobsByWorkerBodySchema,
         response: {
@@ -293,11 +293,12 @@ export async function readQueuesRoute(app: FastifyInstance) {
           workerName: string;
           statuses?: JobType[];
           queues?: string[];
+          limit?: number | 'all';
         }
       }>,
       reply
     ) => {
-      const { workerName, statuses, queues } = request.body;
+      const { workerName, statuses, queues, limit } = request.body;
 
       const queueService = await QueueService.getQueueService();
 
@@ -308,6 +309,7 @@ export async function readQueuesRoute(app: FastifyInstance) {
       const result = await queueService.rerunJobsByWorkerName(workerName, {
         queueNames: resolvedQueues,
         statuses,
+        limit,
       });
 
       return reply.send(result);

src/schemas/request.ts

@@ -41,6 +41,7 @@ export const rerunJobsByWorkerBodySchema = z.object({
     workerName: z.string().describe('Name of the worker / pipeline step to re-run (e.g. "scope1+2")'),
     statuses: z.array(jobStatusSchema).optional().describe('Optional list of job statuses to consider (defaults to completed and failed jobs)'),
     queues: z.array(z.string()).optional().describe('Optional list of queue names to restrict the rerun to (defaults to all known queues)'),
+    limit: z.union([z.number(), z.literal('all')]).optional().default(5).describe('Maximum number of companies to rerun (defaults to 5). Use "all" to rerun all companies.'),
 });
 
 export const rerunAndSaveQueueJobBodySchema = z.object({

src/services/QueueService.ts

@@ -400,12 +400,16 @@ export class QueueService {
      *
      * By default it will re-run jobs that are either completed or failed, since
      * waiting/active jobs are already in progress.
+     * 
+     * Limits to specified number of companies (default 5) and only reruns the latest threadId per company.
+     * Use limit: 'all' to rerun all companies.
      */
     public async rerunJobsByWorkerName(
         workerName: string,
         options?: {
             queueNames?: string[];
             statuses?: JobType[];
+            limit?: number | 'all';
         }
     ): Promise<{ totalMatched: number; perQueue: Record<string, number> }> {
         const queueNames = options?.queueNames && options.queueNames.length > 0
@@ -416,12 +420,50 @@ export class QueueService {
             ? options.statuses
             : (['completed', 'failed'] as JobType[]);
 
-        console.info('[QueueService] rerunJobsByWorkerName: Starting', {
+        const limit = options?.limit ?? 5;
+
+        console.info('[QueueService] rerunJobsByWorkerName: Starting (using rerun-and-save)', {
+            workerName,
+            queueNames,
+            statuses,
+            limit
+        });
+
+        const { allMatchingJobInfos, perQueue, totalMatched } = await this.collectMatchingJobsByWorker(
             workerName,
             queueNames,
             statuses
+        );
+
+        const jobsToRerun = this.selectLatestThreadIdPerCompany(allMatchingJobInfos);
+        const limitedJobsToRerun = limit === 'all' 
+            ? jobsToRerun 
+            : this.limitToMostRecent(jobsToRerun, limit);
+
+        await this.rerunSelectedJobs(limitedJobsToRerun, workerName);
+
+        console.info('[QueueService] rerunJobsByWorkerName: Completed (using rerun-and-save)', {
+            workerName,
+            totalMatched,
+            uniqueCompanies: jobsToRerun.length,
+            jobsRerun: limitedJobsToRerun.length,
+            limit,
+            perQueue
         });
 
+        return { totalMatched, perQueue };
+    }
+
+    private async collectMatchingJobsByWorker(
+        workerName: string,
+        queueNames: string[],
+        statuses: JobType[]
+    ): Promise<{
+        allMatchingJobInfos: Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }>;
+        perQueue: Record<string, number>;
+        totalMatched: number;
+    }> {
+        const allMatchingJobInfos: Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }> = [];
         const perQueue: Record<string, number> = {};
         let totalMatched = 0;
 
@@ -441,28 +483,85 @@ export class QueueService {
             });
 
             for (const job of matchingJobs) {
-                try {
-                    await this.rerunJob(queueName, job.id!);
-                } catch (error) {
-                    console.error('[QueueService] rerunJobsByWorkerName: Failed to rerun job', {
-                        queueName,
-                        jobId: job.id,
-                        error
-                    });
-                }
+                const jobInfo = this.extractJobInfo(job, queueName);
+                allMatchingJobInfos.push(jobInfo);
             }
 
             perQueue[queueName] = matchingJobs.length;
             totalMatched += matchingJobs.length;
         }
 
-        console.info('[QueueService] rerunJobsByWorkerName: Completed', {
-            workerName,
-            totalMatched,
-            perQueue
+        return { allMatchingJobInfos, perQueue, totalMatched };
+    }
+
+    private extractJobInfo(
+        job: Job,
+        queueName: string
+    ): { job: Job; queueName: string; companyName: string; threadId: string; timestamp: number } {
+        const jobData: any = job.data ?? {};
+        const companyName = jobData.companyName ?? jobData.threadId ?? 'unknown';
+        const threadId = jobData.threadId ?? job.id ?? 'unknown';
+        const timestamp = job.timestamp ?? 0;
+
+        return {
+            job,
+            queueName,
+            companyName,
+            threadId,
+            timestamp
+        };
+    }
+
+    private selectLatestThreadIdPerCompany(
+        jobInfos: Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }>
+    ): Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }> {
+        const jobsByCompany = new Map<string, typeof jobInfos[0]>();
+
+        for (const jobInfo of jobInfos) {
+            const existing = jobsByCompany.get(jobInfo.companyName);
+            if (!existing || jobInfo.timestamp > existing.timestamp) {
+                jobsByCompany.set(jobInfo.companyName, jobInfo);
+            }
+        }
+
+        return Array.from(jobsByCompany.values());
+    }
+
+    private limitToMostRecent(
+        jobInfos: Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }>,
+        limit: number
+    ): Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }> {
+        return jobInfos
+            .sort((firstJob, secondJob) => secondJob.timestamp - firstJob.timestamp)
+            .slice(0, limit);
+    }
+
+    private async rerunSelectedJobs(
+        jobInfos: Array<{ job: Job; queueName: string; companyName: string; threadId: string; timestamp: number }>,
+        workerName: string
+    ): Promise<void> {
+        console.info('[QueueService] rerunJobsByWorkerName: Deduplicated and limited', {
+            jobsToRerun: jobInfos.length
         });
 
-        return { totalMatched, perQueue };
+        for (const jobInfo of jobInfos) {
+            try {
+                await this.rerunExtractEmissionsFromFollowup(
+                    jobInfo.queueName,
+                    jobInfo.job.id!,
+                    [workerName]
+                );
+            } catch (error) {
+                console.error('[QueueService] rerunJobsByWorkerName: Failed to rerun and save job', {
+                    queueName: jobInfo.queueName,
+                    jobId: jobInfo.job.id,
+                    companyName: jobInfo.companyName,
+                    threadId: jobInfo.threadId,
+                    workerName,
+                    error
+                });
+            }
+        }
     }
 }