github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Feature: generic fetch requests

Browse Source
This commit is contained in:
Eric Allam
2023-01-21 13:52:42 -08:00
parent 14b5cce734
commit c69c3705c5
29 changed files with 1372 additions and 541 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/quick-pets-refuse.md
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/sdk": patch
---
Added context.fetch to make generic fetch requests using Trigger.dev
apps/webapp/app/models/fetchRequest.server.ts
+16
View File
@@ -0,0 +1,16 @@
import { prisma } from "~/db.server";
import type { FetchRequest } from ".prisma/client";
export type { FetchRequest };
export async function findFetchRequestById(id: string) {
return prisma.fetchRequest.findUnique({
where: {
id,
},
include: {
step: true,
run: true,
},
});
}
apps/webapp/app/models/workflowRunPresenter.server.ts
+67 -2
View File
@@ -1,13 +1,16 @@
import type { SecureString } from "@trigger.dev/common-schemas";
import { FetchResponseSchema } from "@trigger.dev/common-schemas";
import {
CustomEventSchema,
ErrorSchema,
FetchRequestSchema,
LogMessageSchema,
TriggerMetadataSchema,
WaitSchema,
} from "@trigger.dev/common-schemas";
import type { DisplayProperties } from "internal-integrations";
import { slack, shopify } from "internal-integrations";
import type { Provider } from "@trigger.dev/providers";
import type { DisplayProperties } from "internal-integrations";
import { shopify, slack } from "internal-integrations";
import invariant from "tiny-invariant";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
@@ -113,6 +116,31 @@ async function parseStep(
...base,
type: "DISCONNECTION" as const,
};
case "FETCH_REQUEST":
invariant(
original.fetchRequest,
`Fetch request is missing from run step ${original.id}}`
);
const fetchRequest = FetchRequestSchema.parse(original.input);
const lastFetchResponse = original.fetchRequest.responses[0];
const lastResponse = lastFetchResponse
? FetchResponseSchema.parse(lastFetchResponse.output)
: undefined;
return {
...base,
type: "FETCH_REQUEST" as const,
title: `${fetchRequest.method} ${fetchRequest.url}`,
input: {
headers: obfuscateHeaders(fetchRequest.headers),
body: fetchRequest.body,
},
output: original.output,
requestStatus: original.fetchRequest.status,
retryCount: original.fetchRequest.retryCount,
lastResponse,
};
case "INTEGRATION_REQUEST":
invariant(
original.integrationRequest,
@@ -197,9 +225,46 @@ function getWorkflowRun(prismaClient: PrismaClient, id: string) {
},
},
},
fetchRequest: {
include: {
responses: {
orderBy: { createdAt: "desc" },
take: 1,
},
},
},
},
orderBy: { ts: "asc" },
},
},
});
}
function obfuscateHeaders(
headers?: Record<string, string | SecureString>
): Record<string, string> {
if (!headers) {
return {};
}
return Object.fromEntries(
Object.entries(headers).map(([key, value]) => [
key,
typeof value === "string" ? value : obfuscateSecureString(value),
])
);
}
// SecureString is an object with { strings: string[]; interpolations: string[]; }
// So we need to build up a string from strings and replace interpolations with ****
function obfuscateSecureString(value: SecureString) {
let result = "";
for (let i = 0; i < value.strings.length; i++) {
result += value.strings[i];
if (i < value.interpolations.length) {
result += "********";
}
}
return result;
}
apps/webapp/app/routes/__app/orgs/$organizationSlug/workflows/$workflowSlug/runs/$runId.tsx
+96 -1
View File
@@ -40,7 +40,7 @@ import { TriggerTypeIcon } from "~/components/triggers/TriggerIcons";
import { triggerLabel } from "~/components/triggers/triggerLabel";
import { useCurrentOrganization } from "~/hooks/useOrganizations";
import { useCurrentWorkflow } from "~/hooks/useWorkflows";
import { WorkflowRunStatus } from "~/models/workflowRun.server";
import type { WorkflowRunStatus } from "~/models/workflowRun.server";
import { WorkflowRunPresenter } from "~/models/workflowRunPresenter.server";
import { requireUserId } from "~/services/session.server";
import { dateDifference, formatDateTime } from "~/utils";
@@ -430,6 +430,15 @@ function StepHeader({ step }: { step: Step }) {
integration={step.service.connection?.title}
/>
);
case "FETCH_REQUEST":
return (
<PanelHeader
icon={stepInfo[step.type].icon}
title={step.title}
startedAt={step.startedAt}
finishedAt={step.finishedAt}
/>
);
default:
return (
<PanelHeader
@@ -464,6 +473,8 @@ function StepBody({ step }: { step: Step }) {
return <CustomEventStep event={step} />;
case "INTEGRATION_REQUEST":
return <IntegrationRequestStep request={step} />;
case "FETCH_REQUEST":
return <FetchRequestStep request={step} />;
case "DURABLE_DELAY":
return <DelayStep step={step} />;
}
@@ -705,6 +716,86 @@ function IntegrationRequestStep({
);
}
function FetchRequestStep({
request,
}: {
request: StepType<Step, "FETCH_REQUEST">;
}) {
const organization = useCurrentOrganization();
invariant(organization, "Organization must be set");
return (
<>
<div className="mt-4">
{request.input && (
<>
<InputTitle />
<CodeBlock code={stringifyCode(request.input)} align="top" />
</>
)}
</div>
<div className="mt-4">
{request.requestStatus === "ERROR" ? (
<div>
<div className="flex gap-2 mb-2 mt-3 ">
<ExclamationTriangleIcon className="h-5 w-5 text-rose-500" />
<Body size="small" className="text-rose-500">
Failed with error:
</Body>
</div>
<CodeBlock
code={request.output ? stringifyCode(request.output) : ""}
align="top"
maxHeight="200px"
className="border border-rose-500"
/>
</div>
) : request.requestStatus === "RETRYING" ? (
<div>
<div className="flex gap-2 mb-2 mt-3 ">
<ExclamationTriangleIcon className="h-5 w-5 text-rose-500" />
<Body size="small" className="text-rose-500">
{request.lastResponse ? (
<>Got a {request.lastResponse.status} response, retrying...</>
) : (
<>Retrying...</>
)}
</Body>
</div>
{request.output ? (
<CodeBlock
code={request.output ? stringifyCode(request.output) : ""}
align="top"
maxHeight="200px"
className="border border-rose-500"
/>
) : null}
</div>
) : (
request.output && (
<>
<div className="flex justify-between">
<OutputTitle />
{request.retryCount > 0 && (
<Body size="small" className="text-slate-400">
{request.retryCount} retries
</Body>
)}
</div>
<CodeBlock
code={stringifyCode(request.output)}
align="top"
maxHeight="200px"
/>
</>
)
)}
</div>
</>
);
}
function Error({ error }: { error: Run["error"] }) {
if (!error) return null;
@@ -753,6 +844,10 @@ const stepInfo: Record<Step["type"], { label: string; icon: ReactNode }> = {
label: "Disconnected",
icon: <ExclamationCircleIcon className={styleClass} />,
},
FETCH_REQUEST: {
label: "Fetch request",
icon: <GlobeAltIcon className={styleClass} />,
},
} as const;
type LogLevel = StepType<Step, "LOG_MESSAGE">["input"]["level"];
apps/webapp/app/services/fetches/createFetchRequest.server.ts
+100
View File
@@ -0,0 +1,100 @@
import type { FetchRequest } from "@trigger.dev/common-schemas";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
import type { WorkflowRunStep } from "~/models/workflowRun.server";
import { createStepOnce } from "~/models/workflowRunStep.server";
import { taskQueue } from "../messageBroker.server";
export class CreateFetchRequest {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
async call(
key: string,
runId: string,
apiKey: string,
timestamp: string,
data: FetchRequest
) {
const environment = await this.#prismaClient.runtimeEnvironment.findUnique({
where: {
apiKey,
},
include: {
organization: true,
},
});
if (!environment) {
throw new Error("Invalid API key");
}
const workflowRun = await this.#prismaClient.workflowRun.findUnique({
where: {
id: runId,
},
include: {
workflow: true,
},
});
if (!workflowRun) {
throw new Error("Invalid workflow run ID");
}
if (workflowRun.workflow.organizationId !== environment.organizationId) {
throw new Error("Invalid workflow run ID");
}
const idempotentStep = await createStepOnce(workflowRun.id, key, {
type: "FETCH_REQUEST",
input: data,
context: {},
status: "PENDING",
ts: timestamp,
});
if (idempotentStep.status === "EXISTING") {
return this.#handleExistingStep(idempotentStep.step);
}
const workflowRunStep = idempotentStep.step;
// Create the integration request
const fetchRequest = await this.#prismaClient.fetchRequest.create({
data: {
fetch: data,
runId: workflowRun.id,
stepId: workflowRunStep.id,
status: "PENDING",
},
});
await taskQueue.publish("FETCH_REQUEST_CREATED", {
id: fetchRequest.id,
});
return fetchRequest;
}
async #handleExistingStep(step: WorkflowRunStep) {
const fetchRequest = await this.#prismaClient.fetchRequest.findUnique({
where: {
stepId: step.id,
},
});
if (!fetchRequest) {
return;
}
if (fetchRequest.status === "SUCCESS" || fetchRequest.status === "ERROR") {
await taskQueue.publish("RESOLVE_FETCH_REQUEST", {
id: fetchRequest.id,
});
}
}
}
apps/webapp/app/services/fetches/performFetchRequest.server.ts
+281
View File
@@ -0,0 +1,281 @@
import type { FetchRequest } from ".prisma/client";
import type { SecureString } from "@trigger.dev/common-schemas";
import { FetchRequestSchema } from "@trigger.dev/common-schemas";
import type {
NormalizedResponse,
PerformedRequestResponse,
} from "internal-integrations";
import type { z } from "zod";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
const RETRYABLE_STATUS_CODES = [408, 429, 500, 502, 503, 504];
type CallResponse =
| {
stop: true;
}
| {
stop: false;
retryInSeconds: number;
};
export class PerformFetchRequest {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
async call(id: string): Promise<CallResponse> {
const fetchRequest = await this.#prismaClient.fetchRequest.findUnique({
where: { id },
});
if (!fetchRequest) {
return { stop: true };
}
const performedRequest = await this.#performRequest(fetchRequest);
if (performedRequest.ok) {
return this.#completeWithSuccess(fetchRequest, performedRequest.response);
} else if (performedRequest.isRetryable) {
return this.#attemptRetry(fetchRequest, performedRequest.response);
} else {
return this.#completeWithFailure(fetchRequest, performedRequest.response);
}
}
async #completeWithSuccess(
fetchRequest: FetchRequest,
response: NormalizedResponse
) {
await this.#createResponse(fetchRequest, response);
await this.#prismaClient.fetchRequest.update({
where: {
id: fetchRequest.id,
},
data: {
status: "SUCCESS",
},
});
await this.#prismaClient.workflowRunStep.update({
where: {
id: fetchRequest.stepId,
},
data: {
status: "SUCCESS",
output: response.output,
context: response.context,
finishedAt: new Date(),
},
});
return { stop: true as const };
}
async #completeWithFailure(
fetchRequest: FetchRequest,
response: NormalizedResponse
) {
await this.#createResponse(fetchRequest, response);
await this.#prismaClient.fetchRequest.update({
where: {
id: fetchRequest.id,
},
data: {
status: "ERROR",
},
});
await this.#prismaClient.workflowRunStep.update({
where: {
id: fetchRequest.stepId,
},
data: {
status: "ERROR",
output: response.output,
context: response.context,
finishedAt: new Date(),
},
});
return { stop: true as const };
}
async #attemptRetry(
fetchRequest: FetchRequest,
response: NormalizedResponse
) {
if (fetchRequest.retryCount >= 10) {
await this.#prismaClient.fetchRequest.update({
where: {
id: fetchRequest.id,
},
data: {
retryCount: {
increment: 1,
},
},
});
return this.#completeWithFailure(fetchRequest, response);
}
await this.#createResponse(fetchRequest, response);
const updatedFetchRequest = await this.#prismaClient.fetchRequest.update({
where: {
id: fetchRequest.id,
},
data: {
status: "RETRYING",
retryCount: {
increment: 1,
},
},
});
return {
stop: false as const,
retryInSeconds: this.#calculateRetryInSeconds(
updatedFetchRequest.retryCount
),
};
}
// Exponential backoff with a configurable factor and a configurable maximum
#calculateRetryInSeconds(
retryCount: number,
options: { factor: number; maxTimeout: number; minTimeout: number } = {
factor: 1.8,
minTimeout: 1000,
maxTimeout: 60000,
}
) {
const timeout = options.factor ** retryCount * options.minTimeout;
return Math.min(timeout, options.maxTimeout) / 1000;
}
async #createResponse(
fetchRequest: FetchRequest,
response: NormalizedResponse
) {
const integrationResponse = await this.#prismaClient.fetchResponse.create({
data: {
request: {
connect: {
id: fetchRequest.id,
},
},
context: response.context,
output: response.output ? response.output : undefined,
},
});
return integrationResponse;
}
async #performRequest(
fetchRequest: FetchRequest
): Promise<PerformedRequestResponse> {
const request = FetchRequestSchema.parse(fetchRequest.fetch);
const requestInit = createFetchRequestInit(request);
const response = await fetch(request.url, requestInit);
const body = await this.#safeGetJson(response);
if (response.ok) {
return {
ok: true,
isRetryable: false,
response: {
output: {
status: response.status,
headers: headersToRecord(response.headers),
body,
},
context: {},
},
};
}
// Only retry on retryable status codes
return {
ok: false,
isRetryable: RETRYABLE_STATUS_CODES.includes(response.status),
response: {
output: {
status: response.status,
headers: headersToRecord(response.headers),
body,
},
context: {},
},
};
}
#safeGetJson = async (response: Response) => {
try {
return await response.json();
} catch (error) {
return undefined;
}
};
}
type FetchRequestOptions = z.infer<typeof FetchRequestSchema>;
function createFetchRequestInit(request: FetchRequestOptions): RequestInit {
const headers = normalizeHeaders(request.headers);
return {
method: request.method,
headers,
body: request.body ? JSON.stringify(request.body) : undefined,
};
}
function normalizeHeaders(
headers: FetchRequestOptions["headers"]
): Record<string, string> {
if (!headers) {
return {};
}
return Object.fromEntries(
Object.entries(headers).map(([key, value]) => [
key,
typeof value === "string" ? value : normalizeSecureString(value),
])
);
}
function normalizeSecureString(value: SecureString): string {
let result = "";
for (let i = 0; i < value.strings.length; i++) {
result += value.strings[i];
if (i < value.interpolations.length) {
result += value.interpolations[i];
}
}
return result;
}
function headersToRecord(headers: Headers): Record<string, string> {
const result: Record<string, string> = {};
headers.forEach((value, key) => {
result[key] = value;
});
return result;
}
apps/webapp/app/services/fetches/startFetchRequest.server.ts
+38
View File
@@ -0,0 +1,38 @@
import type { WorkflowRunStep } from ".prisma/client";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
import type { FetchRequest } from "~/models/fetchRequest.server";
import { requestTaskQueue } from "../messageBroker.server";
export class StartFetchRequest {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
async call(request: FetchRequest, step: WorkflowRunStep) {
await this.#prismaClient.fetchRequest.update({
where: {
id: request.id,
},
data: {
status: "FETCHING",
},
});
await this.#prismaClient.workflowRunStep.update({
where: {
id: step.id,
},
data: {
status: "RUNNING",
startedAt: new Date(),
},
});
requestTaskQueue.publish("PERFORM_FETCH_REQUEST", {
id: request.id,
});
}
}
apps/webapp/app/services/messageBroker.server.ts
+129
View File
@@ -1,4 +1,5 @@
import {
FetchOutputSchema,
JsonSchema,
ScheduledEventPayloadSchema,
} from "@trigger.dev/common-schemas";
@@ -44,6 +45,10 @@ import { WorkflowRunDisconnected } from "./runs/runDisconnected.server";
import { DeliverScheduledEvent } from "./scheduler/deliverScheduledEvent.server";
import { RegisterSchedulerSource } from "./scheduler/registerSchedulerSource.server";
import { WorkflowRunTriggerTimeout } from "./runs/runTriggerTimeout.server";
import { CreateFetchRequest } from "./fetches/createFetchRequest.server";
import { findFetchRequestById } from "~/models/fetchRequest.server";
import { StartFetchRequest } from "./fetches/startFetchRequest.server";
import { PerformFetchRequest } from "./fetches/performFetchRequest.server";
let pulsarClient: PulsarClient;
let triggerPublisher: ZodPublisher<TriggerCatalog>;
@@ -227,6 +232,19 @@ function createCommandSubscriber() {
return true;
},
SEND_FETCH_REQUEST: async (id, data, properties) => {
const service = new CreateFetchRequest();
await service.call(
data.key,
properties["x-workflow-run-id"],
properties["x-api-key"],
properties["x-timestamp"],
data.fetch
);
return true;
},
TRIGGER_CUSTOM_EVENT: async (id, data, properties) => {
await triggerEventInRun(
data.key,
@@ -276,6 +294,10 @@ const RequestCatalog = {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
PERFORM_FETCH_REQUEST: {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
};
function createRequestTaskQueue() {
@@ -312,6 +334,30 @@ function createRequestTaskQueue() {
{ deliverAfter: response.retryInSeconds * 1000 }
);
return true;
}
},
PERFORM_FETCH_REQUEST: async (id, data, properties) => {
const service = new PerformFetchRequest();
const response = await service.call(data.id);
if (response.stop) {
await taskQueue.publish("RESOLVE_FETCH_REQUEST", {
id: data.id,
});
return true;
} else {
await pubSub.publish(
"PERFORM_FETCH_REQUEST",
{
id: data.id,
},
{},
{ deliverAfter: response.retryInSeconds * 1000 }
);
return true;
}
},
@@ -350,6 +396,14 @@ const taskQueueCatalog = {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
FETCH_REQUEST_CREATED: {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
RESOLVE_FETCH_REQUEST: {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
RESOLVE_DELAY: {
data: z.object({ id: z.string() }),
properties: z.object({}),
@@ -481,6 +535,81 @@ function createTaskQueue() {
return true;
},
FETCH_REQUEST_CREATED: async (id, data, properties) => {
const fetchRequest = await findFetchRequestById(data.id);
if (!fetchRequest) {
return true;
}
const service = new StartFetchRequest();
await service.call(fetchRequest, fetchRequest.step);
return true;
},
RESOLVE_FETCH_REQUEST: async (id, data, properties) => {
const fetchRequest = await findFetchRequestById(data.id);
if (!fetchRequest) {
return true;
}
const run = await findWorklowRunById(fetchRequest.runId);
if (!run) {
return true;
}
if (
fetchRequest.status !== "SUCCESS" &&
fetchRequest.status !== "ERROR"
) {
return true;
}
if (fetchRequest.status === "SUCCESS") {
const output = fetchRequest.step.output as z.infer<
typeof FetchOutputSchema
>;
await commandResponsePublisher.publish(
"RESOLVE_FETCH_REQUEST",
{
id: fetchRequest.id,
key: fetchRequest.step.idempotencyKey,
output: {
...output,
ok: true,
},
},
{
"x-workflow-run-id": run.id,
"x-api-key": run.environment.apiKey,
"x-org-id": run.environment.organizationId,
"x-workflow-id": run.workflowId,
"x-env": run.environment.slug,
}
);
} else {
await commandResponsePublisher.publish(
"REJECT_FETCH_REQUEST",
{
id: fetchRequest.id,
key: fetchRequest.step.idempotencyKey,
error: fetchRequest.step.output as z.infer<typeof JsonSchema>,
},
{
"x-workflow-run-id": run.id,
"x-api-key": run.environment.apiKey,
"x-org-id": run.environment.organizationId,
"x-workflow-id": run.workflowId,
"x-env": run.environment.slug,
}
);
}
return true;
},
EXTERNAL_SOURCE_UPSERTED: async (id, data, properties) => {
const service = new RegisterExternalSource();
apps/webapp/app/services/requests/performIntegrationRequest.server.ts
+3 -3
View File
@@ -188,14 +188,14 @@ export class PerformIntegrationRequest {
#calculateRetryInSeconds(
retryCount: number,
options: { factor: number; maxTimeout: number; minTimeout: number } = {
factor: 2,
factor: 1.8,
minTimeout: 1000,
maxTimeout: Infinity,
maxTimeout: 60000,
}
) {
const timeout = options.factor ** retryCount * options.minTimeout;
return Math.min(timeout, options.maxTimeout);
return Math.min(timeout, options.maxTimeout) / 1000;
}
async #createResponse(
apps/webapp/prisma/migrations/20230120225200_add_fetch_request_step_type/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "WorkflowRunStepType" ADD VALUE 'FETCH_REQUEST';
apps/webapp/prisma/migrations/20230120234147_add_fetch_request_models/migration.sql
+41
View File
@@ -0,0 +1,41 @@
-- CreateEnum
CREATE TYPE "FetchRequestStatus" AS ENUM ('PENDING', 'FETCHING', 'RETRYING', 'SUCCESS', 'ERROR');
-- CreateTable
CREATE TABLE "FetchRequest" (
"id" TEXT NOT NULL,
"fetch" JSONB NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) NOT NULL,
"status" "FetchRequestStatus" NOT NULL DEFAULT 'PENDING',
"runId" TEXT NOT NULL,
"stepId" TEXT NOT NULL,
"retryCount" INTEGER NOT NULL DEFAULT 0,
"error" JSONB,
"response" JSONB,
CONSTRAINT "FetchRequest_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "FetchResponse" (
"id" TEXT NOT NULL,
"requestId" TEXT NOT NULL,
"output" JSONB NOT NULL,
"context" JSONB NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "FetchResponse_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "FetchRequest_stepId_key" ON "FetchRequest"("stepId");
-- AddForeignKey
ALTER TABLE "FetchRequest" ADD CONSTRAINT "FetchRequest_runId_fkey" FOREIGN KEY ("runId") REFERENCES "WorkflowRun"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "FetchRequest" ADD CONSTRAINT "FetchRequest_stepId_fkey" FOREIGN KEY ("stepId") REFERENCES "WorkflowRunStep"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "FetchResponse" ADD CONSTRAINT "FetchResponse_requestId_fkey" FOREIGN KEY ("requestId") REFERENCES "FetchRequest"("id") ON DELETE CASCADE ON UPDATE CASCADE;
apps/webapp/prisma/schema.prisma
+48 -3
View File
@@ -411,9 +411,10 @@ model WorkflowRun {
timedOutAt DateTime?
timedOutReason String?
isTest Boolean @default(false)
requests IntegrationRequest[]
delays DurableDelay[]
isTest Boolean @default(false)
requests IntegrationRequest[]
delays DurableDelay[]
fetchRequests FetchRequest[]
}
enum WorkflowRunStatus {
@@ -449,6 +450,7 @@ model WorkflowRunStep {
integrationRequest IntegrationRequest?
delay DurableDelay?
fetchRequest FetchRequest?
@@unique([runId, idempotencyKey])
}
@@ -467,4 +469,47 @@ enum WorkflowRunStepType {
CUSTOM_EVENT
INTEGRATION_REQUEST
DISCONNECTION
FETCH_REQUEST
}
model FetchRequest {
id String @id @default(cuid())
fetch Json
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
status FetchRequestStatus @default(PENDING)
run WorkflowRun @relation(fields: [runId], references: [id], onDelete: Cascade, onUpdate: Cascade)
runId String
step WorkflowRunStep @relation(fields: [stepId], references: [id], onDelete: Cascade, onUpdate: Cascade)
stepId String @unique
retryCount Int @default(0)
error Json?
response Json?
responses FetchResponse[]
}
enum FetchRequestStatus {
PENDING
FETCHING
RETRYING
SUCCESS
ERROR
}
model FetchResponse {
id String @id @default(cuid())
request FetchRequest @relation(fields: [requestId], references: [id], onDelete: Cascade, onUpdate: Cascade)
requestId String
output Json
context Json
createdAt DateTime @default(now())
}
apps/wss/src/runController.ts
+54
View File
@@ -136,6 +136,60 @@ export class WorkflowRunController {
},
});
return success;
},
RESOLVE_FETCH_REQUEST: async (id, data, properties) => {
if (properties["x-workflow-run-id"] !== this.#runId) {
return true;
}
this.#logger.debug(
"Received resolve fetch request",
id,
data,
properties
);
const success = await this.#hostRPC.send("RESOLVE_FETCH_REQUEST", {
id: data.id,
output: data.output,
key: data.key,
meta: {
workflowId: properties["x-workflow-id"],
organizationId: properties["x-org-id"],
environment: properties["x-env"],
apiKey: properties["x-api-key"],
runId: properties["x-workflow-run-id"],
},
});
return success;
},
REJECT_FETCH_REQUEST: async (id, data, properties) => {
if (properties["x-workflow-run-id"] !== this.#runId) {
return true;
}
this.#logger.debug(
"Received reject fetch request",
id,
data,
properties
);
const success = await this.#hostRPC.send("REJECT_FETCH_REQUEST", {
id: data.id,
key: data.key,
error: data.error,
meta: {
workflowId: properties["x-workflow-id"],
organizationId: properties["x-org-id"],
environment: properties["x-env"],
apiKey: properties["x-api-key"],
runId: properties["x-workflow-run-id"],
},
});
return success;
},
},
apps/wss/src/server.ts
+15
View File
@@ -153,6 +153,21 @@ export class TriggerServer {
return !!response;
},
SEND_FETCH: async (request) => {
const runController = this.#runControllers.get(request.runId);
if (!runController) {
// TODO: need to recover from this issue by trying to reconnect
return false;
}
const response = await runController.publish("SEND_FETCH_REQUEST", {
key: request.key,
fetch: request.fetch,
});
return !!response;
},
SEND_EVENT: async (request) => {
const runController = this.#runControllers.get(request.runId);
examples/fetch-playground/package.json
+19
View File
@@ -0,0 +1,19 @@
{
"private": true,
"name": "@examples/fetch-playground",
"version": "0.0.1",
"description": "A fetch playground for testing Trigger.dev functionality",
"dependencies": {
"@trigger.dev/integrations": "workspace:*",
"@trigger.dev/sdk": "workspace:*",
"zod": "^3.20.2"
},
"devDependencies": {
"@trigger.dev/tsconfig": "workspace:*",
"@types/node": "16",
"tsx": "^3.12.0"
},
"scripts": {
"dev": "tsx src/index.ts"
}
}
examples/fetch-playground/src/index.ts
+43
View File
@@ -0,0 +1,43 @@
import { Trigger, customEvent } from "@trigger.dev/sdk";
import { z } from "zod";
const TOKEN = "abc123";
new Trigger({
id: "fetch-playground",
name: "Fetch Playground",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
logLevel: "debug",
on: customEvent({
name: "playground.fetch",
schema: z.object({
url: z.string().default("http://localhost:8888"),
path: z.string().default("/"),
method: z
.enum(["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"])
.default("GET"),
headers: z.record(z.string()).optional(),
body: z.any().optional(),
}),
}),
run: async (event, ctx) => {
await ctx.logger.info("Received the playground.fetch event", {
event,
wallTime: new Date(),
});
const response = await ctx.fetch("do-fetch", `${event.url}${event.path}`, {
method: event.method,
responseSchema: z.any(),
headers: event.headers,
body: event.body ? JSON.stringify(event.body) : undefined,
});
await ctx.logger.info("Received the fetch response", {
response,
wallTime: new Date(),
});
},
}).listen();
examples/fetch-playground/tsconfig.json
+12
View File
@@ -0,0 +1,12 @@
{
"extends": "@trigger.dev/tsconfig/node16.json",
"include": ["src/**/*.ts"],
"compilerOptions": {
"experimentalDecorators": true,
"emitDecoratorMetadata": true,
"lib": ["esnext", "dom"],
"outDir": "lib",
"moduleResolution": "node"
},
"exclude": ["node_modules", "**/*.test.*"]
}
examples/playground/src/index.ts
+27 -441
View File
@@ -1,457 +1,43 @@
import { Trigger, customEvent, scheduleEvent } from "@trigger.dev/sdk";
import { github, slack } from "@trigger.dev/integrations";
import { Trigger, customEvent } from "@trigger.dev/sdk";
import { z } from "zod";
// // Workflow that sends a message to Slack
// // new Trigger({
// // id: "playground-1",
// // name: "Post to Slack immediately",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: customEvent({
// // name: "playground",
// // schema: z.object({
// // id: z.string(),
// // }),
// // }),
// // run: async (event, ctx) => {
// // await ctx.logger.info(
// // "This workflow will post a message to Slack immediately and display an Error, Info, Debug, and Warning message."
// // );
// new Trigger({
// id: "playground-1",
// name: "Post to Slack immediately",
// apiKey: "trigger_dev_zC25mKNn6c0q",
// endpoint: "ws://localhost:8889/ws",
// on: customEvent({
// name: "playground",
// schema: z.object({
// id: z.string(),
// }),
// }),
// run: async (event, ctx) => {
// await ctx.logger.info(
// "This workflow will post a message to Slack immediately and display an Error, Info, Debug, and Warning message."
// );
// await ctx.logger.error("Error message!", { event });
// await ctx.logger.info("Info message", { event });
// await ctx.logger.debug("Debug message");
// await ctx.logger.warn("Warning message!");
// const response = await slack.postMessage("send-to-slack", {
// channel: "test-integrations",
// text: `This is a message from the "Posts to Slack" workflow ${event.id}`,
// });
// return response.message;
// },
// }).listen();
// // Webhook workflow that sends a message to Slack when a Github issue is created after 2 delays
// new Trigger({
// id: "playground-2",
// name: "Posts to Slack after a GitHub Issue created",
// apiKey: "trigger_dev_zC25mKNn6c0q",
// endpoint: "ws://localhost:8889/ws",
// on: github.events.repoIssueEvent({ repo: "triggerdotdev/trigger.dev" }),
// run: async (event, ctx) => {
// await ctx.logger.info(
// "This workflow will post to Slack when a GitHub Issue created or modified after 2 delays."
// );
// await ctx.waitFor("initial-wait", { seconds: 30 });
// await ctx.waitUntil("wait-until", new Date(Date.now() + 1000 * 30));
// await ctx.logger.info("Both types of delay happened");
// const response = await slack.postMessage("send-to-slack", {
// channel: "test-integrations",
// text: `This is a message posts after an Issue was created on GitHub ${event.action}`,
// });
// await ctx.logger.debug("Debug message");
// await ctx.logger.warn("Warning message!");
// return response.message;
// },
// }).listen();
// // Workflow that shows all the log types with a 5 second delay between each
// new Trigger({
// id: "playground-3",
// name: "All log types with a 5 second delay between each",
// apiKey: "trigger_dev_zC25mKNn6c0q",
// endpoint: "ws://localhost:8889/ws",
// on: customEvent({
// name: "playground",
// schema: z.object({
// id: z.string(),
// }),
// }),
// run: async (event, ctx) => {
// await ctx.logger.info(
// "This workflow prints all the log types with a 5 second delay between each."
// );
// await ctx.logger.info(
// "Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it."
// );
// await ctx.waitFor("first-wait", { seconds: 5 });
// await ctx.logger.error(
// "This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message!",
// { event }
// );
// await ctx.waitFor("second-wait", { seconds: 5 });
// await ctx.logger.info(
// "This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message."
// );
// await ctx.waitFor("third-wait", { seconds: 5 });
// await ctx.logger.debug(
// "This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. "
// );
// // await ctx.logger.error("Error message!", { event });
// // await ctx.logger.info("Info message", { event });
// // await ctx.logger.debug("Debug message");
// // await ctx.logger.warn("Warning message!");
// // const response = await slack.postMessage("send-to-slack", {
// // channel: "test-integrations",
// // text: `This is a message from the "Posts to Slack" workflow ${event.id}`,
// // });
// // return response.message;
// // },
// // }).listen();
// // // Webhook workflow that sends a message to Slack when a Github issue is created after 2 delays
// // new Trigger({
// // id: "playground-2",
// // name: "Posts to Slack after a GitHub Issue created",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: github.events.repoIssueEvent({ repo: "triggerdotdev/trigger.dev" }),
// // run: async (event, ctx) => {
// // await ctx.logger.info(
// // "This workflow will post to Slack when a GitHub Issue created or modified after 2 delays."
// // );
// // await ctx.waitFor("initial-wait", { seconds: 30 });
// // await ctx.waitUntil("wait-until", new Date(Date.now() + 1000 * 30));
// // await ctx.logger.info("Both types of delay happened");
// // const response = await slack.postMessage("send-to-slack", {
// // channel: "test-integrations",
// // text: `This is a message posts after an Issue was created on GitHub ${event.action}`,
// // });
// // await ctx.logger.debug("Debug message");
// // await ctx.logger.warn("Warning message!");
// // return response.message;
// // },
// // }).listen();
// // // Workflow that shows all the log types with a 5 second delay between each
// // new Trigger({
// // id: "playground-3",
// // name: "All log types with a 5 second delay between each",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: customEvent({
// // name: "playground",
// // schema: z.object({
// // id: z.string(),
// // }),
// // }),
// // run: async (event, ctx) => {
// // await ctx.logger.info(
// // "This workflow prints all the log types with a 5 second delay between each."
// // );
// // await ctx.logger.info(
// // "Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it."
// // );
// // await ctx.waitFor("first-wait", { seconds: 5 });
// // await ctx.logger.error(
// // "This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message!",
// // { event }
// // );
// // await ctx.waitFor("second-wait", { seconds: 5 });
// // await ctx.logger.info(
// // "This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message."
// // );
// // await ctx.waitFor("third-wait", { seconds: 5 });
// // await ctx.logger.debug(
// // "This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. "
// // );
// // await ctx.waitFor("fourth-wait", { seconds: 5 });
// // await ctx.logger.warn(
// // "This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! This is a really long Warning message! "
// // );
// // await ctx.waitFor("fifth-wait", { seconds: 5 });
// // const response = await slack.postMessage("send-to-slack", {
// // channel: "test-integrations",
// // text: `This test displays all the log types in the webapp ${event.id}`,
// // });
// // return response.message;
// // },
// // }).listen();
// // // Workflow that sends 2 messages to Slack with a 10 second delay between both
// // new Trigger({
// // id: "playground-4",
// // name: "Post to Slack twice after 10 second delays",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: customEvent({
// // name: "playground",
// // schema: z.object({
// // id: z.string(),
// // }),
// // }),
// // run: async (event, ctx) => {
// // await ctx.logger.info(
// // "This workflow sends 2 messages to Slack with a 10 second delay between both."
// // );
// // await ctx.waitFor("initial-wait", { seconds: 10 });
// // await slack.postMessage("send-to-slack-1", {
// // channel: "test-integrations",
// // text: `This is test message 1/2 from "Post to Slack twice after 10 second delays" workflow ${event.id}`,
// // });
// // await ctx.waitFor("second-wait", { seconds: 10 });
// // const response = await slack.postMessage("send-to-slack-2", {
// // channel: "test-integrations",
// // text: `This is test message 2/2 from "Post to Slack twice after 10 second delays" workflow ${event.id}`,
// // });
// // return response.message;
// // },
// // }).listen();
// // // Workflow that send a message to Slack after 1 hour
// // new Trigger({
// // id: "playground-5",
// // name: "Post to Slack after 1 hour",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: customEvent({
// // name: "playground",
// // schema: z.object({
// // id: z.string(),
// // }),
// // }),
// // run: async (event, ctx) => {
// // await ctx.logger.info(
// // "This workflow posts a message to Slack after a 1 hour delay"
// // );
// // await ctx.waitUntil("wait-until", new Date(Date.now() + 1000 * 60 * 60));
// // await ctx.logger.info("The workflow resumed after the 1 hour delay", {
// // event,
// // });
// // const response = await slack.postMessage("send-to-slack", {
// // channel: "test-integrations",
// // text: `This message was sent after a 1 hour delay ${event.id}`,
// // });
// // return response.message;
// // },
// // }).listen();
// // // Workflow that send a message to Slack after 24 hours
// // new Trigger({
// // id: "playground-6",
// // name: "Post to Slack after a 24 hour delay",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: customEvent({
// // name: "playground",
// // schema: z.object({
// // id: z.string(),
// // }),
// // }),
// // run: async (event, ctx) => {
// // await ctx.logger.info(
// // "This workflow posts a message to Slack after a 24 hour delay"
// // );
// // await ctx.waitUntil(
// // "wait-until",
// // new Date(Date.now() + 1000 * 60 * 60 * 24)
// // );
// // await ctx.logger.info("The workflow resumed after the 24 hour delay", {
// // event,
// // });
// // const response = await slack.postMessage("send-to-slack", {
// // channel: "test-integrations",
// // text: `This message was sent after a 24 hour delay ${event.id}`,
// // });
// // return response.message;
// // },
// // }).listen();
// // // Workflow that send a message to Slack after 24 hours
// // new Trigger({
// // id: "playground-7",
// // name: "Post to Slack many times in a loop",
// // apiKey: "trigger_dev_zC25mKNn6c0q",
// // endpoint: "ws://localhost:8889/ws",
// // on: customEvent({
// // name: "playground",
// // schema: z.object({
// // id: z.string(),
// // }),
// // }),
// // run: async (event, ctx) => {
// // for (let index = 0; index < 4; index++) {
// // const response = await slack.postMessage(`send-to-slack-${index}`, {
// // channel: "test-integrations",
// // text: `This is a post to Slack many times in a loop ${index} ${event.id}`,
// // });
// // }
// // return {};
// // },
// // }).listen();
// const postMessage = new Trigger({
// id: "new-user",
// name: "New user slack message",
// apiKey: "trigger_development_lwlXEjyhSNF4",
// endpoint: "ws://localhost:8889/ws",
// logLevel: "info",
// on: customEvent({
// name: "user.created",
// schema: z.object({
// name: z.string(),
// email: z.string(),
// paidPlan: z.boolean(),
// }),
// }),
// run: async (event, ctx) => {
// await ctx.logger.info("This log will appear on the Trigger.dev run page");
// //send a message to the #new-users Slack channel with user details
// const response = await slack.postMessage("send-to-slack", {
// channel: "new-users",
// text: `New user: ${event.name} (${event.email}) signed up. ${
// event.paidPlan ? "They are paying" : "They are on the free plan"
// }.`,
// });
// return response.message;
// },
// });
// //this workflow will now connect and start listening for events
// postMessage.listen();
// new Trigger({
// id: "playground-7",
// name: "Post to Slack many times in a loop",
// apiKey: "trigger_dev_zC25mKNn6c0q",
// endpoint: "ws://localhost:8889/ws",
// on: customEvent({
// name: "playground",
// schema: z.object({
// id: z.string(),
// }),
// }),
// run: async (event, ctx) => {
// for (let index = 0; index < 4; index++) {
// const response = await slack.postMessage(`send-to-slack-${index}`, {
// channel: "test-integrations",
// text: `This is a post to Slack many times in a loop ${index} ${event.id}`,
// });
// }
// return {};
// },
// }).listen();
const TOKEN = "abc123";
new Trigger({
id: "scheduled-workflow",
name: "Scheduled Workflow",
id: "fetch-playground",
name: "Fetch Playground",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
logLevel: "debug",
triggerTTL: 5,
on: scheduleEvent({ rateOf: { minutes: 4 } }),
on: customEvent({
name: "playground.fetch",
schema: z.object({
url: z.string().default("http://localhost:8888"),
path: z.string().default("/"),
method: z
.enum(["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"])
.default("GET"),
headers: z.record(z.string()).optional(),
body: z.any().optional(),
}),
}),
run: async (event, ctx) => {
await ctx.logger.info("Received the scheduled event", {
await ctx.logger.info("Received the playground.fetch event", {
event,
wallTime: new Date(),
});
return { foo: "bar" };
},
}).listen();
new Trigger({
id: "cron-scheduled-workflow",
name: "Cron Scheduled Workflow",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
logLevel: "debug",
on: scheduleEvent({ cron: "0 * * * *" }),
run: async (event, ctx) => {
await ctx.logger.info("Received the cron scheduled event", {
event,
wallTime: new Date(),
const response = await ctx.fetch("do-fetch", `${event.url}${event.path}`, {
method: event.method,
responseSchema: z.any(),
headers: event.headers,
body: event.body ? JSON.stringify(event.body) : undefined,
});
return { foo: "bar" };
await ctx.logger.info("Received the fetch response", {
response,
wallTime: new Date(),
});
},
}).listen();
packages/common-schemas/src/fetch.ts
+41
View File
@@ -0,0 +1,41 @@
import { z } from "zod";
export const SecureStringSchema = z.object({
__secureString: z.literal(true),
strings: z.array(z.string()),
interpolations: z.array(z.string()),
});
export type SecureString = z.infer<typeof SecureStringSchema>;
export const FetchRequestSchema = z.object({
url: z.string(),
headers: z.record(z.union([z.string(), SecureStringSchema])).optional(),
method: z.enum([
"GET",
"POST",
"PUT",
"DELETE",
"PATCH",
"HEAD",
"OPTIONS",
"TRACE",
]),
body: z.any(),
});
export const FetchOutputSchema = z.object({
status: z.number(),
ok: z.boolean(),
headers: z.record(z.string()),
body: z.any().optional(),
});
export type FetchRequest = z.infer<typeof FetchRequestSchema>;
export type FetchOutput = z.infer<typeof FetchOutputSchema>;
export const FetchResponseSchema = z.object({
status: z.number(),
headers: z.record(z.string()),
body: z.any().optional(),
});
packages/common-schemas/src/index.ts
+1
View File
@@ -4,3 +4,4 @@ export * from "./logs";
export * from "./waits";
export * from "./events";
export * from "./triggers";
export * from "./fetch";
packages/internal-bridge/src/schemas/host.ts
+31 -1
View File
@@ -1,5 +1,5 @@
import { z } from "zod";
import { JsonSchema } from "@trigger.dev/common-schemas";
import { FetchOutputSchema, JsonSchema } from "@trigger.dev/common-schemas";
export const HostRPCSchema = {
TRIGGER_WORKFLOW: {
@@ -62,6 +62,36 @@ export const HostRPCSchema = {
}),
response: z.boolean(),
},
RESOLVE_FETCH_REQUEST: {
request: z.object({
id: z.string(),
key: z.string(),
output: FetchOutputSchema,
meta: z.object({
environment: z.string(),
workflowId: z.string(),
organizationId: z.string(),
apiKey: z.string(),
runId: z.string(),
}),
}),
response: z.boolean(),
},
REJECT_FETCH_REQUEST: {
request: z.object({
id: z.string(),
key: z.string(),
error: JsonSchema.default({}),
meta: z.object({
environment: z.string(),
workflowId: z.string(),
organizationId: z.string(),
apiKey: z.string(),
runId: z.string(),
}),
}),
response: z.boolean(),
},
};
export type HostRPC = typeof HostRPCSchema;
packages/internal-bridge/src/schemas/server.ts
+10
View File
@@ -1,5 +1,6 @@
import {
CustomEventSchema,
FetchRequestSchema,
TriggerMetadataSchema,
WaitSchema,
} from "@trigger.dev/common-schemas";
@@ -28,6 +29,15 @@ export const ServerRPCSchema = {
}),
response: z.boolean(),
},
SEND_FETCH: {
request: z.object({
runId: z.string(),
key: z.string(),
fetch: FetchRequestSchema,
timestamp: z.string(),
}),
response: z.boolean(),
},
SEND_LOG: {
request: z.object({
runId: z.string(),
packages/internal-platform/src/messages/catalogs/commandResponses.ts
+2
View File
@@ -1,9 +1,11 @@
import { commandResponses as integrationRequests } from "../schemas/integrationRequests";
import { commandResponses as delays } from "../schemas/delays";
import { commandResponses as fetchRequests } from "../schemas/fetchRequests";
const Catalog = {
...integrationRequests,
...delays,
...fetchRequests,
};
export default Catalog;
packages/internal-platform/src/messages/catalogs/commands.ts
+2
View File
@@ -3,6 +3,7 @@ import { commands as workflowRuns } from "../schemas/workflowRuns";
import { commands as logs } from "../schemas/logs";
import { commands as customEvents } from "../schemas/customEvents";
import { commands as delays } from "../schemas/delays";
import { commands as fetchRequests } from "../schemas/fetchRequests";
const Catalog = {
...integrationRequests,
@@ -10,6 +11,7 @@ const Catalog = {
...logs,
...customEvents,
...delays,
...fetchRequests,
};
export default Catalog;
packages/internal-platform/src/messages/schemas/fetchRequests.ts
+39
View File
@@ -0,0 +1,39 @@
import {
FetchOutputSchema,
FetchRequestSchema,
JsonSchema,
} from "@trigger.dev/common-schemas";
import { z } from "zod";
import {
WorkflowRunEventPropertiesSchema,
WorkflowSendRunEventPropertiesSchema,
} from "../sharedSchemas";
export const commandResponses = {
RESOLVE_FETCH_REQUEST: {
data: z.object({
id: z.string(),
key: z.string(),
output: FetchOutputSchema,
}),
properties: WorkflowRunEventPropertiesSchema,
},
REJECT_FETCH_REQUEST: {
data: z.object({
id: z.string(),
key: z.string(),
error: JsonSchema.default({}),
}),
properties: WorkflowRunEventPropertiesSchema,
},
};
export const commands = {
SEND_FETCH_REQUEST: {
data: z.object({
key: z.string(),
fetch: FetchRequestSchema,
}),
properties: WorkflowSendRunEventPropertiesSchema,
},
};
packages/trigger-sdk/src/client.ts
+88
View File
@@ -1,3 +1,4 @@
import { FetchOutput } from "@trigger.dev/common-schemas";
import {
HostRPCSchema,
Logger,
@@ -45,6 +46,14 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
}
>();
#fetchCallbacks = new Map<
string,
{
resolve: (output: FetchOutput) => void;
reject: (err?: any) => void;
}
>();
constructor(trigger: Trigger<TSchema>, options: TriggerOptions<TSchema>) {
this.#trigger = trigger;
this.#options = options;
@@ -232,6 +241,54 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
return true;
},
RESOLVE_FETCH_REQUEST: async (data) => {
this.#logger.debug("Handling RESOLVE_FETCH_REQUEST", data);
const fetchCallbacks = this.#fetchCallbacks.get(
messageKey(data.meta.runId, data.key)
);
if (!fetchCallbacks) {
this.#logger.debug(
`Could not find fetch callbacks for request ID ${messageKey(
data.meta.runId,
data.key
)}. This can happen when a workflow run is resumed`
);
return true;
}
const { resolve } = fetchCallbacks;
resolve(data.output);
return true;
},
REJECT_FETCH_REQUEST: async (data) => {
this.#logger.debug("Handling REJECT_FETCH_REQUEST", data);
const fetchCallbacks = this.#fetchCallbacks.get(
messageKey(data.meta.runId, data.key)
);
if (!fetchCallbacks) {
this.#logger.debug(
`Could not find fetch callbacks for request ID ${messageKey(
data.meta.runId,
data.key
)}. This can happen when a workflow run is resumed`
);
return true;
}
const { reject } = fetchCallbacks;
reject(data.error);
return true;
},
TRIGGER_WORKFLOW: async (data) => {
this.#logger.debug("Handling TRIGGER_WORKFLOW", data);
@@ -307,6 +364,37 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
return;
},
fetch: async (key, url, options) => {
const result = new Promise<FetchOutput>((resolve, reject) => {
this.#fetchCallbacks.set(messageKey(data.id, key), {
resolve,
reject,
});
});
await serverRPC.send("SEND_FETCH", {
runId: data.id,
key,
fetch: {
url: url.toString(),
method: options.method,
headers: options.headers,
body: options.body,
},
timestamp: String(highPrecisionTimestamp()),
});
const response = await result;
return {
status: response.status,
ok: response.ok,
headers: response.headers,
body: response.body
? (options.responseSchema ?? z.any()).parse(response.body)
: undefined,
};
},
};
const eventData = this.#options.on.schema.parse(data.trigger.input);
packages/trigger-sdk/src/index.ts
+23
View File
@@ -3,7 +3,30 @@ export * from "./trigger";
export * from "./customEvents";
import { triggerRunLocalStorage } from "./localStorage";
import { SecureString } from "./types";
export function getTriggerRun() {
return triggerRunLocalStorage.getStore();
}
/*
* This function is used to create a secure string that can be used in the headers of a fetch request.
* It is used to prevent the string from being logged in trigger.dev.
* You can use it like this:
*
* await ctx.fetch("https://example.com", {
* headers: {
* Authorization: secureString`Bearer ${ACCESS_TOKEN}`,
* },
* })
*/
export function secureString(
strings: TemplateStringsArray,
...interpolations: string[]
): SecureString {
return {
__secureString: true,
strings: strings.raw as string[],
interpolations,
};
}
packages/trigger-sdk/src/types.ts
+38 -1
View File
@@ -1,6 +1,12 @@
import { SerializableCustomEventSchema } from "@trigger.dev/common-schemas";
import {
SerializableCustomEventSchema,
SerializableJsonSchema,
SecureString,
} from "@trigger.dev/common-schemas";
import { z } from "zod";
export type { SecureString };
export type TriggerCustomEvent = z.infer<typeof SerializableCustomEventSchema>;
export type WaitForOptions = {
@@ -10,6 +16,32 @@ export type WaitForOptions = {
days?: number;
};
export type FetchOptions<
TResponseBodySchema extends z.ZodTypeAny = z.ZodTypeAny
> = {
method:
| "GET"
| "POST"
| "PUT"
| "DELETE"
| "PATCH"
| "HEAD"
| "OPTIONS"
| "TRACE";
body?: z.infer<typeof SerializableJsonSchema>;
headers?: Record<string, string | SecureString>;
responseSchema?: TResponseBodySchema;
};
export type FetchResponse<
TResponseBodySchema extends z.ZodTypeAny = z.ZodTypeAny
> = {
ok: boolean;
body?: z.infer<TResponseBodySchema>;
headers: Record<string, string>;
status: number;
};
export interface TriggerContext {
id: string;
environment: string;
@@ -19,6 +51,11 @@ export interface TriggerContext {
sendEvent(key: string, event: TriggerCustomEvent): Promise<void>;
waitFor(key: string, options: WaitForOptions): Promise<void>;
waitUntil(key: string, date: Date): Promise<void>;
fetch<TBodySchema extends z.ZodTypeAny = z.ZodTypeAny>(
key: string,
url: string | URL,
options: FetchOptions<TBodySchema>
): Promise<FetchResponse<TBodySchema>>;
}
export interface TriggerLogger {
pnpm-lock.yaml
Generated
+101 -89
View File
@@ -180,7 +180,7 @@ importers:
'@aws-sdk/client-s3': 3.245.0
'@aws-sdk/s3-request-presigner': 3.245.0
'@cfworker/json-schema': 1.12.5
'@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde
'@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu
'@codemirror/commands': 6.1.3
'@codemirror/lang-javascript': 6.1.2
'@codemirror/lang-json': 6.0.1
@@ -205,7 +205,7 @@ importers:
'@tanstack/react-table': 8.7.6_biqbaboplfbrettd7655fr4n2y
'@trigger.dev/common-schemas': link:../../packages/common-schemas
'@trigger.dev/providers': link:../../packages/trigger-providers
'@uiw/react-codemirror': 4.19.5_aguurb4bmecpxzejz52amioxne
'@uiw/react-codemirror': 4.19.5_k4ec5g7vuuzzonc3d6xbjnmmle
bcryptjs: 2.4.3
classnames: 2.3.2
clsx: 1.2.1
@@ -411,6 +411,23 @@ importers:
config-packages/tsconfig:
specifiers: {}
examples/fetch-playground:
specifiers:
'@trigger.dev/integrations': workspace:*
'@trigger.dev/sdk': workspace:*
'@trigger.dev/tsconfig': workspace:*
'@types/node': '16'
tsx: ^3.12.0
zod: ^3.20.2
dependencies:
'@trigger.dev/integrations': link:../../packages/trigger-integrations
'@trigger.dev/sdk': link:../../packages/trigger-sdk
zod: 3.20.2
devDependencies:
'@trigger.dev/tsconfig': link:../../config-packages/tsconfig
'@types/node': 16.18.11
tsx: 3.12.2
examples/github-webhook:
specifiers:
'@trigger.dev/integrations': workspace:*
@@ -649,7 +666,7 @@ importers:
'@urql/core': 3.1.1_graphql@16.6.0
debug: 4.3.4
graphql: 16.6.0
urql: 3.0.3_onqnqwb3ubg5opvemcqf7c2qhy
urql: 3.0.3_graphql@16.6.0
zod: 3.20.2
devDependencies:
'@trigger.dev/providers': link:../trigger-providers
@@ -3399,13 +3416,12 @@ packages:
prettier: 2.8.2
dev: false
/@codemirror/autocomplete/6.4.0_eo6pz6bvsllvatnnwfprpuflde:
/@codemirror/autocomplete/6.4.0_czcfkg2f66rxeiodoti7r2gulu:
resolution: {integrity: sha512-HLF2PnZAm1s4kGs30EiqKMgD7XsYaQ0XJnMR0rofEWQ5t5D60SfqpDIkIh1ze5tiEbyUWm8+VJ6W1/erVvBMIA==}
peerDependencies:
'@codemirror/language': ^6.0.0
'@codemirror/state': ^6.0.0
'@codemirror/view': ^6.0.0
'@lezer/common': ^1.0.0
dependencies:
'@codemirror/language': 6.3.2
'@codemirror/state': 6.2.0
@@ -3425,7 +3441,7 @@ packages:
/@codemirror/lang-javascript/6.1.2:
resolution: {integrity: sha512-OcwLfZXdQ1OHrLiIcKCn7MqZ7nx205CMKlhe+vL88pe2ymhT9+2P+QhwkYGxMICj8TDHyp8HFKVwpiisUT7iEQ==}
dependencies:
'@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde
'@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu
'@codemirror/language': 6.3.2
'@codemirror/lint': 6.1.0
'@codemirror/state': 6.2.0
@@ -4663,7 +4679,7 @@ packages:
eslint: 8.31.0
eslint-import-resolver-node: 0.3.6
eslint-import-resolver-typescript: 3.5.3_hnftvkj7qg3s6bbigj4pr6djxy
eslint-plugin-import: 2.27.4_qdjeohovcytra7xto5vgmxssaq
eslint-plugin-import: 2.27.4_2ac3tknkazjoq5fxmuugu665ny
eslint-plugin-jest: 26.9.0_ohsifnwenhmxgcp7mend4dnv74
eslint-plugin-jest-dom: 4.0.3_eslint@8.31.0
eslint-plugin-jsx-a11y: 6.7.1_eslint@8.31.0
@@ -5795,18 +5811,17 @@ packages:
eslint-visitor-keys: 3.3.0
dev: true
/@uiw/codemirror-extensions-basic-setup/4.19.5_tbeldtdcrf45b35pezgkzq2u4e:
/@uiw/codemirror-extensions-basic-setup/4.19.5_wd2tsis3in55bkaiwnc2c46tom:
resolution: {integrity: sha512-1zt7ZPJ01xKkSW/KDy0FZNga0bngN1fC594wCVG7FBi60ehfcAucpooQ+JSPScKXopxcb+ugPKZvVLzr9/OfzA==}
peerDependencies:
'@codemirror/autocomplete': '>=6.0.0'
'@codemirror/commands': '>=6.0.0'
'@codemirror/language': '>=6.0.0'
'@codemirror/lint': '>=6.0.0'
'@codemirror/search': '>=6.0.0'
'@codemirror/state': '>=6.0.0'
'@codemirror/view': '>=6.0.0'
dependencies:
'@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde
'@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu
'@codemirror/commands': 6.1.3
'@codemirror/language': 6.3.2
'@codemirror/lint': 6.1.0
@@ -5815,14 +5830,11 @@ packages:
'@codemirror/view': 6.7.2
dev: false
/@uiw/react-codemirror/4.19.5_aguurb4bmecpxzejz52amioxne:
/@uiw/react-codemirror/4.19.5_k4ec5g7vuuzzonc3d6xbjnmmle:
resolution: {integrity: sha512-ZCHh8d7beXbF8/t7F1+yHht6A9Y6CdKeOkZq4A09lxJEnyTQrj1FMf2zvfaqc7K23KNjkTCtSlbqKKbVDgrWaw==}
peerDependencies:
'@babel/runtime': '>=7.11.0'
'@codemirror/state': '>=6.0.0'
'@codemirror/theme-one-dark': '>=6.0.0'
'@codemirror/view': '>=6.0.0'
codemirror: '>=6.0.0'
react: '>=16.8.0'
react-dom: '>=16.8.0'
dependencies:
@@ -5831,14 +5843,13 @@ packages:
'@codemirror/state': 6.2.0
'@codemirror/theme-one-dark': 6.1.0
'@codemirror/view': 6.7.2
'@uiw/codemirror-extensions-basic-setup': 4.19.5_tbeldtdcrf45b35pezgkzq2u4e
codemirror: 6.0.1_@lezer+common@1.0.2
'@uiw/codemirror-extensions-basic-setup': 4.19.5_wd2tsis3in55bkaiwnc2c46tom
codemirror: 6.0.1
react: 18.2.0
react-dom: 18.2.0_react@18.2.0
transitivePeerDependencies:
- '@codemirror/autocomplete'
- '@codemirror/language'
- '@codemirror/lint'
- '@codemirror/search'
dev: false
@@ -7164,18 +7175,16 @@ packages:
engines: {node: '>=0.10.0'}
dev: false
/codemirror/6.0.1_@lezer+common@1.0.2:
/codemirror/6.0.1:
resolution: {integrity: sha512-J8j+nZ+CdWmIeFIGXEFbFPtpiYacFMDR8GlHK3IyHQJMCaVRfGx9NT+Hxivv1ckLWPvNdZqndbr/7lVhrf/Svg==}
dependencies:
'@codemirror/autocomplete': 6.4.0_eo6pz6bvsllvatnnwfprpuflde
'@codemirror/autocomplete': 6.4.0_czcfkg2f66rxeiodoti7r2gulu
'@codemirror/commands': 6.1.3
'@codemirror/language': 6.3.2
'@codemirror/lint': 6.1.0
'@codemirror/search': 6.2.3
'@codemirror/state': 6.2.0
'@codemirror/view': 6.7.2
transitivePeerDependencies:
- '@lezer/common'
dev: false
/collection-visit/1.0.0:
@@ -8494,7 +8503,7 @@ packages:
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
eslint-import-resolver-typescript: 2.7.1_hnftvkj7qg3s6bbigj4pr6djxy
eslint-plugin-import: 2.27.4_eslint@8.31.0
eslint-plugin-import: 2.27.4_2es4x7ly2gmvaoztzzbbx3tgsy
eslint-plugin-jsx-a11y: 6.7.1_eslint@8.31.0
eslint-plugin-react: 7.31.8_eslint@8.31.0
eslint-plugin-react-hooks: 4.6.0_eslint@8.31.0
@@ -8550,7 +8559,7 @@ packages:
dependencies:
debug: 4.3.4
eslint: 8.31.0
eslint-plugin-import: 2.27.4_eslint@8.31.0
eslint-plugin-import: 2.27.4_2es4x7ly2gmvaoztzzbbx3tgsy
glob: 7.2.3
is-glob: 4.0.3
resolve: 1.22.1
@@ -8569,7 +8578,7 @@ packages:
debug: 4.3.4
enhanced-resolve: 5.12.0
eslint: 8.31.0
eslint-plugin-import: 2.27.4_qdjeohovcytra7xto5vgmxssaq
eslint-plugin-import: 2.27.4_2ac3tknkazjoq5fxmuugu665ny
get-tsconfig: 4.3.0
globby: 13.1.3
is-core-module: 2.11.0
@@ -8579,35 +8588,7 @@ packages:
- supports-color
dev: true
/eslint-module-utils/2.7.4_py7h5widiydke2tta7oabbx7ru:
resolution: {integrity: sha512-j4GT+rqzCoRKHwURX7pddtIPGySnX9Si/cgMI5ztrcqOPtk5dDEeZ34CQVPphnqkJytlc97Vuk05Um2mJ3gEQA==}
engines: {node: '>=4'}
peerDependencies:
'@typescript-eslint/parser': '*'
eslint: '*'
eslint-import-resolver-node: '*'
eslint-import-resolver-typescript: '*'
eslint-import-resolver-webpack: '*'
peerDependenciesMeta:
'@typescript-eslint/parser':
optional: true
eslint:
optional: true
eslint-import-resolver-node:
optional: true
eslint-import-resolver-typescript:
optional: true
eslint-import-resolver-webpack:
optional: true
dependencies:
debug: 3.2.7
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
transitivePeerDependencies:
- supports-color
dev: true
/eslint-module-utils/2.7.4_sqt5xxn4ciiurbqrzlaarm6ama:
/eslint-module-utils/2.7.4_co4ldsxivxjwelu6vt6qmibemq:
resolution: {integrity: sha512-j4GT+rqzCoRKHwURX7pddtIPGySnX9Si/cgMI5ztrcqOPtk5dDEeZ34CQVPphnqkJytlc97Vuk05Um2mJ3gEQA==}
engines: {node: '>=4'}
peerDependencies:
@@ -8632,6 +8613,37 @@ packages:
debug: 3.2.7
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
eslint-import-resolver-typescript: 2.7.1_hnftvkj7qg3s6bbigj4pr6djxy
transitivePeerDependencies:
- supports-color
dev: true
/eslint-module-utils/2.7.4_v73lhamtbyinynmwa5fn7kpmfq:
resolution: {integrity: sha512-j4GT+rqzCoRKHwURX7pddtIPGySnX9Si/cgMI5ztrcqOPtk5dDEeZ34CQVPphnqkJytlc97Vuk05Um2mJ3gEQA==}
engines: {node: '>=4'}
peerDependencies:
'@typescript-eslint/parser': '*'
eslint: '*'
eslint-import-resolver-node: '*'
eslint-import-resolver-typescript: '*'
eslint-import-resolver-webpack: '*'
peerDependenciesMeta:
'@typescript-eslint/parser':
optional: true
eslint:
optional: true
eslint-import-resolver-node:
optional: true
eslint-import-resolver-typescript:
optional: true
eslint-import-resolver-webpack:
optional: true
dependencies:
'@typescript-eslint/parser': 5.48.1_iukboom6ndih5an6iafl45j2fe
debug: 3.2.7
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
eslint-import-resolver-typescript: 3.5.3_hnftvkj7qg3s6bbigj4pr6djxy
transitivePeerDependencies:
- supports-color
dev: true
@@ -8656,39 +8668,7 @@ packages:
regexpp: 3.2.0
dev: true
/eslint-plugin-import/2.27.4_eslint@8.31.0:
resolution: {integrity: sha512-Z1jVt1EGKia1X9CnBCkpAOhWy8FgQ7OmJ/IblEkT82yrFU/xJaxwujaTzLWqigewwynRQ9mmHfX9MtAfhxm0sA==}
engines: {node: '>=4'}
peerDependencies:
'@typescript-eslint/parser': '*'
eslint: ^2 || ^3 || ^4 || ^5 || ^6 || ^7.2.0 || ^8
peerDependenciesMeta:
'@typescript-eslint/parser':
optional: true
dependencies:
array-includes: 3.1.6
array.prototype.flat: 1.3.1
array.prototype.flatmap: 1.3.1
debug: 3.2.7
doctrine: 2.1.0
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
eslint-module-utils: 2.7.4_py7h5widiydke2tta7oabbx7ru
has: 1.0.3
is-core-module: 2.11.0
is-glob: 4.0.3
minimatch: 3.1.2
object.values: 1.1.6
resolve: 1.22.1
semver: 6.3.0
tsconfig-paths: 3.14.1
transitivePeerDependencies:
- eslint-import-resolver-typescript
- eslint-import-resolver-webpack
- supports-color
dev: true
/eslint-plugin-import/2.27.4_qdjeohovcytra7xto5vgmxssaq:
/eslint-plugin-import/2.27.4_2ac3tknkazjoq5fxmuugu665ny:
resolution: {integrity: sha512-Z1jVt1EGKia1X9CnBCkpAOhWy8FgQ7OmJ/IblEkT82yrFU/xJaxwujaTzLWqigewwynRQ9mmHfX9MtAfhxm0sA==}
engines: {node: '>=4'}
peerDependencies:
@@ -8706,7 +8686,40 @@ packages:
doctrine: 2.1.0
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
eslint-module-utils: 2.7.4_sqt5xxn4ciiurbqrzlaarm6ama
eslint-module-utils: 2.7.4_v73lhamtbyinynmwa5fn7kpmfq
has: 1.0.3
is-core-module: 2.11.0
is-glob: 4.0.3
minimatch: 3.1.2
object.values: 1.1.6
resolve: 1.22.1
semver: 6.3.0
tsconfig-paths: 3.14.1
transitivePeerDependencies:
- eslint-import-resolver-typescript
- eslint-import-resolver-webpack
- supports-color
dev: true
/eslint-plugin-import/2.27.4_2es4x7ly2gmvaoztzzbbx3tgsy:
resolution: {integrity: sha512-Z1jVt1EGKia1X9CnBCkpAOhWy8FgQ7OmJ/IblEkT82yrFU/xJaxwujaTzLWqigewwynRQ9mmHfX9MtAfhxm0sA==}
engines: {node: '>=4'}
peerDependencies:
'@typescript-eslint/parser': '*'
eslint: ^2 || ^3 || ^4 || ^5 || ^6 || ^7.2.0 || ^8
peerDependenciesMeta:
'@typescript-eslint/parser':
optional: true
dependencies:
'@typescript-eslint/parser': 5.48.1_iukboom6ndih5an6iafl45j2fe
array-includes: 3.1.6
array.prototype.flat: 1.3.1
array.prototype.flatmap: 1.3.1
debug: 3.2.7
doctrine: 2.1.0
eslint: 8.31.0
eslint-import-resolver-node: 0.3.7
eslint-module-utils: 2.7.4_co4ldsxivxjwelu6vt6qmibemq
has: 1.0.3
is-core-module: 2.11.0
is-glob: 4.0.3
@@ -15945,7 +15958,7 @@ packages:
prepend-http: 2.0.0
dev: true
/urql/3.0.3_onqnqwb3ubg5opvemcqf7c2qhy:
/urql/3.0.3_graphql@16.6.0:
resolution: {integrity: sha512-aVUAMRLdc5AOk239DxgXt6ZxTl/fEmjr7oyU5OGo8uvpqu42FkeJErzd2qBzhAQ3DyusoZIbqbBLPlnKo/yy2A==}
peerDependencies:
graphql: ^0.11.0 || ^0.12.0 || ^0.13.0 || ^14.0.0 || ^15.0.0 || ^16.0.0
@@ -15953,7 +15966,6 @@ packages:
dependencies:
'@urql/core': 3.1.1_graphql@16.6.0
graphql: 16.6.0
react: 18.2.0
wonka: 6.1.2
dev: false