Implement sendEvent using the new Tasks API (#113 )

apps/webapp/app/routes/api/v3/executions.$executionId.tasks.$id.complete.ts

@@ -0,0 +1,142 @@
+import { RuntimeEnvironment } from ".prisma/client";
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import {
+  CompleteTaskBodyInput,
+  CompleteTaskBodyInputSchema,
+  CompleteTaskBodyOutput,
+  RunTaskBodyOutput,
+} from "@trigger.dev/internal";
+import { RunTaskBodyOutputSchema } from "@trigger.dev/internal";
+import { z } from "zod";
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger";
+
+const ParamsSchema = z.object({
+  executionId: z.string(),
+  id: z.string(),
+});
+
+export async function action({ request, params }: ActionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const authenticatedEnv = await authenticateApiRequest(request);
+
+  if (!authenticatedEnv) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const { executionId, id } = ParamsSchema.parse(params);
+
+  // Now parse the request body
+  const anyBody = await request.json();
+
+  logger.debug("CompleteExecutionTaskService.call() request body", {
+    body: anyBody,
+    executionId,
+    id,
+  });
+
+  const body = CompleteTaskBodyInputSchema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new CompleteExecutionTaskService();
+
+  try {
+    const task = await service.call(
+      authenticatedEnv,
+      executionId,
+      id,
+      body.data
+    );
+
+    logger.debug("CompleteExecutionTaskService.call() response body", {
+      executionId,
+      id,
+      task,
+    });
+
+    if (!task) {
+      return json({ message: "Task not found" }, { status: 404 });
+    }
+
+    return json(task);
+  } catch (error) {
+    if (error instanceof Error) {
+      return json({ error: error.message }, { status: 400 });
+    }
+
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}
+
+export class CompleteExecutionTaskService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    environment: RuntimeEnvironment,
+    executionId: string,
+    id: string,
+    taskBody: CompleteTaskBodyOutput
+  ) {
+    // Using a transaction, we'll first check to see if the task already exists and return if if it does
+    // If it doesn't exist, we'll create it and return it
+    const task = await this.#prismaClient.$transaction(async (prisma) => {
+      const existingTask = await prisma.task.findUnique({
+        where: {
+          id,
+        },
+        include: {
+          execution: true,
+        },
+      });
+
+      if (!existingTask) {
+        return;
+      }
+
+      if (existingTask.executionId !== executionId) {
+        return;
+      }
+
+      if (existingTask.execution.environmentId !== environment.id) {
+        return;
+      }
+
+      if (
+        existingTask.status === "COMPLETED" ||
+        existingTask.status === "ERRORED"
+      ) {
+        return existingTask;
+      }
+
+      const task = await prisma.task.update({
+        where: {
+          id,
+        },
+        data: {
+          status: "COMPLETED",
+          output: taskBody.output ?? undefined,
+          completedAt: new Date(),
+        },
+      });
+
+      return task;
+    });
+
+    return task;
+  }
+}

examples/smoke-test/src/index.ts

@@ -32,8 +32,24 @@ new Job({
       myDate: new Date(),
     });
 
+    await ctx.sendEvent("Event 1", {
+      name: "smoke.test",
+      payload: { foo: "bar" },
+      source: "smoke-test",
+    });
+
     await ctx.wait("⏲⏲", 10);
 
+    await ctx.sendEvent(
+      "Event 2",
+      {
+        name: "smoke.test.delayed",
+        payload: { foo: "bar", delayed: true },
+        source: "smoke-test",
+      },
+      { deliverAfter: 30 }
+    );
+
     return { foo: "bar" };
   },
 }).registerWith(client);

packages/internal/src/schemas/api.ts

