Add new batch methods for triggering multiple different tasks in a si…

…ngle batch

.changeset/perfect-onions-call.md

@@ -0,0 +1,80 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Added new batch.trigger and batch.triggerByTask methods that allows triggering multiple different tasks in a single batch:
+
+```ts
+import { batch } from '@trigger.dev/sdk/v3';
+import type { myTask1, myTask2 } from './trigger/tasks';
+
+// Somewhere in your backend code
+const response = await batch.trigger<typeof myTask1 | typeof myTask2>([
+  { id: 'task1', payload: { foo: 'bar' } },
+  { id: 'task2', payload: { baz: 'qux' } },
+]);
+
+for (const run of response.runs) {
+  if (run.ok) {
+    console.log(run.output);
+  } else {
+    console.error(run.error);
+  }
+}
+```
+
+Or if you are inside of a task, you can use `triggerByTask`:
+
+```ts
+import { batch, task, runs } from '@trigger.dev/sdk/v3';
+
+export const myParentTask = task({
+  id: 'myParentTask',
+  run: async () => {
+    const response = await batch.triggerByTask([
+      { task: myTask1, payload: { foo: 'bar' } },
+      { task: myTask2, payload: { baz: 'qux' } },
+    ]);
+
+    const run1 = await runs.retrieve(response.runs[0]);
+    console.log(run1.output) // typed as { foo: string }
+
+    const run2 = await runs.retrieve(response.runs[1]);
+    console.log(run2.output) // typed as { baz: string }
+
+    const response2 = await batch.triggerByTaskAndWait([
+      { task: myTask1, payload: { foo: 'bar' } },
+      { task: myTask2, payload: { baz: 'qux' } },
+    ]);
+
+    if (response2.runs[0].ok) {
+      console.log(response2.runs[0].output) // typed as { foo: string }
+    }
+
+    if (response2.runs[1].ok) {
+      console.log(response2.runs[1].output) // typed as { baz: string }
+    }
+  }
+});
+
+export const myTask1 = task({
+  id: 'myTask1',
+  run: async () => {
+    return {
+      foo: 'bar'
+    }
+  }
+});
+
+export const myTask2 = task({
+  id: 'myTask2',
+  run: async () => {
+    return {
+      baz: 'qux'
+    }
+  }
+});
+
+```

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -1038,13 +1038,15 @@ class SharedQueueTasks {
         id: attempt.taskRun.friendlyId,
         output: attempt.output ?? undefined,
         outputType: attempt.outputType,
+        taskIdentifier: attempt.taskRun.taskIdentifier,
       };
       return success;
     } else {
       const failure: TaskRunFailedExecutionResult = {
         ok,
         id: attempt.taskRun.friendlyId,
         error: attempt.error as TaskRunError,
+        taskIdentifier: attempt.taskRun.taskIdentifier,
       };
       return failure;
     }

packages/cli-v3/src/dev/backgroundWorker.ts

@@ -498,6 +498,7 @@ export class BackgroundWorker {
         ok: false,
         retry: undefined,
         error: TaskRunProcess.parseExecuteError(e),
+        taskIdentifier: payload.execution.task.id,
       };
     }
   }

packages/cli-v3/src/entryPoints/deploy-run-worker.ts

@@ -216,6 +216,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
+            taskIdentifier: execution.task.id,
           },
         });
 
@@ -246,6 +247,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
+              taskIdentifier: execution.task.id,
             },
           });
 
@@ -277,6 +279,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
+              taskIdentifier: execution.task.id,
             },
           });
 
@@ -303,6 +306,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
+              taskIdentifier: execution.task.id,
             },
           });
 
@@ -357,6 +361,7 @@ const zodIpc = new ZodIpcConnection({
                   usage: {
                     durationMs: usageSample.cpuTime,
                   },
+                  taskIdentifier: execution.task.id,
                 },
               });
             }
@@ -380,6 +385,7 @@ const zodIpc = new ZodIpcConnection({
                 usage: {
                   durationMs: usageSample.cpuTime,
                 },
+                taskIdentifier: execution.task.id,
               },
             });
           }
