v3: fix triggering tasks with custom queues (#1242)

* add missing dotenv requires in catalog scripts

* pass task queue options to all task trigger functions

* remove unnecessary include

* fallback to background worker task queue options

* add changeset

Author: nicktrn

.changeset/few-poems-vanish.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix trigger functions for custom queues

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

@@ -274,14 +274,6 @@ export class SharedQueueConsumer {
           where: {
             id: message.messageId,
           },
-          include: {
-            lockedToVersion: {
-              include: {
-                deployment: true,
-                tasks: true,
-              },
-            },
-          },
         });
 
         if (!existingTaskRun) {

apps/webapp/app/v3/models/workerDeployment.server.ts

@@ -1,6 +1,8 @@
 import type { Prettify } from "@trigger.dev/core";
+import { BackgroundWorker } from "@trigger.dev/database";
 import { CURRENT_DEPLOYMENT_LABEL } from "~/consts";
 import { Prisma, prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 
 export type CurrentWorkerDeployment = Prettify<
   NonNullable<Awaited<ReturnType<typeof findCurrentWorkerDeployment>>>
@@ -42,6 +44,25 @@ export async function findCurrentWorkerDeployment(
   return promotion?.deployment;
 }
 
+export async function findCurrentWorkerFromEnvironment(
+  environment: Pick<AuthenticatedEnvironment, "id" | "type">
+): Promise<BackgroundWorker | null> {
+  if (environment.type === "DEVELOPMENT") {
+    const latestDevWorker = await prisma.backgroundWorker.findFirst({
+      where: {
+        runtimeEnvironmentId: environment.id,
+      },
+      orderBy: {
+        createdAt: "desc",
+      },
+    });
+    return latestDevWorker;
+  } else {
+    const deployment = await findCurrentWorkerDeployment(environment.id);
+    return deployment?.worker ?? null;
+  }
+}
+
 export async function getWorkerDeploymentFromWorker(
   workerId: string
 ): Promise<WorkerDeploymentWithWorkerTasks | undefined> {

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

@@ -1,5 +1,6 @@
 import {
   IOPacket,
+  QueueOptions,
   SemanticInternalAttributes,
   TriggerTaskRequestBody,
   packetRequiresOffloading,
@@ -18,6 +19,7 @@ import { BaseService, ServiceValidationError } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
 import { createTag, MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
+import { findCurrentWorkerFromEnvironment } from "../models/workerDeployment.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -211,7 +213,9 @@ export class TriggerTaskService extends BaseService {
                   })
                 : undefined;
 
-              let queueName = sanitizeQueueName(body.options?.queue?.name ?? `task/${taskId}`);
+              let queueName = sanitizeQueueName(
+                await this.#getQueueName(taskId, environment, body.options?.queue?.name)
+              );
 
               // Check that the queuename is not an empty string
               if (!queueName) {
@@ -399,6 +403,57 @@ export class TriggerTaskService extends BaseService {
     });
   }
 
+  async #getQueueName(taskId: string, environment: AuthenticatedEnvironment, queueName?: string) {
+    if (queueName) {
+      return queueName;
+    }
+
+    const defaultQueueName = `task/${taskId}`;
+
+    const worker = await findCurrentWorkerFromEnvironment(environment);
+
+    if (!worker) {
+      logger.debug("Failed to get queue name: No worker found", {
+        taskId,
+        environmentId: environment.id,
+      });
+
+      return defaultQueueName;
+    }
+
+    const task = await this._prisma.backgroundWorkerTask.findUnique({
+      where: {
+        workerId_slug: {
+          workerId: worker.id,
+          slug: taskId,
+        },
+      },
+    });
+
+    if (!task) {
+      console.log("Failed to get queue name: No task found", {
+        taskId,
+        environmentId: environment.id,
+      });
+
+      return defaultQueueName;
+    }
+
+    const queueConfig = QueueOptions.optional().safeParse(task.queueConfig);
+
+    if (!queueConfig.success) {
+      console.log("Failed to get queue name: Invalid queue config", {
+        taskId,
+        environmentId: environment.id,
+        queueConfig: task.queueConfig,
+      });
+
+      return defaultQueueName;
+    }
+
+    return queueConfig.data?.name ?? defaultQueueName;
+  }
+
   async #handlePayloadPacket(
     payload: any,
     payloadType: string,

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

@@ -476,6 +476,13 @@ export function createTask<
 >(
   params: TaskOptions<TIdentifier, TInput, TOutput, TInitOutput>
 ): Task<TIdentifier, TInput, TOutput> {
+  const customQueue = params.queue
+    ? queue({
+        name: params.queue?.name ?? `task/${params.id}`,
+        ...params.queue,
+      })
+    : undefined;
+
   const task: Task<TIdentifier, TInput, TOutput> = {
     id: params.id,
     trigger: async (payload, options) => {
@@ -487,7 +494,10 @@ export function createTask<
           : `trigger()`,
         params.id,
         payload,
-        options
+        {
+          queue: customQueue,
+          ...options,
+        }
       );
     },
     batchTrigger: async (items) => {
@@ -498,7 +508,9 @@ export function createTask<
           ? `${taskMetadata.exportName}.batchTrigger()`
           : `batchTrigger()`,
         params.id,
-        items
+        items,
+        undefined,
+        customQueue
       );
     },
     triggerAndWait: async (payload, options) => {
@@ -510,7 +522,10 @@ export function createTask<
           : `triggerAndWait()`,
         params.id,
         payload,
-        options
+        {
+          queue: customQueue,
+          ...options,
+        }
       );
     },
     batchTriggerAndWait: async (items) => {
@@ -521,7 +536,9 @@ export function createTask<
           ? `${taskMetadata.exportName}.batchTriggerAndWait()`
           : `batchTriggerAndWait()`,
         params.id,
-        items
+        items,
+        undefined,
+        customQueue
       );
     },
   };
