Support custom queue in TriggerTaskService

Author: matt-aitken

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

@@ -4,13 +4,12 @@ import {
   TriggerTaskRequestBody,
   packetRequiresOffloading,
 } from "@trigger.dev/core/v3";
-import { nanoid } from "nanoid";
 import { createHash } from "node:crypto";
-import { $transaction } from "~/db.server";
+import { $transaction, prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { eventRepository } from "../eventRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "~/v3/marqs/index.server";
 import { uploadToObjectStore } from "../r2.server";
 import { BaseService } from "./baseService.server";
 
@@ -112,7 +111,12 @@ export class TriggerTaskService extends BaseService {
               select: { lastNumber: true },
             });
 
-            const queueName = body.options?.queue?.name ?? `task/${taskId}`;
+            let queueName = sanitizeQueueName(body.options?.queue?.name ?? `task/${taskId}`);
+
+            // Check that the queuename is not an empty string
+            if (!queueName) {
+              queueName = sanitizeQueueName(`task/${taskId}`);
+            }
 
             event.setAttribute("queueName", queueName);
             span.setAttribute("queueName", queueName);
@@ -183,6 +187,41 @@ export class TriggerTaskService extends BaseService {
               }
             }
 
+            if (body.options?.queue) {
+              const concurrencyLimit = body.options.queue.concurrencyLimit
+                ? Math.max(0, body.options.queue.concurrencyLimit)
+                : null;
+              const taskQueue = await prisma.taskQueue.upsert({
+                where: {
+                  runtimeEnvironmentId_name: {
+                    runtimeEnvironmentId: environment.id,
+                    name: queueName,
+                  },
+                },
+                update: {
+                  concurrencyLimit,
+                  rateLimit: body.options.queue.rateLimit,
+                },
+                create: {
+                  friendlyId: generateFriendlyId("queue"),
+                  name: queueName,
+                  concurrencyLimit,
+                  runtimeEnvironmentId: environment.id,
+                  projectId: environment.projectId,
+                  rateLimit: body.options.queue.rateLimit,
+                  type: "NAMED",
+                },
+              });
+
+              if (typeof taskQueue.concurrencyLimit === "number") {
+                await marqs?.updateQueueConcurrencyLimits(
+                  environment,
+                  taskQueue.name,
+                  taskQueue.concurrencyLimit
+                );
+              }
+            }
+
             return taskRun;
           });