Skip to content

Commit aa74e68

Browse files
cardernecarderne
authored
feat(sdk): add bulk replay to api and sdk (#4105)
## Summary Adds SDK and API support for run bulk actions. You can now create bulk cancel or replay actions from `@trigger.dev/sdk` using run IDs or the same filters as `runs.list()`, then retrieve, list, poll, or abort the action by its `bulk_` handle. Tests, docs, changesets added. ## Design The dashboard bulk action service now accepts structured filters instead of reading directly from a dashboard request, so the dashboard and API share the same creation path. Replay actions created through the API are attributed with the existing `api` trigger source, while dashboard-created actions keep `dashboard`. The SDK exposes the new surface under `runs.bulk.*`, including `targetRegion` for replay region overrides and cursor pagination for listing bulk actions. ## Filters and runIds Nuance on filters. If `filter` is provided, it MUST have at least one key. This is to remove the footgun of passing no filter and selecting all runs. ```typescript { action: "cancel", runIds: ["run_1"] } // valid { action: "cancel", runIds: [] } // invalid, min(1) { action: "cancel", filter: { status: "FAILED" } } // valid { action: "cancel", filter: {} } // invalid { action: "cancel", filter: {}, runIds: ["run_1"] } // invalid ```
1 parent d59743b commit aa74e68

21 files changed

Lines changed: 1765 additions & 56 deletions

File tree

‎.changeset/bulk-actions-sdk-api.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
"@trigger.dev/core": patch
3+
"@trigger.dev/sdk": patch
4+
---
5+
6+
Add SDK and API client helpers for run bulk actions.

‎.server-changes/bulk-actions-api-sdk.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: webapp
3+
type: feature
4+
---
5+
6+
Add API and SDK support for creating, listing, retrieving, polling, and aborting run bulk actions.

‎apps/webapp/app/env.server.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2007,6 +2007,10 @@ const EnvironmentSchema = z
20072007
BULK_ACTION_BATCH_SIZE: z.coerce.number().int().default(100),
20082008
BULK_ACTION_BATCH_DELAY_MS: z.coerce.number().int().default(200),
20092009
BULK_ACTION_SUBBATCH_CONCURRENCY: z.coerce.number().int().default(5),
2010+
/// Max number of concurrent in-flight (PENDING) bulk replays per environment.
2011+
BULK_ACTION_MAX_CONCURRENT_REPLAYS: z.coerce.number().int().default(3),
2012+
/// Max number of explicit run IDs accepted in a single bulk action create request.
2013+
BULK_ACTION_MAX_RUN_IDS: z.coerce.number().int().default(500),
20102014

20112015
// AI Run Filter
20122016
AI_RUN_FILTER_MODEL: z.string().optional(),

‎apps/webapp/app/presenters/v3/ApiBulkActionPresenter.server.ts‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
import {
2+
type BulkActionGroup,
3+
type BulkActionStatus,
4+
type BulkActionType,
5+
} from "@trigger.dev/database";
6+
import { z } from "zod";
7+
import { BasePresenter } from "./basePresenter.server";
8+
9+
const DEFAULT_PAGE_SIZE = 25;
10+
const MAX_PAGE_SIZE = 100;
11+
12+
export const ApiBulkActionListSearchParams = z.object({
13+
"page[size]": z.coerce.number().int().positive().min(1).max(MAX_PAGE_SIZE).optional(),
14+
"page[after]": z.string().optional(),
15+
"page[before]": z.string().optional(),
16+
});
17+
18+
export type ApiBulkActionListSearchParams = z.infer<typeof ApiBulkActionListSearchParams>;
19+
20+
type BulkActionListCursor = {
21+
createdAt: Date;
22+
id: string;
23+
};
24+
25+
type BulkActionRow = Pick<
26+
BulkActionGroup,
27+
| "id"
28+
| "friendlyId"
29+
| "name"
30+
| "status"
31+
| "type"
32+
| "createdAt"
33+
| "completedAt"
34+
| "totalCount"
35+
| "successCount"
36+
| "failureCount"
37+
>;
38+
39+
export class ApiBulkActionPresenter extends BasePresenter {
40+
public async list(environmentId: string, searchParams: ApiBulkActionListSearchParams) {
41+
const pageSize = searchParams["page[size]"] ?? DEFAULT_PAGE_SIZE;
42+
const after = searchParams["page[after]"];
43+
const before = searchParams["page[before]"];
44+
45+
if (after && before) {
46+
throw new Error("Only one of page[after] or page[before] can be provided");
47+
}
48+
49+
const cursor = decodeCursor(after ?? before);
50+
const direction = before ? "backward" : "forward";
51+
52+
const where = {
53+
environmentId,
54+
...(cursor
55+
? direction === "forward"
56+
? {
57+
OR: [
58+
{ createdAt: { lt: cursor.createdAt } },
59+
{ createdAt: cursor.createdAt, id: { lt: cursor.id } },
60+
],
61+
}
62+
: {
63+
OR: [
64+
{ createdAt: { gt: cursor.createdAt } },
65+
{ createdAt: cursor.createdAt, id: { gt: cursor.id } },
66+
],
67+
}
68+
: {}),
69+
};
70+
71+
const rows = await this._replica.bulkActionGroup.findMany({
72+
select: bulkActionSelect,
73+
where,
74+
orderBy:
75+
direction === "forward"
76+
? [{ createdAt: "desc" }, { id: "desc" }]
77+
: [{ createdAt: "asc" }, { id: "asc" }],
78+
take: pageSize + 1,
79+
});
80+
81+
const hasMore = rows.length > pageSize;
82+
const pageRows = rows.slice(0, pageSize);
83+
const dataRows = direction === "forward" ? pageRows : [...pageRows].reverse();
84+
85+
const first = dataRows.at(0);
86+
const last = dataRows.at(-1);
87+
88+
return {
89+
data: dataRows.map(apiBulkActionObject),
90+
pagination: {
91+
next: last && (hasMore || direction === "backward") ? encodeCursor(last) : undefined,
92+
previous:
93+
first &&
94+
((direction === "forward" && Boolean(after)) || (direction === "backward" && hasMore))
95+
? encodeCursor(first)
96+
: undefined,
97+
},
98+
};
99+
}
100+
}
101+
102+
export const bulkActionSelect = {
103+
id: true,
104+
friendlyId: true,
105+
name: true,
106+
status: true,
107+
type: true,
108+
createdAt: true,
109+
completedAt: true,
110+
totalCount: true,
111+
successCount: true,
112+
failureCount: true,
113+
} as const;
114+
115+
export function apiBulkActionObject(row: BulkActionRow) {
116+
return {
117+
id: row.friendlyId,
118+
name: row.name ?? undefined,
119+
type: row.type as BulkActionType,
120+
status: row.status as BulkActionStatus,
121+
counts: {
122+
total: row.totalCount,
123+
success: row.successCount,
124+
failure: row.failureCount,
125+
},
126+
createdAt: row.createdAt,
127+
completedAt: row.completedAt ?? undefined,
128+
};
129+
}
130+
131+
function encodeCursor(row: Pick<BulkActionRow, "createdAt" | "id">) {
132+
return Buffer.from(JSON.stringify({ createdAt: row.createdAt.getTime(), id: row.id })).toString(
133+
"base64url"
134+
);
135+
}
136+
137+
function decodeCursor(cursor: string | undefined): BulkActionListCursor | undefined {
138+
if (!cursor) {
139+
return undefined;
140+
}
141+
142+
try {
143+
const parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as {
144+
createdAt?: unknown;
145+
id?: unknown;
146+
};
147+
if (typeof parsed.createdAt !== "number" || typeof parsed.id !== "string") {
148+
throw new Error("Invalid cursor");
149+
}
150+
151+
return { createdAt: new Date(parsed.createdAt), id: parsed.id };
152+
} catch {
153+
throw new Error("Invalid cursor");
154+
}
155+
}

‎apps/webapp/app/routes/api.v1.bulk-actions.$bulkActionId.abort.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
import { json } from "@remix-run/server-runtime";
2+
import { z } from "zod";
3+
import { $replica } from "~/db.server";
4+
import { logger } from "~/services/logger.server";
5+
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
6+
import { BulkActionService } from "~/v3/services/bulk/BulkActionV2.server";
7+
import { ServiceValidationError } from "~/v3/services/common.server";
8+
9+
const ParamsSchema = z.object({
10+
bulkActionId: z.string(),
11+
});
12+
13+
const { action } = createActionApiRoute(
14+
{
15+
params: ParamsSchema,
16+
corsStrategy: "none",
17+
authorization: {
18+
action: "write",
19+
resource: () => ({ type: "runs" }),
20+
},
21+
findResource: async (params, auth) => {
22+
return $replica.bulkActionGroup.findFirst({
23+
select: { id: true },
24+
where: {
25+
friendlyId: params.bulkActionId,
26+
environmentId: auth.environment.id,
27+
},
28+
});
29+
},
30+
},
31+
async ({ params, authentication }) => {
32+
const service = new BulkActionService();
33+
34+
try {
35+
const result = await service.abort(params.bulkActionId, authentication.environment.id);
36+
return json({ id: result.bulkActionId });
37+
} catch (error) {
38+
if (error instanceof ServiceValidationError) {
39+
return json({ error: error.message }, { status: error.status ?? 400 });
40+
}
41+
42+
logger.error("Failed to abort API bulk action", { error });
43+
return json({ error: "Failed to abort bulk action" }, { status: 500 });
44+
}
45+
}
46+
);
47+
48+
export { action };

‎apps/webapp/app/routes/api.v1.bulk-actions.$bulkActionId.ts‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
import { json } from "@remix-run/server-runtime";
2+
import { z } from "zod";
3+
import { $replica } from "~/db.server";
4+
import {
5+
apiBulkActionObject,
6+
bulkActionSelect,
7+
} from "~/presenters/v3/ApiBulkActionPresenter.server";
8+
import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
9+
10+
const ParamsSchema = z.object({
11+
bulkActionId: z.string(),
12+
});
13+
14+
export const loader = createLoaderApiRoute(
15+
{
16+
params: ParamsSchema,
17+
corsStrategy: "none",
18+
authorization: {
19+
action: "read",
20+
resource: () => ({ type: "runs" }),
21+
},
22+
findResource: async (params, auth) => {
23+
return $replica.bulkActionGroup.findFirst({
24+
select: bulkActionSelect,
25+
where: {
26+
friendlyId: params.bulkActionId,
27+
environmentId: auth.environment.id,
28+
},
29+
});
30+
},
31+
},
32+
async ({ resource }) => {
33+
return json(apiBulkActionObject(resource));
34+
}
35+
);

0 commit comments

Comments
 (0)