@@ -758,7 +775,8 @@ async function batchTrigger_internal<TPayload, TOutput>(
   name: string,
   id: string,
   items: Array<BatchItem<TPayload>>,
-  requestOptions?: ApiRequestOptions
+  requestOptions?: ApiRequestOptions,
+  queue?: QueueOptions
 ): Promise<BatchRunHandle<TOutput>> {
   const apiClient = apiClientManager.client;
 
@@ -776,7 +794,7 @@ async function batchTrigger_internal<TPayload, TOutput>(
           return {
             payload: payloadPacket.data,
             options: {
-              queue: item.options?.queue,
+              queue: item.options?.queue ?? queue,
               concurrencyKey: item.options?.concurrencyKey,
               test: taskContext.ctx?.run.isTest,
               payloadType: payloadPacket.dataType,
@@ -919,7 +937,8 @@ async function batchTriggerAndWait_internal<TPayload, TOutput>(
   name: string,
   id: string,
   items: Array<BatchItem<TPayload>>,
-  requestOptions?: ApiRequestOptions
+  requestOptions?: ApiRequestOptions,
+  queue?: QueueOptions
 ): Promise<BatchResult<TOutput>> {
   const ctx = taskContext.ctx;
 
@@ -947,7 +966,7 @@ async function batchTriggerAndWait_internal<TPayload, TOutput>(
                 payload: payloadPacket.data,
                 options: {
                   lockToVersion: taskContext.worker?.version,
-                  queue: item.options?.queue,
+                  queue: item.options?.queue ?? queue,
                   concurrencyKey: item.options?.concurrencyKey,
                   test: taskContext.ctx?.run.isTest,
                   payloadType: payloadPacket.dataType,

references/v3-catalog/package.json

@@ -7,8 +7,8 @@
   },
   "scripts": {
     "dev:trigger": "triggerdev dev",
-    "management": "ts-node -r tsconfig-paths/register ./src/management.ts",
-    "queues": "ts-node -r tsconfig-paths/register ./src/queues.ts",
+    "management": "ts-node -r dotenv/config -r tsconfig-paths/register ./src/management.ts",
+    "queues": "ts-node -r dotenv/config -r tsconfig-paths/register ./src/queues.ts",
     "build:client": "tsup-node ./src/clientUsage.ts --format esm,cjs",
     "client": "ts-node -r dotenv/config -r tsconfig-paths/register ./src/clientUsage.ts",
     "triggerWithLargePayload": "ts-node -r dotenv/config -r tsconfig-paths/register ./src/triggerWithLargePayload.ts",