WIP on retrieving a queue using the SDK/API

Author: matt-aitken

apps/webapp/app/presenters/v3/QueueListPresenter.server.ts

@@ -1,9 +1,8 @@
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { determineEngineVersion } from "~/v3/engineVersion.server";
 import { engine } from "~/v3/runEngine.server";
 import { BasePresenter } from "./basePresenter.server";
-import { type TaskQueueType } from "@trigger.dev/database";
-import { assertExhaustive } from "@trigger.dev/core";
-import { determineEngineVersion } from "~/v3/engineVersion.server";
+import { toQueueItem } from "./QueueRetrievePresenter.server";
 
 const DEFAULT_ITEMS_PER_PAGE = 25;
 const MAX_ITEMS_PER_PAGE = 100;
@@ -57,6 +56,7 @@ export class QueueListPresenter extends BasePresenter {
         runtimeEnvironmentId: environment.id,
       },
       select: {
+        friendlyId: true,
         name: true,
         concurrencyLimit: true,
         type: true,
@@ -80,23 +80,15 @@ export class QueueListPresenter extends BasePresenter {
     ]);
 
     // Transform queues to include running and queued counts
-    return queues.map((queue) => ({
-      name: queue.name.replace(/^task\//, ""),
-      type: queueTypeFromType(queue.type),
-      running: results[1][queue.name] ?? 0,
-      queued: results[0][queue.name] ?? 0,
-      concurrencyLimit: queue.concurrencyLimit ?? null,
-    }));
-  }
-}
-
-export function queueTypeFromType(type: TaskQueueType) {
-  switch (type) {
-    case "NAMED":
-      return "custom" as const;
-    case "VIRTUAL":
-      return "task" as const;
-    default:
-      assertExhaustive(type);
+    return queues.map((queue) =>
+      toQueueItem({
+        friendlyId: queue.friendlyId,
+        name: queue.name,
+        type: queue.type,
+        running: results[1][queue.name] ?? 0,
+        queued: results[0][queue.name] ?? 0,
+        concurrencyLimit: queue.concurrencyLimit ?? null,
+      })
+    );
   }
 }

apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts

@@ -0,0 +1,108 @@
+import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { engine } from "~/v3/runEngine.server";
+import { BasePresenter } from "./basePresenter.server";
+import { type TaskQueueType } from "@trigger.dev/database";
+import { assertExhaustive } from "@trigger.dev/core";
+import { determineEngineVersion } from "~/v3/engineVersion.server";
+import { type QueueItem, type RetrieveQueueParam } from "@trigger.dev/core/v3";
+
+export class QueueRetrievePresenter extends BasePresenter {
+  public async call({
+    environment,
+    queueInput,
+  }: {
+    environment: AuthenticatedEnvironment;
+    queueInput: RetrieveQueueParam;
+  }) {
+    //check the engine is the correct version
+    const engineVersion = await determineEngineVersion({ environment });
+
+    if (engineVersion === "V1") {
+      return {
+        success: false as const,
+        code: "engine-version",
+      };
+    }
+
+    const queue = await this.getQueue(environment, queueInput);
+    if (!queue) {
+      return {
+        success: false as const,
+        code: "queue-not-found",
+      };
+    }
+
+    const results = await Promise.all([
+      engine.lengthOfQueues(environment, [queue.name]),
+      engine.currentConcurrencyOfQueues(environment, [queue.name]),
+    ]);
+
+    // Transform queues to include running and queued counts
+    return {
+      success: true as const,
+      queue: toQueueItem({
+        friendlyId: queue.friendlyId,
+        name: queue.name,
+        type: queue.type,
+        running: results[1]?.[queue.name] ?? 0,
+        queued: results[0]?.[queue.name] ?? 0,
+        concurrencyLimit: queue.concurrencyLimit ?? null,
+      }),
+    };
+  }
+
+  private async getQueue(environment: AuthenticatedEnvironment, queue: RetrieveQueueParam) {
+    if (typeof queue === "string") {
+      return this._replica.taskQueue.findFirst({
+        where: {
+          friendlyId: queue,
+          runtimeEnvironmentId: environment.id,
+        },
+      });
+    }
+
+    const queueName =
+      queue.type === "task" ? `task/${queue.name.replace(/^task\//, "")}` : queue.name;
+    return this._replica.taskQueue.findFirst({
+      where: {
+        name: queueName,
+        runtimeEnvironmentId: environment.id,
+      },
+    });
+  }
+}
+
+function queueTypeFromType(type: TaskQueueType) {
+  switch (type) {
+    case "NAMED":
+      return "custom" as const;
+    case "VIRTUAL":
+      return "task" as const;
+    default:
+      assertExhaustive(type);
+  }
+}
+
+/**
+ * Converts raw queue data into a standardized QueueItem format
+ * @param data Raw queue data containing required queue properties
+ * @returns A validated QueueItem object
+ */
+export function toQueueItem(data: {
+  friendlyId: string;
+  name: string;
+  type: TaskQueueType;
+  running: number;
+  queued: number;
+  concurrencyLimit: number | null;
+}): QueueItem {
+  return {
+    id: data.friendlyId,
+    //remove the task/ prefix if it exists
+    name: data.name.replace(/^task\//, ""),
+    type: queueTypeFromType(data.type),
+    running: data.running,
+    queued: data.queued,
+    concurrencyLimit: data.concurrencyLimit,
+  };
+}

apps/webapp/app/routes/api.v1.queues.$queueParam.ts

@@ -0,0 +1,45 @@
+import { json } from "@remix-run/server-runtime";
+import { type QueueItem, type RetrieveQueueParam, RetrieveQueueType } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { QueueRetrievePresenter } from "~/presenters/v3/QueueRetrievePresenter.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+const SearchParamsSchema = z.object({
+  type: RetrieveQueueType.default("id"),
+});
+
+export const loader = createLoaderApiRoute(
+  {
+    params: z.object({
+      queueParam: z.string(),
+    }),
+    searchParams: SearchParamsSchema,
+    findResource: async () => 1, // This is a dummy function, we don't need to find a resource
+  },
+  async ({ params, searchParams, authentication }) => {
+    const input: RetrieveQueueParam =
+      searchParams.type === "id"
+        ? params.queueParam
+        : {
+            type: searchParams.type,
+            name: decodeURIComponent(params.queueParam),
+          };
+
+    const presenter = new QueueRetrievePresenter();
+    const result = await presenter.call({
+      environment: authentication.environment,
+      queueInput: input,
+    });
+
+    if (!result.success) {
+      if (result.code === "queue-not-found") {
+        return json({ error: result.code }, { status: 404 });
+      }
+
+      return json({ error: result.code }, { status: 400 });
+    }
+
+    const q: QueueItem = result.queue;
+    return json(q);
+  }
+);