@@ -28,13 +28,14 @@ export const GetJobsResponseSchema = z.object({
 export const RawEventSchema = z.object({
   id: z.string().default(() => ulid()),
   name: z.string(),
-  source: z.string(),
+  source: z.string().default("trigger.dev"),
   payload: DeserializedJsonSchema,
   context: DeserializedJsonSchema.optional(),
   timestamp: z.string().datetime().optional(),
 });
 
 export type RawEvent = z.infer<typeof RawEventSchema>;
+export type SendEvent = z.input<typeof RawEventSchema>;
 
 export const ApiEventLogSchema = z.object({
   id: z.string(),
@@ -154,3 +155,18 @@ export const RunTaskBodyOutputSchema = RunTaskBodyInputSchema.extend({
 });
 
 export type RunTaskBodyOutput = z.infer<typeof RunTaskBodyOutputSchema>;
+
+export const CompleteTaskBodyInputSchema = RunTaskBodyInputSchema.pick({
+  displayProperties: true,
+  description: true,
+  params: true,
+}).extend({
+  output: SerializableJsonSchema.optional().transform((v) =>
+    DeserializedJsonSchema.parse(JSON.parse(JSON.stringify(v)))
+  ),
+});
+
+export type CompleteTaskBodyInput = z.input<typeof CompleteTaskBodyInputSchema>;
+export type CompleteTaskBodyOutput = z.infer<
+  typeof CompleteTaskBodyInputSchema
+>;

packages/trigger-sdk/src/apiClient.ts

@@ -1,8 +1,11 @@
 import type {
   ApiEventLog,
+  CompleteTaskBodyInput,
   CreateExecutionBody,
   LogMessage,
   RunTaskBodyInput,
+  SendEvent,
+  SendEventOptions,
   ServerTask,
 } from "@trigger.dev/internal";
 import { Logger, LogLevel } from "@trigger.dev/internal";
@@ -184,6 +187,78 @@ export class ApiClient {
     return await response.json();
   }
 
+  async completeTask(
+    executionId: string,
+    id: string,
+    task: CompleteTaskBodyInput
+  ): Promise<ServerTask> {
+    const apiKey = await this.#apiKey();
+
+    this.#logger.debug("Complete Task", {
+      task,
+    });
+
+    const response = await fetch(
+      `${this.#apiUrl}/api/v3/executions/${executionId}/tasks/${id}/complete`,
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          Authorization: `Bearer ${apiKey}`,
+        },
+        body: JSON.stringify(task),
+      }
+    );
+
+    if (response.status >= 400 && response.status < 500) {
+      const body = await response.json();
+
+      throw new Error(body.error);
+    }
+
+    if (response.status !== 200) {
+      throw new Error(
+        `Failed to create execution, got status code ${response.status}`
+      );
+    }
+
+    return await response.json();
+  }
+
+  async sendEvent(
+    event: SendEvent,
+    options: SendEventOptions = {}
+  ): Promise<ApiEventLog> {
+    const apiKey = await this.#apiKey();
+
+    this.#logger.debug("Sending event", {
+      event,
+    });
+
+    const response = await fetch(`${this.#apiUrl}/api/v3/events`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${apiKey}`,
+      },
+      body: JSON.stringify({ event, options }),
+    });
+
+    if (response.status >= 400 && response.status < 500) {
+      const body = await response.json();
+
+      throw new Error(body.error);
+    }
+
+    if (response.status !== 200) {
+      throw new Error(
+        `Failed to create execution, got status code ${response.status}`
+      );
+    }
+
+    return await response.json();
+  }
+
   async #apiKey() {
     const apiKey = getApiKey(this.#options.apiKey);
 

packages/trigger-sdk/src/io.ts

@@ -3,6 +3,7 @@ import {
   IOTask,
   Logger,
   LogLevel,
+  SerializableJson,
   ServerTask,
 } from "@trigger.dev/internal";
 import { webcrypto } from "node:crypto";
@@ -40,7 +41,7 @@ export class IO {
     }
   }
 
-  async runTask<T = void>(
+  async runTask<T extends SerializableJson | void = void>(
     key: string | any[],
     options: IOTask,
     callback: (task: ServerTask) => Promise<T>
@@ -94,14 +95,23 @@ export class IO {
       throw new ResumeWithTask(task);
     }
 
-    const result = await callback(task);
+    try {
+      const result = await callback(task);
 
-    this.#logger.debug("Using task output", {
-      idempotencyKey,
-      task,
-    });
+      this.#logger.debug("Completing using output", {
+        idempotencyKey,
+        task,
+      });
 
-    return result;
+      await this.#apiClient.completeTask(this.#id, task.id, {
+        output: result ?? undefined,
+      });
+
+      return result;
+    } catch (error) {
+      // TODO: implement this
+      throw error;
+    }
   }
 
   #addToCachedTasks(task: ServerTask) {

packages/trigger-sdk/src/triggerClient.ts

@@ -315,6 +315,18 @@ export class TriggerClient {
           async (task) => {}
         );
       },
+      sendEvent: async (key, event, options) => {
+        return await io.runTask(
+          key,
+          {
+            name: "sendEvent",
+            params: { event, options },
+          },
+          async (task) => {
+            return await this.#client.sendEvent(event, options);
+          }
+        );
+      },
     };
   }
 }

packages/trigger-sdk/src/types.ts

@@ -1,4 +1,10 @@
-import type { SecureString } from "@trigger.dev/internal";
+import type {
+  ApiEventLog,
+  RawEvent,
+  SecureString,
+  SendEvent,
+  SendEventOptions,
+} from "@trigger.dev/internal";
 
 export type { SecureString };
 
@@ -11,8 +17,12 @@ export interface TriggerContext {
   isTest: boolean;
   logger: TaskLogger;
   signal: AbortSignal;
-  wait(key: string, seconds: number): Promise<void>;
-  // sendEvent(key: string, event: TriggerCustomEvent): Promise<void>;
+  wait(key: string | any[], seconds: number): Promise<void>;
+  sendEvent(
+    key: string | any[],
+    event: SendEvent,
+    options?: SendEventOptions
+  ): Promise<ApiEventLog>;
   // waitUntil(key: string, date: Date): Promise<void>;
   // runOnce<T extends TriggerRunOnceCallback>(
   //   key: string,