import { json } from "@remix-run/server-runtime"; import type { CreateBatchResponse } from "@trigger.dev/core/v3"; import { CreateBatchRequestBody, generateJWT } from "@trigger.dev/core/v3"; import { env } from "~/env.server"; import { BatchRateLimitExceededError } from "~/runEngine/concerns/batchLimits.server"; import { CreateBatchService } from "~/runEngine/services/createBatch.server"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { getOneTimeUseToken } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server"; import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server"; import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server"; import { clientSafeErrorMessage } from "~/utils/prismaErrors"; import { handleRequestIdempotency, saveRequestIdempotency, } from "~/utils/requestIdempotency.server"; import { sanitizeTriggerSource } from "~/utils/triggerSource"; import { ServiceValidationError } from "~/v3/services/baseService.server"; import { OutOfEntitlementError } from "~/v3/services/triggerTask.server"; import { engine } from "~/v3/runEngine.server"; import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger"; /** * Phase 1 of 2-phase batch API: Create a batch. * * POST /api/v3/batches * * Creates a batch record and optionally blocks the parent run for batchTriggerAndWait. * Items are streamed separately via POST /api/v3/batches/:batchId/items */ const { action, loader } = createActionApiRoute( { headers: HeadersSchema, body: CreateBatchRequestBody, allowJWT: true, maxContentLength: 131_072, // 128KB is plenty for the batch metadata authorization: { action: "batchTrigger", // No specific tasks to authorize at batch creation time — tasks are // validated when items are streamed. Collection-level check. resource: () => ({ type: "tasks" }), }, corsStrategy: "all", }, async ({ body, headers, authentication }) => { // Validate runCount if (body.runCount <= 0) { return json({ error: "runCount must be a positive integer" }, { status: 400 }); } // Check runCount against limit if (body.runCount > env.STREAMING_BATCH_MAX_ITEMS) { return json( { error: `Batch runCount of ${body.runCount} exceeds maximum allowed of ${env.STREAMING_BATCH_MAX_ITEMS}.`, }, { status: 400 } ); } const { "trigger-version": triggerVersion, "x-trigger-span-parent-as-link": spanParentAsLink, "x-trigger-worker": isFromWorker, "x-trigger-client": triggerClient, "x-trigger-realtime-streams-version": realtimeStreamsVersion, "x-trigger-source": triggerSourceHeader, traceparent, tracestate, } = headers; const oneTimeUseToken = await getOneTimeUseToken(authentication); logger.debug("Create batch request", { runCount: body.runCount, parentRunId: body.parentRunId, resumeParentOnCompletion: body.resumeParentOnCompletion, idempotencyKey: body.idempotencyKey, triggerVersion, isFromWorker, triggerClient, }); // Handle idempotency for the batch creation const cachedResponse = await handleRequestIdempotency< { friendlyId: string; runCount: number }, CreateBatchResponse >(body.idempotencyKey, { requestType: "create-batch", findCachedEntity: async (cachedRequestId) => { const batch = await engine.runStore.findBatchTaskRunById(cachedRequestId); if (!batch || batch.runtimeEnvironmentId !== authentication.environment.id) return null; return batch; }, buildResponse: (cachedBatch) => ({ id: cachedBatch.friendlyId, runCount: cachedBatch.runCount, isCached: true, }), buildResponseHeaders: async (responseBody) => { return await responseHeaders(responseBody, authentication.environment, triggerClient); }, }); if (cachedResponse) { return cachedResponse; } const traceContext = isFromWorker ? { traceparent, tracestate } : { external: { traceparent, tracestate } }; const service = new CreateBatchService(); service.onBatchTaskRunCreated.attachOnce(async (batch) => { await saveRequestIdempotency(body.idempotencyKey, "create-batch", batch.id); }); try { const batch = await service.call(authentication.environment, body, { triggerVersion: triggerVersion ?? undefined, traceContext, spanParentAsLink: spanParentAsLink === 1, oneTimeUseToken, realtimeStreamsVersion: determineRealtimeStreamsVersion( realtimeStreamsVersion ?? undefined ), triggerSource: isFromWorker ? "sdk" : (sanitizeTriggerSource(triggerSourceHeader) ?? "api"), }); const $responseHeaders = await responseHeaders( batch, authentication.environment, triggerClient ); return json(batch, { status: 202, headers: $responseHeaders, }); } catch (error) { if (error instanceof BatchRateLimitExceededError) { logger.info("Batch rate limit exceeded", { limit: error.limit, remaining: error.remaining, resetAt: error.resetAt.toISOString(), itemCount: error.itemCount, }); return json( { error: error.message }, { status: 429, headers: { "X-RateLimit-Limit": error.limit.toString(), "X-RateLimit-Remaining": Math.max(0, error.remaining).toString(), "X-RateLimit-Reset": Math.floor(error.resetAt.getTime() / 1000).toString(), "Retry-After": Math.max( 1, Math.ceil((error.resetAt.getTime() - Date.now()) / 1000) ).toString(), }, } ); } // Customer-facing validation/quota failures (invalid batch shape, // entitlements exhausted). The handler returns 422 with the message; // system handles it gracefully, no alert needed. if (error instanceof ServiceValidationError) { logger.warn("Create batch error", { error: error.message }); return json({ error: error.message }, { status: error.status ?? 422 }); } if (error instanceof OutOfEntitlementError) { logger.warn("Create batch error", { error: error.message }); return json({ error: error.message }, { status: 422 }); } logger.error("Create batch error", { error: { message: (error as Error).message, stack: (error as Error).stack, }, }); if (error instanceof Error) { return json( { error: clientSafeErrorMessage(error) }, { status: 500, headers: { "x-should-retry": "false" } } ); } return json({ error: "Something went wrong" }, { status: 500 }); } } ); async function responseHeaders( batch: CreateBatchResponse, 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}`, `write:batch:${batch.id}`], }; const jwt = await generateJWT({ secretKey: extractJwtSigningSecretKey(environment), payload: claims, expirationTime: "1h", }); return { "x-trigger-jwt-claims": claimsHeader, "x-trigger-jwt": jwt, }; } return { "x-trigger-jwt-claims": claimsHeader, }; } export { action, loader };