@@ -402,6 +408,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
+            taskIdentifier: execution.task.id,
           },
         });
       }

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -194,6 +194,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
+            taskIdentifier: execution.task.id,
           },
         });
 
@@ -222,6 +223,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
+            taskIdentifier: execution.task.id,
           },
         });
 
@@ -247,6 +249,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
+            taskIdentifier: execution.task.id,
           },
         });
 
@@ -273,6 +276,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
+            taskIdentifier: execution.task.id,
           },
         });
 
@@ -324,6 +328,7 @@ const zodIpc = new ZodIpcConnection({
                 usage: {
                   durationMs: usageSample.cpuTime,
                 },
+                taskIdentifier: execution.task.id,
               },
             });
           }
@@ -347,6 +352,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: usageSample.cpuTime,
               },
+              taskIdentifier: execution.task.id,
             },
           });
         }

packages/core/src/v3/types/tasks.ts

@@ -49,18 +49,24 @@ export class SubtaskUnwrapError extends Error {
   }
 }
 
-export class TaskRunPromise<T> extends Promise<TaskRunResult<T>> {
+export class TaskRunPromise<TIdentifier extends string, TOutput> extends Promise<
+  TaskRunResult<TIdentifier, TOutput>
+> {
   constructor(
     executor: (
-      resolve: (value: TaskRunResult<T> | PromiseLike<TaskRunResult<T>>) => void,
+      resolve: (
+        value:
+          | TaskRunResult<TIdentifier, TOutput>
+          | PromiseLike<TaskRunResult<TIdentifier, TOutput>>
+      ) => void,
       reject: (reason?: any) => void
     ) => void,
-    private readonly taskId: string
+    private readonly taskId: TIdentifier
   ) {
     super(executor);
   }
 
-  unwrap(): Promise<T> {
+  unwrap(): Promise<TOutput> {
     return this.then((result) => {
       if (result.ok) {
         return result.output;
@@ -415,38 +421,101 @@ export type RunHandleTaskIdentifier<TRunHandle> = TRunHandle extends RunHandle<
   ? TTaskIdentifier
   : never;
 
-export type TaskRunResult<TOutput = any> =
+export type TaskRunResult<TIdentifier extends string, TOutput = any> =
   | {
       ok: true;
       id: string;
-      taskIdentifier: string;
+      taskIdentifier: TIdentifier;
       output: TOutput;
     }
   | {
       ok: false;
       id: string;
-      taskIdentifier: string;
+      taskIdentifier: TIdentifier;
       error: unknown;
     };
 
-export type BatchResult<TOutput = any> = {
+export type AnyTaskRunResult = TaskRunResult<string, any>;
+
+export type TaskRunResultFromTask<TTask extends AnyTask> = TTask extends Task<
+  infer TIdentifier,
+  any,
+  infer TOutput
+>
+  ? TaskRunResult<TIdentifier, TOutput>
+  : never;
+
+export type BatchResult<TIdentifier extends string, TOutput = any> = {
+  id: string;
+  runs: TaskRunResult<TIdentifier, TOutput>[];
+};
+
+export type BatchByIdResult<TTask extends AnyTask> = {
+  id: string;
+  runs: Array<TaskRunResultFromTask<TTask>>;
+};
+
+export type BatchByTaskResult<TTasks extends readonly AnyTask[]> = {
   id: string;
-  runs: TaskRunResult<TOutput>[];
+  runs: {
+    [K in keyof TTasks]: TaskRunResultFromTask<TTasks[K]>;
+  };
 };
 
+/**
+ * A BatchRunHandle can be used to retrieve the runs of a batch trigger in a typesafe manner.
+ */
+// export type BatchTasksRunHandle<TTasks extends readonly AnyTask[]> = BrandedRun<
+//   {
+//     batchId: string;
+//     isCached: boolean;
+//     idempotencyKey?: string;
+//     runs: {
+//       [K in keyof TTasks]: BatchedRunHandle<
+//         TaskIdentifier<TTasks[K]>,
+//         TaskPayload<TTasks[K]>,
+//         TaskOutput<TTasks[K]>
+//       >;
+//     };
+//     publicAccessToken: string;
+//   },
+//   any,
+//   any
+// >;
+
+export type BatchTasksResult<TTasks extends readonly AnyTask[]> = BatchTasksRunHandle<TTasks>;
+
 export type BatchItem<TInput> = { payload: TInput; options?: TriggerOptions };
 
 export type BatchTriggerAndWaitItem<TInput> = {
   payload: TInput;
   options?: TriggerAndWaitOptions;
 };
 
-export type BatchAllItem<TRunTypes extends AnyRunTypes> = {
-  task: TRunTypes["taskIdentifier"];
+export type BatchByIdItem<TRunTypes extends AnyRunTypes> = {
+  id: TRunTypes["taskIdentifier"];
+  payload: TRunTypes["payload"];
+  options?: TriggerOptions;
+};
+
+export type BatchByIdAndWaitItem<TRunTypes extends AnyRunTypes> = {
+  id: TRunTypes["taskIdentifier"];
   payload: TRunTypes["payload"];
+  options?: TriggerAndWaitOptions;
+};
+
+export type BatchByTaskItem<TTask extends AnyTask> = {
+  task: TTask;
+  payload: TaskPayload<TTask>;
   options?: TriggerOptions;
 };
 
+export type BatchByTaskAndWaitItem<TTask extends AnyTask> = {
+  task: TTask;
+  payload: TaskPayload<TTask>;
+  options?: TriggerAndWaitOptions;
+};
+
 export interface Task<TIdentifier extends string, TInput = void, TOutput = any> {
   /**
    * The id of the task.
@@ -497,7 +566,10 @@ export interface Task<TIdentifier extends string, TInput = void, TOutput = any>
    * }
    * ```
    */
-  triggerAndWait: (payload: TInput, options?: TriggerAndWaitOptions) => TaskRunPromise<TOutput>;
+  triggerAndWait: (
+    payload: TInput,
+    options?: TriggerAndWaitOptions
+  ) => TaskRunPromise<TIdentifier, TOutput>;
 
   /**
    * Batch trigger multiple task runs with the given payloads, and wait for the results. Returns the results of the task runs.
@@ -521,7 +593,7 @@ export interface Task<TIdentifier extends string, TInput = void, TOutput = any>
    */
   batchTriggerAndWait: (
     items: Array<BatchTriggerAndWaitItem<TInput>>
-  ) => Promise<BatchResult<TOutput>>;
+  ) => Promise<BatchResult<TIdentifier, TOutput>>;
 }
 
 export interface TaskWithSchema<
@@ -758,3 +830,27 @@ export type RunHandleFromTypes<TRunTypes extends AnyRunTypes> = RunHandle<
 export type BatchRunHandleFromTypes<TRunTypes extends AnyRunTypes> = TRunTypes extends AnyRunTypes
   ? BatchRunHandle<TRunTypes["taskIdentifier"], TRunTypes["payload"], TRunTypes["output"]>
   : never;
+
+/**
+ * A BatchRunHandle can be used to retrieve the runs of a batch trigger in a typesafe manner.
+ */
+export type BatchTasksRunHandle<TTasks extends readonly AnyTask[]> = BrandedRun<
+  {
+    batchId: string;
+    isCached: boolean;
+    idempotencyKey?: string;
+    runs: {
+      [K in keyof TTasks]: BatchedRunHandle<
+        TaskIdentifier<TTasks[K]>,
+        TaskPayload<TTasks[K]>,
+        TaskOutput<TTasks[K]>
+      >;
+    };
+    publicAccessToken: string;
+  },
+  any,
+  any
+>;
+
+export type BatchTasksRunHandleFromTypes<TTasks extends readonly AnyTask[]> =
+  BatchTasksRunHandle<TTasks>;