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

Add support for durable delays (either durations or scheduled times)

Browse Source
This commit is contained in:
Eric Allam
2023-01-02 16:17:28 +00:00
parent 1fcc27b990
commit d660c581f4
27 changed files with 390 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/coordinator/src/server.ts
+33 -1
View File
@@ -111,7 +111,7 @@ export class TriggerServer {
"INITIALIZE_DELAY",
{
id: data.waitId,
delay: data.delay,
config: data.config,
},
{
"x-api-key": this.#apiKey,
@@ -373,6 +373,38 @@ export class TriggerServer {
subscriptionInitialPosition: "Earliest",
},
handlers: {
RESOLVE_DELAY: async (id, data, properties) => {
this.#logger.debug("Received resolve delay", id, data, properties);
if (!this.#serverRPC) {
throw new Error("Cannot resolve delay without an RPC connection");
}
// If the API keys don't match, then we should ignore it
// This ensures the workflow is triggered for the correct environment
if (properties["x-api-key"] !== this.#apiKey) {
return true;
}
// If the workflow id is not the same as the workflow id
// that we are listening for, then we should ignore it
if (properties["x-workflow-id"] !== this.#workflowId) {
return true;
}
const success = await this.#serverRPC.send("RESOLVE_DELAY", {
id: data.id,
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;
},
RESOLVE_INTEGRATION_REQUEST: async (id, data, properties) => {
this.#logger.debug(
"Received finish integration request",
apps/webapp/app/services/delays/initiateDelay.server.ts
+55 -1
View File
@@ -1,5 +1,11 @@
import type { WaitSchema } from "@trigger.dev/common-schemas";
import type { z } from "zod";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
import { calculateDurationInMs } from "~/utils/delays";
import { internalPubSub } from "../messageBroker.server";
type DelayConfig = z.infer<typeof WaitSchema>;
export class InitiateDelay {
#prismaClient: PrismaClient;
@@ -8,5 +14,53 @@ export class InitiateDelay {
this.#prismaClient = prismaClient;
}
async call(runId: string, delay: { id: string; seconds: number }) {}
async call(runId: string, delay: { id: string; config: DelayConfig }) {
const delayUntil = this.#calculateDelayUntil(delay.config);
// Make sure the delay is not more than 1 year in the future
if (delayUntil.getTime() > Date.now() + 365 * 24 * 60 * 60 * 1000) {
throw new Error(
`Delay is more than 1 year in the future, which is the maximum allowed by trigger.dev`
);
}
const workflowStep = await this.#prismaClient.workflowRunStep.create({
data: {
runId,
type: "DURABLE_DELAY",
input: delay.config,
context: { id: delay.id, delayUntil: delayUntil.toISOString() },
status: "RUNNING",
startedAt: new Date(),
},
});
// Create the durable delay
const durableDelay = await this.#prismaClient.durableDelay.create({
data: {
id: delay.id,
runId,
stepId: workflowStep.id,
delayUntil: this.#calculateDelayUntil(delay.config),
},
});
await internalPubSub.publish(
"RESOLVE_DELAY",
{
id: delay.id,
},
{},
{ deliverAt: durableDelay.delayUntil.getTime() }
);
}
#calculateDelayUntil(config: DelayConfig): Date {
switch (config.type) {
case "DELAY":
return new Date(Date.now() + calculateDurationInMs(config));
case "SCHEDULE_FOR":
return new Date(config.scheduledFor);
}
}
}
apps/webapp/app/services/delays/resolveDelay.server.ts
+33
View File
@@ -0,0 +1,33 @@
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
export class ResolveDelay {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
async call(id: string) {
const delay = await this.#prismaClient.durableDelay.update({
where: { id },
data: { resolvedAt: new Date() },
include: {
step: {
include: {
run: {
include: { environment: true },
},
},
},
},
});
await this.#prismaClient.workflowRunStep.update({
where: { id: delay.step.id },
data: { status: "SUCCESS", finishedAt: delay.resolvedAt },
});
return delay.step.run;
}
}
apps/webapp/app/services/messageBroker.server.ts
+24 -1
View File
@@ -28,6 +28,7 @@ import { PerformIntegrationRequest } from "./requests/performIntegrationRequest.
import { StartIntegrationRequest } from "./requests/startIntegrationRequest.server";
import { WaitForConnection } from "./requests/waitForConnection.server";
import { HandleNewServiceConnection } from "./externalServices/handleNewConnection.server";
import { ResolveDelay } from "./delays/resolveDelay.server";
let pulsarClient: PulsarClient;
let triggerPublisher: ZodPublisher<PlatformCatalog>;
@@ -183,7 +184,7 @@ async function createTriggerSubscriber() {
await service.call(properties["x-workflow-run-id"], {
id: data.id,
seconds: data.delay,
config: data.config,
});
return true;
@@ -273,6 +274,10 @@ const InternalCatalog = {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
RESOLVE_DELAY: {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
};
async function createInternalPubSub() {
@@ -288,6 +293,24 @@ async function createInternalPubSub() {
},
schema: InternalCatalog,
handlers: {
RESOLVE_DELAY: async (id, data, properties) => {
const service = new ResolveDelay();
const run = await service.call(data.id);
triggerPublisher.publish(
"RESOLVE_DELAY",
{ id: data.id },
{
"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;
},
INTEGRATION_REQUEST_CREATED: async (id, data, properties) => {
const integrationRequest = await findIntegrationRequestById(data.id);
apps/webapp/app/utils/delays.ts
+13
View File
@@ -0,0 +1,13 @@
export const calculateDurationInMs = (options: {
seconds?: number;
minutes?: number;
hours?: number;
days?: number;
}) => {
return (
(options?.seconds ?? 0) * 1000 +
(options?.minutes ?? 0) * 60 * 1000 +
(options?.hours ?? 0) * 60 * 60 * 1000 +
(options?.days ?? 0) * 24 * 60 * 60 * 1000
);
};
apps/webapp/prisma/migrations/20230102142756_add_durable_delay_model/migration.sql
+20
View File
@@ -0,0 +1,20 @@
-- CreateTable
CREATE TABLE "DurableDelay" (
"id" TEXT NOT NULL,
"runId" TEXT NOT NULL,
"stepId" TEXT NOT NULL,
"delayUntil" TIMESTAMP(3) NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"resolvedAt" TIMESTAMP(3),
CONSTRAINT "DurableDelay_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "DurableDelay_stepId_key" ON "DurableDelay"("stepId");
-- AddForeignKey
ALTER TABLE "DurableDelay" ADD CONSTRAINT "DurableDelay_runId_fkey" FOREIGN KEY ("runId") REFERENCES "WorkflowRun"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "DurableDelay" ADD CONSTRAINT "DurableDelay_stepId_fkey" FOREIGN KEY ("stepId") REFERENCES "WorkflowRunStep"("id") ON DELETE CASCADE ON UPDATE CASCADE;
apps/webapp/prisma/schema.prisma
+20 -3
View File
@@ -59,11 +59,11 @@ model APIConnection {
apiIdentifier String
status APIConnectionStatus @default(CREATED)
scopes String[]
authenticationMethod APIAuthenticationMethod @default(OAUTH)
authenticationConfig Json?
type APIConnectionType
type APIConnectionType
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
@@ -293,6 +293,21 @@ model IntegrationResponse {
createdAt DateTime @default(now())
}
model DurableDelay {
id String @id
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
delayUntil DateTime
createdAt DateTime @default(now())
resolvedAt DateTime?
}
model TriggerEvent {
id String @id @default(cuid())
service String
@@ -351,6 +366,7 @@ model WorkflowRun {
isTest Boolean @default(false)
requests IntegrationRequest[]
delays DurableDelay[]
}
enum WorkflowRunStatus {
@@ -380,6 +396,7 @@ model WorkflowRunStep {
status WorkflowRunStepStatus @default(PENDING)
integrationRequest IntegrationRequest?
delay DurableDelay?
}
enum WorkflowRunStepStatus {
@@ -395,4 +412,4 @@ enum WorkflowRunStepType {
DURABLE_DELAY
CUSTOM_EVENT
INTEGRATION_REQUEST
}
}
examples/send-to-slack/src/index.ts
+9 -1
View File
@@ -17,13 +17,21 @@ const trigger = new Trigger({
}),
}),
run: async (event, ctx) => {
// await ctx.waitFor(60);
await ctx.logger.info(
"Received domain.created event, waiting for 60 seconds..."
);
await ctx.waitFor({ seconds: 60 });
await ctx.logger.info("Posting to Slack...");
const response = await slack.postMessage({
channel: "test-integrations",
text: `New domain created: ${event.domain} by customer ${event.customerId}`,
});
await ctx.logger.info("Posted to Slack!");
return response.message;
},
});
packages/common-schemas/src/waits.ts
+4 -1
View File
@@ -2,7 +2,10 @@ import { z } from "zod";
export const DelaySchema = z.object({
type: z.literal("DELAY"),
durationInMs: z.number(),
seconds: z.number().optional(),
minutes: z.number().optional(),
hours: z.number().optional(),
days: z.number().optional(),
});
export const ScheduledForSchema = z.object({
packages/internal-bridge/src/schemas/host.ts
+13
View File
@@ -32,6 +32,19 @@ export const HostRPCSchema = {
}),
response: z.boolean(),
},
RESOLVE_DELAY: {
request: z.object({
id: z.string(),
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
+2 -1
View File
@@ -1,6 +1,7 @@
import {
CustomEventSchema,
TriggerMetadataSchema,
WaitSchema,
} from "@trigger.dev/common-schemas";
import { z } from "zod";
@@ -9,7 +10,7 @@ export const ServerRPCSchema = {
request: z.object({
id: z.string(),
waitId: z.string(),
delay: z.number(),
config: WaitSchema,
}),
response: z.boolean(),
},
packages/internal-platform/src/messages/catalogs/coordinator.ts
+10 -14
View File
@@ -1,19 +1,15 @@
import sendIntegrationRequest from "../schemas/sendIntegrationRequest";
import startWorklowRun from "../schemas/startWorkflowRun";
import failWorkflowRun from "../schemas/failWorkflowRun";
import completeWorkflowRun from "../schemas/completeWorkflowRun";
import logMessage from "../schemas/logMessage";
import triggerCustomEvent from "../schemas/triggerCustomEvent";
import awaits from "../schemas/awaits";
import { coordinator as integrationRequests } from "../schemas/integrationRequests";
import { coordinator as workflowRuns } from "../schemas/workflowRuns";
import { coordinator as logs } from "../schemas/logs";
import { coordinator as customEvents } from "../schemas/customEvents";
import { coordinator as delays } from "../schemas/delays";
const Catalog = {
...sendIntegrationRequest,
...startWorklowRun,
...failWorkflowRun,
...completeWorkflowRun,
...logMessage,
...triggerCustomEvent,
...awaits,
...integrationRequests,
...workflowRuns,
...logs,
...customEvents,
...delays,
};
export default Catalog;
packages/internal-platform/src/messages/catalogs/platform.ts
+6 -4
View File
@@ -1,9 +1,11 @@
import triggerWorkflow from "../schemas/triggerWorkflow";
import finishIntegrationRequest from "../schemas/finishIntegrationRequest";
import { platform as workflows } from "../schemas/workflows";
import { platform as integrationRequests } from "../schemas/integrationRequests";
import { platform as delays } from "../schemas/delays";
const Catalog = {
...triggerWorkflow,
...finishIntegrationRequest,
...workflows,
...integrationRequests,
...delays,
};
export default Catalog;
packages/internal-platform/src/messages/schemas/awaits.ts
-15
View File
@@ -1,15 +0,0 @@
import { WaitSchema } from "@trigger.dev/common-schemas";
import { z } from "zod";
import { WorkflowSendRunEventPropertiesSchema } from "../sharedSchemas";
const Catalog = {
INITIALIZE_DELAY: {
data: z.object({
id: z.string(),
delay: z.number(),
}),
properties: WorkflowSendRunEventPropertiesSchema,
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/completeWorkflowRun.ts
-16
View File
@@ -1,16 +0,0 @@
import { z } from "zod";
const Catalog = {
COMPLETE_WORKFLOW_RUN: {
data: z.object({
id: z.string(),
output: z.string(),
}),
properties: z.object({
"x-workflow-id": z.string(),
"x-api-key": z.string(),
}),
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/triggerCustomEvent.ts → packages/internal-platform/src/messages/schemas/customEvents.ts
+1 -3
View File
@@ -1,7 +1,7 @@
import { z } from "zod";
import { CustomEventSchema } from "@trigger.dev/common-schemas";
const Catalog = {
export const coordinator = {
TRIGGER_CUSTOM_EVENT: {
data: z.object({
id: z.string(),
@@ -13,5 +13,3 @@ const Catalog = {
}),
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/delays.ts
+25
View File
@@ -0,0 +1,25 @@
import { WaitSchema } from "@trigger.dev/common-schemas";
import { z } from "zod";
import {
WorkflowRunEventPropertiesSchema,
WorkflowSendRunEventPropertiesSchema,
} from "../sharedSchemas";
export const coordinator = {
INITIALIZE_DELAY: {
data: z.object({
id: z.string(),
config: WaitSchema,
}),
properties: WorkflowSendRunEventPropertiesSchema,
},
};
export const platform = {
RESOLVE_DELAY: {
data: z.object({
id: z.string(),
}),
properties: WorkflowRunEventPropertiesSchema,
},
};
packages/internal-platform/src/messages/schemas/failWorkflowRun.ts
-17
View File
@@ -1,17 +0,0 @@
import { z } from "zod";
import { ErrorSchema } from "@trigger.dev/common-schemas";
const Catalog = {
FAIL_WORKFLOW_RUN: {
data: z.object({
id: z.string(),
error: ErrorSchema,
}),
properties: z.object({
"x-workflow-id": z.string(),
"x-api-key": z.string(),
}),
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/finishIntegrationRequest.ts
-15
View File
@@ -1,15 +0,0 @@
import { JsonSchema } from "@trigger.dev/common-schemas";
import { z } from "zod";
import { WorkflowRunEventPropertiesSchema } from "../sharedSchemas";
const Catalog = {
RESOLVE_INTEGRATION_REQUEST: {
data: z.object({
id: z.string(),
output: JsonSchema.default({}),
}),
properties: WorkflowRunEventPropertiesSchema,
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/integrationRequests.ts
+28
View File
@@ -0,0 +1,28 @@
import { JsonSchema } from "@trigger.dev/common-schemas";
import { z } from "zod";
import {
WorkflowRunEventPropertiesSchema,
WorkflowSendRunEventPropertiesSchema,
} from "../sharedSchemas";
export const platform = {
RESOLVE_INTEGRATION_REQUEST: {
data: z.object({
id: z.string(),
output: JsonSchema.default({}),
}),
properties: WorkflowRunEventPropertiesSchema,
},
};
export const coordinator = {
SEND_INTEGRATION_REQUEST: {
data: z.object({
id: z.string(),
service: z.string(),
endpoint: z.string(),
params: z.any(),
}),
properties: WorkflowSendRunEventPropertiesSchema,
},
};
packages/internal-platform/src/messages/schemas/logMessage.ts → packages/internal-platform/src/messages/schemas/logs.ts
+1 -3
View File
@@ -1,7 +1,7 @@
import { LogMessageSchema } from "@trigger.dev/common-schemas";
import { z } from "zod";
const Catalog = {
export const coordinator = {
LOG_MESSAGE: {
data: z.object({
id: z.string(),
@@ -13,5 +13,3 @@ const Catalog = {
}),
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/sendIntegrationRequest.ts
-16
View File
@@ -1,16 +0,0 @@
import { z } from "zod";
import { WorkflowSendRunEventPropertiesSchema } from "../sharedSchemas";
const Catalog = {
SEND_INTEGRATION_REQUEST: {
data: z.object({
id: z.string(),
service: z.string(),
endpoint: z.string(),
params: z.any(),
}),
properties: WorkflowSendRunEventPropertiesSchema,
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/startWorkflowRun.ts
-15
View File
@@ -1,15 +0,0 @@
import { z } from "zod";
const Catalog = {
START_WORKFLOW_RUN: {
data: z.object({
id: z.string(),
}),
properties: z.object({
"x-workflow-id": z.string(),
"x-api-key": z.string(),
}),
},
};
export default Catalog;
packages/internal-platform/src/messages/schemas/workflowRuns.ts
+34
View File
@@ -0,0 +1,34 @@
import { ErrorSchema } from "@trigger.dev/common-schemas";
import { z } from "zod";
export const coordinator = {
COMPLETE_WORKFLOW_RUN: {
data: z.object({
id: z.string(),
output: z.string(),
}),
properties: z.object({
"x-workflow-id": z.string(),
"x-api-key": z.string(),
}),
},
FAIL_WORKFLOW_RUN: {
data: z.object({
id: z.string(),
error: ErrorSchema,
}),
properties: z.object({
"x-workflow-id": z.string(),
"x-api-key": z.string(),
}),
},
START_WORKFLOW_RUN: {
data: z.object({
id: z.string(),
}),
properties: z.object({
"x-workflow-id": z.string(),
"x-api-key": z.string(),
}),
},
};
packages/internal-platform/src/messages/schemas/triggerWorkflow.ts → packages/internal-platform/src/messages/schemas/workflows.ts
+1 -3
View File
@@ -8,11 +8,9 @@ export const TriggerWorkflowMessageSchema = z.object({
context: JsonSchema.default({}),
});
const Catalog = {
export const platform = {
TRIGGER_WORKFLOW: {
data: TriggerWorkflowMessageSchema,
properties: WorkflowEventPropertiesSchema,
},
};
export default Catalog;
packages/trigger-sdk/src/client.ts
+49 -3
View File
@@ -10,7 +10,7 @@ import {
} from "internal-bridge";
import * as pkg from "../package.json";
import { Trigger, TriggerOptions } from "./trigger";
import { TriggerContext } from "./types";
import { TriggerContext, WaitForOptions } from "./types";
import { ContextLogger } from "./logger";
import { triggerRunLocalStorage } from "./localStorage";
import { ulid } from "ulid";
@@ -109,6 +109,23 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
sender: ServerRPCSchema,
receiver: HostRPCSchema,
handlers: {
RESOLVE_DELAY: async (data) => {
console.log(`RESOLVE_DELAY(${data.id})`);
const waitCallbacks = this.#waitForCallbacks.get(data.id);
if (!waitCallbacks) {
throw new Error(
`Could not find wait callbacks for wait ID ${data.id}`
);
}
const { resolve, reject } = waitCallbacks;
resolve();
return true;
},
RESOLVE_REQUEST: async (data) => {
const requestCallbacks = this.#responseCompleteCallbacks.get(data.id);
@@ -148,7 +165,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
event: JSON.parse(JSON.stringify(event)),
});
},
waitFor: async (seconds: number) => {
waitFor: async (options: WaitForOptions) => {
const waitId = ulid();
const result = new Promise<void>((resolve, reject) => {
@@ -161,7 +178,36 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
await serverRPC.send("INITIALIZE_DELAY", {
id: data.id,
waitId,
delay: seconds,
config: {
type: "DELAY",
seconds: options.seconds,
minutes: options.minutes,
hours: options.hours,
days: options.days,
},
});
await result;
return;
},
waitUntil: async (date: Date) => {
const waitId = ulid();
const result = new Promise<void>((resolve, reject) => {
this.#waitForCallbacks.set(waitId, {
resolve,
reject,
});
});
await serverRPC.send("INITIALIZE_DELAY", {
id: data.id,
waitId,
config: {
type: "SCHEDULE_FOR",
scheduledFor: date.toISOString(),
},
});
await result;
packages/trigger-sdk/src/types.ts
+9 -1
View File
@@ -3,6 +3,13 @@ import { z } from "zod";
type CustomEvent = z.infer<typeof SerializableCustomEventSchema>;
export type WaitForOptions = {
seconds?: number;
minutes?: number;
hours?: number;
days?: number;
};
export interface TriggerContext {
id: string;
environment: string;
@@ -10,7 +17,8 @@ export interface TriggerContext {
organizationId: string;
logger: TriggerLogger;
fireEvent(event: CustomEvent): Promise<void>;
waitFor(seconds: number): Promise<void>;
waitFor(options: WaitForOptions): Promise<void>;
waitUntil(date: Date): Promise<void>;
}
export interface TriggerLogger {