Extract common trigger code into internal functions and add a tasks.batchTriggerAndWait function

.changeset/large-seahorses-cheat.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Extract common trigger code into internal functions and add a tasks.batchTriggerAndWait function

docs/v3/triggering.mdx

@@ -3,16 +3,18 @@ title: "Triggering"
 description: "Tasks need to be triggered to run."
 ---
 
-There are currently six ways you can trigger tasks:
+These are the different ways you can trigger tasks:
 
 | Function                         | Where does this work? | What it does                                                                                                                       |
 | -------------------------------- | --------------------- | ---------------------------------------------------------------------------------------------------------------------------------- |
 | `yourTask.trigger()`             | Anywhere              | Triggers a task and gets a handle you can use to monitor and manage the run. It does not wait for the result.                      |
 | `yourTask.batchTrigger()`        | Anywhere              | Triggers a task multiple times and gets a handle you can use to monitor and manage the runs. It does not wait for the results.     |
-| `yourTask.triggerAndWait()`      | Inside a task         | Triggers a task and then waits until it's complete. You get the result data to continue with.                                      |
+| `yourTask.triggerAndWait()`      | Inside task           | Triggers a task and then waits until it's complete. You get the result data to continue with.                                      |
-| `yourTask.batchTriggerAndWait()` | Inside a task         | Triggers a task multiple times in parallel and then waits until they're all complete. You get the resulting data to continue with. |
+| `yourTask.batchTriggerAndWait()` | Inside task           | Triggers a task multiple times in parallel and then waits until they're all complete. You get the resulting data to continue with. |
-| `tasks.trigger()`                | Outside of a task     | Triggers a task and gets a handle you can use to fetch and manage the run.                                                         |
+| `tasks.trigger()`                | Anywhere              | Triggers a task and gets a handle you can use to fetch and manage the run.                                                         |
-| `tasks.batchTrigger()`           | Outside of a task     | Triggers a task multiple times and gets a handle you can use to fetch and manage the runs.                                         |
+| `tasks.batchTrigger()`           | Anywhere              | Triggers a task multiple times and gets a handle you can use to fetch and manage the runs.                                         |
+| `tasks.triggerAndWait()`         | Inside task           | Triggers a task and then waits until it's complete. You get the result data to continue with                                       |
+| `tasks.batchTriggerAndWait()`    | Inside task           | Triggers a task multiple times in parallel and then waits until they're all complete. You get the resulting data to continue with. |
 
 Additionally, [scheduled tasks](/v3/tasks-scheduled) get automatically triggered on their schedule and [webhooks](/v3/tasks-webhooks) when receiving a webhook.
 

packages/trigger-sdk/src/v3/shared.ts

