WIP (fly.io multi-tenant idea)

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.changeset/config.json

@@ -20,5 +20,8 @@
     "webapp",
     "emails",
     "@trigger.dev/database"
-  ]
+  ],
+  "___experimentalUnsafeOptions_WILL_CHANGE_IN_PATCH": {
+    "onlyUpdatePeerDependentsWhenOutOfRange": true
+  }
 }

.changeset/witty-weeks-fly.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Background tasks

.github/workflows/release.yml

@@ -8,7 +8,7 @@ on:
       - ".github/workflows/release.yml"
       - "packages/**"
       - "!packages/**/*.md"
-      - "changesets/**"
+      - ".changeset/**"
       - "integrations/**"
       - "!integrations/**/*.md"
       - "pnpm-lock.yaml"

.vscode/launch.json

@@ -5,29 +5,28 @@
   "version": "0.2.0",
   "configurations": [
     {
-      "command": "pnpm run dev --filter webapp",
-      "name": "Run webapp",
-      "request": "launch",
       "type": "node-terminal",
-      "cwd": "${workspaceFolder}"
+      "request": "launch",
+      "name": "Debug WebApp",
+      "command": "pnpm run dev --filter webapp",
+      "envFile": "${workspaceFolder}/apps/webapp/.env",
+      "cwd": "${workspaceFolder}",
+      "sourceMaps": true
     },
     {
       "type": "chrome",
       "request": "launch",
       "name": "Chrome webapp",
-      "url": "http://localhost:3000",
+      "url": "http://localhost:3030",
       "webRoot": "${workspaceFolder}/apps/webapp/app"
     },
     {
-      "type": "node",
+      "type": "node-terminal",
       "request": "launch",
-      "name": "Debug Current Test File",
-      "autoAttachChildProcesses": true,
-      "skipFiles": ["<node_internals>/**", "**/node_modules/**"],
-      "program": "${workspaceRoot}/node_modules/vitest/vitest.mjs",
-      "args": ["run", "${relativeFile}"],
-      "smartStep": true,
-      "console": "integratedTerminal"
+      "name": "Debug Deploy CLI",
+      "command": "pnpm exec trigger-cli deploy --tag 0.0.0-background-tasks-20230906212613",
+      "cwd": "${workspaceFolder}/examples/nextjs-background-tasks",
+      "sourceMaps": true
     }
   ]
 }

apps/webapp/app/components/code/CodeBlock.tsx

@@ -1,10 +1,9 @@
+import { Clipboard, ClipboardCheck } from "lucide-react";
 import type { Language, PrismTheme } from "prism-react-renderer";
 import Highlight, { defaultProps } from "prism-react-renderer";
 import { forwardRef, useCallback, useState } from "react";
 import { cn } from "~/utils/cn";
 import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "../primitives/Tooltip";
-import { ClipboardDocumentCheckIcon, ClipboardIcon } from "@heroicons/react/24/solid";
-import { Clipboard, ClipboardCheck, ClipboardCheckIcon } from "lucide-react";
 
 //This is a fork of https://github.com/mantinedev/mantine/blob/master/src/mantine-prism/src/Prism/Prism.tsx
 //it didn't support highlighting lines by dimming the rest of the code, or animations on the highlighting
@@ -192,6 +191,9 @@ export const CodeBlock = forwardRef<HTMLDivElement, CodeBlockProps>(
       Array.from({ length: end - start + 1 }, (_, i) => start + i)
     );
 
+    // if there are more than 1000 lines, don't highlight
+    const shouldHighlight = lineCount <= 1000;
+
     return (
       <div
         className={cn("relative overflow-hidden rounded-md border border-slate-800", className)}
@@ -229,99 +231,113 @@ export const CodeBlock = forwardRef<HTMLDivElement, CodeBlockProps>(
           </TooltipProvider>
         )}
 
-        <Highlight {...defaultProps} theme={theme} code={code} language={language}>
-          {({
-            className: inheritedClassName,
-            style: inheritedStyle,
-            tokens,
-            getLineProps,
-            getTokenProps,
-          }) => (
-            <div
-              dir="ltr"
-              className="overflow-auto px-2 py-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-slate-700"
-              style={{
-                maxHeight,
-              }}
-            >
-              <pre
-                className={cn(
-                  "relative mr-2 font-mono text-xs leading-relaxed",
-                  inheritedClassName
-                )}
-                style={inheritedStyle}
+        {shouldHighlight ? (
+          <Highlight {...defaultProps} theme={theme} code={code} language={language}>
+            {({
+              className: inheritedClassName,
+              style: inheritedStyle,
+              tokens,
+              getLineProps,
+              getTokenProps,
+            }) => (
+              <div
                 dir="ltr"
+                className="overflow-auto px-2 py-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-slate-700"
+                style={{
+                  maxHeight,
+                }}
               >
-                {tokens
-                  .map((line, index) => {
-                    if (
-                      index === tokens.length - 1 &&
-                      line.length === 1 &&
-                      line[0].content === "\n"
-                    ) {
-                      return null;
-                    }
+                <pre
+                  className={cn(
+                    "relative mr-2 font-mono text-xs leading-relaxed",
+                    inheritedClassName
+                  )}
+                  style={inheritedStyle}
+                  dir="ltr"
+                >
+                  {tokens
+                    .map((line, index) => {
+                      if (
+                        index === tokens.length - 1 &&
+                        line.length === 1 &&
+                        line[0].content === "\n"
+                      ) {
+                        return null;
+                      }
 
-                    const lineNumber = index + 1;
-                    const lineProps = getLineProps({ line, key: index });
+                      const lineNumber = index + 1;
+                      const lineProps = getLineProps({ line, key: index });
 
-                    let hasAnyHighlights = highlightLines ? highlightLines.length > 0 : false;
+                      let hasAnyHighlights = highlightLines ? highlightLines.length > 0 : false;
 
-                    let shouldDim = hasAnyHighlights;
-                    if (hasAnyHighlights && highlightLines?.includes(lineNumber)) {
-                      shouldDim = false;
-                    }
+                      let shouldDim = hasAnyHighlights;
+                      if (hasAnyHighlights && highlightLines?.includes(lineNumber)) {
+                        shouldDim = false;
+                      }
 
-                    return (
-                      <div
-                        key={lineProps.key}
-                        {...lineProps}
-                        className={cn(
-                          "flex w-full justify-start transition-opacity duration-500",
-                          lineProps.className
-                        )}
-                        style={{
-                          opacity: shouldDim ? dimAmount : undefined,
-                          ...lineProps.style,
-                        }}
-                      >
-                        {showLineNumbers && (
-                          <div
-                            className={
-                              "mr-2 flex-none select-none text-right text-slate-500 transition-opacity duration-500"
-                            }
-                            style={{
-                              width: `calc(8 * ${maxLineWidth / 16}rem)`,
-                            }}
-                          >
-                            {lineNumber}
+                      return (
+                        <div
+                          key={lineProps.key}
+                          {...lineProps}
+                          className={cn(
+                            "flex w-full justify-start transition-opacity duration-500",
+                            lineProps.className
+                          )}
+                          style={{
+                            opacity: shouldDim ? dimAmount : undefined,
+                            ...lineProps.style,
+                          }}
+                        >
+                          {showLineNumbers && (
+                            <div
+                              className={
+                                "mr-2 flex-none select-none text-right text-slate-500 transition-opacity duration-500"
+                              }
+                              style={{
+                                width: `calc(8 * ${maxLineWidth / 16}rem)`,
+                              }}
+                            >
+                              {lineNumber}
+                            </div>
+                          )}
+
+                          <div className="flex-1">
+                            {line.map((token, key) => {
+                              const tokenProps = getTokenProps({ token, key });
+                              return (
+                                <span
+                                  key={tokenProps.key}
+                                  {...tokenProps}
+                                  style={{
+                                    color: tokenProps?.style?.color as string,
+                                    ...tokenProps.style,
+                                  }}
+                                />
+                              );
+                            })}
                           </div>
-                        )}
-
-                        <div className="flex-1">
-                          {line.map((token, key) => {
-                            const tokenProps = getTokenProps({ token, key });
-                            return (
-                              <span
-                                key={tokenProps.key}
-                                {...tokenProps}
-                                style={{
-                                  color: tokenProps?.style?.color as string,
-                                  ...tokenProps.style,
-                                }}
-                              />
-                            );
-                          })}
+                          <div className="w-4 flex-none" />
                         </div>
-                        <div className="w-4 flex-none" />
-                      </div>
-                    );
-                  })
-                  .filter(Boolean)}
-              </pre>
-            </div>
-          )}
-        </Highlight>
+                      );
+                    })
+                    .filter(Boolean)}
+                </pre>
+              </div>
+            )}
+          </Highlight>
+        ) : (
+          <div
+            dir="ltr"
+            className="overflow-auto px-2 py-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-slate-700"
+            style={{
+              maxHeight,
+            }}
+          >
+            <pre className="relative mr-2 p-2 font-mono text-xs leading-relaxed" dir="ltr">
+              {code}
+            </pre>
+          </div>
+        )}
       </div>
     );
   }

apps/webapp/app/components/integrations/CustomHelp.tsx

@@ -25,8 +25,6 @@ client.defineJob({
     //wrap an SDK call in io.runTask so it's resumable and displays in logs
     const repo = await io.runTask(
       "Get repo",
-      //you can add metadata to the task to improve the display in the logs
-      { name: "Get repo", icon: "github" },
       async () => {
         //this is the regular GitHub SDK
         const response = await octokit.rest.repos.get({
@@ -34,7 +32,9 @@ client.defineJob({
           repo: "trigger.dev",
         });
         return response.data;
-      }
+      },
+      //you can add metadata to the task to improve the display in the logs
+      { name: "Get repo", icon: "github" }
     );
   },
 });
@@ -60,13 +60,13 @@ client.defineJob({
     //wrap anything in io.runTask so it's resumable and displays in logs
     const repo = await io.runTask(
       "Get org",
-      //you can add metadata to the task to improve the display in the logs
-      { name: "Get org", icon: "github" },
       async () => {
         //you can use fetch, axios, or any other library to make requests
         const response = await fetch('https://api.github.com/orgs/nodejs');
         return response.json();
-      }
+      },
+      //you can add metadata to the task to improve the display in the logs
+      { name: "Get org", icon: "github" }
     );
   },
 });

apps/webapp/app/components/run/RunCompletedDetail.tsx

