Merge pull request #4 from Klimatbyran/force-reindex-and-error-handling

added force reindex flag to pass to pipeline + error handling

Author: eli6

src/lib/bullmq.ts

@@ -7,6 +7,7 @@ export const JOB_STATUS = ['active', 'waiting', 'waiting-children', 'prioritized
 export const QUEUE_NAMES = {
   NLM_PARSE_PDF: 'nlmParsePDF',
   DOCLING_PARSE_PDF: 'doclingParsePDF',
+  PARSE_PDF: 'parsePdf',
   NLM_EXTRACT_TABLES: 'nlmExtractTables',
   INDEX_MARKDOWN: 'indexMarkdown',
   PRECHECK: 'precheck',

src/routes/readQueues.ts

@@ -2,7 +2,7 @@ import { FastifyInstance, FastifyRequest } from "fastify";
 import { QueueService } from "../services/QueueService";
 import { AddJobBody, BaseJob } from "../schemas/types";
 import { error404ResponseSchema, queueAddJobResponseSchema, queueJobResponseSchema, queueResponseSchema, queueStatsResponseSchema } from "../schemas/response";
-import { JOB_STATUS, STATUS } from "../lib/bullmq";
+import { STATUS, QUEUE_NAMES } from "../lib/bullmq";
 import { JobType } from "bullmq";
 import { addQueueJobBodySchema, readQueueJobPathParamsSchema, readQueuePathParamsSchema, readQueueQueryStringSchema, readQueueStatsQueryStringSchema } from "../schemas/request";
 
@@ -67,7 +67,7 @@ export async function readQueuesRoute(app: FastifyInstance) {
     {
       schema: {
         summary: 'Add job to a queue',
-        description: '',
+        description: 'Enqueue one or more URLs into the specified queue. Optional flags include autoApprove and forceReindex (alias: force-reindex).',
         tags: ['Queues'],
         params: readQueuePathParamsSchema,
         body: addQueueJobBodySchema,
@@ -84,11 +84,25 @@ export async function readQueuesRoute(app: FastifyInstance) {
       reply
     ) => {
       const { name } = request.params;
-      const { urls, autoApprove} = request.body;
+      const resolvedName = Object.values(QUEUE_NAMES).find(q => q.toLowerCase() === name.toLowerCase());
+      if (!resolvedName) {
+        return reply.status(400).send({ error: `Unknown queue '${name}'. Valid queues: ${Object.values(QUEUE_NAMES).join(', ')}` });
+      }
+      const { urls, autoApprove, forceReindex } = request.body;
+      // Log enqueue request (sanitized)
+      app.log.info(
+        {
+          queue: name,
+          urlsCount: Array.isArray(urls) ? urls.length : 0,
+          autoApprove: !!autoApprove,
+          forceReindex: !!forceReindex,
+        },
+        'Enqueue request received'
+      );
       const queueService = await QueueService.getQueueService();
       const addedJobs: BaseJob[] = [];
       for(const url of urls) {
-        const addedJob = await queueService.addJob(name, url, autoApprove);
+        const addedJob = await queueService.addJob(resolvedName, url, autoApprove, { forceReindex });
         addedJobs.push(addedJob);
       }
       return reply.send(addedJobs);

src/schemas/request.ts

@@ -22,7 +22,11 @@ export const readQueueJobPathParamsSchema = z.object({
     id: z.string()
 });
 
+// Accept both camelCase and kebab-case for the reindex flag.
+// Normalization to camelCase is done in the route handler to avoid schema transforms
+// that can accidentally mark fields as required in JSON schema.
 export const addQueueJobBodySchema = z.object({
     autoApprove: z.boolean().optional().default(false),
     urls: z.array(string().url()),
+    forceReindex: z.boolean().optional().describe('Re-index markdown even if already indexed'),
 }).required();

src/services/ProcessService.ts

@@ -24,6 +24,9 @@ export class ProcessService {
 
     public async getProcesses(): Promise<Process[]> {
         const jobs = await this.queueService.getDataJobs(undefined, undefined);
+        // Debug: log number of jobs fetched across all queues
+        // Using console here; Fastify logger isn't directly available in service layer
+        console.info('[ProcessService] getProcesses: jobs fetched', { count: jobs.length });
         const jobProcesses: Record<string, DataJob[]> = {};
         for(const job of jobs) {
             if(!jobProcesses[job.data.id ?? job.data.threadId ?? "unknown"]) {
@@ -35,6 +38,7 @@ export class ProcessService {
         for(const jobProcess of Object.values(jobProcesses)) {
             processes.push(this.createProcess(jobProcess));
         }
+        console.info('[ProcessService] getProcesses: processes built', { count: processes.length });
         return processes;
     }
 
@@ -55,7 +59,9 @@ export class ProcessService {
                     companyProcesses[company].wikidataId = process.wikidataId;    
                 }
             }
-            return Object.values(companyProcesses);
+            const grouped = Object.values(companyProcesses);
+            console.info('[ProcessService] getProcessesGroupedByCompany: companies grouped', { count: grouped.length });
+            return grouped;
         }
 
     private createProcess(jobs: DataJob[]): Process {

src/services/QueueService.ts

@@ -30,7 +30,12 @@ export class QueueService {
         if(this.queues === undefined)
             this.queues = await this.getQueues();
 
-        return this.queues[name];
+        const queue = this.queues[name];
+        if (!queue) {
+            const available = Object.keys(this.queues).join(', ');
+            throw new Error(`Unknown queue: ${name}. Available: ${available}`);
+        }
+        return queue;
     }
 
     public async getJobs(queueNames?: string[], status?: string, processId?: string): Promise<BaseJob[]> {
@@ -42,11 +47,14 @@ export class QueueService {
         for(const queueName of queueNames) {
              const queue = await this.getQueue(queueName);
             const rawJobs = await queue.getJobs(queryStatus as JobType[]);
-            if(processId) {
-                rawJobs.filter(job => job.data.id === processId || job.data.threadId === processId);
-            }
+            const filteredRawJobs = processId
+                ? rawJobs.filter(job => {
+                    const pid = job.data?.runId ?? job.data?.id ?? job.data?.threadId;
+                    return pid === processId;
+                })
+                : rawJobs;
             const transformedJobs = await Promise.all(
-                rawJobs.map(job => transformJobtoBaseJob(job))
+                filteredRawJobs.map(job => transformJobtoBaseJob(job))
             );
             jobs.push(...transformedJobs);
         }
@@ -62,11 +70,14 @@ export class QueueService {
         for(const queueName of queueNames) {
              const queue = await this.getQueue(queueName);
             const rawJobs = await queue.getJobs(queryStatus as JobType[]);
-            if(processId) {
-                rawJobs.filter(job => job.data.id === processId || job.data.threadId === processId);
-            }
+            const filteredRawJobs = processId
+                ? rawJobs.filter(job => {
+                    const pid = job.data?.runId ?? job.data?.id ?? job.data?.threadId;
+                    return pid === processId;
+                })
+                : rawJobs;
             const transformedJobs = await Promise.all(
-                rawJobs.map(async job => {
+                filteredRawJobs.map(async job => {
                     const dataJob: DataJob = await transformJobtoBaseJob(job);                    
                     dataJob.data = job.data;        
                     dataJob.returnvalue = job.returnvalue;
@@ -78,10 +89,15 @@ export class QueueService {
         return jobs;
     }
 
-    public async addJob(queueName: string, url: string, autoApprove: boolean = false): Promise<BaseJob> {
+    public async addJob(queueName: string, url: string, autoApprove: boolean = false, options?: { forceReindex?: boolean }): Promise<BaseJob> {
         const queue = await this.getQueue(queueName);
         const id = crypto.randomUUID();
-        const job = await queue.add('download ' + url.slice(-20), { url: url.trim(), autoApprove, id });
+        const job = await queue.add('download ' + url.slice(-20), {
+            url: url.trim(),
+            autoApprove,
+            id,
+            ...(options?.forceReindex !== undefined && { forceReindex: options.forceReindex })
+        });
         return transformJobtoBaseJob(job);
     }