Initial work on the delay functionality

apps/coordinator/src/server.ts

@@ -89,6 +89,39 @@ export class TriggerServer {
       sender: HostRPCSchema,
       receiver: ServerRPCSchema,
       handlers: {
+        INITIALIZE_DELAY: async (data) => {
+          if (!this.#triggerPublisher) {
+            // TODO: need to recover from this issue by trying to reconnect
+            return false;
+          }
+
+          if (!this.#organizationId) {
+            // TODO: this should never really happen
+            throw new Error(
+              "Cannot complete workflow run without an organization ID"
+            );
+          }
+
+          if (!this.#workflowId) {
+            // TODO: this should never really happen
+            throw new Error("Cannot send log without a workflow ID");
+          }
+
+          const response = await this.#triggerPublisher.publish(
+            "INITIALIZE_DELAY",
+            {
+              id: data.waitId,
+              delay: data.delay,
+            },
+            {
+              "x-api-key": this.#apiKey,
+              "x-workflow-id": this.#workflowId,
+              "x-workflow-run-id": data.id,
+            }
+          );
+
+          return !!response;
+        },
         SEND_REQUEST: async (data) => {
           if (!this.#triggerPublisher) {
             // TODO: need to recover from this issue by trying to reconnect

apps/webapp/app/models/workflowRun.server.ts

@@ -134,24 +134,6 @@ export async function logMessageInRun(
   });
 }
 
-export async function initiateWaitInRun(
-  id: string,
-  wait: z.infer<typeof WaitSchema>,
-  apiKey: string
-) {
-  const workflowRun = await findWorkflowRunScopedToApiKey(id, apiKey);
-
-  await prisma.workflowRunStep.create({
-    data: {
-      runId: workflowRun.id,
-      type: "DURABLE_DELAY",
-      input: wait,
-      context: {},
-      startedAt: new Date(),
-    },
-  });
-}
-
 async function findWorkflowRunScopedToApiKey(id: string, apiKey: string) {
   const workflowRun = await prisma.workflowRun.findFirst({
     where: { id },

apps/webapp/app/services/delays/initiateDelay.server.ts

@@ -0,0 +1,12 @@
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+
+export class InitiateDelay {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  async call(runId: string, delay: { id: string; seconds: number }) {}
+}

apps/webapp/app/services/messageBroker.server.ts

@@ -16,11 +16,11 @@ import {
   completeWorkflowRun,
   failWorkflowRun,
   findWorklowRunById,
-  initiateWaitInRun,
   logMessageInRun,
   startWorkflowRun,
   triggerEventInRun,
 } from "~/models/workflowRun.server";
+import { InitiateDelay } from "./delays/initiateDelay.server";
 import { DispatchEvent } from "./events/dispatch.server";
 import { RegisterExternalSource } from "./externalSources/registerExternalSource.server";
 import { CreateIntegrationRequest } from "./requests/createIntegrationRequest.server";
@@ -177,8 +177,13 @@ async function createTriggerSubscriber() {
 
         return true;
       },
-      INITIATE_WAIT: async (id, data, properties) => {
-        await initiateWaitInRun(data.id, data.wait, properties["x-api-key"]);
+      INITIALIZE_DELAY: async (id, data, properties) => {
+        const service = new InitiateDelay();
+
+        await service.call(properties["x-workflow-run-id"], {
+          id: data.id,
+          seconds: data.delay,
+        });
 
         return true;
       },

examples/send-to-slack/src/index.ts

@@ -17,6 +17,8 @@ const trigger = new Trigger({
     }),
   }),
   run: async (event, ctx) => {
+    await ctx.waitFor(60);
+
     const response = await slack.postMessage({
       channel: "test-integrations",
       text: `New domain created: ${event.domain} by customer ${event.customerId}`,

packages/internal-bridge/src/schemas/server.ts

@@ -5,6 +5,14 @@ import {
 import { z } from "zod";
 
 export const ServerRPCSchema = {
+  INITIALIZE_DELAY: {
+    request: z.object({
+      id: z.string(),
+      waitId: z.string(),
+      delay: z.number(),
+    }),
+    response: z.boolean(),
+  },
   SEND_REQUEST: {
     request: z.object({
       id: z.string(),

packages/internal-platform/src/messages/schemas/awaits.ts

@@ -1,16 +1,14 @@
 import { WaitSchema } from "@trigger.dev/common-schemas";
 import { z } from "zod";
+import { WorkflowSendRunEventPropertiesSchema } from "../sharedSchemas";
 
 const Catalog = {
-  INITIATE_WAIT: {
+  INITIALIZE_DELAY: {
     data: z.object({
       id: z.string(),
-      wait: WaitSchema,
-    }),
-    properties: z.object({
-      "x-workflow-id": z.string(),
-      "x-api-key": z.string(),
+      delay: z.number(),
     }),
+    properties: WorkflowSendRunEventPropertiesSchema,
   },
 };
 

packages/trigger-sdk/src/client.ts

@@ -15,12 +15,6 @@ import { ContextLogger } from "./logger";
 import { triggerRunLocalStorage } from "./localStorage";
 import { ulid } from "ulid";
 
-type RequestResponse = {
-  body?: any;
-  headers: Record<string, string>;
-  status: number;
-};
-
 export class TriggerClient<TSchema extends z.ZodTypeAny> {
   #trigger: Trigger<TSchema>;
   #options: TriggerOptions<TSchema>;
@@ -43,6 +37,14 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
     }
   >();
 
+  #waitForCallbacks = new Map<
+    string,
+    {
+      resolve: () => void;
+      reject: (err?: any) => void;
+    }
+  >();
+
   constructor(trigger: Trigger<TSchema>, options: TriggerOptions<TSchema>) {
     this.#trigger = trigger;
     this.#options = options;
@@ -146,6 +148,26 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
                 event: JSON.parse(JSON.stringify(event)),
               });
             },
+            waitFor: async (seconds: number) => {
+              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,
+                delay: seconds,
+              });
+
+              await result;
+
+              return;
+            },
           };
 
           const eventData = this.#options.on.schema.parse(data.trigger.input);

packages/trigger-sdk/src/types.ts

@@ -10,6 +10,7 @@ export interface TriggerContext {
   organizationId: string;
   logger: TriggerLogger;
   fireEvent(event: CustomEvent): Promise<void>;
+  waitFor(seconds: number): Promise<void>;
 }
 
 export interface TriggerLogger {