@@ -54,7 +54,7 @@ export function RunCompletedDetail({ run }: { run: MatchedRun }) {
         <RunPanelDivider />
         {run.error && <RunPanelError text={run.error.message} stackTrace={run.error.stack} />}
         {run.output ? (
-          <CodeBlock language="json" code={run.output} />
+          <CodeBlock language="json" code={run.output} maxLines={36} />
         ) : (
           run.output === null && <Paragraph variant="small">This run returned nothing</Paragraph>
         )}

apps/webapp/app/components/run/TaskDetail.tsx

@@ -1,4 +1,3 @@
-import { DetailedTask } from "~/presenters/TaskDetailsPresenter.server";
 import {
   RunPanel,
   RunPanelBody,
@@ -29,22 +28,16 @@ import {
 } from "../primitives/Table";
 import { TaskAttemptStatusLabel } from "./TaskAttemptStatus";
 import { TaskStatusIcon } from "./TaskStatus";
+import { ClientOnly } from "remix-utils";
+import { Spinner } from "../primitives/Spinner";
+import type { DetailedTask } from "~/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam.tasks.$taskParam/route";
 
 export function TaskDetail({ task }: { task: DetailedTask }) {
-  const {
-    name,
-    description,
-    icon,
-    startedAt,
-    completedAt,
-    status,
-    delayUntil,
-    params,
-    properties,
-    output,
-    style,
-    attempts,
-  } = task;
+  const { name, description, icon, status, params, properties, output, style, attempts } = task;
+
+  const startedAt = task.startedAt ? new Date(task.startedAt) : undefined;
+  const completedAt = task.completedAt ? new Date(task.completedAt) : undefined;
+  const delayUntil = task.delayUntil ? new Date(task.delayUntil) : undefined;
 
   return (
     <RunPanel selected={false}>
@@ -150,7 +143,9 @@ export function TaskDetail({ task }: { task: DetailedTask }) {
         <div className="mt-4 flex flex-col gap-2">
           <Header3>Output</Header3>
           {output ? (
-            <CodeBlock code={JSON.stringify(output, null, 2)} />
+            <ClientOnly fallback={<Spinner />}>
+              {() => <CodeBlock code={output} maxLines={35} />}
+            </ClientOnly>
           ) : (
             <Paragraph variant="small">No output</Paragraph>
           )}

apps/webapp/app/env.server.ts

@@ -40,6 +40,14 @@ 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"),
+  // Docker Registry
+  DOCKER_REGISTRY_HOST: z.string().optional(),
+  DOCKER_REGISTRY_USERNAME: z.string().optional(),
+  DOCKER_REGISTRY_PASSWORD: z.string().optional(),
+  // Fly Background Task Provider
+  FLY_IO_API_TOKEN: z.string().optional(),
+  FLY_IO_API_URL: z.string().url().optional(),
+  FLY_IO_ORG_SLUG: z.string().optional(),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

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

@@ -0,0 +1,136 @@
+import {
+  BackgroundTaskSecret,
+  BackgroundTaskVersion,
+  SecretReference,
+} from "@trigger.dev/database";
+import { z } from "zod";
+import { $transaction, PrismaClientOrTransaction } from "~/db.server";
+import { SecretStore, getSecretStore } from "~/services/secrets/secretStore.server";
+
+const BackgroundTaskSecretSchema = z.object({
+  secret: z.string(),
+});
+
+export async function createBackgroundTaskSecret(
+  prisma: PrismaClientOrTransaction,
+  version: BackgroundTaskVersion,
+  key: string,
+  value: string
+) {
+  const secretKey = `${version.environmentId}:${key}`;
+
+  return await $transaction(prisma, async (tx) => {
+    const newSecret = await tx.backgroundTaskSecret.create({
+      data: {
+        key,
+        backgroundTaskVersion: {
+          connect: {
+            id: version.id,
+          },
+        },
+        secretReference: {
+          connectOrCreate: {
+            where: {
+              key: secretKey,
+            },
+            create: {
+              key: secretKey,
+              provider: "DATABASE",
+            },
+          },
+        },
+      },
+      include: {
+        secretReference: true,
+      },
+    });
+
+    const secretStoreProvider = getSecretStore("DATABASE", { prismaClient: tx });
+    const secretStore = new SecretStore(secretStoreProvider);
+
+    await secretStore.setSecret(secretKey, { secret: value });
+
+    return newSecret;
+  });
+}
+
+export async function updateBackgroundTaskSecret(
+  prisma: PrismaClientOrTransaction,
+  version: BackgroundTaskVersion,
+  key: string,
+  value: string
+) {
+  const secretKey = `${version.environmentId}:${key}`;
+
+  return await $transaction(prisma, async (tx) => {
+    const updatedSecret = await tx.backgroundTaskSecret.upsert({
+      where: {
+        backgroundTaskVersionId_key: {
+          backgroundTaskVersionId: version.id,
+          key,
+        },
+      },
+      create: {
+        key,
+        backgroundTaskVersion: {
+          connect: {
+            id: version.id,
+          },
+        },
+        secretReference: {
+          connectOrCreate: {
+            where: {
+              key: secretKey,
+            },
+            create: {
+              key: secretKey,
+              provider: "DATABASE",
+            },
+          },
+        },
+      },
+      update: {},
+      include: {
+        secretReference: true,
+      },
+    });
+
+    const secretStoreProvider = getSecretStore("DATABASE", { prismaClient: tx });
+    const secretStore = new SecretStore(secretStoreProvider);
+
+    await secretStore.setSecret(secretKey, { secret: value });
+
+    return updatedSecret;
+  });
+}
+
+export async function deleteBackgroundTaskSecret(prisma: PrismaClientOrTransaction, id: string) {
+  return await $transaction(prisma, async (tx) => {
+    const secret = await tx.backgroundTaskSecret.delete({
+      where: {
+        id,
+      },
+      include: {
+        secretReference: true,
+      },
+    });
+
+    if (!secret) {
+      return;
+    }
+
+    const secretStoreProvider = getSecretStore("DATABASE", { prismaClient: tx });
+    const secretStore = new SecretStore(secretStoreProvider);
+
+    await secretStore.deleteSecret(secret.secretReference.key);
+  });
+}
+
+export async function resolveBackgroundTaskSecret(reference: SecretReference) {
+  const secretStoreProvider = getSecretStore(reference.provider);
+  const secretStore = new SecretStore(secretStoreProvider);
+
+  const secretRecord = await secretStore.getSecret(BackgroundTaskSecretSchema, reference.key);
+
+  return secretRecord?.secret;
+}

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

@@ -28,6 +28,7 @@ export type EnqueueRunExecutionV2Options = {
   resumeTaskId?: string;
   isRetry?: boolean;
   skipRetrying?: boolean;
+  executionCount?: number;
 };
 
 export async function enqueueRunExecutionV2(
@@ -44,10 +45,11 @@ export async function enqueueRunExecutionV2(
       isRetry: typeof options.isRetry === "boolean" ? options.isRetry : false,
     },
     {
-      queueName: `job:${run.jobId}:env:${run.environmentId}`,
       tx,
       runAt: options.runAt,
-      jobKey: `job_run:${run.id}`,
+      jobKey: `job_run:${run.id}:${options.executionCount ?? 0}${
+        options.resumeTaskId ? `:task:${options.resumeTaskId}` : ""
+      }`,
       maxAttempts: options.skipRetrying ? 1 : undefined,
     }
   );

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

@@ -0,0 +1,61 @@
+import { ExternalAccount, Integration, TriggerSource } from "@trigger.dev/database";
+import { ConnectionAuth } from "@trigger.dev/core";
+import { PrismaClientOrTransaction } from "~/db.server";
+import { integrationAuthRepository } from "~/services/externalApis/integrationAuthRepository.server";
+import { logger } from "~/services/logger.server";
+
+type ResolvableTriggerSource = TriggerSource & {
+  integration: Integration;
+  externalAccount: ExternalAccount | null;
+};
+
+export async function resolveSourceConnection(
+  tx: PrismaClientOrTransaction,
+  source: ResolvableTriggerSource
+): Promise<ConnectionAuth | undefined> {
+  if (source.integration.authSource !== "HOSTED") return;
+
+  const connection = await getConnection(tx, source);
+
+  if (!connection) {
+    logger.error(
+      `Integration connection not found for source ${source.id}, integration ${source.integration.id}`
+    );
+    return;
+  }
+
+  const response = await integrationAuthRepository.getCredentials(connection);
+
+  if (!response) {
+    return;
+  }
+
+  return {
+    type: "oauth2",
+    scopes: response.scopes,
+    accessToken: response.accessToken,
+  };
+}
+
+function getConnection(tx: PrismaClientOrTransaction, source: ResolvableTriggerSource) {
+  if (source.externalAccount) {
+    return tx.integrationConnection.findFirst({
+      where: {
+        integrationId: source.integration.id,
+        externalAccountId: source.externalAccount.id,
+      },
+      include: {
+        dataReference: true,
+      },
+    });
+  }
+
+  return tx.integrationConnection.findFirst({
+    where: {
+      integrationId: source.integration.id,
+    },
+    include: {
+      dataReference: true,
+    },
+  });
+}

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

@@ -1,5 +1,6 @@
 import type { Task, TaskAttempt } from "@trigger.dev/database";
 import { ServerTask } from "@trigger.dev/core";
+import { PrismaClientOrTransaction } from "~/db.server";
 
 export type TaskWithAttempts = Task & { attempts: TaskAttempt[] };
 
@@ -25,3 +26,20 @@ export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask
     operation: task.operation,
   };
 }
+
+export type KitchenSinkTask = NonNullable<Awaited<ReturnType<typeof findKitchenSinkTask>>>;
+
+export async function findKitchenSinkTask(prisma: PrismaClientOrTransaction, id: string) {
+  return prisma.task.findUnique({
+    where: { id },
+    include: {
+      attempts: true,
+      run: {
+        include: {
+          environment: true,
+          queue: true,
+        },
+      },
+    },
+  });
+}

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

@@ -1,4 +1,4 @@
-import { DisplayPropertiesSchema, StyleSchema } from "@trigger.dev/core";
+import { StyleSchema } from "@trigger.dev/core";
 import { PrismaClient, prisma } from "~/db.server";
 import { mergeProperties } from "~/utils/mergeProperties.server";
 
@@ -7,8 +7,6 @@ type DetailsProps = {
   userId: string;
 };
 
-export type DetailedTask = NonNullable<Awaited<ReturnType<TaskDetailsPresenter["call"]>>>;
-
 export class TaskDetailsPresenter {
   #prismaClient: PrismaClient;
 
@@ -87,6 +85,7 @@ export class TaskDetailsPresenter {
 
     return {
       ...task,
+      output: task.output ? JSON.stringify(task.output, null, 2) : undefined,
       connection: task.runConnection,
       params: task.params as Record<string, any>,
       properties: mergeProperties(task.properties, task.outputProperties),

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

@@ -62,6 +62,22 @@ export class TriggerSourcePresenter {
             },
           },
         },
+        dynamicTrigger: {
+          select: {
+            id: true,
+            slug: true,
+            sourceRegistrationJob: {
+              select: {
+                job: {
+                  select: {
+                    id: true,
+                    slug: true,
+                  },
+                },
+              },
+            },
+          },
+        },
       },
       where: {
         id: triggerSourceId,
@@ -73,10 +89,15 @@ export class TriggerSourcePresenter {
     }
 
     const runListPresenter = new RunListPresenter(this.#prismaClient);
-    const runList = trigger.sourceRegistrationJob
+    const jobSlug = getJobSlug(
+      trigger.sourceRegistrationJob?.job.slug,
+      trigger.dynamicTrigger?.sourceRegistrationJob?.job.slug
+    );
+
+    const runList = jobSlug
       ? await runListPresenter.call({
           userId,
-          jobSlug: trigger.sourceRegistrationJob.job.slug,
+          jobSlug,
           organizationSlug,
           projectSlug,
           direction,
@@ -95,7 +116,21 @@ export class TriggerSourcePresenter {
         params: trigger.params,
         registrationJob: trigger.sourceRegistrationJob?.job,
         runList,
+        dynamic: trigger.dynamicTrigger
+          ? { id: trigger.dynamicTrigger.id, slug: trigger.dynamicTrigger.slug }
+          : undefined,
       },
     };
   }
 }
+
+function getJobSlug(
+  sourceRegistrationJobSlug: string | undefined,
+  dynamicSourceRegistrationJobSlug: string | undefined
+) {
+  if (sourceRegistrationJobSlug) {
+    return sourceRegistrationJobSlug;
+  }
+
+  return dynamicSourceRegistrationJobSlug;
+}

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

@@ -1,5 +1,7 @@
-import { LoaderArgs } from "@remix-run/server-runtime";
-import { typedjson, useTypedLoaderData } from "remix-typedjson";
+import { Await, useLoaderData } from "@remix-run/react";
+import { LoaderArgs, SerializeFrom, defer } from "@remix-run/server-runtime";
+import { Suspense } from "react";
+import { Spinner } from "~/components/primitives/Spinner";
 import { TaskDetail } from "~/components/run/TaskDetail";
 import { TaskDetailsPresenter } from "~/presenters/TaskDetailsPresenter.server";
 import { requireUserId } from "~/services/session.server";
@@ -10,23 +12,26 @@ export const loader = async ({ request, params }: LoaderArgs) => {
   const { taskParam } = TaskParamsSchema.parse(params);
 
   const presenter = new TaskDetailsPresenter();
-  const task = await presenter.call({
+  const taskPromise = presenter.call({
     userId,
     id: taskParam,
   });
 
-  if (!task) {
-    throw new Response(null, {
-      status: 404,
-    });
-  }
-
-  return typedjson({
-    task,
+  return defer({
+    taskPromise,
   });
 };
 
+export type DetailedTask = NonNullable<Awaited<SerializeFrom<typeof loader>["taskPromise"]>>;
+
 export default function Page() {
-  const { task } = useTypedLoaderData<typeof loader>();
-  return <TaskDetail task={task} />;
+  const { taskPromise } = useLoaderData<typeof loader>();
+
+  return (
+    <Suspense fallback={<Spinner />}>
+      <Await resolve={taskPromise} errorElement={<p>Error loading task!</p>}>
+        {(resolvedTask) => resolvedTask && <TaskDetail task={resolvedTask as any} />}
+      </Await>
+    </Suspense>
+  );
 }

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

@@ -43,6 +43,7 @@ import { parse } from "@conform-to/zod";
 import { z } from "zod";
 import { ActivateSourceService } from "~/services/sources/activateSource.server";
 import { redirectWithSuccessMessage } from "~/models/message.server";
+import { nanoid } from "nanoid";
 
 export const loader = async ({ request, params }: LoaderArgs) => {
   const user = await requireUser(request);
@@ -90,7 +91,7 @@ export const action: ActionFunction = async ({ request, params }) => {
   try {
     const service = new ActivateSourceService();
 
-    const result = await service.call(triggerParam, submission.value.jobId);
+    const result = await service.call(triggerParam);
 
     return redirectWithSuccessMessage(
       externalTriggerPath({ slug: organizationSlug }, { slug: projectParam }, { id: triggerParam }),
@@ -167,6 +168,17 @@ export default function Page() {
                 <NamedIcon name={trigger.active ? "active" : "inactive"} className="h-4 w-4" />
               }
             />
+            {trigger.dynamic && (
+              <PageInfoProperty
+                label="Dynamic"
+                value={
+                  <span className="flex items-center gap-0.5">
+                    <NamedIcon name="dynamic" className="h-4 w-4" />
+                    {trigger.dynamic.slug}
+                  </span>
+                }
+              />
+            )}
             <PageInfoProperty
               label="Environment"
               value={<EnvironmentLabel environment={trigger.environment} />}
@@ -206,7 +218,7 @@ export default function Page() {
                   </Button>
                 </Callout>
               </Form>
-            ) : (
+            ) : trigger.dynamic ? null : (
               <Callout variant="error" className="justiy-between mb-4 items-center">
                 This External Trigger hasn't registered successfully. Contact support for help:{" "}
                 {trigger.id}

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

@@ -1,5 +1,7 @@
-import { LoaderArgs } from "@remix-run/server-runtime";
-import { typedjson, useTypedLoaderData } from "remix-typedjson";
+import { Await, useLoaderData } from "@remix-run/react";
+import { LoaderArgs, defer } from "@remix-run/server-runtime";
+import { Suspense } from "react";
+import { Spinner } from "~/components/primitives/Spinner";
 import { TaskDetail } from "~/components/run/TaskDetail";
 import { TaskDetailsPresenter } from "~/presenters/TaskDetailsPresenter.server";
 import { requireUserId } from "~/services/session.server";
@@ -10,23 +12,24 @@ export const loader = async ({ request, params }: LoaderArgs) => {
   const { taskParam } = TriggerSourceRunTaskParamsSchema.parse(params);
 
   const presenter = new TaskDetailsPresenter();
-  const task = await presenter.call({
+  const taskPromise = presenter.call({
     userId,
     id: taskParam,
   });
 
-  if (!task) {
-    throw new Response(null, {
-      status: 404,
-    });
-  }
-
-  return typedjson({
-    task,
+  return defer({
+    taskPromise,
   });
 };
 
 export default function Page() {
-  const { task } = useTypedLoaderData<typeof loader>();
-  return <TaskDetail task={task} />;
+  const { taskPromise } = useLoaderData<typeof loader>();
+
+  return (
+    <Suspense fallback={<Spinner />}>
+      <Await resolve={taskPromise} errorElement={<p>Error loading task!</p>}>
+        {(resolvedTask) => resolvedTask && <TaskDetail task={resolvedTask as any} />}
+      </Await>
+    </Suspense>
+  );
 }

apps/webapp/app/routes/_docker.v2.$.ts

@@ -0,0 +1,10 @@
+import { ActionArgs, LoaderArgs } from "@remix-run/server-runtime";
+import { proxyToRegistry } from "~/services/docker/registryProxy.server";
+
+export async function action({ request }: ActionArgs) {
+  return await proxyToRegistry(request);
+}
+
+export async function loader({ request, params }: LoaderArgs) {
+  return await proxyToRegistry(request);
+}

apps/webapp/app/routes/_docker.v2.ts

@@ -0,0 +1,6 @@
+import { LoaderArgs } from "@remix-run/server-runtime";
+import { proxyToRegistry } from "~/services/docker/registryProxy.server";
+
+export async function loader({ request }: LoaderArgs) {
+  return await proxyToRegistry(request);
+}

apps/webapp/app/routes/api.v1.$endpointSlug.sources.$id.ts

@@ -1,10 +1,10 @@
 import type { ActionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { UpdateTriggerSourceBodySchema } from "@trigger.dev/core";
+import { UpdateTriggerSourceBodyV1Schema } from "@trigger.dev/core";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { UpdateSourceService } from "~/services/sources/updateSource.server";
+import { UpdateSourceServiceV1 } from "~/services/sources/updateSourceV1.server";
 
 const ParamsSchema = z.object({
   endpointSlug: z.string(),
@@ -40,13 +40,13 @@ export async function action({ request, params }: ActionArgs) {
   // Now parse the request body
   const anyBody = await request.json();
 
-  const body = UpdateTriggerSourceBodySchema.safeParse(anyBody);
+  const body = UpdateTriggerSourceBodyV1Schema.safeParse(anyBody);
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
   }
 
-  const service = new UpdateSourceService();
+  const service = new UpdateSourceServiceV1();
 
   try {
     const source = await service.call({

apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts

@@ -1,10 +1,10 @@
 import type { ActionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { RegisterTriggerBodySchema } from "@trigger.dev/core";
+import { RegisterTriggerBodySchemaV1 } from "@trigger.dev/core";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { RegisterTriggerSourceService } from "~/services/triggers/registerTriggerSource.server";
+import { RegisterTriggerSourceServiceV1 } from "~/services/triggers/registerTriggerSourceV1.server";
 
 const ParamsSchema = z.object({
   endpointSlug: z.string(),
@@ -41,13 +41,13 @@ export async function action({ request, params }: ActionArgs) {
   // Now parse the request body
   const anyBody = await request.json();
 
-  const body = RegisterTriggerBodySchema.safeParse(anyBody);
+  const body = RegisterTriggerBodySchemaV1.safeParse(anyBody);
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
   }
 
-  const service = new RegisterTriggerSourceService();
+  const service = new RegisterTriggerSourceServiceV1();
 
   try {
     const registration = await service.call({

apps/webapp/app/routes/api.v1.background.artifacts.$id.images.ts

@@ -0,0 +1,66 @@
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { CreateBackgroundTaskImageRequestBodySchema } from "@trigger.dev/core";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { CreateBackgroundTaskImageService } from "~/services/backgroundTasks/createBackgroundTaskImage.server";
+
+const ParamsSchema = z.object({
+  id: z.string(),
+});
+
+export async function action({ request, params }: ActionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return json({ error: "Invalid request params" }, { status: 400 });
+  }
+
+  // Now parse the request body
+  const anyBody = await request.json();
+
+  const body = CreateBackgroundTaskImageRequestBodySchema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new CreateBackgroundTaskImageService();
+
+  try {
+    const image = await service.call(
+      authenticationResult.environment,
+      parsedParams.data.id,
+      body.data
+    );
+
+    if (!image) {
+      return json(
+        {
+          error: `Unable to create background task image`,
+        },
+        { status: 500 }
+      );
+    }
+
+    return json(image);
+  } catch (error) {
+    if (error instanceof Error) {
+      return json({ error: error.message }, { status: 400 });
+    }
+
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}

apps/webapp/app/routes/api.v1.background.artifacts.$id[.tgz].ts

@@ -0,0 +1,142 @@
+import { Response } from "@remix-run/node";
+import { LoaderArgs, json } from "@remix-run/server-runtime";
+import { BackgroundTask, BackgroundTaskArtifact, PrismaClient } from "@trigger.dev/database";
+import archiver from "archiver";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+
+const ParamsSchema = z.object({
+  id: z.string(),
+});
+
+export async function loader({ params }: LoaderArgs) {
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    return json({ error: "Invalid request params" }, { status: 400 });
+  }
+
+  const service = new GenerateBackgroundTaskArtifactArchiveService();
+
+  const results = await service.call(parsedParams.data.id);
+
+  if (!results) {
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+
+  return new Response(results.archive, {
+    headers: {
+      "Content-Disposition": `attachment; filename="${results.name}"`,
+      "Content-Type": "application/gzip",
+    },
+  });
+}
+
+class GenerateBackgroundTaskArtifactArchiveService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    const artifact = await this.#prismaClient.backgroundTaskArtifact.findUnique({
+      where: {
+        id,
+      },
+      include: {
+        backgroundTask: true,
+      },
+    });
+
+    if (!artifact) {
+      return;
+    }
+
+    const archive = archiver("tar", {
+      gzip: true,
+      zlib: { level: 9 }, // Sets the compression level
+    });
+
+    function addFileToContext(contents: string, path: string) {
+      // Append files to the archive
+      archive.append(contents, { name: `ctx/${path}` });
+    }
+
+    // Good practice to catch warnings (ie stat failures and other non-blocking errors)
+    archive.on("warning", function (err) {
+      if (err.code === "ENOENT") {
+        // log warning
+        console.warn(err);
+      } else {
+        // throw error
+        throw err;
+      }
+    });
+
+    // Good practice to catch this error explicitly
+    archive.on("error", function (err) {
+      throw err;
+    });
+
+    addFileToContext(artifact.bundle, `src/${artifact.fileName}`);
+    addFileToContext(
+      JSON.stringify(this.#generatePackageJson(artifact, artifact.backgroundTask)),
+      "package.json"
+    );
+    addFileToContext(this.#generateDockerfile(artifact, artifact.backgroundTask), "Dockerfile");
+    addFileToContext(this.#generateIndexJs(artifact, artifact.backgroundTask), "src/index.js");
+
+    // Finalize the archive
+    archive.finalize();
+
+    return {
+      archive,
+      name: `${artifact.id}.tar.gz`,
+    };
+  }
+
+  #generatePackageJson(artifact: BackgroundTaskArtifact, task: BackgroundTask) {
+    return {
+      name: task.slug,
+      version: artifact.version,
+      description: `Trigger background task ${task.slug}`,
+      main: "src/index.js",
+      scripts: {
+        start: "node src/index.js",
+      },
+      dependencies: artifact.dependencies,
+      engines: {
+        node: this.#getNodeVersion(artifact.nodeVersion),
+      },
+    };
+  }
+
+  #generateDockerfile(artifact: BackgroundTaskArtifact, task: BackgroundTask) {
+    return `FROM amd64/node:${this.#getNodeVersion(artifact.nodeVersion)}-bullseye-slim
+    
+    WORKDIR /usr/src/app
+    
+    COPY ctx/package*.json ./
+    
+    RUN npm install
+    
+    COPY ctx/. .
+    
+    CMD [ "npm", "start" ]
+    `;
+  }
+
+  #generateIndexJs(artifact: BackgroundTaskArtifact, task: BackgroundTask) {
+    return `
+      const task = require("./${artifact.fileName}").default; 
+      console.log(task);
+      console.log(process.env);
+    `;
+  }
+
+  // replace the v if it exists
+  #getNodeVersion(version: string) {
+    return version.replace("v", "");
+  }
+}

apps/webapp/app/routes/api.v1.background.tasks.ts

@@ -0,0 +1,60 @@
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { DeployBackgroundTaskRequestBodySchema } from "@trigger.dev/core";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { DeployBackgroundTaskService } from "~/services/backgroundTasks/deployBackgroundTask.server";
+
+export async function action({ request, params }: ActionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const 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();
+
+  const body = DeployBackgroundTaskRequestBodySchema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new DeployBackgroundTaskService();
+
+  try {
+    const results = await service.call(authenticationResult.environment, body.data);
+
+    if (!results) {
+      return json(
+        {
+          error: `Unable to deploy background task, Task with ID = ${body.data.id} not found`,
+        },
+        { status: 500 }
+      );
+    }
+
+    const { artifact, imageConfig } = results;
+
+    return json({
+      id: artifact.id,
+      hash: artifact.hash,
+      image: imageConfig.image,
+      tag: imageConfig.tag,
+      createdAt: artifact.createdAt,
+      updatedAt: artifact.updatedAt,
+    });
+  } catch (error) {
+    if (error instanceof Error) {
+      return json({ error: error.message }, { status: 400 });
+    }
+
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}

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

@@ -183,7 +183,7 @@ export class RunTaskService {
             },
           },
           parent: taskBody.parentId ? { connect: { id: taskBody.parentId } } : undefined,
-          name: taskBody.name,
+          name: taskBody.name ?? "Task",
           description: taskBody.description,
           status,
           startedAt: new Date(),

apps/webapp/app/routes/api.v2.$endpointSlug.sources.$id.ts

@@ -0,0 +1,72 @@
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { UpdateTriggerSourceBodyV2Schema } from "@trigger.dev/core";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { UpdateSourceServiceV2 } from "~/services/sources/updateSourceV2.server";
+
+const ParamsSchema = z.object({
+  endpointSlug: z.string(),
+  id: z.string(),
+});
+
+export async function action({ request, params }: ActionArgs) {
+  logger.info("Updating source", { url: request.url });
+
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "PUT") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    logger.info("Invalid params", { params });
+
+    return json({ error: "Invalid params" }, { status: 400 });
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    logger.info("Invalid or missing api key", { url: request.url });
+    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 = UpdateTriggerSourceBodyV2Schema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new UpdateSourceServiceV2();
+
+  try {
+    const source = await service.call({
+      environment: authenticatedEnv,
+      payload: body.data,
+      endpointSlug: parsedParams.data.endpointSlug,
+      id: parsedParams.data.id,
+    });
+
+    return json(source);
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Error activating http source", {
+        url: request.url,
+        error: error.message,
+      });
+
+      return json({ error: error.message }, { status: 400 });
+    }
+
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}

apps/webapp/app/routes/api.v2.$endpointSlug.triggers.$id.registrations.$key.ts

@@ -0,0 +1,106 @@
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import {
+  REGISTER_SOURCE_EVENT_V2,
+  RegisterSourceEventV2,
+  RegisterTriggerBodySchemaV2,
+} from "@trigger.dev/core";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { IngestSendEvent } from "~/services/events/ingestSendEvent.server";
+import { logger } from "~/services/logger.server";
+import { RegisterTriggerSourceServiceV2 } from "~/services/triggers/registerTriggerSourceV2.server";
+
+const ParamsSchema = z.object({
+  endpointSlug: z.string(),
+  id: z.string(),
+  key: z.string(),
+});
+
+export async function action({ request, params }: ActionArgs) {
+  logger.info("Registering trigger", { url: request.url });
+
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "PUT") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    logger.info("Invalid params", { params });
+
+    return json({ error: "Invalid params" }, { status: 400 });
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    logger.info("Invalid or missing api key", { url: request.url });
+    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 = RegisterTriggerBodySchemaV2.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new RegisterTriggerSourceServiceV2();
+
+  try {
+    const registration = await service.call({
+      environment: authenticatedEnv,
+      payload: body.data,
+      endpointSlug: parsedParams.data.endpointSlug,
+      id: parsedParams.data.id,
+      key: parsedParams.data.key,
+    });
+
+    if (!registration) {
+      return json({ error: "Could not register trigger" }, { status: 500 });
+    }
+
+    //the source is already active
+    if (registration.source.active) {
+      return json(registration);
+    }
+
+    const payload: RegisterSourceEventV2 = {
+      ...registration,
+      dynamicTriggerId: parsedParams.data.id,
+    };
+
+    const ingestEventService = new IngestSendEvent();
+    await ingestEventService.call(
+      authenticatedEnv,
+      {
+        id: registration.id,
+        name: REGISTER_SOURCE_EVENT_V2,
+        source: "trigger.dev",
+        payload,
+      }
+      //todo accountId?
+      // {accountId: body.data.accountId}
+    );
+
+    return json(registration);
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Error registering trigger", {
+        url: request.url,
+        error: error.message,
+      });
+
+      return json({ error: error.message }, { status: 400 });
+    }
+
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}

apps/webapp/app/routes/healthcheck.tsx

@@ -7,11 +7,15 @@ export const loader: LoaderFunction = async ({ request }) => {
 
   try {
     const url = new URL("/", `http://${host}`);
+
+    if (request.headers.get("x-forwarded-proto") === "https") {
+      url.protocol = "https:";
+    }
     // if we can connect to the database and make a simple query
     // and make a HEAD request to ourselves, then we're good.
     await Promise.all([
       prisma.user.count(),
-      fetch(url.toString(), { method: "HEAD" }).then((r) => {
+      fetch(url.href, { method: "HEAD" }).then((r) => {
         if (!r.ok) return Promise.reject(r);
       }),
     ]);

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

@@ -26,30 +26,44 @@ export async function authenticateApiRequest(
     return;
   }
 
+  return await authenticateApiKey(result, { allowPublicKey });
+}
+
+export async function authenticateApiKey(
+  apiKey: string,
+  { allowPublicKey = false }: { allowPublicKey?: boolean } = {}
+): Promise<ApiAuthenticationResult | undefined> {
+  const type = isPublicApiKey(apiKey) ? ("PUBLIC" as const) : ("PRIVATE" as const);
+
   //if it's a public API key and we don't allow public keys, return
   if (!allowPublicKey) {
-    const environment = await findEnvironmentByApiKey(result.apiKey);
+    const environment = await findEnvironmentByApiKey(apiKey);
     if (!environment) return;
+
     return {
-      ...result,
+      apiKey,
+      type,
       environment,
     };
   }
 
-  switch (result.type) {
+  switch (type) {
     case "PUBLIC": {
-      const environment = await findEnvironmentByPublicApiKey(result.apiKey);
+      const environment = await findEnvironmentByPublicApiKey(apiKey);
       if (!environment) return;
       return {
-        ...result,
+        apiKey,
+        type,
         environment,
       };
     }
     case "PRIVATE": {
-      const environment = await findEnvironmentByApiKey(result.apiKey);
+      const environment = await findEnvironmentByApiKey(apiKey);
       if (!environment) return;
+
       return {
-        ...result,
+        apiKey,
+        type,
         environment,
       };
     }
@@ -69,6 +83,6 @@ export function getApiKeyFromRequest(request: Request) {
   }
 
   const apiKey = authorization.data.replace(/^Bearer /, "");
-  const type = isPublicApiKey(apiKey) ? ("PUBLIC" as const) : ("PRIVATE" as const);
-  return { apiKey, type };
+
+  return apiKey;
 }

apps/webapp/app/services/backgroundTasks/assignOperationToPool.server.ts

@@ -0,0 +1,66 @@
+import {
+  BackgroundTaskImage,
+  BackgroundTaskOperation,
+  BackgroundTaskVersion,
+} from "@trigger.dev/database";
+import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server";
+import { $transaction, prisma } from "~/db.server";
+import { workerQueue } from "../worker.server";
+import { AutoScalePoolService } from "./autoScalePool.server";
+
+export class AssignOperationToPoolService {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    operation: BackgroundTaskOperation,
+    version: BackgroundTaskVersion,
+    image: BackgroundTaskImage
+  ) {
+    return await $transaction(this.#prismaClient, async (tx) => {
+      const pool = await tx.backgroundTaskMachinePool.upsert({
+        where: {
+          backgroundTaskVersionId_imageId: {
+            backgroundTaskVersionId: version.id,
+            imageId: image.id,
+          },
+        },
+        create: {
+          backgroundTaskVersionId: version.id,
+          imageId: image.id,
+          backgroundTaskId: operation.backgroundTaskId,
+          provider: image.provider,
+          region: version.region,
+          cpu: version.cpu,
+          memory: version.memory,
+          concurrency: version.concurrency,
+          diskSize: version.diskSize,
+        },
+        update: {
+          region: version.region,
+          cpu: version.cpu,
+          memory: version.memory,
+          concurrency: version.concurrency,
+          diskSize: version.diskSize,
+        },
+      });
+
+      const updatedOperation = await tx.backgroundTaskOperation.update({
+        where: {
+          id: operation.id,
+        },
+        data: {
+          status: "ASSIGNED_TO_POOL",
+          poolId: pool.id,
+        },
+      });
+
+      await AutoScalePoolService.enqueue(pool, tx, true);
+
+      return updatedOperation;
+    });
+  }
+}

apps/webapp/app/services/backgroundTasks/autoScalePool.server.ts

@@ -0,0 +1,191 @@
+import {
+  BackgroundTask,
+  BackgroundTaskMachine,
+  BackgroundTaskMachinePool,
+} from "@trigger.dev/database";
+import type { PrismaClientOrTransaction } from "~/db.server";
+import { prisma } from "~/db.server";
+import { workerQueue } from "../worker.server";
+import { backgroundTaskProvider } from "./provider.server";
+import { CreateExternalMachineService } from "./createExternalMachine.server";
+
+const frequency = 1000 * 30; // 30 seconds
+
+export class AutoScalePoolService {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    const pool = await this.#prismaClient.backgroundTaskMachinePool.findUniqueOrThrow({
+      where: {
+        id,
+      },
+      include: {
+        _count: {
+          select: {
+            operations: {
+              where: {
+                status: "ASSIGNED_TO_POOL",
+              },
+            },
+          },
+        },
+        machines: true,
+        backgroundTask: true,
+        backgroundTaskVersion: true,
+      },
+    });
+
+    // Index the machines to gather the current state of the pool
+
+    // If there are no operations, we can re-enqueue this job
+    if (pool._count.operations === 0) {
+      return await AutoScalePoolService.enqueue(pool, this.#prismaClient);
+    }
+
+    // TODO: we probably should just list all the machines for the app?
+
+    const machines = await this.#autoScaleMachines(
+      pool.backgroundTaskVersion.concurrency,
+      pool,
+      pool.machines,
+      pool.backgroundTask
+    );
+
+    // We need to create any pending machines
+    const pendingMachines = machines.filter((machine) => machine.status === "PENDING");
+
+    for (const pendingMachine of pendingMachines) {
+      await CreateExternalMachineService.enqueue(pendingMachine, this.#prismaClient);
+    }
+
+    await backgroundTaskProvider.cleanupForTask(pool.backgroundTask);
+
+    await AutoScalePoolService.enqueue(pool, this.#prismaClient);
+  }
+
+  // If there are operations, we need to scale up the pool
+  // Machines will automatically be restarted when they are returned to the pool
+  // So we just need to make sure at least one machine is running
+  // And if there is not, we need to start one
+  // Status
+  // machine statutes:
+  // PENDING - The record has been created, but the machine has not on the provider
+  // CREATED - The machine has been created on the provider
+  // STARTING
+  // STARTED - The machine is running
+  // STOPPING
+  // STOPPED
+  // DESTROYING
+  // DESTROYED - The machine has been destroyed on the provider
+  // REPLACING - The machine config is being updated on the provider
+  async #autoScaleMachines(
+    target: number,
+    pool: BackgroundTaskMachinePool,
+    existingMachines: BackgroundTaskMachine[],
+    task: BackgroundTask
+  ): Promise<BackgroundTaskMachine[]> {
+    const pendingMachines: BackgroundTaskMachine[] = [];
+
+    // We need to update the pool to have the correct number of machines
+    if (existingMachines.length < target) {
+      const machinesToCreate = target - existingMachines.length;
+
+      for (let i = 0; i < machinesToCreate; i++) {
+        const pendingMachine = await this.#prismaClient.backgroundTaskMachine.create({
+          data: {
+            provider: backgroundTaskProvider.name,
+            poolId: pool.id,
+            backgroundTaskId: pool.backgroundTaskId,
+            backgroundTaskVersionId: pool.backgroundTaskVersionId,
+            backgroundTaskImageId: pool.imageId,
+          },
+        });
+
+        pendingMachines.push(pendingMachine);
+      }
+    }
+
+    const updatedExistingMachines = (
+      await Promise.all(existingMachines.map(async (machine) => this.#indexMachine(machine, task)))
+    ).filter(Boolean);
+
+    const replacedMachines: BackgroundTaskMachine[] = [];
+
+    if (updatedExistingMachines.length < existingMachines.length) {
+      const machinesToReplace = existingMachines.length - updatedExistingMachines.length;
+
+      for (let i = 0; i < machinesToReplace; i++) {
+        const pendingMachine = await this.#prismaClient.backgroundTaskMachine.create({
+          data: {
+            provider: backgroundTaskProvider.name,
+            poolId: pool.id,
+            backgroundTaskId: pool.backgroundTaskId,
+            backgroundTaskVersionId: pool.backgroundTaskVersionId,
+            backgroundTaskImageId: pool.imageId,
+          },
+        });
+
+        replacedMachines.push(pendingMachine);
+      }
+    }
+
+    return [...pendingMachines, ...updatedExistingMachines, ...replacedMachines];
+  }
+
+  async #indexMachine(
+    machine: BackgroundTaskMachine,
+    task: BackgroundTask
+  ): Promise<BackgroundTaskMachine | undefined> {
+    // Using the provider get updated information about the machine (if it has an externalId)
+    if (!machine.externalId) {
+      return machine;
+    }
+
+    const externalMachine = await backgroundTaskProvider.getMachineForTask(
+      machine.externalId,
+      task
+    );
+
+    if (!externalMachine) {
+      await this.#prismaClient.backgroundTaskMachine.delete({
+        where: {
+          id: machine.id,
+        },
+      });
+
+      return;
+    }
+
+    return await this.#prismaClient.backgroundTaskMachine.update({
+      where: {
+        id: machine.id,
+      },
+      data: {
+        status: externalMachine.status,
+        data: externalMachine.data,
+      },
+    });
+  }
+
+  static async enqueue(
+    pool: BackgroundTaskMachinePool,
+    tx: PrismaClientOrTransaction = prisma,
+    force = false
+  ) {
+    return await workerQueue.enqueue(
+      "autoScalePool",
+      {
+        id: pool.id,
+      },
+      {
+        tx,
+        jobKey: `scale:${pool.id}`,
+        runAt: force ? new Date() : new Date(Date.now() + frequency),
+      }
+    );
+  }
+}

apps/webapp/app/services/backgroundTasks/createBackgroundTaskImage.server.ts

@@ -0,0 +1,62 @@
+import { CreateBackgroundTaskImageRequestBody } from "@trigger.dev/core";
+import { PrismaClient } from "@trigger.dev/database";
+import { prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { backgroundTaskProvider } from "./provider.server";
+
+export class CreateBackgroundTaskImageService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    environment: AuthenticatedEnvironment,
+    id: string,
+    payload: CreateBackgroundTaskImageRequestBody
+  ) {
+    // Find the artifact
+    const artifact = await this.#prismaClient.backgroundTaskArtifact.findUnique({
+      where: {
+        id,
+      },
+      include: {
+        backgroundTask: true,
+      },
+    });
+
+    if (!artifact) {
+      return;
+    }
+
+    if (artifact.backgroundTask.projectId !== environment.projectId) {
+      return;
+    }
+
+    const image = await this.#prismaClient.backgroundTaskImage.upsert({
+      where: {
+        backgroundTaskArtifactId_digest: {
+          backgroundTaskArtifactId: artifact.id,
+          digest: payload.digest,
+        },
+      },
+      create: {
+        backgroundTaskArtifactId: artifact.id,
+        backgroundTaskId: artifact.backgroundTaskId,
+        digest: payload.digest,
+        name: payload.name,
+        tag: payload.tag,
+        size: payload.size,
+        provider: backgroundTaskProvider.name,
+      },
+      update: {
+        name: payload.name,
+        tag: payload.tag,
+        size: payload.size,
+      },
+    });
+
+    return image;
+  }
+}

apps/webapp/app/services/backgroundTasks/createExternalMachine.server.ts

@@ -0,0 +1,98 @@
+import { BackgroundTaskMachine } from "@trigger.dev/database";
+import type { PrismaClientOrTransaction } from "~/db.server";
+import { prisma } from "~/db.server";
+import { workerQueue } from "../worker.server";
+import { backgroundTaskProvider } from "./provider.server";
+import { resolveBackgroundTaskSecret } from "~/models/backgroundTaskSecret.server";
+import { env } from "~/env.server";
+import { ExternalMachineConfig } from "./providers/types";
+
+export class CreateExternalMachineService {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    const machine = await this.#prismaClient.backgroundTaskMachine.findUniqueOrThrow({
+      where: {
+        id,
+      },
+      include: {
+        pool: {
+          include: {
+            image: true,
+            backgroundTask: true,
+            backgroundTaskVersion: {
+              include: {
+                environment: true,
+                secrets: {
+                  include: {
+                    secretReference: true,
+                  },
+                },
+              },
+            },
+          },
+        },
+      },
+    });
+
+    const envVars: Record<string, string> = {};
+
+    for (const secret of machine.pool.backgroundTaskVersion.secrets) {
+      const secretValue = await resolveBackgroundTaskSecret(secret.secretReference);
+
+      if (!secretValue) {
+        continue;
+      }
+
+      envVars[secret.key] = secretValue;
+    }
+
+    envVars["TRIGGER_API_KEY"] = machine.pool.backgroundTaskVersion.environment.apiKey;
+    envVars["TRIGGER_API_URL"] = env.APP_ORIGIN;
+    envVars["TRIGGER_POOL_ID"] = machine.pool.id;
+    envVars["TRIGGER_MACHINE_ID"] = machine.id;
+
+    const config: ExternalMachineConfig = {
+      cpus: machine.pool.cpu,
+      memory: machine.pool.memory,
+      diskSize: machine.pool.diskSize,
+      region: machine.pool.region,
+      env: envVars,
+      image: `${backgroundTaskProvider.registry}/${machine.pool.image.name}:${machine.pool.image.tag}@${machine.pool.image.digest}`,
+    };
+
+    const externalMachine = await backgroundTaskProvider.createMachineForTask(
+      machine.id,
+      machine.pool.backgroundTask,
+      config
+    );
+
+    await this.#prismaClient.backgroundTaskMachine.update({
+      where: {
+        id: machine.id,
+      },
+      data: {
+        externalId: externalMachine.id,
+        status: externalMachine.status,
+        data: externalMachine.data,
+      },
+    });
+  }
+
+  static async enqueue(machine: BackgroundTaskMachine, tx: PrismaClientOrTransaction = prisma) {
+    return await workerQueue.enqueue(
+      "createExternalMachine",
+      {
+        id: machine.id,
+      },
+      {
+        tx,
+        jobKey: `createMachine:${machine.id}`,
+      }
+    );
+  }
+}

apps/webapp/app/services/backgroundTasks/deployBackgroundTask.server.ts

@@ -0,0 +1,86 @@
+import { DeployBackgroundTaskRequestBody } from "@trigger.dev/core";
+import { PrismaClient } from "@trigger.dev/database";
+import nodeCrypto from "node:crypto";
+import { prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { backgroundTaskProvider } from "./provider.server";
+
+export class DeployBackgroundTaskService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    environment: AuthenticatedEnvironment,
+    payload: DeployBackgroundTaskRequestBody
+  ) {
+    const hash = this.#hashPayload(payload);
+
+    const backgroundTask = await this.#prismaClient.backgroundTask.findUnique({
+      where: {
+        projectId_slug: {
+          projectId: environment.projectId,
+          slug: payload.id,
+        },
+      },
+    });
+
+    if (!backgroundTask) {
+      return;
+    }
+
+    const artifact = await this.#prismaClient.backgroundTaskArtifact.upsert({
+      where: {
+        backgroundTaskId_version_hash: {
+          backgroundTaskId: backgroundTask.id,
+          version: payload.version,
+          hash,
+        },
+      },
+      create: {
+        backgroundTaskId: backgroundTask.id,
+        fileName: payload.fileName,
+        version: payload.version,
+        hash,
+        bundle: payload.bundle,
+        nodeVersion: payload.nodeVersion,
+        dependencies: payload.dependencies,
+        sourcemap: payload.sourcemap,
+      },
+      update: {
+        fileName: payload.fileName,
+        bundle: payload.bundle,
+        nodeVersion: payload.nodeVersion,
+        dependencies: payload.dependencies,
+        sourcemap: payload.sourcemap,
+      },
+    });
+
+    const imageConfig = await backgroundTaskProvider.prepareArtifact(backgroundTask, artifact);
+
+    return {
+      artifact,
+      imageConfig,
+    };
+  }
+
+  #hashPayload(payload: DeployBackgroundTaskRequestBody) {
+    // Create a hash out of the bundle, the nodeVersion, and a determinstically list of dependencies
+    // This will allow us to determine if the bundle has changed
+    const hash = nodeCrypto.createHash("sha256");
+
+    hash.update(payload.bundle);
+    hash.update(payload.nodeVersion);
+
+    const dependencies = Object.keys(payload.dependencies).sort();
+
+    for (const dependency of dependencies) {
+      hash.update(dependency);
+      hash.update(payload.dependencies[dependency]);
+    }
+
+    return hash.digest("hex");
+  }
+}

apps/webapp/app/services/backgroundTasks/disableBackgroundTask.server.ts

@@ -0,0 +1,64 @@
+import type { BackgroundTaskVersion } from "@trigger.dev/database";
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
+import type { AuthenticatedEnvironment } from "../apiAuth.server";
+
+export type DisableBackgroundTaskServiceOptions = {
+  slug: string;
+  version: string;
+};
+
+export class DisableBackgroundTaskService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    endpointIdOrEndpoint: string | ExtendedEndpoint,
+    options: DisableBackgroundTaskServiceOptions
+  ) {
+    const endpoint =
+      typeof endpointIdOrEndpoint === "string"
+        ? await findEndpoint(endpointIdOrEndpoint)
+        : endpointIdOrEndpoint;
+
+    return this.#disableBackgroundTask(endpoint.environment, options);
+  }
+
+  async #disableBackgroundTask(
+    environment: AuthenticatedEnvironment,
+    options: DisableBackgroundTaskServiceOptions
+  ): Promise<BackgroundTaskVersion | undefined> {
+    const backgroundTask = await this.#prismaClient.backgroundTask.findUnique({
+      where: {
+        projectId_slug: {
+          projectId: environment.projectId,
+          slug: options.slug,
+        },
+      },
+    });
+
+    if (!backgroundTask) {
+      return;
+    }
+
+    const backgroundTaskVersion = await this.#prismaClient.backgroundTaskVersion.findUnique({
+      where: {
+        backgroundTaskId_version_environmentId: {
+          backgroundTaskId: backgroundTask.id,
+          version: options.version,
+          environmentId: environment.id,
+        },
+      },
+    });
+
+    if (!backgroundTaskVersion) {
+      return;
+    }
+
+    // TODO: Disable background task
+  }
+}

apps/webapp/app/services/backgroundTasks/executeBackgroundTaskOperation.server.ts

@@ -0,0 +1,55 @@
+import { PrismaClient } from "@trigger.dev/database";
+import { prisma } from "~/db.server";
+import { AssignOperationToPoolService } from "./assignOperationToPool.server";
+
+export class ExecuteBackgroundTaskOperationService {
+  #prismaClient: PrismaClient;
+  #assignOperationToPoolService = new AssignOperationToPoolService();
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    const operation = await this.#prismaClient.backgroundTaskOperation.findUniqueOrThrow({
+      where: {
+        id,
+      },
+      include: {
+        backgroundTask: true,
+        backgroundTaskVersion: true,
+      },
+    });
+
+    // Find the BackgroundTaskImage
+    const image = await this.#prismaClient.backgroundTaskImage.findFirst({
+      where: {
+        backgroundTaskId: operation.backgroundTaskId,
+        tag: operation.backgroundTaskVersion.version,
+      },
+      orderBy: {
+        createdAt: "desc",
+      },
+    });
+
+    // If the image is not found, we need to wait for it to be deployed
+    if (!image) {
+      await this.#prismaClient.backgroundTaskOperation.update({
+        where: {
+          id,
+        },
+        data: {
+          status: "WAITING_ON_IMAGE",
+        },
+      });
+
+      return;
+    }
+
+    return await this.#assignOperationToPoolService.call(
+      operation,
+      operation.backgroundTaskVersion,
+      image
+    );
+  }
+}

apps/webapp/app/services/backgroundTasks/initializeBackgroundTaskOperation.server.ts

@@ -0,0 +1,104 @@
+import { BackgroundTaskOperationParamsSchema } from "@trigger.dev/core";
+import { PrismaClient, RuntimeEnvironmentType, Task } from "@trigger.dev/database";
+import { $transaction, prisma } from "~/db.server";
+import { enqueueRunExecutionV2 } from "~/models/jobRunExecution.server";
+import { KitchenSinkTask } from "~/models/task.server";
+import { workerQueue } from "../worker.server";
+
+export class InitializeBackgroundTaskOperationService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(task: KitchenSinkTask) {
+    const params = BackgroundTaskOperationParamsSchema.safeParse(task.params);
+    // We need to create a new background task operation
+
+    if (!params.success) {
+      await this.#resumeTaskWithError(task, params.error.message);
+      return;
+    }
+
+    const backgroundTask = await this.#prismaClient.backgroundTask.findUnique({
+      where: {
+        projectId_slug: {
+          projectId: task.run.projectId,
+          slug: params.data.id,
+        },
+      },
+      include: {
+        versions: {
+          where: {
+            version: params.data.version,
+          },
+        },
+      },
+    });
+
+    if (!backgroundTask) {
+      await this.#resumeTaskWithError(task, `Background task ${params.data.id} not found`);
+      return;
+    }
+
+    const version = backgroundTask.versions[0];
+
+    if (!version) {
+      await this.#resumeTaskWithError(
+        task,
+        `Background task ${params.data.id} version ${params.data.version} not found`
+      );
+      return;
+    }
+
+    await $transaction(this.#prismaClient, async (tx) => {
+      const operation = await tx.backgroundTaskOperation.create({
+        data: {
+          backgroundTaskId: backgroundTask.id,
+          backgroundTaskVersionId: version.id,
+          taskId: task.id,
+          payload: params.data.payload,
+        },
+      });
+
+      await workerQueue.enqueue(
+        "executeBackgroundTaskOperation",
+        {
+          id: operation.id,
+        },
+        { tx }
+      );
+
+      return operation;
+    });
+  }
+
+  async #resumeTaskWithError(task: KitchenSinkTask, message: string) {
+    await $transaction(this.#prismaClient, async (tx) => {
+      await tx.task.update({
+        where: { id: task.id },
+        data: {
+          status: "ERRORED",
+          completedAt: new Date(),
+          output: { message },
+        },
+      });
+
+      await tx.taskAttempt.updateMany({
+        where: {
+          taskId: task.id,
+          status: "PENDING",
+        },
+        data: {
+          status: "ERRORED",
+          error: message,
+        },
+      });
+
+      await enqueueRunExecutionV2(task.run, prisma, {
+        skipRetrying: task.run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
+      });
+    });
+  }
+}

apps/webapp/app/services/backgroundTasks/provider.server.ts

@@ -0,0 +1,56 @@
+import { BackgroundTask, BackgroundTaskProviderStrategy } from "@trigger.dev/database";
+import { env } from "~/env.server";
+import { FlyBackgroundTaskProvider } from "./providers/fly.server";
+import { BackgroundTaskProvider, ExternalMachine, ExternalMachineConfig } from "./providers/types";
+
+export class UnsupportedBackgroundTaskProvider implements BackgroundTaskProvider {
+  async prepareArtifact(task: BackgroundTask): Promise<any> {
+    throw new Error("Unsupported background task provider");
+  }
+
+  get name(): BackgroundTaskProviderStrategy {
+    return "UNSUPPORTED";
+  }
+
+  get defaultRegion(): string {
+    return "UNSUPPORTED";
+  }
+
+  get registry(): string {
+    return "UNSUPPORTED";
+  }
+
+  async getMachineForTask(id: string, task: BackgroundTask): Promise<ExternalMachine | undefined> {
+    throw new Error("Unsupported background task provider");
+  }
+
+  getMachinesForTask(task: BackgroundTask): Promise<Array<ExternalMachine>> {
+    throw new Error("Unsupported background task provider");
+  }
+
+  createMachineForTask(
+    id: string,
+    task: BackgroundTask,
+    config: ExternalMachineConfig
+  ): Promise<ExternalMachine> {
+    throw new Error("Unsupported background task provider");
+  }
+
+  cleanupForTask(task: BackgroundTask): Promise<void> {
+    throw new Error("Unsupported background task provider");
+  }
+}
+
+let backgroundTaskProvider: BackgroundTaskProvider;
+
+if (env.FLY_IO_API_TOKEN && env.FLY_IO_API_URL && env.FLY_IO_ORG_SLUG) {
+  backgroundTaskProvider = new FlyBackgroundTaskProvider(
+    env.FLY_IO_API_URL,
+    env.FLY_IO_ORG_SLUG,
+    env.FLY_IO_API_TOKEN
+  );
+} else {
+  backgroundTaskProvider = new UnsupportedBackgroundTaskProvider();
+}
+
+export { backgroundTaskProvider };

