+
diff --git a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts
index 0fe9e3f3652..de88628a834 100644
--- a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts
+++ b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts
@@ -1,11 +1,16 @@
-import { TaskQueueType } from "@trigger.dev/database";
+import type { RunEngine } from "@internal/run-engine";
+import { Prisma, TaskQueueType } from "@trigger.dev/database";
+import { type PrismaClientOrTransaction } from "~/db.server";
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 { toQueueItem } from "./QueueRetrievePresenter.server";
+import type { QueueListPagination } from "./queueListPagination.server";
-const DEFAULT_ITEMS_PER_PAGE = 25;
+type QueueListEngine = Pick
;
+
+export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25;
const MAX_ITEMS_PER_PAGE = 100;
const typeToDBQueueType: Record<"task" | "custom", TaskQueueType> = {
@@ -13,12 +18,51 @@ const typeToDBQueueType: Record<"task" | "custom", TaskQueueType> = {
custom: TaskQueueType.NAMED,
};
+const queueListSelect = {
+ friendlyId: true,
+ name: true,
+ orderableName: true,
+ concurrencyLimit: true,
+ concurrencyLimitBase: true,
+ concurrencyLimitOverriddenAt: true,
+ concurrencyLimitOverriddenBy: true,
+ type: true,
+ paused: true,
+} satisfies Prisma.TaskQueueSelect;
+
+function buildQueueListWhere(
+ environmentId: string,
+ query: string | undefined,
+ type: "task" | "custom" | undefined
+): Prisma.TaskQueueWhereInput {
+ const trimmedQuery = query?.trim();
+
+ return {
+ runtimeEnvironmentId: environmentId,
+ version: "V2",
+ name: trimmedQuery
+ ? {
+ contains: trimmedQuery,
+ mode: "insensitive",
+ }
+ : undefined,
+ type: type ? typeToDBQueueType[type] : undefined,
+ };
+}
+
export class QueueListPresenter extends BasePresenter {
private readonly perPage: number;
+ private readonly engineClient: QueueListEngine;
- constructor(perPage: number = DEFAULT_ITEMS_PER_PAGE) {
- super();
+ constructor(
+ perPage: number = QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE,
+ prismaClient?: PrismaClientOrTransaction,
+ replicaClient?: PrismaClientOrTransaction,
+ engineClient: QueueListEngine = engine
+ ) {
+ super(prismaClient, replicaClient);
this.perPage = Math.min(perPage, MAX_ITEMS_PER_PAGE);
+ this.engineClient = engineClient;
}
public async call({
@@ -33,26 +77,14 @@ export class QueueListPresenter extends BasePresenter {
perPage?: number;
type?: "task" | "custom";
}) {
- const hasFilters = (query !== undefined && query.length > 0) || type !== undefined;
-
- // Get total count for pagination
- const totalQueues = await this._replica.taskQueue.count({
- where: {
- runtimeEnvironmentId: environment.id,
- version: "V2",
- name: query
- ? {
- contains: query,
- mode: "insensitive",
- }
- : undefined,
- type: type ? typeToDBQueueType[type] : undefined,
- },
- });
+ const hasFilters = Boolean(query?.trim()) || type !== undefined;
- //check the engine is the correct version
const engineVersion = await determineEngineVersion({ environment });
if (engineVersion === "V1") {
+ const totalQueues = await this._replica.taskQueue.count({
+ where: buildQueueListWhere(environment.id, query, type),
+ });
+
if (totalQueues === 0) {
const oldQueue = await this._replica.taskQueue.findFirst({
where: {
@@ -78,10 +110,30 @@ export class QueueListPresenter extends BasePresenter {
};
}
+ if (hasFilters) {
+ const { queues, hasMore } = await this.getFilteredQueues(environment, query, page, type);
+
+ return {
+ success: true as const,
+ queues,
+ pagination: {
+ mode: "filtered" as const,
+ currentPage: page,
+ hasMore,
+ },
+ hasFilters,
+ };
+ }
+
+ const totalQueues = await this._replica.taskQueue.count({
+ where: buildQueueListWhere(environment.id, query, type),
+ });
+
return {
success: true as const,
- queues: await this.getQueuesWithPagination(environment, query, page, type),
+ queues: await this.getUnfilteredQueues(environment, page, type),
pagination: {
+ mode: "unfiltered" as const,
currentPage: page,
totalPages: Math.ceil(totalQueues / this.perPage),
count: totalQueues,
@@ -91,35 +143,38 @@ export class QueueListPresenter extends BasePresenter {
};
}
- private async getQueuesWithPagination(
+ private async getFilteredQueues(
environment: AuthenticatedEnvironment,
query: string | undefined,
page: number,
type: "task" | "custom" | undefined
) {
const queues = await this._replica.taskQueue.findMany({
- where: {
- runtimeEnvironmentId: environment.id,
- version: "V2",
- name: query
- ? {
- contains: query,
- mode: "insensitive",
- }
- : undefined,
- type: type ? typeToDBQueueType[type] : undefined,
- },
- select: {
- friendlyId: true,
- name: true,
- orderableName: true,
- concurrencyLimit: true,
- concurrencyLimitBase: true,
- concurrencyLimitOverriddenAt: true,
- concurrencyLimitOverriddenBy: true,
- type: true,
- paused: true,
+ where: buildQueueListWhere(environment.id, query, type),
+ select: queueListSelect,
+ orderBy: {
+ orderableName: "asc",
},
+ skip: (page - 1) * this.perPage,
+ take: this.perPage + 1,
+ });
+
+ const hasMore = queues.length > this.perPage;
+
+ return {
+ queues: await this.enrichQueues(environment, queues.slice(0, this.perPage)),
+ hasMore,
+ };
+ }
+
+ private async getUnfilteredQueues(
+ environment: AuthenticatedEnvironment,
+ page: number,
+ type: "task" | "custom" | undefined
+ ) {
+ const queues = await this._replica.taskQueue.findMany({
+ where: buildQueueListWhere(environment.id, undefined, type),
+ select: queueListSelect,
orderBy: {
orderableName: "asc",
},
@@ -127,12 +182,29 @@ export class QueueListPresenter extends BasePresenter {
take: this.perPage,
});
- const results = await Promise.all([
- engine.lengthOfQueues(
+ return this.enrichQueues(environment, queues);
+ }
+
+ private async enrichQueues(
+ environment: AuthenticatedEnvironment,
+ queues: {
+ friendlyId: string;
+ name: string;
+ orderableName: string | null;
+ concurrencyLimit: number | null;
+ concurrencyLimitBase: number | null;
+ concurrencyLimitOverriddenAt: Date | null;
+ concurrencyLimitOverriddenBy: string | null;
+ type: TaskQueueType;
+ paused: boolean;
+ }[]
+ ) {
+ const [queuedByQueue, runningByQueue] = await Promise.all([
+ this.engineClient.lengthOfQueues(
environment,
queues.map((q) => q.name)
),
- engine.currentConcurrencyOfQueues(
+ this.engineClient.currentConcurrencyOfQueues(
environment,
queues.map((q) => q.name)
),
@@ -149,14 +221,13 @@ export class QueueListPresenter extends BasePresenter {
const overriddenByMap = new Map(overriddenByUsers.map((u) => [u.id, u]));
- // Transform queues to include running and queued counts
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,
+ running: runningByQueue[queue.name] ?? 0,
+ queued: queuedByQueue[queue.name] ?? 0,
concurrencyLimit: queue.concurrencyLimit ?? null,
concurrencyLimitBase: queue.concurrencyLimitBase ?? null,
concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null,
diff --git a/apps/webapp/app/presenters/v3/queueListPagination.server.ts b/apps/webapp/app/presenters/v3/queueListPagination.server.ts
new file mode 100644
index 00000000000..b366ebe0f4e
--- /dev/null
+++ b/apps/webapp/app/presenters/v3/queueListPagination.server.ts
@@ -0,0 +1,43 @@
+export type QueueListFilteredPagination = {
+ mode: "filtered";
+ currentPage: number;
+ hasMore: boolean;
+};
+
+export type QueueListUnfilteredPagination = {
+ mode: "unfiltered";
+ currentPage: number;
+ totalPages: number;
+ count: number;
+};
+
+export type QueueListPagination = QueueListFilteredPagination | QueueListUnfilteredPagination;
+
+export type OffsetLimitPagination = {
+ currentPage: number;
+ totalPages: number;
+ count: number;
+};
+
+/** Maps presenter pagination to the public API / SDK offset-limit contract. */
+export function toOffsetLimitQueueListPagination(
+ pagination: QueueListPagination,
+ options: { itemsOnPage: number; perPage: number }
+): OffsetLimitPagination {
+ if (pagination.mode === "unfiltered") {
+ return {
+ currentPage: pagination.currentPage,
+ totalPages: pagination.totalPages,
+ count: pagination.count,
+ };
+ }
+
+ return {
+ currentPage: pagination.currentPage,
+ totalPages: pagination.hasMore ? pagination.currentPage + 1 : pagination.currentPage,
+ count:
+ (pagination.currentPage - 1) * options.perPage +
+ options.itemsOnPage +
+ (pagination.hasMore ? 1 : 0),
+ };
+}
diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx
index debab4683ce..b6afe46be4f 100644
--- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx
+++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx
@@ -440,7 +440,10 @@ export default function Page() {