Update pnpm lock file

chore: Update version for release

apps/webapp/app/env.server.ts

@@ -11,7 +11,10 @@ const EnvironmentSchema = z.object({
   SESSION_SECRET: z.string(),
   MAGIC_LINK_SECRET: z.string(),
   ENCRYPTION_KEY: z.string(),
-  WHITELISTED_EMAILS: z.string().refine(isValidRegex, "WHITELISTED_EMAILS must be a valid regex.").optional(),
+  WHITELISTED_EMAILS: z
+    .string()
+    .refine(isValidRegex, "WHITELISTED_EMAILS must be a valid regex.")
+    .optional(),
   REMIX_APP_PORT: z.string().optional(),
   LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ORIGIN: z.string().default("http://localhost:3030"),
@@ -42,7 +45,11 @@ const EnvironmentSchema = z.object({
   EXECUTION_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
   WORKER_ENABLED: z.string().default("true"),
   EXECUTION_WORKER_ENABLED: z.string().default("true"),
+  TASK_OPERATION_WORKER_ENABLED: z.string().default("true"),
+  TASK_OPERATION_WORKER_CONCURRENCY: z.coerce.number().int().default(10),
+  TASK_OPERATION_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
   GRACEFUL_SHUTDOWN_TIMEOUT: z.coerce.number().int().default(60000),
+  DISABLE_SSE: z.string().optional(),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/hooks/useEventSource.tsx

@@ -0,0 +1,38 @@
+import { useEffect, useState } from "react";
+
+type EventSourceOptions = {
+  init?: EventSourceInit;
+  event?: string;
+};
+
+/**
+ * Subscribe to an event source and return the latest event.
+ * @param url The URL of the event source to connect to
+ * @param options The options to pass to the EventSource constructor
+ * @returns The last event received from the server
+ */
+export function useEventSource(
+  url: string | URL,
+  { event = "message", init }: EventSourceOptions = {}
+) {
+  const [data, setData] = useState<string | null>(null);
+
+  useEffect(() => {
+    const eventSource = new EventSource(url, init);
+    eventSource.addEventListener(event ?? "message", handler);
+
+    // rest data if dependencies change
+    setData(null);
+
+    function handler(event: MessageEvent) {
+      setData(event.data || "UNKNOWN_EVENT_DATA");
+    }
+
+    return () => {
+      eventSource.removeEventListener(event ?? "message", handler);
+      eventSource.close();
+    };
+  }, [url, event, init]);
+
+  return data;
+}

apps/webapp/app/hooks/useProjectSetupComplete.ts

@@ -1,9 +1,9 @@
 import { useEffect } from "react";
-import { useEventSource } from "remix-utils/sse/react";
 import { projectPath, projectStreamingPath } from "~/utils/pathBuilder";
 import { useProject } from "./useProject";
 import { useOrganization } from "./useOrganizations";
 import { useNavigate } from "@remix-run/react";
+import { useEventSource } from "./useEventSource";
 
 export function useProjectSetupComplete() {
   const project = useProject();

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

@@ -0,0 +1,22 @@
+import { z } from "zod";
+
+export const JobVersionDispatchableSchema = z.object({
+  type: z.literal("JOB_VERSION"),
+  id: z.string(),
+});
+
+export const DynamicTriggerDispatchableSchema = z.object({
+  type: z.literal("DYNAMIC_TRIGGER"),
+  id: z.string(),
+});
+
+export const EphemeralDispatchableSchema = z.object({
+  type: z.literal("EPHEMERAL"),
+  url: z.string(),
+});
+
+export const DispatchableSchema = z.discriminatedUnion("type", [
+  JobVersionDispatchableSchema,
+  DynamicTriggerDispatchableSchema,
+  EphemeralDispatchableSchema,
+]);

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

@@ -16,6 +16,7 @@ export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask
     description: task.description,
     params: task.params as any,
     output: task.output as any,
+    context: task.context as any,
     properties: task.properties as any,
     style: task.style as any,
     error: task.error,

apps/webapp/app/platform/zodWorker.server.ts

@@ -14,7 +14,6 @@ import { run as graphileRun, parseCronItems } from "graphile-worker";
 import omit from "lodash.omit";
 import { z } from "zod";
 import { PrismaClient, PrismaClientOrTransaction } from "~/db.server";
-import { PgListenService } from "~/services/db/pgListen.server";
 import { workerLogger as logger } from "~/services/logger.server";
 
 export interface MessageCatalogSchema {
@@ -167,21 +166,6 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
 
     this.#runner?.events.on("pool:listen:success", async ({ workerPool, client }) => {
       this.#logDebug("pool:listen:success");
-
-      // hijack client instance to listen and react to incoming NOTIFY events
-      const pgListen = new PgListenService(client, this.#name, logger);
-
-      await pgListen.on("trigger:graphile:migrate", async ({ latestMigration }) => {
-        this.#logDebug("Detected incoming migration", { latestMigration });
-
-        if (latestMigration > 10) {
-          // already migrated past v0.14 - nothing to do
-          return;
-        }
-
-        // simulate SIGTERM to trigger graceful shutdown
-        this._handleSignal("SIGTERM");
-      });
     });
 
     this.#runner?.events.on("pool:listen:error", ({ error }) => {

apps/webapp/app/presenters/EnvironmentsStreamPresenter.server.ts

@@ -1,7 +1,7 @@
 import { PrismaClient, prisma } from "~/db.server";
 import { Project } from "~/models/project.server";
 import { User } from "~/models/user.server";
-import { sse } from "~/utils/sse";
+import { sse } from "~/utils/sse.server";
 
 type EnvironmentSignalsMap = {
   [x: string]: {

apps/webapp/app/presenters/RunStreamPresenter.server.ts

@@ -1,6 +1,6 @@
 import { JobRun } from "@trigger.dev/database";
 import { PrismaClient, prisma } from "~/db.server";
-import { sse } from "~/utils/sse";
+import { sse } from "~/utils/sse.server";
 
 export class RunStreamPresenter {
   #prismaClient: PrismaClient;

apps/webapp/app/routes/_app._index/route.tsx

@@ -1,19 +1,26 @@
 import { LoaderFunctionArgs, redirect } from "@remix-run/server-runtime";
+import { getUsersInvites } from "~/models/member.server";
 import { SelectBestProjectPresenter } from "~/presenters/SelectBestProjectPresenter.server";
-import { requireUserId } from "~/services/session.server";
-import { newOrganizationPath, projectPath } from "~/utils/pathBuilder";
+import { requireUser } from "~/services/session.server";
+import { invitesPath, newOrganizationPath, projectPath } from "~/utils/pathBuilder";
 
 //this loader chooses the best project to redirect you to, ideally based on the cookie
 export const loader = async ({ request }: LoaderFunctionArgs) => {
-  const userId = await requireUserId(request);
+  const user = await requireUser(request);
+
+  //if there are invites then we should redirect to the invites page
+  const invites = await getUsersInvites({ email: user.email });
+  if (invites.length > 0) {
+    return redirect(invitesPath());
+  }
 
   const presenter = new SelectBestProjectPresenter();
   try {
-    const { project, organization } = await presenter.call({ userId, request });
-
+    const { project, organization } = await presenter.call({ userId: user.id, request });
+    //redirect them to the most appropriate project
     return redirect(projectPath(organization, project));
   } catch (e) {
-    //this should only happen if the user has no projects
+    //this should only happen if the user has no projects, and no invites
     return redirect(newOrganizationPath());
   }
 };

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.environments/ConfigureEndpointSheet.tsx

@@ -2,7 +2,7 @@ import { conform, useForm } from "@conform-to/react";
 import { parse } from "@conform-to/zod";
 import { useFetcher, useRevalidator } from "@remix-run/react";
 import { useEffect } from "react";
-import { useEventSource } from "remix-utils/sse/react";
+import { useEventSource } from "~/hooks/useEventSource";
 import { InlineCode } from "~/components/code/InlineCode";
 import {
   EndpointIndexStatusIcon,

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.environments/FirstEndpointSheet.tsx

@@ -51,7 +51,7 @@ export function FirstEndpointSheet({ projectId, environments }: FirstEndpointShe
   return (
     <Sheet>
       <SheetTrigger>
-        <Button variant="secondary/medium">Add your first endpoint</Button>
+        <ButtonContent variant="secondary/medium">Add your first endpoint</ButtonContent>
       </SheetTrigger>
       <SheetContent size="lg">
         <SheetHeader>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.environments/route.tsx

@@ -2,7 +2,7 @@ import { useRevalidator } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { useEffect, useMemo, useState } from "react";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
-import { useEventSource } from "remix-utils/sse/react";
+import { useEventSource } from "~/hooks/useEventSource";
 import {
   EndpointIndexStatusIcon,
   EndpointIndexStatusLabel,

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx

@@ -2,7 +2,7 @@ import { useRevalidator } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { Fragment, useEffect } from "react";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
-import { useEventSource } from "remix-utils/sse/react";
+import { useEventSource } from "~/hooks/useEventSource";
 import { BreadcrumbLink } from "~/components/navigation/Breadcrumb";
 import { BreadcrumbIcon } from "~/components/primitives/BreadcrumbIcon";
 import { RunOverview } from "~/components/run/RunOverview";

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx

@@ -1,6 +1,5 @@
 import { useForm } from "@conform-to/react";
 import { parse } from "@conform-to/zod";
-import { ClipboardIcon } from "@heroicons/react/20/solid";
 import { ClockIcon, CodeBracketIcon } from "@heroicons/react/24/outline";
 import { Form, useActionData, useSubmit } from "@remix-run/react";
 import { ActionFunction, LoaderFunctionArgs, json } from "@remix-run/server-runtime";
@@ -33,10 +32,14 @@ import { redirectBackWithErrorMessage, redirectWithSuccessMessage } from "~/mode
 import { TestJobPresenter } from "~/presenters/TestJobPresenter.server";
 import { TestJobService } from "~/services/jobs/testJob.server";
 import { requireUserId } from "~/services/session.server";
-import { cn } from "~/utils/cn";
 import { Handle } from "~/utils/handle";
 import { isValidIcon } from "~/utils/icon";
-import { JobParamsSchema, jobRunDashboardPath, trimTrailingSlash } from "~/utils/pathBuilder";
+import {
+  JobParamsSchema,
+  docsPath,
+  jobRunDashboardPath,
+  trimTrailingSlash,
+} from "~/utils/pathBuilder";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
@@ -139,7 +142,7 @@ export default function Page() {
     setDefaultJson(code);
   }, []);
 
-  const [selectedEnvironmentId, setSelectedEnvironmentId] = useState<string>(environments[0].id);
+  const [selectedEnvironmentId, setSelectedEnvironmentId] = useState(environments.at(0)?.id);
   const selectedEnvironment = environments.find((e) => e.id === selectedEnvironmentId);
 
   const currentJson = useRef<string>(defaultJson);
@@ -147,6 +150,10 @@ export default function Page() {
 
   const submitForm = useCallback(
     (e: React.FormEvent<HTMLFormElement>) => {
+      if (!selectedEnvironmentId) {
+        return;
+      }
+
       submit(
         {
           payload: currentJson.current,
@@ -175,10 +182,33 @@ export default function Page() {
 
   if (environments.length === 0) {
     return (
-      <Callout variant="warning">
-        Can't run a test when there are no environments. This shouldn't happen, please contact
-        support.
-      </Callout>
+      <div className="flex flex-col gap-4">
+        <Callout variant="info">
+          There are no environments that you can test this job with – you can't run Tests against
+          your teammates' Dev environments. You should run the code locally (using the CLI) so that
+          this Job will be associated with your Dev environment. This also means that this Job
+          hasn't been deployed to Staging or Prod yet.
+        </Callout>
+        <div>
+          <Header2 spacing>Useful guides</Header2>
+          <div className="flex gap-2">
+            <LinkButton
+              to={docsPath("documentation/guides/cli#dev-command")}
+              variant="secondary/small"
+              LeadingIcon="docs"
+            >
+              Using the CLI
+            </LinkButton>
+            <LinkButton
+              to={docsPath("documentation/guides/deployment")}
+              variant="secondary/small"
+              LeadingIcon="docs"
+            >
+              Deploying your Jobs
+            </LinkButton>
+          </div>
+        </div>
+      </div>
     );
   }
 

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam/route.tsx

@@ -2,7 +2,7 @@ import { useRevalidator } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { Fragment, useEffect } from "react";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
-import { useEventSource } from "remix-utils/sse/react";
+import { useEventSource } from "~/hooks/useEventSource";
 import { BreadcrumbLink } from "~/components/navigation/Breadcrumb";
 import { BreadcrumbIcon } from "~/components/primitives/BreadcrumbIcon";
 import { RunOverview } from "~/components/run/RunOverview";

apps/webapp/app/routes/admin.api.v1.snapshot.ts

@@ -0,0 +1,59 @@
+import path from "path";
+import os from "os";
+import fs from "fs";
+import v8 from "v8";
+import { PassThrough } from "stream";
+import { json, type DataFunctionArgs } from "@remix-run/node";
+import { prisma } from "~/db.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { requireUser } from "~/services/session.server";
+
+// Format date as yyyy-MM-dd HH_mm_ss_SSS
+function formatDate(date: Date) {
+  const year = date.getFullYear();
+  const month = date.getMonth() + 1;
+  const day = date.getDate();
+  const hours = date.getHours();
+  const minutes = date.getMinutes();
+  const seconds = date.getSeconds();
+  const milliseconds = date.getMilliseconds();
+
+  return `${year}-${month.toString().padStart(2, "0")}-${day.toString().padStart(2, "0")} ${hours
+    .toString()
+    .padStart(2, "0")}_${minutes.toString().padStart(2, "0")}_${seconds
+    .toString()
+    .padStart(2, "0")}_${milliseconds.toString().padStart(3, "0")}`;
+}
+
+export async function loader({ request }: DataFunctionArgs) {
+  const user = await requireUser(request);
+
+  if (!user.admin) {
+    throw new Response("You must be an admin to perform this action", { status: 403 });
+  }
+
+  const host = request.headers.get("X-Forwarded-Host") ?? request.headers.get("host");
+
+  const tempDir = os.tmpdir();
+  const filepath = path.join(tempDir, `${host}-${formatDate(new Date())}.heapsnapshot`);
+
+  const snapshotPath = v8.writeHeapSnapshot(filepath);
+  if (!snapshotPath) {
+    throw new Response("No snapshot saved", { status: 500 });
+  }
+
+  const body = new PassThrough();
+  const stream = fs.createReadStream(snapshotPath);
+  stream.on("open", () => stream.pipe(body));
+  stream.on("error", (err) => body.end(err));
+  stream.on("end", () => body.end());
+
+  return new Response(body as any, {
+    status: 200,
+    headers: {
+      "Content-Type": "application/octet-stream",
+      "Content-Disposition": `attachment; filename="${path.basename(snapshotPath)}"`,
+      "Content-Length": (await fs.promises.stat(snapshotPath)).size.toString(),
+    },
+  });
+}

apps/webapp/app/routes/api.v1.event-dispatchers.ephemeral.ts

@@ -0,0 +1,59 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import {
+  EphemeralEventDispatcherRequestBodySchema,
+  InvokeJobRequestBodySchema,
+} from "@trigger.dev/core";
+import { z } from "zod";
+import { PrismaErrorSchema } from "~/db.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { CreateEphemeralEventDispatcherService } from "~/services/dispatchers/createEphemeralEventDispatcher.server";
+import { InvokeJobService } from "~/services/jobs/invokeJob.server";
+import { logger } from "~/services/logger.server";
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
+  }
+
+  // Now parse the request body
+  const anyBody = await request.json();
+
+  logger.debug("CreateEphemeralEventDispatcherService.call() request body", {
+    body: anyBody,
+  });
+
+  const body = EphemeralEventDispatcherRequestBodySchema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new CreateEphemeralEventDispatcherService();
+
+  try {
+    const dispatcher = await service.call(authenticationResult.environment, body.data);
+
+    if (!dispatcher) {
+      return json({ error: "Could not create Event Dispatcher" }, { status: 500 });
+    }
+
+    return json({ id: dispatcher.id });
+  } catch (error) {
+    const prismaError = PrismaErrorSchema.safeParse(error);
+    // Record not found in the database
+    if (prismaError.success && prismaError.data.code === "P2005") {
+      return json({ error: "Dispatcher not found" }, { status: 404 });
+    } else {
+      return json({ error: "Internal Server Error" }, { status: 500 });
+    }
+  }
+}

apps/webapp/app/routes/api.v1.events.bulk.ts

@@ -0,0 +1,49 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { SendBulkEventsBodySchema } from "@trigger.dev/core";
+import { generateErrorMessage } from "zod-error";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { IngestSendEvent } from "~/services/events/ingestSendEvent.server";
+import { eventRecordToApiJson } from "~/api.server";
+import { EventRecord } from "@trigger.dev/database";
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  // Now parse the request body
+  const anyBody = await request.json();
+
+  const body = SendBulkEventsBodySchema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ message: generateErrorMessage(body.error.issues) }, { status: 422 });
+  }
+
+  const service = new IngestSendEvent();
+
+  const events: EventRecord[] = [];
+
+  for (const event of body.data.events) {
+    const eventRecord = await service.call(authenticatedEnv, event, body.data.options);
+
+    if (!eventRecord) {
+      return json({ error: "Failed to create event during bulk ingest" }, { status: 500 });
+    }
+
+    events.push(eventRecord);
+  }
+
+  return json(events.map(eventRecordToApiJson));
+}

apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts

@@ -1,6 +1,5 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { TaskStatus } from "@trigger.dev/database";
 import {
   API_VERSIONS,
   RunTaskBodyOutput,
@@ -8,15 +7,16 @@ import {
   RunTaskResponseWithCachedTasksBody,
   ServerTask,
 } from "@trigger.dev/core";
+import { TaskStatus } from "@trigger.dev/database";
 import { z } from "zod";
 import { $transaction, PrismaClient, prisma } from "~/db.server";
+import { env } from "~/env.server";
 import { prepareTasksForCaching, taskWithAttemptsToServerTask } from "~/models/task.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { ulid } from "~/services/ulid.server";
-import { workerQueue } from "~/services/worker.server";
 import { generateSecret } from "~/services/sources/utils.server";
-import { env } from "~/env.server";
+import { ulid } from "~/services/ulid.server";
+import { taskOperationWorker, workerQueue } from "~/services/worker.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -302,7 +302,7 @@ export class RunTaskService {
 
       if (task.status === "RUNNING" && typeof taskBody.operation === "string") {
         // We need to schedule the operation
-        await workerQueue.enqueue(
+        await taskOperationWorker.enqueue(
           "performTaskOperation",
           {
             id: task.id,

apps/webapp/app/routes/resources.environments.$environmentParam.endpoint.stream.ts

@@ -2,7 +2,7 @@ import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { requireUserId } from "~/services/session.server";
-import { sse } from "~/utils/sse";
+import { sse } from "~/utils/sse.server";
 
 export async function loader({ request, params }: LoaderFunctionArgs) {
   await requireUserId(request);

apps/webapp/app/routes/resources.projects.$projectId.jobs.stream.ts

@@ -2,7 +2,7 @@ import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { requireUserId } from "~/services/session.server";
-import { sse } from "~/utils/sse";
+import { sse } from "~/utils/sse.server";
 
 export async function loader({ request, params }: LoaderFunctionArgs) {
   await requireUserId(request);

apps/webapp/app/routes/tests.sse.stream.ts

@@ -1,7 +1,7 @@
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { logger } from "~/services/logger.server";
-import { sse } from "~/utils/sse";
+import { sse } from "~/utils/sse.server";
 
 export async function loader({ request }: LoaderFunctionArgs) {
   const url = new URL(request.url);

apps/webapp/app/routes/tests.sse.tsx

@@ -1,6 +1,6 @@
 import { useLoaderData } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
-import { useEventSource } from "remix-utils/sse/react";
+import { useEventSource } from "~/hooks/useEventSource";
 import { z } from "zod";
 
 export async function loader({ request }: LoaderFunctionArgs) {

apps/webapp/app/services/dispatchers/createEphemeralEventDispatcher.server.ts

@@ -0,0 +1,68 @@
+import { EphemeralEventDispatcherRequestBody } from "@trigger.dev/core";
+import { $transaction, PrismaClient, prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { ExpireDispatcherService } from "./expireDispatcher.server";
+
+export class CreateEphemeralEventDispatcherService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    environment: AuthenticatedEnvironment,
+    data: EphemeralEventDispatcherRequestBody
+  ) {
+    return await $transaction(this.#prismaClient, async (tx) => {
+      const existingDispatcher = await tx.eventDispatcher.findUnique({
+        where: {
+          dispatchableId_environmentId: {
+            dispatchableId: data.url,
+            environmentId: environment.id,
+          },
+        },
+      });
+
+      if (existingDispatcher) {
+        return existingDispatcher;
+      }
+
+      const externalAccount = data.accountId
+        ? await this.#prismaClient.externalAccount.upsert({
+            where: {
+              environmentId_identifier: {
+                environmentId: environment.id,
+                identifier: data.accountId,
+              },
+            },
+            create: {
+              environmentId: environment.id,
+              organizationId: environment.organizationId,
+              identifier: data.accountId,
+            },
+            update: {},
+          })
+        : undefined;
+
+      const dispatcher = await tx.eventDispatcher.create({
+        data: {
+          dispatchableId: data.url,
+          environmentId: environment.id,
+          source: data.source ?? "trigger.dev",
+          payloadFilter: data.filter,
+          contextFilter: data.contextFilter,
+          dispatchable: { url: data.url, type: "EPHEMERAL" },
+          enabled: true,
+          event: typeof data.name === "string" ? [data.name] : data.name,
+          manual: false,
+          externalAccountId: externalAccount?.id,
+        },
+      });
+
+      await ExpireDispatcherService.enqueue(dispatcher.id, data.timeoutInSeconds, tx);
+
+      return dispatcher;
+    });
+  }
+}

apps/webapp/app/services/dispatchers/expireDispatcher.server.ts

@@ -0,0 +1,36 @@
+import { PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { workerQueue } from "../worker.server";
+
+export class ExpireDispatcherService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    await this.#prismaClient.eventDispatcher.delete({
+      where: {
+        id,
+      },
+    });
+  }
+
+  static async dequeue(id: string, tx?: PrismaClientOrTransaction) {
+    await workerQueue.dequeue(`expire:${id}`, { tx });
+  }
+
+  static async enqueue(id: string, timeoutInSeconds: number, tx?: PrismaClientOrTransaction) {
+    await workerQueue.enqueue(
+      "expireDispatcher",
+      {
+        id,
+      },
+      {
+        tx,
+        runAt: new Date(Date.now() + 1000 * timeoutInSeconds),
+        jobKey: `expire:${id}`,
+      }
+    );
+  }
+}

apps/webapp/app/services/dispatchers/invokeEphemeralEventDispatcher.server.ts

@@ -0,0 +1,108 @@
+import { PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { taskOperationWorker } from "../worker.server";
+import { EphemeralDispatchableSchema } from "~/models/eventDispatcher.server";
+import { fetch } from "@whatwg-node/fetch";
+import { ExpireDispatcherService } from "./expireDispatcher.server";
+
+export class InvokeEphemeralDispatcherService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string, eventRecordId: string) {
+    const eventDispatcher = await this.#prismaClient.eventDispatcher.findUnique({
+      where: {
+        id,
+      },
+    });
+
+    if (!eventDispatcher) {
+      return;
+    }
+
+    if (!eventDispatcher.enabled) {
+      return;
+    }
+
+    const eventRecord = await this.#prismaClient.eventRecord.findUnique({
+      where: {
+        id: eventRecordId,
+      },
+      include: {
+        externalAccount: true,
+      },
+    });
+
+    if (!eventRecord) {
+      return;
+    }
+
+    if (eventRecord.cancelledAt) {
+      return;
+    }
+
+    const dispatchable = EphemeralDispatchableSchema.safeParse(eventDispatcher.dispatchable);
+
+    if (!dispatchable.success) {
+      return;
+    }
+
+    const url = dispatchable.data.url;
+
+    const body = {
+      id: eventRecord.eventId,
+      source: eventRecord.source,
+      name: eventRecord.name,
+      payload: eventRecord.payload,
+      context: eventRecord.context,
+      timestamp: eventRecord.timestamp,
+      accountId: eventRecord.externalAccount ? eventRecord.externalAccount.identifier : undefined,
+    };
+
+    const abortController = new AbortController();
+
+    const timeoutId = setTimeout(() => {
+      abortController.abort();
+    }, 5000);
+
+    const response = await fetch(url, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json; charset=utf-8",
+      },
+      body: JSON.stringify(body),
+      signal: abortController.signal,
+    });
+
+    clearTimeout(timeoutId);
+
+    if (!response.ok) {
+      throw new Error(
+        `Failed to invoke ephemeral dispatcher: ${response.statusText} [${response.status}]`
+      );
+    }
+
+    // Run the expire dispatcher service
+    await ExpireDispatcherService.enqueue(id, 0);
+  }
+
+  static async dequeue(id: string, tx?: PrismaClientOrTransaction) {
+    await taskOperationWorker.dequeue(`invoke:ephemeral:${id}`, { tx });
+  }
+
+  static async enqueue(id: string, eventRecordId: string, tx?: PrismaClientOrTransaction) {
+    await taskOperationWorker.enqueue(
+      "invokeEphemeralDispatcher",
+      {
+        id,
+        eventRecordId,
+      },
+      {
+        tx,
+        jobKey: `invoke:ephemeral:${id}`,
+      }
+    );
+  }
+}

apps/webapp/app/services/events/deliverEvent.server.ts

@@ -96,6 +96,13 @@ export class DeliverEventService {
       return true;
     }
 
+    if (
+      dispatcher.externalAccountId &&
+      dispatcher.externalAccountId !== eventRecord.externalAccountId
+    ) {
+      return false;
+    }
+
     const payloadFilter = EventFilterSchema.safeParse(dispatcher.payloadFilter ?? {});
 
     const contextFilter = EventFilterSchema.safeParse(dispatcher.contextFilter ?? {});

apps/webapp/app/services/events/invokeDispatcher.server.ts

@@ -3,21 +3,8 @@ import type { PrismaClientOrTransaction } from "~/db.server";
 import { prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { CreateRunService } from "~/services/runs/createRun.server";
-
-const JobVersionDispatchableSchema = z.object({
-  type: z.literal("JOB_VERSION"),
-  id: z.string(),
-});
-
-const DynamicTriggerDispatchableSchema = z.object({
-  type: z.literal("DYNAMIC_TRIGGER"),
-  id: z.string(),
-});
-
-const DispatchableSchema = z.discriminatedUnion("type", [
-  JobVersionDispatchableSchema,
-  DynamicTriggerDispatchableSchema,
-]);
+import { InvokeEphemeralDispatcherService } from "../dispatchers/invokeEphemeralEventDispatcher.server";
+import { DispatchableSchema } from "~/models/eventDispatcher.server";
 
 export class InvokeDispatcherService {
   #prismaClient: PrismaClientOrTransaction;
@@ -142,6 +129,11 @@ export class InvokeDispatcherService {
           });
         }
 
+        break;
+      }
+      case "EPHEMERAL": {
+        await InvokeEphemeralDispatcherService.enqueue(eventDispatcher.id, eventRecord.id);
+
         break;
       }
     }

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

@@ -3,24 +3,16 @@ import { Logger } from "@trigger.dev/core";
 import { sensitiveDataReplacer } from "./sensitiveDataReplacer";
 import { singleton } from "~/utils/singleton";
 
-export const logger = singleton(
-  "logger",
-  () =>
-    new Logger(
-      "webapp",
-      (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
-      ["examples", "output", "connectionString", "payload"],
-      sensitiveDataReplacer
-    )
+export const logger = new Logger(
+  "webapp",
+  (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
+  ["examples", "output", "connectionString", "payload"],
+  sensitiveDataReplacer
 );
 
-export const workerLogger = singleton(
-  "worker-logger",
-  () =>
-    new Logger(
-      "worker",
-      (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
-      ["examples", "output", "connectionString"],
-      sensitiveDataReplacer
-    )
+export const workerLogger = new Logger(
+  "worker",
+  (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
+  ["examples", "output", "connectionString"],
+  sensitiveDataReplacer
 );

apps/webapp/app/services/tasks/performTaskOperation.server.ts

@@ -1,20 +1,26 @@
 import {
   FetchOperationSchema,
+  FetchPollOperationSchema,
   FetchRequestInit,
   FetchRetryOptions,
   FetchRetryStrategy,
   RedactString,
   RetryOptions,
+  calculateResetAt,
   calculateRetryAt,
+  eventFilterMatches,
+  responseFilterMatches,
 } from "@trigger.dev/core";
 import { type Task } from "@trigger.dev/database";
 import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
 import { formatUnknownError } from "~/utils/formatErrors.server";
 import { safeJsonFromResponse } from "~/utils/json";
 import { logger } from "../logger.server";
-import { workerQueue } from "../worker.server";
+import { taskOperationWorker, workerQueue } from "../worker.server";
 import { ResumeTaskService } from "./resumeTask.server";
 import { fetch } from "@whatwg-node/fetch";
+import { fromZodError } from "zod-validation-error";
+import { ulid } from "../ulid.server";
 
 type FoundTask = Awaited<ReturnType<typeof findTask>>;
 
@@ -32,16 +38,166 @@ export class PerformTaskOperationService {
       return;
     }
 
+    if (task.status === "CANCELED") {
+      return;
+    }
+
     if (task.status === "COMPLETED" || task.status === "ERRORED") {
       return await this.#resumeRunExecution(task, this.#prismaClient);
     }
 
     if (!task.operation) {
-      return await this.#resumeTask(task, null, 0);
+      return await this.#resumeTask(task, null, null, 200, "fetch", 0);
     }
 
     switch (task.operation) {
-      case "fetch": {
+      case "fetch-poll": {
+        const pollOperation = FetchPollOperationSchema.safeParse(task.params);
+
+        if (!pollOperation.success) {
+          return await this.#resumeTaskWithError(
+            task,
+            fromZodError(pollOperation.error, {
+              prefix: "Invalid fetch poll params",
+            }).message
+          );
+        }
+
+        const { url, requestInit, timeout, interval, responseFilter, requestTimeout } =
+          pollOperation.data;
+
+        // check if we need to fail the task because it's timed out
+        const startedAt = task.startedAt;
+
+        if (!startedAt) {
+          return await this.#resumeTaskWithError(task, {
+            message: "Task has not been started",
+          });
+        }
+
+        if (Date.now() - startedAt.getTime() > timeout * 1000) {
+          return await this.#resumeTaskWithError(task, {
+            message: `Task timed out after ${timeout} seconds`,
+          });
+        }
+
+        const startTimeInMs = performance.now();
+
+        const abortController = new AbortController();
+
+        // calculate the actual timeout. If timeoutInMs is undefined, we use the default of 5s
+        // Also make sure the timeout is at least 1s, but not bigger than 5s
+        const actualTimeoutInMs = Math.min(
+          Math.max(requestTimeout?.durationInMs ?? 5000, 1000),
+          5000
+        );
+
+        const timeoutId = setTimeout(() => {
+          abortController.abort();
+        }, actualTimeoutInMs);
+
+        try {
+          logger.debug("PerformTaskOperationService.call poll request", {
+            task,
+            actualTimeoutInMs,
+            url,
+            responseFilter,
+          });
+
+          const startedAt = new Date();
+
+          const method = requestInit?.method ?? "GET";
+
+          const response = await fetch(url, {
+            method,
+            headers: normalizeHeaders(requestInit?.headers ?? {}),
+            body: requestInit?.body,
+            signal: abortController.signal,
+          });
+
+          clearTimeout(timeoutId);
+
+          const durationInMs = Math.floor(performance.now() - startTimeInMs);
+
+          const headers = Object.fromEntries(response.headers.entries());
+
+          logger.debug("PerformTaskOperationService.call poll response", {
+            url,
+            requestInit,
+            statusCode: response.status,
+            headers: Object.fromEntries(response.headers.entries()),
+            durationInMs,
+          });
+
+          const matchResult = await responseFilterMatches(response, responseFilter);
+
+          await this.#prismaClient.task.create({
+            data: {
+              id: ulid(),
+              idempotencyKey: ulid(),
+              runId: task.runId,
+              parentId: task.id,
+              name: "poll attempt",
+              icon: "activity",
+              status: "COMPLETED",
+              noop: true,
+              style: { style: "minimal", variant: "info" },
+              description: `${method} ${url} ${response.status}`,
+              params: {
+                status: response.status,
+                headers,
+                body: matchResult.body as any,
+              },
+              startedAt,
+              completedAt: new Date(),
+            },
+          });
+
+          if (matchResult.match) {
+            logger.debug("PerformTaskOperationService.call poll response matched", {
+              url,
+              matchResult,
+            });
+
+            return await this.#resumeTask(
+              task,
+              matchResult.body,
+              Object.fromEntries(response.headers.entries()),
+              response.status,
+              "fetch",
+              durationInMs
+            );
+          } else {
+            const retryAt = new Date(Date.now() + interval * 1000);
+
+            return await this.#retryTask(task, retryAt);
+          }
+        } catch (error) {
+          if (error instanceof Error && error.name === "AbortError") {
+            const durationInMs = Math.floor(performance.now() - startTimeInMs);
+
+            logger.debug("PerformTaskOperationService.call poll timed out", {
+              url,
+              durationInMs,
+              error,
+            });
+
+            const retryAt = this.#calculateRetryForTimeout(task, requestTimeout?.retry);
+
+            if (retryAt) {
+              return await this.#retryTask(task, retryAt);
+            }
+
+            return await this.#resumeTaskWithError(task, {
+              message: `Fetch timed out after ${actualTimeoutInMs.toFixed(0)}ms`,
+            });
+          }
+
+          throw error;
+        }
+      }
+      case "fetch":
+      case "fetch-response": {
         const fetchOperation = FetchOperationSchema.safeParse(task.params);
 
         if (!fetchOperation.success) {
@@ -97,7 +253,7 @@ export class PerformTaskOperationService {
           });
 
           if (!response.ok) {
-            const retryAt = this.#calculateRetryForResponse(task, retry, response);
+            const retryAt = this.#calculateRetryForResponse(task, retry, response, jsonBody);
 
             if (retryAt) {
               return await this.#retryTaskWithError(
@@ -117,7 +273,14 @@ export class PerformTaskOperationService {
             }
           }
 
-          return await this.#resumeTask(task, jsonBody, durationInMs);
+          return await this.#resumeTask(
+            task,
+            jsonBody,
+            Object.fromEntries(response.headers.entries()),
+            response.status,
+            task.operation,
+            durationInMs
+          );
         } catch (error) {
           if (error instanceof Error && error.name === "AbortError") {
             const durationInMs = Math.floor(performance.now() - startTimeInMs);
@@ -157,13 +320,14 @@ export class PerformTaskOperationService {
   #calculateRetryForResponse(
     task: NonNullable<FoundTask>,
     retry: FetchRetryOptions | undefined,
-    response: Response
+    response: Response,
+    body: any
   ): Date | undefined {
     if (!retry) {
       return;
     }
 
-    const strategy = this.#getRetryStrategyForStatusCode(response.status, retry);
+    const strategy = this.#getRetryStrategyForResponse(response, body, retry);
 
     if (!strategy) {
       return;
@@ -180,11 +344,10 @@ export class PerformTaskOperationService {
         return calculateRetryAt(strategy, task.attempts.length - 1);
       }
       case "headers": {
-        const remaining = response.headers.get(strategy.remainingHeader);
         const resetAt = response.headers.get(strategy.resetHeader);
 
-        if (typeof remaining === "string" && typeof resetAt === "string" && remaining === "0") {
-          return new Date(Number(resetAt) * 1000 + addJitterInMs());
+        if (typeof resetAt === "string") {
+          return calculateResetAt(resetAt, strategy.resetFormat);
         }
       }
     }
@@ -201,8 +364,9 @@ export class PerformTaskOperationService {
     return calculateRetryAt(retry, task.attempts.length - 1);
   }
 
-  #getRetryStrategyForStatusCode(
-    statusCode: number,
+  #getRetryStrategyForResponse(
+    response: Response,
+    body: any,
     retry: FetchRetryOptions
   ): FetchRetryStrategy | undefined {
     const statusCodes = Object.keys(retry);
@@ -211,7 +375,19 @@ export class PerformTaskOperationService {
       const statusRange = statusCodes[i];
       const strategy = retry[statusRange];
 
-      if (isStatusCodeInRange(statusCode, statusRange)) {
+      if (isStatusCodeInRange(response.status, statusRange)) {
+        if (strategy.bodyFilter) {
+          if (!body) {
+            continue;
+          }
+
+          if (eventFilterMatches(body, strategy.bodyFilter)) {
+            return strategy;
+          } else {
+            continue;
+          }
+        }
+
         return strategy;
       }
     }
@@ -248,16 +424,26 @@ export class PerformTaskOperationService {
         },
       });
 
-      await workerQueue.enqueue(
+      await taskOperationWorker.enqueue(
         "performTaskOperation",
         {
           id: task.id,
         },
-        { tx, runAt: retryAt }
+        { tx, runAt: retryAt, jobKey: `operation:${task.id}` }
       );
     });
   }
 
+  async #retryTask(task: Task, retryAt: Date) {
+    await taskOperationWorker.enqueue(
+      "performTaskOperation",
+      {
+        id: task.id,
+      },
+      { runAt: retryAt, jobKey: `operation:${task.id}` }
+    );
+  }
+
   async #resumeTaskWithError(task: NonNullable<FoundTask>, output: any) {
     await $transaction(this.#prismaClient, async (tx) => {
       await tx.task.update({
@@ -284,7 +470,14 @@ export class PerformTaskOperationService {
     });
   }
 
-  async #resumeTask(task: NonNullable<FoundTask>, output: any, durationInMs: number) {
+  async #resumeTask(
+    task: NonNullable<FoundTask>,
+    output: any,
+    context: any,
+    status: number,
+    operation: "fetch" | "fetch-response",
+    durationInMs: number
+  ) {
     await $transaction(this.#prismaClient, async (tx) => {
       await tx.taskAttempt.updateMany({
         where: {
@@ -296,12 +489,22 @@ export class PerformTaskOperationService {
         },
       });
 
+      const taskOutput =
+        operation === "fetch"
+          ? output
+          : {
+              data: output,
+              headers: context,
+              status,
+            };
+
       await tx.task.update({
         where: { id: task.id },
         data: {
           status: "COMPLETED",
           completedAt: new Date(),
-          output: output ? output : undefined,
+          output: taskOutput,
+          context: context ? context : undefined,
           run: {
             update: {
               executionDuration: {

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

@@ -24,6 +24,8 @@ import { ProbeEndpointService } from "./endpoints/probeEndpoint.server";
 import { DeliverRunSubscriptionService } from "./runs/deliverRunSubscription.server";
 import { DeliverRunSubscriptionsService } from "./runs/deliverRunSubscriptions.server";
 import { ResumeTaskService } from "./tasks/resumeTask.server";
+import { ExpireDispatcherService } from "./dispatchers/expireDispatcher.server";
+import { InvokeEphemeralDispatcherService } from "./dispatchers/invokeEphemeralEventDispatcher.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -40,9 +42,6 @@ const workerCatalog = {
   processCallbackTimeout: z.object({
     id: z.string(),
   }),
-  performTaskOperation: z.object({
-    id: z.string(),
-  }),
   deliverHttpSourceRequest: z.object({ id: z.string() }),
   refreshOAuthToken: z.object({
     organizationId: z.string(),
@@ -93,6 +92,9 @@ const workerCatalog = {
   resumeTask: z.object({
     id: z.string(),
   }),
+  expireDispatcher: z.object({
+    id: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -108,12 +110,24 @@ const executionWorkerCatalog = {
   }),
 };
 
+const taskOperationWorkerCatalog = {
+  performTaskOperation: z.object({
+    id: z.string(),
+  }),
+  invokeEphemeralDispatcher: z.object({
+    id: z.string(),
+    eventRecordId: z.string(),
+  }),
+};
+
 let workerQueue: ZodWorker<typeof workerCatalog>;
 let executionWorker: ZodWorker<typeof executionWorkerCatalog>;
+let taskOperationWorker: ZodWorker<typeof taskOperationWorkerCatalog>;
 
 declare global {
   var __worker__: ZodWorker<typeof workerCatalog>;
   var __executionWorker__: ZodWorker<typeof executionWorkerCatalog>;
+  var __taskOperationWorker__: ZodWorker<typeof taskOperationWorkerCatalog>;
 }
 
 // this is needed because in development we don't want to restart
@@ -123,6 +137,7 @@ declare global {
 if (env.NODE_ENV === "production") {
   workerQueue = getWorkerQueue();
   executionWorker = getExecutionWorkerQueue();
+  taskOperationWorker = getTaskOperationWorkerQueue();
 } else {
   if (!global.__worker__) {
     global.__worker__ = getWorkerQueue();
@@ -134,6 +149,12 @@ if (env.NODE_ENV === "production") {
   }
 
   executionWorker = global.__executionWorker__;
+
+  if (!global.__taskOperationWorker__) {
+    global.__taskOperationWorker__ = getTaskOperationWorkerQueue();
+  }
+
+  taskOperationWorker = global.__taskOperationWorker__;
 }
 
 export async function init() {
@@ -148,6 +169,10 @@ export async function init() {
   if (env.EXECUTION_WORKER_ENABLED === "true") {
     await executionWorker.initialize();
   }
+
+  if (env.TASK_OPERATION_WORKER_ENABLED === "true") {
+    await taskOperationWorker.initialize();
+  }
 }
 
 function getWorkerQueue() {
@@ -286,15 +311,6 @@ function getWorkerQueue() {
           await service.call(payload.id);
         },
       },
-      performTaskOperation: {
-        priority: 0, // smaller number = higher priority
-        maxAttempts: 3,
-        handler: async (payload, job) => {
-          const service = new PerformTaskOperationService();
-
-          await service.call(payload.id);
-        },
-      },
       scheduleEmail: {
         priority: 100,
         maxAttempts: 3,
@@ -375,6 +391,15 @@ function getWorkerQueue() {
         handler: async (payload, job) => {
           const service = new ResumeTaskService();
 
+          return await service.call(payload.id);
+        },
+      },
+      expireDispatcher: {
+        priority: 10,
+        maxAttempts: 3,
+        handler: async (payload) => {
+          const service = new ExpireDispatcherService();
+
           return await service.call(payload.id);
         },
       },
@@ -428,4 +453,41 @@ function getExecutionWorkerQueue() {
   });
 }
 
-export { executionWorker, workerQueue };
+function getTaskOperationWorkerQueue() {
+  return new ZodWorker({
+    name: "taskOperationWorker",
+    prisma,
+    runnerOptions: {
+      connectionString: env.DATABASE_URL,
+      concurrency: env.TASK_OPERATION_WORKER_CONCURRENCY,
+      pollInterval: env.TASK_OPERATION_WORKER_POLL_INTERVAL,
+      noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
+      schema: env.WORKER_SCHEMA,
+      maxPoolSize: env.TASK_OPERATION_WORKER_CONCURRENCY,
+    },
+    shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
+    schema: taskOperationWorkerCatalog,
+    tasks: {
+      performTaskOperation: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 3,
+        handler: async (payload, job) => {
+          const service = new PerformTaskOperationService();
+
+          await service.call(payload.id);
+        },
+      },
+      invokeEphemeralDispatcher: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 10,
+        handler: async (payload, job) => {
+          const service = new InvokeEphemeralDispatcherService();
+
+          await service.call(payload.id, payload.eventRecordId);
+        },
+      },
+    },
+  });
+}
+
+export { executionWorker, workerQueue, taskOperationWorker };

apps/webapp/app/utils/sse.server.ts

@@ -1,4 +1,5 @@
 import { eventStream } from "remix-utils/sse/server";
+import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 
 type SseProps = {
@@ -17,20 +18,11 @@ type Event = {
 };
 
 export function sse({ request, pingInterval = 1000, updateInterval = 348, run }: SseProps) {
-  let pinger: NodeJS.Timer | undefined = undefined;
-  let updater: NodeJS.Timer | undefined = undefined;
+  if (env.DISABLE_SSE === "1" || env.DISABLE_SSE === "true") {
+    return new Response("SSE disabled", { status: 200 });
+  }
 
-  const abort = () => {
-    if (pinger) {
-      clearInterval(pinger);
-    }
-
-    if (updater) {
-      clearInterval(updater);
-    }
-  };
-
-  return eventStream(request.signal, (send) => {
+  return eventStream(request.signal, (send, close) => {
     const safeSend = (args: { event?: string; data: string }) => {
       try {
         send(args);
@@ -53,18 +45,34 @@ export function sse({ request, pingInterval = 1000, updateInterval = 348, run }:
           });
         }
 
-        abort();
+        close();
       }
     };
 
-    pinger = setInterval(() => {
+    const pinger = setInterval(() => {
+      if (request.signal.aborted) {
+        return close();
+      }
+
       safeSend({ event: "ping", data: new Date().toISOString() });
     }, pingInterval);
 
-    updater = setInterval(async () => {
-      run(safeSend, abort);
+    const updater = setInterval(() => {
+      if (request.signal.aborted) {
+        return close();
+      }
+
+      run(safeSend, close);
     }, updateInterval);
 
-    return abort;
+    const timeout = setTimeout(() => {
+      close();
+    }, 60 * 1000); // 1 minute
+
+    return () => {
+      clearInterval(updater);
+      clearInterval(pinger);
+      clearTimeout(timeout);
+    };
   });
 }

apps/webapp/package.json

@@ -102,7 +102,7 @@
     "react-use": "^17.4.0",
     "recharts": "^2.8.0",
     "remix-auth": "^3.6.0",
-    "remix-auth-email-link": "^2.1.0",
+    "remix-auth-email-link": "2.0.2",
     "remix-auth-github": "^1.6.0",
     "remix-typedjson": "0.3.1",
     "remix-utils": "^7.1.0",

apps/webapp/server.ts

@@ -20,7 +20,9 @@ app.use((req, res, next) => {
   next();
 });
 
-app.use(compression());
+if (process.env.DISABLE_COMPRESSION !== "1") {
+  app.use(compression());
+}
 
 // http://expressjs.com/en/advanced/best-practice-security.html#at-a-minimum-disable-x-powered-by-header
 app.disable("x-powered-by");

docs/_snippets/send-events-params.mdx

@@ -0,0 +1,52 @@
+<ParamField body="events" type="array" required>
+  <Expandable title="event properties" defaultOpen>
+    <ParamField body="name" type="string" required>
+      The `name` property must exactly match any subscriptions you want to
+      trigger.
+    </ParamField>
+    <ParamField body="payload" type="any">
+      The `payload` property will be sent to any matching Jobs and will appear
+      as the `payload` param of the `run()` function. You can leave this
+      parameter out if you just want to trigger a Job without any input data.
+    </ParamField>
+    <ParamField body="context" type="any">
+      The optional `context` property will be sent to any matching Jobs and will
+      be passed through as the `context.event.context` param of the `run()`
+      function. This is optional but can be useful if you want to pass through
+      some additional context to the Job.
+    </ParamField>
+    <ParamField body="id" type="string">
+      The `id` property uniquely identify this particular event. If unset it
+      will be set automatically using `ulid`.
+    </ParamField>
+    <ParamField body="timestamp" type="Date">
+      This is optional, it defaults to the current timestamp. Usually you would
+      only set this if you have a timestamp that you wish to pass through, e.g.
+      you receive a timestamp from a service and you want the same timestamp to
+      be used in your Job.
+    </ParamField>
+    <ParamField body="source" type="string">
+      This is optional, it defaults to "trigger.dev". It can be useful to set
+      this as you can filter events using this in the `eventTrigger()`.
+    </ParamField>
+  </Expandable>
+</ParamField>
+
+<ParamField body="options" type="object">
+  <Expandable title="properties" defaultOpen>
+    <ParamField body="deliverAt" type="Date">
+      An optional Date when you want the event to Trigger Jobs. The event will
+      be sent to the platform immediately but won't be acted upon until the
+      specified time.
+    </ParamField>
+    <ParamField body="deliverAfter" type="number">
+      An optional number of seconds you want to wait for the event to Trigger
+      any relevant Jobs. The event will be sent to the platform immediately but
+      won't be acted upon until the specified time.
+    </ParamField>
+    <ParamField body="accountId" type="string">
+      This optional param will be used by the Trigger.dev Connect feature, which
+      is coming soon.
+    </ParamField>
+  </Expandable>
+</ParamField>

docs/_snippets/send-events-return.mdx

@@ -0,0 +1,29 @@
+<ResponseField name="events" type="array">
+  <Expandable title="properties" defaultOpen>
+    <ResponseField name="id" type="string" required>
+      The `id` of the event that was sent.
+    </ResponseField>
+    <ResponseField name="name" type="string" required>
+      The `name` of the event that was sent.
+    </ResponseField>
+    <ResponseField name="payload" type="any" required>
+      The `payload` of the event that was sent
+    </ResponseField>
+    <ResponseField name="timestamp" type="Date" required>
+      The `timestamp` of the event that was sent
+    </ResponseField>
+    <ResponseField name="context" type="any">
+      The `context` of the event that was sent. Is `undefined` if no context was
+      set when sending the event.
+    </ResponseField>
+    <ResponseField name="deliverAt" type="Date">
+      The timestamp when the event will be delivered to any matching Jobs. Is
+      `undefined` if `deliverAt` or `deliverAfter` wasn't set when sending the
+      event.
+    </ResponseField>
+    <ResponseField name="deliveredAt" type="Date">
+      The timestamp when the event was delivered. Is `undefined` if `deliverAt`
+      or `deliverAfter` were set when sending the event.
+    </ResponseField>
+  </Expandable>
+</ResponseField>

docs/documentation/concepts/http-endpoints.mdx

@@ -3,13 +3,11 @@ title: HTTP endpoints
 description: HTTP endpoints allow you to trigger your Jobs from any webhooks.
 ---
 
-<Warning>This feature is in beta and not yet available to use on the Trigger.dev Cloud.</Warning>
-
 Sometimes you want to subscribe to changes from an API, and we don't have [an Integration](/integrations/introduction) for it yet. That's when you can use `defineHttpEndpoint` to receive webhooks, verify them, and create an [HTTP Trigger](/documentation/concepts/triggers/http).
 
 ## Defining an HTTP endpoint
 
-Defining an HTTP endpoint creates a URL and secret which you'll enter into Cal.com's website. It also attaches a `verify` function that is called when a webhook is received. It's compulsory to return a result from this function – 90% of the time you can use our `verifyRequestSignature` helper function.
+We'll use Cal.com as an example:
 
 ```ts
 const caldotcom = client.defineHttpEndpoint({
@@ -35,6 +33,10 @@ const caldotcom = client.defineHttpEndpoint({
 });
 ```
 
+When this code runs (and you're running the CLI dev command) the HTTP endpoint will be created and be visible in the Trigger.dev dashboard.
+
+The `verify` function is compulsory and is automatically called when a webhook is received. It's required to return a result from this function – 90% of the time you can use our `verifyRequestSignature` helper function.
+
 ## Getting the URL and secret
 
 In our dashboard, you can navigate to the HTTP endpoints page. From there you can select your endpoint and copy the URL (1) and secret (2) for the appropriate Environment.

docs/documentation/concepts/triggers/http.mdx

@@ -4,8 +4,6 @@ sidebarTitle: "HTTP"
 description: "HTTP Triggers allow you to trigger your Jobs from any webhooks."
 ---
 
-<Warning>This feature is in beta and not yet available to use on the Trigger.dev Cloud.</Warning>
-
 Sometimes you want to subscribe to changes from an API, and we don't have [an Integration](/integrations/introduction) for it yet. That's when you can use `defineHttpEndpoint` to receive webhooks, verify them, and create an HTTP Trigger.
 
 You should read the [HTTP endpoint](/documentation/concepts/http-endpoints) documentation to understand how to create an HTTP endpoint.

docs/documentation/concepts/triggers/invoke.mdx

@@ -4,8 +4,6 @@ sidebarTitle: "Manual Invoke"
 description: "Invoke Jobs manually using the invoke Trigger"
 ---
 
-<Warning>This feature is in beta and not yet available to use on the Trigger.dev Cloud.</Warning>
-
 Sometimes it makes sense to be able to invoke a Job manually, without having to specify an event, especially for cases where you want to get notified when the invoked Job Run is complete.
 
 To specify that a job is manually invokable, you can use the `invokeTrigger()` function when defining a job:

docs/documentation/guides/task-library.mdx

@@ -0,0 +1,255 @@
+---
+title: "Task Library"
+description: "These are the built-in tasks that are available to use in your Jobs."
+---
+
+Welcome to the Trigger.dev Task Library 📚. You may be wondering, what is a Task and why are there a library of them? Well you see, Trigger.dev works by divvying up a long-running job execution into a bunch of little tasks, each one taking less time then a single serverless function execution. 💫
+
+You can define and run your own tasks easily using [io.runTask()](/sdk/io/runtask), or you can use one of our [Integrations](/integrations/introduction) which are tasks for specific APIs, like OpenAI or Stripe.
+
+<Note>Read more about how Tasks work [here](/documentation/concepts/tasks).</Note>
+
+We also have a growing library of built-in tasks that you can use in your Jobs through the `io` object. These tasks are designed to be generic and reusable, and are a great way to get started with Trigger.dev.
+
+<Info>
+  You may notice that I'm using emojis for all the cache keys below, which is totally 💯% fine as
+  long as they are unique inside a run. Read more about how cache keys work
+  [here](/documentation/concepts/tasks#task-cache-keys)
+</Info>
+
+## `wait`
+
+This task allows you to resume executing your job after a certain amount of time has passed:
+
+```ts
+await io.wait("⏰", 60); // wait 60 seconds
+```
+
+Internally this task is considered a "noop", and noop tasks have no output.
+
+[reference docs](/sdk/io/wait)
+
+## `waitForRequest`
+
+You supply this task with a callback to receive a URL. When a POST request is made to that URL, the JSON body of the request becomes the task output.
+
+The example below uses `waitForRequest` to capture a Screenshot of a website using [ScreenshotOne.com](https://screenshotone.com/) and passes the callback URL to the webhook URL to get notified when the screenshot is finished:
+
+```ts
+const result = await io.waitForRequest<ScreenshotResponse>(
+  "📸",
+  async (url) => {
+    await fetch(`https://api.screenshotone.com/take`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        webhook_url: url, // this is the URL that will be called when the screenshot is ready
+        access_key: "my-access-key",
+        url: "https://trigger.dev",
+        store: "true",
+        storage_path: "my-screeshots",
+        response_type: "json",
+        async: "true",
+        storage_return_location: "true",
+      }),
+    });
+  },
+  {
+    timeoutInSeconds: 300, // wait up to 5 minutes for the screenshot to be ready
+  }
+);
+```
+
+We actually originally built this task for our [Replicate integration](/integrations/apis/replicate), which accepts a callback URL to notify you when a prediction is ready. So this allows you to write very succinct code to create a prediction and wait for it's results:
+
+```ts
+const sdPrediction = await io.replicate.predictions.createAndAwait("🧑‍🎨", {
+  version: "ac732df83cea7fff18b8472768c88ad041fa750ff7682a21affe81863cbe77e4",
+  input: {
+    prompt: "What is the meaning of life?",
+  },
+});
+```
+
+[reference docs](/sdk/io/wait-for-request)
+
+## `waitForEvent`
+
+This task allows you to wait for an event to be sent. To read about how events work, check out the [Events](/documentation/concepts/triggers/events) documentation.
+
+```ts
+const event = await io.waitForEvent(
+  "🥂",
+  {
+    name: "user.created",
+    schema: z.object({
+      id: z.string(),
+      createdAt: z.coerce.date(),
+      isAdmin: z.boolean(),
+    }),
+    filter: {
+      isAdmin: [true], // Only wait for events where isAdmin is true
+    },
+  },
+  {
+    timeoutInSeconds: 60 * 60, // Wait for up to an hour
+  }
+);
+```
+
+The event object returned from this task is the full event object that was sent, including `id`, `name`, `payload`, `context`, and more.
+
+[reference docs](/sdk/io/wait-for-event)
+
+## `backgroundFetch`
+
+This task allows you to perform a `fetch` request in the background, and then resume the execution of your job after the request has completed.
+
+```ts
+const body = io.backgroundFetch<MyResponseData>("🕸️", "https://example.com/api/endpoint", {
+  method: "POST",
+  headers: {
+    "Content-Type": "application/json",
+    Authorization: redactString`Bearer ${auth.apiKey}`,
+  },
+  body: JSON.stringify({ foo: "bar" }),
+});
+```
+
+This is useful for when an API is slow to respond and might not finish before your serverless function times out. We created this task to power our [OpenAI integration](/integrations/apis/openai), which can sometimes take more than a minute to respond:
+
+```ts
+// This uses backgroundFetch under the hood
+await io.openai.chat.completions.backgroundCreate("💬", {
+  model: "gpt-3.5-turbo",
+  messages: [
+    {
+      role: "user",
+      content: "Create a good programming joke about background jobs",
+    },
+  ],
+});
+```
+
+[reference docs](/sdk/io/backgroundfetch)
+
+## `backgroundPoll`
+
+This task is similar to `backgroundFetch`, but instead of waiting for a single request to complete, it will poll a URL until it returns a certain value.
+
+```ts
+const result = await io.backgroundPoll<{ foo: string }>("🔃", {
+  url: "https://example.com/api/endpoint",
+  interval: 10, // every 10 seconds
+  timeout: 300, // stop polling after 5 minutes
+  responseFilter: {
+    // stop polling once this filter matches
+    status: [200],
+    body: {
+      status: ["SUCCESS"],
+    },
+  },
+});
+```
+
+## `logger`
+
+The logger object allows you to log messages to the Trigger.dev console. This is useful for debugging your jobs, or just to see what's going on inside your job.
+
+```ts
+await io.logger.info("This is an info message");
+```
+
+You can optionally pass a `context` object to the logger, which will be displayed in the console:
+
+```ts
+await io.logger.info("This is an info message", {
+  foo: "bar",
+});
+```
+
+We support the following log levels:
+
+- `io.logger.debug()`
+- `io.logger.info()`
+- `io.logger.warn()`
+- `io.logger.error()`
+
+<Note>
+  You may notice these tasks don't include cache keys. We automatically create a cache key for you
+  based on the message and the log-level
+</Note>
+
+[reference docs](/sdk/io/logger)
+
+## `random`
+
+Use this task to generate a random number that stays stable during run retries/resumes:
+
+```ts
+const randomNumber = await io.random("🎲", {
+  min: 1,
+  max: 100,
+});
+```
+
+[reference docs](/sdk/io/random)
+
+## `sendEvent`
+
+This task allows you to send an event from inside your job run.
+
+If you want to send an event from outside a run (e.g. just from your backend) you should use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent) instead.
+
+```ts
+await io.sendEvent("🚚", {
+  id: "e_1234567890",
+  name: "new.user",
+  payload: {
+    userId: "u_1234567890",
+  },
+});
+```
+
+[reference docs](/sdk/io/sendevent)
+
+## `getEvent`
+
+This task allows you to get an event by ID from inside your job run.
+
+If you want to get an event from outside a run (e.g. just from your backend) you should use [client.getEvent()](/sdk/triggerclient/instancemethods/getevent) instead.
+
+```ts
+const event = await io.getEvent("📥", "e_1234567890");
+```
+
+[reference docs](/sdk/io/getevent)
+
+## `cancelEvent`
+
+If you send an event that has a delivery date in the future, you can use this task to cancel it.
+
+```ts
+await io.sendEvent(
+  "🚚",
+  {
+    id: "e_1234567890",
+    name: "new.user",
+    payload: {
+      userId: "u_1234567890",
+    },
+  },
+  {
+    deliverAt: new Date(Date.now() + 1000 * 60 * 60 * 24), // deliver in 24 hours
+  }
+);
+
+// Later on, if you want to cancel the event:
+await io.cancelEvent("🚫", "e_1234567890");
+```
+
+## `createStatus`
+
+Coming soon

docs/integrations/apis/openai-tasks.mdx

@@ -1,643 +0,0 @@
----
-title: OpenAI tasks
-sidebarTitle: Tasks
----
-
-Tasks are executed after the job is triggered and are the main building blocks of a job. You can string together as many tasks as you want.
-
----
-
-## All tasks
-
-### `createCompletion`
-
-Generates text completions as per given prompt. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/chat)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code demonstrates using OpenAI's text completion with the "davinci" model.
-    // It generates text based on the given prompt.
-    await io.openai.createCompletion("completion", {
-      model: "davinci",
-      prompt: "Once upon a time",
-    });
-},
-```
-
-### `backgroundCreateCompletion`
-
-Generates text completions in the background. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/chat)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background text completion using the "gpt-3.5-turbo" model.
-    // It generates text based on the provided programming task and logs the result.
-    const programmingTask = `Create a function that checks if a string is a palindrome.`;
-
-    const response = await io.openai.backgroundCreateCompletion("background-completion", {
-      model: "gpt-3.5-turbo",
-      prompt: `Coding task: ${programmingTask}\n\n`,
-    });
-
-    await io.logger.info("codeSnippet", response.choices[0]?.text);
-},
-```
-
-You can also pass an optional third parameter to `backgroundCreateCompletion` to specify OpenAI request options:
-
-```ts requestOptions.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background text completion using the "gpt-3.5-turbo" model.
-    // It generates text based on the provided programming task and logs the result.
-    const programmingTask = `Create a function that checks if a string is a palindrome.`;
-
-    const response = await io.openai.backgroundCreateCompletion("background-completion", {
-      model: "gpt-3.5-turbo",
-      prompt: `Coding task: ${programmingTask}\n\n`,
-    }, {
-      headers: {
-        "User-Agent": "my-user-agent"
-      }
-    });
-
-    await io.logger.info("codeSnippet", response.choices[0]?.text);
-},
-```
-
-This task is implemented using [io.backgroundFetch()](/sdk/io/backgroundfetch) and so you can also pass a 4th parameter customizing the retry and timeout options:
-
-```ts fetchOptions.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background text completion using the "gpt-3.5-turbo" model.
-    // It generates text based on the provided programming task and logs the result.
-    const programmingTask = `Create a function that checks if a string is a palindrome.`;
-
-    const response = await io.openai.backgroundCreateCompletion("background-completion", {
-      model: "gpt-3.5-turbo",
-      prompt: `Coding task: ${programmingTask}\n\n`,
-    }, {
-      headers: {
-        "User-Agent": "my-user-agent"
-      }
-    }, {
-      timeout: {
-        durationInMs: 10000,
-        retry: {
-          limit: 3,
-          minTimeoutInMs: 1000,
-          factor: 2,
-        }
-      }
-    });
-
-    await io.logger.info("codeSnippet", response.choices[0]?.text);
-},
-```
-
-### `createChatCompletion`
-
-Generates text completions in a conversational context. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/chat/create)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code demonstrates chat completion with the "gpt-3.5-turbo" model.
-    // It simulates a conversation by providing messages and receiving a chat response.
-    await io.openai.createChatCompletion("chat-completion", {
-      model: "gpt-3.5-turbo",
-      messages: [
-        {
-          role: "user",
-          content: "Create a good programming joke about background jobs",
-        },
-      ],
-    });
-},
-```
-
-### `backgroundCreateChatCompletion`
-
-Generates text completions in a conversational context in the background. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/chat/object)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background chat completion using the "gpt-3.5-turbo" model.
-    // It simulates a conversation with a user message and logs the response choices.
-    const response = await io.openai.backgroundCreateChatCompletion("background-chat-completion", {
-      model: "gpt-3.5-turbo",
-      messages: [
-        {
-          role: "user",
-          content: "Create a good programming joke about background jobs",
-        },
-      ],
-    });
-
-    await io.logger.info("choices", response.choices);
-},
-```
-
-You can also use the more "fluent" pattern used by the OpenAI SDK:
-
-```ts fluent.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background chat completion using the "gpt-3.5-turbo" model.
-    // It simulates a conversation with a user message and logs the response choices.
-    const response = await io.openai.chat.completions.backgroundCreate("background-chat-completion", {
-      model: "gpt-3.5-turbo",
-      messages: [
-        {
-          role: "user",
-          content: "Create a good programming joke about background jobs",
-        },
-      ],
-    });
-
-    await io.logger.info("choices", response.choices);
-},
-```
-
-Additionally, you can pass an optional third parameter to specify OpenAI request options:
-
-```ts requestOptions.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background text completion using the "gpt-3.5-turbo" model.
-    // It generates text based on the provided programming task and logs the result.
-    const programmingTask = `Create a function that checks if a string is a palindrome.`;
-
-    const response = await io.openai.chat.completions.backgroundCreate("background-completion", {
-      model: "gpt-3.5-turbo",
-      messages: [
-        {
-          role: "user",
-          content: "Create a good programming joke about background jobs",
-        },
-      ],
-    }, {
-      headers: {
-        "User-Agent": "my-user-agent"
-      }
-    });
-
-    await io.logger.info("choices", response.choices);
-},
-```
-
-This task is implemented using [io.backgroundFetch()](/sdk/io/backgroundfetch) and so you can also pass a 4th parameter customizing the retry and timeout options:
-
-```ts fetchOptions.ts
-run: async (payload, io, ctx) => {
-    // This code showcases background text completion using the "gpt-3.5-turbo" model.
-    // It generates text based on the provided programming task and logs the result.
-    const programmingTask = `Create a function that checks if a string is a palindrome.`;
-
-    const response = await io.openai.chat.completions.backgroundCreate("background-completion", {
-      model: "gpt-3.5-turbo",
-      messages: [
-        {
-          role: "user",
-          content: "Create a good programming joke about background jobs",
-        },
-      ],
-    }, {
-      headers: {
-        "User-Agent": "my-user-agent"
-      }
-    }, {
-      timeout: {
-        durationInMs: 10000,
-        retry: {
-          limit: 3,
-          minTimeoutInMs: 1000,
-          factor: 2,
-        }
-      }
-    });
-
-    await io.logger.info("choices", response.choices);
-},
-```
-
-### `retrieveModel`
-
-Retrieves a specific model by ID. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/models/retrieve)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // In this code snippet, we retrieve detailed information about a specific OpenAI model.
-
-    // Specify the ID of the model you want to retrieve. Replace 'your_model_id' with the actual model ID.
-    const modelIdToRetrieve = "your_model_id";
-
-    try {
-        // Retrieve the model information using the OpenAI API
-        const retrievedModel = await io.openai.retrieveModel("get-model", {
-            model: modelIdToRetrieve,
-        });
-
-        // Log the detailed model information
-        await io.logger.info("retrievedModel", retrievedModel);
-    } catch (error) {
-        // Handle errors, such as if the model with the provided ID does not exist.
-        await io.logger.error("Error retrieving model:", error.message);
-    }
-},
-```
-
-### `listModels`
-
-Lists the available models. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/models/list)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code lists available models without retrieving detailed information.
-    const models = await io.openai.listModels("list-models");
-},
-```
-
-### `createEdit`
-
-Edits a given text prompt. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/edits/create)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code snippet demonstrates using the OpenAI API to create an edit task.
-
-    // Specify the task parameters:
-    const editTaskParams = {
-        model: "text-davinci-edit-001", // Replace with the desired model
-        input: "Thsi is ridddled with erors", // Replace with the input text
-        instruction: "Fix the spelling errors", // Replace with the editing instruction
-    };
-
-    try {
-        // Create an edit task using the OpenAI API
-        const editResponse = await io.openai.createEdit("edit", editTaskParams);
-
-        // Log the response
-        await io.logger.info("editResponse", editResponse);
-    } catch (error) {
-        // Handle any potential errors that may occur during the API request.
-        await io.logger.error("Error creating edit task:", error.message);
-    }
-},
-```
-
-### `createImage`
-
-Generates images from textual descriptions. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/images/create)
-
-```ts example.ts
-  run: async (payload, io, ctx) => {
-    const imageResults = await io.openai.createImage("image", {
-      prompt: "A hedgehog wearing a party hat",
-      n: 2,
-      size: "256x256",
-      response_format: "url",
-    });
-```
-
-### `createImageEdit`
-
-Creates an edited or extended image given an original image and a prompt. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/images/createEdit)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the parameters for the image edit
-  const imageEditParams = {
-    style: "data:image/png;base64,base64_encoded_style_image",
-    content: "data:image/png;base64,base64_encoded_content_image",
-  };
-
-  // Create the image edit using the OpenAI API
-  const imageEditResponse = await io.openai.createImageEdit(imageEditParams);
-
-  // Log the response
-  await io.logger.info("imageEditResponse", imageEditResponse);
-},
-```
-
-### `createImageVariation`
-
-Creates a variation of a given image. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/images/createVariation)
-
-```ts example.ts
-    run: async (payload, io, ctx) => {
-        // Specify the parameters for creating an image variation
-        const imageVariationParams = {
-        image: "data:image/png;base64,base64_encoded_image",
-        variation: "brightness(1.2) contrast(0.8) rotate(45deg)",
-        };
-
-        // Create the image variation using the OpenAI API
-        const imageVariationResponse = await io.openai.createImageVariation(imageVariationParams);
-
-        // Log the response
-        await io.logger.info("imageVariationResponse", imageVariationResponse);
-    },
-```
-
-### `createEmbedding`
-
-Generates embeddings for a given text. [Official OpenAI Docs](hhttps://platform.openai.com/docs/api-reference/embeddings/object)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-    // This code snippet demonstrates using the OpenAI API to create a text embedding.
-
-    // Specify the task parameters:
-    const embeddingTaskParams = {
-        model: "text-embedding-ada-002", // Replace with the desired model
-        input: "The food was delicious and the waiter...", // Replace with the input text
-    };
-
-    try {
-        // Create a text embedding using the OpenAI API
-        const embeddingResponse = await io.openai.createEmbedding("embedding", embeddingTaskParams);
-
-        // Log the response
-        await io.logger.info("embeddingResponse", embeddingResponse);
-    } catch (error) {
-        // Handle any potential errors that may occur during the API request.
-        await io.logger.error("Error creating text embedding:", error.message);
-    }
-},
-```
-
-### `createFile`
-
-Uploads a file to the OpenAI API. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/files/object)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the parameters for creating a file
-  const fileParams = {
-    name: "example.txt",
-    content: "This is the content of the file.",
-  };
-
-  // Create the file using the OpenAI API
-  const fileResponse = await io.openai.createFile(fileParams);
-
-  // Log the response
-  await io.logger.info("fileResponse", fileResponse);
-},
-```
-
-### `listFiles`
-
-Lists the uploaded files. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/files/list)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // List the files available in your OpenAI account
-  const fileListResponse = await io.openai.listFiles();
-
-  // Log the list of files
-  await io.logger.info("fileListResponse", fileListResponse);
-},
-```
-
-### `createFineTuneFile`
-
-Uploads a file for fine-tuning a model. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the parameters for creating a fine-tune file
-  const fineTuneFileParams = {
-    model: "text-davinci-002",
-    prompt: "Translate English to French: 'Hello, world.'",
-    language: "en",
-    description: "Fine-tune file for translation task",
-  };
-
-  // Create the fine-tune file using the OpenAI API
-  const fineTuneFileResponse = await io.openai.createFineTuneFile(fineTuneFileParams);
-
-  // Log the response
-  await io.logger.info("fineTuneFileResponse", fineTuneFileResponse);
-},
-```
-
-### `createFineTune`
-
-Fine-tunes a model on a given task. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/create)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the parameters for creating a fine-tune task
-  const fineTuneParams = {
-    model: "text-davinci-002",
-    dataset: "your_dataset_id",
-    description: "Fine-tune task for custom dataset",
-  };
-
-  // Create the fine-tune task using the OpenAI API
-  const fineTuneResponse = await io.openai.createFineTune(fineTuneParams);
-
-  // Log the response
-  await io.logger.info("fineTuneResponse", fineTuneResponse);
-},
-```
-
-### `listFineTunes`
-
-Lists the available fine-tunes. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/list)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // List the fine-tunes available in your OpenAI account
-  const fineTunesListResponse = await io.openai.listFineTunes();
-
-  // Log the list of fine-tunes
-  await io.logger.info("fineTunesListResponse", fineTunesListResponse);
-},
-```
-
-### `retrieveFineTune`
-
-Retrieves a specific fine-tune by ID. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/retrieve)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the ID of the fine-tune you want to retrieve
-  const fineTuneId = "your_fine_tune_id"; // Replace with the actual fine-tune ID
-
-  // Retrieve the fine-tune using the OpenAI API
-  const retrievedFineTune = await io.openai.retrieveFineTune(fineTuneId);
-
-  // Log the retrieved fine-tune
-  await io.logger.info("retrievedFineTune", retrievedFineTune);
-},
-```
-
-### `cancelFineTune`
-
-Cancels a specific fine-tune by ID. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/cancel)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the ID of the fine-tune you want to cancel
-  const fineTuneIdToCancel = "your_fine_tune_id"; // Replace with the actual fine-tune ID
-
-  // Cancel the specified fine-tune using the OpenAI API
-  const cancellationResponse = await io.openai.cancelFineTune(fineTuneIdToCancel);
-
-  // Log the cancellation response
-  await io.logger.info("cancellationResponse", cancellationResponse);
-},
-```
-
-### `createFineTuningJob`
-
-Creates a job that fine-tunes a specified model from a given dataset. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/create)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the parameters for creating a fine-tuning job
-  const fineTuningJobParams = {
-    fineTuneId: "your_fine_tune_id", // Replace with the actual fine-tune ID
-    datasetId: "your_dataset_id", // Replace with the ID of your dataset
-    model: "text-davinci-002", // Replace with the model for fine-tuning
-    n_examples: 100, // Replace with the number of examples
-  };
-
-  // Create the fine-tuning job using the OpenAI API
-  const fineTuningJobResponse = await io.openai.createFineTuningJob(fineTuningJobParams);
-
-  // Log the response
-  await io.logger.info("fineTuningJobResponse", fineTuningJobResponse);
-},
-```
-
-### `retrieveFineTuningJob`
-
-Get info about a fine-tuning job. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/retrieve)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the ID of the fine-tuning job you want to retrieve
-  const fineTuningJobId = "your_fine_tuning_job_id"; // Replace with the actual job ID
-
-  // Retrieve the fine-tuning job using the OpenAI API
-  const retrievedJob = await io.openai.retrieveFineTuningJob(fineTuningJobId);
-
-  // Log the retrieved job
-  await io.logger.info("retrievedJob", retrievedJob);
-},
-```
-
-### `cancelFineTuningJob`
-
-Cancel a fine-tuning job. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/cancel)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the ID of the fine-tuning job you want to cancel
-  const fineTuningJobIdToCancel = "your_fine_tuning_job_id"; // Replace with the actual job ID
-
-  // Cancel the specified fine-tuning job using the OpenAI API
-  const cancellationResponse = await io.openai.cancelFineTuningJob(fineTuningJobIdToCancel);
-
-  // Log the cancellation response
-  await io.logger.info("cancellationResponse", cancellationResponse);
-},
-```
-
-### `listFineTuningJobEvents`
-
-List events for a fine-tuning job. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/list-events)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // Specify the ID of the fine-tuning job for which you want to list events
-  const fineTuningJobId = "your_fine_tuning_job_id"; // Replace with the actual job ID
-
-  // List events for the specified fine-tuning job using the OpenAI API
-  const eventsListResponse = await io.openai.listFineTuningJobEvents(fineTuningJobId);
-
-  // Log the list of events
-  await io.logger.info("eventsListResponse", eventsListResponse);
-},
-```
-
-### `listFineTuningJobs`
-
-List fine tuning jobs. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/list)
-
-```ts example.ts
-run: async (payload, io, ctx) => {
-  // List the fine-tuning jobs available in your OpenAI account
-  const jobsListResponse = await io.openai.listFineTuningJobs();
-
-  // Log the list of fine-tuning jobs
-  await io.logger.info("jobsListResponse", jobsListResponse);
-},
-```
-
-## Example usage
-
-In this example we'll create a task that generates a random joke using OpenAI GPT 3.5 .
-
-```ts example.ts
-import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
-import { OpenAI } from "@trigger.dev/openai";
-import { z } from "zod";
-
-// Initialize a TriggerClient with the ID "jobs-showcase"
-const client = new TriggerClient({ id: "jobs-showcase" });
-
-// Create an instance of the OpenAI client and provide the OpenAI API key from environment variables
-const openai = new OpenAI({
-  id: "openai",
-  apiKey: process.env.OPENAI_API_KEY!, // Replace with your actual OpenAI API key
-});
-
-// Define a job that uses OpenAI GPT-3.5 Turbo to tell jokes
-client.defineJob({
-  id: "openai-tell-me-a-joke",
-  name: "OpenAI: tell me a joke",
-  version: "1.0.0",
-  trigger: eventTrigger({
-    name: "openai.tasks", // Define the trigger event name
-    schema: z.object({
-      jokePrompt: z.string(), // Expect a joke prompt as input
-    }),
-  }),
-  integrations: {
-    openai, // Use the OpenAI integration for this job
-  },
-  run: async (payload, io, ctx) => {
-    // Retrieve information about the GPT-3.5 Turbo model
-    await io.openai.retrieveModel("get-model", {
-      model: "gpt-3.5-turbo",
-    });
-
-    // List available models (optional, for reference)
-    const models = await io.openai.listModels("list-models");
-
-    // Generate a joke in the background using the chat conversation format
-    const jokeResult = await io.openai.backgroundCreateChatCompletion(
-      "background-chat-completion",
-      {
-        model: "gpt-3.5-turbo",
-        messages: [
-          {
-            role: "user",
-            content: payload.jokePrompt, // User-provided joke prompt
-          },
-        ],
-      }
-    );
-
-    // Return the generated joke as the result
-    return {
-      joke: jokeResult.choices[0]?.message?.content,
-    };
-  },
-});
-
-// These lines are specific to the Express framework and can be removed if not needed
-import { createExpressServer } from "@trigger.dev/express";
-createExpressServer(client);
-```

docs/integrations/apis/openai.mdx

@@ -49,11 +49,71 @@ const openai = new OpenAI({
 
 ## Tasks
 
-Once you have set up a OpenAI client, you can use it to create tasks.
+Once you have set up a OpenAI client, you can add it to your job and start using the provided tasks:
+
+```ts
+client.defineJob({
+  id: "openai-job",
+  name: "OpenAI Job",
+  version: "1.0.0",
+  trigger: invokeTrigger(),
+  integrations: {
+    openai, // Add the OpenAI client as an integration
+  },
+  run: async (payload, io, ctx) => {
+    // Now you can access it through the io object
+    const completion = await io.openai.chat.completions.create("completion", {
+      model: "gpt-3.5-turbo",
+      messages: [
+        {
+          role: "user",
+          content: "Create a good programming joke about background jobs",
+        },
+      ],
+    });
+  },
+});
+```
+
+As you can see above, we've replicated the API of the [OpenAI TypeScript SDK](https://github.com/openai/openai-node), with a crucial difference of adding the [Task Cache Key](https://trigger.dev/docs/documentation/concepts/tasks#task-cache-keys) as the first parameter.
+
+We've also added a few convenience methods to make it easier to work with the OpenAI API, especially in a serverless environment. For example, you can run a Chat Completion task in the background with [backgroundCreate()](/integrations/apis/openai/chat#completions-backgroundcreate):
+
+```ts
+const completion = await io.openai.chat.completions.backgroundCreate("completion", {
+  model: "gpt-3.5-turbo",
+  messages: [
+    {
+      role: "user",
+      content: "Create a good programming joke about background jobs",
+    },
+  ],
+});
+```
+
+See our full task reference below:
 
 <CardGroup>
-  <Card title="Tasks" icon="sparkles" href="/integrations/apis/openai-tasks">
-    Perform different AI-powered tasks using OpenAI.
+  <Card title="Chat Completions" icon="sparkles" href="/integrations/apis/openai/chat">
+    Given a list of messages comprising a conversation, the model will return a response
+  </Card>
+  <Card title="Assistants (Beta)" icon="arrows-spin" href="/integrations/apis/openai/assistants">
+    Build assistants that can call models and use tools to perform tasks
+  </Card>
+  <Card title="Files" icon="file" href="/integrations/apis/openai/files">
+    Upload files to use with assistants and fine-tuning
+  </Card>
+  <Card title="Images" icon="image" href="/integrations/apis/openai/images">
+    Given a prompt and/or an input image, the model will generate a new image
+  </Card>
+  <Card title="Fine Tuning Jobs" icon="vial" href="/integrations/apis/openai/fine-tunes">
+    Manage fine-tuning jobs to tailor a model to your specific training data
+  </Card>
+  <Card title="Models" icon="server" href="/integrations/apis/openai/models">
+    List and describe the various models available in the API
+  </Card>
+  <Card title="Completions (Legacy)" icon="scroll" href="/integrations/apis/openai/completions">
+    Given a prompt, the model will return one or more predicted completions.
   </Card>
 </CardGroup>
 

docs/integrations/apis/openai/assistants.mdx

@@ -0,0 +1,287 @@
+---
+title: Assistant Tasks
+sidebarTitle: Assitants (Beta)
+---
+
+<Note>
+  This feature is currently marked as a "Beta" by OpenAI. Make sure to check our their [How
+  Assistants Work](https://platform.openai.com/docs/assistants/how-it-works) and [Assistants
+  Overview](https://platform.openai.com/docs/assistants/overview) guides.
+</Note>
+
+## Assistants
+
+Build assistants that can call models and use tools to perform tasks. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/assistants)
+
+### `create()`
+
+Create an assistant with a model and instructions. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/createAssistant)
+
+```ts example.ts
+const file = await io.openai.files.createAndWaitForProcessing("upload-file", {
+  purpose: "assistants",
+  file: fs.createReadStream("./fixtures/mydata.csv"),
+});
+
+const assistant = await io.openai.beta.assistants.create("create-assistant", {
+  name: "Data visualizer",
+  description:
+    "You are great at creating beautiful data visualizations. You analyze data present in .csv files, understand trends, and come up with data visualizations relevant to those trends. You also share a brief text summary of the trends observed.",
+  model: "gpt-4-1106-preview",
+  tools: [{ type: "code_interpreter" }],
+  file_ids: [file.id],
+});
+```
+
+## Threads
+
+Create threads that assistants can interact with. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/threads/createThread)
+
+### `create()`
+
+Create a thread. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/assistants/createAssistant)
+
+```ts example.ts
+const thread = await io.openai.beta.threads.create("create-thread", {
+  messages: [
+    {
+      role: "user",
+      content: "Create 3 data visualizations based on the trends in this file.",
+      file_ids: [fileId],
+    },
+  ],
+});
+```
+
+### `createAndRun()`
+
+Create a thread and run it in one task.
+
+```ts example.ts
+const run = await io.openai.beta.threads.createAndRun("create-and-run-thread", {
+  assistant_id: "asst_abc123",
+  thread: {
+    messages: [
+      {
+        role: "user",
+        content: "Create 3 data visualizations based on the trends in this file.",
+        file_ids: [fileId],
+      },
+    ],
+  },
+});
+```
+
+### `createAndRunUntilCompletion()`
+
+Create a thread and runs it in one task, and only returns when the run is completed by polling in the background using [io.backgroundPoll()](/sdk/io/background-poll).
+
+```ts example.ts
+const run = await io.openai.beta.threads.createAndRunUntilCompletion("create-thread", {
+  assistant_id: "asst_abc123",
+  thread: {
+    messages: [
+      {
+        role: "user",
+        content: "Create 3 data visualizations based on the trends in this file.",
+        file_ids: [fileId],
+      },
+    ],
+  },
+});
+
+if (run.status !== "completed") {
+  throw new Error(`Run finished with status ${run.status}: ${JSON.stringify(run.last_error)}`);
+}
+
+// List all messages in the thread
+const messages = await io.openai.beta.threads.messages.list("list-messages", run.thread_id);
+```
+
+### `retrieve()`
+
+Retrieves a thread. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/threads/getThread)
+
+```ts example.ts
+const thread = await io.openai.beta.threads.retrieve("get-thread", "thread_abc123");
+```
+
+### `update()`
+
+Modifies a thread. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/threads/modifyThread)
+
+```ts example.ts
+await io.openai.beta.threads.update("update-thread", "thread_abc123", {
+  metadata: {
+    foo: "bar",
+  },
+});
+```
+
+### `del()`
+
+Deletes a thread. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/threads/deleteThread)
+
+```ts example.ts
+const deletedThread = await io.openai.beta.threads.del("update-thread", "thread_abc123");
+```
+
+## Messages
+
+Create messages within threads. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/messages)
+
+### `list()`
+
+List all messages in a thread.
+
+```ts example.ts
+const messages = await io.openai.beta.threads.messages.list("list-messages", "thread_abc123");
+```
+
+### `create()`
+
+Create a message. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/messages/createMessage)
+
+```ts example.ts
+const thread = await io.openai.beta.threads.create("get-thread");
+
+const message = await io.openai.beta.threads.messages.create("create-message", thread.id, {
+  role: "user",
+  content: "Create 3 data visualizations based on the trends in this file.",
+  file_ids: [fileId],
+});
+```
+
+### `retrieve()`
+
+Retrieve a message. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/messages/getMessage)
+
+```ts example.ts
+const message = await io.openai.beta.threads.messages.retrieve(
+  "get-message",
+  "thread_abc123",
+  "message_abc123"
+);
+```
+
+### `update()`
+
+Update a message. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/messages/modifyMessage)
+
+```ts example.ts
+await io.openai.beta.threads.messages.update("update-message", thread.id, message.id, {
+  metadata: {
+    foo: "bar",
+  },
+});
+```
+
+## Runs
+
+Represents an execution run on a thread. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/runs)
+
+### `list()`
+
+List all runs belonging to a thread.
+
+```ts example.ts
+const runs = await io.openai.beta.threads.runs.list("list-runs", "thread_abc123");
+```
+
+### `create()`
+
+Create a run. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/runs/createRun)
+
+```ts example.ts
+const run = await io.openai.beta.threads.runs.create("create-run", "thread_abc123", {
+  assistant_id: payload.id,
+});
+```
+
+### `createAndWaitForCompletion()`
+
+Create a run and only return when the run is completed by polling in the background using [io.backgroundPoll()](/sdk/io/background-poll).
+
+```ts example.ts
+const run = await io.openai.beta.threads.runs.createAndWaitForCompletion(
+  "create-run",
+  "thread_abc123",
+  {
+    assistant_id: payload.id,
+  }
+);
+
+if (run.status !== "completed") {
+  throw new Error(`Run finished with status ${run.status}: ${JSON.stringify(run.last_error)}`);
+}
+
+const messages = await io.openai.beta.threads.messages.list("list-messages", run.thread_id);
+```
+
+### `waitForCompletion()`
+
+Wait for a run to complete by polling in the background using [io.backgroundPoll()](/sdk/io/background-poll).
+
+```ts example.ts
+const run = await io.openai.beta.threads.runs.create("create-run", "thread_abc123", {
+  assistant_id: payload.id,
+});
+
+const completedRun = await io.openai.beta.threads.runs.waitForCompletion(
+  "wait-for-completion",
+  "thread_abc123",
+  run.id
+);
+
+if (completedRun.status !== "completed") {
+  throw new Error(
+    `Run finished with status ${completedRun.status}: ${JSON.stringify(completedRun.last_error)}`
+  );
+}
+
+const messages = await io.openai.beta.threads.messages.list("list-messages", run.thread_id);
+```
+
+### `retrieve()`
+
+Retrieve a run. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/runs/getRun)
+
+```ts example.ts
+const run = await io.openai.beta.threads.runs.retrieve("get-run", "thread_abc123", "run_abc123");
+```
+
+### `cancel()`
+
+Cancels a run that is `in_progress`. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/runs/cancelRun)
+
+```ts example.ts
+const run = await io.openai.beta.threads.runs.cancel("cancel-run", "thread_abc123", "run_abc123");
+```
+
+### `submitToolOutputs()`
+
+When a run has the `status: "requires_action"` and `required_action.type` is `submit_tool_outputs`, this endpoint can be used to submit the outputs from the tool calls once they're all completed. All outputs must be submitted in a single request. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/runs/submitToolOutputs)
+
+```ts example.ts
+const run = await io.openai.beta.threads.runs.submitToolOutputs(
+  "submit-tool-outputs",
+  "thread_abc123",
+  "run_abc123",
+  {
+    tool_outputs: [
+      {
+        tool_call_id: "tool_run_abc123",
+        output: "This is the output of the tool call.",
+      },
+    ],
+  }
+);
+```
+
+### `list()`
+
+Returns all runs belonging to a thread. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/runs/listRuns)
+
+```ts example.ts
+const runs = await io.openai.beta.threads.runs.list("list-runs", "thread_abc123");
+```

docs/integrations/apis/openai/chat.mdx

@@ -0,0 +1,38 @@
+---
+title: Chat Completion Tasks
+sidebarTitle: Chat Completions
+---
+
+Given a list of messages comprising a conversation, the model will return a response. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/chat)
+
+### `completions.create()`
+
+Creates a model response for the given chat conversation. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/chat/create)
+
+```ts example.ts
+await io.openai.chat.completions.create("chat-completion", {
+  model: "gpt-3.5-turbo",
+  messages: [
+    {
+      role: "user",
+      content: "Create a good programming joke about background jobs",
+    },
+  ],
+});
+```
+
+### `completions.backgroundCreate()`
+
+Creates a model response for the given chat conversation, but runs the request in the background using [io.backgroundFetch()](/sdk/io/backgroundfetch)
+
+```ts example.ts
+await io.openai.chat.completions.create("chat-completion", {
+  model: "gpt-3.5-turbo",
+  messages: [
+    {
+      role: "user",
+      content: "Create a good programming joke about background jobs",
+    },
+  ],
+});
+```

docs/integrations/apis/openai/completions.mdx

@@ -0,0 +1,19 @@
+---
+title: Completion Tasks
+sidebarTitle: Completions (Legacy)
+---
+
+Given a prompt, the model will return one or more predicted completions, and can also return the probabilities of alternative tokens at each position. We recommend most users use the Chat Completions API. [Learn more](https://platform.openai.com/docs/deprecations/2023-07-06-gpt-and-embeddings)
+
+### `create()`
+
+<Warning>This is a legacy API</Warning>
+
+Creates a completion for the provided prompt and parameters. [Official OpenAI docs](https://platform.openai.com/docs/api-reference/completions/create)
+
+```ts example.ts
+const completion = await io.openai.completions.create("completion", {
+  model: "text-davinci-003",
+  prompt: "Create a good programming joke about Tasks",
+});
+```

docs/integrations/apis/openai/files.mdx

@@ -0,0 +1,62 @@
+---
+title: File Tasks
+sidebarTitle: Files
+---
+
+Files are used to upload documents that can be used with features like [Assistants](https://platform.openai.com/docs/api-reference/assistants) and [Fine-tuning](https://platform.openai.com/docs/api-reference/fine-tuning). [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/files)
+
+### `list()`
+
+Returns a list of files that belong to the user's organization. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/files/list)
+
+```ts example.ts
+await io.openai.files.list("list-files");
+await io.openai.files.list("list-files", { purpose: "assistants" }); // gets only assistant files
+```
+
+### `create()`
+
+Upload a file that can be used across various endpoints/features. The size of all the files uploaded by one organization can be up to 100 GB.
+
+The size of individual files for can be a maximum of `512MB`. See the [Assistants Tools guide](https://platform.openai.com/docs/assistants/tools) to learn more about the types of files supported. The Fine-tuning API only supports `.jsonl` files.
+
+[Official OpenAI Docs](https://platform.openai.com/docs/api-reference/files/create)
+
+```ts example.ts
+const file = await io.openai.files.create("upload-file", {
+  purpose: "assistants",
+  file: fs.createReadStream("./fixtures/mydata.csv"),
+});
+```
+
+### `createAndWaitForProcessing()`
+
+Upload a file and will return when the file is processed by polling in the background using [io.backgroundPoll()](/sdk/io/background-poll).
+
+```ts example.ts
+const file = await io.openai.files.createAndWaitForProcessing("upload-file", {
+  purpose: "assistants",
+  file: fs.createReadStream("./fixtures/mydata.csv"),
+});
+```
+
+### `waitForProcessing()`
+
+Will return when the file is processed by polling in the background using [io.backgroundPoll()](/sdk/io/background-poll).
+
+```ts example.ts
+const file = await io.openai.files.create("upload-file", {
+  purpose: "assistants",
+  file: fs.createReadStream("./fixtures/mydata.csv"),
+});
+
+const processedFile = await io.openai.files.waitForProcessing("wait", file.id);
+```
+
+### `retrieve()`
+
+Returns information about a specific file. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/files/retrieve)
+
+```ts example.ts
+await io.openai.files.retrieve("retrieve-file", "file-id");
+```

docs/integrations/apis/openai/fine-tunes.mdx

@@ -0,0 +1,59 @@
+---
+title: Fine Tuning Tasks
+sidebarTitle: Fine Tunes
+---
+
+Manage fine-tuning jobs to tailor a model to your specific training data. See the related guide [Fine Tuning models](https://platform.openai.com/docs/guides/fine-tuning) and view the [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning).
+
+### `jobs.create()`
+
+Creates a job that fine-tunes a specified model from a given dataset.
+
+Response includes details of the enqueued job including job status and the name of the fine-tuned models once complete.
+
+You must first upload a dataset to the API before creating a fine-tuning job. See our [OpenAI File Tasks](/integrations/apis/openai/files#createandwaitforprocessing) for more information.
+
+```ts example.ts
+const file = await io.openai.files.create("upload-file", {
+  purpose: "fine-tune",
+  file: fs.createReadStream("./mydata.jsonl"),
+});
+
+const fineTuning = await io.openai.fineTuning.jobs.create("fine-tuning", {
+  training_file: file.id,
+  model: "gpt-3.5-turbo",
+  suffix: "my-model",
+});
+```
+
+### `jobs.list()`
+
+List your organization's fine-tuning jobs. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/list)
+
+```ts example.ts
+const fts = await io.openai.fineTuning.jobs.list("list");
+```
+
+### `jobs.retrieve()`
+
+Get info about a fine-tuning job. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/retrieve)
+
+```ts example.ts
+const fineTuning = await io.openai.fineTuning.jobs.retrieve("fine-tuning", "ft_1234");
+```
+
+### `jobs.cancel()`
+
+Immediately cancel a fine-tune job. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/cancel)
+
+```ts example.ts
+const fineTuning = await io.openai.fineTuning.jobs.cancel("fine-tuning", "ft_1234");
+```
+
+### `jobs.listEvents()`
+
+Get status updates for a fine-tuning job. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/fine-tuning/list-events)
+
+```ts example.ts
+const events = await io.openai.fineTuning.jobs.listEvents("fine-tuning", { id: "ft_1234" });
+```

docs/integrations/apis/openai/images.mdx

@@ -0,0 +1,55 @@
+---
+title: Image Tasks
+sidebarTitle: Images
+---
+
+Given a prompt and/or an input image, the model will generate a new image. See the [Image generation guide](https://platform.openai.com/docs/guides/images) and the [Official OpenAI docs](https://platform.openai.com/docs/api-reference/images).
+
+### `create()`
+
+Creates an image given a prompt. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/images/create)
+
+```ts example.ts
+await io.openai.images.create("dalle-3", {
+  model: "dall-e-3",
+  prompt:
+    "I would like to generate an image of an american giraffe riding a bycicle in a suburban neighborhood, into the sunset.",
+});
+```
+
+### `backgroundCreate()`
+
+Creates a an image given a prompt, but runs the request in the background using [io.backgroundFetch()](/sdk/io/backgroundfetch)
+
+```ts example.ts
+await io.openai.images.backgroundCreate("dalle-3", {
+  model: "dall-e-3",
+  prompt:
+    "I would like to generate an image of an american giraffe riding a bycicle in a suburban neighborhood, into the sunset.",
+});
+```
+
+### `edit()`
+
+Creates an edited or extended image given an original image and a prompt. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/images/createEdit)
+
+```ts example.ts
+await io.openai.images.edit("dalle-2", {
+  model: "dall-e-2",
+  image: fs.createReadStream("./giraffe.jpg"),
+  prompt: "A painting of a giraffe in a suburban neighborhood",
+  response_format: "url",
+});
+```
+
+### `createVariation()`
+
+Creates a variation of a given image. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/images/createVariation)
+
+```ts example.ts
+await io.openai.images.createVariation("dalle-3", {
+  model: "dall-e-2",
+  image: fs.createReadStream("./giraffe.jpg"),
+  response_format: "url",
+});
+```

docs/integrations/apis/openai/models.mdx

@@ -0,0 +1,14 @@
+---
+title: Model Tasks
+sidebarTitle: Models
+---
+
+List and describe the various models available in the API. You can refer to the Models documentation to understand what models are available and the differences between them. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/models)
+
+### `list`
+
+Lists the available models. [Official OpenAI Docs](https://platform.openai.com/docs/api-reference/models/list)
+
+```ts example.ts
+const models = await io.openai.models.list("list-models");
+```

docs/mint.json

@@ -1,6 +1,9 @@
 {
   "$schema": "https://mintlify.com/schema.json",
   "name": "Trigger.dev",
+  "openapi": [
+    "/openapi.yml"
+  ],
   "logo": {
     "dark": "/logo/dark.png",
     "light": "/logo/light.png",
@@ -95,6 +98,7 @@
           ]
         },
         "documentation/guides/writing-jobs-step-by-step",
+        "documentation/guides/task-library",
         "documentation/guides/video-walkthrough"
       ]
     },
@@ -251,7 +255,10 @@
       "pages": [
         {
           "group": "Airtable",
-          "pages": ["integrations/apis/airtable", "integrations/apis/airtable-tasks"]
+          "pages": [
+            "integrations/apis/airtable",
+            "integrations/apis/airtable-tasks"
+          ]
         },
         {
           "group": "GitHub",
@@ -264,18 +271,33 @@
         "integrations/apis/linear",
         {
           "group": "OpenAI",
-          "pages": ["integrations/apis/openai", "integrations/apis/openai-tasks"]
+          "pages": [
+            "integrations/apis/openai",
+            "integrations/apis/openai/chat",
+            "integrations/apis/openai/assistants",
+            "integrations/apis/openai/files",
+            "integrations/apis/openai/images",
+            "integrations/apis/openai/fine-tunes",
+            "integrations/apis/openai/models",
+            "integrations/apis/openai/completions"
+          ]
         },
         {
           "group": "Plain",
-          "pages": ["integrations/apis/plain", "integrations/apis/plain-tasks"]
+          "pages": [
+            "integrations/apis/plain",
+            "integrations/apis/plain-tasks"
+          ]
         },
         "integrations/apis/replicate",
         "integrations/apis/resend",
         "integrations/apis/sendgrid",
         {
           "group": "Slack",
-          "pages": ["integrations/apis/slack", "integrations/apis/slack-tasks"]
+          "pages": [
+            "integrations/apis/slack",
+            "integrations/apis/slack-tasks"
+          ]
         },
         "integrations/apis/stripe",
         {
@@ -302,6 +324,7 @@
               "group": "Instance methods",
               "pages": [
                 "sdk/triggerclient/instancemethods/sendevent",
+                "sdk/triggerclient/instancemethods/sendevents",
                 "sdk/triggerclient/instancemethods/getevent",
                 "sdk/triggerclient/instancemethods/cancel-event",
                 "sdk/triggerclient/instancemethods/cancel-runs-for-event",
@@ -322,11 +345,15 @@
           "pages": [
             "sdk/io/overview",
             "sdk/io/runtask",
-            "sdk/io/wait",
-            "sdk/io/logger",
             "sdk/io/sendevent",
+            "sdk/io/sendevents",
+            "sdk/io/wait",
+            "sdk/io/wait-for-event",
+            "sdk/io/wait-for-request",
             "sdk/io/backgroundfetch",
+            "sdk/io/background-poll",
             "sdk/io/random",
+            "sdk/io/logger",
             "sdk/io/try",
             "sdk/io/registerinterval",
             "sdk/io/unregisterinterval",
@@ -350,7 +377,10 @@
             "sdk/dynamictrigger/constructor",
             {
               "group": "Instance methods",
-              "pages": ["sdk/dynamictrigger/register", "sdk/dynamictrigger/unregister"]
+              "pages": [
+                "sdk/dynamictrigger/register",
+                "sdk/dynamictrigger/unregister"
+              ]
             }
           ]
         },
@@ -361,7 +391,10 @@
             "sdk/dynamicschedule/constructor",
             {
               "group": "Instance methods",
-              "pages": ["sdk/dynamicschedule/register", "sdk/dynamicschedule/unregister"]
+              "pages": [
+                "sdk/dynamicschedule/register",
+                "sdk/dynamicschedule/unregister"
+              ]
             }
           ]
         },
@@ -371,6 +404,12 @@
         "sdk/verify-request-signature"
       ]
     },
+    {
+      "group": "HTTP Reference",
+      "pages": [
+        "sdk/api-reference/events/create-an-event"
+      ]
+    },
     {
       "group": "React SDK",
       "pages": [
@@ -383,7 +422,9 @@
     },
     {
       "group": "Overview",
-      "pages": ["examples/introduction"]
+      "pages": [
+        "examples/introduction"
+      ]
     }
   ],
   "footerSocials": {
@@ -396,4 +437,4 @@
       "apiKey": "phc_hwYmedO564b3Ik8nhA4Csrb5SueY0EwFJWCbseGwWW"
     }
   }
-}
+}

docs/openapi.yml

@@ -0,0 +1,137 @@
+openapi: 3.0.0
+info:
+  title: Trigger.dev API
+  description: API for triggering events in Trigger.dev
+  version: 1.0.0
+servers:
+  - url: https://api.trigger.dev
+    description: Trigger.dev API server
+security:
+  - BearerAuth: []
+paths:
+  /api/v1/events:
+    post:
+      operationId: sendEvent
+      externalDocs:
+        description: Find more info here
+        url: "https://trigger.dev/docs/api/events/send-event"
+      tags:
+        - Events
+      summary: Create an event
+      description: Send an event to Trigger.dev to trigger job runs through eventTrigger()
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/EventRequest"
+      responses:
+        "200":
+          description: Event successfully sent
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/EventResponse"
+        "400":
+          description: Invalid request
+        "401":
+          description: Unauthorized - API key is missing or invalid
+        "422":
+          description: Invalid request body
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Error"
+
+components:
+  schemas:
+    Error:
+      type: object
+      properties:
+        message:
+          type: string
+    EventRequest:
+      type: object
+      properties:
+        event:
+          type: object
+          required:
+            - name
+          properties:
+            name:
+              type: string
+              description: The name of the event
+            payload:
+              type: object
+              additionalProperties: true
+              description: The payload of the event
+            context:
+              type: object
+              additionalProperties: true
+              description: An optional context object
+            id:
+              type: string
+              description: Unique identifier for the event. Auto-generated if not provided. If you provide an ID that already exists, the event will not be redelivered.
+            timestamp:
+              type: string
+              format: date-time
+              description: Event timestamp. Defaults to current timestamp if not provided.
+            source:
+              type: string
+              description: Event source, default is 'trigger.dev'.
+        options:
+          type: object
+          properties:
+            deliverAt:
+              type: string
+              format: date-time
+              description: Optional Date to deliver the event.
+            deliverAfter:
+              type: integer
+              description: Optional delay in seconds before delivering the event.
+            accountId:
+              type: string
+              description: Optional account ID to associate with the event.
+    EventResponse:
+      type: object
+      properties:
+        id:
+          type: string
+          description: The ID of the event that was sent.
+        name:
+          type: string
+          description: The name of the event that was sent.
+        payload:
+          $ref: "#/components/schemas/DeserializedJson"
+        context:
+          $ref: "#/components/schemas/DeserializedJson"
+          nullable: true
+          description: The context of the event that was sent. Null if no context was set.
+        timestamp:
+          type: string
+          format: date-time
+          description: The timestamp of the event that was sent.
+        deliverAt:
+          type: string
+          format: date-time
+          nullable: true
+          description: The timestamp when the event will be delivered. Null if not applicable.
+        deliveredAt:
+          type: string
+          format: date-time
+          nullable: true
+          description: The timestamp when the event was delivered. Null if not applicable.
+        cancelledAt:
+          type: string
+          format: date-time
+          nullable: true
+          description: The timestamp when the event was cancelled. Null if the event wasn't cancelled.
+    DeserializedJson:
+      type: object
+      additionalProperties: true
+      description: A JSON object that represents the deserialized payload or context.
+  securitySchemes:
+    BearerAuth:
+      type: http
+      scheme: bearer
+      bearerFormat: JWT

docs/sdk/api-reference/events/create-an-event.mdx

@@ -0,0 +1,3 @@
+---
+openapi: post /api/v1/events
+---

docs/sdk/invoke-trigger.mdx

@@ -3,8 +3,6 @@ title: "invokeTrigger()"
 description: "Use `invokeTrigger()` to allow a Job to be manually triggered using `Job.invoke()`"
 ---
 
-<Warning>This feature is in beta and not yet available to use on the Trigger.dev Cloud.</Warning>
-
 Setting `invokeTrigger()` on a job allows you to manually trigger the job using `Job.invoke()`, either from your backend or from inside another job. See our [Invoke Trigger guide](/documentation/guides/invoke) for more info.
 
 ## Parameters

docs/sdk/io/background-poll.mdx

@@ -0,0 +1,137 @@
+---
+title: "io.backgroundPoll()"
+sidebarTitle: "backgroundPoll()"
+description: "`io.backgroundPoll()` allows you to fetch data from a URL on an interval."
+---
+
+## Parameters
+
+<Snippet file="stable-key-param.mdx" />
+
+<ResponseField name="url" type="string" required>
+  The url to fetch.
+</ResponseField>
+
+<ResponseField name="interval" type="number" required>
+  The interval in seconds to wait between requests. Minimum interval is 10 seconds and maximum is 5
+  minutes.
+</ResponseField>
+
+<ResponseField name="timeout" type="number" required>
+  The timeout in seconds before aborting the polling. Minimum timeout is 30 seconds and maximum is 1
+  hour.
+</ResponseField>
+
+<ResponseField name="requestInit" type="RequestInit">
+  Options for the fetch request
+
+<Expandable title="options" defaultOpen>
+  <ResponseField name="method" type="string">
+    The HTTP method to use for the request.
+  </ResponseField>
+  <ResponseField name="headers" type="object">
+    Any headers to send with the request. Note that you can use [redactString](sdk/redactString) to
+    prevent sensitive information from being stored (e.g. in the logs), like API keys and tokens.
+  </ResponseField>
+  <ResponseField name="body" type="string | ArrayBuffer">
+    The body of the request.
+  </ResponseField>
+</Expandable>
+
+</ResponseField>
+
+<ResponseField name="responseFilter" type="ResponseFilter">
+  Allows you to filter the response to determine when to stop polling.
+
+<Expandable title="options" defaultOpen>
+  <ResponseField name="status" type="string[]">
+    An array of status codes to match against. 
+  </ResponseField>
+  <ResponseField name="headers" type="EventFilter">
+    An object of header key/values to match. This uses the [EventFilter matching syntax](/documentation/guides/event-filter)
+
+```ts example.ts
+filter: {
+  header: {
+    "content-type": [{ $startsWith: "application/json" }],
+  },
+},
+```
+
+  </ResponseField>
+  <ResponseField name="body" type="EventFilter">
+    An [EventFilter](/documentation/guides/event-filter) object to match against the response body. This will only be applied if the response body is JSON.
+  </ResponseField>
+</Expandable>
+
+</ResponseField>
+
+<ResponseField name="requestTimeout" type="object">
+  An optional object to specify a timeout for each individual request.
+
+<Expandable title="options" defaultOpen>
+  <ResponseField name="durationInMs" type="number" required>
+    The timeout in milliseconds before aborting the request.
+  </ResponseField>
+  
+  <ResponseField name="retry" type="RetryOptions">
+    
+    <Expandable title="options">
+
+{" "}
+
+<ResponseField name="limit" type="number">
+  The maximum number of times to retry the request.
+</ResponseField>
+<ResponseField name="minTimeoutInMs" type="number">
+  The minimum amount of time to wait before retrying the request.
+</ResponseField>
+<ResponseField name="maxTimeoutInMs" type="number">
+  The maximum amount of time to wait before retrying the request.
+</ResponseField>
+<ResponseField name="factor" type="number">
+  The exponential factor to use when calculating the next retry time.
+</ResponseField>
+<ResponseField name="randomize" type="boolean">
+  Whether to randomize the retry time.
+</ResponseField>
+
+    </Expandable>
+
+  </ResponseField>
+</Expandable>
+
+</ResponseField>
+
+## Returns
+
+A `Promise` that resolves with the JSON response body of the matching background fetch request. You can specify the type of the response body as a generic parameter.
+
+<RequestExample>
+
+```ts polling
+client.defineJob({
+  id: "background-poll-job",
+  name: "Background Poll Job",
+  version: "0.0.1",
+  trigger: invokeTrigger({
+    schema: z.object({ url: z.string().url() }),
+  }),
+  run: async (payload, io, ctx) => {
+    const result = await io.backgroundPoll<{ foo: string }>("poll", {
+      url: payload.url,
+      interval: 10, // every 10 seconds
+      timeout: 300, // stop polling after 5 minutes
+      responseFilter: {
+        // stop polling once this filter matches
+        status: [200],
+        body: {
+          status: ["SUCCESS"],
+        },
+      },
+    });
+  },
+});
+```
+
+</RequestExample>

docs/sdk/io/backgroundfetch.mdx

@@ -4,7 +4,7 @@ sidebarTitle: "backgroundFetch()"
 description: "`io.backgroundFetch()` allows you to fetch data from a URL that can take longer that the serverless timeout. The actual `fetch` request is performed on the Trigger.dev platform, and the response is sent back to you. An example use case is fetching data from a slow API, like some AI endpoints."
 ---
 
-This is used inside the OpenAI Integration for Tasks like `backgroundCreateChatCompletion` and `backgroundCreateCompletion`.
+This is used inside the OpenAI Integration for Tasks like [Chat Completions Background Create](/integrations/apis/openai/chat#completions-backgroundcreate)
 
 ## Parameters
 
@@ -140,7 +140,7 @@ An individual retrying strategy can be one of two types:
 
 ## Returns
 
-A `Promise` that resolves after the specified amount of time.
+A `Promise` that resolves with the JSON response body of the background fetch request. You can specify the type of the response body as a generic parameter.
 
 <RequestExample>
 

docs/sdk/io/overview.mdx

@@ -26,20 +26,38 @@ Used to send log messages to the [Run log](/documentation/guides/viewing-runs).
 
 `io.runTask()` allows you to run a [Task](/documentation/concepts/tasks) from inside a Job run. A Task is a resumable unit of a Run that can be retried, resumed and is logged. [Integrations](/integrations) use Tasks internally to perform their actions.
 
-### [wait()](/sdk/io/wait)
-
-Waits for a certain amount of time before continuing the Job. Delays works even if you're on a serverless platform with timeouts, or if your server goes down. They utilize [resumability](/documentation/concepts/resumability) to ensure that the Run can be resumed after the delay.
-
 ### [sendEvent()](/sdk/io/sendevent)
 
 `io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name).
 
 If you want to send an event from outside a run (e.g. just from your backend) you can use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent).
 