apps/webapp/app/services/backgroundTasks/providers/fly.server.ts

@@ -0,0 +1,569 @@
+import {
+  BackgroundTask,
+  BackgroundTaskArtifact,
+  BackgroundTaskMachineStatus,
+  BackgroundTaskProviderStrategy,
+} from "@trigger.dev/database";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { ZodResponse, zodfetch } from "~/zodfetch.server";
+import { BackgroundTaskProvider, ExternalMachine, ExternalMachineConfig } from "./types";
+import retry from "async-retry";
+import AsyncRetry from "async-retry";
+
+const FlyAppSchema = z.object({
+  name: z.string(),
+  organization: z.object({
+    name: z.string(),
+    slug: z.string(),
+  }),
+  status: z.string(),
+});
+
+const FlyCreateAppSchema = z.object({
+  app_name: z.string(),
+  org_slug: z.string(),
+  network: z.string().optional(),
+});
+
+const FlyCheckStatusSchema = z.object({
+  name: z.string(),
+  output: z.string(),
+  status: z.string(),
+  updated_at: z.coerce.date().optional(),
+});
+
+const FlyMachineGuestSchema = z.object({
+  cpu_kind: z.enum(["shared", "dedicated"]),
+  cpus: z.number(),
+  memory_mb: z.number(),
+});
+
+const FlyMachineMetricsSchema = z.object({
+  path: z.string(),
+  port: z.number(),
+});
+
+const FlyMachineMountSchema = z.object({
+  encrypted: z.boolean().optional(),
+  name: z.string().optional(),
+  path: z.string(),
+  volume: z.string(),
+  size_gb: z.number().optional(),
+});
+
+const FlyMachineProcessSchema = z.object({
+  cmd: z.array(z.string()).optional(),
+  entrypoint: z.string().optional(),
+  env: z.record(z.string()).default({}),
+  exec: z.array(z.string()).optional(),
+  user: z.string().optional(),
+});
+
+const FlyMachineRestartSchema = z.object({
+  max_retries: z.number().optional(),
+  policy: z.string().optional(),
+});
+
+const FlyMachineHTTPHeaderSchema = z.array(
+  z.object({ name: z.string(), values: z.array(z.string()) })
+);
+
+const FlyMachineCheckSchema = z.object({
+  grace_period: z.string().optional(),
+  headers: z.array(FlyMachineHTTPHeaderSchema).default([]),
+  interval: z.string().optional(),
+  method: z.string().optional(),
+  path: z.string().optional(),
+  port: z.number().optional(),
+  protocol: z.string().optional(),
+  timeouyt: z.string().optional(),
+  tls_server_name: z.string().optional(),
+  tls_skip_verify: z.boolean().optional(),
+  type: z.string().optional(),
+});
+
+const FlyMachineServiceConcurrencySchema = z.object({
+  hard_limit: z.number().optional(),
+  soft_limit: z.number().optional(),
+  type: z.string(),
+});
+
+const FlyHTTPOptionsSchema = z.object({
+  compress: z.boolean().optional(),
+  response: z
+    .object({
+      headers: z.record(z.string()).default({}),
+    })
+    .optional(),
+});
+
+const FlyMachinePortSchema = z.object({
+  end_port: z.number(),
+  force_https: z.boolean().optional(),
+  handlers: z.array(z.string()).default([]),
+  http_options: FlyHTTPOptionsSchema.optional(),
+});
+
+const FlyMachineServiceSchema = z.object({
+  autostart: z.boolean().optional(),
+  autostop: z.boolean().optional(),
+  checks: z.array(FlyMachineCheckSchema).default([]),
+  concurrency: FlyMachineServiceConcurrencySchema,
+  force_instance_description: z.string().optional(),
+  force_instance_key: z.string().optional(),
+  internal_port: z.number().optional(),
+  min_machines_running: z.number().optional(),
+  ports: z.array(FlyMachinePortSchema).default([]),
+  protocol: z.string().optional(),
+});
+
+const FlyMachineStaticSchema = z.object({
+  guest_path: z.string(),
+  url_prefix: z.string(),
+});
+
+const FlyMachineStopConfigSchema = z.object({
+  signal: z.string().optional(),
+  timeout: z.string().optional(),
+});
+
+const FlyMachineConfigSchema = z.object({
+  auto_destroy: z.boolean().optional(),
+  env: z.record(z.string()).default({}),
+  checks: z.record(FlyMachineCheckSchema).default({}),
+  metadata: z.record(z.string()).default({}),
+  guest: FlyMachineGuestSchema,
+  image: z.string(),
+  metrics: FlyMachineMetricsSchema.optional(),
+  mounts: z.array(FlyMachineMountSchema).default([]),
+  processes: z.array(FlyMachineProcessSchema).default([]),
+  restart: FlyMachineRestartSchema.default({}),
+  services: z.array(FlyMachineServiceSchema).default([]),
+  standbys: z.array(z.string()).default([]),
+  statics: z.array(FlyMachineStaticSchema).default([]),
+  stop_config: FlyMachineStopConfigSchema.optional(),
+});
+
+const FlyMachineImageRefSchema = z.object({
+  digest: z.string(),
+  registry: z.string(),
+  repository: z.string(),
+  tag: z.string(),
+  labels: z.record(z.string()).nullable().default({}),
+});
+
+const FlyMachineStateSchema = z.enum([
+  "created",
+  "starting",
+  "started",
+  "stopping",
+  "stopped",
+  "destroying",
+  "destroyed",
+  "replacing",
+]);
+
+const FlyMachineEventSchema = z.object({
+  id: z.string(),
+  type: z.string(),
+  status: z.string(),
+  source: z.string(),
+  timestamp: z.coerce.date(),
+  request: z.any(),
+});
+
+const FlyMachineSchema = z.object({
+  id: z.string(),
+  instance_id: z.string(),
+  name: z.string(),
+  nonce: z.string().optional(),
+  private_ip: z.string(),
+  region: z.string(),
+  state: FlyMachineStateSchema,
+  config: FlyMachineConfigSchema,
+  checks: z.array(FlyCheckStatusSchema).optional(),
+  events: z.array(FlyMachineEventSchema).default([]),
+  image_ref: FlyMachineImageRefSchema,
+  created_at: z.coerce.date(),
+  updated_at: z.coerce.date().optional(),
+});
+
+const FlyVolumeSchema = z.object({
+  id: z.string(),
+  name: z.string(),
+  state: z.string(),
+  region: z.string(),
+  size_gb: z.number(),
+  encrypted: z.boolean(),
+  created_at: z.coerce.date(),
+  attached_machine_id: z.string().nullable().optional(),
+  attached_alloc_id: z.string().nullable().optional(),
+  blocks: z.number(),
+  block_size: z.number(),
+  blocks_free: z.number(),
+  blocks_avail: z.number(),
+  fstype: z.string(),
+  host_dedication_key: z.string().nullable().optional(),
+});
+
+const FlyCreateVolumeSchema = z.object({
+  name: z.string(),
+  region: z.string(),
+  size_gb: z.number(),
+  machines_only: z.boolean().optional(),
+  encrypted: z.boolean().optional(),
+});
+
+const FlyCreateMachineSchema = z.object({
+  name: z.string(),
+  lease_ttl: z.number().optional(),
+  region: z.string(),
+  config: FlyMachineConfigSchema,
+  skip_launch: z.boolean().optional(),
+  skip_service_registration: z.boolean().optional(),
+});
+
+export class FlyBackgroundTaskProvider implements BackgroundTaskProvider {
+  private readonly _logger = logger.child("FlyBackgroundTaskProvider");
+
+  get name(): BackgroundTaskProviderStrategy {
+    return "FLY_IO";
+  }
+
+  get registry(): string {
+    return "registry.fly.io";
+  }
+
+  constructor(
+    private readonly url: string,
+    private readonly org: string,
+    private readonly token: string
+  ) {}
+
+  get defaultRegion(): string {
+    return "iad";
+  }
+
+  async prepareArtifact(
+    task: BackgroundTask,
+    artifact: BackgroundTaskArtifact
+  ): Promise<{ image: string; tag: string }> {
+    // Check that the app has been created
+    const app = await this.#getApp(this.#appNameForTask(task));
+
+    if (app) {
+      return {
+        image: app.name,
+        tag: artifact.version,
+      };
+    }
+
+    // Create the app
+    const created = await this.#createApp({
+      app_name: this.#appNameForTask(task),
+      network: this.#networkNameForTask(task),
+      org_slug: this.org,
+    });
+
+    if (!created) {
+      throw new Error("Failed to create app");
+    }
+
+    return {
+      image: this.#appNameForTask(task),
+      tag: artifact.version,
+    };
+  }
+
+  async getMachineForTask(id: string, task: BackgroundTask): Promise<ExternalMachine | undefined> {
+    const response = await this.#fetch(
+      FlyMachineSchema,
+      `/v1/apps/${this.#appNameForTask(task)}/machines/${id}`
+    );
+
+    if (!response.ok) {
+      return;
+    }
+
+    return this.#flyMachineToExternalMachine(response.data);
+  }
+
+  async getMachinesForTask(task: BackgroundTask): Promise<Array<ExternalMachine>> {
+    const response = await this.#fetch(
+      z.array(FlyMachineSchema),
+      `/v1/apps/${this.#appNameForTask(task)}/machines`
+    );
+
+    if (!response.ok) {
+      return [];
+    }
+
+    return response.data.map((machine) => this.#flyMachineToExternalMachine(machine));
+  }
+
+  async createMachineForTask(
+    id: string,
+    task: BackgroundTask,
+    config: ExternalMachineConfig
+  ): Promise<ExternalMachine> {
+    // We have to create a volume first
+    const volume = await this.#createVolume(this.#appNameForTask(task), {
+      name: id,
+      region: config.region,
+      size_gb: config.diskSize,
+      encrypted: true,
+      machines_only: true,
+    });
+
+    const machine = await this.#createMachine(this.#appNameForTask(task), {
+      name: id,
+      region: config.region,
+      config: {
+        image: config.image,
+        env: config.env,
+        guest: {
+          cpu_kind: "shared",
+          cpus: config.cpus,
+          memory_mb: config.memory,
+        },
+        auto_destroy: false,
+        mounts: [
+          {
+            volume: volume.id,
+            path: "/data",
+          },
+        ],
+      },
+    });
+
+    return this.#flyMachineToExternalMachine(machine);
+  }
+
+  async cleanupForTask(task: BackgroundTask): Promise<void> {
+    const volumes = await this.#listVolumes(this.#appNameForTask(task));
+
+    if (!volumes) {
+      return;
+    }
+
+    // Destroy any volumn created more than 30 seconds ago that doesn't have a machine attached
+    const destroyableVolumes = volumes.filter(
+      (volume) =>
+        volume.created_at.getTime() < Date.now() - 30 * 1000 &&
+        !volume.attached_machine_id &&
+        !volume.attached_alloc_id &&
+        volume.state !== "pending_destroy"
+    );
+
+    this._logger.debug("cleanupForTask", {
+      volumesToDestroy: destroyableVolumes.length,
+    });
+
+    for (const volume of destroyableVolumes) {
+      await this.#destroyVolume(this.#appNameForTask(task), volume.id);
+    }
+  }
+
+  #flyMachineToExternalMachine(flyMachine: z.output<typeof FlyMachineSchema>): ExternalMachine {
+    return {
+      id: flyMachine.id,
+      status: this.#flyStateToStatus(flyMachine.state),
+      data: flyMachine,
+    };
+  }
+
+  #flyStateToStatus(state: z.infer<typeof FlyMachineStateSchema>): BackgroundTaskMachineStatus {
+    const mappings: Record<z.infer<typeof FlyMachineStateSchema>, BackgroundTaskMachineStatus> = {
+      created: "CREATED",
+      starting: "STARTING",
+      started: "STARTED",
+      stopping: "STOPPING",
+      stopped: "STOPPED",
+      destroying: "DESTROYING",
+      destroyed: "DESTROYED",
+      replacing: "REPLACING",
+    };
+
+    return mappings[state];
+  }
+
+  async #getApp(appName: string) {
+    const response = await this.#fetch(FlyAppSchema, `/v1/apps/${appName}`);
+
+    if (!response.ok) {
+      return;
+    }
+
+    return response.data;
+  }
+
+  async #createApp(body: z.input<typeof FlyCreateAppSchema>) {
+    const response = await this.#fetch(z.any(), "/v1/apps", {
+      method: "POST",
+      body: JSON.stringify(body),
+    });
+
+    return response.ok;
+  }
+
+  async #createMachine(
+    appName: string,
+    body: z.input<typeof FlyCreateMachineSchema>
+  ): Promise<z.output<typeof FlyMachineSchema>> {
+    const response = await this.#fetch(
+      FlyMachineSchema,
+      `/v1/apps/${appName}/machines`,
+      {
+        method: "POST",
+        body: JSON.stringify(body),
+      },
+      {
+        retries: 5,
+      }
+    );
+
+    if (!response.ok) {
+      throw new Error("Failed to create machine");
+    }
+
+    return response.data;
+  }
+
+  async #createVolume(
+    appName: string,
+    body: z.input<typeof FlyCreateVolumeSchema>
+  ): Promise<z.output<typeof FlyVolumeSchema>> {
+    const response = await this.#fetch(
+      FlyVolumeSchema,
+      `/v1/apps/${appName}/volumes`,
+      {
+        method: "POST",
+        body: JSON.stringify(body),
+      },
+      {
+        retries: 5,
+      }
+    );
+
+    if (!response.ok) {
+      throw new Error("Failed to create volume");
+    }
+
+    return response.data;
+  }
+
+  async #listVolumes(
+    appName: string
+  ): Promise<Array<z.output<typeof FlyVolumeSchema>> | undefined> {
+    const response = await this.#fetch(z.array(FlyVolumeSchema), `/v1/apps/${appName}/volumes`, {
+      method: "GET",
+    });
+
+    if (!response.ok) {
+      return;
+    }
+
+    return response.data;
+  }
+
+  async #destroyVolume(appName: string, id: string): Promise<boolean> {
+    const response = await this.#fetch(z.any(), `/v1/apps/${appName}/volumes/${id}`, {
+      method: "DELETE",
+    });
+
+    return response.ok;
+  }
+
+  async #fetch<TResponseSchema extends z.ZodTypeAny>(
+    schema: TResponseSchema,
+    path: string,
+    requestInit?: RequestInit,
+    retryOptions?: AsyncRetry.Options
+  ): Promise<ZodResponse<TResponseSchema>> {
+    const headers = new Headers(requestInit?.headers ?? {});
+
+    // Add the common headers
+    headers.set("Authorization", `Bearer ${this.token}`);
+    headers.set("Accept", "application/json");
+    headers.set("User-Agent", "Trigger.dev/2.1.0");
+
+    if (requestInit?.body) {
+      headers.set("Content-Type", "application/json; charset=utf-8");
+    }
+
+    if (retryOptions) {
+      return await retry(
+        async (bail) => {
+          const response = await zodfetch(schema, `${this.url}${path}`, {
+            ...requestInit,
+            headers,
+          });
+
+          if (!response.ok) {
+            if (response.status === 429 || response.status >= 500) {
+              throw new Error(
+                `[${response.status}] Request ${
+                  requestInit?.method ?? "GET"
+                } ${path} failed: ${JSON.stringify(response.error)}`
+              );
+            }
+
+            bail(
+              new Error(
+                `[${response.status}] Request ${
+                  requestInit?.method ?? "GET"
+                } ${path} failed: ${JSON.stringify(response.error)}`
+              )
+            );
+            return response;
+          }
+
+          return response;
+        },
+        {
+          ...retryOptions,
+          onRetry: (e, attempt) => {
+            this._logger.debug("fetch.retry", {
+              url: `${this.url}${path}`,
+              attempt,
+              response: {
+                ok: response.ok,
+                status: response.status,
+                error: response.ok ? undefined : response.error,
+                err: {
+                  message: e.message,
+                  stack: e.stack,
+                },
+              },
+            });
+          },
+        }
+      );
+    }
+
+    const response = await zodfetch(schema, `${this.url}${path}`, {
+      ...requestInit,
+      headers,
+    });
+
+    this._logger.debug("fetch", {
+      url: `${this.url}${path}`,
+      response: {
+        ok: response.ok,
+        status: response.status,
+        error: response.ok ? undefined : response.error,
+      },
+    });
+
+    return response;
+  }
+
+  #appNameForTask(task: BackgroundTask): string {
+    return `${task.id}-${task.slug}`;
+  }
+
+  #networkNameForTask(task: BackgroundTask): string {
+    return task.projectId;
+  }
+}

apps/webapp/app/services/backgroundTasks/providers/types.ts

@@ -0,0 +1,43 @@
+import type {
+  BackgroundTask,
+  BackgroundTaskArtifact,
+  BackgroundTaskMachineStatus,
+  BackgroundTaskProviderStrategy,
+} from "@trigger.dev/database";
+
+export type ExternalMachine = {
+  id: string;
+  status: BackgroundTaskMachineStatus;
+  data: any;
+};
+
+export type ExternalMachineConfig = {
+  cpus: number;
+  memory: number;
+  diskSize: number;
+  region: string;
+  image: string;
+  env: Record<string, string>;
+};
+
+export interface BackgroundTaskProvider {
+  prepareArtifact(
+    task: BackgroundTask,
+    artifact: BackgroundTaskArtifact
+  ): Promise<{ image: string; tag: string }>;
+
+  get name(): BackgroundTaskProviderStrategy;
+  get defaultRegion(): string;
+  get registry(): string;
+
+  getMachineForTask(id: string, task: BackgroundTask): Promise<ExternalMachine | undefined>;
+  getMachinesForTask(task: BackgroundTask): Promise<Array<ExternalMachine>>;
+
+  createMachineForTask(
+    id: string,
+    task: BackgroundTask,
+    config: ExternalMachineConfig
+  ): Promise<ExternalMachine>;
+
+  cleanupForTask(task: BackgroundTask): Promise<void>;
+}

apps/webapp/app/services/backgroundTasks/registerBackgroundTask.server.ts

@@ -0,0 +1,221 @@
+import { BackgroundTaskMetadata } from "@trigger.dev/core";
+import type { BackgroundTaskVersion, Endpoint } from "@trigger.dev/database";
+import { DEFAULT_MAX_CONCURRENT_RUNS } from "~/consts";
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+import {
+  createBackgroundTaskSecret,
+  deleteBackgroundTaskSecret,
+  updateBackgroundTaskSecret,
+} from "~/models/backgroundTaskSecret.server";
+import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
+import type { AuthenticatedEnvironment } from "../apiAuth.server";
+import { backgroundTaskProvider } from "./provider.server";
+
+export class RegisterBackgroundTaskService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    endpointIdOrEndpoint: string | ExtendedEndpoint,
+    metadata: BackgroundTaskMetadata
+  ) {
+    const endpoint =
+      typeof endpointIdOrEndpoint === "string"
+        ? await findEndpoint(endpointIdOrEndpoint)
+        : endpointIdOrEndpoint;
+
+    return this.#upsertBackgroundTask(endpoint, endpoint.environment, metadata);
+  }
+
+  async #upsertBackgroundTask(
+    endpoint: Endpoint,
+    environment: AuthenticatedEnvironment,
+    metadata: BackgroundTaskMetadata
+  ): Promise<BackgroundTaskVersion | undefined> {
+    // Check the background task doesn't already exist and is deleted
+    const existingBackgroundTask = await this.#prismaClient.backgroundTask.findUnique({
+      where: {
+        projectId_slug: {
+          projectId: environment.projectId,
+          slug: metadata.id,
+        },
+      },
+    });
+
+    if (existingBackgroundTask && existingBackgroundTask.deletedAt && !metadata.enabled) {
+      return;
+    }
+
+    const backgroundTask = await this.#prismaClient.backgroundTask.upsert({
+      where: {
+        projectId_slug: {
+          projectId: environment.projectId,
+          slug: metadata.id,
+        },
+      },
+      create: {
+        organization: {
+          connect: {
+            id: environment.organizationId,
+          },
+        },
+        project: {
+          connect: {
+            id: environment.projectId,
+          },
+        },
+        slug: metadata.id,
+        title: metadata.name,
+      },
+      update: {
+        title: metadata.name,
+        deletedAt: metadata.enabled ? null : undefined,
+      },
+    });
+
+    const backgroundTaskVersion = await this.#prismaClient.backgroundTaskVersion.upsert({
+      where: {
+        backgroundTaskId_version_environmentId: {
+          backgroundTaskId: backgroundTask.id,
+          version: metadata.version,
+          environmentId: environment.id,
+        },
+      },
+      create: {
+        backgroundTask: {
+          connect: {
+            id: backgroundTask.id,
+          },
+        },
+        endpoint: {
+          connect: {
+            id: endpoint.id,
+          },
+        },
+        environment: {
+          connect: {
+            id: environment.id,
+          },
+        },
+        organization: {
+          connect: {
+            id: environment.organizationId,
+          },
+        },
+        project: {
+          connect: {
+            id: environment.projectId,
+          },
+        },
+        version: metadata.version,
+        cpu: metadata.cpu,
+        memory: metadata.memory,
+        concurrency: metadata.concurrency ?? DEFAULT_MAX_CONCURRENT_RUNS,
+        region: metadata.region ?? backgroundTaskProvider.defaultRegion,
+        diskSize: metadata.diskSizeInGB,
+      },
+      update: {
+        cpu: metadata.cpu,
+        memory: metadata.memory,
+        concurrency: metadata.concurrency ?? DEFAULT_MAX_CONCURRENT_RUNS,
+        region: metadata.region,
+        diskSize: metadata.diskSizeInGB,
+        endpoint: {
+          connect: {
+            id: endpoint.id,
+          },
+        },
+      },
+    });
+
+    // Count the number of job instances that have higher version numbers
+    const laterVersionCount = await this.#prismaClient.backgroundTaskVersion.count({
+      where: {
+        backgroundTaskId: backgroundTask.id,
+        version: {
+          gt: metadata.version,
+        },
+        environmentId: environment.id,
+      },
+    });
+
+    // If there are no later versions, then we can upsert the latest BackgroundTaskAlias
+    if (laterVersionCount === 0) {
+      await this.#prismaClient.backgroundTaskAlias.upsert({
+        where: {
+          backgroundTaskId_environmentId_name: {
+            backgroundTaskId: backgroundTask.id,
+            environmentId: environment.id,
+            name: "latest",
+          },
+        },
+        create: {
+          backgroundTaskId: backgroundTask.id,
+          versionId: backgroundTaskVersion.id,
+          environmentId: environment.id,
+          name: "latest",
+          value: backgroundTaskVersion.version,
+        },
+        update: {
+          versionId: backgroundTaskVersion.id,
+          value: backgroundTaskVersion.version,
+        },
+      });
+    }
+
+    // Now we need to register the background task secrets
+    // 1. Add new secrets
+    // 2. Remove old secrets
+    // 3. Update existing secrets
+
+    const existingSecrets = await this.#prismaClient.backgroundTaskSecret.findMany({
+      where: {
+        backgroundTaskVersionId: backgroundTaskVersion.id,
+      },
+      select: {
+        id: true,
+        key: true,
+      },
+    });
+
+    const metadataSecrets = metadata.secrets ?? {};
+
+    const existingSecretKeys = existingSecrets.map((s) => s.key);
+    const newSecretKeys = Object.keys(metadataSecrets);
+
+    const secretsToRemove = existingSecrets.filter((s) => !newSecretKeys.includes(s.key));
+    const secretsToCreate = newSecretKeys.filter((k) => !existingSecretKeys.includes(k));
+    const secretsToUpdate = newSecretKeys.filter((k) => existingSecretKeys.includes(k));
+
+    // 1. Add new secrets
+    for (const secretKey of secretsToCreate) {
+      await createBackgroundTaskSecret(
+        this.#prismaClient,
+        backgroundTaskVersion,
+        secretKey,
+        metadataSecrets[secretKey]
+      );
+    }
+
+    // 2. Remove old secrets
+    for (const secret of secretsToRemove) {
+      await deleteBackgroundTaskSecret(this.#prismaClient, secret.id);
+    }
+
+    // 3. Update existing secrets
+    for (const secretKey of secretsToUpdate) {
+      await updateBackgroundTaskSecret(
+        this.#prismaClient,
+        backgroundTaskVersion,
+        secretKey,
+        metadataSecrets[secretKey]
+      );
+    }
+
+    return backgroundTaskVersion;
+  }
+}

apps/webapp/app/services/docker/registryProxy.server.ts

@@ -0,0 +1,146 @@
+import { env } from "~/env.server";
+import { authenticateApiKey } from "../apiAuth.server";
+import { logger } from "../logger.server";
+
+export class RegistryProxy {
+  constructor(public readonly host: string, private auth: { username: string; password: string }) {}
+
+  public async call(request: Request) {
+    return await this.#proxyRequest(request);
+  }
+
+  // Proxies the request to the registry
+  async #proxyRequest(request: Request) {
+    const credentials = this.#getBasicAuthCredentials(request);
+
+    if (!credentials) {
+      logger.debug("Returning 401 because credentials are missing");
+
+      return new Response("Unauthorized", {
+        status: 401,
+        headers: {
+          "WWW-Authenticate": 'Basic realm="Access to the registry"',
+        },
+      });
+    }
+
+    // Authenticate the request
+    const authenticatedEnv = await authenticateApiKey(credentials.password, {
+      allowPublicKey: false,
+    });
+
+    if (!authenticatedEnv) {
+      return new Response("Unauthorized", {
+        status: 401,
+      });
+    }
+
+    // construct a new url based on the url passed in and the registry url
+    const proxiedUrl = new URL(request.url);
+    proxiedUrl.host = this.host;
+
+    // Update the protocol to https if there is the x-forwarded-proto header
+    if (request.headers.get("x-forwarded-proto") === "https") {
+      proxiedUrl.protocol = "https:";
+    }
+
+    const updatedHeaders = this.#updateHeaders(request.headers);
+
+    const response = await fetch(proxiedUrl, {
+      method: request.method,
+      headers: updatedHeaders,
+      body: request.body,
+    });
+
+    const updatedResponseHeaders = this.#updateResponseHeaders(response.headers, request.url);
+
+    logger.debug("proxied request/response", {
+      proxiedUrl,
+      status: response.status,
+      statusText: response.statusText,
+      method: request.method,
+      requestHeaders: Object.fromEntries(updatedHeaders.entries()),
+      responseHeaders: Object.fromEntries(updatedResponseHeaders.entries()),
+    });
+
+    return new Response(response.body, {
+      status: response.status,
+      statusText: response.statusText,
+      headers: updatedResponseHeaders,
+    });
+  }
+
+  #getBasicAuthCredentials(request: Request) {
+    const authHeader = request.headers.get("authorization");
+
+    if (!authHeader) {
+      return;
+    }
+
+    const [type, credentials] = authHeader.split(" ");
+
+    if (type.toLowerCase() !== "basic") {
+      return;
+    }
+
+    const decoded = Buffer.from(credentials, "base64").toString("utf-8");
+    const [username, password] = decoded.split(":");
+
+    return { username, password };
+  }
+
+  // Updates the headers to be sent to the registry
+  // adds the docker auth
+  #updateHeaders(headers: Headers): Headers {
+    const newHeaders = new Headers(headers);
+
+    // Remove host, connection, accept-encoding, content-length, and authorization headers
+    newHeaders.delete("host");
+    newHeaders.delete("connection");
+    newHeaders.delete("accept-encoding");
+    newHeaders.delete("authorization");
+    newHeaders.delete("content-length");
+
+    newHeaders.set(
+      "authorization",
+      `Basic ${Buffer.from(`${this.auth.username}:${this.auth.password}`).toString("base64")}`
+    );
+
+    return newHeaders;
+  }
+
+  // Updates the headers to be sent back to the client
+  #updateResponseHeaders(headers: Headers, proxyUrl: string): Headers {
+    const newHeaders = new Headers(headers);
+
+    // Rewrite location headers to point to the proxy
+    if (headers.has("location")) {
+      const location = headers.get("location");
+
+      if (location) {
+        const proxiedLocation = new URL(location);
+        proxiedLocation.host = new URL(proxyUrl).host;
+
+        newHeaders.set("location", proxiedLocation.href);
+      }
+    }
+
+    return newHeaders;
+  }
+}
+
+export async function proxyToRegistry(request: Request) {
+  if (!env.DOCKER_REGISTRY_HOST || !env.DOCKER_REGISTRY_USERNAME || !env.DOCKER_REGISTRY_PASSWORD) {
+    return new Response(
+      "Could not proxy to the registry, please double check your DOCKER_REGISTRY_* env vars",
+      { status: 500 }
+    );
+  }
+
+  const registryProxy = new RegistryProxy(env.DOCKER_REGISTRY_HOST, {
+    username: env.DOCKER_REGISTRY_USERNAME,
+    password: env.DOCKER_REGISTRY_PASSWORD,
+  });
+
+  return await registryProxy.call(request);
+}

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

