feat: task updates & jobId

Author: dobosmarton

app/api/articles/[id]/similars/route.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from 'crypto';
 import { NextRouteFunction } from '@/lib/route-validator.server';
 import * as similarityService from '@/lib/article-metadata/similarity.server';
 import * as redis from '@/lib/redis';
@@ -10,7 +11,10 @@ const generateEmbeddingAsync = redis.cacheableFunction<string, { id: string }>(
   redis.asyncEmbeddingGenerationSchema,
   { ex: 60 * 60 }
 )(async (metadataId) =>
-  generateEmbeddingsFromPdf.trigger({ id: metadataId }, { idempotencyKey: `similarity-${metadataId}` })
+  generateEmbeddingsFromPdf.trigger(
+    { id: metadataId, jobId: randomUUID() },
+    { idempotencyKey: `similarity-${metadataId}` }
+  )
 );
 
 const getSimilarArticlesByMetadataId: NextRouteFunction<Params> = async (_, { params }) => {

trigger/ai/embedding-from-pdf.ts

@@ -9,15 +9,17 @@ export const generateEmbeddingsFromPdf = task({
     const payload = metadataIdPayloadSchema.parse(_payload);
     logger.info(`Generate Embedding from pdf - Id: ${payload.id}`, { time: new Date().toISOString() });
 
-    const pdfDocs = await loadPdf.triggerAndWait(payload.id);
+    const pdfDocs = await loadPdf.triggerAndWait(payload.id, {
+      idempotencyKey: `load-pdf-${payload.jobId}-${payload.id}`,
+    });
 
     if (!pdfDocs.ok || !pdfDocs.output) {
       throw new Error(`PDF not found for metadata id: ${payload.id}`);
     }
 
     await generateEmbedding.trigger(
-      { itemId: payload.id, doc: pdfDocs.output.doc },
-      { idempotencyKey: `generate-metadata-embedding-${payload.id}` }
+      { itemId: payload.id, doc: pdfDocs.output.doc, jobId: payload.jobId },
+      { idempotencyKey: `generate-metadata-embedding-${payload.jobId}-${payload.id}` }
     );
 
     logger.info(`Add Embeddings - Done`, { time: new Date().toISOString() });

trigger/ai/generate-ai-content.ts

@@ -1,23 +1,26 @@
 import { logger, task } from '@trigger.dev/sdk/v3';
 import * as articleMetadataService from '@/lib/article-metadata/metadata.server';
-import { GenerateAiContent, generateAiContentSchema } from '../schema';
+import { GenerateAiContentPayload, generateAiContentPayloadSchema } from '../schema';
 import { generateEmbeddingsFromPdf } from './embedding-from-pdf';
 
 export const generateAIContent = task({
   id: 'generate-ai-content',
-  run: async (_payload: GenerateAiContent) => {
-    const payload = generateAiContentSchema.parse(_payload);
+  run: async (_payload: GenerateAiContentPayload) => {
+    const payload = generateAiContentPayloadSchema.parse(_payload);
 
     const metadataList = await articleMetadataService.getArticleMetadataIdsWithZeroEmbeddingsByIds(
-      payload.map((payloadItem) => payloadItem.externalId)
+      payload.data.map((payloadItem) => payloadItem.externalId)
     );
 
     logger.info(`Fetched metadata count: ${metadataList.length} - Done`, {
       time: new Date().toISOString(),
     });
 
     await generateEmbeddingsFromPdf.batchTrigger(
-      metadataList.map((item) => ({ payload: { id: item.id, externalId: item.external_id } }))
+      metadataList.map((item) => ({
+        payload: { id: item.id, externalId: item.external_id, jobId: payload.jobId },
+        options: { idempotencyKey: `generate-embedding-from-pdf-${payload.jobId}-${item.id}` },
+      }))
     );
   },
 });

trigger/research-sync/add-article-metadata-batch.ts

@@ -20,7 +20,10 @@ export const addArticleMetadaBatch = task({
       return;
     }
 
-    await generateAIContent.trigger(newIds.map((id) => ({ externalId: id })));
+    await generateAIContent.trigger(
+      { data: newIds.map((id) => ({ externalId: id })), jobId: payload.jobId },
+      { idempotencyKey: `generate-ai-content-${payload.jobId}-${payload.batchIndex}` }
+    );
 
     logger.info(`Add Article Metadata Batch - Done`, { time: new Date().toISOString() });
   },

trigger/research-sync/scheduler.ts

@@ -9,9 +9,8 @@ export const researchSync = schedules.task({
     const startDate = new Date();
     startDate.setDate(startDate.getDate() - 1);
 
-    const jobId = randomUUID();
-
     await syncMetadata.trigger({
+      jobId: payload.externalId ?? randomUUID(),
       startDate: startDate.toISOString().split('T')[0],
     });
 
@@ -23,6 +22,7 @@ export const researchSync = schedules.task({
 
 export const researchSyncSchedule = schedules.create({
   task: researchSync.id,
+  externalId: randomUUID(),
   deduplicationKey: 'research-sync-scheduler',
   // At minute 0 past every 6th hour
   cron: '0 */8 * * *',

trigger/research-sync/sync-metadata.ts

@@ -68,7 +68,9 @@ export const syncMetadata = task({
       throw new Error(`Error fetching data from OAI-PMH server for ${payload.startDate} - ${payload.untilDate}`);
     }
 
-    const parsedData = await parseMetadata.triggerAndWait(responseTextData);
+    const parsedData = await parseMetadata.triggerAndWait(responseTextData, {
+      idempotencyKey: `parse-metadata-${payload.jobId}-${payload.startDate}-${payload.untilDate}`,
+    });
 
     if (!parsedData.ok) {
       throw new Error('Error parsing data');
@@ -83,7 +85,7 @@ export const syncMetadata = task({
     // send the batch to the add_article_metadata_batch event
     // no need to wait for the result
     await addArticleMetadaBatch.batchTrigger(
-      batches.map((batch, index) => ({ payload: { batch, batchIndex: index } }))
+      batches.map((batch, index) => ({ payload: { batch, batchIndex: index, jobId: payload.jobId } }))
     );
 
     const token = getResumptionToken(parsedData.output.resumptionToken, parsedData.output.records.length);
@@ -93,11 +95,17 @@ export const syncMetadata = task({
     if (token) {
       // send the resumption token to the same event
       // no need to wait for the result
-      await syncMetadata.trigger({
-        startDate: payload.startDate,
-        untilDate: payload.untilDate,
-        resumptionToken: token,
-      });
+      await syncMetadata.trigger(
+        {
+          jobId: payload.jobId,
+          startDate: payload.startDate,
+          untilDate: payload.untilDate,
+          resumptionToken: token,
+        },
+        {
+          idempotencyKey: `sync-metadata-${payload.jobId}-${payload.startDate}-${payload.untilDate}`,
+        }
+      );
     }
   },
 });

trigger/schema/article-metadata.ts

@@ -46,11 +46,13 @@ const metadataBatchSchema = z
   }));
 
 export const addArticleMetadaBatchPayloadSchema = z.object({
+  jobId: z.string(),
   batchIndex: z.number(),
   batch: z.array(metadataBatchSchema),
 });
 
 export const metadataIdPayloadSchema = z.object({
+  jobId: z.string(),
   id: z.string(),
 });
 

trigger/schema/generate-ai-content.ts

@@ -1,9 +1,12 @@
 import { z } from 'zod';
 
-export const generateAiContentSchema = z.array(
-  z.object({
-    externalId: z.string(),
-  })
-);
+export const generateAiContentPayloadSchema = z.object({
+  jobId: z.string(),
+  data: z.array(
+    z.object({
+      externalId: z.string(),
+    })
+  ),
+});
 
-export type GenerateAiContent = z.input<typeof generateAiContentSchema>;
+export type GenerateAiContentPayload = z.input<typeof generateAiContentPayloadSchema>;

trigger/schema/generate-embedding.ts

@@ -1,6 +1,7 @@
 import { z } from 'zod';
 
 export const generateEmbeddingPayloadSchema = z.object({
+  jobId: z.string(),
   itemId: z.string(),
   // Document<PDFMetadata<{ metadata_id: string }>>[]
   doc: z.array(

trigger/schema/research-sync.ts

@@ -1,6 +1,7 @@
 import { z } from 'zod';
 
 export const researchSyncPayloadSchema = z.object({
+  jobId: z.string(),
   startDate: z.string().regex(/^[0-9]{4}-[0-9]{2}-[0-9]{2}$/),
   untilDate: z
     .string()