(null);
@@ -487,7 +489,7 @@ function TasksTreeView({
scrollToNode,
virtualizer,
} = useTree({
- tree: events,
+ tree: showDebug ? events : events.filter((event) => !event.data.isDebug),
selectedId,
// collapsedIds,
onSelectedIdChanged,
@@ -512,6 +514,15 @@ function TasksTreeView({
+ {isAdmin && (
+
setShowDebug(e.valueOf())}
+ />
+ )}
{
);
}
- logger.error("Failed to start a test run", { error: e });
+ logger.error("Failed to start a test run", { error: e instanceof Error ? e.message : e });
return redirectBackWithErrorMessage(
request,
diff --git a/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts b/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts
index 471481a16..3bb2ecf66 100644
--- a/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts
+++ b/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts
@@ -3,6 +3,7 @@ import { z } from "zod";
import { prisma } from "~/db.server";
import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
import { marqs } from "~/v3/marqs/index.server";
+import { updateEnvConcurrencyLimits } from "~/v3/runQueue.server";
const ParamsSchema = z.object({
environmentId: z.string(),
@@ -60,7 +61,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
},
});
- await marqs?.updateEnvConcurrencyLimits(environment);
+ await updateEnvConcurrencyLimits(environment);
return json({ success: true });
}
diff --git a/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.concurrency.ts b/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.concurrency.ts
index 51d292eb0..d6491bcc4 100644
--- a/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.concurrency.ts
+++ b/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.concurrency.ts
@@ -3,6 +3,7 @@ import { z } from "zod";
import { prisma } from "~/db.server";
import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
import { marqs } from "~/v3/marqs/index.server";
+import { updateEnvConcurrencyLimits } from "~/v3/runQueue.server";
const ParamsSchema = z.object({
organizationId: z.string(),
@@ -97,7 +98,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
},
});
- await marqs?.updateEnvConcurrencyLimits({ ...modifiedEnvironment, organization });
+ await updateEnvConcurrencyLimits({ ...modifiedEnvironment, organization });
}
return json({ success: true });
diff --git a/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.environments.staging.ts b/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.environments.staging.ts
index c4088257a..8483058f3 100644
--- a/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.environments.staging.ts
+++ b/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.environments.staging.ts
@@ -4,6 +4,7 @@ import { prisma } from "~/db.server";
import { createEnvironment } from "~/models/organization.server";
import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
import { marqs } from "~/v3/marqs/index.server";
+import { updateEnvConcurrencyLimits } from "~/v3/runQueue.server";
const ParamsSchema = z.object({
organizationId: z.string(),
@@ -58,10 +59,10 @@ export async function action({ request, params }: ActionFunctionArgs) {
if (!stagingEnvironment) {
const staging = await createEnvironment(organization, project, "STAGING");
- await marqs?.updateEnvConcurrencyLimits({ ...staging, organization, project });
+ await updateEnvConcurrencyLimits({ ...staging, organization, project });
created++;
} else {
- await marqs?.updateEnvConcurrencyLimits({ ...stagingEnvironment, organization, project });
+ await updateEnvConcurrencyLimits({ ...stagingEnvironment, organization, project });
}
}
diff --git a/apps/webapp/app/routes/admin.api.v1.workers.ts b/apps/webapp/app/routes/admin.api.v1.workers.ts
new file mode 100644
index 000000000..185c9cc4d
--- /dev/null
+++ b/apps/webapp/app/routes/admin.api.v1.workers.ts
@@ -0,0 +1,65 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { WorkerGroupService } from "~/v3/services/worker/workerGroupService.server";
+
+const RequestBodySchema = z.object({
+ name: z.string().optional(),
+ description: z.string().optional(),
+ projectId: z.string().optional(),
+ makeDefault: z.boolean().optional(),
+});
+
+export async function action({ request }: ActionFunctionArgs) {
+ // Next authenticate the request
+ const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+ if (!authenticationResult) {
+ return json({ error: "Invalid or Missing API key" }, { status: 401 });
+ }
+
+ const user = await prisma.user.findUnique({
+ where: {
+ id: authenticationResult.userId,
+ },
+ });
+
+ if (!user) {
+ return json({ error: "Invalid or Missing API key" }, { status: 401 });
+ }
+
+ if (!user.admin) {
+ return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+ }
+
+ try {
+ const rawBody = await request.json();
+ const { name, description, projectId, makeDefault } = RequestBodySchema.parse(rawBody ?? {});
+
+ const service = new WorkerGroupService();
+ const { workerGroup, token } = await service.createWorkerGroup({
+ name,
+ description,
+ });
+
+ if (makeDefault && projectId) {
+ await prisma.project.update({
+ where: {
+ id: projectId,
+ },
+ data: {
+ defaultWorkerGroupId: workerGroup.id,
+ engine: "V2",
+ },
+ });
+ }
+
+ return json({
+ token,
+ workerGroup,
+ });
+ } catch (error) {
+ return json({ error: error instanceof Error ? error.message : error }, { status: 400 });
+ }
+}
diff --git a/apps/webapp/app/routes/api.v1.batches.$batchId.ts b/apps/webapp/app/routes/api.v1.batches.$batchId.ts
index 365f9caa2..150978331 100644
--- a/apps/webapp/app/routes/api.v1.batches.$batchId.ts
+++ b/apps/webapp/app/routes/api.v1.batches.$batchId.ts
@@ -34,6 +34,7 @@ export const loader = createLoaderApiRoute(
createdAt: batch.createdAt,
updatedAt: batch.updatedAt,
runCount: batch.runCount,
+ runs: batch.runIds,
});
}
);
diff --git a/apps/webapp/app/routes/api.v1.deployments.latest.ts b/apps/webapp/app/routes/api.v1.deployments.latest.ts
new file mode 100644
index 000000000..6f31f58fc
--- /dev/null
+++ b/apps/webapp/app/routes/api.v1.deployments.latest.ts
@@ -0,0 +1,41 @@
+import { LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import { WorkerInstanceGroupType } from "@trigger.dev/database";
+import { prisma } from "~/db.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+
+export async function loader({ request }: LoaderFunctionArgs) {
+ // Next authenticate the request
+ const authenticationResult = await authenticateApiRequest(request);
+
+ if (!authenticationResult) {
+ logger.info("Invalid or missing api key", { url: request.url });
+ return json({ error: "Invalid or Missing API key" }, { status: 401 });
+ }
+
+ const authenticatedEnv = authenticationResult.environment;
+
+ const deployment = await prisma.workerDeployment.findFirst({
+ where: {
+ type: WorkerInstanceGroupType.UNMANAGED,
+ environmentId: authenticatedEnv.id,
+ },
+ orderBy: {
+ createdAt: "desc",
+ },
+ });
+
+ if (!deployment) {
+ return json({ error: "Deployment not found" }, { status: 404 });
+ }
+
+ return json({
+ id: deployment.friendlyId,
+ status: deployment.status,
+ contentHash: deployment.contentHash,
+ shortCode: deployment.shortCode,
+ version: deployment.version,
+ imageReference: deployment.imageReference,
+ errorData: deployment.errorData,
+ });
+}
diff --git a/apps/webapp/app/routes/api.v1.deployments.ts b/apps/webapp/app/routes/api.v1.deployments.ts
index 2f4b9bdd5..c3dcfb13d 100644
--- a/apps/webapp/app/routes/api.v1.deployments.ts
+++ b/apps/webapp/app/routes/api.v1.deployments.ts
@@ -6,6 +6,7 @@ import {
import { env } from "~/env.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
import { InitializeDeploymentService } from "~/v3/services/initializeDeployment.server";
export async function action({ request, params }: ActionFunctionArgs) {
@@ -33,18 +34,30 @@ export async function action({ request, params }: ActionFunctionArgs) {
const service = new InitializeDeploymentService();
- const { deployment, imageTag } = await service.call(authenticatedEnv, body.data);
+ try {
+ const { deployment, imageTag } = await service.call(authenticatedEnv, body.data);
- const responseBody: InitializeDeploymentResponseBody = {
- id: deployment.friendlyId,
- contentHash: deployment.contentHash,
- shortCode: deployment.shortCode,
- version: deployment.version,
- externalBuildData:
- deployment.externalBuildData as InitializeDeploymentResponseBody["externalBuildData"],
- imageTag,
- registryHost: body.data.registryHost ?? env.DEPLOY_REGISTRY_HOST,
- };
+ const responseBody: InitializeDeploymentResponseBody = {
+ id: deployment.friendlyId,
+ contentHash: deployment.contentHash,
+ shortCode: deployment.shortCode,
+ version: deployment.version,
+ externalBuildData:
+ deployment.externalBuildData as InitializeDeploymentResponseBody["externalBuildData"],
+ imageTag,
+ registryHost: body.data.registryHost ?? env.DEPLOY_REGISTRY_HOST,
+ };
- return json(responseBody, { status: 200 });
+ return json(responseBody, { status: 200 });
+ } catch (error) {
+ if (error instanceof ServiceValidationError) {
+ return json({ error: error.message }, { status: 400 });
+ } else if (error instanceof Error) {
+ logger.error("Error initializing deployment", { error: error.message });
+ return json({ error: `Internal server error: ${error.message}` }, { status: 500 });
+ } else {
+ logger.error("Error initializing deployment", { error: String(error) });
+ return json({ error: "Internal server error" }, { status: 500 });
+ }
+ }
}
diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts
index a55724344..cc364f343 100644
--- a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts
+++ b/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts
@@ -1,5 +1,9 @@
import { json } from "@remix-run/server-runtime";
-import { generateJWT as internal_generateJWT, TriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import {
+ generateJWT as internal_generateJWT,
+ RunEngineVersionSchema,
+ TriggerTaskRequestBody,
+} from "@trigger.dev/core/v3";
import { TaskRun } from "@trigger.dev/database";
import { z } from "zod";
import { env } from "~/env.server";
@@ -21,6 +25,7 @@ export const HeadersSchema = z.object({
"x-trigger-span-parent-as-link": z.coerce.number().nullish(),
"x-trigger-worker": z.string().nullish(),
"x-trigger-client": z.string().nullish(),
+ "x-trigger-engine-version": RunEngineVersionSchema.nullish(),
traceparent: z.string().optional(),
tracestate: z.string().optional(),
});
@@ -49,6 +54,7 @@ const { action, loader } = createActionApiRoute(
tracestate,
"x-trigger-worker": isFromWorker,
"x-trigger-client": triggerClient,
+ "x-trigger-engine-version": engineVersion,
} = headers;
const service = new TriggerTaskService();
@@ -74,14 +80,20 @@ const { action, loader } = createActionApiRoute(
const idempotencyKeyExpiresAt = resolveIdempotencyKeyTTL(idempotencyKeyTTL);
- const result = await service.call(params.taskId, authentication.environment, body, {
- idempotencyKey: idempotencyKey ?? undefined,
- idempotencyKeyExpiresAt: idempotencyKeyExpiresAt,
- triggerVersion: triggerVersion ?? undefined,
- traceContext,
- spanParentAsLink: spanParentAsLink === 1,
- oneTimeUseToken,
- });
+ const result = await service.call(
+ params.taskId,
+ authentication.environment,
+ body,
+ {
+ idempotencyKey: idempotencyKey ?? undefined,
+ idempotencyKeyExpiresAt: idempotencyKeyExpiresAt,
+ triggerVersion: triggerVersion ?? undefined,
+ traceContext,
+ spanParentAsLink: spanParentAsLink === 1,
+ oneTimeUseToken,
+ },
+ engineVersion ?? undefined
+ );
if (!result) {
return json({ error: "Task not found" }, { status: 404 });
diff --git a/apps/webapp/app/routes/api.v1.tasks.batch.ts b/apps/webapp/app/routes/api.v1.tasks.batch.ts
index 550442798..d26c55f1c 100644
--- a/apps/webapp/app/routes/api.v1.tasks.batch.ts
+++ b/apps/webapp/app/routes/api.v1.tasks.batch.ts
@@ -67,6 +67,7 @@ const { action, loader } = createActionApiRoute(
"x-trigger-span-parent-as-link": spanParentAsLink,
"x-trigger-worker": isFromWorker,
"x-trigger-client": triggerClient,
+ "x-trigger-engine-version": engineVersion,
"batch-processing-strategy": batchProcessingStrategy,
traceparent,
tracestate,
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts
new file mode 100644
index 000000000..e7109f67b
--- /dev/null
+++ b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts
@@ -0,0 +1,68 @@
+import { json } from "@remix-run/server-runtime";
+import {
+ CompleteWaitpointTokenRequestBody,
+ CompleteWaitpointTokenResponseBody,
+ conditionallyExportPacket,
+ stringifyIO,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/apps";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ params: z.object({
+ waitpointFriendlyId: z.string(),
+ }),
+ body: CompleteWaitpointTokenRequestBody,
+ maxContentLength: env.TASK_PAYLOAD_MAXIMUM_SIZE,
+ method: "POST",
+ },
+ async ({ authentication, body, params }) => {
+ // Resume tokens are actually just waitpoints
+ const waitpointId = WaitpointId.toId(params.waitpointFriendlyId);
+
+ try {
+ //check permissions
+ const waitpoint = await $replica.waitpoint.findFirst({
+ where: {
+ id: waitpointId,
+ environmentId: authentication.environment.id,
+ },
+ });
+
+ if (!waitpoint) {
+ throw json({ error: "Waitpoint not found" }, { status: 404 });
+ }
+
+ const stringifiedData = await stringifyIO(body.data);
+ const finalData = await conditionallyExportPacket(
+ stringifiedData,
+ `${waitpointId}/waitpoint/token`
+ );
+
+ const result = await engine.completeWaitpoint({
+ id: waitpointId,
+ output: finalData.data
+ ? { type: finalData.dataType, value: finalData.data, isError: false }
+ : undefined,
+ });
+
+ return json(
+ {
+ success: true,
+ },
+ { status: 200 }
+ );
+ } catch (error) {
+ logger.error("Failed to complete waitpoint token", { error });
+ throw json({ error: "Failed to complete waitpoint token" }, { status: 500 });
+ }
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts
new file mode 100644
index 000000000..7603551fe
--- /dev/null
+++ b/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts
@@ -0,0 +1,43 @@
+import { json } from "@remix-run/server-runtime";
+import {
+ CreateWaitpointTokenRequestBody,
+ CreateWaitpointTokenResponseBody,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/apps";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { parseDelay } from "~/utils/delays";
+import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ body: CreateWaitpointTokenRequestBody,
+ maxContentLength: 1024 * 10, // 10KB
+ method: "POST",
+ },
+ async ({ authentication, body }) => {
+ const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
+ ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
+ : undefined;
+
+ const timeout = await parseDelay(body.timeout);
+
+ const result = await engine.createManualWaitpoint({
+ environmentId: authentication.environment.id,
+ projectId: authentication.environment.projectId,
+ idempotencyKey: body.idempotencyKey,
+ idempotencyKeyExpiresAt,
+ timeout,
+ });
+
+ return json(
+ {
+ id: WaitpointId.toFriendlyId(result.waitpoint.id),
+ isCached: result.isCached,
+ },
+ { status: 200 }
+ );
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/api.v1.workers.ts b/apps/webapp/app/routes/api.v1.workers.ts
new file mode 100644
index 000000000..4008d64f1
--- /dev/null
+++ b/apps/webapp/app/routes/api.v1.workers.ts
@@ -0,0 +1,73 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import {
+ WorkersCreateRequestBody,
+ WorkersCreateResponseBody,
+ WorkersListResponseBody,
+} from "@trigger.dev/core/v3";
+import {
+ createActionApiRoute,
+ createLoaderApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+import { WorkerGroupService } from "~/v3/services/worker/workerGroupService.server";
+
+export const loader = createLoaderApiRoute(
+ {
+ corsStrategy: "all",
+ findResource: async () => 1, // This is a dummy function, we don't need to find a resource
+ },
+ async ({
+ authentication,
+ }): Promise> => {
+ if (authentication.environment.project.engine !== "V2") {
+ return json({ error: "Not supported for V1 projects" }, { status: 400 });
+ }
+
+ const service = new WorkerGroupService();
+ const workers = await service.listWorkerGroups({
+ projectId: authentication.environment.projectId,
+ });
+
+ return json(
+ workers.map((w) => ({
+ type: w.type,
+ name: w.name,
+ description: w.description,
+ isDefault: w.id === authentication.environment.project.defaultWorkerGroupId,
+ updatedAt: w.updatedAt,
+ }))
+ );
+ }
+);
+
+export const { action } = createActionApiRoute(
+ {
+ corsStrategy: "all",
+ body: WorkersCreateRequestBody,
+ },
+ async ({
+ authentication,
+ body,
+ }): Promise> => {
+ if (authentication.environment.project.engine !== "V2") {
+ return json({ error: "Not supported" }, { status: 400 });
+ }
+
+ const service = new WorkerGroupService();
+ const { workerGroup, token } = await service.createWorkerGroup({
+ projectId: authentication.environment.projectId,
+ organizationId: authentication.environment.organizationId,
+ name: body.name,
+ description: body.description,
+ });
+
+ return json({
+ token: {
+ plaintext: token.plaintext,
+ },
+ workerGroup: {
+ name: workerGroup.name,
+ description: workerGroup.description,
+ },
+ });
+ }
+);
diff --git a/apps/webapp/app/routes/api.v2.batches.$batchId.ts b/apps/webapp/app/routes/api.v2.batches.$batchId.ts
new file mode 100644
index 000000000..150978331
--- /dev/null
+++ b/apps/webapp/app/routes/api.v2.batches.$batchId.ts
@@ -0,0 +1,40 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+const ParamsSchema = z.object({
+ batchId: z.string(),
+});
+
+export const loader = createLoaderApiRoute(
+ {
+ params: ParamsSchema,
+ allowJWT: true,
+ corsStrategy: "all",
+ findResource: (params, auth) => {
+ return $replica.batchTaskRun.findFirst({
+ where: {
+ friendlyId: params.batchId,
+ runtimeEnvironmentId: auth.environment.id,
+ },
+ });
+ },
+ authorization: {
+ action: "read",
+ resource: (batch) => ({ batch: batch.friendlyId }),
+ superScopes: ["read:runs", "read:all", "admin"],
+ },
+ },
+ async ({ resource: batch }) => {
+ return json({
+ id: batch.friendlyId,
+ status: batch.status,
+ idempotencyKey: batch.idempotencyKey ?? undefined,
+ createdAt: batch.createdAt,
+ updatedAt: batch.updatedAt,
+ runCount: batch.runCount,
+ runs: batch.runIds,
+ });
+ }
+);
diff --git a/apps/webapp/app/routes/api.v2.tasks.batch.ts b/apps/webapp/app/routes/api.v2.tasks.batch.ts
new file mode 100644
index 000000000..e2c13e1aa
--- /dev/null
+++ b/apps/webapp/app/routes/api.v2.tasks.batch.ts
@@ -0,0 +1,152 @@
+import { json } from "@remix-run/server-runtime";
+import {
+ BatchTriggerTaskV3RequestBody,
+ BatchTriggerTaskV3Response,
+ generateJWT,
+} from "@trigger.dev/core/v3";
+import { env } from "~/env.server";
+import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { BatchProcessingStrategy } from "~/v3/services/batchTriggerV3.server";
+import { BatchTriggerV4Service } from "~/v3/services/batchTriggerV4.server";
+import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
+import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
+
+const { action, loader } = createActionApiRoute(
+ {
+ headers: HeadersSchema.extend({
+ "batch-processing-strategy": BatchProcessingStrategy.nullish(),
+ }),
+ body: BatchTriggerTaskV3RequestBody,
+ allowJWT: true,
+ maxContentLength: env.BATCH_TASK_PAYLOAD_MAXIMUM_SIZE,
+ authorization: {
+ action: "batchTrigger",
+ resource: (_, __, ___, body) => ({
+ tasks: Array.from(new Set(body.items.map((i) => i.task))),
+ }),
+ superScopes: ["write:tasks", "admin"],
+ },
+ corsStrategy: "all",
+ },
+ async ({ body, headers, params, authentication }) => {
+ if (!body.items.length) {
+ return json({ error: "Batch cannot be triggered with no items" }, { status: 400 });
+ }
+
+ // Check the there are fewer than MAX_BATCH_V2_TRIGGER_ITEMS items
+ if (body.items.length > env.MAX_BATCH_V2_TRIGGER_ITEMS) {
+ return json(
+ {
+ error: `Batch size of ${body.items.length} is too large. Maximum allowed batch size is ${env.MAX_BATCH_V2_TRIGGER_ITEMS}.`,
+ },
+ { status: 400 }
+ );
+ }
+
+ const {
+ "trigger-version": triggerVersion,
+ "x-trigger-span-parent-as-link": spanParentAsLink,
+ "x-trigger-worker": isFromWorker,
+ "x-trigger-client": triggerClient,
+ "x-trigger-engine-version": engineVersion,
+ "batch-processing-strategy": batchProcessingStrategy,
+ traceparent,
+ tracestate,
+ } = headers;
+
+ const oneTimeUseToken = await getOneTimeUseToken(authentication);
+
+ logger.debug("Batch trigger request", {
+ triggerVersion,
+ spanParentAsLink,
+ isFromWorker,
+ triggerClient,
+ traceparent,
+ tracestate,
+ batchProcessingStrategy,
+ });
+
+ const traceContext =
+ traceparent && isFromWorker // If the request is from a worker, we should pass the trace context
+ ? { traceparent, tracestate }
+ : undefined;
+
+ const service = new BatchTriggerV4Service(batchProcessingStrategy ?? undefined);
+
+ try {
+ const batch = await service.call(authentication.environment, body, {
+ triggerVersion: triggerVersion ?? undefined,
+ traceContext,
+ spanParentAsLink: spanParentAsLink === 1,
+ oneTimeUseToken,
+ });
+
+ const $responseHeaders = await responseHeaders(
+ batch,
+ authentication.environment,
+ triggerClient
+ );
+
+ return json(batch, { status: 202, headers: $responseHeaders });
+ } catch (error) {
+ logger.error("Batch trigger error", {
+ error: {
+ message: (error as Error).message,
+ stack: (error as Error).stack,
+ },
+ });
+
+ if (error instanceof ServiceValidationError) {
+ return json({ error: error.message }, { status: 422 });
+ } else if (error instanceof OutOfEntitlementError) {
+ return json({ error: error.message }, { status: 422 });
+ } else if (error instanceof Error) {
+ return json(
+ { error: error.message },
+ { status: 500, headers: { "x-should-retry": "false" } }
+ );
+ }
+
+ return json({ error: "Something went wrong" }, { status: 500 });
+ }
+ }
+);
+
+async function responseHeaders(
+ batch: BatchTriggerTaskV3Response,
+ environment: AuthenticatedEnvironment,
+ triggerClient?: string | null
+): Promise> {
+ const claimsHeader = JSON.stringify({
+ sub: environment.id,
+ pub: true,
+ });
+
+ if (triggerClient === "browser") {
+ const claims = {
+ sub: environment.id,
+ pub: true,
+ scopes: [`read:batch:${batch.id}`],
+ };
+
+ const jwt = await generateJWT({
+ secretKey: environment.apiKey,
+ payload: claims,
+ expirationTime: "1h",
+ });
+
+ return {
+ "x-trigger-jwt-claims": claimsHeader,
+ "x-trigger-jwt": jwt,
+ };
+ }
+
+ return {
+ "x-trigger-jwt-claims": claimsHeader,
+ };
+}
+
+export { action, loader };
diff --git a/apps/webapp/app/routes/engine.v1.dev.config.ts b/apps/webapp/app/routes/engine.v1.dev.config.ts
new file mode 100644
index 000000000..501b6e80d
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.config.ts
@@ -0,0 +1,32 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { DevConfigResponseBody } from "@trigger.dev/core/v3/schemas";
+import { z } from "zod";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const loader = createLoaderApiRoute(
+ {
+ findResource: async () => 1,
+ headers: z.object({
+ "x-forwarded-for": z.string().optional(),
+ }),
+ },
+ async ({ authentication }): Promise> => {
+ logger.debug("Get dev settings", { environmentId: authentication.environment.id });
+
+ try {
+ return json({
+ environmentId: authentication.environment.id,
+ dequeueIntervalWithRun: env.DEV_DEQUEUE_INTERVAL_WITH_RUN,
+ dequeueIntervalWithoutRun: env.DEV_DEQUEUE_INTERVAL_WITHOUT_RUN,
+ });
+ } catch (error) {
+ logger.error("Failed to get dev settings", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.dev.dequeue.ts b/apps/webapp/app/routes/engine.v1.dev.dequeue.ts
new file mode 100644
index 000000000..0f10c3dee
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.dequeue.ts
@@ -0,0 +1,92 @@
+import { json } from "@remix-run/server-runtime";
+import { DequeuedMessage, DevDequeueRequestBody, MachineResources } from "@trigger.dev/core/v3";
+import { BackgroundWorkerId } from "@trigger.dev/core/v3/apps";
+import { env } from "~/env.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ body: DevDequeueRequestBody,
+ maxContentLength: 1024 * 10, // 10KB
+ method: "POST",
+ },
+ async ({ authentication, body }) => {
+ //we won't return more runs than this in one API call
+ let maxDequeueCount = env.DEV_DEQUEUE_MAX_RUNS_PER_PULL;
+
+ //we can't use more than the max resources
+ const availableResources = body.maxResources ?? {
+ cpu: 8,
+ memory: 16,
+ };
+
+ let dequeuedMessages: DequeuedMessage[] = [];
+
+ //we need to check the current worker, because a run might have been locked to it
+ const workers = body.oldWorkers.concat(body.currentWorker);
+
+ //first we want to clear out old runs
+ for (const worker of workers) {
+ //dequeue
+ const latestResult = await engine.dequeueFromBackgroundWorkerMasterQueue({
+ consumerId: authentication.environment.id,
+ //specific version
+ backgroundWorkerId: BackgroundWorkerId.toId(worker),
+ maxRunCount: maxDequeueCount,
+ maxResources: availableResources,
+ });
+
+ //add runs to the array
+ dequeuedMessages.push(...latestResult);
+
+ //update availableResources
+ const consumedResources = latestResult.reduce(
+ (acc, r) => {
+ return {
+ cpu: acc.cpu + r.run.machine.cpu,
+ memory: acc.memory + r.run.machine.memory,
+ };
+ },
+ { cpu: 0, memory: 0 }
+ );
+ updateAvailableResources(availableResources, consumedResources);
+
+ //update maxDequeueCount
+ maxDequeueCount -= latestResult.length;
+
+ //if we have no resources left, we exit the loop
+ if (!hasAvailableResources(availableResources)) break;
+ //we've already dequeued the max number of runs
+ if (maxDequeueCount <= 0) break;
+ }
+
+ //dequeue from the current version if we still have space
+ if (hasAvailableResources(availableResources) && maxDequeueCount > 0) {
+ const latestResult = await engine.dequeueFromEnvironmentMasterQueue({
+ consumerId: authentication.environment.id,
+ //current dev version (no specific version specified)
+ environmentId: authentication.environment.id,
+ maxRunCount: maxDequeueCount,
+ maxResources: availableResources,
+ });
+ dequeuedMessages.push(...latestResult);
+ }
+
+ return json({ dequeuedMessages }, { status: 200 });
+ }
+);
+
+function updateAvailableResources(
+ availableResources: MachineResources,
+ resources: MachineResources
+) {
+ availableResources.cpu -= resources.cpu;
+ availableResources.memory -= resources.memory;
+}
+
+function hasAvailableResources(availableResources: MachineResources) {
+ return availableResources.cpu > 0 && availableResources.memory > 0;
+}
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.dev.presence.ts b/apps/webapp/app/routes/engine.v1.dev.presence.ts
new file mode 100644
index 000000000..9e3add8c7
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.presence.ts
@@ -0,0 +1,81 @@
+import { json } from "@remix-run/server-runtime";
+import { Redis } from "ioredis";
+import { env } from "~/env.server";
+import { DevPresenceStream } from "~/presenters/v3/DevPresenceStream.server";
+import { authenticateApiRequestWithFailure } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { createSSELoader } from "~/utils/sse";
+
+const redis = new Redis({
+ port: env.RUN_ENGINE_DEV_PRESENCE_REDIS_PORT ?? undefined,
+ host: env.RUN_ENGINE_DEV_PRESENCE_REDIS_HOST ?? undefined,
+ username: env.RUN_ENGINE_DEV_PRESENCE_REDIS_USERNAME ?? undefined,
+ password: env.RUN_ENGINE_DEV_PRESENCE_REDIS_PASSWORD ?? undefined,
+ enableAutoPipelining: true,
+ ...(env.RUN_ENGINE_DEV_PRESENCE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+});
+
+export const loader = createSSELoader({
+ timeout: env.DEV_PRESENCE_TTL_MS,
+ interval: env.DEV_PRESENCE_POLL_INTERVAL_MS,
+ debug: true,
+ handler: async ({ id, controller, debug, request }) => {
+ const authentication = await authenticateApiRequestWithFailure(request);
+
+ if (!authentication.ok) {
+ throw json({ error: "Invalid or Missing API key" }, { status: 401 });
+ }
+
+ const environmentId = authentication.environment.id;
+
+ const presenceKey = DevPresenceStream.getPresenceKey(environmentId);
+ const presenceChannel = DevPresenceStream.getPresenceChannel(environmentId);
+
+ return {
+ beforeStream: async () => {
+ logger.debug("Start dev presence SSE session", {
+ environmentId,
+ presenceKey,
+ presenceChannel,
+ });
+ },
+ initStream: async ({ send }) => {
+ //todo set a string instead, with the expire on the same call
+ //won't need multi
+
+ // Set initial presence with more context
+ await redis.setex(presenceKey, env.DEV_PRESENCE_TTL_MS / 1000, Date.now().toString());
+
+ // Publish presence update
+ await redis.publish(
+ presenceChannel,
+ JSON.stringify({
+ type: "connected",
+ environmentId,
+ timestamp: Date.now(),
+ })
+ );
+
+ send({ event: "start", data: `Started ${id}` });
+ },
+ iterator: async ({ send, date }) => {
+ await redis.setex(presenceKey, env.DEV_PRESENCE_TTL_MS / 1000, date.toISOString());
+
+ send({ event: "time", data: new Date().toISOString() });
+ },
+ cleanup: async () => {
+ await redis.del(presenceKey);
+
+ // Publish disconnect event
+ await redis.publish(
+ presenceChannel,
+ JSON.stringify({
+ type: "disconnected",
+ environmentId,
+ timestamp: Date.now(),
+ })
+ );
+ },
+ };
+ },
+});
diff --git a/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.logs.debug.ts b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.logs.debug.ts
new file mode 100644
index 000000000..d804f49b7
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.logs.debug.ts
@@ -0,0 +1,74 @@
+import { TypedResponse } from "@remix-run/server-runtime";
+import { assertExhaustive } from "@trigger.dev/core";
+import { RunId } from "@trigger.dev/core/v3/apps";
+import {
+ WorkerApiDebugLogBody,
+ WorkerApiRunAttemptStartResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { recordRunDebugLog } from "~/v3/eventRepository.server";
+
+const { action } = createActionApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ }),
+ body: WorkerApiDebugLogBody,
+ method: "POST",
+ },
+ async ({
+ authentication,
+ body,
+ params,
+ }): Promise> => {
+ const { runFriendlyId } = params;
+
+ try {
+ const run = await prisma.taskRun.findFirst({
+ where: {
+ friendlyId: params.runFriendlyId,
+ runtimeEnvironmentId: authentication.environment.id,
+ },
+ });
+
+ if (!run) {
+ throw new Response("You don't have permissions for this run", { status: 401 });
+ }
+
+ const eventResult = await recordRunDebugLog(
+ RunId.fromFriendlyId(runFriendlyId),
+ body.message,
+ {
+ attributes: {
+ properties: body.properties,
+ },
+ startTime: body.time,
+ }
+ );
+
+ if (eventResult.success) {
+ return new Response(null, { status: 204 });
+ }
+
+ switch (eventResult.code) {
+ case "FAILED_TO_RECORD_EVENT":
+ return new Response(null, { status: 400 }); // send a 400 to prevent retries
+ case "RUN_NOT_FOUND":
+ return new Response(null, { status: 404 });
+ default:
+ return assertExhaustive(eventResult.code);
+ }
+ } catch (error) {
+ logger.error("Failed to record dev log", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts
new file mode 100644
index 000000000..13752e36f
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts
@@ -0,0 +1,64 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { assertExhaustive } from "@trigger.dev/core";
+import { RunId, SnapshotId } from "@trigger.dev/core/v3/apps";
+import {
+ WorkerApiDebugLogBody,
+ WorkerApiRunAttemptCompleteRequestBody,
+ WorkerApiRunAttemptCompleteResponseBody,
+ WorkerApiRunAttemptStartResponseBody,
+ WorkloadHeartbeatResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { recordRunDebugLog } from "~/v3/eventRepository.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ body: WorkerApiRunAttemptCompleteRequestBody,
+ method: "POST",
+ },
+ async ({
+ authentication,
+ body,
+ params,
+ }): Promise> => {
+ const { completion } = body;
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ try {
+ const run = await prisma.taskRun.findFirst({
+ where: {
+ friendlyId: params.runFriendlyId,
+ runtimeEnvironmentId: authentication.environment.id,
+ },
+ });
+
+ if (!run) {
+ throw new Response("You don't have permissions for this run", { status: 401 });
+ }
+
+ const completeResult = await engine.completeRunAttempt({
+ runId: RunId.toId(runFriendlyId),
+ snapshotId: SnapshotId.toId(snapshotFriendlyId),
+ completion,
+ });
+
+ return json({ result: completeResult });
+ } catch (error) {
+ logger.error("Failed to complete dev attempt", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts
new file mode 100644
index 000000000..25787eee0
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts
@@ -0,0 +1,103 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { MachinePreset } from "@trigger.dev/core/v3";
+import { RunId, SnapshotId } from "@trigger.dev/core/v3/apps";
+import {
+ WorkerApiRunAttemptStartRequestBody,
+ WorkerApiRunAttemptStartResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { RuntimeEnvironment } from "@trigger.dev/database";
+import { defaultMachine } from "@trigger.dev/platform/v3";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { generateJWTTokenForEnvironment } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import {
+ createActionApiRoute,
+ createActionWorkerApiRoute,
+} from "~/services/routeBuilders/apiBuilder.server";
+import { resolveVariablesForEnvironment } from "~/v3/environmentVariables/environmentVariablesRepository.server";
+import { machinePresetFromName } from "~/v3/machinePresets.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ body: WorkerApiRunAttemptStartRequestBody,
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ method: "POST",
+ },
+ async ({
+ authentication,
+ body,
+ params,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ try {
+ const run = await prisma.taskRun.findFirst({
+ where: {
+ friendlyId: params.runFriendlyId,
+ runtimeEnvironmentId: authentication.environment.id,
+ },
+ });
+
+ if (!run) {
+ throw new Response("You don't have permissions for this run", { status: 401 });
+ }
+
+ const engineResult = await engine.startRunAttempt({
+ runId: RunId.toId(runFriendlyId),
+ snapshotId: SnapshotId.toId(snapshotFriendlyId),
+ });
+
+ const defaultMachinePreset = machinePresetFromName(defaultMachine);
+
+ const envVars = await getEnvVars(
+ authentication.environment,
+ engineResult.run.id,
+ engineResult.execution.machine ?? defaultMachinePreset
+ );
+
+ return json({
+ ...engineResult,
+ envVars,
+ });
+ } catch (error) {
+ logger.error("Failed to record dev log", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
+
+async function getEnvVars(
+ environment: RuntimeEnvironment,
+ runId: string,
+ machinePreset: MachinePreset
+): Promise> {
+ const variables = await resolveVariablesForEnvironment(environment);
+
+ const jwt = await generateJWTTokenForEnvironment(environment, {
+ run_id: runId,
+ machine_preset: machinePreset.name,
+ });
+
+ variables.push(
+ ...[
+ { key: "TRIGGER_JWT", value: jwt },
+ { key: "TRIGGER_RUN_ID", value: runId },
+ { key: "TRIGGER_MACHINE_PRESET", value: machinePreset.name },
+ ]
+ );
+
+ return variables.reduce((acc: Record, curr) => {
+ acc[curr.key] = curr.value;
+ return acc;
+ }, {});
+}
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.heartbeat.ts b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.heartbeat.ts
new file mode 100644
index 000000000..2bf88ff1a
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.heartbeat.ts
@@ -0,0 +1,59 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { assertExhaustive } from "@trigger.dev/core";
+import { RunId, SnapshotId } from "@trigger.dev/core/v3/apps";
+import {
+ WorkerApiDebugLogBody,
+ WorkerApiRunAttemptStartResponseBody,
+ WorkloadHeartbeatResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { recordRunDebugLog } from "~/v3/eventRepository.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ method: "POST",
+ },
+ async ({
+ authentication,
+ body,
+ params,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ try {
+ const run = await prisma.taskRun.findFirst({
+ where: {
+ friendlyId: params.runFriendlyId,
+ runtimeEnvironmentId: authentication.environment.id,
+ },
+ });
+
+ if (!run) {
+ throw new Response("You don't have permissions for this run", { status: 401 });
+ }
+
+ await engine.heartbeatRun({
+ runId: RunId.toId(runFriendlyId),
+ snapshotId: SnapshotId.toId(snapshotFriendlyId),
+ });
+
+ return json({ ok: true });
+ } catch (error) {
+ logger.error("Failed to heartbeat dev run", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.latest.ts b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.latest.ts
new file mode 100644
index 000000000..91664888a
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.latest.ts
@@ -0,0 +1,55 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { RunId } from "@trigger.dev/core/v3/apps";
+import { WorkerApiRunLatestSnapshotResponseBody } from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { engine } from "~/v3/runEngine.server";
+
+export const loader = createLoaderApiRoute(
+ {
+ findResource: async () => 1,
+ params: z.object({
+ runFriendlyId: z.string(),
+ }),
+ },
+ async ({
+ authentication,
+ params,
+ }): Promise> => {
+ logger.debug("dev: Get latest snapshot", {
+ environmentId: authentication.environment.id,
+ params,
+ });
+
+ try {
+ const run = await prisma.taskRun.findFirst({
+ where: {
+ friendlyId: params.runFriendlyId,
+ runtimeEnvironmentId: authentication.environment.id,
+ },
+ });
+
+ if (!run) {
+ throw new Response("You don't have permissions for this run", { status: 401 });
+ }
+
+ const executionData = await engine.getRunExecutionData({
+ runId: RunId.toId(params.runFriendlyId),
+ });
+
+ if (!executionData) {
+ throw new Error("Failed to retrieve latest snapshot");
+ }
+
+ return json({ execution: executionData });
+ } catch (error) {
+ logger.error("Failed to get latest snapshot", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.wait.duration.ts b/apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.wait.duration.ts
new file mode 100644
index 000000000..04ef8c5aa
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.wait.duration.ts
@@ -0,0 +1,75 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WaitForDurationRequestBody, WaitForDurationResponseBody } from "@trigger.dev/core/v3";
+import { RunId } from "@trigger.dev/core/v3/apps";
+
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ body: WaitForDurationRequestBody,
+ params: z.object({
+ runFriendlyId: z.string(),
+ }),
+ method: "POST",
+ },
+ async ({ authentication, body, params }): Promise> => {
+ const { runFriendlyId } = params;
+ const runId = RunId.toId(runFriendlyId);
+
+ try {
+ const run = await prisma.taskRun.findFirst({
+ where: {
+ id: runId,
+ runtimeEnvironmentId: authentication.environment.id,
+ },
+ });
+
+ if (!run) {
+ throw new Response("You don't have permissions for this run", { status: 401 });
+ }
+
+ const idempotencyKeyExpiresAt = body.idempotencyKeyTTL
+ ? resolveIdempotencyKeyTTL(body.idempotencyKeyTTL)
+ : undefined;
+
+ const { waitpoint } = await engine.createDateTimeWaitpoint({
+ projectId: authentication.environment.project.id,
+ environmentId: authentication.environment.id,
+ completedAfter: body.date,
+ idempotencyKey: body.idempotencyKey,
+ idempotencyKeyExpiresAt: idempotencyKeyExpiresAt,
+ });
+
+ const waitResult = await engine.blockRunWithWaitpoint({
+ runId: run.id,
+ waitpoints: waitpoint.id,
+ environmentId: authentication.environment.id,
+ projectId: authentication.environment.project.id,
+ organizationId: authentication.environment.organization.id,
+ releaseConcurrency: {
+ releaseQueue: true,
+ },
+ });
+
+ return json({
+ waitUntil: body.date,
+ waitpoint: {
+ id: waitpoint.friendlyId,
+ },
+ });
+ } catch (error) {
+ logger.error("Failed to wait for duration dev", {
+ environmentId: authentication.environment.id,
+ error,
+ });
+ throw error;
+ }
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.waitpoints.tokens.$waitpointFriendlyId.wait.ts b/apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.waitpoints.tokens.$waitpointFriendlyId.wait.ts
new file mode 100644
index 000000000..e34e25529
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.runs.$runFriendlyId.waitpoints.tokens.$waitpointFriendlyId.wait.ts
@@ -0,0 +1,58 @@
+import { json } from "@remix-run/server-runtime";
+import { WaitForWaitpointTokenResponseBody } from "@trigger.dev/core/v3";
+import { RunId, WaitpointId } from "@trigger.dev/core/v3/apps";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { engine } from "~/v3/runEngine.server";
+
+const { action } = createActionApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ waitpointFriendlyId: z.string(),
+ }),
+ maxContentLength: 1024 * 10, // 10KB
+ method: "POST",
+ },
+ async ({ authentication, body, params }) => {
+ // Resume tokens are actually just waitpoints
+ const waitpointId = WaitpointId.toId(params.waitpointFriendlyId);
+ const runId = RunId.toId(params.runFriendlyId);
+
+ try {
+ //check permissions
+ const waitpoint = await $replica.waitpoint.findFirst({
+ where: {
+ id: waitpointId,
+ environmentId: authentication.environment.id,
+ },
+ });
+
+ if (!waitpoint) {
+ throw json({ error: "Waitpoint not found" }, { status: 404 });
+ }
+
+ const result = await engine.blockRunWithWaitpoint({
+ runId,
+ waitpoints: [waitpointId],
+ environmentId: authentication.environment.id,
+ projectId: authentication.environment.project.id,
+ organizationId: authentication.environment.organization.id,
+ });
+
+ return json(
+ {
+ success: true,
+ },
+ { status: 200 }
+ );
+ } catch (error) {
+ logger.error("Failed to wait for waitpoint", { runId, waitpointId, error });
+ throw json({ error: "Failed to wait for waitpoint token" }, { status: 500 });
+ }
+ }
+);
+
+export { action };
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.connect.ts b/apps/webapp/app/routes/engine.v1.worker-actions.connect.ts
new file mode 100644
index 000000000..247984d45
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.connect.ts
@@ -0,0 +1,19 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkerApiConnectRequestBody, WorkerApiConnectResponseBody } from "@trigger.dev/core/v3/workers";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ body: WorkerApiConnectRequestBody,
+ },
+ async ({ authenticatedWorker, body }): Promise> => {
+ await authenticatedWorker.connect(body.metadata);
+ return json({
+ ok: true,
+ workerGroup: {
+ type: authenticatedWorker.type,
+ name: authenticatedWorker.name,
+ },
+ });
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.deployments.$deploymentFriendlyId.dequeue.ts b/apps/webapp/app/routes/engine.v1.worker-actions.deployments.$deploymentFriendlyId.dequeue.ts
new file mode 100644
index 000000000..fbfa19466
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.deployments.$deploymentFriendlyId.dequeue.ts
@@ -0,0 +1,68 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { CURRENT_DEPLOYMENT_LABEL } from "@trigger.dev/core/v3/apps";
+import { WorkerApiDequeueResponseBody } from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { $replica, prisma } from "~/db.server";
+import { createLoaderWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const loader = createLoaderWorkerApiRoute(
+ {
+ params: z.object({
+ deploymentFriendlyId: z.string(),
+ }),
+ searchParams: z.object({
+ maxRunCount: z.coerce.number().optional(),
+ }),
+ },
+ async ({
+ authenticatedWorker,
+ params,
+ searchParams,
+ }): Promise> => {
+ const deployment = await $replica.workerDeployment.findUnique({
+ where: {
+ friendlyId: params.deploymentFriendlyId,
+ },
+ include: {
+ worker: true,
+ },
+ });
+
+ if (!deployment) {
+ throw new Error("Deployment not found");
+ }
+
+ if (!deployment.worker) {
+ throw new Error("Worker not found");
+ }
+
+ const dequeuedMessages = (await isCurrentDeployment(deployment.id, deployment.environmentId))
+ ? await authenticatedWorker.dequeueFromEnvironment(
+ deployment.worker.id,
+ deployment.environmentId
+ )
+ : await authenticatedWorker.dequeueFromVersion(
+ deployment.worker.id,
+ searchParams.maxRunCount
+ );
+
+ return json(dequeuedMessages);
+ }
+);
+
+async function isCurrentDeployment(deploymentId: string, environmentId: string): Promise {
+ const promotion = await prisma.workerDeploymentPromotion.findUnique({
+ where: {
+ environmentId_label: {
+ environmentId,
+ label: CURRENT_DEPLOYMENT_LABEL,
+ },
+ },
+ });
+
+ if (!promotion) {
+ return false;
+ }
+
+ return promotion.deploymentId === deploymentId;
+}
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.dequeue.ts b/apps/webapp/app/routes/engine.v1.worker-actions.dequeue.ts
new file mode 100644
index 000000000..4dd0798ad
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.dequeue.ts
@@ -0,0 +1,16 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkerApiDequeueRequestBody, WorkerApiDequeueResponseBody } from "@trigger.dev/core/v3/workers";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ body: WorkerApiDequeueRequestBody,
+ },
+ async ({ authenticatedWorker, body }): Promise> => {
+ return json(
+ await authenticatedWorker.dequeue({
+ maxResources: body.maxResources,
+ })
+ );
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.heartbeat.ts b/apps/webapp/app/routes/engine.v1.worker-actions.heartbeat.ts
new file mode 100644
index 000000000..111cab601
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.heartbeat.ts
@@ -0,0 +1,13 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkerApiHeartbeatResponseBody, WorkerApiHeartbeatRequestBody } from "@trigger.dev/core/v3/workers";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ body: WorkerApiHeartbeatRequestBody,
+ },
+ async ({ authenticatedWorker }): Promise> => {
+ await authenticatedWorker.heartbeatWorkerInstance();
+ return json({ ok: true });
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.logs.debug.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.logs.debug.ts
new file mode 100644
index 000000000..2ec26906b
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.logs.debug.ts
@@ -0,0 +1,38 @@
+import { assertExhaustive } from "@trigger.dev/core";
+import { RunId } from "@trigger.dev/core/v3/apps";
+import { WorkerApiDebugLogBody } from "@trigger.dev/core/v3/runEngineWorker";
+import { z } from "zod";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { recordRunDebugLog } from "~/v3/eventRepository.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ }),
+ body: WorkerApiDebugLogBody,
+ },
+ async ({ body, params }): Promise => {
+ const { runFriendlyId } = params;
+
+ const eventResult = await recordRunDebugLog(RunId.fromFriendlyId(runFriendlyId), body.message, {
+ attributes: {
+ properties: body.properties,
+ },
+ startTime: body.time,
+ });
+
+ if (eventResult.success) {
+ return new Response(null, { status: 204 });
+ }
+
+ switch (eventResult.code) {
+ case "FAILED_TO_RECORD_EVENT":
+ return new Response(null, { status: 400 }); // send a 400 to prevent retries
+ case "RUN_NOT_FOUND":
+ return new Response(null, { status: 404 });
+ default:
+ return assertExhaustive(eventResult.code);
+ }
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts
new file mode 100644
index 000000000..81f53280e
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts
@@ -0,0 +1,33 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import {
+ WorkerApiRunAttemptCompleteRequestBody,
+ WorkerApiRunAttemptCompleteResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ body: WorkerApiRunAttemptCompleteRequestBody,
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ },
+ async ({
+ authenticatedWorker,
+ body,
+ params,
+ }): Promise> => {
+ const { completion } = body;
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ const completeResult = await authenticatedWorker.completeRunAttempt({
+ runFriendlyId,
+ snapshotFriendlyId,
+ completion,
+ });
+
+ return json({ result: completeResult });
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts
new file mode 100644
index 000000000..f7a8e874e
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts
@@ -0,0 +1,32 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import {
+ WorkerApiRunAttemptStartRequestBody,
+ WorkerApiRunAttemptStartResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ body: WorkerApiRunAttemptStartRequestBody,
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ },
+ async ({
+ authenticatedWorker,
+ body,
+ params,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ const runExecutionData = await authenticatedWorker.startRunAttempt({
+ runFriendlyId,
+ snapshotFriendlyId,
+ isWarmStart: body.isWarmStart,
+ });
+
+ return json(runExecutionData);
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.continue.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.continue.ts
new file mode 100644
index 000000000..b35f26a10
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.continue.ts
@@ -0,0 +1,34 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkerApiContinueRunExecutionRequestBody } from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { createLoaderWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const loader = createLoaderWorkerApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ },
+ async ({
+ authenticatedWorker,
+ params,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ logger.debug("Continuing run execution", { runFriendlyId, snapshotFriendlyId });
+
+ try {
+ const continuationResult = await authenticatedWorker.continueRunExecution({
+ runFriendlyId,
+ snapshotFriendlyId,
+ });
+
+ return json(continuationResult);
+ } catch (error) {
+ logger.error("Failed to suspend run", { runFriendlyId, snapshotFriendlyId, error });
+ throw error;
+ }
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.heartbeat.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.heartbeat.ts
new file mode 100644
index 000000000..0942bcde8
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.heartbeat.ts
@@ -0,0 +1,26 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkloadHeartbeatResponseBody } from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ },
+ async ({
+ authenticatedWorker,
+ params,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ await authenticatedWorker.heartbeatRun({
+ runFriendlyId,
+ snapshotFriendlyId,
+ });
+
+ return json({ ok: true });
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.restore.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.restore.ts
new file mode 100644
index 000000000..8f892288e
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.restore.ts
@@ -0,0 +1,51 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import {
+ WorkerApiSuspendRunRequestBody,
+ WorkerApiSuspendRunResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ body: WorkerApiSuspendRunRequestBody,
+ },
+ async ({
+ authenticatedWorker,
+ params,
+ body,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ logger.debug("Restoring run", { runFriendlyId, snapshotFriendlyId, body });
+
+ if (!body.success) {
+ // TODO: we could create a debug span here
+ logger.error("Failed to restore run", {
+ runFriendlyId,
+ snapshotFriendlyId,
+ error: body.error,
+ });
+
+ return json({ ok: true });
+ }
+
+ try {
+ await authenticatedWorker.createCheckpoint({
+ runFriendlyId,
+ snapshotFriendlyId,
+ checkpoint: body.checkpoint,
+ });
+
+ return json({ ok: true });
+ } catch (error) {
+ logger.error("Failed to restore run", { runFriendlyId, snapshotFriendlyId, error });
+ throw error;
+ }
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.suspend.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.suspend.ts
new file mode 100644
index 000000000..323c98405
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.suspend.ts
@@ -0,0 +1,51 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import {
+ WorkerApiSuspendRunRequestBody,
+ WorkerApiSuspendRunResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const action = createActionWorkerApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ snapshotFriendlyId: z.string(),
+ }),
+ body: WorkerApiSuspendRunRequestBody,
+ },
+ async ({
+ authenticatedWorker,
+ params,
+ body,
+ }): Promise> => {
+ const { runFriendlyId, snapshotFriendlyId } = params;
+
+ logger.debug("Suspending run", { runFriendlyId, snapshotFriendlyId, body });
+
+ if (!body.success) {
+ // TODO: we could create a debug span here
+ logger.error("Failed to suspend run", {
+ runFriendlyId,
+ snapshotFriendlyId,
+ error: body.error,
+ });
+
+ return json({ ok: true });
+ }
+
+ try {
+ await authenticatedWorker.createCheckpoint({
+ runFriendlyId,
+ snapshotFriendlyId,
+ checkpoint: body.checkpoint,
+ });
+
+ return json({ ok: true });
+ } catch (error) {
+ logger.error("Failed to suspend run", { runFriendlyId, snapshotFriendlyId, error });
+ throw error;
+ }
+ }
+);
diff --git a/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.latest.ts b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.latest.ts
new file mode 100644
index 000000000..d480acf01
--- /dev/null
+++ b/apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.latest.ts
@@ -0,0 +1,28 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkerApiRunLatestSnapshotResponseBody } from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { createLoaderWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const loader = createLoaderWorkerApiRoute(
+ {
+ params: z.object({
+ runFriendlyId: z.string(),
+ }),
+ },
+ async ({
+ authenticatedWorker,
+ params,
+ }): Promise> => {
+ const { runFriendlyId } = params;
+
+ const executionData = await authenticatedWorker.getLatestSnapshot({
+ runFriendlyId,
+ });
+
+ if (!executionData) {
+ throw new Error("Failed to retrieve latest snapshot");
+ }
+
+ return json({ execution: executionData });
+ }
+);
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.waitpoints.$waitpointFriendlyId.complete/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.waitpoints.$waitpointFriendlyId.complete/route.tsx
new file mode 100644
index 000000000..56ebef072
--- /dev/null
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.waitpoints.$waitpointFriendlyId.complete/route.tsx
@@ -0,0 +1,410 @@
+import { env } from "~/env.server";
+import { parse } from "@conform-to/zod";
+import { Form, useLocation, useNavigation, useSubmit } from "@remix-run/react";
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import {
+ conditionallyExportPacket,
+ IOPacket,
+ stringifyIO,
+ timeoutError,
+} from "@trigger.dev/core/v3";
+import { WaitpointId } from "@trigger.dev/core/v3/apps";
+import { Waitpoint } from "@trigger.dev/database";
+import { useCallback, useRef } from "react";
+import { z } from "zod";
+import { AnimatedHourglassIcon } from "~/assets/icons/AnimatedHourglassIcon";
+import { JSONEditor } from "~/components/code/JSONEditor";
+import { Button } from "~/components/primitives/Buttons";
+import { DateTime } from "~/components/primitives/DateTime";
+import { Paragraph } from "~/components/primitives/Paragraph";
+import { InfoIconTooltip } from "~/components/primitives/Tooltip";
+import { LiveCountdown } from "~/components/runs/v3/LiveTimer";
+import { $replica } from "~/db.server";
+import { useOrganization } from "~/hooks/useOrganizations";
+import { useProject } from "~/hooks/useProject";
+import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
+import { logger } from "~/services/logger.server";
+import { requireUserId } from "~/services/session.server";
+import { ProjectParamSchema, v3RunsPath } from "~/utils/pathBuilder";
+import { engine } from "~/v3/runEngine.server";
+
+const CompleteWaitpointFormData = z.discriminatedUnion("type", [
+ z.object({
+ type: z.literal("MANUAL"),
+ payload: z.string().optional(),
+ isTimeout: z.string().optional(),
+ successRedirect: z.string(),
+ failureRedirect: z.string(),
+ }),
+ z.object({
+ type: z.literal("DATETIME"),
+ successRedirect: z.string(),
+ failureRedirect: z.string(),
+ }),
+]);
+
+const Params = ProjectParamSchema.extend({
+ waitpointFriendlyId: z.string(),
+});
+
+export const action = async ({ request, params }: ActionFunctionArgs) => {
+ const userId = await requireUserId(request);
+ const { organizationSlug, projectParam, waitpointFriendlyId } = Params.parse(params);
+
+ const formData = await request.formData();
+ const submission = parse(formData, { schema: CompleteWaitpointFormData });
+
+ if (!submission.value) {
+ return json(submission);
+ }
+
+ try {
+ //first check that the user has access to the project
+ const project = await $replica.project.findUnique({
+ where: {
+ slug: projectParam,
+ organization: {
+ members: {
+ some: {
+ userId,
+ },
+ },
+ },
+ },
+ select: { id: true },
+ });
+
+ if (!project) {
+ throw new Error("Project not found");
+ }
+
+ const waitpointId = WaitpointId.toId(waitpointFriendlyId);
+
+ const waitpoint = await $replica.waitpoint.findFirst({
+ select: {
+ projectId: true,
+ },
+ where: {
+ id: waitpointId,
+ },
+ });
+
+ if (waitpoint?.projectId !== project.id) {
+ return redirectWithErrorMessage(
+ submission.value.failureRedirect,
+ request,
+ "No waitpoint found"
+ );
+ }
+
+ switch (submission.value.type) {
+ case "DATETIME": {
+ const result = await engine.completeWaitpoint({
+ id: waitpointId,
+ });
+
+ return redirectWithSuccessMessage(
+ submission.value.successRedirect,
+ request,
+ "Waitpoint skipped"
+ );
+ }
+ case "MANUAL": {
+ if (submission.value.isTimeout) {
+ try {
+ const result = await engine.completeWaitpoint({
+ id: waitpointId,
+ output: {
+ type: "application/json",
+ value: JSON.stringify(timeoutError(new Date())),
+ isError: true,
+ },
+ });
+
+ return redirectWithSuccessMessage(
+ submission.value.successRedirect,
+ request,
+ "Waitpoint timed out"
+ );
+ } catch (e) {
+ return redirectWithErrorMessage(
+ submission.value.failureRedirect,
+ request,
+ "Invalid payload, must be valid JSON"
+ );
+ }
+ }
+
+ try {
+ if (
+ submission.value.payload &&
+ submission.value.payload.length > env.TASK_PAYLOAD_MAXIMUM_SIZE
+ ) {
+ return redirectWithErrorMessage(
+ submission.value.failureRedirect,
+ request,
+ "Payload is too large"
+ );
+ }
+
+ const data = submission.value.payload ? JSON.parse(submission.value.payload) : {};
+ const stringifiedData = await stringifyIO(data);
+ const finalData = await conditionallyExportPacket(
+ stringifiedData,
+ `${waitpointId}/waitpoint/token`
+ );
+
+ const result = await engine.completeWaitpoint({
+ id: waitpointId,
+ output: finalData.data
+ ? { type: finalData.dataType, value: finalData.data, isError: false }
+ : undefined,
+ });
+
+ return redirectWithSuccessMessage(
+ submission.value.successRedirect,
+ request,
+ "Waitpoint completed"
+ );
+ } catch (e) {
+ return redirectWithErrorMessage(
+ submission.value.failureRedirect,
+ request,
+ "Invalid payload, must be valid JSON"
+ );
+ }
+ }
+ }
+ } catch (error: any) {
+ logger.error("Failed to complete waitpoint", error);
+
+ const errorMessage = `Something went wrong. Please try again.`;
+ return redirectWithErrorMessage(
+ v3RunsPath({ slug: organizationSlug }, { slug: projectParam }),
+ request,
+ errorMessage
+ );
+ }
+};
+
+type FormWaitpoint = Pick;
+
+export function CompleteWaitpointForm({ waitpoint }: { waitpoint: FormWaitpoint }) {
+ const navigation = useNavigation();
+ const submit = useSubmit();
+ const isLoading = navigation.state !== "idle";
+ const organization = useOrganization();
+ const project = useProject();
+
+ return (
+
+ {waitpoint.type === "DATETIME" ? (
+ waitpoint.completedAfter ? (
+
+ ) : (
+ <>Waitpoint doesn't have a complete date
+ )
+ ) : (
+
+ )}
+
+ );
+}
+
+function CompleteDateTimeWaitpointForm({
+ waitpoint,
+}: {
+ waitpoint: { friendlyId: string; completedAfter: Date };
+}) {
+ const location = useLocation();
+ const navigation = useNavigation();
+ const isLoading = navigation.state !== "idle";
+ const organization = useOrganization();
+ const project = useProject();
+
+ const timeToComplete = waitpoint.completedAfter.getTime() - Date.now();
+ if (timeToComplete < 0) {
+ return (
+
+ );
+ }
+
+ return (
+
+ );
+}
+
+function CompleteManualWaitpointForm({ waitpoint }: { waitpoint: { friendlyId: string } }) {
+ const location = useLocation();
+ const navigation = useNavigation();
+ const submit = useSubmit();
+ const isLoading = navigation.state !== "idle";
+ const organization = useOrganization();
+ const project = useProject();
+ const currentJson = useRef("{\n\n}");
+ const formAction = `/resources/orgs/${organization.slug}/projects/${project.slug}/waitpoints/${waitpoint.friendlyId}/complete`;
+
+ const submitForm = useCallback(
+ (e: React.FormEvent) => {
+ const formData = new FormData(e.currentTarget);
+ const data: Record = {
+ type: formData.get("type") as string,
+ failureRedirect: formData.get("failureRedirect") as string,
+ successRedirect: formData.get("successRedirect") as string,
+ };
+
+ data.payload = currentJson.current;
+
+ submit(data, {
+ action: formAction,
+ method: "post",
+ });
+ e.preventDefault();
+ },
+ [currentJson]
+ );
+
+ return (
+ <>
+
+
+ );
+}
+
+export function ForceTimeout({ waitpoint }: { waitpoint: { friendlyId: string } }) {
+ const location = useLocation();
+ const navigation = useNavigation();
+ const isLoading = navigation.state !== "idle";
+ const organization = useOrganization();
+ const project = useProject();
+ const formAction = `/resources/orgs/${organization.slug}/projects/${project.slug}/waitpoints/${waitpoint.friendlyId}/complete`;
+
+ return (
+
+ );
+}
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
index 04afd2c37..805cf89b9 100644
--- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
@@ -59,6 +59,7 @@ import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import { formatCurrencyAccurate } from "~/utils/numberFormatter";
import {
+ docsPath,
v3BatchPath,
v3RunDownloadLogsPath,
v3RunPath,
@@ -67,6 +68,11 @@ import {
v3SchedulePath,
v3SpanParamsSchema,
} from "~/utils/pathBuilder";
+import { SpanLink } from "~/v3/eventRepository.server";
+import {
+ CompleteWaitpointForm,
+ ForceTimeout,
+} from "../resources.orgs.$organizationSlug.projects.$projectParam.waitpoints.$waitpointFriendlyId.complete/route";
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
@@ -170,7 +176,6 @@ function SpanBody({
runParam?: string;
closePanel?: () => void;
}) {
- const isAdmin = useHasAdminAccess();
const organization = useOrganization();
const project = useProject();
const { value, replace } = useSearchParams();
@@ -190,7 +195,7 @@ function SpanBody({
className="h-4 min-h-4 w-4 min-w-4"
/>
-
+
{runParam && closePanel && (
@@ -226,179 +231,67 @@ function SpanBody({
-
-
- {tab === "detail" ? (
-
-
-
- Status
-
-
-
-
-
- Task
-
-
- {span.taskSlug}
-
- }
- content={`Filter runs by ${span.taskSlug}`}
- />
-
-
- {span.idempotencyKey && (
-
- Idempotency key
- {span.idempotencyKey}
-
- )}
-
- Version
-
- {span.workerVersion ? (
- span.workerVersion
- ) : (
-
- Never started
-
-
- )}
-
-
-
-
- ) : (
-
- {span.level === "TRACE" ? (
- <>
-
-
-
-
+ {tab === "detail" ? (
+
+
+
+ Status
+
+
-
- ) : (
-
-
}
- variant="dot-solid"
+
+
+
+ Task
+
+
+ {span.taskSlug}
+
+ }
+ content={`Filter runs by ${span.taskSlug}`}
/>
-
-
Triggered runs
-
-
-
- Run #
- Task
- Version
- Created at
-
-
-
- {span.triggeredRuns.map((run) => {
- const path = v3RunSpanPath(
- organization,
- project,
- { friendlyId: run.friendlyId },
- { spanId: run.spanId }
- );
- return (
-
-
- {run.number}
-
-
- {run.taskIdentifier}
-
-
- {run.lockedToVersion?.version ?? "–"}
-
-
-
-
-
- );
- })}
-
-
-
- {span.events.length > 0 &&
}
- {span.properties !== undefined && (
-
)}
-
+
+ Version
+
+ {span.workerVersion ? (
+ span.workerVersion
+ ) : (
+
+ Never started
+
+
+ )}
+
+
+
+
+ ) : (
+
+ )}
- {run.taskIdentifier}
+
+ {run.taskIdentifier}
+ {run.isCached ? " (cached)" : null}
+
{runParam && closePanel && (
@@ -602,6 +499,22 @@ function RunBody({
)}
+
+ Idempotency
+
+ {run.idempotencyKey ? run.idempotencyKey : "–"}
+ {run.idempotencyKey && (
+
+ Expires:{" "}
+ {run.idempotencyKeyExpiresAt ? (
+
+ ) : (
+ "–"
+ )}
+
+ )}
+
+
Version
@@ -634,6 +547,22 @@ function RunBody({
)}
+
+ Engine version
+ {run.engine}
+
+ {isAdmin && (
+ <>
+
+ Primary master queue
+ {run.masterQueue}
+
+
+ Secondary master queue
+ {run.secondaryMasterQueue}
+
+
+ )}
Test run
@@ -771,12 +700,13 @@ function RunBody({
+ {run.error && }
+
{run.payload !== undefined && (
)}
- {run.error !== undefined ? (
-
- ) : run.output !== undefined ? (
+
+ {run.error === undefined && run.output !== undefined ? (
) : null}
@@ -787,12 +717,17 @@ function RunBody({
{run.friendlyId !== runParam && (
- Focus on run
+ {run.isCached ? "Jump to original run" : "Focus on run"}
)}
@@ -925,3 +860,216 @@ function PacketDisplay({
}
}
}
+
+function SpanEntity({ span }: { span: Span }) {
+ const isAdmin = useHasAdminAccess();
+
+ const organization = useOrganization();
+ const project = useProject();
+
+ if (!span.entity) {
+ //normal span
+ return (
+
+ {span.level === "TRACE" ? (
+ <>
+
+
+
+
+
+ ) : (
+
+ }
+ variant="dot-solid"
+ />
+
+ )}
+
+
+ Message
+ {span.message}
+
+ {span.triggeredRuns.length > 0 && (
+
+
+
Triggered runs
+
+
+
+ Run #
+ Task
+ Version
+ Created at
+
+
+
+ {span.triggeredRuns.map((run) => {
+ const path = v3RunSpanPath(
+ organization,
+ project,
+ { friendlyId: run.friendlyId },
+ { spanId: run.spanId }
+ );
+ return (
+
+
+ {run.number}
+
+
+ {run.taskIdentifier}
+
+
+ {run.lockedToVersion?.version ?? "–"}
+
+
+
+
+
+ );
+ })}
+
+
+
+ {span.events.length > 0 &&
}
+ {span.properties !== undefined ? (
+
+ ) : null}
+
+
+
+
Waitpoint
+
+ A waitpoint pauses your code from continuing until the conditions are met.{" "}
+ View docs.
+
+
+
+
+ Status
+
+
+
+
+
+ ID
+
+ {span.entity.object.friendlyId}
+
+
+
+ Idempotency key
+
+
+
+ {span.entity.object.userProvidedIdempotencyKey
+ ? span.entity.object.idempotencyKey
+ : "–"}
+
+
+ {span.entity.object.idempotencyKeyExpiresAt ? (
+ <>
+ TTL:
+
+ ) : null}
+
+
+ {span.entity.object.completedAfter ? (
+
+ ) : (
+ "–"
+ )}
+ {span.entity.object.status === "PENDING" && (
+
+ )}
+
+
+
+
+ )}
+ {span.entity.object.status === "PENDING" ? null : span.entity.object.isTimeout ? (
+ <>
+ ) : span.entity.object.output ? (
+
+ ) : span.entity.object.completedAfter ? (
+
+ Completed at
+
+
+
+
+ ) : (
+ "Completed with no output"
+ )}
+
+
+ {span.entity.object.status === "PENDING" && (
+
+
+
+ )}
+