+### [wait()](/sdk/io/wait)
+
+Waits for a certain amount of time before continuing the Job. Delays works even if you're on a serverless platform with timeouts, or if your server goes down. They utilize [resumability](/documentation/concepts/resumability) to ensure that the Run can be resumed after the delay.
+
+### [waitForEvent()](/sdk/io/wait-for-event)
+
+`io.waitForEvent()` allows you to pause the execution of a run until an event is received, and receive the event data.
+
+### [waitForRequest()](/sdk/io/wait-for-request)
+
+`io.waitForRequest()` allows you to pause the execution of a run until the provided URL is requested, and receive the request data.
+
+### [sendEvents()](/sdk/io/sendevents)
+
+`io.sendEvents()` allows you to send multiple events from inside a Job run. The sent events will trigger any Jobs that are listening for those events (based on the name).
+
+If you want to send multiple events from outside a run (e.g. just from your backend) you can use [client.sendEvents()](/sdk/triggerclient/instancemethods/sendevents).
+
 ### [backgroundFetch()](/sdk/io/backgroundfetch)
 
 `io.backgroundFetch()` allows you to fetch data from a URL that can take longer that the serverless timeout. The actual `fetch` request is performed on the Trigger.dev platform, and the response is sent back to you. An example use case is fetching data from a slow API, like some AI endpoints.
 