@@ -1,6 +1,7 @@
 import {
   ApiEventLog,
   DeliverEventResponseSchema,
+  DeserializedJson,
   ErrorWithStackSchema,
   HttpSourceRequest,
   HttpSourceResponseSchema,
@@ -9,8 +10,8 @@ import {
   PongResponseSchema,
   PreprocessRunBody,
   PreprocessRunResponseSchema,
-  RegisterTriggerBody,
-  RegisterTriggerBodySchema,
+  RegisterTriggerBodySchemaV1,
+  RegisterTriggerBodyV1,
   RunJobBody,
   RunJobResponseSchema,
   ValidateResponse,
@@ -18,6 +19,8 @@ import {
 } from "@trigger.dev/core";
 import { safeBodyFromResponse, safeParseBodyFromResponse } from "~/utils/json";
 import { logger } from "./logger.server";
+import { ConnectionAuth } from "@trigger.dev/core";
+import { performance } from "node:perf_hooks";
 
 export class EndpointApiError extends Error {
   constructor(message: string, stack?: string) {
@@ -28,10 +31,7 @@ export class EndpointApiError extends Error {
 }
 
 export class EndpointApi {
-  constructor(
-    private apiKey: string,
-    private url: string
-  ) {}
+  constructor(private apiKey: string, private url: string) {}
 
   async ping(endpointId: string): Promise<PongResponse> {
     const response = await safeFetch(this.url, {
@@ -165,9 +165,7 @@ export class EndpointApi {
   }
 
   async executeJobRequest(options: RunJobBody) {
-    logger.debug("executeJobRequest()", {
-      options,
-    });
+    const startTimeInMs = performance.now();
 
     const response = await safeFetch(this.url, {
       method: "POST",
@@ -183,6 +181,7 @@ export class EndpointApi {
       response,
       parser: RunJobResponseSchema,
       errorParser: ErrorWithStackSchema,
+      durationInMs: Math.floor(performance.now() - startTimeInMs),
     };
   }
 
@@ -200,7 +199,7 @@ export class EndpointApi {
     return { response, parser: PreprocessRunResponseSchema };
   }
 
-  async initializeTrigger(id: string, params: any): Promise<RegisterTriggerBody | undefined> {
+  async initializeTrigger(id: string, params: any): Promise<RegisterTriggerBodyV1 | undefined> {
     const response = await safeFetch(this.url, {
       method: "POST",
       headers: {
@@ -234,7 +233,7 @@ export class EndpointApi {
       body: anyBody,
     });
 
-    return RegisterTriggerBodySchema.parse(anyBody);
+    return RegisterTriggerBodySchemaV1.parse(anyBody);
   }
 
   async deliverHttpSourceRequest(options: {
@@ -244,6 +243,8 @@ export class EndpointApi {
     params: any;
     data: any;
     request: HttpSourceRequest;
+    auth?: ConnectionAuth;
+    metadata?: any;
   }) {
     const response = await safeFetch(this.url, {
       method: "POST",
@@ -258,7 +259,9 @@ export class EndpointApi {
         "x-ts-http-url": options.request.url,
         "x-ts-http-method": options.request.method,
         "x-ts-http-headers": JSON.stringify(options.request.headers),
+        ...(options.auth && { "x-ts-auth": JSON.stringify(options.auth) }),
         ...(options.dynamicId && { "x-ts-dynamic-id": options.dynamicId }),
+        ...(options.metadata && { "x-ts-metadata": JSON.stringify(options.metadata) }),
       },
       body: options.request.rawBody,
     });

apps/webapp/app/services/endpoints/indexEndpoint.server.ts

@@ -4,16 +4,22 @@ import { findEndpoint } from "~/models/endpoint.server";
 import { EndpointApi } from "../endpointApi.server";
 import { RegisterJobService } from "../jobs/registerJob.server";
 import { logger } from "../logger.server";
-import { RegisterSourceService } from "../sources/registerSource.server";
+import { RegisterSourceServiceV1 } from "../sources/registerSourceV1.server";
 import { RegisterDynamicScheduleService } from "../triggers/registerDynamicSchedule.server";
 import { RegisterDynamicTriggerService } from "../triggers/registerDynamicTrigger.server";
 import { DisableJobService } from "../jobs/disableJob.server";
+import { RegisterSourceServiceV2 } from "../sources/registerSourceV2.server";
+import { RegisterBackgroundTaskService } from "../backgroundTasks/registerBackgroundTask.server";
+import { DisableBackgroundTaskService } from "../backgroundTasks/disableBackgroundTask.server";
 
 export class IndexEndpointService {
   #prismaClient: PrismaClient;
   #registerJobService = new RegisterJobService();
   #disableJobService = new DisableJobService();
-  #registerSourceService = new RegisterSourceService();
+  #registerSourceServiceV1 = new RegisterSourceServiceV1();
+  #registerSourceServiceV2 = new RegisterSourceServiceV2();
+  #registerBackgroundTaskService = new RegisterBackgroundTaskService();
+  #disableBackgroundTaskService = new DisableBackgroundTaskService();
   #registerDynamicTriggerService = new RegisterDynamicTriggerService();
   #registerDynamicScheduleService = new RegisterDynamicScheduleService();
 
@@ -38,7 +44,13 @@ export class IndexEndpointService {
       throw new Error(indexResponse.error);
     }
 
-    const { jobs, sources, dynamicTriggers, dynamicSchedules } = indexResponse.data;
+    const {
+      jobs,
+      sources,
+      dynamicTriggers,
+      dynamicSchedules,
+      backgroundTasks = [],
+    } = indexResponse.data;
 
     logger.debug("Indexing endpoint", {
       endpointId: endpoint.id,
@@ -51,15 +63,18 @@ export class IndexEndpointService {
         sources: sources.length,
         dynamicTriggers: dynamicTriggers.length,
         dynamicSchedules: dynamicSchedules.length,
+        backgroundTasks: backgroundTasks.length,
       },
     });
 
     const indexStats = {
       jobs: 0,
+      backgroundTasks: 0,
       sources: 0,
       dynamicTriggers: 0,
       dynamicSchedules: 0,
       disabledJobs: 0,
+      disabedBackgroundTasks: 0,
     };
 
     const existingJobs = await this.#prismaClient.job.findMany({
@@ -154,9 +169,113 @@ export class IndexEndpointService {
       }
     }
 
+    const existingBackgroundTasks = await this.#prismaClient.backgroundTask.findMany({
+      where: {
+        projectId: endpoint.projectId,
+        deletedAt: null,
+      },
+      include: {
+        aliases: {
+          where: {
+            name: "latest",
+            environmentId: endpoint.environmentId,
+          },
+          include: {
+            version: true,
+          },
+          take: 1,
+        },
+      },
+    });
+
+    for (const backgroundTask of backgroundTasks) {
+      if (!backgroundTask.enabled) {
+        const disabledBackgroundTask = await this.#disableBackgroundTaskService
+          .call(endpoint, { slug: backgroundTask.id, version: backgroundTask.version })
+          .catch((error) => {
+            logger.error("Failed to disable background task", {
+              endpointId: endpoint.id,
+              backgroundTask,
+              error,
+            });
+
+            return;
+          });
+
+        if (disabledBackgroundTask) {
+          indexStats.disabledJobs++;
+        }
+      } else {
+        try {
+          const registeredVersion = await this.#registerBackgroundTaskService.call(
+            endpoint,
+            backgroundTask
+          );
+
+          if (registeredVersion) {
+            indexStats.backgroundTasks++;
+          }
+        } catch (error) {
+          logger.error("Failed to register background task", {
+            endpointId: endpoint.id,
+            backgroundTask,
+            error,
+          });
+        }
+      }
+    }
+
+    const missingBackgroundTasks = existingBackgroundTasks.filter((backgroundTask) => {
+      return !backgroundTasks.find((b) => b.id === backgroundTask.slug);
+    });
+
+    if (missingBackgroundTasks.length > 0) {
+      logger.debug("Disabling missing background tasks", {
+        endpointId: endpoint.id,
+        missingIds: missingBackgroundTasks.map((job) => job.slug),
+      });
+
+      for (const backgroundTask of missingBackgroundTasks) {
+        const latestVersion = backgroundTask.aliases[0]?.version;
+
+        if (!latestVersion) {
+          continue;
+        }
+
+        const disabledBackgroundTask = await this.#disableBackgroundTaskService
+          .call(endpoint, {
+            slug: backgroundTask.slug,
+            version: latestVersion.version,
+          })
+          .catch((error) => {
+            logger.error("Failed to disable background task", {
+              endpointId: endpoint.id,
+              backgroundTask,
+              error,
+            });
+
+            return;
+          });
+
+        if (disabledBackgroundTask) {
+          indexStats.disabledJobs++;
+        }
+      }
+    }
+
     for (const source of sources) {
       try {
-        await this.#registerSourceService.call(endpoint, source);
+        switch (source.version) {
+          default:
+          case "1": {
+            await this.#registerSourceServiceV1.call(endpoint, source);
+            break;
+          }
+          case "2": {
+            await this.#registerSourceServiceV2.call(endpoint, source);
+            break;
+          }
+        }
 
         indexStats.sources++;
       } catch (error) {

apps/webapp/app/services/externalApis/integrationCatalog.server.ts

@@ -1,11 +1,12 @@
+import { airtable } from "./integrations/airtable";
 import { github } from "./integrations/github";
 import { openai } from "./integrations/openai";
 import { plain } from "./integrations/plain";
 import { resend } from "./integrations/resend";
+import { sendgrid } from "./integrations/sendgrid";
 import { slack } from "./integrations/slack";
 import { stripe } from "./integrations/stripe";
-import { sendgrid } from "./integrations/sendgrid";
-import { supabaseManagement, supabase } from "./integrations/supabase";
+import { supabase, supabaseManagement } from "./integrations/supabase";
 import { typeform } from "./integrations/typeform";
 import type { Integration } from "./types";
 
@@ -30,6 +31,7 @@ export class IntegrationCatalog {
 }
 
 export const integrationCatalog = new IntegrationCatalog({
+  airtable,
   github,
   openai,
   plain,

apps/webapp/app/services/externalApis/integrations/airtable.ts

@@ -1,4 +1,52 @@
-import type { Integration } from "../types";
+import type { HelpSample, Integration } from "../types";
+
+function usageSample(hasApiKey: boolean): HelpSample {
+  return {
+    title: "Using the client",
+    code: `
+import { Airtable } from "@trigger.dev/airtable";
+
+const airtable = new Airtable({
+  id: "__SLUG__"${hasApiKey ? ",\n  token: process.env.AIRTABLE_TOKEN!" : ""}
+});
+
+//you can define your Airtable table types
+type LaunchGoalsAndOkRs = {
+  "Launch goals"?: string;
+  DRI?: Collaborator;
+  Team?: string;
+  Status?: "On track" | "In progress" | "At risk";
+  "Key results"?: Array<string>;
+  "Features (from 💻 Features table)"?: Array<string>;
+  "Status (from 💻 Features)": Array<"Live" | "Complete" | "In progress" | "Planning" | "In reviews">;
+};
+
+client.defineJob({
+  id: "airtable-example-1",
+  name: "Airtable Example 1: getRecords",
+  version: "0.1.0",
+  trigger: eventTrigger({
+    name: "airtable.example",
+    schema: z.object({
+      baseId: z.string(),
+      tableName: z.string(),
+    }),
+  }),
+  integrations: {
+    airtable,
+  },
+  run: async (payload, io, ctx) => {
+    //then you can set the types for your table, so you get type safety
+    const table = io.airtable.base(payload.baseId).table<LaunchGoalsAndOkRs>(payload.tableName);
+
+    const records = await table.getRecords("muliple records", { fields: ["Status"] });
+    //this will be type checked
+    await io.logger.log(records[0].fields.Status ?? "no status");
+  },
+});
+  `,
+  };
+}
 
 export const airtable: Integration = {
   identifier: "airtable",
@@ -70,18 +118,13 @@ export const airtable: Integration = {
         },
       ],
       help: {
-        samples: [
-          {
-            title: "Creating the client",
-            code: `
-import { Airtable } from "@trigger.dev/airtable";
-
-const airtable = new Airtable({
-  id: "__SLUG__"
-});
-`,
-          },
-        ],
+        samples: [usageSample(false)],
+      },
+    },
+    apiKey: {
+      type: "apikey",
+      help: {
+        samples: [usageSample(true)],
       },
     },
   },

apps/webapp/app/services/externalApis/integrations/github.ts

@@ -43,14 +43,14 @@ const usageSample: HelpSample = {
       //wrap the SDK call in runTask
       const { data } = await io.runTask(
         "create-card",
-        { name: "Create card" },
         async () => {
           //create a project card using the underlying client
           return io.github.client.rest.projects.createCard({
             column_id: 123,
             note: "test",
           });
-        }
+        },
+        { name: "Create card" }
       );
   
       //log the url of the created card

apps/webapp/app/services/runs/performRunExecutionV2.server.ts

@@ -20,6 +20,13 @@ import { logger } from "../logger.server";
 type FoundRun = NonNullable<Awaited<ReturnType<typeof findRun>>>;
 type FoundTask = FoundRun["tasks"][number];
 
+export type PerformRunExecutionV2Input = {
+  id: string;
+  reason: "PREPROCESS" | "EXECUTE_JOB";
+  isRetry: boolean;
+  resumeTaskId?: string;
+};
+
 export class PerformRunExecutionV2Service {
   #prismaClient: PrismaClient;
 
@@ -27,25 +34,20 @@ export class PerformRunExecutionV2Service {
     this.#prismaClient = prismaClient;
   }
 
-  public async call(
-    id: string,
-    reason: "PREPROCESS" | "EXECUTE_JOB",
-    isRetry: boolean = false,
-    resumeTaskId?: string
-  ) {
-    const run = await findRun(this.#prismaClient, id);
+  public async call(input: PerformRunExecutionV2Input) {
+    const run = await findRun(this.#prismaClient, input.id);
 
     if (!run) {
       return;
     }
 
-    switch (reason) {
+    switch (input.reason) {
       case "PREPROCESS": {
         await this.#executePreprocessing(run);
         break;
       }
       case "EXECUTE_JOB": {
-        await this.#executeJob(run, isRetry, resumeTaskId);
+        await this.#executeJob(run, input);
         break;
       }
     }
@@ -141,7 +143,9 @@ export class PerformRunExecutionV2Service {
       });
     }
   }
-  async #executeJob(run: FoundRun, isRetry: boolean, resumeTaskId?: string) {
+  async #executeJob(run: FoundRun, input: PerformRunExecutionV2Input) {
+    const { isRetry, resumeTaskId } = input;
+
     if (run.status === "CANCELED") {
       await this.#cancelExecution(run);
       return;
@@ -152,21 +156,27 @@ export class PerformRunExecutionV2Service {
 
     const startedAt = new Date();
 
-    await this.#prismaClient.jobRun.update({
+    const { executionCount } = await this.#prismaClient.jobRun.update({
       where: {
         id: run.id,
       },
       data: {
         status: run.status === "QUEUED" ? "STARTED" : run.status,
         startedAt: run.startedAt ?? new Date(),
+        executionCount: {
+          increment: 1,
+        },
+      },
+      select: {
+        executionCount: true,
       },
     });
 
     const connections = await resolveRunConnections(run.runConnections);
 
     if (!connections.success) {
-      return this.#failRunExecutionWithRetry({
-        message: `Could not resolve all connections for run ${run.id}, attempting to retry`,
+      return this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
+        message: `Could not resolve all connections for run ${run.id}. This should not happen`,
       });
     }
 
@@ -195,7 +205,7 @@ export class PerformRunExecutionV2Service {
 
     const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext);
 
-    const { response, parser, errorParser } = await client.executeJobRequest({
+    const { response, parser, errorParser, durationInMs } = await client.executeJobRequest({
       event,
       job: {
         id: run.version.job.slug,
@@ -261,50 +271,82 @@ export class PerformRunExecutionV2Service {
 
       // Only retry if the error isn't a 4xx
       if (response.status >= 400 && response.status <= 499 && response.status !== 408) {
-        return await this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
-          message: `Endpoint responded with ${response.status} status code`,
-        });
+        return await this.#failRunExecution(
+          this.#prismaClient,
+          "EXECUTE_JOB",
+          run,
+          {
+            message: `Endpoint responded with ${response.status} status code`,
+          },
+          "FAILURE",
+          durationInMs
+        );
       } else {
-        return await this.#failRunExecutionWithRetry({
-          message: `Endpoint responded with ${response.status} status code`,
-        });
+        // If the error is a 504 timeout, we should mark this execution as succeeded (by not throwing an error) and enqueue a new execution
+        if (response.status === 504) {
+          return await this.#resumeRunExecutionAfterTimeout(
+            this.#prismaClient,
+            run,
+            input,
+            durationInMs,
+            executionCount
+          );
+        } else {
+          return await this.#failRunExecutionWithRetry({
+            message: `Endpoint responded with ${response.status} status code`,
+          });
+        }
       }
     }
 
     const safeBody = safeJsonZodParse(parser, rawBody);
 
     if (!safeBody) {
-      return await this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
-        message: "Endpoint responded with invalid JSON",
-      });
+      return await this.#failRunExecution(
+        this.#prismaClient,
+        "EXECUTE_JOB",
+        run,
+        {
+          message: "Endpoint responded with invalid JSON",
+        },
+        "FAILURE",
+        durationInMs
+      );
     }
 
     if (!safeBody.success) {
-      return await this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
-        message: generateErrorMessage(safeBody.error.issues),
-      });
+      return await this.#failRunExecution(
+        this.#prismaClient,
+        "EXECUTE_JOB",
+        run,
+        {
+          message: generateErrorMessage(safeBody.error.issues),
+        },
+        "FAILURE",
+        durationInMs
+      );
     }
 
     const status = safeBody.data.status;
 
     switch (status) {
       case "SUCCESS": {
-        await this.#completeRunWithSuccess(run, safeBody.data);
+        await this.#completeRunWithSuccess(run, safeBody.data, durationInMs);
 
         break;
       }
       case "RESUME_WITH_TASK": {
-        await this.#resumeRunWithTask(run, safeBody.data, isRetry);
+        await this.#resumeRunWithTask(run, safeBody.data, isRetry, durationInMs, executionCount);
 
         break;
       }
       case "ERROR": {
-        await this.#failRunWithError(run, safeBody.data);
+        await this.#failRunWithError(run, safeBody.data, durationInMs);
 
         break;
       }
       case "RETRY_WITH_TASK": {
-        await this.#retryRunWithTask(run, safeBody.data, isRetry);
+        await this.#retryRunWithTask(run, safeBody.data, isRetry, durationInMs, executionCount);
 
         break;
       }
@@ -319,19 +361,37 @@ export class PerformRunExecutionV2Service {
     }
   }
 
-  async #completeRunWithSuccess(run: FoundRun, data: RunJobSuccess) {
+  async #completeRunWithSuccess(run: FoundRun, data: RunJobSuccess, durationInMs: number) {
     await this.#prismaClient.jobRun.update({
       where: { id: run.id },
       data: {
         completedAt: new Date(),
         status: "SUCCESS",
         output: data.output ?? undefined,
+        executionDuration: {
+          increment: durationInMs,
+        },
       },
     });
   }
 
-  async #resumeRunWithTask(run: FoundRun, data: RunJobResumeWithTask, isRetry: boolean) {
+  async #resumeRunWithTask(
+    run: FoundRun,
+    data: RunJobResumeWithTask,
+    isRetry: boolean,
+    durationInMs: number,
+    executionCount: number
+  ) {
     return await $transaction(this.#prismaClient, async (tx) => {
+      await tx.jobRun.update({
+        where: { id: run.id },
+        data: {
+          executionDuration: {
+            increment: durationInMs,
+          },
+        },
+      });
+
       // If the task has an operation, then the next performRunExecution will occur
       // when that operation has finished
       if (!data.task.operation) {
@@ -340,12 +400,13 @@ export class PerformRunExecutionV2Service {
           resumeTaskId: data.task.id,
           isRetry,
           skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
+          executionCount,
         });
       }
     });
   }
 
-  async #failRunWithError(execution: FoundRun, data: RunJobError) {
+  async #failRunWithError(execution: FoundRun, data: RunJobError, durationInMs: number) {
     return await $transaction(this.#prismaClient, async (tx) => {
       if (data.task) {
         await tx.task.update({
@@ -360,11 +421,24 @@ export class PerformRunExecutionV2Service {
         });
       }
 
-      await this.#failRunExecution(tx, "EXECUTE_JOB", execution, data.error ?? undefined);
+      await this.#failRunExecution(
+        tx,
+        "EXECUTE_JOB",
+        execution,
+        data.error ?? undefined,
+        "FAILURE",
+        durationInMs
+      );
     });
   }
 
-  async #retryRunWithTask(run: FoundRun, data: RunJobRetryWithTask, isRetry: boolean) {
+  async #retryRunWithTask(
+    run: FoundRun,
+    data: RunJobRetryWithTask,
+    isRetry: boolean,
+    durationInMs: number,
+    executionCount: number
+  ) {
     return await $transaction(this.#prismaClient, async (tx) => {
       // We need to check for an existing task attempt
       const existingAttempt = await tx.taskAttempt.findFirst({
@@ -405,6 +479,13 @@ export class PerformRunExecutionV2Service {
         },
         data: {
           status: "WAITING",
+          run: {
+            update: {
+              executionDuration: {
+                increment: durationInMs,
+              },
+            },
+          },
         },
       });
 
@@ -413,6 +494,55 @@ export class PerformRunExecutionV2Service {
         resumeTaskId: data.task.id,
         isRetry,
         skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
+        executionCount,
+      });
+    });
+  }
+
+  async #resumeRunExecutionAfterTimeout(
+    prisma: PrismaClientOrTransaction,
+    run: FoundRun,
+    input: PerformRunExecutionV2Input,
+    durationInMs: number,
+    executionCount: number
+  ) {
+    await $transaction(prisma, async (tx) => {
+      const executionDuration = run.executionDuration + durationInMs;
+
+      // If the execution duration is greater than the maximum execution time, we need to fail the run
+      if (executionDuration >= run.organization.maximumExecutionTimePerRunInMs) {
+        await this.#failRunExecution(
+          tx,
+          "EXECUTE_JOB",
+          run,
+          {
+            message: `Execution timed out after ${
+              run.organization.maximumExecutionTimePerRunInMs / 1000
+            } seconds`,
+          },
+          "TIMED_OUT",
+          durationInMs
+        );
+        return;
+      }
+
+      await tx.jobRun.update({
+        where: {
+          id: run.id,
+        },
+        data: {
+          executionDuration: {
+            increment: durationInMs,
+          },
+        },
+      });
+
+      // The run has timed out, so we need to enqueue a new execution
+      await enqueueRunExecutionV2(run, tx, {
+        resumeTaskId: input.resumeTaskId,
+        isRetry: input.isRetry,
+        skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
+        executionCount,
       });
     });
   }