apps/webapp/app/routes/api.v1.queues.ts

@@ -1,4 +1,5 @@
 import { json } from "@remix-run/server-runtime";
+import { type QueueItem } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { QueueListPresenter } from "~/presenters/v3/QueueListPresenter.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
@@ -27,7 +28,7 @@ export const loader = createLoaderApiRoute(
         return json({ error: result.code }, { status: 400 });
       }
 
-      const queues = await result.queues;
+      const queues: QueueItem[] = await result.queues;
       return json({ data: queues, pagination: result.pagination }, { status: 200 });
     } catch (error) {
       if (error instanceof ServiceValidationError) {

apps/webapp/app/services/routeBuilders/apiBuilder.server.ts

@@ -39,7 +39,12 @@ type ApiKeyRouteBuilderOptions<
     params: TParamsSchema extends z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>
       ? z.infer<TParamsSchema>
       : undefined,
-    authentication: ApiAuthenticationResultSuccess
+    authentication: ApiAuthenticationResultSuccess,
+    searchParams: TSearchParamsSchema extends
+      | z.ZodFirstPartySchemaTypes
+      | z.ZodDiscriminatedUnion<any, any>
+      ? z.infer<TSearchParamsSchema>
+      : undefined
   ) => Promise<TResource | undefined>;
   shouldRetryNotFound?: boolean;
   authorization?: {
@@ -179,7 +184,7 @@ export function createLoaderApiRoute<
       }
 
       // Find the resource
-      const resource = await findResource(parsedParams, authenticationResult);
+      const resource = await findResource(parsedParams, authenticationResult, parsedSearchParams);
 
       if (!resource) {
         return await wrapResponse(

packages/core/src/v3/apiClient/index.ts

@@ -25,6 +25,7 @@ import {
   ReplayRunResponse,
   RescheduleRunRequestBody,
   RetrieveBatchV2Response,
+  RetrieveQueueParam,
   RetrieveRunResponse,
   ScheduleObject,
   TaskRunExecutionResult,
@@ -744,6 +745,21 @@ export class ApiClient {
     );
   }
 
+  retrieveQueue(queue: RetrieveQueueParam, requestOptions?: ZodFetchOptions) {
+    const type = typeof queue === "string" ? "id" : queue.type;
+    const value = typeof queue === "string" ? queue : queue.name;
+
+    return zodfetch(
+      QueueItem,
+      `${this.baseUrl}/api/v1/queues/${encodeURIComponent(value)}?type=${type}`,
+      {
+        method: "GET",
+        headers: this.#getHeaders(false),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
   subscribeToRun<TRunTypes extends AnyRunTypes>(
     runId: string,
     options?: {

packages/core/src/v3/schemas/queues.ts

@@ -1,14 +1,21 @@
 import { z } from "zod";
 
+const queueTypes = ["task", "custom"] as const;
+
 /**
  * The type of queue, either "task" or "custom"
  * "task" are created automatically for each task.
  * "custom" are created by you explicitly in your code.
  * */
-export const QueueType = z.enum(["task", "custom"]);
+export const QueueType = z.enum(queueTypes);
 export type QueueType = z.infer<typeof QueueType>;
 
+export const RetrieveQueueType = z.enum([...queueTypes, "id"]);
+export type RetrieveQueueType = z.infer<typeof RetrieveQueueType>;
+
 export const QueueItem = z.object({
+  /** The queue id, e.g. queue_12345 */
+  id: z.string(),
   /** The queue name */
   name: z.string(),
   /**
@@ -35,3 +42,35 @@ export const ListQueueOptions = z.object({
 });
 
 export type ListQueueOptions = z.infer<typeof ListQueueOptions>;
+
+/**
+ * When retrieving a queue you can either use the queue id,
+ * or the type and name.
+ *
+ * @example
+ *
+ * ```ts
+ * // Use a queue id (they start with queue_
+ * const q1 = await queues.retrieve("queue_12345");
+ *
+ * // Or use the type and name
+ * // The default queue for your "my-task-id"
+ * const q2 = await queues.retrieve({ type: "task", name: "my-task-id"});
+ *
+ * // The custom queue you defined in your code
+ * const q3 = await queues.retrieve({ type: "custom", name: "my-custom-queue" });
+ * ```
+ */
+export const RetrieveQueueParam = z.union([
+  z.string(),
+  z.object({
+    /** "task" or "custom" */
+    type: QueueType,
+    /** The name of your queue.
+     * For "task" type it will be the task id, for "custom" it will be the name you specified.
+     * */
+    name: z.string(),
+  }),
+]);
+
+export type RetrieveQueueParam = z.infer<typeof RetrieveQueueParam>;