+### [backgroundPoll()](/sdk/io/background-poll)
+
+`io.backgroundPoll()` allows you to fetch data from a URL on an interval, in the background.
+
 ### [random()](/sdk/io/random)
 
 `io.random()` is identical to `Math.random()` when called without options but ensures your random numbers are not regenerated on resume or retry. It will return a pseudo-random floating-point number between optional `min` (default: 0, inclusive) and `max` (default: 1, exclusive). Can optionally `round` to the nearest integer.

docs/sdk/io/sendevent.mdx

@@ -1,13 +1,15 @@
 ---
 title: "io.sendEvent()"
 sidebarTitle: "sendEvent()"
-description: "`io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name)."
+description: "`io.sendEvent()` allows you to send an event from inside a Job run. The sent event will trigger any Jobs that are listening for that event (based on the name)."
 ---
 
 If you want to send an event from outside a run (e.g. just from your backend) you should use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent) instead.
 
 Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
 
+For multiple events, use [io.sendEvents()](/sdk/io/sendevents) instead.
+
 ## Parameters
 
 <Snippet file="stable-key-param.mdx" />

docs/sdk/io/sendevents.mdx

@@ -0,0 +1,71 @@
+---
+title: "io.sendEvents()"
+sidebarTitle: "sendEvents()"
+description: "`io.sendEvents()` allows you to send multiple events from inside a Job run. The sent events will trigger any Jobs that are listening for those events (based on the name)."
+---
+
+If you want to send multiple events from outside a run (e.g. just from your backend) you should use [client.sendEvents()](/sdk/triggerclient/instancemethods/sendevents) instead.
+
+Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
+
+For single events, use [io.sendEvent()](/sdk/io/sendevent) instead.
+
+## Parameters
+
+<Snippet file="stable-key-param.mdx" />
+
+<Snippet file="send-events-params.mdx" />
+
+## Returns
+
+<Snippet file="send-events-return.mdx" />
+
+<RequestExample>
+
+```ts Send multiple events
+//this Job sends multiple events that triggers the second job
+client.defineJob({
+  id: "job-1",
+  name: "First job",
+  version: "0.0.1",
+  trigger: cronTrigger({
+    cron: "0 9 * * *", // 9am every day (UTC)
+  }),
+  run: async (payload, io, ctx) => {
+    //sends "new.user" events with a userId in the payload
+    await io.sendEvents("send-events", [
+      {
+        name: "new.user",
+        payload: {
+          userId: "u_12345",
+        },
+      },
+      {
+        name: "new.user",
+        payload: {
+          userId: "u_67890",
+        },
+      },
+    ]);
+  },
+});
+
+client.defineJob({
+  id: "job-2",
+  name: "Second job",
+  version: "0.0.1",
+  //subscribes to the "new.user" event
+  trigger: eventTrigger({
+    name: "new.user",
+    schema: z.object({
+      userId: z.string(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.log("New user created", { userId: payload.userId });
+    //do stuff with the new user
+  },
+});
+```
+
+</RequestExample>