@@ -426,7 +556,8 @@ export class PerformRunExecutionV2Service {
     reason: "EXECUTE_JOB" | "PREPROCESS",
     run: FoundRun,
     output: Record<string, any>,
-    status: "FAILURE" | "ABORTED" = "FAILURE"
+    status: "FAILURE" | "ABORTED" | "TIMED_OUT" = "FAILURE",
+    durationInMs: number = 0
   ): Promise<void> {
     await $transaction(prisma, async (tx) => {
       switch (reason) {
@@ -438,6 +569,9 @@ export class PerformRunExecutionV2Service {
               completedAt: new Date(),
               status,
               output,
+              executionDuration: {
+                increment: durationInMs,
+              },
             },
           });
 
@@ -510,38 +644,23 @@ function prepareTasksForRun(possibleTasks: FoundTask[]): CachedTask[] {
     return cachedTaskSizes.get(taskId)!;
   }
 
-  // Create a dynamic programming array to store intermediate results
-  const dp: number[][] = [];
-  for (let i = 0; i <= tasks.length; i++) {
-    dp[i] = [];
-    for (let j = 0; j <= TOTAL_CACHED_TASK_BYTE_LIMIT; j++) {
-      dp[i][j] = 0;
-    }
-  }
-
-  // Fill the dynamic programming array
-  for (let i = 1; i <= tasks.length; i++) {
-    const task = tasks[i - 1];
+  // Prepare tasks and calculate their sizes
+  const availableTasks = tasks.map((task) => {
     const cachedTask = getCachedTask(task);
-    const taskSize = getCachedTaskSize(cachedTask);
-    for (let j = 0; j <= TOTAL_CACHED_TASK_BYTE_LIMIT; j++) {
-      if (taskSize <= j) {
-        dp[i][j] = Math.max(dp[i - 1][j], dp[i - 1][j - taskSize] + taskSize);
-      } else {
-        dp[i][j] = dp[i - 1][j];
-      }
-    }
-  }
+    return { task: cachedTask, size: getCachedTaskSize(cachedTask) };
+  });
 
-  // Traverse the dynamic programming array to find the included tasks
+  // Sort tasks in ascending order by size
+  availableTasks.sort((a, b) => a.size - b.size);
+
+  // Select tasks using greedy approach
   const tasksToRun: CachedTask[] = [];
-  let j = TOTAL_CACHED_TASK_BYTE_LIMIT;
-  for (let i = tasks.length; i > 0 && j > 0; i--) {
-    if (dp[i][j] !== dp[i - 1][j]) {
-      const task = tasks[i - 1];
-      const cachedTask = getCachedTask(task);
-      tasksToRun.unshift(cachedTask);
-      j -= getCachedTaskSize(cachedTask);
+  let remainingSize = TOTAL_CACHED_TASK_BYTE_LIMIT;
+
+  for (const { task, size } of availableTasks) {
+    if (size <= remainingSize) {
+      tasksToRun.push(task);
+      remainingSize -= size;
     }
   }
 
@@ -571,7 +690,6 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
       endpoint: true,
       organization: true,
       externalAccount: true,
-      queue: true,
       runConnections: {
         include: {
           integration: true,
@@ -588,6 +706,14 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
             in: ["COMPLETED"],
           },
         },
+        select: {
+          id: true,
+          idempotencyKey: true,
+          status: true,
+          noop: true,
+          output: true,
+          parentId: true,
+        },
       },
       event: true,
       version: {

apps/webapp/app/services/secrets/secretStore.server.ts

@@ -19,6 +19,7 @@ type ProviderInitializationOptions = {
 export interface SecretStoreProvider {
   getSecret<T>(schema: z.Schema<T>, key: string): Promise<T | undefined>;
   setSecret<T extends object>(key: string, value: T): Promise<void>;
+  deleteSecret(key: string): Promise<boolean>;
 }
 
 /** The SecretStore will use the passed in provider. We do NOT recommend using "DATABASE" outside of localhost. */
@@ -42,6 +43,10 @@ export class SecretStore {
   setSecret<T extends object>(key: string, value: T): Promise<void> {
     return this.provider.setSecret(key, value);
   }
+
+  deleteSecret<T extends object>(key: string): Promise<boolean> {
+    return this.provider.deleteSecret(key);
+  }
 }
 
 const EncryptedSecretValueSchema = z.object({
@@ -116,6 +121,16 @@ class PrismaSecretStore implements SecretStoreProvider {
     });
   }
 
+  async deleteSecret(key: string): Promise<boolean> {
+    const result = await this.#prismaClient.secretStore.delete({
+      where: {
+        key,
+      },
+    });
+
+    return !!result;
+  }
+
   async #decrypt(nonce: string, ciphertext: string, tag: string): Promise<string> {
     const decipher = nodeCrypto.createDecipheriv(
       "aes-256-gcm",
@@ -154,7 +169,7 @@ class PrismaSecretStore implements SecretStoreProvider {
 
 export function getSecretStore<
   K extends SecretStoreOptions,
-  TOptions extends ProviderInitializationOptions[K],
+  TOptions extends ProviderInitializationOptions[K]
 >(provider: K, options?: TOptions): SecretStore {
   switch (provider) {
     case "DATABASE": {

apps/webapp/app/services/sources/activateSource.server.ts

@@ -1,13 +1,21 @@
+import {
+  REGISTER_SOURCE_EVENT_V2,
+  REGISTER_SOURCE_EVENT_V1,
+  RegisterTriggerSource,
+  RegisterSourceEventV1,
+  RegisterSourceEventV2,
+  RegisterSourceEventOptions,
+  RegisteredOptionsDiff,
+} from "@trigger.dev/core";
+import type { SecretReference, TriggerSource, TriggerSourceOption } from "@trigger.dev/database";
+import { z } from "zod";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
-import type { TriggerSource, SecretReference, TriggerSourceEvent } from "@trigger.dev/database";
-import type { AuthenticatedEnvironment } from "../apiAuth.server";
 import { env } from "~/env.server";
-import { REGISTER_SOURCE_EVENT, RegisterTriggerSource } from "@trigger.dev/core";
-import { SecretStoreProvider, getSecretStore } from "../secrets/secretStore.server";
-import { SecretStore } from "../secrets/secretStore.server";
-import { z } from "zod";
+import type { AuthenticatedEnvironment } from "../apiAuth.server";
 import { IngestSendEvent } from "../events/ingestSendEvent.server";
+import { getSecretStore } from "../secrets/secretStore.server";
+import { nanoid } from "nanoid";
 
 export class ActivateSourceService {
   #prismaClient: PrismaClient;
@@ -16,7 +24,7 @@ export class ActivateSourceService {
     this.#prismaClient = prismaClient;
   }
 
-  public async call(id: string, jobId: string, orphanedEvents?: Array<string>) {
+  public async call(id: string, jobId?: string, orphanedOptions?: Record<string, string[]>) {
     const triggerSource = await this.#prismaClient.triggerSource.findUniqueOrThrow({
       where: {
         id,
@@ -29,12 +37,12 @@ export class ActivateSourceService {
             project: true,
           },
         },
-        events: true,
+        options: true,
         secretReference: true,
       },
     });
 
-    const eventId = `${id}:${jobId}`;
+    const eventId = `${id}:${jobId ?? nanoid()}`;
 
     // TODO: support more channels
     switch (triggerSource.channel) {
@@ -42,10 +50,10 @@ export class ActivateSourceService {
         await this.#activateHttpSource(
           triggerSource.environment,
           triggerSource,
-          triggerSource.events,
+          triggerSource.options,
           triggerSource.secretReference,
           eventId,
-          orphanedEvents
+          orphanedOptions
         );
       }
     }
@@ -54,10 +62,10 @@ export class ActivateSourceService {
   async #activateHttpSource(
     environment: AuthenticatedEnvironment,
     triggerSource: TriggerSource,
-    events: Array<TriggerSourceEvent>,
+    options: Array<TriggerSourceOption>,
     secretReference: SecretReference,
     eventId: string,
-    orphanedEvents?: Array<string>
+    orphanedOptions?: Record<string, string[]>
   ) {
     const secretStore = getSecretStore(secretReference.provider);
     const httpSecret = await secretStore.getSecret(
@@ -71,14 +79,6 @@ export class ActivateSourceService {
       throw new Error("HTTP Secret not found");
     }
 
-    const eventNames = triggerSource.active
-      ? events.filter((e) => e.registered).map((e) => e.name)
-      : events.map((e) => e.name);
-
-    const missingEvents = triggerSource.active
-      ? events.filter((e) => !e.registered).map((e) => e.name)
-      : [];
-
     const service = new IngestSendEvent();
 
     const source: RegisterTriggerSource = {
@@ -92,17 +92,85 @@ export class ActivateSourceService {
       },
     };
 
-    await service.call(environment, {
-      id: eventId,
-      name: REGISTER_SOURCE_EVENT,
-      source: "trigger.dev",
-      payload: {
-        id: triggerSource.id,
-        source,
-        events: eventNames,
-        missingEvents,
-        orphanedEvents: orphanedEvents ?? [],
-      },
-    });
+    switch (triggerSource.version) {
+      case "1": {
+        const events = triggerSource.active
+          ? options.filter((e) => e.registered).map((e) => e.value)
+          : options.map((e) => e.value);
+        const missingEvents = triggerSource.active
+          ? options.filter((e) => !e.registered).map((e) => e.value)
+          : [];
+        const orphanedEvents = orphanedOptions
+          ? Object.values(orphanedOptions).flatMap((vals) => vals)
+          : [];
+
+        const payload: RegisterSourceEventV1 = {
+          id: triggerSource.id,
+          source,
+          events,
+          missingEvents,
+          orphanedEvents,
+        };
+
+        await service.call(environment, {
+          id: eventId,
+          name: REGISTER_SOURCE_EVENT_V1,
+          source: "trigger.dev",
+          payload,
+        });
+        break;
+      }
+      case "2": {
+        //group the options by the name
+        const optionsRecord = options.reduce((acc, option) => {
+          if (!acc[option.name]) {
+            acc[option.name] = [];
+          }
+
+          acc[option.name].push(option);
+
+          return acc;
+        }, {} as Record<string, Array<TriggerSourceOption>>);
+
+        //for each of the optionsRecord, create the diff
+        const payloadOptions = Object.entries(optionsRecord).reduce(
+          (acc, [key, value]) => ({
+            ...acc,
+            [key]: getOptionsDiff(triggerSource.active, orphanedOptions?.[key] ?? [], value),
+          }),
+          {} as Record<string, RegisteredOptionsDiff>
+        ) as RegisterSourceEventOptions;
+
+        const payload: RegisterSourceEventV2 = {
+          id: triggerSource.id,
+          source,
+          options: payloadOptions,
+        };
+
+        await service.call(environment, {
+          id: eventId,
+          name: REGISTER_SOURCE_EVENT_V2,
+          source: "trigger.dev",
+          payload,
+        });
+      }
+    }
   }
 }
+
+function getOptionsDiff(
+  sourceIsActive: boolean,
+  orphaned: string[],
+  options: Array<TriggerSourceOption>
+): RegisteredOptionsDiff {
+  const desired = sourceIsActive
+    ? options.filter((e) => e.registered).map((e) => e.value)
+    : options.map((e) => e.value);
+  const missing = sourceIsActive ? options.filter((e) => !e.registered).map((e) => e.value) : [];
+
+  return {
+    desired: [...new Set(desired)],
+    missing: [...new Set(missing)],
+    orphaned: [...new Set(orphaned)],
+  };
+}

apps/webapp/app/services/sources/deliverHttpSourceRequest.server.ts

@@ -4,6 +4,9 @@ import { prisma } from "~/db.server";
 import { EndpointApi } from "../endpointApi.server";
 import { IngestSendEvent } from "../events/ingestSendEvent.server";
 import { getSecretStore } from "../secrets/secretStore.server";
+import { resolveApiConnection, resolveRunConnection } from "~/models/runConnection.server";
+import { ConnectionAuth } from "@trigger.dev/sdk";
+import { resolveSourceConnection } from "~/models/sourceConnection.server";
 
 export class DeliverHttpSourceRequestService {
   #prismaClient: PrismaClient;
@@ -28,7 +31,11 @@ export class DeliverHttpSourceRequestService {
             secretReference: true,
             dynamicTrigger: true,
             externalAccount: true,
-            integration: true,
+            integration: {
+              include: {
+                connections: true,
+              },
+            },
           },
         },
       },
@@ -51,14 +58,14 @@ export class DeliverHttpSourceRequestService {
       throw new Error(`Secret not found for ${httpSourceRequest.source.key}`);
     }
 
-    // TODO: implement auth for http source requests
+    const auth = await resolveSourceConnection(this.#prismaClient, httpSourceRequest.source);
 
     const clientApi = new EndpointApi(
       httpSourceRequest.environment.apiKey,
       httpSourceRequest.endpoint.url
     );
 
-    const { response, events } = await clientApi.deliverHttpSourceRequest({
+    const { response, events, metadata } = await clientApi.deliverHttpSourceRequest({
       key: httpSourceRequest.source.key,
       dynamicId: httpSourceRequest.source.dynamicTrigger?.slug,
       secret: secret.secret,
@@ -70,6 +77,8 @@ export class DeliverHttpSourceRequestService {
         headers: httpSourceRequest.headers as Record<string, string>,
         rawBody: httpSourceRequest.body,
       },
+      auth,
+      metadata: httpSourceRequest.source.metadata,
     });
 
     await this.#prismaClient.httpSourceRequestDelivery.update({
@@ -81,6 +90,17 @@ export class DeliverHttpSourceRequestService {
       },
     });
 
+    if (metadata) {
+      await this.#prismaClient.triggerSource.update({
+        where: {
+          id: httpSourceRequest.source.id,
+        },
+        data: {
+          metadata: metadata,
+        },
+      });
+    }
+
     const ingestService = new IngestSendEvent();
 
     for (const event of events) {

apps/webapp/app/services/sources/handleHttpSource.server.ts

@@ -54,7 +54,7 @@ export class HandleHttpSourceService {
             id: delivery.id,
           },
           {
-            queueName: `endpoint-${triggerSource.endpointId}`,
+            queueName: `deliver:${triggerSource.id}`,
             tx,
             maxAttempts:
               triggerSource.environment.type === RuntimeEnvironmentType.DEVELOPMENT ? 1 : undefined,

apps/webapp/app/services/sources/registerSourceV1.server.ts

@@ -1,5 +1,5 @@
 import type { Endpoint } from "@trigger.dev/database";
-import type { SourceMetadata } from "@trigger.dev/core";
+import type { SourceMetadataV1 } from "@trigger.dev/core";
 import { $transaction, PrismaClientOrTransaction } from "~/db.server";
 import { prisma } from "~/db.server";
 import type { AuthenticatedEnvironment } from "../apiAuth.server";
@@ -7,7 +7,7 @@ import { workerQueue } from "../worker.server";
 import { generateSecret } from "./utils.server";
 import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
 
-export class RegisterSourceService {
+export class RegisterSourceServiceV1 {
   #prismaClient: PrismaClientOrTransaction;
 
   constructor(prismaClient: PrismaClientOrTransaction = prisma) {
@@ -16,7 +16,7 @@ export class RegisterSourceService {
 
   public async call(
     endpointIdOrEndpoint: string | ExtendedEndpoint,
-    metadata: SourceMetadata,
+    metadata: SourceMetadataV1,
     dynamicTriggerId?: string,
     accountId?: string,
     dynamicSource?: { id: string; metadata: any }
@@ -39,7 +39,7 @@ export class RegisterSourceService {
   async #upsertSource(
     endpoint: Endpoint,
     environment: AuthenticatedEnvironment,
-    metadata: SourceMetadata,
+    metadata: SourceMetadataV1,
     dynamicTriggerId?: string,
     accountId?: string,
     dynamicSource?: { id: string; metadata: any }
@@ -121,9 +121,10 @@ export class RegisterSourceService {
                 }
               : undefined,
             externalAccount: externalAccount ? { connect: { id: externalAccount.id } } : undefined,
-            events: {
+            options: {
               create: metadata.events.map((event) => ({
-                name: event,
+                name: "event",
+                value: event,
               })),
             },
             secretReference: {
@@ -175,7 +176,7 @@ export class RegisterSourceService {
                 : undefined,
           },
           include: {
-            events: true,
+            options: true,
             secretReference: true,
           },
         });
@@ -200,22 +201,24 @@ export class RegisterSourceService {
         const newEvents = new Set<string>(metadata.events);
         const orphanedEvents = new Set<string>();
 
-        for (const event of triggerSource.events) {
-          if (!newEvents.has(event.name)) {
-            orphanedEvents.add(event.name);
+        for (const option of triggerSource.options) {
+          if (!newEvents.has(option.value)) {
+            orphanedEvents.add(option.value);
           }
         }
 
         for (const event of newEvents) {
-          await tx.triggerSourceEvent.upsert({
+          await tx.triggerSourceOption.upsert({
             where: {
-              name_sourceId: {
-                name: event,
+              name_value_sourceId: {
+                name: "event",
+                value: event,
                 sourceId: triggerSource.id,
               },
             },
             create: {
-              name: event,
+              name: "event",
+              value: event,
               source: {
                 connect: {
                   id: triggerSource.id,
@@ -249,7 +252,7 @@ export class RegisterSourceService {
         id: id,
       },
       include: {
-        events: true,
+        options: true,
         secretReference: true,
         integration: true,
       },
@@ -261,11 +264,12 @@ export class RegisterSourceService {
 
     const triggerIsActive = triggerSource.active;
     const triggerHasOrphanedEvents = orphanedEvents.length > 0;
-    const triggerHasUnregisteredEvents = triggerSource.events.some((event) => !event.registered);
+    const triggerHasUnregisteredEvents = triggerSource.options.some((option) => !option.registered);
 
     if (!triggerIsActive || triggerHasOrphanedEvents || triggerHasUnregisteredEvents) {
       // We need to re-activate the source, and there could be orphaned events
       await workerQueue.enqueue("activateSource", {
+        version: "1",
         id: triggerSource.id,
         orphanedEvents: orphanedEvents,
       });
@@ -277,7 +281,7 @@ export class RegisterSourceService {
   async #findOrCreateIntegration(
     tx: PrismaClientOrTransaction,
     organizationId: string,
-    config: SourceMetadata["integration"]
+    config: SourceMetadataV1["integration"]
   ) {
     if (config.authSource === "HOSTED") {
       return tx.integration.findUnique({

apps/webapp/app/services/sources/registerSourceV2.server.ts

@@ -0,0 +1,361 @@
+import type { Endpoint } from "@trigger.dev/database";
+import type { SourceMetadataV2 } from "@trigger.dev/core";
+import { $transaction, PrismaClientOrTransaction } from "~/db.server";
+import { prisma } from "~/db.server";
+import type { AuthenticatedEnvironment } from "../apiAuth.server";
+import { workerQueue } from "../worker.server";
+import { generateSecret } from "./utils.server";
+import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
+
+export class RegisterSourceServiceV2 {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    endpointIdOrEndpoint: string | ExtendedEndpoint,
+    metadata: SourceMetadataV2,
+    dynamicTriggerId?: string,
+    accountId?: string,
+    dynamicSource?: { id: string; metadata: any }
+  ) {
+    const endpoint =
+      typeof endpointIdOrEndpoint === "string"
+        ? await findEndpoint(endpointIdOrEndpoint)
+        : endpointIdOrEndpoint;
+
+    return this.#upsertSource(
+      endpoint,
+      endpoint.environment,
+      metadata,
+      dynamicTriggerId,
+      accountId,
+      dynamicSource
+    );
+  }
+
+  async #upsertSource(
+    endpoint: Endpoint,
+    environment: AuthenticatedEnvironment,
+    metadata: SourceMetadataV2,
+    dynamicTriggerId?: string,
+    accountId?: string,
+    dynamicSource?: { id: string; metadata: any }
+  ) {
+    const key = [dynamicTriggerId, dynamicSource?.id, metadata.key].filter(Boolean).join(":");
+
+    const registrationJob = metadata.registerSourceJob
+      ? await this.#prismaClient.job.findUnique({
+          where: {
+            projectId_slug: {
+              projectId: endpoint.projectId,
+              slug: metadata.registerSourceJob.id,
+            },
+          },
+        })
+      : undefined;
+
+    const source = await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        const integration = await this.#findOrCreateIntegration(
+          tx,
+          environment.organizationId,
+          metadata.integration
+        );
+
+        if (!integration) {
+          throw new Error("Integration not found");
+        }
+
+        const externalAccount = accountId
+          ? await tx.externalAccount.findUniqueOrThrow({
+              where: {
+                environmentId_identifier: {
+                  environmentId: environment.id,
+                  identifier: accountId,
+                },
+              },
+            })
+          : undefined;
+
+        // options
+        const createOptions = Object.entries(metadata.options).flatMap(([name, values]) => {
+          const uniqueValues = [...new Set(values)];
+          return uniqueValues.map((value) => ({ name, value }));
+        });
+
+        const triggerSource = await tx.triggerSource.upsert({
+          where: {
+            key_environmentId: {
+              environmentId: environment.id,
+              key,
+            },
+          },
+          create: {
+            version: "2",
+            params: metadata.params,
+            key,
+            channel: metadata.channel,
+            organization: {
+              connect: {
+                id: environment.organizationId,
+              },
+            },
+            endpoint: {
+              connect: {
+                id: endpoint.id,
+              },
+            },
+            project: {
+              connect: {
+                id: environment.projectId,
+              },
+            },
+            environment: {
+              connect: {
+                id: environment.id,
+              },
+            },
+            integration: { connect: { id: integration.id } },
+            dynamicTrigger: dynamicTriggerId
+              ? {
+                  connect: {
+                    id: dynamicTriggerId,
+                  },
+                }
+              : undefined,
+            externalAccount: externalAccount ? { connect: { id: externalAccount.id } } : undefined,
+            options: {
+              create: createOptions,
+            },
+            secretReference: {
+              connectOrCreate: {
+                where: {
+                  key: `${endpoint.id}:${key}`,
+                },
+                create: {
+                  key: `${endpoint.id}:${key}`,
+                  provider: "DATABASE",
+                },
+              },
+            },
+            dynamicSourceId: dynamicSource?.id,
+            dynamicSourceMetadata: dynamicSource?.metadata,
+            sourceRegistrationJob:
+              registrationJob && metadata.registerSourceJob
+                ? {
+                    connect: {
+                      jobId_version_environmentId: {
+                        jobId: registrationJob.id,
+                        version: metadata.registerSourceJob.version,
+                        environmentId: endpoint.environmentId,
+                      },
+                    },
+                  }
+                : undefined,
+          },
+          update: {
+            version: "2",
+            endpoint: {
+              connect: {
+                id: endpoint.id,
+              },
+            },
+            integration: { connect: { id: integration.id } },
+            dynamicSourceId: dynamicSource?.id,
+            dynamicSourceMetadata: dynamicSource?.metadata,
+            sourceRegistrationJob:
+              registrationJob && metadata.registerSourceJob
+                ? {
+                    connect: {
+                      jobId_version_environmentId: {
+                        jobId: registrationJob.id,
+                        version: metadata.registerSourceJob.version,
+                        environmentId: endpoint.environmentId,
+                      },
+                    },
+                  }
+                : undefined,
+          },
+          include: {
+            options: true,
+            secretReference: true,
+          },
+        });
+
+        switch (metadata.channel) {
+          case "HTTP": {
+            await tx.secretStore.upsert({
+              where: {
+                key: triggerSource.secretReference.key,
+              },
+              create: {
+                key: triggerSource.secretReference.key,
+                value: {
+                  secret: generateSecret(),
+                },
+              },
+              update: {},
+            });
+          }
+        }
+
+        // Collect the options that are no longer being used so we can remove them
+        const newOptions = metadata.options;
+        const orphanedOptions: Record<string, string[]> = {};
+        for (const event of triggerSource.options) {
+          const values = newOptions[event.name];
+          if (values === undefined) {
+            orphanedOptions[event.name] = [event.value];
+            continue;
+          }
+
+          if (values!.includes(event.value)) {
+            orphanedOptions[event.name] = [...values, event.value];
+          }
+        }
+
+        //add or update the options
+        const flatOptions = Object.entries(newOptions).flatMap(([name, values]) =>
+          values.map((v) => ({ name, value: v }))
+        );
+        for (const { name, value } of flatOptions) {
+          await tx.triggerSourceOption.upsert({
+            where: {
+              name_value_sourceId: {
+                name,
+                value,
+                sourceId: triggerSource.id,
+              },
+            },
+            create: {
+              name,
+              value,
+              source: {
+                connect: {
+                  id: triggerSource.id,
+                },
+              },
+            },
+            update: {},
+          });
+        }
+
+        return {
+          id: triggerSource.id,
+          orphanedOptions,
+        };
+      },
+      { timeout: 15000 }
+    );
+
+    if (!source) {
+      return;
+    }
+
+    const { id, orphanedOptions } = source;
+
+    // We need to activate the source if:
+    // 1. It's not active
+    // 2. There are orphaned events
+    // 3. There are trigger events that are not registered
+    const triggerSource = await this.#prismaClient.triggerSource.findUniqueOrThrow({
+      where: {
+        id: id,
+      },
+      include: {
+        options: true,
+        secretReference: true,
+        integration: true,
+      },
+    });
+
+    if (dynamicTriggerId) {
+      return triggerSource;
+    }
+
+    const triggerIsActive = triggerSource.active;
+    const triggerHasOrphanedEvents = Object.keys(orphanedOptions).length > 0;
+    const triggerHasUnregisteredEvents = triggerSource.options.some((option) => !option.registered);
+
+    if (!triggerIsActive || triggerHasOrphanedEvents || triggerHasUnregisteredEvents) {
+      // We need to re-activate the source, and there could be orphaned events
+      await workerQueue.enqueue("activateSource", {
+        version: "2",
+        id: triggerSource.id,
+        orphanedOptions,
+      });
+    }
+
+    return triggerSource;
+  }
+
+  async #findOrCreateIntegration(
+    tx: PrismaClientOrTransaction,
+    organizationId: string,
+    config: SourceMetadataV2["integration"]
+  ) {
+    if (config.authSource === "HOSTED") {
+      return tx.integration.findUnique({
+        where: {
+          organizationId_slug: {
+            organizationId,
+            slug: config.id,
+          },
+        },
+      });
+    } else {
+      return tx.integration.upsert({
+        where: {
+          organizationId_slug: {
+            organizationId,
+            slug: config.id,
+          },
+        },
+        create: {
+          slug: config.id,
+          title: config.metadata.name,
+          authSource: "LOCAL",
+          connectionType: "DEVELOPER",
+          organization: {
+            connect: {
+              id: organizationId,
+            },
+          },
+          definition: {
+            connectOrCreate: {
+              where: {
+                id: config.metadata.id,
+              },
+              create: {
+                id: config.metadata.id,
+                name: config.metadata.name,
+                instructions: config.metadata.instructions,
+              },
+            },
+          },
+        },
+        update: {
+          title: config.metadata.name,
+          authSource: "LOCAL",
+          connectionType: "DEVELOPER",
+          definition: {
+            connectOrCreate: {
+              where: {
+                id: config.metadata.id,
+              },
+              create: {
+                id: config.metadata.id,
+                name: config.metadata.name,
+                instructions: config.metadata.instructions,
+              },
+            },
+          },
+        },
+      });
+    }
+  }
+}

apps/webapp/app/services/sources/updateSourceV1.server.ts

@@ -1,10 +1,10 @@
 import type { RuntimeEnvironment } from "@trigger.dev/database";
-import type { TriggerSource, UpdateTriggerSourceBody } from "@trigger.dev/core";
+import type { TriggerSource, UpdateTriggerSourceBodyV1 } from "@trigger.dev/core";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
 import { getSecretStore } from "../secrets/secretStore.server";
 
-export class UpdateSourceService {
+export class UpdateSourceServiceV1 {
   #prismaClient: PrismaClient;
 
   constructor(prismaClient: PrismaClient = prisma) {
@@ -18,7 +18,7 @@ export class UpdateSourceService {
     endpointSlug,
   }: {
     environment: RuntimeEnvironment;
-    payload: UpdateTriggerSourceBody;
+    payload: UpdateTriggerSourceBodyV1;
     id: string;
     endpointSlug: string;
   }): Promise<TriggerSource> {
@@ -55,10 +55,11 @@ export class UpdateSourceService {
     });
 
     for (const event of payload.registeredEvents) {
-      await this.#prismaClient.triggerSourceEvent.update({
+      await this.#prismaClient.triggerSourceOption.update({
         where: {
-          name_sourceId: {
-            name: event,
+          name_value_sourceId: {
+            name: "event",
+            value: event,
             sourceId: triggerSource.id,
           },
         },

apps/webapp/app/services/sources/updateSourceV2.server.ts

@@ -0,0 +1,93 @@
+import type { TriggerSource, UpdateTriggerSourceBodyV2 } from "@trigger.dev/core";
+import type { RuntimeEnvironment } from "@trigger.dev/database";
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+import { getSecretStore } from "../secrets/secretStore.server";
+
+export class UpdateSourceServiceV2 {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call({
+    environment,
+    payload,
+    id,
+    endpointSlug,
+  }: {
+    environment: RuntimeEnvironment;
+    payload: UpdateTriggerSourceBodyV2;
+    id: string;
+    endpointSlug: string;
+  }): Promise<TriggerSource> {
+    const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({
+      where: {
+        environmentId_slug: {
+          environmentId: environment.id,
+          slug: endpointSlug,
+        },
+      },
+    });
+
+    const triggerSource = await this.#prismaClient.triggerSource.findUniqueOrThrow({
+      where: {
+        key_environmentId: {
+          environmentId: environment.id,
+          key: id,
+        },
+      },
+      include: {
+        secretReference: true,
+      },
+    });
+
+    await this.#prismaClient.triggerSource.update({
+      where: {
+        id: triggerSource.id,
+      },
+      data: {
+        active: true,
+        channelData: payload.data as any,
+        endpointId: endpoint.id,
+      },
+    });
+
+    const flatOptions = Object.entries(payload.options).flatMap(([name, value]) =>
+      value.map((v) => ({
+        name,
+        value: v,
+      }))
+    );
+
+    for (const { name, value } of flatOptions) {
+      await this.#prismaClient.triggerSourceOption.update({
+        where: {
+          name_value_sourceId: {
+            name,
+            value,
+            sourceId: triggerSource.id,
+          },
+        },
+        data: {
+          registered: true,
+        },
+      });
+    }
+
+    if (payload.secret) {
+      // We need to update the secret reference in the store
+      const secretStore = getSecretStore(triggerSource.secretReference.provider);
+
+      await secretStore.setSecret<{ secret: string }>(triggerSource.secretReference.key, {
+        secret: payload.secret,
+      });
+    }
+
+    return {
+      id: triggerSource.id,
+      key: triggerSource.key,
+    };
+  }
+}

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

@@ -1,5 +1,3 @@
-import { env } from "process";
-import { Run } from "~/presenters/RunPresenter.server";
 import {
   FetchOperationSchema,
   FetchRequestInit,
@@ -11,13 +9,13 @@ import {
 import { RuntimeEnvironmentType, type Task } from "@trigger.dev/database";
 import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
 import { enqueueRunExecutionV2 } from "~/models/jobRunExecution.server";
+import { KitchenSinkTask, findKitchenSinkTask } from "~/models/task.server";
 import { formatUnknownError } from "~/utils/formatErrors.server";
 import { safeJsonFromResponse } from "~/utils/json";
+import { InitializeBackgroundTaskOperationService } from "../backgroundTasks/initializeBackgroundTaskOperation.server";
 import { logger } from "../logger.server";
 import { workerQueue } from "../worker.server";
 
-type FoundTask = Awaited<ReturnType<typeof findTask>>;
-
 export class PerformTaskOperationService {
   #prismaClient: PrismaClient;
 
@@ -26,7 +24,7 @@ export class PerformTaskOperationService {
   }
 
   public async call(id: string) {
-    const task = await findTask(this.#prismaClient, id);
+    const task = await findKitchenSinkTask(this.#prismaClient, id);
 
     if (!task) {
       return;
@@ -95,6 +93,11 @@ export class PerformTaskOperationService {
 
         return await this.#resumeTask(task, jsonBody);
       }
+      case "backgroundTask": {
+        const service = new InitializeBackgroundTaskOperationService();
+
+        return await service.call(task);
+      }
       default: {
         await this.#resumeTaskWithError(task, {
           message: `Unknown operation: ${task.operation}`,
@@ -104,7 +107,7 @@ export class PerformTaskOperationService {
   }
 
   #calculateRetryForResponse(
-    task: NonNullable<FoundTask>,
+    task: NonNullable<KitchenSinkTask>,
     retry: FetchRetryOptions | undefined,
     response: Response
   ): Date | undefined {
@@ -194,7 +197,7 @@ export class PerformTaskOperationService {
     });
   }
 
-  async #resumeTaskWithError(task: NonNullable<FoundTask>, output: any) {
+  async #resumeTaskWithError(task: NonNullable<KitchenSinkTask>, output: any) {
     await $transaction(this.#prismaClient, async (tx) => {
       await tx.task.update({
         where: { id: task.id },
@@ -220,7 +223,7 @@ export class PerformTaskOperationService {
     });
   }
 
-  async #resumeTask(task: NonNullable<FoundTask>, output: any) {
+  async #resumeTask(task: NonNullable<KitchenSinkTask>, output: any) {
     await $transaction(this.#prismaClient, async (tx) => {
       await tx.taskAttempt.updateMany({
         where: {
@@ -245,7 +248,7 @@ export class PerformTaskOperationService {
     });
   }
 
-  async #resumeRunExecution(task: NonNullable<FoundTask>, prisma: PrismaClientOrTransaction) {
+  async #resumeRunExecution(task: NonNullable<KitchenSinkTask>, prisma: PrismaClientOrTransaction) {
     await enqueueRunExecutionV2(task.run, prisma, {
       skipRetrying: task.run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
     });
@@ -278,21 +281,6 @@ function hydrateRedactedString(value: RedactString): string {
   return result;
 }
 
-async function findTask(prisma: PrismaClient, id: string) {
-  return prisma.task.findUnique({
-    where: { id },
-    include: {
-      attempts: true,
-      run: {
-        include: {
-          environment: true,
-          queue: true,
-        },
-      },
-    },
-  });
-}
-
 // Add a random number of ms between 0ms and 5000ms
 function addJitterInMs() {
   return Math.floor(Math.random() * 5000);

apps/webapp/app/services/triggers/initializeTrigger.server.ts

@@ -1,14 +1,14 @@
-import { InitializeTriggerBody, REGISTER_SOURCE_EVENT } from "@trigger.dev/core";
+import { InitializeTriggerBody, REGISTER_SOURCE_EVENT_V1 } from "@trigger.dev/core";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "../apiAuth.server";
 import { EndpointApi } from "../endpointApi.server";
-import { RegisterTriggerSourceService } from "./registerTriggerSource.server";
+import { RegisterTriggerSourceServiceV1 } from "./registerTriggerSourceV1.server";
 import { IngestSendEvent } from "../events/ingestSendEvent.server";
 
 export class InitializeTriggerService {
   #prismaClient: PrismaClient;
-  #registerTriggerSource = new RegisterTriggerSourceService();
+  #registerTriggerSource = new RegisterTriggerSourceServiceV1();
   #sendEvent = new IngestSendEvent();
 
   constructor(prismaClient: PrismaClient = prisma) {
@@ -71,7 +71,7 @@ export class InitializeTriggerService {
       environment,
       {
         id: registration.id,
-        name: REGISTER_SOURCE_EVENT,
+        name: REGISTER_SOURCE_EVENT_V1,
         source: "trigger.dev",
         payload: {
           ...registration,

apps/webapp/app/services/triggers/registerTriggerSourceV1.server.ts

@@ -1,12 +1,12 @@
-import { RegisterSourceEvent, RegisterTriggerBody } from "@trigger.dev/core";
+import { RegisterSourceEventV1, RegisterTriggerBodyV1 } from "@trigger.dev/core";
 import { z } from "zod";
 import { $transaction, PrismaClient, prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "../apiAuth.server";
 import { getSecretStore } from "../secrets/secretStore.server";
-import { RegisterSourceService } from "../sources/registerSource.server";
+import { RegisterSourceServiceV1 } from "../sources/registerSourceV1.server";
 
-export class RegisterTriggerSourceService {
+export class RegisterTriggerSourceServiceV1 {
   #prismaClient: PrismaClient;
 
   constructor(prismaClient: PrismaClient = prisma) {
@@ -23,13 +23,13 @@ export class RegisterTriggerSourceService {
     registrationMetadata,
   }: {
     environment: AuthenticatedEnvironment;
-    payload: RegisterTriggerBody;
+    payload: RegisterTriggerBodyV1;
     id: string;
     endpointSlug: string;
     key: string;
     accountId?: string;
     registrationMetadata?: any;
-  }): Promise<RegisterSourceEvent | undefined> {
+  }): Promise<RegisterSourceEventV1 | undefined> {
     const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({
       where: {
         environmentId_slug: {
@@ -52,7 +52,7 @@ export class RegisterTriggerSourceService {
     return await $transaction(
       this.#prismaClient,
       async (tx) => {
-        const service = new RegisterSourceService(tx);
+        const service = new RegisterSourceServiceV1(tx);
 
         const triggerSource = await service.call(
           endpoint.id,
@@ -76,7 +76,7 @@ export class RegisterTriggerSourceService {
           create: {
             dispatchableId: triggerSource.id,
             environmentId: environment.id,
-            event: Array.isArray(payload.rule.event) ? payload.rule.event : [payload.rule.event], 
+            event: Array.isArray(payload.rule.event) ? payload.rule.event : [payload.rule.event],
             source: payload.rule.source,
             payloadFilter: payload.rule.payload,
             contextFilter: payload.rule.context,
@@ -141,7 +141,7 @@ export class RegisterTriggerSourceService {
             },
             clientId: triggerSource.integration.slug,
           },
-          events: triggerSource.events.map((e) => e.name),
+          events: triggerSource.options.map((e) => e.value),
           missingEvents: [],
           orphanedEvents: [],
         };

apps/webapp/app/services/triggers/registerTriggerSourceV2.server.ts

@@ -0,0 +1,170 @@
+import {
+  RegisterSourceEventOptions,
+  RegisterSourceEventV2,
+  RegisterTriggerBodyV2,
+  RegisteredOptionsDiff,
+} from "@trigger.dev/core";
+import { z } from "zod";
+import { $transaction, PrismaClient, prisma } from "~/db.server";
+import { env } from "~/env.server";
+import { AuthenticatedEnvironment } from "../apiAuth.server";
+import { getSecretStore } from "../secrets/secretStore.server";
+import { RegisterSourceServiceV2 } from "../sources/registerSourceV2.server";
+
+export class RegisterTriggerSourceServiceV2 {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call({
+    environment,
+    payload,
+    endpointSlug,
+    id,
+    key,
+    accountId,
+    registrationMetadata,
+  }: {
+    environment: AuthenticatedEnvironment;
+    payload: RegisterTriggerBodyV2;
+    id: string;
+    endpointSlug: string;
+    key: string;
+    accountId?: string;
+    registrationMetadata?: any;
+  }): Promise<RegisterSourceEventV2 | undefined> {
+    const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({
+      where: {
+        environmentId_slug: {
+          environmentId: environment.id,
+          slug: endpointSlug,
+        },
+      },
+    });
+
+    const dynamicTrigger = await this.#prismaClient.dynamicTrigger.findUniqueOrThrow({
+      where: {
+        endpointId_slug_type: {
+          endpointId: endpoint.id,
+          slug: id,
+          type: "EVENT",
+        },
+      },
+    });
+
+    return await $transaction(
+      this.#prismaClient,
+      async (tx) => {
+        const service = new RegisterSourceServiceV2(tx);
+
+        const triggerSource = await service.call(
+          endpoint.id,
+          payload.source,
+          dynamicTrigger.id,
+          accountId,
+          { id: key, metadata: registrationMetadata }
+        );
+
+        if (!triggerSource) {
+          return;
+        }
+
+        const eventDispatcher = await tx.eventDispatcher.upsert({
+          where: {
+            dispatchableId_environmentId: {
+              dispatchableId: triggerSource.id,
+              environmentId: environment.id,
+            },
+          },
+          create: {
+            dispatchableId: triggerSource.id,
+            environmentId: environment.id,
+            event: Array.isArray(payload.rule.event) ? payload.rule.event : [payload.rule.event],
+            source: payload.rule.source,
+            payloadFilter: payload.rule.payload,
+            contextFilter: payload.rule.context,
+            dispatchable: {
+              type: "DYNAMIC_TRIGGER",
+              id: dynamicTrigger.id,
+            },
+          },
+          update: {
+            event: Array.isArray(payload.rule.event) ? payload.rule.event : [payload.rule.event],
+            source: payload.rule.source,
+            payloadFilter: payload.rule.payload,
+            contextFilter: payload.rule.context,
+            dispatchable: {
+              type: "DYNAMIC_TRIGGER",
+              id: dynamicTrigger.id,
+            },
+          },
+        });
+
+        const registration = await tx.dynamicTriggerRegistration.upsert({
+          where: {
+            key_dynamicTriggerId: {
+              key,
+              dynamicTriggerId: dynamicTrigger.id,
+            },
+          },
+          create: {
+            key,
+            dynamicTriggerId: dynamicTrigger.id,
+            sourceId: triggerSource.id,
+            eventDispatcherId: eventDispatcher.id,
+            metadata: registrationMetadata,
+          },
+          update: {
+            metadata: registrationMetadata,
+          },
+        });
+
+        const secretStore = getSecretStore(triggerSource.secretReference.provider, {
+          prismaClient: tx,
+        });
+
+        const { secret } = await secretStore.getSecretOrThrow(
+          z.object({
+            secret: z.string(),
+          }),
+          triggerSource.secretReference.key
+        );
+
+        //turn into required format
+        const optionsArray = Object.entries(payload.source.options).flatMap(([name, values]) => {
+          return { name, values };
+        });
+        const options = optionsArray.reduce((acc, { name, values }) => {
+          acc[name] = {
+            desired: [...new Set(values)],
+            missing: [],
+            orphaned: [],
+          };
+          return acc;
+        }, {} as Record<string, { desired: string[]; missing: string[]; orphaned: string[] }>) as RegisterSourceEventOptions;
+
+        const data: RegisterSourceEventV2 = {
+          id: registration.id,
+          source: {
+            key: triggerSource.key,
+            active: triggerSource.active,
+            params: triggerSource.params,
+            secret,
+            data: triggerSource.channelData as any,
+            channel: {
+              type: "HTTP",
+              url: `${env.APP_ORIGIN}/api/v1/sources/http/${triggerSource.id}`,
+            },
+            clientId: triggerSource.integration.slug,
+          },
+          options,
+        };
+
+        return data;
+      },
+      { timeout: 15000 }
+    );
+  }
+}

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

@@ -19,6 +19,10 @@ import { DeliverScheduledEventService } from "./schedules/deliverScheduledEvent.
 import { ActivateSourceService } from "./sources/activateSource.server";
 import { DeliverHttpSourceRequestService } from "./sources/deliverHttpSourceRequest.server";
 import { PerformTaskOperationService } from "./tasks/performTaskOperation.server";
+import { addMissingVersionField } from "@trigger.dev/core";
+import { ExecuteBackgroundTaskOperationService } from "./backgroundTasks/executeBackgroundTaskOperation.server";
+import { AutoScalePoolService } from "./backgroundTasks/autoScalePool.server";
+import { CreateExternalMachineService } from "./backgroundTasks/createExternalMachine.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -37,11 +41,21 @@ const workerCatalog = {
     organizationId: z.string(),
     connectionId: z.string(),
   }),
-  activateSource: z.object({
-    id: z.string(),
-    orphanedEvents: z.array(z.string()).optional(),
-  }),
-
+  activateSource: z.preprocess(
+    addMissingVersionField,
+    z.discriminatedUnion("version", [
+      z.object({
+        version: z.literal("1"),
+        id: z.string(),
+        orphanedEvents: z.array(z.string()).optional(),
+      }),
+      z.object({
+        version: z.literal("2"),
+        id: z.string(),
+        orphanedOptions: z.record(z.string(), z.array(z.string())).optional(),
+      }),
+    ])
+  ),
   deliverEvent: z.object({ id: z.string() }),
   "events.invokeDispatcher": z.object({
     id: z.string(),
@@ -57,6 +71,15 @@ const workerCatalog = {
   connectionCreated: z.object({
     id: z.string(),
   }),
+  executeBackgroundTaskOperation: z.object({
+    id: z.string(),
+  }),
+  autoScalePool: z.object({
+    id: z.string(),
+  }),
+  createExternalMachine: z.object({
+    id: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -187,10 +210,27 @@ function getWorkerQueue() {
       activateSource: {
         priority: 10, // smaller number = higher priority
         maxAttempts: 3,
-        handler: async (payload, job) => {
+        handler: async (payload, graphileJob) => {
           const service = new ActivateSourceService();
-
-          await service.call(payload.id, job.id, payload.orphanedEvents);
+          switch (payload.version) {
+            case "1": {
+              //change the input data to match the new schema
+              await service.call(
+                payload.id,
+                graphileJob.id,
+                payload.orphanedEvents
+                  ? {
+                      event: payload.orphanedEvents,
+                    }
+                  : undefined
+              );
+              break;
+            }
+            case "2": {
+              await service.call(payload.id, graphileJob.id, payload.orphanedOptions);
+              break;
+            }
+          }
         },
       },
       deliverHttpSourceRequest: {
@@ -257,6 +297,33 @@ function getWorkerQueue() {
           });
         },
       },
+      executeBackgroundTaskOperation: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new ExecuteBackgroundTaskOperationService();
+
+          await service.call(payload.id);
+        },
+      },
+      autoScalePool: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new AutoScalePoolService();
+
+          await service.call(payload.id);
+        },
+      },
+      createExternalMachine: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new CreateExternalMachineService();
+
+          await service.call(payload.id);
+        },
+      },
     },
   });
 }
@@ -292,7 +359,12 @@ function getExecutionWorkerQueue() {
         handler: async (payload, job) => {
           const service = new PerformRunExecutionV2Service();
 
-          await service.call(payload.id, payload.reason, payload.isRetry, payload.resumeTaskId);
+          await service.call({
+            id: payload.id,
+            reason: payload.reason,
+            resumeTaskId: payload.resumeTaskId,
+            isRetry: payload.isRetry,
+          });
         },
       },
     },

apps/webapp/app/zodfetch.server.ts

@@ -0,0 +1,108 @@
+import { z } from "zod";
+import { safeJsonParse } from "./utils/json";
+import { ErrorWithStackSchema } from "../../../packages/core/src";
+
+export type ZodResponse<TResponseSchema extends z.ZodTypeAny> =
+  | {
+      ok: true;
+      data: z.output<TResponseSchema>;
+      status: number;
+      headers: Headers;
+    }
+  | {
+      ok: false;
+      error: { message: string; name?: string; stack?: string };
+      status: number;
+      headers: Headers;
+    };
+
+const CommonErrorSchema = z.object({
+  error: z.string(),
+});
+
+export async function zodfetch<TResponseSchema extends z.ZodTypeAny>(
+  schema: TResponseSchema,
+  url: string,
+  requestInit?: RequestInit
+): Promise<ZodResponse<TResponseSchema>> {
+  const response = await fetch(url, requestInit);
+  const contentType = response.headers.get("content-type");
+
+  if (!response.ok) {
+    // Check to see if we have a JSON body
+    if (contentType?.includes("application/json")) {
+      const rawJsonBody = await response.text();
+      const jsonBody = safeJsonParse(rawJsonBody);
+
+      if (!jsonBody) {
+        return {
+          ok: false,
+          error: { message: "Failed to parse JSON response" },
+          status: response.status,
+          headers: response.headers,
+        };
+      }
+
+      const parsed = ErrorWithStackSchema.safeParse(jsonBody);
+
+      if (parsed.success) {
+        return {
+          ok: false,
+          error: parsed.data,
+          status: response.status,
+          headers: response.headers,
+        };
+      }
+
+      const commonParsed = CommonErrorSchema.safeParse(jsonBody);
+
+      if (commonParsed.success) {
+        return {
+          ok: false,
+          error: { message: commonParsed.data.error, name: response.statusText },
+          status: response.status,
+          headers: response.headers,
+        };
+      }
+
+      return {
+        ok: false,
+        error: jsonBody as any,
+        status: response.status,
+        headers: response.headers,
+      };
+    }
+
+    return {
+      ok: false,
+      error: { message: response.statusText },
+      status: response.status,
+      headers: response.headers,
+    };
+  }
+
+  const rawJsonBody = await response.text();
+  const jsonBody = safeJsonParse(rawJsonBody);
+
+  if (!jsonBody) {
+    return {
+      ok: true,
+      status: response.status,
+      headers: response.headers,
+      data: null as any,
+    };
+  }
+
+  const parsed = schema.safeParse(jsonBody);
+
+  if (!parsed.success) {
+    throw new Error(`Failed to parse response: ${parsed.error.message}`);
+  }
+
+  return {
+    ok: true,
+    data: parsed.data,
+    status: response.status,
+    headers: response.headers,
+  };
+}

apps/webapp/package.json

@@ -60,10 +60,12 @@
     "@remix-run/server-runtime": "1.19.2-pre.0",
     "@team-plain/typescript-sdk": "^2.2.0",
     "@trigger.dev/companyicons": "^1.5.14",
-    "@trigger.dev/database": "workspace:*",
     "@trigger.dev/core": "workspace:*",
+    "@trigger.dev/database": "workspace:*",
     "@trigger.dev/sdk": "workspace:*",
     "@uiw/react-codemirror": "^4.19.5",
+    "archiver": "^6.0.1",
+    "async-retry": "^1.3.3",
     "class-variance-authority": "^0.5.2",
     "clsx": "^1.2.1",
     "compression": "^1.7.4",
@@ -133,6 +135,8 @@
     "@tailwindcss/typography": "^0.5.9",
     "@total-typescript/ts-reset": "^0.4.2",
     "@trigger.dev/tailwind-config": "workspace:*",
+    "@types/archiver": "^5.3.2",
+    "@types/async-retry": "^1.4.5",
     "@types/bcryptjs": "^2.4.2",
     "@types/compression": "^1.7.2",
     "@types/eslint": "^8.4.6",
@@ -148,6 +152,7 @@
     "@types/qs": "^6.9.7",
     "@types/react": "18.2.17",
     "@types/react-dom": "18.2.7",
+    "@types/retry": "^0.12.2",
     "@types/semver": "^7.3.13",
     "@types/simple-oauth2": "^5.0.4",
     "@types/slug": "^5.0.3",

apps/webapp/prisma/seedCloud.ts

@@ -87,7 +87,7 @@ export async function seedCloud(prisma: PrismaClient) {
     },
     create: {
       apiKey: "tr_prod_bNaLxayOXqoj",
-      pkApiKey: "pk_dev_323f3650218e370508cf",
+      pkApiKey: "pk_dev_323f3650218e378191cf",
       slug: "prod",
       type: "PRODUCTION",
       project: {

docs/documentation/concepts/client-adaptors.mdx

@@ -16,7 +16,7 @@ export const client = new TriggerClient({
 });
 ```
 
-View the [Client API Reference](//sdk/triggerclient) for more information.
+View the [Client API Reference](/sdk/triggerclient) for more information.
 
 ## Adaptors
 

docs/documentation/concepts/tasks.mdx

@@ -44,9 +44,13 @@ client.defineJob({
     await io.wait("wait", 60 * 60 * 3); // wait for 3 hours
 
     // You can wrap your own code in a Task, for retrying, resumability and logging
-    const response = await io.runTask("my-task", { name: "My Task" }, async () => {
-      return await longRunningCode(payload.userId);
-    });
+    const response = await io.runTask(
+      "my-task",
+      async () => {
+        return await longRunningCode(payload.userId);
+      },
+      { name: "My Task" }
+    );
 
     return response;
   },

docs/documentation/guides/create-a-job.mdx

@@ -32,10 +32,9 @@ client.defineJob({
 The `id` and `name` are important because they are used to create and identify your Job in the app.
 
 <Note>
-  This Job must be imported in the `trigger` file in order to be registered when
-  the CLI dev command is run. This can be found in either the
-  `app/api/trigger/route.ts` file if you're using the Next.js App Router, or
-  `pages/api/trigger.ts` if you're using the Next.js Pages Router.
+  This Job must be imported in the `trigger` file in order to be registered when the CLI dev command
+  is run. This can be found in either the `app/api/trigger/route.ts` file if you're using the
+  Next.js App Router, or `pages/api/trigger.ts` if you're using the Next.js Pages Router.
 </Note>
 
 ### 2. Choose a Trigger
@@ -112,8 +111,8 @@ This is what kicks-off a Job. There are a few different types of Triggers you ca
 > A Task is a resumable unit of a Run that can be retried, resumed and is logged.
 
 <Info>
-  You can use just regular code in your Jobs. But you don't get the benefits of
-  retrying, logging and resumability. More info on [Tasks vs regular
+  You can use just regular code in your Jobs. But you don't get the benefits of retrying, logging
+  and resumability. More info on [Tasks vs regular
   code](/documentation/concepts/tasks#tasks-vs-regular-code).
 </Info>
 
@@ -121,21 +120,21 @@ You can string together multiple Tasks and regular code in any order you want.
 
 **Useful built-in Tasks:**
 
-| Task                                        | Description                                                             | Task code                                                                                  |
-| ------------------------------------------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ |
-| [Delay](/documentation/concepts/delays)     | Wait for a period of time                                               | `await io.wait("wait", 60);`                                                               |
-| [Log](/sdk/io/logger)                       | Log a message                                                           | `await io.logger.log("Hello");`                                                            |
-| [Send Event](/sdk/io/sendevent)             | Send an event (for eventTrigger)                                        | `await io.sendEvent("my-event", { name: "my.event", payload: { hello: "world" } });`       |
-| [Run task](/sdk/io/runtask)                 | Wrap your own code in this to create a Task                             | `await io.runTask("My Task", { name: "My Task" }, async () => { console.log("Hello"); });` |
-| [Background fetch](/sdk/io/backgroundfetch) | Fetch data from a URL that can take longer that the serverless timeout. | `await io.backgroundFetch("fetch-some-data", { url: "https://example.com" });`             |
+| Task                                        | Description                                                             | Task code                                                                            |
+| ------------------------------------------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
+| [Delay](/documentation/concepts/delays)     | Wait for a period of time                                               | `await io.wait("wait", 60);`                                                         |
+| [Log](/sdk/io/logger)                       | Log a message                                                           | `await io.logger.log("Hello");`                                                      |
+| [Send Event](/sdk/io/sendevent)             | Send an event (for eventTrigger)                                        | `await io.sendEvent("my-event", { name: "my.event", payload: { hello: "world" } });` |
+| [Run task](/sdk/io/runtask)                 | Wrap your own code in this to create a Task                             | `await io.runTask("My Task", async () => { console.log("Hello"); });`                |
+| [Background fetch](/sdk/io/backgroundfetch) | Fetch data from a URL that can take longer that the serverless timeout. | `await io.backgroundFetch("fetch-some-data", { url: "https://example.com" });`       |
 
 For a full list of built-in Tasks, see the [io SDK reference](/sdk/io).
 
 **Integration Task examples:**
 
 <Info>
-  To use our integrations you will need to set them up in the app first. Our
-  guide is [here](/documentation/guides/using-integrations).
+  To use our integrations you will need to set them up in the app first. Our guide is
+  [here](/documentation/guides/using-integrations).
 </Info>
 
 <AccordionGroup>
@@ -229,10 +228,9 @@ yarn dlx @trigger.dev/cli@latest dev
 This will register all of your Jobs, they should appear in your dashboard.
 
 <Note>
-  Not seeing your Job in the web app? It might be because you forgot to import
-  it. This will need to be either in `app/api/trigger/route.ts` file if you're
-  using the Next,js App Router, or `pages/api/trigger.ts` if you're using the
-  Next,js Pages Router.
+  Not seeing your Job in the web app? It might be because you forgot to import it. This will need to
+  be either in `app/api/trigger/route.ts` file if you're using the Next,js App Router, or
+  `pages/api/trigger.ts` if you're using the Next,js Pages Router.
 </Note>
 
 If you are having trouble getting your job running, please reach out to us and we will help you fix any issues:

docs/documentation/guides/video-walkthrough.mdx

@@ -11,9 +11,9 @@ You can view the [source code here](https://github.com/triggerdotdev/examples/tr
   className="w-full aspect-video"
   src="https://www.youtube.com/embed/uocBQt2HeQo"
   title="Create a serverless background job in 10 mins"
-  frameborder="0"
+  frameBorder="0"
   allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture"
-  allowfullscreen
+  allowFullScreen
 ></iframe>
 
 We cover how to:

docs/examples/introduction.mdx

@@ -25,5 +25,3 @@ Click the links below to view the Job code. You can also easily test these Jobs
 | [GitHub new star alert](https://github.com/triggerdotdev/examples/blob/main/github/src/jobs/newStarAlert.ts)                                    | When a repo is starred a message is logged with the new Stargazers count.                                                                                     | [GitHub](/integrations/apis/github)                                         |
 | [Send a Slack message when an event is received](https://github.com/triggerdotdev/examples/blob/main/slack/src/jobs/sendSlackMessage.ts)        | Sends a Slack message to a specific channel when an event is received.                                                                                        | [Slack](/integrations/apis/slack)                                           |
 | [Send an email using Resend](https://github.com/triggerdotdev/examples/blob/main/resend/src/jobs/resendBasicEmail.ts)                           | Send a basic email using Resend                                                                                                                               | [Resend](/integrations/apis/resend)                                         |
-
-If you have any ideas for Jobs you would like to build, you can fill in the [Job request form](https://bcymafitv0e.typeform.com/to/YLUKy9my#source=example-jobs-docs).

docs/integrations/apis/github.mdx

@@ -82,14 +82,17 @@ client.defineJob({
     github,
   },
   run: async (payload, io, ctx) => {
-    //wrap the SDK call in runTask
-    const { data } = await io.runTask("create-card", { name: "Create card" }, async () => {
-      //create a project card using the underlying client
-      return io.github.client.rest.projects.createCard({
-        column_id: 123,
-        note: "test",
-      });
-    });
+    //io.github.runTask allows you to use the underlying SDK client
+    const { data } = await io.github.runTask(
+      "create-card",
+      async (client) => {
+        return client.rest.projects.createCard({
+          column_id: 123,
+          note: "test",
+        });
+      },
+      { name: "Create card" }
+    );
 
     //log the url of the created card
     await io.logger.info(data.url);

docs/integrations/apis/plain.mdx

@@ -131,7 +131,7 @@ client.defineJob({
 
 ## Using the underlying client
 
-You can use the underlying client to do anything [@team-plain/typescript-sdk](https://github.com/team-plain/typescript-sdk) supports, but make sure to wrap it in a task:
+You can use the underlying client to do anything [@team-plain/typescript-sdk](https://github.com/team-plain/typescript-sdk) supports by using runTask:
 
 ```ts
 import { Plain } from "@trigger.dev/plain";
@@ -151,21 +151,14 @@ client.defineJob({
     name: "plain.client",
   }),
   run: async (payload, io, ctx) => {
-    const issue = await io.runTask(
+    const result = await io.plain.runTask(
       "create-issue",
-      { name: "Create issue", icon: "plain" },
-      async () => {
-        const result = await io.plain.client.createIssue({
+      async (client) =>
+        client.createIssue({
           customerId: "abcdefghij",
           issueTypeId: "123456",
-        });
-
-        if (result.error) {
-          throw result.error;
-        }
-
-        return result.data;
-      }
+        }),
+      { name: "Create issue" }
     );
   },
 });

docs/integrations/apis/stripe.mdx

@@ -215,7 +215,7 @@ If there are any tasks missing that you'd like to see added, please [open a new
 
 ## Using the underlying Stripe client
 
-You can use the underlying client to do anything the [stripe-node](https://github.com/stripe/stripe-node) client supports by using the `client` property on the integration:
+You can use the underlying client to do anything the [stripe-node](https://github.com/stripe/stripe-node) client supports by using `runTask` on the integration:
 
 ```ts
 const stripe = new Stripe({
@@ -234,20 +234,25 @@ client.defineJob({
     stripe,
   },
   run: async (payload, io, ctx) => {
-    await io.runTask("create-price", { name: "Create Price" }, async (task) => {
-      return stripe.client.prices.create(
-        {
-          unit_amount: 2000,
-          currency: "usd",
-          product_data: {
-            name: "T-shirt",
+    const price = await io.stripe.runTask(
+      "create-price",
+      async (client, task) => {
+        return client.prices.create(
+          {
+            unit_amount: 2000,
+            currency: "usd",
+            product_data: {
+              name: "T-shirt",
+            },
           },
-        },
-        {
-          idempotencyKey: task.idempotencyKey,
-        }
-      );
-    });
+          {
+            idempotencyKey: task.idempotencyKey,
+          }
+        );
+      },
+      //this is optional, it will appear on the Run page
+      { name: "Create Price" }
+    );
   },
 });
 ```

docs/integrations/apis/typeform.mdx

@@ -119,7 +119,7 @@ client.defineJob({
 
 ## Using the underlying client
 
-You can use the underlying client to do anything [@typeform/api-client](https://www.npmjs.com/package/@typeform/api-client) supports, but make sure to wrap it in a task:
+You can use the underlying client to do anything [@typeform/api-client](https://www.npmjs.com/package/@typeform/api-client) supports:
 
 ```ts
 import { Typeform } from "@trigger.dev/typeform";
@@ -139,14 +139,28 @@ client.defineJob({
     name: "typeform.client",
   }),
   run: async (payload, io, ctx) => {
-    const form = await io.runTask(
+    const form = await io.typeform.runTask(
       "create-form",
-      { name: "Create Form", icon: "typeform" },
-      async () => {
-        return io.typeform.client.forms.create({
-          data: { ... }
-        })
-      }
+      async (client) => {
+        return client.forms.create({
+          data: {
+            title: "My Form",
+            fields: [
+              {
+                title: "What is your name?",
+                type: "short_text",
+                ref: "name",
+              },
+              {
+                title: "What is your email?",
+                type: "email",
+                ref: "email",
+              },
+            ],
+          },
+        });
+      },
+      { name: "Create Form" }
     );
   },
 });

docs/integrations/create-anatomy.mdx

@@ -0,0 +1,44 @@
+---
+title: The structure of an integration
+description: "What files are in an Integration package and what do they do?"
+---
+
+```sh
+stripe
+  ├── README.md
+  ├── package.json
+  ├── tsup.config.ts
+  ├── src
+  │   ├── index.ts
+  │   ├── tasks.ts
+  │   └── types.ts
+  └── tsconfig.json
+```
+
+Once you've created your Integration package, you can start developing it. In this stage of development, you'll mainly be making changes to the `src/index.ts`, `src/types.ts`, and `src/tasks.ts` files. Below is an explanation of each of the files and their roles:
+
+### `src/index.ts`
+
+This is the entry point of the Integration package and where the client lives. See [the Client docs](/integrations/create-client).
+
+### `src/types.ts`
+
+This file contains all of the types that are used in the Integration package. It's a good idea to keep all of the types in this file so that they can be easily imported into both the `src/index.ts` and `src/tasks.ts` files.
+
+You'll want to export the following types from this file
+
+- **SDK Client Type** - Export the type of the underlying SDK client that will be used both in authenticated tasks (more below) and the main Integration class.
+- **Authenticated Task Param Types** - Export a single type for each authenticated task input params
+- **Authenticated Task Response Types** - Export a single type for each authenticated task output response
+
+### `src/tasks.ts`
+
+This file contains all of the authenticated tasks that the Integration will support. Tasks are the main way that developers will interact with your Integration. They are the actions that developers will be able to perform in their jobs.
+
+For more about how tasks work, see the [tasks documentation](/documentation/concepts/tasks).
+
+See the guide for creating [authenticated tasks](/integrations/create-tasks) for more information.
+
+### Triggers
+
+See the guide for creating [triggers](/integrations/create-triggers) for more information.

docs/integrations/create-client.mdx

@@ -0,0 +1,99 @@
+---
+title: Creating the client
+description: "Each integration has a client that performs the actual HTTP requests. Usually this is a wrapper around an official SDK provided by the service."
+---
+
+### `src/index.ts`
+
+This is where the client should live.
+
+<Tip>
+  We're adopting the naming convention of naming the class after the service, without a suffix or
+  prefix. We prefer the exported name be `Slack` instead of something like `SlackIntegration` or
+  `SlackConnector`
+</Tip>
+
+<Accordion title="Example: OpenAI">
+
+```ts
+import type { IntegrationClient, TriggerIntegration } from "@trigger.dev/sdk";
+import { Configuration, OpenAIApi } from "openai";
+import * as tasks from "./tasks";
+import { OpenAIIntegrationOptions } from "./types";
+
+export class OpenAI implements TriggerIntegration<IntegrationClient<OpenAIApi, typeof tasks>> {
+  client: IntegrationClient<OpenAIApi, typeof tasks>;
+
+  constructor(private options: OpenAIIntegrationOptions) {
+    this.client = {
+      tasks,
+      usesLocalAuth: true,
+      client: new OpenAIApi(
+        new Configuration({
+          apiKey: options.apiKey,
+          organization: options.organization,
+        })
+      ),
+      auth: {
+        apiKey: options.apiKey,
+        organization: options.organization,
+      },
+    };
+  }
+
+  get id() {
+    return this.options.id;
+  }
+
+  get metadata() {
+    return { id: "openai", name: "OpenAI" };
+  }
+}
+```
+
+</Accordion>
+
+The `TriggerIntegration` interface requires three properties to be implemented:
+
+<ParamField body="id" type="string" required>
+  The `id` that uniquely identifies the Integration. This should always be passed through the
+  constructor options.
+</ParamField>
+
+<ParamField body="metadata" type="object" required>
+  <Expandable title="properties">
+    <ParamField body="id" type="string" required>
+      A unique identifier for the Integration. For example, the OpenAI Integration has an id of
+      `"openai"`.
+    </ParamField>
+    <ParamField body="name" type="string" required>
+      The name of the Integration. For example, the OpenAI Integration has a name of `"OpenAI"`.
+    </ParamField>
+  </Expandable>
+</ParamField>
+
+<ParamField body="client" type="IntegrationClient" required>
+  An IntegrationClient object that contains either the underlying SDK client (if
+  using local auth) or a `clientFactory` function to create new clients using
+  authenticated credentials.
+
+{" "}
+
+  <Expandable title="properties" defaultOpen>
+    <ParamField body="usesLocalAuth" type="boolean" required>
+      Specifies whether the client uses local auth or not. If `true`, the `client` property should contain the underlying SDK client. If `false`, the `client` property should contain a `clientFactory` function that can be used to create new clients using authenticated credentials.
+    </ParamField>
+    <ParamField body="tasks" type="Record<string, AuthenticatedTask>" required>
+      An object that contains all of the authenticated tasks that are supported by the Integration. The keys of the object should be the names of the tasks and the values should be the authenticated tasks. More on authenticated tasks below.
+    </ParamField>
+    <ParamField body="clientFactory" type="function">
+      A function that takes in an `auth` object (of type [ConnectionAuth](/sdk/connection-auth)) and returns a new SDK client that is authenticated with the credentials in the `auth` object. This should only be set if `usesLocalAuth` is `false`.
+    </ParamField>
+    <ParamField body="client" type="SDK Client">
+      The underlying SDK client that will be used to make requests to the service. This should only be set if `usesLocalAuth` is `true`.
+    </ParamField>
+    <ParamField body="auth" type="any">
+      The authenticated credentials that were used to create the `client`. This should only be set if `usesLocalAuth` is `true`. This is used to access the credentials inside of authenticated tasks, but is only necessary in specific cases (like when doing a [`backgroundFetch`](/sdk/io/backgroundfetch) as a subtask)
+    </ParamField>
+  </Expandable>
+</ParamField>

docs/integrations/create-oauth.mdx

@@ -0,0 +1,6 @@
+---
+title: OAuth
+description: "OAuth is used to authenticate users by passing them to a third-party service, such as Google or Facebook, to log in."
+---
+
+<Note>The guide on adding a new OAuth app to Trigger.dev is coming soon</Note>

docs/integrations/create-publishing.mdx

@@ -0,0 +1,17 @@
+---
+title: Publishing
+description: "Publishing an official integration package to the public registry."
+---
+
+## 1. Check your integration
+
+Ensure that you have done the following:
+
+- [ ] Create the integrations in the `integrations` folder.
+- [ ] Created a file in the `examples/job-catalog` folder with some example jobs.
+- [ ] Have exported types.
+- [ ] Avoid using `any` types and `@ts-ignore` comments. If there are any, please explain them in the PR.
+
+## 2. Create a Pull Request
+
+[Create a Pull Request](https://github.com/triggerdotdev/trigger.dev/pulls) in the Trigger.dev repository.

docs/integrations/create-setup.mdx

@@ -0,0 +1,60 @@
+---
+title: Setup the package
+description: "How to create the folders and basic files for an integration package."
+---
+
+Before you embark on creating an Integration, you have to decide where the Integration code will be located. You have two options, either in the [Trigger.dev monorepo](https://github.com/triggerdotdev/trigger.dev) and namespaced under the `@trigger.dev` NPM organization, or in a separate repository you control and published independently.
+
+## Using OpenAI to generate the initial code
+
+When you use the `@trigger.dev/cli create-integration` command you can pass in an OpenAI API key to generate the initial code for your Integration. Use the `-o` option to do this.
+
+```bash
+npx @trigger.dev/cli create-integration -o=sk-abcdefghijk integrations/stripe
+```
+
+## In the Trigger.dev monorepo
+
+Before you can create an Integration in the Trigger.dev monorepo, you'll need to follow our [contributing guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md) to get your local environment setup.
+
+Once you've forked the repository and cloned it locally, create a new branch for your Integration and prefix it with the `integrations/` namespace. For example, if you were creating an Integration for [Stripe](https://stripe.com), you would create a branch named `integrations/stripe`.
+
+```bash
+git checkout -b integrations/stripe
+```
+
+Now you are ready to create your Integration. We've created a CLI tool to help you scaffold out the Integration package. You can run the following command to create a new Integration package in the `integrations` directory:
+
+```bash
+npx @trigger.dev/cli create-integration integrations/stripe
+```
+
+This will ask you a few questions about your Integration:
+
+1. `What is the name of your Integration package?` - This is the name of the NPM package that will be created. It should be prefixed with `@trigger.dev/integration-` and be all lowercase. For example, if you were creating an Integration for Stripe, you would enter `@trigger.dev/stripe`.
+2. `What is the name of the npm package of the Integration?` - This is the name of the NPM package that the Integration will be wrapping. For example, if you were creating an Integration for Stripe, you would enter `stripe`.
+
+From this point, the CLI will create the Integration package for you and install all of the dependencies. It creates the following package structure:
+
+```bash
+integrations
+└── stripe
+    ├── README.md
+    ├── package.json
+    ├── tsup.config.ts
+    ├── src
+    │   ├── index.ts
+    │   ├── tasks.ts
+    │   └── types.ts
+    └── tsconfig.json
+```
+
+Next, head to the [How to develop an Integration package](#how-to-develop-an-integration-package) section to learn how to develop your Integration.
+
+## In your own repository
+
+You can also create an Integration in your own repository and publish it independently. This is useful if you want to keep your Integration code separate from the Trigger.dev monorepo. In this case, you should use the `@trigger.dev/cli` to create the Integration package in your repository or to start a new one:
+
+```bash
+npx @trigger.dev/cli create-integration my-internal-integration
+```

docs/integrations/create-tasks.mdx

@@ -0,0 +1,428 @@
+---
+title: Authenticated Tasks
+description: "Tasks are the main way that developers will interact with your Integration. They are the actions that developers will be able to perform in their jobs."
+---
+
+## `src/tasks.ts`
+
+This file contains all of the authenticated tasks that the Integration will support.
+
+## Authenticated tasks
+
+Authenticated tasks are the backbone of an Integration and so we're going to cover them in more detail before moving on the main Integration class.
+
+Authenticated tasks are a specially crafted object that allows the `@trigger.dev/sdk` to run the task seeded with an authenticated SDK client.
+
+For example, here is the `getForm` authenticated task defined in the `@trigger.dev/typeform` Integration package:
+
+<CodeGroup>
+
+```ts tasks.ts
+import type { AuthenticatedTask } from "@trigger.dev/sdk";
+import type { GetFormParams, GetFormResponse, TypeformSDK } from "./types";
+
+export const getForm: AuthenticatedTask<TypeformSDK, GetFormParams, GetFormResponse> = {
+  init: (params) => {
+    return {
+      name: "Get Form",
+      params,
+      icon: "typeform",
+      properties: [
+        {
+          label: "Form ID",
+          text: params.uid,
+        },
+      ],
+    };
+  },
+  run: async (params, client) => {
+    return client.forms.get(params);
+  },
+};
+```
+
+```ts types.ts
+import { Prettify } from "@trigger.dev/integration-kit";
+import { createClient } from "@typeform/api-client";
+
+export type TypeformSDK = ReturnType<typeof createClient>;
+
+export type GetFormParams = {
+  uid: string;
+};
+
+export type GetFormResponse = Prettify<Typeform.Form>;
+```
+
+```ts usage.ts
+client.defineJob({
+  id: "typeform-playground",
+  name: "Typeform Playground",
+  version: "0.1.1",
+  integrations: {
+    typeform,
+  },
+  run: async (payload, io, ctx) => {
+    const form = await io.typeform.getForm("get-form", {
+      uid: payload.formId,
+    });
+  },
+});
+```
+
+</CodeGroup>
+
+The first thing to notice is the explicit typing of the `getForm` export as an `AuthenticatedTask<TypeformSDK, GetFormParams, GetFormResponse>`.
+
+- The first type parameter is the type of the SDK client that will be used to run the task. In this case, it's the `TypeformSDK` type that is exported from the `src/types.ts` file.
+- The second type parameter is the type of the input params that will be passed to the task. The `params` argument in the `run` and `init` functions will be typed as this type parameter.
+- The third type parameter is the type of the output response that will be returned from the task. The return type of the `run` function needs to match this type.
+
+If you take a look at the `usage.ts` file above, you can see how this task is used in a job. The `io.typeform.getForm` function is typed as returning `Promise<GetFormResponse>` and the `params` argument is typed as `GetFormParams`.
+
+<Note>
+  Notice how the params are the _second_ argument to `getForm`, that's because the first argument is
+  always the task key. See our [Keys and Resumability docs](/documentation/concepts/resumability)
+  for more on why this is important
+</Note>
+
+#### `run` function
+
+The `run` function is the main function that will be called when the task is run. It's an async function that takes up to 5 arguments:
+
+<ParamField body="params" type="type parameter" required>
+  The input params that were passed to the task. This is the second argument to the `getForm`
+  function in the example above.
+</ParamField>
+
+<ParamField body="client" type="type parameter" required>
+  The authenticated SDK client that was seeded into the task.
+</ParamField>
+
+<ParamField body="task" type="Task">
+  The underlying [task](/documentation/concepts/tasks) object
+</ParamField>
+
+<ParamField body="io" type="IO">
+  The [IO](/sdk/io/overview) object that can be used to run subtasks using
+  [`io.runTask`](/sdk/io/runtask)
+</ParamField>
+
+<ParamField body="auth" type="ConnectionAuth">
+  If for some reason you need to access the auth object that was used to seed the SDK client, you
+  can access it here. The `AuthenticatedTask` generic type takes an optional 4th type parameter that
+  allows you to specify the auth type
+</ParamField>
+
+#### `init` function
+
+The `init` function is used to initialize the task. It's a synchronous function that takes a single argument:
+
+<ParamField body="params" type="type parameter" required>
+  The input params that were passed to the task. This is the second argument to the `getForm`
+  function in the example above.
+</ParamField>
+
+#### `onError` function
+
+Authenticated tasks take an optional `onError` function that can be used to handle errors that occur during executing of the `run` function of the task. It takes two arguments:
+
+<ParamField body="error" type="unknown" required>
+  The error that was thrown during the execution of the `run` function.
+</ParamField>
+
+<ParamField body="task" type="Task" required>
+  The underlying [task](/documentation/concepts/tasks) object
+</ParamField>
+
+The `onError` function allows you to reformated errors that occur during the execution of the `run` function. For example, all the tasks in `@trigger.dev/openai` specify the following `onError` function:
+
+<CodeGroup>
+
+```ts tasks.ts
+import { OpenAIErrorSchema } from "./types";
+
+function onTaskError(error: unknown) {
+  const openAIError = OpenAIErrorSchema.safeParse(error);
+
+  if (!openAIError.success) {
+    return;
+  }
+
+  const { message, code, type } = openAIError.data.response.data.error;
+
+  return new Error(`${type}: ${message}${code ? ` (${code})` : ""}`);
+}
+```
+
+```ts types.ts
+const OpenAIErrorSchema = z.object({
+  response: z.object({
+    data: z.object({
+      error: z.object({
+        code: z.string().nullable().optional(),
+        message: z.string(),
+        type: z.string(),
+      }),
+    }),
+  }),
+});
+```
+
+</CodeGroup>
+
+You can also use the `onError` function to specify a specific time the task should be retried. The `@trigger.dev/github` Integration uses this to retry rate-limited requests:
+
+```ts
+function isRequestError(error: unknown): error is RequestError {
+  return typeof error === "object" && error !== null && "status" in error;
+}
+
+function onError(error: unknown) {
+  if (!isRequestError(error)) {
+    return;
+  }
+
+  // Check if this is a rate limit error
+  if (error.status === 403 && error.response) {
+    const rateLimitRemaining = error.response.headers["x-ratelimit-remaining"];
+    const rateLimitReset = error.response.headers["x-ratelimit-reset"];
+
+    if (rateLimitRemaining === "0" && rateLimitReset) {
+      const resetDate = new Date(Number(rateLimitReset) * 1000);
+
+      return {
+        retryAt: resetDate,
+        error,
+      };
+    }
+  }
+}
+```
+
+### Triggers
+
+<Note>The guide on creating Integration triggers is coming soon</Note>
+
+## Testing an Integration package
+
+<Note>This section is coming soon</Note>
+
+## Publishing an Integration package
+
+<Note>This section is coming soon</Note>
+
+## Example authenticated tasks
+
+### OpenAI examples
+
+<CodeGroup>
+
+```ts retrieveModel
+export const retrieveModel: AuthenticatedTask<
+  OpenAIClientType,
+  Prettify<RetrieveModelRequest>,
+  RetrieveModelResponseData
+> = {
+  onError: onTaskError,
+  run: async (params, client) => {
+    return client.retrieveModel(params.model).then((res) => res.data);
+  },
+  init: (params) => {
+    return {
+      name: "Retrieve model",
+      params,
+      icon: "openai",
+      properties: [
+        {
+          label: "Model id",
+          text: params.model,
+        },
+      ],
+    };
+  },
+};
+```
+
+```ts createCompletion
+export const createCompletion: AuthenticatedTask<
+  OpenAIClientType,
+  Prettify<CreateCompletionRequest>,
+  Prettify<Awaited<ReturnType<OpenAIClientType["createCompletion"]>>["data"]>
+> = {
+  run: async (params, client, task) => {
+    const response = await client.createCompletion(params);
+
+    task.outputProperties = createTaskUsageProperties(response.data.usage);
+
+    return response.data;
+  },
+  init: (params) => {
+    return {
+      name: "Completion",
+      params,
+      icon: "openai",
+      properties: [
+        {
+          label: "model",
+          text: params.model,
+        },
+      ],
+    };
+  },
+};
+```
+
+```ts backgroundCompletion
+export const backgroundCreateCompletion: AuthenticatedTask<
+  OpenAIClientType,
+  Prettify<CreateCompletionRequest>,
+  Prettify<CreateCompletionResponseData>,
+  OpenAIIntegrationAuth
+> = {
+  run: async (params, client, task, io, auth) => {
+    const response = await io.backgroundFetch<CreateCompletionResponseData>(
+      "background",
+      "https://api.openai.com/v1/completions",
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          Authorization: redactString`Bearer ${auth.apiKey}`,
+          ...(auth.organization ? { "OpenAI-Organization": auth.organization } : {}),
+        },
+        body: JSON.stringify(params),
+      }
+    );
+
+    task.outputProperties = createTaskUsageProperties(response.usage);
+
+    return response;
+  },
+  init: (params) => {
+    return {
+      name: "Background Completion",
+      params,
+      icon: "openai",
+      properties: [
+        {
+          label: "model",
+          text: params.model,
+        },
+      ],
+    };
+  },
+};
+```
+
+</CodeGroup>
+
+### GitHub examples
+
+<CodeGroup>
+
+```ts createIssue
+const createIssue: GithubAuthenticatedTask<
+  { title: string; owner: string; repo: string },
+  OctokitClient["rest"]["issues"]["create"]
+> = {
+  onError,
+  run: async (params, client, task, io) => {
+    return client.rest.issues
+      .create({
+        owner: params.owner,
+        repo: params.repo,
+        title: params.title,
+      })
+      .then((res) => res.data);
+  },
+  init: (params) => {
+    return {
+      name: "Create Issue",
+      params,
+      properties: [
+        ...repoProperties(params),
+        {
+          label: "Title",
+          text: params.title,
+        },
+      ],
+      retry: {
+        limit: 3,
+        factor: 2,
+        minTimeoutInMs: 500,
+        maxTimeoutInMs: 30000,
+        randomize: true,
+      },
+    };
+  },
+};
+```
+
+```ts createIssueCommentWithReaction
+const createIssueCommentWithReaction: GithubAuthenticatedTask<
+  {
+    body: string;
+    owner: string;
+    repo: string;
+    issueNumber: number;
+    reaction: ReactionContent;
+  },
+  OctokitClient["rest"]["issues"]["createComment"]
+> = {
+  onError,
+  run: async (params, client, task, io, auth) => {
+    const comment = await io.runTask(
+      `Comment on Issue #${params.issueNumber}`,
+      async (t) => {
+        return createIssueComment.run(params, client, t, io, auth);
+      },
+      createIssueComment.init(params)
+    );
+
+    await io.runTask(
+      `React with ${params.reaction}`,
+      addIssueCommentReaction.init({
+        owner: params.owner,
+        repo: params.repo,
+        commentId: comment.id,
+        content: params.reaction,
+      }),
+      async (t) => {
+        return addIssueCommentReaction.run(
+          {
+            owner: params.owner,
+            repo: params.repo,
+            commentId: comment.id,
+            content: params.reaction,
+          },
+          client,
+          t,
+          io,
+          auth
+        );
+      }
+    );
+
+    return comment;
+  },
+  init: (params) => {
+    return {
+      name: "Create Issue Comment",
+      params,
+      properties: [
+        {
+          label: "Repo",
+          text: params.repo,
+        },
+        {
+          label: "Issue",
+          text: `#${params.issueNumber}`,
+        },
+      ],
+    };
+  },
+};
+```
+
+</CodeGroup>

docs/integrations/create-triggers.mdx

@@ -0,0 +1,6 @@
+---
+title: Triggers
+description: "Triggers cause a Job Run to start. Webhooks and polling are the most relevant for integrations."
+---
+
+<Note>The guide on creating Integration triggers is coming soon</Note>

docs/integrations/create.mdx

@@ -1,5 +1,5 @@
 ---
-title: Create an Integration
+title: Introduction
 description: "You can create Integrations of your own."
 ---
 
@@ -7,592 +7,10 @@ Before creating an Integration, make sure to read the [Integration overview](/do
 
 What we'll cover in this guide:
 
-- [Creating an Integration package](#creating-an-integration-package)
-- [Developing an Integration package](#developing-an-integration-package)
-- [Testing an Integration package](#testing-an-integration-package)
-- [Publishing an Integration package](#publishing-an-integration-package)
-
-## Creating an Integration package
-
-Before you embark on creating an Integration, you have to decide where the Integration code will be located. You have two options, either in the [Trigger.dev monorepo](https://github.com/triggerdotdev/trigger.dev) and namespaced under the `@trigger.dev` NPM organization, or in a separate repository you control and published independently.
-
-### In the Trigger.dev monorepo
-
-Before you can create an Integration in the Trigger.dev monorepo, you'll need to follow our [contributing guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md) to get your local environment setup.
-
-Once you've forked the repository and cloned it locally, create a new branch for your Integration and prefix it with the `integrations/` namespace. For example, if you were creating an Integration for [Stripe](https://stripe.com), you would create a branch named `integrations/stripe`.
-
-```bash
-git checkout -b integrations/stripe
-```
-
-Now you are ready to create your Integration. We've created a CLI tool to help you scaffold out the Integration package. You can run the following command to create a new Integration package in the `integrations` directory:
-
-```bash
-npx @trigger.dev/cli create-integration integrations/stripe
-```
-
-This will ask you a few questions about your Integration:
-
-1. `What is the name of your Integration package?` - This is the name of the NPM package that will be created. It should be prefixed with `@trigger.dev/integration-` and be all lowercase. For example, if you were creating an Integration for Stripe, you would enter `@trigger.dev/stripe`.
-2. `What is the name of the npm package of the Integration?` - This is the name of the NPM package that the Integration will be wrapping. For example, if you were creating an Integration for Stripe, you would enter `stripe`.
-
-From this point, the CLI will create the Integration package for you and install all of the dependencies. It creates the following package structure:
-
-```bash
-integrations
-└── stripe
-    ├── README.md
-    ├── package.json
-    ├── tsup.config.ts
-    ├── src
-    │   ├── index.ts
-    │   ├── tasks.ts
-    │   └── types.ts
-    └── tsconfig.json
-```
-
-Next, head to the [How to develop an Integration package](#how-to-develop-an-integration-package) section to learn how to develop your Integration.
-
-### In your own repository
-
-You can also create an Integration in your own repository and publish it independently. This is useful if you want to keep your Integration code separate from the Trigger.dev monorepo. In this case, you should use the `@trigger.dev/cli` to create the Integration package in your repository or to start a new one:
-
-```bash
-npx @trigger.dev/cli create-integration my-internal-integration
-```
-
-## Developing an Integration package
-
-Once you've created your Integration package, you can start developing it. In this stage of development, you'll mainly be making changes to the `src/index.ts`, `src/types.ts`, and `src/tasks.ts` files. Below is an explanation of each of the files and their roles:
-
-### `src/index.ts`
-
-This is the entry point of the Integration package. It exports a main "integration" class that implements the `TriggerIntegration` interface. For example, the `@trigger.dev/github` Integration exports a `Github` class that implements.
-
-<Tip>
-  We're adopting the naming convention of naming the class after the service, without a suffix or
-  prefix. We prefer the exported name be `Slack` instead of something like `SlackIntegration` or
-  `SlackConnector`
-</Tip>
-
-<Accordion title="Example: OpenAI">
-
-```ts
-import type { IntegrationClient, TriggerIntegration } from "@trigger.dev/sdk";
-import { Configuration, OpenAIApi } from "openai";
-import * as tasks from "./tasks";
-import { OpenAIIntegrationOptions } from "./types";
-
-export class OpenAI implements TriggerIntegration<IntegrationClient<OpenAIApi, typeof tasks>> {
-  client: IntegrationClient<OpenAIApi, typeof tasks>;
-
-  constructor(private options: OpenAIIntegrationOptions) {
-    this.client = {
-      tasks,
-      usesLocalAuth: true,
-      client: new OpenAIApi(
-        new Configuration({
-          apiKey: options.apiKey,
-          organization: options.organization,
-        })
-      ),
-      auth: {
-        apiKey: options.apiKey,
-        organization: options.organization,
-      },
-    };
-  }
-
-  get id() {
-    return this.options.id;
-  }
-
-  get metadata() {
-    return { id: "openai", name: "OpenAI" };
-  }
-}
-```
-
-</Accordion>
-
-The `TriggerIntegration` interface requires three properties to be implemented:
-
-<ParamField body="id" type="string" required>
-  The `id` that uniquely identifies the Integration. This should always be passed through the
-  constructor options.
-</ParamField>
-
-<ParamField body="metadata" type="object" required>
-  <Expandable title="properties">
-    <ParamField body="id" type="string" required>
-      A unique identifier for the Integration. For example, the OpenAI Integration has an id of
-      `"openai"`.
-    </ParamField>
-    <ParamField body="name" type="string" required>
-      The name of the Integration. For example, the OpenAI Integration has a name of `"OpenAI"`.
-    </ParamField>
-  </Expandable>
-</ParamField>
-
-<ParamField body="client" type="IntegrationClient" required>
-  An IntegrationClient object that contains either the underlying SDK client (if
-  using local auth) or a `clientFactory` function to create new clients using
-  authenticated credentials.
-
-{" "}
-
-  <Expandable title="properties" defaultOpen>
-    <ParamField body="usesLocalAuth" type="boolean" required>
-      Specifies whether the client uses local auth or not. If `true`, the `client` property should contain the underlying SDK client. If `false`, the `client` property should contain a `clientFactory` function that can be used to create new clients using authenticated credentials.
-    </ParamField>
-    <ParamField body="tasks" type="Record<string, AuthenticatedTask>" required>
-      An object that contains all of the authenticated tasks that are supported by the Integration. The keys of the object should be the names of the tasks and the values should be the authenticated tasks. More on authenticated tasks below.
-    </ParamField>
-    <ParamField body="clientFactory" type="function">
-      A function that takes in an `auth` object (of type [ConnectionAuth](/sdk/connection-auth)) and returns a new SDK client that is authenticated with the credentials in the `auth` object. This should only be set if `usesLocalAuth` is `false`.
-    </ParamField>
-    <ParamField body="client" type="SDK Client">
-      The underlying SDK client that will be used to make requests to the service. This should only be set if `usesLocalAuth` is `true`.
-    </ParamField>
-    <ParamField body="auth" type="any">
-      The authenticated credentials that were used to create the `client`. This should only be set if `usesLocalAuth` is `true`. This is used to access the credentials inside of authenticated tasks, but is only necessary in specific cases (like when doing a [`backgroundFetch`](/sdk/io/backgroundfetch) as a subtask)
-    </ParamField>
-  </Expandable>
-</ParamField>
-
-### `src/types.ts`
-
-This file contains all of the types that are used in the Integration package. It's a good idea to keep all of the types in this file so that they can be easily imported into both the `src/index.ts` and `src/tasks.ts` files.
-
-You'll want to export the following types from this file
-
-- **SDK Client Type** - Export the type of the underlying SDK client that will be used both in authenticated tasks (more below) and the main Integration class.
-- **Authenticated Task Param Types** - Export a single type for each authenticated task input params
-- **Authenticated Task Response Types** - Export a single type for each authenticated task output response
-
-### `src/tasks.ts`
-
-This file contains all of the authenticated tasks that the Integration will support. Tasks are the main way that developers will interact with your Integration. They are the actions that developers will be able to perform in their jobs.
-
-For more about how tasks workd, see the [tasks documentation](/documentation/concepts/tasks).
-
-### Authenticated tasks
-
-Authenticated tasks are the backbone of an Integration and so we're going to cover them in more detail before moving on the main Integration class.
-
-Authenticated tasks are a specially crafted object that allows the `@trigger.dev/sdk` to run the task seeded with an authenticated SDK client.
-
-For example, here is the `getForm` authenticated task defined in the `@trigger.dev/typeform` Integration package:
-
-<CodeGroup>
-
-```ts tasks.ts
-import type { AuthenticatedTask } from "@trigger.dev/sdk";
-import type { GetFormParams, GetFormResponse, TypeformSDK } from "./types";
-
-export const getForm: AuthenticatedTask<TypeformSDK, GetFormParams, GetFormResponse> = {
-  init: (params) => {
-    return {
-      name: "Get Form",
-      params,
-      icon: "typeform",
-      properties: [
-        {
-          label: "Form ID",
-          text: params.uid,
-        },
-      ],
-    };
-  },
-  run: async (params, client) => {
-    return client.forms.get(params);
-  },
-};
-```
-
-```ts types.ts
-import { Prettify } from "@trigger.dev/integration-kit";
-import { createClient } from "@typeform/api-client";
-
-export type TypeformSDK = ReturnType<typeof createClient>;
-
-export type GetFormParams = {
-  uid: string;
-};
-
-export type GetFormResponse = Prettify<Typeform.Form>;
-```
-
-```ts usage.ts
-client.defineJob({
-  id: "typeform-playground",
-  name: "Typeform Playground",
-  version: "0.1.1",
-  integrations: {
-    typeform,
-  },
-  run: async (payload, io, ctx) => {
-    const form = await io.typeform.getForm("get-form", {
-      uid: payload.formId,
-    });
-  },
-});
-```
-
-</CodeGroup>
-
-The first thing to notice is the explicit typing of the `getForm` export as an `AuthenticatedTask<TypeformSDK, GetFormParams, GetFormResponse>`.
-
-- The first type parameter is the type of the SDK client that will be used to run the task. In this case, it's the `TypeformSDK` type that is exported from the `src/types.ts` file.
-- The second type parameter is the type of the input params that will be passed to the task. The `params` argument in the `run` and `init` functions will be typed as this type parameter.
-- The third type parameter is the type of the output response that will be returned from the task. The return type of the `run` function needs to match this type.
-
-If you take a look at the `usage.ts` file above, you can see how this task is used in a job. The `io.typeform.getForm` function is typed as returning `Promise<GetFormResponse>` and the `params` argument is typed as `GetFormParams`.
-
-<Note>
-  Notice how the params are the _second_ argument to `getForm`, that's because the first argument is
-  always the task key. See our [Keys and Resumability docs](/documentation/concepts/resumability)
-  for more on why this is important
-</Note>
-
-#### `run` function
-
-The `run` function is the main function that will be called when the task is run. It's an async function that takes up to 5 arguments:
-
-<ParamField body="params" type="type parameter" required>
-  The input params that were passed to the task. This is the second argument to the `getForm`
-  function in the example above.
-</ParamField>
-
-<ParamField body="client" type="type parameter" required>
-  The authenticated SDK client that was seeded into the task.
-</ParamField>
-
-<ParamField body="task" type="Task">
-  The underlying [task](/documentation/concepts/tasks) object
-</ParamField>
-
-<ParamField body="io" type="IO">
-  The [IO](/sdk/io/overview) object that can be used to run subtasks using
-  [`io.runTask`](/sdk/io/runtask)
-</ParamField>
-
-<ParamField body="auth" type="ConnectionAuth">
-  If for some reason you need to access the auth object that was used to seed the SDK client, you
-  can access it here. The `AuthenticatedTask` generic type takes an optional 4th type parameter that
-  allows you to specify the auth type
-</ParamField>
-
-#### `init` function
-
-The `init` function is used to initialize the task. It's a synchronous function that takes a single argument:
-
-<ParamField body="params" type="type parameter" required>
-  The input params that were passed to the task. This is the second argument to the `getForm`
-  function in the example above.
-</ParamField>
-
-#### `onError` function
-
-Authenticated tasks take an optional `onError` function that can be used to handle errors that occur during executing of the `run` function of the task. It takes two arguments:
-
-<ParamField body="error" type="unknown" required>
-  The error that was thrown during the execution of the `run` function.
-</ParamField>
-
-<ParamField body="task" type="Task" required>
-  The underlying [task](/documentation/concepts/tasks) object
-</ParamField>
-
-The `onError` function allows you to reformated errors that occur during the execution of the `run` function. For example, all the tasks in `@trigger.dev/openai` specify the following `onError` function:
-
-<CodeGroup>
-
-```ts tasks.ts
-import { OpenAIErrorSchema } from "./types";
-
-function onTaskError(error: unknown) {
-  const openAIError = OpenAIErrorSchema.safeParse(error);
-
-  if (!openAIError.success) {
-    return;
-  }
-
-  const { message, code, type } = openAIError.data.response.data.error;
-
-  return new Error(`${type}: ${message}${code ? ` (${code})` : ""}`);
-}
-```
-
-```ts types.ts
-const OpenAIErrorSchema = z.object({
-  response: z.object({
-    data: z.object({
-      error: z.object({
-        code: z.string().nullable().optional(),
-        message: z.string(),
-        type: z.string(),
-      }),
-    }),
-  }),
-});
-```
-
-</CodeGroup>
-
-You can also use the `onError` function to specify a specific time the task should be retried. The `@trigger.dev/github` Integration uses this to retry rate-limited requests:
-
-```ts
-function isRequestError(error: unknown): error is RequestError {
-  return typeof error === "object" && error !== null && "status" in error;
-}
-
-function onError(error: unknown) {
-  if (!isRequestError(error)) {
-    return;
-  }
-
-  // Check if this is a rate limit error
-  if (error.status === 403 && error.response) {
-    const rateLimitRemaining = error.response.headers["x-ratelimit-remaining"];
-    const rateLimitReset = error.response.headers["x-ratelimit-reset"];
-
-    if (rateLimitRemaining === "0" && rateLimitReset) {
-      const resetDate = new Date(Number(rateLimitReset) * 1000);
-
-      return {
-        retryAt: resetDate,
-        error,
-      };
-    }
-  }
-}
-```
-
-### Triggers
-
-<Note>The guide on creating Integration triggers is coming soon</Note>
-
-## Testing an Integration package
-
-<Note>This section is coming soon</Note>
-
-## Publishing an Integration package
-
-<Note>This section is coming soon</Note>
-
-## Example authenticated tasks
-
-### OpenAI examples
-
-<CodeGroup>
-
-```ts retrieveModel
-export const retrieveModel: AuthenticatedTask<
-  OpenAIClientType,
-  Prettify<RetrieveModelRequest>,
-  RetrieveModelResponseData
-> = {
-  onError: onTaskError,
-  run: async (params, client) => {
-    return client.retrieveModel(params.model).then((res) => res.data);
-  },
-  init: (params) => {
-    return {
-      name: "Retrieve model",
-      params,
-      icon: "openai",
-      properties: [
-        {
-          label: "Model id",
-          text: params.model,
-        },
-      ],
-    };
-  },
-};
-```
-
-```ts createCompletion
-export const createCompletion: AuthenticatedTask<
-  OpenAIClientType,
-  Prettify<CreateCompletionRequest>,
-  Prettify<Awaited<ReturnType<OpenAIClientType["createCompletion"]>>["data"]>
-> = {
-  run: async (params, client, task) => {
-    const response = await client.createCompletion(params);
-
-    task.outputProperties = createTaskUsageProperties(response.data.usage);
-
-    return response.data;
-  },
-  init: (params) => {
-    return {
-      name: "Completion",
-      params,
-      icon: "openai",
-      properties: [
-        {
-          label: "model",
-          text: params.model,
-        },
-      ],
-    };
-  },
-};
-```
-
-```ts backgroundCompletion
-export const backgroundCreateCompletion: AuthenticatedTask<
-  OpenAIClientType,
-  Prettify<CreateCompletionRequest>,
-  Prettify<CreateCompletionResponseData>,
-  OpenAIIntegrationAuth
-> = {
-  run: async (params, client, task, io, auth) => {
-    const response = await io.backgroundFetch<CreateCompletionResponseData>(
-      "background",
-      "https://api.openai.com/v1/completions",
-      {
-        method: "POST",
-        headers: {
-          "Content-Type": "application/json",
-          Authorization: redactString`Bearer ${auth.apiKey}`,
-          ...(auth.organization ? { "OpenAI-Organization": auth.organization } : {}),
-        },
-        body: JSON.stringify(params),
-      }
-    );
-
-    task.outputProperties = createTaskUsageProperties(response.usage);
-
-    return response;
-  },
-  init: (params) => {
-    return {
-      name: "Background Completion",
-      params,
-      icon: "openai",
-      properties: [
-        {
-          label: "model",
-          text: params.model,
-        },
-      ],
-    };
-  },
-};
-```
-
-</CodeGroup>
-
-### GitHub examples
-
-<CodeGroup>
-
-```ts createIssue
-const createIssue: GithubAuthenticatedTask<
-  { title: string; owner: string; repo: string },
-  OctokitClient["rest"]["issues"]["create"]
-> = {
-  onError,
-  run: async (params, client, task, io) => {
-    return client.rest.issues
-      .create({
-        owner: params.owner,
-        repo: params.repo,
-        title: params.title,
-      })
-      .then((res) => res.data);
-  },
-  init: (params) => {
-    return {
-      name: "Create Issue",
-      params,
-      properties: [
-        ...repoProperties(params),
-        {
-          label: "Title",
-          text: params.title,
-        },
-      ],
-      retry: {
-        limit: 3,
-        factor: 2,
-        minTimeoutInMs: 500,
-        maxTimeoutInMs: 30000,
-        randomize: true,
-      },
-    };
-  },
-};
-```
-
-```ts createIssueCommentWithReaction
-const createIssueCommentWithReaction: GithubAuthenticatedTask<
-  {
-    body: string;
-    owner: string;
-    repo: string;
-    issueNumber: number;
-    reaction: ReactionContent;
-  },
-  OctokitClient["rest"]["issues"]["createComment"]
-> = {
-  onError,
-  run: async (params, client, task, io, auth) => {
-    const comment = await io.runTask(
-      `Comment on Issue #${params.issueNumber}`,
-      createIssueComment.init(params),
-      async (t) => {
-        return createIssueComment.run(params, client, t, io, auth);
-      }
-    );
-
-    await io.runTask(
-      `React with ${params.reaction}`,
-      addIssueCommentReaction.init({
-        owner: params.owner,
-        repo: params.repo,
-        commentId: comment.id,
-        content: params.reaction,
-      }),
-      async (t) => {
-        return addIssueCommentReaction.run(
-          {
-            owner: params.owner,
-            repo: params.repo,
-            commentId: comment.id,
-            content: params.reaction,
-          },
-          client,
-          t,
-          io,
-          auth
-        );
-      }
-    );
-
-    return comment;
-  },
-  init: (params) => {
-    return {
-      name: "Create Issue Comment",
-      params,
-      properties: [
-        {
-          label: "Repo",
-          text: params.repo,
-        },
-        {
-          label: "Issue",
-          text: `#${params.issueNumber}`,
-        },
-      ],
-    };
-  },
-};
-```
-
-</CodeGroup>
+- [Setup the package](/integrations/create-setup)
+- [The structure of an integration](/integrations/create-anatomy)
+- [OAuth](/integrations/create-oauth)
+- [The integration client](/integrations/create-client)
+- [Creating Tasks](/integrations/create-tasks)
+- [Creating Triggers](/integrations/create-triggers)
+- [Publishing an Integration package](/integrations/create-publishing)

docs/mint.json

@@ -173,7 +173,19 @@
       "group": "Overview",
       "pages": [
         "integrations/introduction",
-        "integrations/create"
+        {
+          "group": "Create an Integration",
+          "pages": [
+            "integrations/create",
+            "integrations/create-setup",
+            "integrations/create-anatomy",
+            "integrations/create-oauth",
+            "integrations/create-client",
+            "integrations/create-tasks",
+            "integrations/create-triggers",
+            "integrations/create-publishing"
+          ]
+        }
       ]
     },
     {

docs/sdk/io/runtask.mdx

@@ -10,12 +10,26 @@ A Task is a resumable unit of a Run that can be retried, resumed and is logged.
 
 <Snippet file="stable-key-param.mdx" />
 
-<ResponseField name="options" type="object" required>
-  The options of how you'd like to run and log the Task. `name` is the only required field.
+<ResponseField name="callback" type="function" required>
+  The callback that will be called when the Task is run, this is where your logic should go. The callback receives
+  the Task and the IO as parameters.
+
+  <Expandable title="arguments">
+    <ResponseField name="task" type="Task">
+      The Task that is running. It has some useful properties like `idempotencyKey` and `attempts`.
+    </ResponseField>
+    <ResponseField name="io" type="IO">
+      [IO](/sdk/io/overview) holds Integrations and useful actions you can perform.
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+
+<ResponseField name="options" type="object">
+  The options of how you'd like to run and log the Task.
 
 <Expandable title="options">
-  <ResponseField name="name" type="string" required>
-    The name of the Task is required. This is displayed on the Task in the logs.
+  <ResponseField name="name" type="string">
+    This is displayed on the Task in the logs.
   </ResponseField>
   <ResponseField name="delayUntil" type="date">
     The Task will wait and only start at the specified Date.
@@ -87,20 +101,6 @@ A Task is a resumable unit of a Run that can be retried, resumed and is logged.
 </Expandable>
 </ResponseField>
 
-<ResponseField name="callback" type="function" required>
-  The callback that will be called when the Task is run, this is where your logic should go. The callback receives
-  the Task and the IO as parameters.
-
-  <Expandable title="arguments">
-    <ResponseField name="task" type="Task">
-      The Task that is running. It has some useful properties like `idempotencyKey` and `attempts`.
-    </ResponseField>
-    <ResponseField name="io" type="IO">
-      [IO](/sdk/io/overview) holds Integrations and useful actions you can perform.
-    </ResponseField>
-  </Expandable>
-</ResponseField>
-
 <ResponseField name="onError" type="function">
   An optional callback that will be called when the Task fails. You can perform
   logic in here and optionally return a custom error object. Returning an object with `{ retryAt: Date, error?: Error }` will retry the Task at the specified Date. You can also just return a new `Error` object to throw a new error. Return nothing to rethrow the original error.

examples/job-catalog/package.json

@@ -13,6 +13,14 @@
     "schedules": "nodemon --watch src/schedules.ts -r tsconfig-paths/register -r dotenv/config src/schedules.ts",
     "stressTest": "nodemon --watch src/stressTest.ts -r tsconfig-paths/register -r dotenv/config src/stressTest.ts",
     "delays": "nodemon --watch src/delays.ts -r tsconfig-paths/register -r dotenv/config src/delays.ts",
+    "airtable": "nodemon --watch src/airtable.ts -r tsconfig-paths/register -r dotenv/config src/airtable.ts",
+    "resend": "nodemon --watch src/resend.ts -r tsconfig-paths/register -r dotenv/config src/resend.ts",
+    "github": "nodemon --watch src/github.ts -r tsconfig-paths/register -r dotenv/config src/github.ts",
+    "plain": "nodemon --watch src/plain.ts -r tsconfig-paths/register -r dotenv/config src/plain.ts",
+    "typeform": "nodemon --watch src/typeform.ts -r tsconfig-paths/register -r dotenv/config src/typeform.ts",
+    "dynamic-schedule": "nodemon --watch src/dynamic-schedule.ts -r tsconfig-paths/register -r dotenv/config src/dynamic-schedule.ts",
+    "dynamic-triggers": "nodemon --watch src/dynamic-triggers.ts -r tsconfig-paths/register -r dotenv/config src/dynamic-triggers.ts",
+    "background-fetch": "nodemon --watch src/background-fetch.ts -r tsconfig-paths/register -r dotenv/config src/background-fetch.ts",
     "dev:trigger": "trigger-cli dev --port 8080"
   },
   "dependencies": {
@@ -29,7 +37,8 @@
     "@trigger.dev/supabase": "workspace:*",
     "@types/node": "20.4.2",
     "typescript": "5.1.6",
-    "zod": "3.21.4"
+    "zod": "3.21.4",
+    "@trigger.dev/airtable": "workspace:*"
   },
   "trigger.dev": {
     "endpointId": "job-catalog"
@@ -43,4 +52,4 @@
     "ts-node": "^10.9.1",
     "tsconfig-paths": "^3.14.1"
   }
-}
+}

examples/job-catalog/src/airtable.ts

@@ -0,0 +1,123 @@
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { createExpressServer } from "@trigger.dev/express";
+import { z } from "zod";
+import { Airtable, Collaborator } from "@trigger.dev/airtable";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+// const airtable = new Airtable({
+//   id: "airtable",
+//   token: process.env["AIRTABLE_TOKEN"],
+// });
+
+const airtable = new Airtable({
+  id: "airtable-oauth",
+});
+
+type Status = "Live" | "Complete" | "In progress" | "Planning" | "In reviews";
+
+type LaunchGoalsAndOkRs = {
+  "Launch goals"?: string;
+  DRI?: Collaborator;
+  Team?: string;
+  Status?: "On track" | "In progress" | "At risk";
+  "Key results"?: Array<string>;
+  "Features (from 💻 Features table)"?: Array<string>;
+  "Status (from 💻 Features)": Array<Status>;
+};
+
+client.defineJob({
+  id: "airtable-example-1",
+  name: "Airtable Example 1: getRecords",
+  version: "0.1.0",
+  trigger: eventTrigger({
+    name: "airtable.example",
+    schema: z.object({
+      baseId: z.string(),
+      tableName: z.string(),
+    }),
+  }),
+  integrations: {
+    airtable,
+  },
+  run: async (payload, io, ctx) => {
+    const table = io.airtable.base(payload.baseId).table<LaunchGoalsAndOkRs>(payload.tableName);
+
+    const records = await table.getRecords("muliple records", { fields: ["Status"] });
+    await io.logger.log(records[0].fields.Status ?? "no status");
+
+    const aRecord = await table.getRecord("single", records[0].id);
+
+    const newRecords = await table.createRecords("create records", [
+      {
+        fields: { "Launch goals": "Created from Trigger.dev", Status: "In progress" },
+      },
+    ]);
+
+    const updatedRecords = await table.updateRecords(
+      "update records",
+      newRecords.map((record) => ({
+        id: record.id,
+        fields: { Status: "At risk" },
+      }))
+    );
+
+    await io.wait("5 secs", 5);
+
+    const deletedRecords = await table.deleteRecords(
+      "delete records",
+      updatedRecords.map((record) => record.id)
+    );
+  },
+});
+
+//todo webhooks require batch support
+// client.defineJob({
+//   id: "airtable-delete-webhooks",
+//   name: "Airtable Example 2: webhook admin",
+//   version: "0.1.0",
+//   trigger: eventTrigger({
+//     name: "airtable.example",
+//     schema: z.object({
+//       baseId: z.string(),
+//       deleteWebhooks: z.boolean().optional(),
+//     }),
+//   }),
+//   integrations: {
+//     airtable,
+//   },
+//   run: async (payload, io, ctx) => {
+//     const webhooks = await io.airtable.webhooks().list("list webhooks", { baseId: payload.baseId });
+
+//     if (payload.deleteWebhooks === true) {
+//       for (const webhook of webhooks.webhooks) {
+//         await io.airtable.webhooks().delete(`delete webhook: ${webhook.id}`, {
+//           baseId: payload.baseId,
+//           webhookId: webhook.id,
+//         });
+//       }
+//     }
+//   },
+// });
+
+//todo changes the structure of the trigger so it's airtable.base("val").onChanges({
+// client.defineJob({
+//   id: "airtable-on-table",
+//   name: "Airtable Example: onTable",
+//   version: "0.1.0",
+//   trigger: airtable.onTableChanges({
+//     baseId: "appSX6ly4nZGfdUSy",
+//     tableId: "tblr5BReu2yeOMk7n",
+//   }),
+//   run: async (payload, io, ctx) => {
+//     await io.logger.log(`transaction number ${payload.baseTransactionNumber}`);
+//   },
+// });
+
+createExpressServer(client);

examples/job-catalog/src/background-fetch.ts

@@ -0,0 +1,41 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { z } from "zod";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+  id: "test-background-fetch-retry",
+  name: "Test background fetch retry",
+  version: "0.0.1",
+  trigger: eventTrigger({
+    name: "test.background-fetch",
+    schema: z.object({
+      url: z.string(),
+      method: z.string().optional(),
+      headers: z.record(z.string()).optional(),
+      body: z.any().optional(),
+      retry: z.any().optional(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    return await io.backgroundFetch<any>(
+      "fetch",
+      payload.url,
+      {
+        method: payload.method ?? "GET",
+        headers: payload.headers,
+        body: payload.body ? JSON.stringify(payload.body) : undefined,
+      },
+      payload.retry
+    );
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/dynamic-schedule.ts

@@ -0,0 +1,72 @@
+import { DynamicSchedule, TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { createExpressServer } from "@trigger.dev/express";
+import { z } from "zod";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+const dynamicSchedule = new DynamicSchedule(client, {
+  id: "dynamic-interval",
+});
+
+client.defineJob({
+  id: "get-user-repo-on-schedule",
+  name: "Get User Repo On Schedule",
+  version: "0.1.1",
+  trigger: dynamicSchedule,
+
+  run: async (payload, io, ctx) => {
+    io.logger.log("Hello World");
+  },
+});
+
+client.defineJob({
+  id: "register-dynamic-interval",
+  name: "Register Dynamic Interval",
+  version: "0.1.1",
+  trigger: eventTrigger({
+    name: "dynamic.interval",
+    schema: z.object({
+      id: z.string(),
+      seconds: z.number().int().positive(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    await io.registerInterval("📆", dynamicSchedule, payload.id, {
+      seconds: payload.seconds,
+    });
+
+    await io.wait("wait", payload.seconds + 10);
+
+    await io.unregisterInterval("❌📆", dynamicSchedule, payload.id);
+  },
+});
+
+client.defineJob({
+  id: "register-dynamic-cron",
+  name: "Register Dynamic Cron",
+  version: "0.1.1",
+  trigger: eventTrigger({
+    name: "dynamic.cron",
+    schema: z.object({
+      id: z.string(),
+      cron: z.string(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    await io.registerCron("📆", dynamicSchedule, payload.id, {
+      cron: payload.cron,
+    });
+
+    await io.wait("wait", 60);
+
+    await io.unregisterCron("❌📆", dynamicSchedule, payload.id);
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/dynamic-triggers.ts

@@ -0,0 +1,76 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { Github, events } from "@trigger.dev/github";
+import { DynamicTrigger, TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { z } from "zod";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+const github = new Github({
+  id: "github-api-key",
+  token: process.env["GITHUB_API_KEY"]!,
+});
+
+const dynamicOnIssueOpenedTrigger = new DynamicTrigger(client, {
+  id: "github-issue-opened",
+  event: events.onIssueOpened,
+  source: github.sources.repo,
+});
+
+const dynamicUserTrigger = new DynamicTrigger(client, {
+  id: "dynamic-user-trigger",
+  event: events.onIssueOpened,
+  source: github.sources.repo,
+});
+
+client.defineJob({
+  id: "register-dynamic-trigger-on-new-repo",
+  name: "Register dynamic trigger on new repo",
+  version: "0.1.1",
+  trigger: eventTrigger({
+    name: "new.repo",
+    schema: z.object({ owner: z.string(), repo: z.string() }),
+  }),
+  run: async (payload, io, ctx) => {
+    return await io.registerTrigger("register-repo", dynamicOnIssueOpenedTrigger, payload.repo, {
+      owner: payload.owner,
+      repo: payload.repo,
+    });
+  },
+});
+
+client.defineJob({
+  id: "listen-for-dynamic-trigger",
+  name: "Listen for dynamic trigger",
+  version: "0.1.1",
+  trigger: dynamicOnIssueOpenedTrigger,
+
+  run: async (payload, io, ctx) => {},
+});
+
+client.defineJob({
+  id: "listen-for-dynamic-trigger-2",
+  name: "Listen for dynamic trigger-2",
+  version: "0.1.1",
+  trigger: dynamicOnIssueOpenedTrigger,
+
+  run: async (payload, io, ctx) => {},
+});
+
+client.defineJob({
+  id: "user-on-issue-opened",
+  name: "user on issue opened",
+  version: "0.1.1",
+  trigger: dynamicUserTrigger,
+
+  run: async (payload, io, ctx) => {
+    await io.logger.info("user-on-issue-opened", { ctx });
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/events.ts

@@ -19,7 +19,7 @@ client.defineJob({
     name: "event.example",
   }),
   run: async (payload, io, ctx) => {
-    await io.runTask("task-example-1", { name: "Task 1" }, async () => {
+    await io.runTask("task-example-1", async () => {
       return {
         message: "Hello World",
       };
@@ -71,28 +71,36 @@ client.defineJob({
   run: async (payload, io, ctx) => {
     await io.wait("sleeping", payload.delay);
 
-    await io.runTask("init", { name: "init" }, async () => {
-      console.log("init function ran", payload.userId);
-    });
+    await io.runTask(
+      "init",
+      async () => {
+        console.log("init function ran", payload.userId);
+      },
+      { name: "init" }
+    );
 
-    await io.runTask("failable", { name: "task-1", retry: { limit: 3 } }, async (task) => {
-      if (task.attempts > 2) {
-        console.log("task succeeded");
-        return {
-          ok: true,
-        };
-      }
-      console.log("task failed");
-      throw new Error(`Task failed on ${task.attempts} attempt(s)`);
-    });
+    await io.runTask(
+      "failable",
+      async (task) => {
+        if (task.attempts > 2) {
+          console.log("task succeeded");
+          return {
+            ok: true,
+          };
+        }
+        console.log("task failed");
+        throw new Error(`Task failed on ${task.attempts} attempt(s)`);
+      },
+      { name: "task-1", retry: { limit: 3 } }
+    );
 
     await io.runTask(
       "log",
-      {
-        name: "log",
-      },
       async () => {
         console.log("hello from the job", payload.userId);
+      },
+      {
+        name: "log",
       }
     );
   },

examples/job-catalog/src/github.ts

@@ -0,0 +1,393 @@
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { createExpressServer } from "@trigger.dev/express";
+import { Github, events } from "@trigger.dev/github";
+import { Slack } from "@trigger.dev/slack";
+import { z } from "zod";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+const githubApiKey = new Github({
+  id: "github-api-key",
+  token: process.env["GITHUB_API_KEY"]!,
+});
+
+const github = new Github({
+  id: "github",
+  octokitRequest: { fetch },
+});
+
+const slack = new Slack({ id: "my-slack-new" });
+
+client.defineJob({
+  id: "github-create-issue",
+  name: "GitHub Integration - Create issue",
+  version: "0.1.0",
+  trigger: eventTrigger({
+    name: "create-issue",
+    schema: z.object({
+      owner: z.string(),
+      repo: z.string(),
+      title: z.string(),
+    }),
+  }),
+  integrations: {
+    github: githubApiKey,
+  },
+  run: async (payload, io, ctx) => {
+    const issue = await io.github.createIssue("create issue", {
+      title: payload.title,
+      owner: payload.owner,
+      repo: payload.repo,
+    });
+
+    const { data } = await io.github.runTask(
+      "create-card",
+      async (client) => {
+        return client.rest.projects.createCard({
+          column_id: 123,
+          note: "test",
+        });
+      },
+      { name: "Create card" }
+    );
+
+    await io.github.createIssueCommentWithReaction("comment on issue with reaction", {
+      body: "This is a comment",
+      owner: payload.owner,
+      repo: payload.repo,
+      issueNumber: issue.number,
+      reaction: "heart",
+    });
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-issue",
+  name: "GitHub Integration - On Issue",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onIssue,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-issue-opened",
+  name: "GitHub Integration - On Issue Opened",
+  version: "0.1.0",
+  integrations: { github: githubApiKey },
+  trigger: githubApiKey.triggers.repo({
+    event: events.onIssueOpened,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.github.addIssueAssignees("add assignee", {
+      owner: payload.repository.owner.login,
+      repo: payload.repository.name,
+      issueNumber: payload.issue.number,
+      assignees: ["matt-aitken"],
+    });
+
+    await io.github.addIssueLabels("add label", {
+      owner: payload.repository.owner.login,
+      repo: payload.repository.name,
+      issueNumber: payload.issue.number,
+      labels: ["bug"],
+    });
+
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-issue-assigned",
+  name: "GitHub Integration - On Issue assigned",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onIssueAssigned,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-issue-commented",
+  name: "GitHub Integration - On Issue commented",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onIssueComment,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "star-slack-notification",
+  name: "New Star Slack Notification",
+  version: "0.1.0",
+  integrations: { slack },
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewStar,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    const response = await io.slack.postMessage("Slack star", {
+      text: `${payload.sender.login} starred ${payload.repository.full_name}.\nTotal: ${payload.repository.stargazers_count}⭐️`,
+      channel: "C04GWUTDC3W",
+    });
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-new-star",
+  name: "GitHub Integration - On New Star",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewStar,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-new-repo",
+  name: "GitHub Integration - On New Repository",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewRepository,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-new-branch-or-tag",
+  name: "GitHub Integration - On New Branch or Tag",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewBranchOrTag,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-new-branch",
+  name: "GitHub Integration - On New Branch",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewBranch,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-push",
+  name: "GitHub Integration - On Push",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onPush,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-pull-request",
+  name: "GitHub Integration - On Pull Request",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onPullRequest,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-pull-request-review",
+  name: "GitHub Integration - On Pull Request Review",
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onPullRequestReview,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-on-pull-request-merge-commit",
+  name: "GitHub Integration - on Pull Request Merge Commit",
+  version: "0.1.0",
+  integrations: { github },
+  trigger: githubApiKey.triggers.repo({
+    event: events.onPullRequest,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+
+    if (payload.pull_request.merged && payload.pull_request.merge_commit_sha) {
+      const commit = await io.github.getCommit("get merge commit", {
+        owner: payload.repository.owner.login,
+        repo: payload.repository.name,
+        commitSHA: payload.pull_request.merge_commit_sha,
+      });
+      await io.logger.info("Merge Commit Details", commit);
+    }
+
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-get-tree",
+  name: "GitHub Integration - Get Tree",
+  version: "0.1.0",
+  integrations: { github },
+  trigger: githubApiKey.triggers.repo({
+    event: events.onPullRequest,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+
+    if (payload.pull_request.merged && payload.pull_request.merge_commit_sha) {
+      const tree = await io.github.getTree("get merge commit", {
+        owner: payload.repository.owner.login,
+        repo: payload.repository.name,
+        treeSHA: payload.pull_request.merge_commit_sha,
+      });
+      await io.logger.info("Tree ", tree);
+    }
+
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-get-reference",
+  name: "GitHub Integration - Get Reference",
+  integrations: { github },
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewBranch,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+
+    const ref = await io.github.getReference("Get reference", {
+      owner: payload.repository.owner.login,
+      repo: payload.repository.name,
+      ref: payload.ref,
+    });
+
+    await io.logger.info("Reference ", ref);
+
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-list-matching-references",
+  name: "GitHub Integration - List Matching References",
+  integrations: { github },
+  version: "0.1.0",
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewBranch,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+
+    const ref = await io.github.listMatchingReferences("List Matching References", {
+      owner: payload.repository.owner.login,
+      repo: payload.repository.name,
+      ref: payload.ref,
+    });
+
+    await io.logger.info("Reference ", ref);
+
+    return { payload, ctx };
+  },
+});
+
+client.defineJob({
+  id: "github-integration-get-tag",
+  name: "GitHub Integration - Get Tag",
+  version: "0.1.0",
+  integrations: { github },
+  trigger: githubApiKey.triggers.repo({
+    event: events.onNewBranchOrTag,
+    owner: "triggerdotdev",
+    repo: "empty",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.info("This is a simple log info message");
+    if (payload.ref_type === "tag") {
+      const tag = io.github.getTag("Get Tag", {
+        owner: payload.repository.owner.login,
+        repo: payload.repository.name,
+        tagSHA: payload.ref,
+      });
+      await io.logger.info("Tag ", tag);
+    }
+    return { payload, ctx };
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/plain.ts

@@ -0,0 +1,91 @@
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { createExpressServer } from "@trigger.dev/express";
+import {
+  ComponentDividerSpacingSize,
+  ComponentTextColor,
+  ComponentTextSize,
+  Plain,
+} from "@trigger.dev/plain";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+const plain = new Plain({
+  id: "plain-1",
+  apiKey: process.env["PLAIN_API_KEY"]!,
+});
+
+client.defineJob({
+  id: "plain-playground",
+  name: "Plain Playground",
+  version: "0.1.1",
+  integrations: {
+    plain,
+  },
+  trigger: eventTrigger({
+    name: "plain.playground",
+  }),
+  run: async (payload, io, ctx) => {
+    const { customer } = await io.plain.upsertCustomer("upsert-customer", {
+      identifier: {
+        emailAddress: "eric@trigger.dev",
+      },
+      onCreate: {
+        email: {
+          email: "eric@trigger.dev",
+          isVerified: true,
+        },
+        fullName: "Eric Allam",
+        externalId: "123",
+      },
+      onUpdate: {
+        fullName: {
+          value: "Eric Allam",
+        },
+        externalId: {
+          value: "123",
+        },
+      },
+    });
+
+    const foundCustomer = await io.plain.getCustomerById("get-customer", {
+      customerId: customer.id,
+    });
+
+    const timelineEntry = await io.plain.upsertCustomTimelineEntry("upsert-timeline-entry", {
+      customerId: customer.id,
+      title: "My timeline entry",
+      components: [
+        {
+          componentText: {
+            text: `This is a nice title`,
+          },
+        },
+        {
+          componentDivider: {
+            dividerSpacingSize: ComponentDividerSpacingSize.M,
+          },
+        },
+        {
+          componentText: {
+            textSize: ComponentTextSize.S,
+            textColor: ComponentTextColor.Muted,
+            text: "External id",
+          },
+        },
+        {
+          componentText: {
+            text: foundCustomer?.externalId ?? "",
+          },
+        },
+      ],
+    });
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/resend.ts

@@ -1,8 +1,16 @@
-import { client } from "@/trigger";
+import { createExpressServer } from "@trigger.dev/express";
 import { Resend } from "@trigger.dev/resend";
-import { Job, eventTrigger } from "@trigger.dev/sdk";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
 import { z } from "zod";
 
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
 const resend = new Resend({
   id: "resend-client",
   apiKey: process.env.RESEND_API_KEY!,
@@ -24,11 +32,15 @@ client.defineJob({
     resend,
   },
   run: async (payload, io, ctx) => {
-    await io.resend.sendEmail("📧", {
+    const response = await io.resend.sendEmail("📧", {
       to: payload.to,
       subject: payload.subject,
       text: payload.text,
       from: "Trigger.dev <hello@email.trigger.dev>",
     });
+
+    await io.logger.info("Sent email", { response });
   },
 });
+
+createExpressServer(client);

examples/job-catalog/src/schedules.ts

@@ -1,5 +1,6 @@
 import { createExpressServer } from "@trigger.dev/express";
-import { TriggerClient, intervalTrigger } from "@trigger.dev/sdk";
+import { Resend } from "@trigger.dev/resend";
+import { TriggerClient, cronTrigger, intervalTrigger } from "@trigger.dev/sdk";
 
 export const client = new TriggerClient({
   id: "job-catalog",
@@ -18,7 +19,7 @@ client.defineJob({
     seconds: 60 * 3, // 3 minutes
   }),
   run: async (payload, io, ctx) => {
-    await io.runTask("task-example-1", { name: "Task 1" }, async () => {
+    await io.runTask("task-example-1", async () => {
       return {
         message: "Hello World",
       };
@@ -30,4 +31,43 @@ client.defineJob({
   },
 });
 
+const resend = new Resend({
+  id: "resend-client",
+  apiKey: process.env.RESEND_API_KEY!,
+});
+
+client.defineJob({
+  id: "weekly-kpi-report",
+  name: "Weekly KPI report",
+  version: "1.0.0",
+  trigger: cronTrigger({
+    // Every Friday at 5pm (UTC)
+    cron: "0 17 * * 5",
+  }),
+  integrations: {
+    resend,
+  },
+  run: async (payload, io, ctx) => {
+    const kpis = await io.runTask("get-kpis", async () => {
+      return await getKpiData();
+    });
+
+    const emailList = ["jen@whatever.com", "ann@whatever.com"];
+
+    await io.resend.sendEmail("send-kpis", {
+      to: emailList,
+      subject: "Weekly KPI report",
+      text: `Users: ${kpis.users}, Revenue: ${kpis.revenue}`,
+      from: "Trigger.dev <hello@email.trigger.dev>",
+    });
+  },
+});
+
+async function getKpiData() {
+  return {
+    users: 100_000,
+    revenue: 1_000_000,
+  };
+}
+
 createExpressServer(client);

examples/job-catalog/src/slack.ts

@@ -23,7 +23,7 @@ client.defineJob({
     slack,
   },
   run: async (payload, io, ctx) => {
-    await io.slack.postMessage("Slack 📝", {
+    const message = await io.slack.postMessage("Slack 📝", {
       channel: "C04GWUTDC3W",
       text: "Welcome to the team, Eric!",
     });

examples/job-catalog/src/stressTest.ts

@@ -19,23 +19,124 @@ client.defineJob({
   run: async (payload, io, ctx) => {
     // Run 10 tasks, each with a 300KB output
     for (let i = 0; i < 10; i++) {
-      await io.runTask(`task-${i}`, { name: `Task ${i}` }, async (task) => {
-        return {
-          output: "a".repeat(300 * 1024),
-        };
-      });
+      await io.runTask(
+        `task-${i}`,
+        async (task) => {
+          return {
+            output: "a".repeat(300 * 1024),
+          };
+        },
+        { name: `Task ${i}` }
+      );
     }
 
     // Now run a single task with 5MB output
-    await io.runTask(`task-5mb`, { name: `Task 5MB` }, async (task) => {
-      return {
-        output: "a".repeat(5 * 1024 * 1024),
-      };
-    });
+    await io.runTask(
+      `task-5mb`,
+      async (task) => {
+        return {
+          output: "a".repeat(5 * 1024 * 1024),
+        };
+      },
+      { name: `Task 5MB` }
+    );
 
     // Now do a wait for 5 seconds
     await io.wait("wait", 5);
   },
 });
 
+client.defineJob({
+  id: "stress-test-2",
+  name: "Stress Test 2",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "stress.test.2",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.runTask(
+      `task-1`,
+      async (task) => {
+        const response = await fetch("https://jsonplaceholder.typicode.com/photos");
+        return await response.json();
+      },
+      { name: `Task 1` }
+    );
+
+    await io.runTask(
+      `task-2`,
+      async (task) => {
+        const response = await fetch("https://jsonplaceholder.typicode.com/comments");
+        return await response.json();
+      },
+      { name: `Task 2` }
+    );
+
+    await io.runTask(
+      `task-3`,
+      async (task) => {
+        const response = await fetch("https://jsonplaceholder.typicode.com/photos");
+
+        return await response.json();
+      },
+      { name: `Task 3` }
+    );
+
+    await io.runTask(
+      `task-4`,
+      async (task) => {
+        const response = await fetch("https://jsonplaceholder.typicode.com/comments");
+
+        return await response.json();
+      },
+      { name: `Task 4` }
+    );
+
+    const response = await io.runTask(
+      `task-5`,
+      async (task) => {
+        const response = await fetch("https://jsonplaceholder.typicode.com/photos");
+
+        return await response.json();
+      },
+      { name: `Task 5` }
+    );
+
+    await io.runTask(
+      `task-6`,
+      async (task) => {
+        const response = await fetch("https://jsonplaceholder.typicode.com/users");
+
+        return await response.json();
+      },
+      { name: `Task 6` }
+    );
+
+    return response;
+  },
+});
+
+client.defineJob({
+  id: "long.running",
+  name: "Long Running Job",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "long.running",
+  }),
+  run: async (payload, io, ctx) => {
+    // Perform X tasks in an iteration, each one taking X milliseconds
+    for (let i = 0; i < payload.iterations; i++) {
+      await io.runTask(
+        `task.${i}`,
+        async (task) => {
+          await new Promise((resolve) => setTimeout(resolve, payload.duration ?? 5000));
+
+          return { i };
+        },
+        { name: `Task ${i}` }
+      );
+    }
+  },
+});
+
 createExpressServer(client);

examples/job-catalog/src/stripe.ts

@@ -40,30 +40,6 @@ client.defineJob({
   },
 });
 
-client.defineJob({
-  id: "stripe-example-1",
-  name: "Stripe Example 1",
-  version: "0.1.0",
-  trigger: eventTrigger({
-    name: "stripe.example",
-    schema: z.object({
-      customerId: z.string(),
-      source: z.string(),
-    }),
-  }),
-  integrations: {
-    stripe,
-  },
-  run: async (payload, io, ctx) => {
-    await io.stripe.createCharge("create-charge", {
-      amount: 100,
-      currency: "usd",
-      source: payload.source,
-      customer: payload.customerId,
-    });
-  },
-});
-
 client.defineJob({
   id: "stripe-create-customer",
   name: "Stripe Create Customer",

examples/job-catalog/src/typeform.ts

@@ -0,0 +1,63 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { Typeform } from "@trigger.dev/typeform";
+import { z } from "zod";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+export const typeform = new Typeform({
+  id: "typeform-1",
+  token: process.env["TYPEFORM_API_KEY"]!,
+});
+
+client.defineJob({
+  id: "typeform-playground",
+  name: "Typeform Playground",
+  version: "0.1.1",
+  integrations: {
+    typeform,
+  },
+  trigger: eventTrigger({
+    name: "typeform.playground",
+    schema: z.object({
+      formId: z.string().optional(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    await io.typeform.listForms("list-forms");
+
+    if (payload.formId) {
+      const form = await io.typeform.getForm("get-form", {
+        uid: payload.formId,
+      });
+
+      const listResponses = await io.typeform.listResponses("list-responses", {
+        uid: payload.formId,
+        pageSize: 50,
+      });
+
+      const allResponses = await io.typeform.getAllResponses("get-all-responses", {
+        uid: payload.formId,
+      });
+    }
+  },
+});
+
+client.defineJob({
+  id: "typeform-webhook-2",
+  name: "Typeform Webhook 2",
+  version: "0.1.1",
+  trigger: typeform.onFormResponse({
+    uid: "KywLXMeB",
+    tag: "tag1",
+  }),
+  run: async (payload, io, ctx) => {},
+});
+
+createExpressServer(client);

examples/job-catalog/tsconfig.json

@@ -1,37 +1,102 @@
 {
   "extends": "@trigger.dev/tsconfig/node18.json",
-  "include": ["./src/**/*.ts"],
+  "include": [
+    "./src/**/*.ts"
+  ],
   "compilerOptions": {
     "baseUrl": ".",
-    "lib": ["DOM", "DOM.Iterable"],
+    "lib": [
+      "DOM",
+      "DOM.Iterable"
+    ],
     "paths": {
-      "@/*": ["./src/*"],
-      "@trigger.dev/sdk": ["../../packages/trigger-sdk/src/index"],
-      "@trigger.dev/sdk/*": ["../../packages/trigger-sdk/src/*"],
-      "@trigger.dev/express": ["../../packages/express/src/index"],
-      "@trigger.dev/express/*": ["../../packages/express/src/*"],
-      "@trigger.dev/core": ["../../packages/core/src/index"],
-      "@trigger.dev/core/*": ["../../packages/core/src/*"],
-      "@trigger.dev/integration-kit": ["../../packages/integration-kit/src/index"],
-      "@trigger.dev/integration-kit/*": ["../../packages/integration-kit/src/*"],
-      "@trigger.dev/github": ["../../integrations/github/src/index"],
-      "@trigger.dev/github/*": ["../../integrations/github/src/*"],
-      "@trigger.dev/slack": ["../../integrations/slack/src/index"],
-      "@trigger.dev/slack/*": ["../../integrations/slack/src/*"],
-      "@trigger.dev/openai": ["../../integrations/openai/src/index"],
-      "@trigger.dev/openai/*": ["../../integrations/openai/src/*"],
-      "@trigger.dev/resend": ["../../integrations/resend/src/index"],
-      "@trigger.dev/resend/*": ["../../integrations/resend/src/*"],
-      "@trigger.dev/typeform": ["../../integrations/typeform/src/index"],
-      "@trigger.dev/typeform/*": ["../../integrations/typeform/src/*"],
-      "@trigger.dev/plain": ["../../integrations/plain/src/index"],
-      "@trigger.dev/plain/*": ["../../integrations/plain/src/*"],
-      "@trigger.dev/supabase": ["../../integrations/supabase/src/index"],
-      "@trigger.dev/supabase/*": ["../../integrations/supabase/src/*"],
-      "@trigger.dev/stripe": ["../../integrations/stripe/src/index"],
-      "@trigger.dev/stripe/*": ["../../integrations/stripe/src/*"],
-      "@trigger.dev/sendgrid": ["../../integrations/sendgrid/src/index"],
-      "@trigger.dev/sendgrid/*": ["../../integrations/sendgrid/src/*"]
+      "@/*": [
+        "./src/*"
+      ],
+      "@trigger.dev/sdk": [
+        "../../packages/trigger-sdk/src/index"
+      ],
+      "@trigger.dev/sdk/*": [
+        "../../packages/trigger-sdk/src/*"
+      ],
+      "@trigger.dev/express": [
+        "../../packages/express/src/index"
+      ],
+      "@trigger.dev/express/*": [
+        "../../packages/express/src/*"
+      ],
+      "@trigger.dev/core": [
+        "../../packages/core/src/index"
+      ],
+      "@trigger.dev/core/*": [
+        "../../packages/core/src/*"
+      ],
+      "@trigger.dev/integration-kit": [
+        "../../packages/integration-kit/src/index"
+      ],
+      "@trigger.dev/integration-kit/*": [
+        "../../packages/integration-kit/src/*"
+      ],
+      "@trigger.dev/github": [
+        "../../integrations/github/src/index"
+      ],
+      "@trigger.dev/github/*": [
+        "../../integrations/github/src/*"
+      ],
+      "@trigger.dev/slack": [
+        "../../integrations/slack/src/index"
+      ],
+      "@trigger.dev/slack/*": [
+        "../../integrations/slack/src/*"
+      ],
+      "@trigger.dev/openai": [
+        "../../integrations/openai/src/index"
+      ],
+      "@trigger.dev/openai/*": [
+        "../../integrations/openai/src/*"
+      ],
+      "@trigger.dev/resend": [
+        "../../integrations/resend/src/index"
+      ],
+      "@trigger.dev/resend/*": [
+        "../../integrations/resend/src/*"
+      ],
+      "@trigger.dev/typeform": [
+        "../../integrations/typeform/src/index"
+      ],
+      "@trigger.dev/typeform/*": [
+        "../../integrations/typeform/src/*"
+      ],
+      "@trigger.dev/plain": [
+        "../../integrations/plain/src/index"
+      ],
+      "@trigger.dev/plain/*": [
+        "../../integrations/plain/src/*"
+      ],
+      "@trigger.dev/supabase": [
+        "../../integrations/supabase/src/index"
+      ],
+      "@trigger.dev/supabase/*": [
+        "../../integrations/supabase/src/*"
+      ],
+      "@trigger.dev/stripe": [
+        "../../integrations/stripe/src/index"
+      ],
+      "@trigger.dev/stripe/*": [
+        "../../integrations/stripe/src/*"
+      ],
+      "@trigger.dev/sendgrid": [
+        "../../integrations/sendgrid/src/index"
+      ],
+      "@trigger.dev/sendgrid/*": [
+        "../../integrations/sendgrid/src/*"
+      ],
+      "@trigger.dev/airtable": [
+        "../../integrations/airtable/src/index"
+      ],
+      "@trigger.dev/airtable/*": [
+        "../../integrations/airtable/src/*"
+      ]
     }
   }
-}
+}

examples/jobs-starter/jobs/starter.ts

@@ -71,7 +71,7 @@ client.defineJob({
   run: async (_payload, io, _ctx) => {
     return await io.runTask(
       "get-stars-count",
-      { name: "Get Trigger.dev stars count" },
+
       async () => {
         try {
           const response = await fetch("https://api.github.com/repos/triggerdotdev/trigger.dev");
@@ -83,7 +83,8 @@ client.defineJob({
 
           return { success: false };
         }
-      }
+      },
+      { name: "Get Trigger.dev stars count" }
     );
   },
 });