@@ -479,346 +479,49 @@ export function createTask<
   const task: Task<TIdentifier, TInput, TOutput> = {
     id: params.id,
     trigger: async (payload, options) => {
-      const apiClient = apiClientManager.client;
-
-      if (!apiClient) {
-        throw apiClientMissingError();
-      }
-
       const taskMetadata = taskCatalog.getTaskMetadata(params.id);
 
-      const payloadPacket = await stringifyIO(payload);
+      return await trigger_internal<TInput, TOutput>(
-
+        taskMetadata && taskMetadata.exportName
-      const handle = await apiClient.triggerTask(
+          ? `${taskMetadata.exportName}.trigger()`
+          : `trigger()`,
         params.id,
-        {
+        payload,
-          payload: payloadPacket.data,
+        options
-          options: {
-            queue: options?.queue ?? params.queue,
-            concurrencyKey: options?.concurrencyKey,
-            test: taskContext.ctx?.run.isTest,
-            payloadType: payloadPacket.dataType,
-            idempotencyKey: await makeKey(options?.idempotencyKey),
-            delay: options?.delay,
-            ttl: options?.ttl,
-            tags: options?.tags,
-            maxAttempts: options?.maxAttempts,
-          },
-        },
-        { spanParentAsLink: true },
-        {
-          name: taskMetadata ? `${taskMetadata.exportName}.trigger()` : `trigger()`,
-          tracer,
-          icon: "trigger",
-          attributes: {
-            [SEMATTRS_MESSAGING_OPERATION]: "publish",
-            ["messaging.client_id"]: taskContext.worker?.id,
-            [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
-            [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
-            ...accessoryAttributes({
-              items: [
-                {
-                  text: params.id,
-                  variant: "normal",
-                },
-              ],
-              style: "codepath",
-            }),
-          },
-          onResponseBody: (body, span) => {
-            body &&
-              typeof body === "object" &&
-              !Array.isArray(body) &&
-              "id" in body &&
-              typeof body.id === "string" &&
-              span.setAttribute("messaging.message.id", body.id);
-          },
-        }
       );
-
-      return handle as RunHandle<TOutput>;
     },
     batchTrigger: async (items) => {
-      const apiClient = apiClientManager.client;
-
-      if (!apiClient) {
-        throw apiClientMissingError();
-      }
-
       const taskMetadata = taskCatalog.getTaskMetadata(params.id);
 
-      const response = await apiClient.batchTriggerTask(
+      return await batchTrigger_internal<TInput, TOutput>(
+        taskMetadata && taskMetadata.exportName
+          ? `${taskMetadata.exportName}.batchTrigger()`
+          : `batchTrigger()`,
         params.id,
-        {
+        items
-          items: await Promise.all(
-            items.map(async (item) => {
-              const payloadPacket = await stringifyIO(item.payload);
-
-              return {
-                payload: payloadPacket.data,
-                options: {
-                  queue: item.options?.queue ?? params.queue,
-                  concurrencyKey: item.options?.concurrencyKey,
-                  test: taskContext.ctx?.run.isTest,
-                  payloadType: payloadPacket.dataType,
-                  idempotencyKey: await makeKey(item.options?.idempotencyKey),
-                  delay: item.options?.delay,
-                  ttl: item.options?.ttl,
-                  tags: item.options?.tags,
-                  maxAttempts: item.options?.maxAttempts,
-                },
-              };
-            })
-          ),
-        },
-        { spanParentAsLink: true },
-        {
-          name: taskMetadata ? `${taskMetadata.exportName}.batchTrigger()` : `batchTrigger()`,
-          icon: "trigger",
-          tracer,
-          attributes: {
-            [SEMATTRS_MESSAGING_OPERATION]: "publish",
-            ["messaging.batch.message_count"]: items.length,
-            ["messaging.client_id"]: taskContext.worker?.id,
-            [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
-            [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
-            ...accessoryAttributes({
-              items: [
-                {
-                  text: params.id,
-                  variant: "normal",
-                },
-              ],
-              style: "codepath",
-            }),
-          },
-        }
       );
-
-      const handle = {
-        batchId: response.batchId,
-        runs: response.runs.map((id) => ({ id })),
-      };
-
-      return handle as BatchRunHandle<TOutput>;
     },
     triggerAndWait: async (payload, options) => {
-      const ctx = taskContext.ctx;
-
-      if (!ctx) {
-        throw new Error("triggerAndWait can only be used from inside a task.run()");
-      }
-
-      const apiClient = apiClientManager.client;
-
-      if (!apiClient) {
-        throw apiClientMissingError();
-      }
-
       const taskMetadata = taskCatalog.getTaskMetadata(params.id);
 
-      const payloadPacket = await stringifyIO(payload);
+      return await triggerAndWait_internal<TInput, TOutput>(
-
+        taskMetadata && taskMetadata.exportName
-      return await tracer.startActiveSpan(
+          ? `${taskMetadata.exportName}.triggerAndWait()`
-        taskMetadata ? `${taskMetadata.exportName}.triggerAndWait()` : `triggerAndWait()`,
+          : `triggerAndWait()`,
-        async (span) => {
+        params.id,
-          const response = await apiClient.triggerTask(params.id, {
+        payload,
-            payload: payloadPacket.data,
+        options
-            options: {
-              dependentAttempt: ctx.attempt.id,
-              lockToVersion: taskContext.worker?.version, // Lock to current version because we're waiting for it to finish
-              queue: options?.queue ?? params.queue,
-              concurrencyKey: options?.concurrencyKey,
-              test: taskContext.ctx?.run.isTest,
-              payloadType: payloadPacket.dataType,
-              idempotencyKey: await makeKey(options?.idempotencyKey),
-              delay: options?.delay,
-              ttl: options?.ttl,
-              tags: options?.tags,
-              maxAttempts: options?.maxAttempts,
-            },
-          });
-
-          span.setAttribute("messaging.message.id", response.id);
-
-          if (options?.idempotencyKey) {
-            // If an idempotency key is provided, we can check if the result is already available
-            const result = await apiClient.getRunResult(response.id);
-
-            if (result) {
-              logger.log(
-                `Result reused from previous task run with idempotency key '${options.idempotencyKey}'.`,
-                {
-                  runId: response.id,
-                  idempotencyKey: options.idempotencyKey,
-                }
-              );
-
-              return await handleTaskRunExecutionResult<TOutput>(result);
-            }
-          }
-
-          const result = await runtime.waitForTask({
-            id: response.id,
-            ctx,
-          });
-
-          return await handleTaskRunExecutionResult<TOutput>(result);
-        },
-        {
-          kind: SpanKind.PRODUCER,
-          attributes: {
-            [SemanticInternalAttributes.STYLE_ICON]: "trigger",
-            [SEMATTRS_MESSAGING_OPERATION]: "publish",
-            ["messaging.client_id"]: taskContext.worker?.id,
-            [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
-            [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
-            ...accessoryAttributes({
-              items: [
-                {
-                  text: params.id,
-                  variant: "normal",
-                },
-              ],
-              style: "codepath",
-            }),
-          },
-        }
       );
     },
     batchTriggerAndWait: async (items) => {
-      const ctx = taskContext.ctx;
-
-      if (!ctx) {
-        throw new Error("batchTriggerAndWait can only be used from inside a task.run()");
-      }
-
-      const apiClient = apiClientManager.client;
-
-      if (!apiClient) {
-        throw apiClientMissingError();
-      }
-
       const taskMetadata = taskCatalog.getTaskMetadata(params.id);
 
-      return await tracer.startActiveSpan(
+      return await batchTriggerAndWait_internal<TInput, TOutput>(
-        taskMetadata ? `${taskMetadata.exportName}.batchTriggerAndWait()` : `batchTriggerAndWait()`,
+        taskMetadata && taskMetadata.exportName
-        async (span) => {
+          ? `${taskMetadata.exportName}.batchTriggerAndWait()`
-          const response = await apiClient.batchTriggerTask(params.id, {
+          : `batchTriggerAndWait()`,
-            items: await Promise.all(
+        params.id,
-              items.map(async (item) => {
+        items
-                const payloadPacket = await stringifyIO(item.payload);
-
-                return {
-                  payload: payloadPacket.data,
-                  options: {
-                    lockToVersion: taskContext.worker?.version,
-                    queue: item.options?.queue ?? params.queue,
-                    concurrencyKey: item.options?.concurrencyKey,
-                    test: taskContext.ctx?.run.isTest,
-                    payloadType: payloadPacket.dataType,
-                    idempotencyKey: await makeKey(item.options?.idempotencyKey),
-                    delay: item.options?.delay,
-                    ttl: item.options?.ttl,
-                    tags: item.options?.tags,
-                    maxAttempts: item.options?.maxAttempts,
-                  },
-                };
-              })
-            ),
-            dependentAttempt: ctx.attempt.id,
-          });
-
-          span.setAttribute("messaging.message.id", response.batchId);
-
-          const getBatchResults = async (): Promise<BatchTaskRunExecutionResult> => {
-            // We need to check if the results are already available, but only if any of the items options has an idempotency key
-            const hasIdempotencyKey = items.some((item) => item.options?.idempotencyKey);
-
-            if (hasIdempotencyKey) {
-              const results = await apiClient.getBatchResults(response.batchId);
-
-              if (results) {
-                return results;
-              }
-            }
-
-            return {
-              id: response.batchId,
-              items: [],
-            };
-          };
-
-          const existingResults = await getBatchResults();
-
-          const incompleteRuns = response.runs.filter(
-            (runId) => !existingResults.items.some((item) => item.id === runId)
-          );
-
-          if (incompleteRuns.length === 0) {
-            logger.log(
-              `Results reused from previous task runs because of the provided idempotency keys.`
-            );
-
-            // All runs are already completed
-            const runs = await handleBatchTaskRunExecutionResult<TOutput>(existingResults.items);
-
-            return {
-              id: existingResults.id,
-              runs,
-            };
-          }
-
-          const result = await runtime.waitForBatch({
-            id: response.batchId,
-            runs: incompleteRuns,
-            ctx,
-          });
-
-          // Combine the already completed runs with the newly completed runs, ordered by the original order
-          const combinedItems: BatchTaskRunExecutionResult["items"] = [];
-
-          for (const runId of response.runs) {
-            const existingItem = existingResults.items.find((item) => item.id === runId);
-
-            if (existingItem) {
-              combinedItems.push(existingItem);
-            } else {
-              const newItem = result.items.find((item) => item.id === runId);
-
-              if (newItem) {
-                combinedItems.push(newItem);
-              }
-            }
-          }
-
-          const runs = await handleBatchTaskRunExecutionResult<TOutput>(combinedItems);
-
-          return {
-            id: result.id,
-            runs,
-          };
-        },
-        {
-          kind: SpanKind.PRODUCER,
-          attributes: {
-            [SEMATTRS_MESSAGING_OPERATION]: "publish",
-            ["messaging.batch.message_count"]: items.length,
-            ["messaging.client_id"]: taskContext.worker?.id,
-            [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
-            [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
-            [SemanticInternalAttributes.STYLE_ICON]: "trigger",
-            ...accessoryAttributes({
-              items: [
-                {
-                  text: params.id,
-                  variant: "normal",
-                },
-              ],
-              style: "codepath",
-            }),
-          },
-        }
       );
     },
   };
@@ -865,7 +568,132 @@ export async function trigger<TTask extends AnyTask>(
   payload: TaskPayload<TTask>,
   options?: TaskRunOptions,
   requestOptions?: ApiRequestOptions
-): Promise<TaskOutputHandle<TTask>> {
+): Promise<RunHandle<TaskOutput<TTask>>> {
+  return await trigger_internal<TaskPayload<TTask>, TaskOutput<TTask>>(
+    "tasks.trigger()",
+    id,
+    payload,
+    options,
+    requestOptions
+  );
+}
+
+/**
+ * Trigger a task with the given payload, and wait for the result. Returns the result of the task run
+ * @param id - The id of the task to trigger
+ * @param payload
+ * @param options - Options for the task run
+ * @returns TaskRunResult
+ * @example
+ * ```ts
+ * import { tasks } from "@trigger.dev/sdk/v3";
+ * const result = await tasks.triggerAndWait("my-task", { foo: "bar" });
+ *
+ * if (result.ok) {
+ *  console.log(result.output);
+ * } else {
+ *  console.error(result.error);
+ * }
+ * ```
+ */
+export async function triggerAndWait<TTask extends AnyTask>(
+  id: TaskIdentifier<TTask>,
+  payload: TaskPayload<TTask>,
+  options?: TaskRunOptions,
+  requestOptions?: ApiRequestOptions
+): Promise<TaskRunResult<TaskOutput<TTask>>> {
+  return await triggerAndWait_internal<TaskPayload<TTask>, TaskOutput<TTask>>(
+    "tasks.triggerAndWait()",
+    id,
+    payload,
+    options,
+    requestOptions
+  );
+}
+
+/**
+ * Batch trigger multiple task runs with the given payloads, and wait for the results. Returns the results of the task runs.
+ * @param id - The id of the task to trigger
+ * @param items
+ * @returns BatchResult
+ * @example
+ *
+ * ```ts
+ * import { tasks } from "@trigger.dev/sdk/v3";
+ *
+ * const result = await tasks.batchTriggerAndWait("my-task", [
+ *  { payload: { foo: "bar" } },
+ *  { payload: { foo: "baz" } },
+ * ]);
+ *
+ * for (const run of result.runs) {
+ *  if (run.ok) {
+ *    console.log(run.output);
+ *  } else {
+ *    console.error(run.error);
+ *  }
+ * }
+ * ```
+ */
+export async function batchTriggerAndWait<TTask extends AnyTask>(
+  id: TaskIdentifier<TTask>,
+  items: Array<BatchItem<TaskPayload<TTask>>>,
+  requestOptions?: ApiRequestOptions
+): Promise<BatchResult<TaskOutput<TTask>>> {
+  return await batchTriggerAndWait_internal<TaskPayload<TTask>, TaskOutput<TTask>>(
+    "tasks.batchTriggerAndWait()",
+    id,
+    items,
+    requestOptions
+  );
+}
+
+/**
+ * Trigger a task by its identifier with the given payload and poll until the run is completed.
+ *
+ * @example
+ *
+ * ```ts
+ * import { tasks, runs } from "@trigger.dev/sdk/v3";
+ * import type { myTask } from "./myTasks"; // Import just the type of the task
+ *
+ * const run = await tasks.triggerAndPoll<typeof myTask>("my-task", { foo: "bar" }); // The id and payload are fully typesafe
+ * console.log(run.output) // The output is also fully typed
+ * ```
+ *
+ * @returns {Run} The completed run, either successful or failed.
+ */
+export async function triggerAndPoll<TTask extends AnyTask>(
+  id: TaskIdentifier<TTask>,
+  payload: TaskPayload<TTask>,
+  options?: TaskRunOptions & PollOptions,
+  requestOptions?: ApiRequestOptions
+): Promise<RetrieveRunResult<RunHandle<TaskOutput<TTask>>>> {
+  const handle = await trigger(id, payload, options, requestOptions);
+
+  return runs.poll(handle, options, requestOptions);
+}
+
+export async function batchTrigger<TTask extends AnyTask>(
+  id: TaskIdentifier<TTask>,
+  items: Array<BatchItem<TaskPayload<TTask>>>,
+  requestOptions?: ApiRequestOptions
+): Promise<BatchRunHandle<TTask>> {
+  return await batchTrigger_internal<TaskPayload<TTask>, TaskOutput<TTask>>(
+    "tasks.batchTrigger()",
+    id,
+    items,
+    requestOptions
+  );
+}
+
+async function trigger_internal<TPayload, TOutput>(
+  name: string,
+  id: string,
+  payload: TPayload,
+  options?: TaskRunOptions,
+  requestOptions?: ApiRequestOptions
+): Promise<RunHandle<TOutput>> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
@@ -894,7 +722,7 @@ export async function trigger<TTask extends AnyTask>(
       spanParentAsLink: true,
     },
     {
-      name: `tasks.trigger()`,
+      name,
       tracer,
       icon: "trigger",
       attributes: {
@@ -923,19 +751,87 @@ export async function trigger<TTask extends AnyTask>(
     }
   );
 
-  return handle as TaskOutputHandle<TTask>;
+  return handle as RunHandle<TOutput>;
 }
 
-export async function triggerAndWait<TTask extends AnyTask>(
+async function batchTrigger_internal<TPayload, TOutput>(
-  id: TaskIdentifier<TTask>,
+  name: string,
-  payload: TaskPayload<TTask>,
+  id: string,
+  items: Array<BatchItem<TPayload>>,
+  requestOptions?: ApiRequestOptions
+): Promise<BatchRunHandle<TOutput>> {
+  const apiClient = apiClientManager.client;
+
+  if (!apiClient) {
+    throw apiClientMissingError();
+  }
+
+  const response = await apiClient.batchTriggerTask(
+    id,
+    {
+      items: await Promise.all(
+        items.map(async (item) => {
+          const payloadPacket = await stringifyIO(item.payload);
+
+          return {
+            payload: payloadPacket.data,
+            options: {
+              queue: item.options?.queue,
+              concurrencyKey: item.options?.concurrencyKey,
+              test: taskContext.ctx?.run.isTest,
+              payloadType: payloadPacket.dataType,
+              idempotencyKey: await makeKey(item.options?.idempotencyKey),
+              delay: item.options?.delay,
+              ttl: item.options?.ttl,
+              tags: item.options?.tags,
+              maxAttempts: item.options?.maxAttempts,
+            },
+          };
+        })
+      ),
+    },
+    { spanParentAsLink: true },
+    {
+      name,
+      tracer,
+      icon: "trigger",
+      attributes: {
+        [SEMATTRS_MESSAGING_OPERATION]: "publish",
+        ["messaging.client_id"]: taskContext.worker?.id,
+        [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
+        ...accessoryAttributes({
+          items: [
+            {
+              text: id,
+              variant: "normal",
+            },
+          ],
+          style: "codepath",
+        }),
+      },
+      ...requestOptions,
+    }
+  );
+
+  const handle = {
+    batchId: response.batchId,
+    runs: response.runs.map((id) => ({ id })),
+  };
+
+  return handle as BatchRunHandle<TOutput>;
+}
+
+async function triggerAndWait_internal<TPayload, TOutput>(
+  name: string,
+  id: string,
+  payload: TPayload,
   options?: TaskRunOptions,
   requestOptions?: ApiRequestOptions
-): Promise<TaskRunResult<TaskOutput<TTask>>> {
+): Promise<TaskRunResult<TOutput>> {
   const ctx = taskContext.ctx;
 
   if (!ctx) {
-    throw new Error("tasks.triggerAndWait can only be used from inside a task.run()");
+    throw new Error("triggerAndWait can only be used from inside a task.run()");
   }
 
   const apiClient = apiClientManager.client;
@@ -947,7 +843,7 @@ export async function triggerAndWait<TTask extends AnyTask>(
   const payloadPacket = await stringifyIO(payload);
 
   return await tracer.startActiveSpan(
-    "tasks.triggerAndWait()",
+    name,
     async (span) => {
       const response = await apiClient.triggerTask(
         id,
@@ -986,7 +882,7 @@ export async function triggerAndWait<TTask extends AnyTask>(
             }
           );
 
-          return await handleTaskRunExecutionResult<TaskOutput<TTask>>(result);
+          return await handleTaskRunExecutionResult<TOutput>(result);
         }
       }
 
@@ -995,7 +891,7 @@ export async function triggerAndWait<TTask extends AnyTask>(
         ctx,
       });
 
-      return await handleTaskRunExecutionResult<TaskOutput<TTask>>(result);
+      return await handleTaskRunExecutionResult<TOutput>(result);
     },
     {
       kind: SpanKind.PRODUCER,
@@ -1019,75 +915,135 @@ export async function triggerAndWait<TTask extends AnyTask>(
   );
 }
 
-/**
+async function batchTriggerAndWait_internal<TPayload, TOutput>(
- * Trigger a task by its identifier with the given payload and poll until the run is completed.
+  name: string,
- *
+  id: string,
- * @example
+  items: Array<BatchItem<TPayload>>,
- *
- * ```ts
- * import { tasks, runs } from "@trigger.dev/sdk/v3";
- * import type { myTask } from "./myTasks"; // Import just the type of the task
- *
- * const run = await tasks.triggerAndPoll<typeof myTask>("my-task", { foo: "bar" }); // The id and payload are fully typesafe
- * console.log(run.output) // The output is also fully typed
- * ```
- *
- * @returns {Run} The completed run, either successful or failed.
- */
-export async function triggerAndPoll<TTask extends AnyTask>(
-  id: TaskIdentifier<TTask>,
-  payload: TaskPayload<TTask>,
-  options?: TaskRunOptions & PollOptions,
   requestOptions?: ApiRequestOptions
-): Promise<RetrieveRunResult<TaskOutputHandle<TTask>>> {
+): Promise<BatchResult<TOutput>> {
-  const handle = await trigger(id, payload, options, requestOptions);
+  const ctx = taskContext.ctx;
 
-  return runs.poll(handle, options, requestOptions);
+  if (!ctx) {
-}
+    throw new Error("batchTriggerAndWait can only be used from inside a task.run()");
+  }
 
-export async function batchTrigger<TTask extends AnyTask>(
-  id: TaskIdentifier<TTask>,
-  items: Array<BatchItem<TaskPayload<TTask>>>,
-  requestOptions?: ApiRequestOptions
-): Promise<TaskBatchOutputHandle<TTask>> {
   const apiClient = apiClientManager.client;
 
   if (!apiClient) {
     throw apiClientMissingError();
   }
 
-  const response = await apiClient.batchTriggerTask(
+  return await tracer.startActiveSpan(
-    id,
+    name,
-    {
+    async (span) => {
-      items: await Promise.all(
+      const response = await apiClient.batchTriggerTask(
-        items.map(async (item) => {
+        id,
-          const payloadPacket = await stringifyIO(item.payload);
+        {
+          items: await Promise.all(
+            items.map(async (item) => {
+              const payloadPacket = await stringifyIO(item.payload);
 
-          return {
+              return {
-            payload: payloadPacket.data,
+                payload: payloadPacket.data,
-            options: {
+                options: {
-              queue: item.options?.queue,
+                  lockToVersion: taskContext.worker?.version,
-              concurrencyKey: item.options?.concurrencyKey,
+                  queue: item.options?.queue,
-              test: taskContext.ctx?.run.isTest,
+                  concurrencyKey: item.options?.concurrencyKey,
-              payloadType: payloadPacket.dataType,
+                  test: taskContext.ctx?.run.isTest,
-              idempotencyKey: await makeKey(item.options?.idempotencyKey),
+                  payloadType: payloadPacket.dataType,
-              delay: item.options?.delay,
+                  idempotencyKey: await makeKey(item.options?.idempotencyKey),
-              ttl: item.options?.ttl,
+                  delay: item.options?.delay,
-              tags: item.options?.tags,
+                  ttl: item.options?.ttl,
-              maxAttempts: item.options?.maxAttempts,
+                  tags: item.options?.tags,
-            },
+                  maxAttempts: item.options?.maxAttempts,
-          };
+                },
-        })
+              };
-      ),
+            })
+          ),
+          dependentAttempt: ctx.attempt.id,
+        },
+        {},
+        requestOptions
+      );
+
+      span.setAttribute("messaging.message.id", response.batchId);
+
+      const getBatchResults = async (): Promise<BatchTaskRunExecutionResult> => {
+        // We need to check if the results are already available, but only if any of the items options has an idempotency key
+        const hasIdempotencyKey = items.some((item) => item.options?.idempotencyKey);
+
+        if (hasIdempotencyKey) {
+          const results = await apiClient.getBatchResults(response.batchId);
+
+          if (results) {
+            return results;
+          }
+        }
+
+        return {
+          id: response.batchId,
+          items: [],
+        };
+      };
+
+      const existingResults = await getBatchResults();
+
+      const incompleteRuns = response.runs.filter(
+        (runId) => !existingResults.items.some((item) => item.id === runId)
+      );
+
+      if (incompleteRuns.length === 0) {
+        logger.log(
+          `Results reused from previous task runs because of the provided idempotency keys.`
+        );
+
+        // All runs are already completed
+        const runs = await handleBatchTaskRunExecutionResult<TOutput>(existingResults.items);
+
+        return {
+          id: existingResults.id,
+          runs,
+        };
+      }
+
+      const result = await runtime.waitForBatch({
+        id: response.batchId,
+        runs: incompleteRuns,
+        ctx,
+      });
+
+      // Combine the already completed runs with the newly completed runs, ordered by the original order
+      const combinedItems: BatchTaskRunExecutionResult["items"] = [];
+
+      for (const runId of response.runs) {
+        const existingItem = existingResults.items.find((item) => item.id === runId);
+
+        if (existingItem) {
+          combinedItems.push(existingItem);
+        } else {
+          const newItem = result.items.find((item) => item.id === runId);
+
+          if (newItem) {
+            combinedItems.push(newItem);
+          }
+        }
+      }
+
+      const runs = await handleBatchTaskRunExecutionResult<TOutput>(combinedItems);
+
+      return {
+        id: result.id,
+        runs,
+      };
     },
-    { spanParentAsLink: true },
     {
-      name: `tasks.batchTrigger()`,
+      kind: SpanKind.PRODUCER,
-      tracer,
-      icon: "trigger",
       attributes: {
+        [SemanticInternalAttributes.STYLE_ICON]: "trigger",
+        ["messaging.batch.message_count"]: items.length,
         [SEMATTRS_MESSAGING_OPERATION]: "publish",
         ["messaging.client_id"]: taskContext.worker?.id,
+        [SEMATTRS_MESSAGING_DESTINATION]: id,
         [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
         ...accessoryAttributes({
           items: [
@@ -1099,16 +1055,8 @@ export async function batchTrigger<TTask extends AnyTask>(
           style: "codepath",
         }),
       },
-      ...requestOptions,
     }
   );
-
-  const handle = {
-    batchId: response.batchId,
-    runs: response.runs.map((id) => ({ id })),
-  };
-
-  return handle as TaskBatchOutputHandle<TTask>;
 }
 
 async function handleBatchTaskRunExecutionResult<TOutput>(

packages/trigger-sdk/src/v3/tasks.ts

@@ -1,5 +1,12 @@
 import { InitOutput } from "@trigger.dev/core/v3";
-import { batchTrigger, createTask, trigger, triggerAndPoll, triggerAndWait } from "./shared";
+import {
+  batchTrigger,
+  batchTriggerAndWait,
+  createTask,
+  trigger,
+  triggerAndPoll,
+  triggerAndWait,
+} from "./shared";
 
 import type {
   TaskOptions,
@@ -65,4 +72,5 @@ export const tasks = {
   triggerAndPoll,
   batchTrigger,
   triggerAndWait,
+  batchTriggerAndWait,
 };

references/v3-catalog/src/trigger/triggerKitchenSink.ts

@@ -0,0 +1,87 @@
+import { tasks, task } from "@trigger.dev/sdk/v3";
+
+export const triggerKitchenSink = task({
+  id: "trigger-kitchen-sink",
+  run: async (payload: { message: string }) => {
+    await triggerKitchenSinkChild.trigger({
+      message: `${payload.message} - 2.b`,
+    });
+
+    await tasks.trigger<typeof triggerKitchenSinkChild>("trigger-kitchen-sink-child", {
+      message: `${payload.message} - 2.c`,
+    });
+
+    await triggerKitchenSinkChild.triggerAndWait({
+      message: `${payload.message} - 2.b`,
+    });
+
+    await tasks.triggerAndWait<typeof triggerKitchenSinkChild>("trigger-kitchen-sink-child", {
+      message: `${payload.message} - 2.c`,
+    });
+
+    await triggerKitchenSinkChild.batchTrigger([
+      {
+        payload: {
+          message: `${payload.message} - 2.c`,
+        },
+      },
+      {
+        payload: {
+          message: `${payload.message} - 2.cc`,
+        },
+      },
+    ]);
+
+    await tasks.batchTrigger<typeof triggerKitchenSinkChild>("trigger-kitchen-sink-child", [
+      {
+        payload: {
+          message: `${payload.message} - 2.c`,
+        },
+      },
+      {
+        payload: {
+          message: `${payload.message} - 2.cc`,
+        },
+      },
+    ]);
+
+    await triggerKitchenSinkChild.batchTriggerAndWait([
+      {
+        payload: {
+          message: `${payload.message} - 2.d`,
+        },
+      },
+      {
+        payload: {
+          message: `${payload.message} - 2.dd`,
+        },
+      },
+    ]);
+
+    await tasks.batchTriggerAndWait<typeof triggerKitchenSinkChild>("trigger-kitchen-sink-child", [
+      {
+        payload: {
+          message: `${payload.message} - 2.d`,
+        },
+      },
+      {
+        payload: {
+          message: `${payload.message} - 2.dd`,
+        },
+      },
+    ]);
+
+    return {
+      hello: "world",
+    };
+  },
+});
+
+export const triggerKitchenSinkChild = task({
+  id: "trigger-kitchen-sink-child",
+  run: async (payload: { message: string }) => {
+    return {
+      foo: payload.message,
+    };
+  },
+});