docs/sdk/io/wait-for-event.mdx

@@ -0,0 +1,105 @@
+---
+title: "io.waitForEvent()"
+sidebarTitle: "waitForEvent()"
+description: "`io.waitForEvent()` waits for the next event to be emitted, and returns the event data"
+---
+
+<Warning>This feature is in beta and has not yet been deployed to the Trigger.dev cloud</Warning>
+
+## Parameters
+
+<Snippet file="stable-key-param.mdx" />
+
+<ResponseField name="event" type="object" required>
+  Specify the options for the event to wait for.
+
+{" "}
+
+<Expandable title="fields" defaultOpen>
+  <ResponseField name="name" type="string | string[]" required>
+    The name(s) of the event to wait for.
+  </ResponseField>
+  <ResponseField name="schema" type="ZodTypeAny">
+    An optional Zod schema to validate the event payload against. If omitted the `event.payload`
+    will be typed as `any`
+  </ResponseField>
+  <ResponseField name="filter" type="EventFilter">
+    An [EventFilter](/documentation/guides/event-filter) to match against the event payload.
+  </ResponseField>
+  <ResponseField name="source" type="string">
+    The source of the event to wait for. If omitted, the event can come from any source.
+  </ResponseField>
+  <ResponseField name="contextFilter" type="EventFilter">
+    An [EventFilter](/documentation/guides/event-filter) to match against the event context.
+  </ResponseField>
+  <ResponseField name="accountId" type="string">
+    The account ID of the event to wait for. If omitted, the event can come from any account.
+  </ResponseField>
+</Expandable>
+
+</ResponseField>
+
+<ResponseField name="options" type="object">
+  Specify the options for the event to wait for.
+
+{" "}
+
+<Expandable title="options" defaultOpen>
+  <ResponseField name="timeoutInSeconds" type="number">
+    The amount of time to wait for the event to be emitted before timing out. The default timeout is
+    1 hour and the maximum timeout is 1 year. If the timeout is reached, the task will fail with an
+    error message and the run will be exited.
+  </ResponseField>
+</Expandable>
+
+</ResponseField>
+
+## Returns
+
+<ResponseField name="id" type="string" required>
+  The ID of the event that was emitted.
+</ResponseField>
+
+<ResponseField name="name" type="string" required>
+  The name of the event that was emitted.
+</ResponseField>
+
+<ResponseField name="payload" type="any" required>
+  The payload of the event that was emitted.
+</ResponseField>
+
+<ResponseField name="context" type="any">
+  The context of the event that was emitted.
+</ResponseField>
+
+<ResponseField name="timestamp" type="Date" required>
+  The timestamp of the event that was emitted.
+</ResponseField>
+
+<ResponseField name="accountId" type="string">
+  The account ID of the event that was emitted.
+</ResponseField>
+
+<RequestExample>
+  
+```ts example.ts
+const event = await io.waitForEvent(
+  "wait",
+  {
+    name: "my.event",
+    schema: z.object({
+      id: z.string(),
+      createdAt: z.coerce.date(),
+      isAdmin: z.boolean(),
+    }),
+    filter: {
+      isAdmin: [true], // Only wait for events where isAdmin is true
+    },
+  },
+  {
+    timeoutInSeconds: 60 * 60, // Wait for up to an hour
+  }
+);
+```
+
+</RequestExample>

