Merge pull request #7 from Klimatbyran/replace-all-emissins-optional-flag

added replace-all-emissions flag for re-runs

Author: eli6

src/routes/readQueues.ts

@@ -67,7 +67,7 @@ export async function readQueuesRoute(app: FastifyInstance) {
     {
       schema: {
         summary: 'Add job to a queue',
-        description: 'Enqueue one or more URLs into the specified queue. Optional flags include autoApprove and forceReindex (alias: force-reindex).',
+        description: 'Enqueue one or more URLs into the specified queue. Optional flags include autoApprove, replaceAllEmissions and forceReindex (alias: force-reindex).',
         tags: ['Queues'],
         params: readQueuePathParamsSchema,
         body: addQueueJobBodySchema,
@@ -88,7 +88,7 @@ export async function readQueuesRoute(app: FastifyInstance) {
       if (!resolvedName) {
         return reply.status(400).send({ error: `Unknown queue '${name}'. Valid queues: ${Object.values(QUEUE_NAMES).join(', ')}` });
       }
-      const { urls, autoApprove, forceReindex, threadId } = request.body as any;
+      const { urls, autoApprove, forceReindex, threadId, replaceAllEmissions } = request.body as any;
 
       // Resolve threadId: accept provided threadId/runId or generate a new one
       const providedOrCreatedThreadId = threadId || `run_${Date.now()}_${Math.random().toString(36).slice(2,8)}`;
@@ -99,13 +99,14 @@ export async function readQueuesRoute(app: FastifyInstance) {
           urlsCount: Array.isArray(urls) ? urls.length : 0,
           autoApprove: !!autoApprove,
           forceReindex: !!forceReindex,
+          replaceAllEmissions: !!replaceAllEmissions,
         },
         'Enqueue request received'
       );
       const queueService = await QueueService.getQueueService();
       const addedJobs: BaseJob[] = [];
       for(const url of urls) {
-        const addedJob = await queueService.addJob(resolvedName, url, autoApprove, { forceReindex, threadId: providedOrCreatedThreadId });
+        const addedJob = await queueService.addJob(resolvedName, url, autoApprove, { forceReindex, threadId: providedOrCreatedThreadId, replaceAllEmissions });
         addedJobs.push(addedJob);
       }
       return reply.send(addedJobs);

src/schemas/request.ts

@@ -30,4 +30,5 @@ export const addQueueJobBodySchema = z.object({
     urls: z.array(string().url()),
     forceReindex: z.boolean().optional().describe('Re-index markdown even if already indexed'),
     threadId: z.string().optional(),
+    replaceAllEmissions: z.boolean().optional().default(false).describe('Replace all scope 1,2,3 emissions and totals (delete old ones from all periods) before adding new ones')
 });

src/schemas/response.ts

@@ -23,6 +23,7 @@ export const baseJobSchema = z.object({
     url: z.string().url().optional(),
     autoApprove: z.boolean().optional().default(false),
     timestamp: z.number(),
+    processedOn: z.number().optional(),
     processId: z.string().optional(),
     threadId: z.string().optional(),
     queue: z.string(),

src/services/QueueService.ts

@@ -89,15 +89,16 @@ export class QueueService {
         return jobs;
     }
 
-    public async addJob(queueName: string, url: string, autoApprove: boolean = false, options?: { forceReindex?: boolean; threadId?: string }): Promise<BaseJob> {
+    public async addJob(queueName: string, url: string, autoApprove: boolean = false, options?: { forceReindex?: boolean; threadId?: string; replaceAllEmissions?: 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,
             ...(options?.threadId ? { threadId: options.threadId } : {}),
-            ...(options?.forceReindex !== undefined ? { forceReindex: options.forceReindex } : {})
+            ...(options?.forceReindex !== undefined ? { forceReindex: options.forceReindex } : {}),
+            ...(options?.replaceAllEmissions !== undefined ? { replaceAllEmissions: options.replaceAllEmissions } : {})
         });
         return transformJobtoBaseJob(job);
     }