Added pause/resume functions to the SDK

Author: matt-aitken

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

@@ -4,7 +4,7 @@ import { engine } from "~/v3/runEngine.server";
 import { BasePresenter } from "./basePresenter.server";
 import { toQueueItem } from "./QueueRetrievePresenter.server";
 
-const DEFAULT_ITEMS_PER_PAGE = 10;
+const DEFAULT_ITEMS_PER_PAGE = 25;
 const MAX_ITEMS_PER_PAGE = 100;
 export class QueueListPresenter extends BasePresenter {
   private readonly perPage: number;

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

@@ -0,0 +1,46 @@
+import { json } from "@remix-run/server-runtime";
+import { type QueueItem, type RetrieveQueueParam, RetrieveQueueType } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { PauseQueueService } from "~/v3/services/pauseQueue.server";
+
+const BodySchema = z.object({
+  type: RetrieveQueueType.default("id"),
+  action: z.enum(["pause", "resume"]),
+});
+
+export const { action } = createActionApiRoute(
+  {
+    body: BodySchema,
+    params: z.object({
+      queueParam: z.string().transform((val) => val.replace(/%2F/g, "/")),
+    }),
+  },
+  async ({ params, body, authentication }) => {
+    const input: RetrieveQueueParam =
+      body.type === "id"
+        ? params.queueParam
+        : {
+            type: body.type,
+            name: decodeURIComponent(params.queueParam).replace(/%2F/g, "/"),
+          };
+
+    const service = new PauseQueueService();
+    const result = await service.call(
+      authentication.environment,
+      input,
+      body.action === "pause" ? "paused" : "resumed"
+    );
+
+    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/v3/services/pauseQueue.server.ts

@@ -1,17 +1,19 @@
-import { type RetrieveQueueParam } from "@trigger.dev/core/v3";
-import { getQueue } from "~/presenters/v3/QueueRetrievePresenter.server";
+import { QueueItem, type RetrieveQueueParam } from "@trigger.dev/core/v3";
+import { getQueue, toQueueItem } from "~/presenters/v3/QueueRetrievePresenter.server";
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { BaseService } from "./baseService.server";
 import { determineEngineVersion } from "../engineVersion.server";
 import { removeQueueConcurrencyLimits, updateQueueConcurrencyLimits } from "../runQueue.server";
+import { engine } from "../runEngine.server";
 
 export type PauseStatus = "paused" | "resumed";
 
 export type PauseQueueResult =
   | {
       success: true;
       state: PauseStatus;
+      queue: QueueItem;
     }
   | {
       success: false;
@@ -46,7 +48,7 @@ export class PauseQueueService extends BaseService {
         };
       }
 
-      await this._prisma.taskQueue.update({
+      const updatedQueue = await this._prisma.taskQueue.update({
         where: {
           id: queue.id,
         },
@@ -71,9 +73,23 @@ export class PauseQueueService extends BaseService {
         environmentId: environment.id,
       });
 
+      const results = await Promise.all([
+        engine.lengthOfQueues(environment, [queue.name]),
+        engine.currentConcurrencyOfQueues(environment, [queue.name]),
+      ]);
+
       return {
         success: true,
         state: action,
+        queue: toQueueItem({
+          friendlyId: updatedQueue.friendlyId,
+          name: updatedQueue.name,
+          type: updatedQueue.type,
+          running: results[1]?.[updatedQueue.name] ?? 0,
+          queued: results[0]?.[updatedQueue.name] ?? 0,
+          concurrencyLimit: updatedQueue.concurrencyLimit ?? null,
+          paused: updatedQueue.paused,
+        }),
       };
     } catch (error) {
       logger.error("PauseQueueService: error updating queue state", {

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

@@ -763,6 +763,32 @@ export class ApiClient {
     );
   }
 
+  pauseQueue(
+    queue: RetrieveQueueParam,
+    action: "pause" | "resume",
+    requestOptions?: ZodFetchOptions
+  ) {
+    const type = typeof queue === "string" ? "id" : queue.type;
+    const value = typeof queue === "string" ? queue : queue.name;
+
+    // Explicitly encode slashes before encoding the rest of the string
+    const encodedValue = encodeURIComponent(value.replace(/\//g, "%2F"));
+
+    return zodfetch(
+      QueueItem,
+      `${this.baseUrl}/api/v1/queues/${encodedValue}/pause`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(false),
+        body: JSON.stringify({
+          type,
+          action,
+        }),
+      },
+      mergeRequestOptions(this.defaultRequestOptions, requestOptions)
+    );
+  }
+
   subscribeToRun<TRunTypes extends AnyRunTypes>(
     runId: string,
     options?: {

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

@@ -86,3 +86,92 @@ export function retrieve(
 
   return apiClient.retrieveQueue(queue, $requestOptions);
 }
+
+/**
+ * Pauses a queue, preventing any new runs from being started.
+ * Runs that are currently running will continue to completion.
+ *
+ * @example
+ * ```ts
+ * // Pause using a queue id
+ * await queues.pause("queue_12345");
+ *
+ * // Or pause using type and name
+ * await queues.pause({ type: "task", name: "my-task-id"});
+ * ```
+ * @param queue - The ID of the queue to pause, or the type and name
+ * @returns The updated queue state
+ */
+export function pause(
+  queue: RetrieveQueueParam,
+  requestOptions?: ApiRequestOptions
+): ApiPromise<QueueItem> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "queues.pause()",
+      icon: "queue",
+      attributes: {
+        ...flattenAttributes({ queue }),
+        ...accessoryAttributes({
+          items: [
+            {
+              text: typeof queue === "string" ? queue : queue.name,
+              variant: "normal",
+            },
+          ],
+          style: "codepath",
+        }),
+      },
+    },
+    requestOptions
+  );
+
+  return apiClient.pauseQueue(queue, "pause", $requestOptions);
+}
+
+/**
+ * Resumes a paused queue, allowing new runs to be started.
+ *
+ * @example
+ * ```ts
+ * // Resume using a queue id
+ * await queues.resume("queue_12345");
+ *
+ * // Or resume using type and name
+ * await queues.resume({ type: "task", name: "my-task-id"});
+ * ```
+ * @param queue - The ID of the queue to resume, or the type and name
+ * @returns The updated queue state
+ */
+export function resume(
+  queue: RetrieveQueueParam,
+  requestOptions?: ApiRequestOptions
+): ApiPromise<QueueItem> {
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const $requestOptions = mergeRequestOptions(
+    {
+      tracer,
+      name: "queues.resume()",
+      icon: "queue",
+      attributes: {
+        ...flattenAttributes({ queue }),
+        ...accessoryAttributes({
+          items: [
+            {
+              text: typeof queue === "string" ? queue : queue.name,
+              variant: "normal",
+            },
+          ],
+          style: "codepath",
+        }),
+      },
+    },
+    requestOptions
+  );
+
+  return apiClient.pauseQueue(queue, "resume", $requestOptions);
+}

references/hello-world/src/trigger/queues.ts

@@ -18,10 +18,24 @@ export const queuesTester = task({
     });
     logger.log("Retrieved from name", { retrievedFromCtxName });
 
+    //pause the queue
+    const pausedQueue = await queues.pause({
+      type: "task",
+      name: "queues-tester",
+    });
+    logger.log("Paused queue", { pausedQueue });
+
     const retrievedFromName = await queues.retrieve({
       type: "task",
       name: "queues-tester",
     });
     logger.log("Retrieved from name", { retrievedFromName });
+
+    //resume the queue
+    const resumedQueue = await queues.resume({
+      type: "task",
+      name: "queues-tester",
+    });
+    logger.log("Resumed queue", { resumedQueue });
   },
 });