docs/sdk/io/wait-for-request.mdx

@@ -0,0 +1,75 @@
+---
+title: "io.waitForRequest()"
+sidebarTitle: "waitForRequest()"
+description: "`io.waitForRequest()` waits for a request to be made to the provided URL."
+---
+
+## Parameters
+
+<Snippet file="stable-key-param.mdx" />
+
+<ResponseField name="callback" type="function" required>
+  A callback function that is called with a single `url` parameter. When the URL is POSTed to, the
+  task will be completed and the POST request body will be returned.
+</ResponseField>
+
+<ResponseField name="options" type="object">
+  <Expandable title="fields" defaultOpen>
+    <ResponseField name="timeoutInSeconds" type="number">
+      The amount of time to wait for the request to be made before timing out. Defaults to 1 hour.
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+
+## Returns
+
+Returns a `Promise` that resolves to the request body when the request is made.
+
+<RequestExample>
+  
+```ts example.ts
+type ScreenshotResponse = {
+  store: {
+    location: string;
+  }
+}
+
+client.defineJob({
+  id: "screenshot-one-example",
+  name: "Screenshot One Example",
+  version: "1.0.0",
+  trigger: invokeTrigger({
+    schema: z.object({
+      url: z.string().url().default("https://trigger.dev"),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    const result = await io.waitForRequest<ScreenshotResponse>(
+      "screenshot-one",
+      async (url) => {
+        await fetch(`https://api.screenshotone.com/take`, {
+          method: "POST",
+          headers: {
+            "Content-Type": "application/json",
+          },
+          body: JSON.stringify({
+            access_key: process.env.SCREENSHOT_ONE_API_KEY,
+            url: payload.url,
+            store: "true",
+            storage_path: "my-screeshots",
+            response_type: "json",
+            async: "true",
+            webhook_url: url, // this is the URL that will be called when the screenshot is ready
+            storage_return_location: "true",
+          }),
+        });
+      },
+      {
+        timeoutInSeconds: 300,
+      }
+    );
+  },
+});
+```
+
+</RequestExample>

docs/sdk/triggerclient/instancemethods/sendevent.mdx

@@ -8,6 +8,8 @@ You can call this function from anywhere in your backend to send an event. The o
 
 Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
 
+For multiple events, use [client.sendEvents()](/sdk/triggerclient/instancemethods/sendevents) instead.
+
 ## Parameters
 
 <Snippet file="send-event-params.mdx" />

docs/sdk/triggerclient/instancemethods/sendevents.mdx

@@ -0,0 +1,83 @@
+---
+title: "TriggerClient: sendEvents() instance method"
+sidebarTitle: "sendEvents()"
+description: "The `sendEvents()` instance method send multiple events that triggers any Jobs that are listening for those events (based on the name)."
+---
+
+You can call this function from anywhere in your backend to send multiple events. The other way to send multiple events is by using [io.sendEvents()](/sdk/io/sendevents) from inside a `run()` function.
+
+Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
+
+For single events, use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent) instead.
+
+## Parameters
+
+<Snippet file="send-events-params.mdx" />
+
+## Returns
+
+<Snippet file="send-events-return.mdx" />
+
+<RequestExample>
+
+```ts Simple example with payloads
+const event = client.sendEvents([
+  {
+    name: "new.user",
+    payload: {
+      userId: "u_12345",
+    },
+  },
+  {
+    name: "new.user",
+    payload: {
+      userId: "u_67890",
+    },
+  },
+]);
+```
+
+```ts Send multiple events with an ID
+const event = client.sendEvents([
+  {
+    id: "e_12345", // You can use this to deduplicate events
+    name: "new.user",
+    payload: {
+      userId: "u_12345",
+    },
+  },
+  {
+    id: "e_67890", // You can use this to deduplicate events
+    name: "new.user",
+    payload: {
+      userId: "u_67890",
+    },
+  },
+]);
+```
+
+```ts Send multiple events to be delivered later
+const event = client.sendEvents(
+  [
+    {
+      id: "e_12345", // You can use this to deduplicate events
+      name: "new.user",
+      payload: {
+        userId: "u_12345",
+      },
+    },
+    {
+      id: "e_67890", // You can use this to deduplicate events
+      name: "new.user",
+      payload: {
+        userId: "u_67890",
+      },
+    },
+  ],
+  {
+    deliverAt: new Date("2023-12-01T00:00:00.000Z"),
+  }
+);
+```
+
+</RequestExample>

docs/sdk/triggerclient/overview.mdx

@@ -38,6 +38,12 @@ Sending an event triggers any Jobs that are listening for that event (based on t
 
 You can call this function from anywhere in your code to send an event. The other way to send an event is by using [io.sendEvent()](/sdk/io) from inside a `run()` function.
 
+#### [sendEvents()](/sdk/triggerclient/instancemethods/sendevents)
+
+Sending multiple events triggers any Jobs that are listening for those events (based on the name). Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
+
+You can call this function from anywhere in your code to send multiple events. The other way to send multiple events is by using [io.sendEvents()](/sdk/io) from inside a `run()` function.
+
 #### [getEvent()](/sdk/triggerclient/instancemethods/getevent)
 
 The `getEvent()` method gets the event details for a given eventId.

integrations/airtable/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/airtable
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/airtable/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/airtable",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "Trigger.dev integration for airtable",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "airtable": "^0.12.1",
     "zod": "3.22.3"
   },

integrations/github/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/github
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/github/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/github",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "The official GitHub integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -29,8 +29,8 @@
     "@octokit/request": "^6.2.5",
     "@octokit/request-error": "^4.0.1",
     "@octokit/webhooks": "^10.4.0",
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "octokit": "^2.0.14",
     "zod": "3.22.3"
   },

integrations/linear/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/linear
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/linear/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/linear",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "Trigger.dev integration for @linear/sdk",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -27,8 +27,8 @@
   },
   "dependencies": {
     "@linear/sdk": "^8.0.0",
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "zod": "3.22.3"
   },
   "engines": {

integrations/openai/CHANGELOG.md

@@ -1,5 +1,18 @@
 # @trigger.dev/slack
 
+## 2.2.6
+
+### Patch Changes
+
+- cb1825bf: OpenAI support for 4.16.0
+- cb1825bf: Add support for background polling and use that in OpenAI integration to power assistants
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/openai/jest.config.js

@@ -0,0 +1,8 @@
+module.exports = {
+  moduleFileExtensions: ["ts", "tsx", "js"],
+  transform: {
+    "^.+\\.(ts|tsx)$": "ts-jest",
+  },
+  testMatch: ["<rootDir>/test/**/*.ts?(x)", "<rootDir>/test/**/?(*.)+(spec|test).ts?(x)"],
+  testEnvironment: "node",
+};

integrations/openai/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/openai",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "The official OpenAI integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -17,18 +17,22 @@
     "@types/node": "18",
     "rimraf": "^3.0.2",
     "tsup": "^6.5.0",
-    "typescript": "^4.9.4"
+    "typescript": "^4.9.4",
+    "@types/jest": "^29.5.3",
+    "jest": "^29.6.2",
+    "ts-jest": "^29.1.1"
   },
   "scripts": {
     "clean": "rimraf dist",
     "build": "npm run clean && npm run build:tsup",
     "build:tsup": "tsup",
-    "typecheck": "tsc --noEmit"
+    "typecheck": "tsc --noEmit",
+    "test": "jest"
   },
   "dependencies": {
-    "openai": "^4.13.0",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
-    "@trigger.dev/integration-kit": "workspace:^2.2.5"
+    "openai": "^4.16.1",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6"
   },
   "engines": {
     "node": ">=18.0.0"

integrations/openai/src/assistants.ts

@@ -0,0 +1,57 @@
+import { IntegrationTaskKey, Prettify } from "@trigger.dev/sdk";
+import { OpenAIRunTask } from "./index";
+import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
+import OpenAI from "openai";
+import { createTaskOutputProperties, handleOpenAIError } from "./taskUtils";
+
+export class Assistants {
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
+
+  async create(
+    key: IntegrationTaskKey,
+    params: Prettify<OpenAI.Beta.AssistantCreateParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Assistant> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.assistants
+          .create(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        const outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        task.outputProperties = [
+          ...(outputProperties ?? []),
+          {
+            label: "assistantId",
+            text: data.id,
+          },
+        ];
+
+        return data;
+      },
+      {
+        name: "Create Assistant",
+        params,
+        properties: [
+          {
+            label: "model",
+            text: params.model,
+          },
+          ...(params.name ? [{ label: "name", text: params.name }] : []),
+          ...(params.file_ids && params.file_ids.length > 0
+            ? [{ label: "files", text: params.file_ids.join(", ") }]
+            : []),
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+}

integrations/openai/src/beta.ts

@@ -0,0 +1,19 @@
+import { Assistants } from "./assistants";
+import { OpenAIRunTask } from "./index";
+import { Threads } from "./threads";
+import { OpenAIIntegrationOptions } from "./types";
+
+export class Beta {
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
+
+  get assistants() {
+    return new Assistants(this.runTask.bind(this), this.options);
+  }
+
+  get threads() {
+    return new Threads(this.runTask.bind(this), this.options);
+  }
+}

integrations/openai/src/chat.ts

@@ -5,7 +5,8 @@ import {
   backgroundTaskRetries,
   createBackgroundFetchHeaders,
   createBackgroundFetchUrl,
-  createTaskUsageProperties,
+  createTaskOutputProperties,
+  handleOpenAIError,
 } from "./taskUtils";
 import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
 import { FetchRetryOptions, FetchTimeoutOptions } from "@trigger.dev/integration-kit";
@@ -25,12 +26,16 @@ export class Chat {
       return this.runTask(
         key,
         async (client, task) => {
-          const response = await client.chat.completions.create(params, {
-            idempotencyKey: task.idempotencyKey,
-            ...options,
-          });
-          task.outputProperties = createTaskUsageProperties(response.usage);
-          return response;
+          const { data, response } = await client.chat.completions
+            .create(params, {
+              idempotencyKey: task.idempotencyKey,
+              ...options,
+            })
+            .withResponse();
+
+          task.outputProperties = createTaskOutputProperties(data.usage, response.headers);
+
+          return data;
         },
         {
           name: "Chat Completion",
@@ -41,7 +46,8 @@ export class Chat {
               text: params.model,
             },
           ],
-        }
+        },
+        handleOpenAIError
       );
     },
 
@@ -61,7 +67,7 @@ export class Chat {
             options
           );
 
-          const response = await io.backgroundFetch<OpenAI.Chat.ChatCompletion>(
+          const response = await io.backgroundFetchResponse<OpenAI.Chat.ChatCompletion>(
             "background",
             url,
             {
@@ -74,13 +80,18 @@ export class Chat {
               ),
               body: JSON.stringify(params),
             },
-            fetchOptions.retries ?? backgroundTaskRetries,
-            fetchOptions.timeout
+            {
+              retry: fetchOptions?.retries ?? backgroundTaskRetries,
+              timeout: fetchOptions?.timeout,
+            }
           );
 
-          task.outputProperties = createTaskUsageProperties(response.usage);
+          task.outputProperties = createTaskOutputProperties(
+            response.data.usage,
+            new Headers(response.headers)
+          );
 
-          return response;
+          return response.data;
         },
         {
           name: "Background Chat Completion",

integrations/openai/src/completions.ts

@@ -5,7 +5,8 @@ import {
   backgroundTaskRetries,
   createBackgroundFetchHeaders,
   createBackgroundFetchUrl,
-  createTaskUsageProperties,
+  createTaskOutputProperties,
+  handleOpenAIError,
 } from "./taskUtils";
 import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
 import { FetchRetryOptions, FetchTimeoutOptions } from "@trigger.dev/integration-kit";
@@ -24,12 +25,16 @@ export class Completions {
     return this.runTask(
       key,
       async (client, task) => {
-        const response = await client.completions.create(params, {
-          idempotencyKey: task.idempotencyKey,
-          ...options,
-        });
-        task.outputProperties = createTaskUsageProperties(response.usage);
-        return response;
+        const { data, response } = await client.completions
+          .create(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(data.usage, response.headers);
+
+        return data;
       },
       {
         name: "Completion",
@@ -40,7 +45,8 @@ export class Completions {
             text: params.model,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -60,7 +66,7 @@ export class Completions {
           options
         );
 
-        const response = await io.backgroundFetch<OpenAI.Completion>(
+        const response = await io.backgroundFetchResponse<OpenAI.Completion>(
           "background",
           url,
           {
@@ -73,13 +79,18 @@ export class Completions {
             ),
             body: JSON.stringify(params),
           },
-          fetchOptions.retries ?? backgroundTaskRetries,
-          fetchOptions.timeout
+          {
+            retry: fetchOptions?.retries ?? backgroundTaskRetries,
+            timeout: fetchOptions?.timeout,
+          }
         );
 
-        task.outputProperties = createTaskUsageProperties(response.usage);
+        task.outputProperties = createTaskOutputProperties(
+          response.data.usage,
+          new Headers(response.headers)
+        );
 
-        return response;
+        return response.data;
       },
       {
         name: "Background Completion",

integrations/openai/src/edits.ts

@@ -2,7 +2,7 @@ import { truncate } from "@trigger.dev/integration-kit";
 import { IntegrationTaskKey, Prettify } from "@trigger.dev/sdk";
 import OpenAI from "openai";
 import { OpenAIRunTask } from "./index";
-import { createTaskUsageProperties } from "./taskUtils";
+import { createTaskOutputProperties, handleOpenAIError } from "./taskUtils";
 import { OpenAIRequestOptions } from "./types";
 
 export class Edits {
@@ -42,18 +42,21 @@ export class Edits {
     return this.runTask(
       key,
       async (client, task) => {
-        const response = await client.edits.create(params, {
-          idempotencyKey: task.idempotencyKey,
-          ...options,
-        });
-        task.outputProperties = createTaskUsageProperties(response.usage);
-        return response;
+        const { data, response } = await client.edits
+          .create(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+        task.outputProperties = createTaskOutputProperties(data.usage, response.headers);
+        return data;
       },
       {
         name: "Create edit",
         params,
         properties,
-      }
+      },
+      handleOpenAIError
     );
   }
 }

integrations/openai/src/embeddings.ts

@@ -1,7 +1,7 @@
 import { IntegrationTaskKey } from "@trigger.dev/sdk";
 import OpenAI from "openai";
 import { OpenAIRunTask } from "./index";
-import { createTaskUsageProperties } from "./taskUtils";
+import { createTaskOutputProperties, handleOpenAIError } from "./taskUtils";
 import { OpenAIRequestOptions } from "./types";
 
 export class Embeddings {
@@ -19,12 +19,14 @@ export class Embeddings {
     return this.runTask(
       key,
       async (client, task) => {
-        const response = await client.embeddings.create(params, {
-          idempotencyKey: task.idempotencyKey,
-          ...options,
-        });
-        task.outputProperties = createTaskUsageProperties(response.usage);
-        return response;
+        const { data, response } = await client.embeddings
+          .create(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+        task.outputProperties = createTaskOutputProperties(data.usage, response.headers);
+        return data;
       },
       {
         name: "Create embedding",
@@ -35,7 +37,8 @@ export class Embeddings {
             text: params.model,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 }

integrations/openai/src/files.ts

@@ -2,12 +2,19 @@ import { fileFromString } from "@trigger.dev/integration-kit";
 import { IntegrationTaskKey } from "@trigger.dev/sdk";
 import OpenAI from "openai";
 import { OpenAIRunTask } from "./index";
-import { OpenAIRequestOptions } from "./types";
+import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
+import {
+  createBackgroundFetchHeaders,
+  createBackgroundFetchUrl,
+  createTaskOutputProperties,
+  handleOpenAIError,
+} from "./taskUtils";
+import { Uploadable } from "openai/uploads";
 
 type CreateFileRequest = {
-  file: string | File;
+  file: string | File | Uploadable;
   fileName?: string;
-  purpose: string;
+  purpose: "fine-tune" | "assistants";
 };
 
 type CreateFineTuneFileRequest = {
@@ -19,11 +26,10 @@ type CreateFineTuneFileRequest = {
 };
 
 export class Files {
-  runTask: OpenAIRunTask;
-
-  constructor(runTask: OpenAIRunTask) {
-    this.runTask = runTask;
-  }
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
 
   create(
     key: IntegrationTaskKey,
@@ -33,7 +39,7 @@ export class Files {
     return this.runTask(
       key,
       async (client, task) => {
-        let file: File;
+        let file: Uploadable;
 
         if (typeof params.file === "string") {
           file = await fileFromString(params.file, params.fileName ?? "file.txt");
@@ -59,24 +65,172 @@ export class Files {
             text: typeof params.file === "string" ? "string" : "File",
           },
         ],
-      }
+      },
+      handleOpenAIError
+    );
+  }
+
+  async createAndWaitForProcessing(
+    key: IntegrationTaskKey,
+    params: CreateFileRequest,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Files.FileObject> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        let file: Uploadable;
+
+        if (typeof params.file === "string") {
+          file = await fileFromString(params.file, params.fileName ?? "file.txt");
+        } else {
+          file = params.file;
+        }
+
+        const { data, response } = await client.files
+          .create(
+            { file, purpose: params.purpose },
+            { idempotencyKey: task.idempotencyKey, ...options }
+          )
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        if (["processed", "error", "deleted"].includes(data.status)) {
+          return data;
+        }
+
+        const url = createBackgroundFetchUrl(
+          client,
+          `/files/${data.id}`,
+          this.options.defaultQuery,
+          options
+        );
+
+        const headers = this.options.defaultHeaders ?? {};
+
+        const processedFile = await io.backgroundPoll<OpenAI.Files.FileObject>("poll", {
+          url,
+          requestInit: {
+            headers: createBackgroundFetchHeaders(client, task.idempotencyKey, headers, options),
+          },
+          interval: 10,
+          timeout: 600,
+          responseFilter: {
+            status: [200],
+            body: {
+              status: ["processed", "error", "deleted"],
+            },
+          },
+        });
+
+        return processedFile;
+      },
+      {
+        name: "Create file and wait for processing",
+        params,
+        properties: [
+          {
+            label: "Purpose",
+            text: params.purpose,
+          },
+          {
+            label: "Input type",
+            text: typeof params.file === "string" ? "string" : "File",
+          },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  async waitForProcessing(
+    key: IntegrationTaskKey,
+    id: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Files.FileObject> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const url = createBackgroundFetchUrl(
+          client,
+          `/files/${id}`,
+          this.options.defaultQuery,
+          options
+        );
+
+        const headers = this.options.defaultHeaders ?? {};
+
+        const processedFile = await io.backgroundPoll<OpenAI.Files.FileObject>("poll", {
+          url,
+          requestInit: {
+            headers: createBackgroundFetchHeaders(client, task.idempotencyKey, headers, options),
+          },
+          interval: 10,
+          timeout: 600,
+          responseFilter: {
+            status: [200],
+            body: {
+              status: ["processed", "error", "deleted"],
+            },
+          },
+        });
+
+        return processedFile;
+      },
+      {
+        name: "Wait for processing",
+        properties: [
+          {
+            label: "fileId",
+            text: id,
+          },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  retrieve(
+    key: IntegrationTaskKey,
+    id: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Files.FileObject> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const response = await client.files.retrieve(id, options);
+        return response;
+      },
+      {
+        name: "Retrieve file",
+        properties: [
+          {
+            label: "fileId",
+            text: id,
+          },
+        ],
+      },
+      handleOpenAIError
     );
   }
 
   list(
     key: IntegrationTaskKey,
+    query?: OpenAI.Files.FileListParams,
     options: OpenAIRequestOptions = {}
   ): Promise<OpenAI.Files.FileObject[]> {
     return this.runTask(
       key,
       async (client, task) => {
-        const response = await client.files.list(options);
+        const response = await client.files.list(query, options);
+
         return response.data;
       },
       {
         name: "List files",
         properties: [],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -107,7 +261,8 @@ export class Files {
             text: params.examples.length.toString(),
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 }

integrations/openai/src/fineTunes.ts

@@ -2,6 +2,7 @@ import { IntegrationTaskKey } from "@trigger.dev/sdk";
 import OpenAI from "openai";
 import { OpenAIRunTask } from "./index";
 import { OpenAIRequestOptions } from "./types";
+import { handleOpenAIError } from "./taskUtils";
 
 type SpecificFineTuneRequest = {
   fineTuneId: string;
@@ -49,7 +50,8 @@ export class FineTunes {
         name: "Create fine tune",
         params,
         properties,
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -66,7 +68,8 @@ export class FineTunes {
       {
         name: "List fine tunes",
         properties: [],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -89,7 +92,8 @@ export class FineTunes {
             text: params.fineTuneId,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -115,7 +119,8 @@ export class FineTunes {
             text: params.fineTuneId,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -138,7 +143,8 @@ export class FineTunes {
             text: params.fineTuneId,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -189,7 +195,8 @@ export class FineTunes {
           name: "Create Fine Tuning Job",
           params,
           properties,
-        }
+        },
+        handleOpenAIError
       );
     },
 
@@ -212,7 +219,8 @@ export class FineTunes {
               text: params.id,
             },
           ],
-        }
+        },
+        handleOpenAIError
       );
     },
 
@@ -238,7 +246,8 @@ export class FineTunes {
               text: params.id,
             },
           ],
-        }
+        },
+        handleOpenAIError
       );
     },
 
@@ -262,7 +271,8 @@ export class FineTunes {
               text: params.id,
             },
           ],
-        }
+        },
+        handleOpenAIError
       );
     },
 
@@ -281,7 +291,8 @@ export class FineTunes {
         {
           name: "List Fine Tuning Jobs",
           params,
-        }
+        },
+        handleOpenAIError
       );
     },
   };

integrations/openai/src/images.ts

@@ -1,33 +1,43 @@
-import { fileFromUrl } from "@trigger.dev/integration-kit";
+import { FetchRetryOptions, FetchTimeoutOptions, fileFromUrl } from "@trigger.dev/integration-kit";
 import { IntegrationTaskKey, Prettify } from "@trigger.dev/sdk";
 import OpenAI from "openai";
 import { OpenAIRunTask } from "./index";
-import { OpenAIRequestOptions } from "./types";
+import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
+import {
+  backgroundTaskRetries,
+  createBackgroundFetchHeaders,
+  createBackgroundFetchUrl,
+  createImageTaskOutputProperties,
+  createTaskOutputProperties,
+  handleOpenAIError,
+} from "./taskUtils";
+import { Uploadable } from "openai/uploads";
 
 export type CreateImageEditRequest = {
-  image: string | File;
+  image: string | File | Uploadable;
   prompt: string;
-  mask?: string | File;
+  mask?: string | File | Uploadable;
   n?: number;
   size?: "256x256" | "512x512" | "1024x1024";
   response_format?: "url" | "b64_json";
   user?: string;
+  model?: (string & {}) | "dall-e-2" | null;
 };
 
 export type CreateImageVariationRequest = {
-  image: string | File;
+  image: string | File | Uploadable;
   n?: number;
   size?: "256x256" | "512x512" | "1024x1024";
   response_format?: "url" | "b64_json";
   user?: string;
+  model?: (string & {}) | "dall-e-2" | null;
 };
 
 export class Images {
-  runTask: OpenAIRunTask;
-
-  constructor(runTask: OpenAIRunTask) {
-    this.runTask = runTask;
-  }
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
 
   generate(
     key: IntegrationTaskKey,
@@ -65,16 +75,89 @@ export class Images {
     return this.runTask(
       key,
       async (client, task) => {
-        return client.images.generate(params, { idempotencyKey: task.idempotencyKey, ...options });
+        const { data, response } = await client.images
+          .generate(params, { idempotencyKey: task.idempotencyKey, ...options })
+          .withResponse();
+
+        task.outputProperties = createImageTaskOutputProperties(data, response.headers);
+
+        return data;
       },
       {
         name: "Create image",
         params,
         properties,
+      },
+      handleOpenAIError
+    );
+  }
+
+  backgroundGenerate(
+    key: IntegrationTaskKey,
+    params: Prettify<OpenAI.Images.ImageGenerateParams>,
+    options: OpenAIRequestOptions = {},
+    fetchOptions: { retries?: FetchRetryOptions; timeout?: FetchTimeoutOptions } = {}
+  ): Promise<OpenAI.Images.ImagesResponse> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const url = createBackgroundFetchUrl(
+          client,
+          "/images/generations",
+          this.options.defaultQuery,
+          options
+        );
+
+        const response = await io.backgroundFetchResponse<OpenAI.Images.ImagesResponse>(
+          "background",
+          url,
+          {
+            method: options.method ?? "POST",
+            headers: createBackgroundFetchHeaders(
+              client,
+              task.idempotencyKey,
+              this.options.defaultHeaders,
+              options
+            ),
+            body: JSON.stringify(params),
+          },
+          {
+            retry: fetchOptions?.retries ?? backgroundTaskRetries,
+            timeout: fetchOptions?.timeout,
+          }
+        );
+
+        task.outputProperties = createImageTaskOutputProperties(
+          response.data,
+          new Headers(response.headers)
+        );
+
+        return response.data;
+      },
+      {
+        name: "Background Image Generate",
+        params,
+        properties: [
+          {
+            label: "model",
+            text: params.model ?? "unknown",
+          },
+        ],
+        retry: {
+          limit: 0,
+        },
       }
     );
   }
 
+  create(...args: Parameters<Images["generate"]>) {
+    return this.generate(...args);
+  }
+
+  backgroundCreate(...args: Parameters<Images["backgroundGenerate"]>) {
+    return this.backgroundGenerate(...args);
+  }
+
   edit(
     key: IntegrationTaskKey,
     params: CreateImageEditRequest,
@@ -87,6 +170,13 @@ export class Images {
       text: params.prompt,
     });
 
+    if (typeof params.model === "string") {
+      properties.push({
+        label: "model",
+        text: params.model,
+      });
+    }
+
     if (params.n) {
       properties.push({
         label: "Number of images",
@@ -123,26 +213,32 @@ export class Images {
           typeof params.image === "string" ? await fileFromUrl(params.image) : params.image;
         const mask = typeof params.mask === "string" ? await fileFromUrl(params.mask) : params.mask;
 
-        const response = await client.images.edit(
-          {
-            image: file,
-            prompt: params.prompt,
-            mask: mask,
-            n: params.n,
-            size: params.size,
-            response_format: params.response_format,
-            user: params.user,
-          },
-          { idempotencyKey: task.idempotencyKey, ...options }
-        );
+        const { data, response } = await client.images
+          .edit(
+            {
+              image: file,
+              prompt: params.prompt,
+              mask: mask,
+              n: params.n,
+              size: params.size,
+              response_format: params.response_format,
+              user: params.user,
+              model: params.model,
+            },
+            { idempotencyKey: task.idempotencyKey, ...options }
+          )
+          .withResponse();
 
-        return response;
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
       },
       {
         name: "Create image edit",
         params,
         properties,
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -153,6 +249,13 @@ export class Images {
   ): Promise<OpenAI.Images.ImagesResponse> {
     let properties = [];
 
+    if (typeof params.model === "string") {
+      properties.push({
+        label: "model",
+        text: params.model,
+      });
+    }
+
     if (params.n) {
       properties.push({
         label: "Number of images",
@@ -188,18 +291,23 @@ export class Images {
         const file =
           typeof params.image === "string" ? await fileFromUrl(params.image) : params.image;
 
-        const response = await client.images.createVariation(
-          {
-            image: file,
-            n: params.n,
-            size: params.size,
-            response_format: params.response_format,
-            user: params.user,
-          },
-          { idempotencyKey: task.idempotencyKey, ...options }
-        );
+        const { data, response } = await client.images
+          .createVariation(
+            {
+              image: file,
+              n: params.n,
+              size: params.size,
+              response_format: params.response_format,
+              user: params.user,
+              model: params.model,
+            },
+            { idempotencyKey: task.idempotencyKey, ...options }
+          )
+          .withResponse();
 
-        return response;
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
       },
       {
         name: "Create image variation",

integrations/openai/src/index.ts

@@ -19,6 +19,7 @@ import { FineTunes } from "./fineTunes";
 import { Images } from "./images";
 import { Models } from "./models";
 import { OpenAIIntegrationOptions } from "./types";
+import { Beta } from "./beta";
 
 export type OpenAIRunTask = InstanceType<typeof OpenAI>["runTask"];
 
@@ -103,7 +104,10 @@ export class OpenAI implements TriggerIntegration {
     options?: RunTaskOptions,
     errorCallback?: RunTaskErrorCallback
   ): Promise<TResult> {
-    if (!this._io) throw new Error("No IO");
+    if (!this._io)
+      throw new Error(
+        "Issue with running task: IO not found. It's possible that you forgot to prefix openai with io. inside a run"
+      );
     if (!this._connectionKey) throw new Error("No connection key");
     return this._io.runTask(
       key,
@@ -113,7 +117,7 @@ export class OpenAI implements TriggerIntegration {
       },
       {
         icon: this._options.icon ?? "openai",
-        retry: retry.standardBackoff,
+        retry: retry.exponentialBackoff,
         ...(options ?? {}),
         connectionKey: this._connectionKey,
       },
@@ -129,6 +133,10 @@ export class OpenAI implements TriggerIntegration {
     return new Completions(this.runTask.bind(this), this._options);
   }
 
+  get beta() {
+    return new Beta(this.runTask.bind(this), this._options);
+  }
+
   get chat() {
     return new Chat(this.runTask.bind(this), this._options);
   }
@@ -138,7 +146,7 @@ export class OpenAI implements TriggerIntegration {
   }
 
   get images() {
-    return new Images(this.runTask.bind(this));
+    return new Images(this.runTask.bind(this), this._options);
   }
 
   get embeddings() {
@@ -146,18 +154,45 @@ export class OpenAI implements TriggerIntegration {
   }
 
   get files() {
-    return new Files(this.runTask.bind(this));
+    return new Files(this.runTask.bind(this), this._options);
   }
 
   get fineTunes() {
+    return this.fineTuning;
+  }
+
+  get fineTuning() {
     return new FineTunes(this.runTask.bind(this));
   }
 
+  /**
+   * @deprecated Please use openai.models.retrieve instead
+   */
   retrieveModel = this.models.retrieve;
+
+  /**
+   * @deprecated Please use openai.models.list instead
+   */
   listModels = this.models.list;
+
+  /**
+   * @deprecated Please use openai.models.delete instead
+   */
   deleteModel = this.models.delete;
+
+  /**
+   * @deprecated Please use openai.models.delete instead
+   */
   deleteFineTune = this.models.delete;
+
+  /**
+   * @deprecated Please use openai.completions.create instead
+   */
   createCompletion = this.completions.create;
+
+  /**
+   * @deprecated Please use openai.chat.completions.create instead
+   */
   createChatCompletion = this.chat.completions.create;
 
   /**
@@ -176,19 +211,82 @@ export class OpenAI implements TriggerIntegration {
     return this.chat.completions.backgroundCreate(...args);
   }
 
+  /**
+   * @deprecated Please use openai.edits.create instead
+   */
   createEdit = this.edits.create;
-  generateImage = this.images.generate;
-  createImage = this.images.generate;
-  createImageEdit = this.images.edit;
-  createImageVariation = this.images.createVariation;
+
+  /**
+   * @deprecated Please use openai.images.generate instead
+   */
+  async generateImage(...args: Parameters<typeof this.images.generate>) {
+    return this.images.generate(...args);
+  }
+
+  /**
+   * @deprecated Please use openai.images.create instead
+   */
+  async createImage(...args: Parameters<typeof this.images.generate>) {
+    return this.images.generate(...args);
+  }
+
+  /**
+   * @deprecated Please use openai.images.edit instead
+   */
+  async createImageEdit(...args: Parameters<typeof this.images.edit>) {
+    return this.images.edit(...args);
+  }
+
+  /**
+   * @deprecated Please use openai.images.createVariation instead
+   */
+  async createImageVariation(...args: Parameters<typeof this.images.createVariation>) {
+    return this.images.createVariation(...args);
+  }
+
+  /**
+   * @deprecated Please use openai.embeddings.create instead
+   */
   createEmbedding = this.embeddings.create;
+
+  /**
+   * @deprecated Please use openai.files.create instead
+   */
   createFile = this.files.create;
+
+  /**
+   * @deprecated Please use openai.files.list instead
+   */
   listFiles = this.files.list;
+
+  /**
+   * @deprecated Please use openai.files.create instead
+   */
   createFineTuneFile = this.files.createFineTune;
+
+  /**
+   * @deprecated Please use openai.fineTuning.create instead
+   */
   createFineTune = this.fineTunes.create;
+
+  /**
+   * @deprecated Please use openai.fineTuning.list instead
+   */
   listFineTunes = this.fineTunes.list;
+
+  /**
+   * @deprecated Please use openai.fineTuning.retrieve instead
+   */
   retrieveFineTune = this.fineTunes.retrieve;
+
+  /**
+   * @deprecated Please use openai.fineTuning.cancel instead
+   */
   cancelFineTune = this.fineTunes.cancel;
+
+  /**
+   * @deprecated Please use openai.fineTuning.listEvents instead
+   */
   listFineTuneEvents = this.fineTunes.listEvents;
 
   /**
@@ -198,10 +296,28 @@ export class OpenAI implements TriggerIntegration {
    * of the fine-tuned models once complete.
    *
    * [Learn more about fine-tuning](https://platform.openai.com/docs/guides/fine-tuning)
+   *
+   * @deprecated Please use openai.fineTuning.jobs.create instead
    */
   createFineTuningJob = this.fineTunes.jobs.create;
+
+  /**
+   * @deprecated Please use openai.fineTuning.jobs.retrieve instead
+   */
   retrieveFineTuningJob = this.fineTunes.jobs.retrieve;
+
+  /**
+   * @deprecated Please use openai.fineTuning.jobs.cancel instead
+   */
   cancelFineTuningJob = this.fineTunes.jobs.cancel;
+
+  /**
+   * @deprecated Please use openai.fineTuning.jobs.listEvents instead
+   */
   listFineTuningJobEvents = this.fineTunes.jobs.listEvents;
+
+  /**
+   * @deprecated Please use openai.fineTuning.jobs.list instead
+   */
   listFineTuningJobs = this.fineTunes.jobs.list;
 }

integrations/openai/src/models.ts

@@ -3,6 +3,7 @@ import { Model } from "openai/resources";
 import { OpenAIRunTask } from "./index";
 import OpenAI from "openai";
 import { OpenAIRequestOptions } from "./types";
+import { handleOpenAIError } from "./taskUtils";
 
 type DeleteFineTunedModelRequest = {
   fineTunedModelId: string;
@@ -33,7 +34,8 @@ export class Models {
             text: params.model,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -47,7 +49,8 @@ export class Models {
       {
         name: "List models",
         properties: [],
-      }
+      },
+      handleOpenAIError
     );
   }
 
@@ -70,7 +73,8 @@ export class Models {
             text: params.fineTunedModelId,
           },
         ],
-      }
+      },
+      handleOpenAIError
     );
   }
 }

integrations/openai/src/taskUtils.ts

@@ -1,12 +1,56 @@
-import OpenAI from "openai";
+import OpenAI, { APIError } from "openai";
 import { OpenAIRequestOptions } from "./types";
 import { redactString } from "@trigger.dev/sdk";
+import { calculateResetAtUtil } from "@trigger.dev/integration-kit";
+import { FetchRetryOptions } from "@trigger.dev/integration-kit";
 
-export function createTaskUsageProperties(
+export function createImageTaskOutputProperties(
+  response: OpenAI.ImagesResponse | undefined,
+  headers?: Headers | undefined
+) {
+  if (!response && !headers) {
+    return;
+  }
+
+  return [...createTaskImageProperties(response), ...createTaskRateLimitProperties(headers)];
+}
+
+function createTaskImageProperties(response: OpenAI.ImagesResponse | undefined) {
+  if (!response) {
+    return [];
+  }
+
+  const imageUrls = response.data.map((image) => image.url).filter(Boolean) as string[];
+
+  if (imageUrls.length === 0) {
+    return [];
+  }
+
+  return [
+    {
+      label: "Images",
+      text: imageUrls[0],
+      imageUrl: imageUrls,
+    },
+  ];
+}
+
+export function createTaskOutputProperties(
+  usage: OpenAI.Completions.CompletionUsage | OpenAI.CreateEmbeddingResponse.Usage | undefined,
+  headers?: Headers | undefined
+) {
+  if (!usage && !headers) {
+    return;
+  }
+
+  return [...createTaskUsageProperties(usage), ...createTaskRateLimitProperties(headers)];
+}
+
+function createTaskUsageProperties(
   usage: OpenAI.Completions.CompletionUsage | OpenAI.CreateEmbeddingResponse.Usage | undefined
 ) {
   if (!usage) {
-    return;
+    return [];
   }
 
   return [
@@ -22,15 +66,117 @@ export function createTaskUsageProperties(
           },
         ]
       : []),
-    {
-      label: "Total Usage",
-      text: String(usage.total_tokens),
-    },
   ];
 }
 
-export function onTaskError(error: unknown) {
-  return;
+function createTaskRateLimitProperties(headers: Headers | undefined) {
+  if (!headers) {
+    return [];
+  }
+
+  const remainingRequests = headers.get("x-ratelimit-remaining-requests");
+  const remainingTokens = headers.get("x-ratelimit-remaining-tokens");
+
+  const resetRequests = headers.get("x-ratelimit-reset-requests");
+  const resetTokens = headers.get("x-ratelimit-reset-tokens");
+
+  return [
+    ...(remainingRequests
+      ? [
+          {
+            label: "Remaining Requests",
+            text: remainingRequests ?? "Unknown",
+          },
+        ]
+      : []),
+    ...(resetRequests
+      ? [
+          {
+            label: "Reset Requests",
+            text: resetRequests ?? "Unknown",
+          },
+        ]
+      : []),
+    ...(remainingTokens
+      ? [
+          {
+            label: "Remaining Tokens",
+            text: remainingTokens ?? "Unknown",
+          },
+        ]
+      : []),
+    ...(resetTokens
+      ? [
+          {
+            label: "Reset Tokens",
+            text: resetTokens ?? "Unknown",
+          },
+        ]
+      : []),
+  ];
+}
+
+export function handleOpenAIError(error: unknown) {
+  if (error instanceof APIError) {
+    const isErrorRetryable = () => {
+      if (typeof error.status !== "number") {
+        return false;
+      }
+
+      if (error.status === 429 && error.type === "insufficient_quota") {
+        return false;
+      }
+
+      return (
+        error.status === 429 ||
+        error.status === 408 ||
+        error.status === 409 ||
+        (error.status >= 500 && error.status <= 599)
+      );
+    };
+
+    const calculateRetryAt = () => {
+      if (error.status !== 429) {
+        return;
+      }
+
+      if (!error.headers) {
+        return;
+      }
+
+      const remainingRequests = error.headers["x-ratelimit-remaining-requests"];
+      const requestResets = error.headers["x-ratelimit-reset-requests"];
+
+      if (typeof remainingRequests === "string" && Number(remainingRequests) === 0) {
+        return calculateResetAt(requestResets);
+      }
+
+      const remainingTokens = error.headers["x-ratelimit-remaining-tokens"];
+      const tokensResets = error.headers["x-ratelimit-reset-tokens"];
+
+      if (typeof remainingTokens === "string" && Number(remainingTokens) === 0) {
+        return calculateResetAt(tokensResets);
+      }
+    };
+
+    return {
+      error,
+      skipRetrying: !isErrorRetryable(),
+      retryAt: calculateRetryAt(),
+    };
+  }
+
+  return error as Error;
+}
+
+// This takes a string in the format of 1s or 6m59s, 1h6m18s and calculates the date
+// If the string is invalid, it returns undefined
+// If the string is null or undefined, it returns undefined
+export function calculateResetAt(
+  resets: string | null | undefined,
+  now: Date = new Date()
+): Date | undefined {
+  return calculateResetAtUtil(resets, "iso_8601_duration_openai_variant", now);
 }
 
 export function createBackgroundFetchUrl(
@@ -83,29 +229,56 @@ export function createBackgroundFetchHeaders(
   };
 }
 
-export const backgroundTaskRetries = {
+export const backgroundTaskRetries: FetchRetryOptions = {
   "500-599": {
     strategy: "backoff",
     limit: 5,
     minTimeoutInMs: 1000,
     maxTimeoutInMs: 30000,
-    factor: 1.8,
+    factor: 2,
     randomize: true,
   },
   "429": {
     strategy: "backoff",
-    limit: 10,
-    minTimeoutInMs: 1000,
-    maxTimeoutInMs: 60000,
-    factor: 2,
-    randomize: true,
+    limit: 0,
+    bodyFilter: {
+      error: {
+        code: ["insufficient_quota"],
+      },
+    },
+  },
+  "429,429": {
+    strategy: "headers",
+    limitHeader: "x-ratelimit-limit-requests",
+    remainingHeader: "x-ratelimit-remaining-requests",
+    resetHeader: "x-ratelimit-reset-requests",
+    resetFormat: "iso_8601_duration_openai_variant",
+    bodyFilter: {
+      error: {
+        code: ["rate_limit_exceeded"],
+        type: ["requests"],
+      },
+    },
+  },
+  "429,429,429": {
+    strategy: "headers",
+    limitHeader: "x-ratelimit-limit-tokens",
+    remainingHeader: "x-ratelimit-remaining-tokens",
+    resetHeader: "x-ratelimit-reset-tokens",
+    resetFormat: "iso_8601_duration_openai_variant",
+    bodyFilter: {
+      error: {
+        code: ["rate_limit_exceeded"],
+        type: ["tokens"],
+      },
+    },
   },
   "408-409": {
     strategy: "backoff",
-    limit: 3,
+    limit: 5,
     minTimeoutInMs: 1000,
     maxTimeoutInMs: 60000,
     factor: 2,
     randomize: true,
   },
-} as const;
+};

integrations/openai/src/threads.ts

@@ -0,0 +1,691 @@
+import { IntegrationTaskKey, Prettify } from "@trigger.dev/sdk";
+import { OpenAIRunTask } from "./index";
+import { OpenAIIntegrationOptions, OpenAIRequestOptions } from "./types";
+import OpenAI from "openai";
+import {
+  createBackgroundFetchHeaders,
+  createBackgroundFetchUrl,
+  createTaskOutputProperties,
+  handleOpenAIError,
+} from "./taskUtils";
+import { RunSubmitToolOutputsParams } from "openai/resources/beta/threads/runs/runs";
+import { ThreadUpdateParams } from "openai/resources/beta/threads/threads";
+
+export class Threads {
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
+
+  /**
+   * Create a thread and run it in one task.
+   */
+  async createAndRun(
+    key: IntegrationTaskKey,
+    params: Prettify<OpenAI.Beta.ThreadCreateAndRunParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.threads
+          .createAndRun(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        const outputProperties = [
+          ...(createTaskOutputProperties(undefined, response.headers) ?? []),
+          {
+            label: "threadId",
+            text: data.thread_id,
+          },
+          {
+            label: "runId",
+            text: data.id,
+          },
+        ];
+
+        task.outputProperties = outputProperties;
+
+        return data;
+      },
+      {
+        name: "Create Thread and Run",
+        params,
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Create a thread and runs it in one task, and only returns when the run is completed by polling in the background.
+   */
+  async createAndRunUntilCompletion(
+    key: IntegrationTaskKey,
+    params: Prettify<OpenAI.Beta.ThreadCreateAndRunParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads
+          .createAndRun(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        const outputProperties = [
+          ...(createTaskOutputProperties(undefined, response.headers) ?? []),
+          {
+            label: "threadId",
+            text: data.thread_id,
+          },
+          {
+            label: "runId",
+            text: data.id,
+          },
+        ];
+
+        task.outputProperties = outputProperties;
+
+        const url = createBackgroundFetchUrl(
+          client,
+          `/threads/${data.thread_id}/runs/${data.id}`,
+          this.options.defaultQuery,
+          options
+        );
+
+        const headers = this.options.defaultHeaders ?? {};
+
+        headers["OpenAI-Beta"] = "assistants=v1";
+
+        const completedRun = await io.backgroundPoll<OpenAI.Beta.Threads.Run>("poll", {
+          url,
+          requestInit: {
+            headers: createBackgroundFetchHeaders(client, task.idempotencyKey, headers, options),
+          },
+          interval: 10,
+          timeout: 600,
+          responseFilter: {
+            status: [200],
+            body: {
+              status: ["completed", "expired", "cancelled", "failed"],
+            },
+          },
+        });
+
+        return completedRun;
+      },
+      {
+        name: "Run Created Thread and Wait for Completion",
+        params,
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Create a thread.
+   */
+  async create(
+    key: IntegrationTaskKey,
+    params: Prettify<OpenAI.Beta.ThreadCreateParams> = {},
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Thread> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.threads
+          .create(params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Create Thread",
+        params,
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Retrieves a thread.
+   */
+  async retrieve(
+    key: IntegrationTaskKey,
+    threadId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Thread> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.threads
+          .retrieve(threadId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Retrieve Thread",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Modifies a thread.
+   */
+  async update(
+    key: IntegrationTaskKey,
+    threadId: string,
+    body: ThreadUpdateParams,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Thread> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.threads
+          .update(threadId, body, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Update Thread",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Delete a thread.
+   */
+  async del(
+    key: IntegrationTaskKey,
+    threadId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.ThreadDeleted> {
+    return this.runTask(
+      key,
+      async (client, task) => {
+        const { data, response } = await client.beta.threads
+          .del(threadId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Delete Thread",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  get runs() {
+    return new Runs(this.runTask.bind(this), this.options);
+  }
+
+  get messages() {
+    return new Messages(this.runTask.bind(this), this.options);
+  }
+}
+
+class Runs {
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
+
+  /**
+   * Creates a run and waits for it to complete by polling in the background.
+   */
+  async createAndWaitForCompletion(
+    key: IntegrationTaskKey,
+    threadId: string,
+    params: Prettify<OpenAI.Beta.Threads.RunCreateParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.runs
+          .create(threadId, params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        const url = createBackgroundFetchUrl(
+          client,
+          `/threads/${threadId}/runs/${data.id}`,
+          this.options.defaultQuery,
+          options
+        );
+
+        const headers = this.options.defaultHeaders ?? {};
+
+        headers["OpenAI-Beta"] = "assistants=v1";
+
+        const completedRun = await io.backgroundPoll<OpenAI.Beta.Threads.Run>("poll", {
+          url,
+          requestInit: {
+            headers: createBackgroundFetchHeaders(client, task.idempotencyKey, headers, options),
+          },
+          interval: 10,
+          timeout: 600,
+          responseFilter: {
+            status: [200],
+            body: {
+              status: ["completed", "expired", "cancelled", "failed"],
+            },
+          },
+        });
+
+        return completedRun;
+      },
+      {
+        name: "Run Thread and Wait for Completion",
+        params,
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Waits for a run to complete by polling in the background.
+   */
+  async waitForCompletion(
+    key: IntegrationTaskKey,
+    threadId: string,
+    runId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const url = createBackgroundFetchUrl(
+          client,
+          `/threads/${threadId}/runs/${runId}`,
+          this.options.defaultQuery,
+          options
+        );
+
+        const headers = this.options.defaultHeaders ?? {};
+
+        headers["OpenAI-Beta"] = "assistants=v1";
+
+        const completedRun = await io.backgroundPoll<OpenAI.Beta.Threads.Run>("poll", {
+          url,
+          requestInit: {
+            headers: createBackgroundFetchHeaders(client, task.idempotencyKey, headers, options),
+          },
+          interval: 10,
+          timeout: 600,
+          responseFilter: {
+            status: [200],
+            body: {
+              status: ["completed", "expired", "cancelled", "failed"],
+            },
+          },
+        });
+
+        return completedRun;
+      },
+      {
+        name: "Wait for Run Completion",
+        properties: [
+          { label: "threadId", text: threadId },
+          { label: "runId", text: runId },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Creates a run.
+   */
+  async create(
+    key: IntegrationTaskKey,
+    threadId: string,
+    params: Prettify<OpenAI.Beta.Threads.RunCreateParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.runs
+          .create(threadId, params, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Run Thread",
+        params,
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Retrieves a run.
+   */
+  async retrieve(
+    key: IntegrationTaskKey,
+    threadId: string,
+    runId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.runs
+          .retrieve(threadId, runId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Retrieve Run",
+        properties: [
+          { label: "threadId", text: threadId },
+          { label: "runId", text: runId },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Cancels a run that is `in_progress`.
+   */
+  async cancel(
+    key: IntegrationTaskKey,
+    threadId: string,
+    runId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.runs
+          .cancel(threadId, runId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Cancel Run",
+        properties: [
+          { label: "threadId", text: threadId },
+          { label: "runId", text: runId },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * When a run has the `status: "requires_action"` and `required_action.type` is
+   * `submit_tool_outputs`, this endpoint can be used to submit the outputs from the
+   * tool calls once they're all completed. All outputs must be submitted in a single
+   * request.
+   */
+  async submitToolOutputs(
+    key: IntegrationTaskKey,
+    threadId: string,
+    runId: string,
+    body: RunSubmitToolOutputsParams,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.runs
+          .submitToolOutputs(threadId, runId, body, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Submit Tool Outputs",
+        properties: [
+          { label: "threadId", text: threadId },
+          { label: "runId", text: runId },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Returns all runs belonging to a thread.
+   */
+  async list(
+    key: IntegrationTaskKey,
+    threadId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.Run[]> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data: page, response } = await client.beta.threads.runs
+          .list(threadId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        const allRuns = [];
+
+        for await (const fineTuningJob of page) {
+          allRuns.push(fineTuningJob);
+        }
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return allRuns;
+      },
+      {
+        name: "List Runs",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+}
+
+class Messages {
+  constructor(
+    private runTask: OpenAIRunTask,
+    private options: OpenAIIntegrationOptions
+  ) {}
+
+  /**
+   * Returns all messages for a given thread.
+   */
+  async list(
+    key: IntegrationTaskKey,
+    threadId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage[]> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data: page, response } = await client.beta.threads.messages
+          .list(threadId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        const allMessages = [];
+
+        for await (const fineTuningJob of page) {
+          allMessages.push(fineTuningJob);
+        }
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return allMessages;
+      },
+      {
+        name: "List Messages",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Create a message.
+   */
+  async create(
+    key: IntegrationTaskKey,
+    threadId: string,
+    body: Prettify<OpenAI.Beta.Threads.MessageCreateParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.messages
+          .create(threadId, body, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Create Message",
+        properties: [{ label: "threadId", text: threadId }],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Retrieve a message.
+   */
+  async retrieve(
+    key: IntegrationTaskKey,
+    threadId: string,
+    messageId: string,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.messages
+          .retrieve(threadId, messageId, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Retrieve Message",
+        properties: [
+          { label: "threadId", text: threadId },
+          { label: "messageId", text: messageId },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+
+  /**
+   * Modifies a message.
+   */
+  async update(
+    key: IntegrationTaskKey,
+    threadId: string,
+    messageId: string,
+    body: Prettify<OpenAI.Beta.Threads.MessageUpdateParams>,
+    options: OpenAIRequestOptions = {}
+  ): Promise<OpenAI.Beta.Threads.ThreadMessage> {
+    return this.runTask(
+      key,
+      async (client, task, io) => {
+        const { data, response } = await client.beta.threads.messages
+          .update(threadId, messageId, body, {
+            idempotencyKey: task.idempotencyKey,
+            ...options,
+          })
+          .withResponse();
+
+        task.outputProperties = createTaskOutputProperties(undefined, response.headers);
+
+        return data;
+      },
+      {
+        name: "Update Message",
+        properties: [
+          { label: "threadId", text: threadId },
+          { label: "messageId", text: messageId },
+        ],
+      },
+      handleOpenAIError
+    );
+  }
+}

integrations/openai/test/calculateResetAt.test.ts

@@ -0,0 +1,18 @@
+import { calculateResetAt } from "../src/taskUtils";
+
+describe("calculateResetAt", () => {
+  it("Should be able to correctly calculate based on a variety of formats", () => {
+    const now = new Date("2023-01-01T00:00:00.000Z");
+
+    expect(calculateResetAt("1s", now)).toEqual(new Date("2023-01-01T00:00:01.000Z"));
+    expect(calculateResetAt("6m59s", now)).toEqual(new Date("2023-01-01T00:06:59.000Z"));
+    expect(calculateResetAt("5m48s", now)).toEqual(new Date("2023-01-01T00:05:48.000Z"));
+    expect(calculateResetAt("1h44m5s", now)).toEqual(new Date("2023-01-01T01:44:05.000Z"));
+    expect(calculateResetAt("1h2s", now)).toEqual(new Date("2023-01-01T01:00:02.000Z"));
+    expect(calculateResetAt("45m", now)).toEqual(new Date("2023-01-01T00:45:00.000Z"));
+    expect(calculateResetAt("23h59m0s", now)).toEqual(new Date("2023-01-01T23:59:00.000Z"));
+    expect(calculateResetAt("1d22h8m1s", now)).toEqual(new Date("2023-01-02T22:08:01.000Z"));
+    expect(calculateResetAt("3h36m7.312s", now)).toEqual(new Date("2023-01-01T03:36:07.312Z"));
+    expect(calculateResetAt("72ms", now)).toEqual(new Date("2023-01-01T00:00:00.072Z"));
+  });
+});

integrations/openai/tsconfig.build.json

@@ -0,0 +1,18 @@
+{
+  "extends": "@trigger.dev/tsconfig/node18.json",
+  "include": ["./src/**/*.ts", "tsup.config.ts"],
+  "compilerOptions": {
+    "lib": ["DOM", "DOM.Iterable", "ES2019"],
+    "paths": {
+      "@trigger.dev/sdk/*": ["../../packages/trigger-sdk/src/*"],
+      "@trigger.dev/sdk": ["../../packages/trigger-sdk/src/index"],
+      "@trigger.dev/integration-kit/*": ["../../packages/integration-kit/src/*"],
+      "@trigger.dev/integration-kit": ["../../packages/integration-kit/src/index"]
+    },
+    "declaration": false,
+    "declarationMap": false,
+    "baseUrl": ".",
+    "stripInternal": true
+  },
+  "exclude": ["node_modules"]
+}

integrations/openai/tsconfig.json

@@ -1,6 +1,6 @@
 {
   "extends": "@trigger.dev/tsconfig/node18.json",
-  "include": ["./src/**/*.ts", "tsup.config.ts"],
+  "include": ["./src/**/*.ts", "tsup.config.ts", "src/globals.d.ts", "./test/**/*.ts"],
   "compilerOptions": {
     "lib": ["DOM", "DOM.Iterable", "ES2019"],
     "paths": {

integrations/plain/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/plain
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/plain/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/plain",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "The official Plain.com integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -24,8 +24,8 @@
     "build:tsup": "tsup"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "@team-plain/typescript-sdk": "^2.7.0"
   },
   "engines": {

integrations/replicate/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/replicate
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/replicate/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/replicate",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "Trigger.dev integration for replicate",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "replicate": "^0.18.1",
     "zod": "3.22.3"
   },

integrations/resend/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/resend
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/resend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/resend",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "The official Resend.com integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -24,8 +24,8 @@
     "build:tsup": "tsup"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "resend": "^2.0.0"
   },
   "engines": {

integrations/sendgrid/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/sendgrid
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/sendgrid/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/sendgrid",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "Trigger.dev integration for @sendgrid/mail",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -27,8 +27,8 @@
   },
   "dependencies": {
     "@sendgrid/mail": "^7.7.0",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
-    "@trigger.dev/integration-kit": "workspace:^2.2.5"
+    "@trigger.dev/sdk": "workspace:^2.2.6",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6"
   },
   "engines": {
     "node": ">=16.8.0"

integrations/slack/CHANGELOG.md

@@ -1,5 +1,15 @@
 # @trigger.dev/slack
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/slack/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/slack",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "The official Slack integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -25,7 +25,7 @@
   },
   "dependencies": {
     "@slack/web-api": "^6.8.1",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "zod": "3.22.3"
   },
   "engines": {

integrations/stripe/CHANGELOG.md

@@ -1,5 +1,16 @@
 # @trigger.dev/stripe
 
+## 2.2.6
+
+### Patch Changes
+
+- Updated dependencies [cb1825bf]
+- Updated dependencies [cb1825bf]
+- Updated dependencies [d0217344]
+- Updated dependencies [cb1825bf]
+  - @trigger.dev/integration-kit@2.2.6
+  - @trigger.dev/sdk@2.2.6
+
 ## 2.2.5
 
 ### Patch Changes

integrations/stripe/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/stripe",
-  "version": "2.2.5",
+  "version": "2.2.6",
   "description": "Trigger.dev integration for stripe",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^2.2.5",
-    "@trigger.dev/sdk": "workspace:^2.2.5",
+    "@trigger.dev/integration-kit": "workspace:^2.2.6",
+    "@trigger.dev/sdk": "workspace:^2.2.6",
     "stripe": "^12.14.0",
     "zod": "3.22.3"
   },