RunEngine 2.0 batch trigger support

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -1,23 +1,23 @@
 import { json } from "@remix-run/server-runtime";
 import {
-  BatchTriggerTaskResponse,
   BatchTriggerTaskV2RequestBody,
   BatchTriggerTaskV2Response,
   generateJWT,
 } from "@trigger.dev/core/v3";
 import { env } from "~/env.server";
+import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { determineEngineVersion } from "~/v3/engineVersion.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
 import {
   BatchProcessingStrategy,
   BatchTriggerV2Service,
 } from "~/v3/services/batchTriggerV2.server";
-import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
-import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
-import { logger } from "~/services/logger.server";
-import { z } from "zod";
+import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 
 const { action, loader } = createActionApiRoute(
   {
@@ -87,7 +87,11 @@ const { action, loader } = createActionApiRoute(
       resolveIdempotencyKeyTTL(idempotencyKeyTTL) ??
       new Date(Date.now() + 24 * 60 * 60 * 1000 * 30);
 
-    const service = new BatchTriggerV2Service(batchProcessingStrategy ?? undefined);
+    const version = await determineEngineVersion({ environment: authentication.environment });
+    const service =
+      version === "V1"
+        ? new BatchTriggerV2Service(batchProcessingStrategy ?? undefined)
+        : new BatchTriggerV3Service(batchProcessingStrategy ?? undefined);
 
     try {
       const batch = await service.call(authentication.environment, body, {

apps/webapp/app/services/worker.server.ts

@@ -56,6 +56,10 @@ import {
 } from "~/v3/services/cancelDevSessionRuns.server";
 import { logger } from "./logger.server";
 import { BatchProcessingOptions, BatchTriggerV2Service } from "~/v3/services/batchTriggerV2.server";
+import {
+  BatchProcessingOptions as BatchProcessingOptionsV3,
+  BatchTriggerV3Service,
+} from "~/v3/services/batchTriggerV3.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -199,6 +203,7 @@ const workerCatalog = {
   }),
   "v3.cancelDevSessionRuns": CancelDevSessionRunsServiceOptions,
   "v3.processBatchTaskRun": BatchProcessingOptions,
+  "v3.processBatchTaskRunV3": BatchProcessingOptionsV3,
 };
 
 const executionWorkerCatalog = {
@@ -735,6 +740,15 @@ function getWorkerQueue() {
         handler: async (payload, job) => {
           const service = new BatchTriggerV2Service(payload.strategy);
 
+          await service.processBatchTaskRun(payload);
+        },
+      },
+      "v3.processBatchTaskRunV3": {
+        priority: 0,
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new BatchTriggerV3Service(payload.strategy);
+
           await service.processBatchTaskRun(payload);
         },
       },

apps/webapp/app/v3/services/batchTriggerV2.server.ts

@@ -49,6 +49,9 @@ export type BatchTriggerTaskServiceOptions = {
   oneTimeUseToken?: string;
 };
 
+/**
+ * Larger batches, used in Run Engine v1
+ */
 export class BatchTriggerV2Service extends BaseService {
   private _batchProcessingStrategy: BatchProcessingStrategy;
 
@@ -787,7 +790,8 @@ export class BatchTriggerV2Service extends BaseService {
         batchId: batch.friendlyId,
         skipChecks: true,
         runFriendlyId: task.runFriendlyId,
-      }
+      },
+      "V1"
     );
 
     if (!run) {