WIP (fly.io multi-tenant idea)

.changeset/witty-weeks-fly.md

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

.vscode/launch.json

@@ -19,6 +19,14 @@
       "name": "Chrome webapp",
       "url": "http://localhost:3030",
       "webRoot": "${workspaceFolder}/apps/webapp/app"
+    },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "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/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/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/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.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/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/endpoints/indexEndpoint.server.ts

@@ -9,6 +9,8 @@ import { RegisterDynamicScheduleService } from "../triggers/registerDynamicSched
 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;
@@ -16,6 +18,8 @@ export class IndexEndpointService {
   #disableJobService = new DisableJobService();
   #registerSourceServiceV1 = new RegisterSourceServiceV1();
   #registerSourceServiceV2 = new RegisterSourceServiceV2();
+  #registerBackgroundTaskService = new RegisterBackgroundTaskService();
+  #disableBackgroundTaskService = new DisableBackgroundTaskService();
   #registerDynamicTriggerService = new RegisterDynamicTriggerService();
   #registerDynamicScheduleService = new RegisterDynamicScheduleService();
 
@@ -40,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,
@@ -53,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({
@@ -156,6 +169,100 @@ 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 {
         switch (source.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/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/worker.server.ts

@@ -20,6 +20,9 @@ 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({
@@ -68,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 = {
@@ -285,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);
+        },
+      },
     },
   });
 }

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",

examples/nextjs-background-tasks/.gitignore

@@ -0,0 +1,35 @@
+# See https://help.github.com/articles/ignoring-files/ for more about ignoring files.
+
+# dependencies
+/node_modules
+/.pnp
+.pnp.js
+
+# testing
+/coverage
+
+# next.js
+/.next/
+/out/
+
+# production
+/build
+
+# misc
+.DS_Store
+*.pem
+
+# debug
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+
+# local env files
+.env*.local
+
+# vercel
+.vercel
+
+# typescript
+*.tsbuildinfo
+next-env.d.ts

examples/nextjs-background-tasks/README.md

@@ -0,0 +1,34 @@
+This is a [Next.js](https://nextjs.org/) project bootstrapped with [`create-next-app`](https://github.com/vercel/next.js/tree/canary/packages/create-next-app).
+
+## Getting Started
+
+First, run the development server:
+
+```bash
+npm run dev
+# or
+yarn dev
+# or
+pnpm dev
+```
+
+Open [http://localhost:3000](http://localhost:3000) with your browser to see the result.
+
+You can start editing the page by modifying `app/page.tsx`. The page auto-updates as you edit the file.
+
+This project uses [`next/font`](https://nextjs.org/docs/basic-features/font-optimization) to automatically optimize and load Inter, a custom Google Font.
+
+## Learn More
+
+To learn more about Next.js, take a look at the following resources:
+
+- [Next.js Documentation](https://nextjs.org/docs) - learn about Next.js features and API.
+- [Learn Next.js](https://nextjs.org/learn) - an interactive Next.js tutorial.
+
+You can check out [the Next.js GitHub repository](https://github.com/vercel/next.js/) - your feedback and contributions are welcome!
+
+## Deploy on Vercel
+
+The easiest way to deploy your Next.js app is to use the [Vercel Platform](https://vercel.com/new?utm_medium=default-template&filter=next.js&utm_source=create-next-app&utm_campaign=create-next-app-readme) from the creators of Next.js.
+
+Check out our [Next.js deployment documentation](https://nextjs.org/docs/deployment) for more details.

examples/nextjs-background-tasks/next.config.js

@@ -0,0 +1,4 @@
+/** @type {import('next').NextConfig} */
+const nextConfig = {}
+
+module.exports = nextConfig

examples/nextjs-background-tasks/package.json

@@ -0,0 +1,32 @@
+{
+  "name": "nextjs-background-tasks",
+  "version": "0.1.0",
+  "private": true,
+  "scripts": {
+    "dev": "next dev",
+    "build": "next build",
+    "start": "next start",
+    "lint": "next lint"
+  },
+  "dependencies": {
+    "@trigger.dev/nextjs": "workspace:*",
+    "@trigger.dev/sdk": "workspace:*",
+    "@types/node": "20.5.9",
+    "@types/react": "18.2.17",
+    "@types/react-dom": "18.2.7",
+    "autoprefixer": "10.4.15",
+    "next": "13.4.19",
+    "postcss": "8.4.29",
+    "react": "18.2.0",
+    "react-dom": "18.2.0",
+    "tailwindcss": "3.3.3",
+    "typescript": "5.2.2",
+    "zod": "3.21.4"
+  },
+  "devDependencies": {
+    "@trigger.dev/cli": "workspace:*"
+  },
+  "trigger.dev": {
+    "endpointId": "background-tasks-z2kp"
+  }
+}

examples/nextjs-background-tasks/postcss.config.js

@@ -0,0 +1,6 @@
+module.exports = {
+  plugins: {
+    tailwindcss: {},
+    autoprefixer: {},
+  },
+}

examples/nextjs-background-tasks/public/next.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" fill="none" viewBox="0 0 394 80"><path fill="#000" d="M262 0h68.5v12.7h-27.2v66.6h-13.6V12.7H262V0ZM149 0v12.7H94v20.4h44.3v12.6H94v21h55v12.6H80.5V0h68.7zm34.3 0h-17.8l63.8 79.4h17.9l-32-39.7 32-39.6h-17.9l-23 28.6-23-28.6zm18.3 56.7-9-11-27.1 33.7h17.8l18.3-22.7z"/><path fill="#000" d="M81 79.3 17 0H0v79.3h13.6V17l50.2 62.3H81Zm252.6-.4c-1 0-1.8-.4-2.5-1s-1.1-1.6-1.1-2.6.3-1.8 1-2.5 1.6-1 2.6-1 1.8.3 2.5 1a3.4 3.4 0 0 1 .6 4.3 3.7 3.7 0 0 1-3 1.8zm23.2-33.5h6v23.3c0 2.1-.4 4-1.3 5.5a9.1 9.1 0 0 1-3.8 3.5c-1.6.8-3.5 1.3-5.7 1.3-2 0-3.7-.4-5.3-1s-2.8-1.8-3.7-3.2c-.9-1.3-1.4-3-1.4-5h6c.1.8.3 1.6.7 2.2s1 1.2 1.6 1.5c.7.4 1.5.5 2.4.5 1 0 1.8-.2 2.4-.6a4 4 0 0 0 1.6-1.8c.3-.8.5-1.8.5-3V45.5zm30.9 9.1a4.4 4.4 0 0 0-2-3.3 7.5 7.5 0 0 0-4.3-1.1c-1.3 0-2.4.2-3.3.5-.9.4-1.6 1-2 1.6a3.5 3.5 0 0 0-.3 4c.3.5.7.9 1.3 1.2l1.8 1 2 .5 3.2.8c1.3.3 2.5.7 3.7 1.2a13 13 0 0 1 3.2 1.8 8.1 8.1 0 0 1 3 6.5c0 2-.5 3.7-1.5 5.1a10 10 0 0 1-4.4 3.5c-1.8.8-4.1 1.2-6.8 1.2-2.6 0-4.9-.4-6.8-1.2-2-.8-3.4-2-4.5-3.5a10 10 0 0 1-1.7-5.6h6a5 5 0 0 0 3.5 4.6c1 .4 2.2.6 3.4.6 1.3 0 2.5-.2 3.5-.6 1-.4 1.8-1 2.4-1.7a4 4 0 0 0 .8-2.4c0-.9-.2-1.6-.7-2.2a11 11 0 0 0-2.1-1.4l-3.2-1-3.8-1c-2.8-.7-5-1.7-6.6-3.2a7.2 7.2 0 0 1-2.4-5.7 8 8 0 0 1 1.7-5 10 10 0 0 1 4.3-3.5c2-.8 4-1.2 6.4-1.2 2.3 0 4.4.4 6.2 1.2 1.8.8 3.2 2 4.3 3.4 1 1.4 1.5 3 1.5 5h-5.8z"/></svg>

examples/nextjs-background-tasks/public/vercel.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" fill="none" viewBox="0 0 283 64"><path fill="black" d="M141 16c-11 0-19 7-19 18s9 18 20 18c7 0 13-3 16-7l-7-5c-2 3-6 4-9 4-5 0-9-3-10-7h28v-3c0-11-8-18-19-18zm-9 15c1-4 4-7 9-7s8 3 9 7h-18zm117-15c-11 0-19 7-19 18s9 18 20 18c6 0 12-3 16-7l-8-5c-2 3-5 4-8 4-5 0-9-3-11-7h28l1-3c0-11-8-18-19-18zm-10 15c2-4 5-7 10-7s8 3 9 7h-19zm-39 3c0 6 4 10 10 10 4 0 7-2 9-5l8 5c-3 5-9 8-17 8-11 0-19-7-19-18s8-18 19-18c8 0 14 3 17 8l-8 5c-2-3-5-5-9-5-6 0-10 4-10 10zm83-29v46h-9V5h9zM37 0l37 64H0L37 0zm92 5-27 48L74 5h10l18 30 17-30h10zm59 12v10l-3-1c-6 0-10 4-10 10v15h-9V17h9v9c0-5 6-9 13-9z"/></svg>

examples/nextjs-background-tasks/src/app/api/trigger/route.ts

@@ -0,0 +1,9 @@
+
+import { createAppRoute } from "@trigger.dev/nextjs";
+import { client } from "@/trigger";
+
+
+import "@/jobs";
+
+//this route is used to send and receive data with Trigger.dev
+export const { POST, dynamic } = createAppRoute(client);

examples/nextjs-background-tasks/src/app/globals.css

@@ -0,0 +1,27 @@
+@tailwind base;
+@tailwind components;
+@tailwind utilities;
+
+:root {
+  --foreground-rgb: 0, 0, 0;
+  --background-start-rgb: 214, 219, 220;
+  --background-end-rgb: 255, 255, 255;
+}
+
+@media (prefers-color-scheme: dark) {
+  :root {
+    --foreground-rgb: 255, 255, 255;
+    --background-start-rgb: 0, 0, 0;
+    --background-end-rgb: 0, 0, 0;
+  }
+}
+
+body {
+  color: rgb(var(--foreground-rgb));
+  background: linear-gradient(
+      to bottom,
+      transparent,
+      rgb(var(--background-end-rgb))
+    )
+    rgb(var(--background-start-rgb));
+}

examples/nextjs-background-tasks/src/app/layout.tsx

@@ -0,0 +1,22 @@
+import './globals.css'
+import type { Metadata } from 'next'
+import { Inter } from 'next/font/google'
+
+const inter = Inter({ subsets: ['latin'] })
+
+export const metadata: Metadata = {
+  title: 'Create Next App',
+  description: 'Generated by create next app',
+}
+
+export default function RootLayout({
+  children,
+}: {
+  children: React.ReactNode
+}) {
+  return (
+    <html lang="en">
+      <body className={inter.className}>{children}</body>
+    </html>
+  )
+}

examples/nextjs-background-tasks/src/app/page.tsx

@@ -0,0 +1,113 @@
+import Image from 'next/image'
+
+export default function Home() {
+  return (
+    <main className="flex min-h-screen flex-col items-center justify-between p-24">
+      <div className="z-10 max-w-5xl w-full items-center justify-between font-mono text-sm lg:flex">
+        <p className="fixed left-0 top-0 flex w-full justify-center border-b border-gray-300 bg-gradient-to-b from-zinc-200 pb-6 pt-8 backdrop-blur-2xl dark:border-neutral-800 dark:bg-zinc-800/30 dark:from-inherit lg:static lg:w-auto  lg:rounded-xl lg:border lg:bg-gray-200 lg:p-4 lg:dark:bg-zinc-800/30">
+          Get started by editing&nbsp;
+          <code className="font-mono font-bold">src/app/page.tsx</code>
+        </p>
+        <div className="fixed bottom-0 left-0 flex h-48 w-full items-end justify-center bg-gradient-to-t from-white via-white dark:from-black dark:via-black lg:static lg:h-auto lg:w-auto lg:bg-none">
+          <a
+            className="pointer-events-none flex place-items-center gap-2 p-8 lg:pointer-events-auto lg:p-0"
+            href="https://vercel.com?utm_source=create-next-app&utm_medium=appdir-template&utm_campaign=create-next-app"
+            target="_blank"
+            rel="noopener noreferrer"
+          >
+            By{' '}
+            <Image
+              src="/vercel.svg"
+              alt="Vercel Logo"
+              className="dark:invert"
+              width={100}
+              height={24}
+              priority
+            />
+          </a>
+        </div>
+      </div>
+
+      <div className="relative flex place-items-center before:absolute before:h-[300px] before:w-[480px] before:-translate-x-1/2 before:rounded-full before:bg-gradient-radial before:from-white before:to-transparent before:blur-2xl before:content-[''] after:absolute after:-z-20 after:h-[180px] after:w-[240px] after:translate-x-1/3 after:bg-gradient-conic after:from-sky-200 after:via-blue-200 after:blur-2xl after:content-[''] before:dark:bg-gradient-to-br before:dark:from-transparent before:dark:to-blue-700 before:dark:opacity-10 after:dark:from-sky-900 after:dark:via-[#0141ff] after:dark:opacity-40 before:lg:h-[360px] z-[-1]">
+        <Image
+          className="relative dark:drop-shadow-[0_0_0.3rem_#ffffff70] dark:invert"
+          src="/next.svg"
+          alt="Next.js Logo"
+          width={180}
+          height={37}
+          priority
+        />
+      </div>
+
+      <div className="mb-32 grid text-center lg:max-w-5xl lg:w-full lg:mb-0 lg:grid-cols-4 lg:text-left">
+        <a
+          href="https://nextjs.org/docs?utm_source=create-next-app&utm_medium=appdir-template&utm_campaign=create-next-app"
+          className="group rounded-lg border border-transparent px-5 py-4 transition-colors hover:border-gray-300 hover:bg-gray-100 hover:dark:border-neutral-700 hover:dark:bg-neutral-800/30"
+          target="_blank"
+          rel="noopener noreferrer"
+        >
+          <h2 className={`mb-3 text-2xl font-semibold`}>
+            Docs{' '}
+            <span className="inline-block transition-transform group-hover:translate-x-1 motion-reduce:transform-none">
+              -&gt;
+            </span>
+          </h2>
+          <p className={`m-0 max-w-[30ch] text-sm opacity-50`}>
+            Find in-depth information about Next.js features and API.
+          </p>
+        </a>
+
+        <a
+          href="https://nextjs.org/learn?utm_source=create-next-app&utm_medium=appdir-template-tw&utm_campaign=create-next-app"
+          className="group rounded-lg border border-transparent px-5 py-4 transition-colors hover:border-gray-300 hover:bg-gray-100 hover:dark:border-neutral-700 hover:dark:bg-neutral-800/30"
+          target="_blank"
+          rel="noopener noreferrer"
+        >
+          <h2 className={`mb-3 text-2xl font-semibold`}>
+            Learn{' '}
+            <span className="inline-block transition-transform group-hover:translate-x-1 motion-reduce:transform-none">
+              -&gt;
+            </span>
+          </h2>
+          <p className={`m-0 max-w-[30ch] text-sm opacity-50`}>
+            Learn about Next.js in an interactive course with&nbsp;quizzes!
+          </p>
+        </a>
+
+        <a
+          href="https://vercel.com/templates?framework=next.js&utm_source=create-next-app&utm_medium=appdir-template&utm_campaign=create-next-app"
+          className="group rounded-lg border border-transparent px-5 py-4 transition-colors hover:border-gray-300 hover:bg-gray-100 hover:dark:border-neutral-700 hover:dark:bg-neutral-800/30"
+          target="_blank"
+          rel="noopener noreferrer"
+        >
+          <h2 className={`mb-3 text-2xl font-semibold`}>
+            Templates{' '}
+            <span className="inline-block transition-transform group-hover:translate-x-1 motion-reduce:transform-none">
+              -&gt;
+            </span>
+          </h2>
+          <p className={`m-0 max-w-[30ch] text-sm opacity-50`}>
+            Explore the Next.js 13 playground.
+          </p>
+        </a>
+
+        <a
+          href="https://vercel.com/new?utm_source=create-next-app&utm_medium=appdir-template&utm_campaign=create-next-app"
+          className="group rounded-lg border border-transparent px-5 py-4 transition-colors hover:border-gray-300 hover:bg-gray-100 hover:dark:border-neutral-700 hover:dark:bg-neutral-800/30"
+          target="_blank"
+          rel="noopener noreferrer"
+        >
+          <h2 className={`mb-3 text-2xl font-semibold`}>
+            Deploy{' '}
+            <span className="inline-block transition-transform group-hover:translate-x-1 motion-reduce:transform-none">
+              -&gt;
+            </span>
+          </h2>
+          <p className={`m-0 max-w-[30ch] text-sm opacity-50`}>
+            Instantly deploy your Next.js site to a shareable URL with Vercel.
+          </p>
+        </a>
+      </div>
+    </main>
+  )
+}

examples/nextjs-background-tasks/src/jobs/examples.ts

@@ -0,0 +1,23 @@
+import { eventTrigger } from "@trigger.dev/sdk";
+import { client } from "@/trigger";
+import task1 from "@/tasks/task1.background";
+
+// Your first job
+// This Job will be triggered by an event, log a joke to the console, and then wait 5 seconds before logging the punchline
+client.defineJob({
+  // This is the unique identifier for your Job, it must be unique across all Jobs in your project
+  id: "example-job",
+  name: "Background Task Usage",
+  version: "0.0.1",
+  // This is triggered by an event using eventTrigger. You can also trigger Jobs with webhooks, on schedules, and more: https://trigger.dev/docs/documentation/concepts/triggers/introduction
+  trigger: eventTrigger({
+    name: "example.event",
+  }),
+  run: async (payload, io, ctx) => {
+    const output = await task1.invoke("task-1", {
+      userName: "ericallam",
+    });
+
+    return { output };
+  },
+});

examples/nextjs-background-tasks/src/jobs/index.ts

@@ -0,0 +1,4 @@
+
+// import all your job files here
+
+export * from "./examples"

examples/nextjs-background-tasks/src/tasks/task1.background.ts

@@ -0,0 +1,28 @@
+// tasks.background.ts
+import { client } from "@/trigger";
+import { z } from "zod";
+
+export default client.defineBackgroundTask({
+  id: "task-1",
+  name: "Task 1",
+  version: "1.0.2",
+  schema: z.object({
+    userName: z.string(),
+  }),
+  cpu: 1,
+  memory: 256,
+  concurrency: 5,
+  secrets: {
+    OPENAI_API_KEY: process.env.OPENAI_API_KEY!,
+  },
+  run: async (payload) => {
+    // This code will run in the background, as it will be bundled and shipped to a trigger.dev background worker
+    await new Promise((resolve) => setTimeout(resolve, 100000));
+
+    return {
+      username: payload.userName,
+      foo: "bar",
+      message: `Task Response for user ${payload.userName}`,
+    };
+  },
+});

examples/nextjs-background-tasks/src/tasks/task2.background.ts

@@ -0,0 +1,14 @@
+import { client } from "@/trigger";
+import { z } from "zod";
+
+export default client.defineBackgroundTask({
+  id: "task-2",
+  name: "Task 2",
+  version: "1.0.0",
+  schema: z.object({
+    id: z.string(),
+  }),
+  run: async (payload) => {
+    return `Task Response for user ${payload.id}`;
+  },
+});

examples/nextjs-background-tasks/src/trigger.ts

@@ -0,0 +1,7 @@
+import { TriggerClient } from "@trigger.dev/sdk";
+
+export const client = new TriggerClient({
+  id: "background-tasks-z2kp",
+  apiKey: process.env.TRIGGER_API_KEY,
+  apiUrl: process.env.TRIGGER_API_URL,
+});

examples/nextjs-background-tasks/tailwind.config.ts

@@ -0,0 +1,20 @@
+import type { Config } from 'tailwindcss'
+
+const config: Config = {
+  content: [
+    './src/pages/**/*.{js,ts,jsx,tsx,mdx}',
+    './src/components/**/*.{js,ts,jsx,tsx,mdx}',
+    './src/app/**/*.{js,ts,jsx,tsx,mdx}',
+  ],
+  theme: {
+    extend: {
+      backgroundImage: {
+        'gradient-radial': 'radial-gradient(var(--tw-gradient-stops))',
+        'gradient-conic':
+          'conic-gradient(from 180deg at 50% 50%, var(--tw-gradient-stops))',
+      },
+    },
+  },
+  plugins: [],
+}
+export default config

examples/nextjs-background-tasks/tsconfig.json

@@ -0,0 +1,27 @@
+{
+  "compilerOptions": {
+    "target": "es5",
+    "lib": ["dom", "dom.iterable", "esnext"],
+    "allowJs": true,
+    "skipLibCheck": true,
+    "strict": true,
+    "noEmit": true,
+    "esModuleInterop": true,
+    "module": "esnext",
+    "moduleResolution": "bundler",
+    "resolveJsonModule": true,
+    "isolatedModules": true,
+    "jsx": "preserve",
+    "incremental": true,
+    "plugins": [
+      {
+        "name": "next"
+      }
+    ],
+    "paths": {
+      "@/*": ["./src/*"]
+    }
+  },
+  "include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"],
+  "exclude": ["node_modules"]
+}

package.json

@@ -39,6 +39,8 @@
     "changeset:release": "pnpm run build --filter \"@trigger.dev/*\" && changeset publish",
     "changeset:beta": "changeset pre enter beta",
     "changeset:normal": "changeset pre exit",
+    "changeset:version:snapshot": "changeset version --snapshot",
+    "changeset:release:snapshot": "pnpm run build --filter \"@trigger.dev/*\" && changeset publish --no-git-tag --snapshot",
     "clean:sourcemaps": "turbo run clean:sourcemaps",
     "storybook": "turbo run storybook"
   },
@@ -72,4 +74,4 @@
       "@changesets/assemble-release-plan@5.2.4": "patches/@changesets__assemble-release-plan@5.2.4.patch"
     }
   }
-}
+}

packages/cli/package.json

@@ -37,11 +37,15 @@
   "devDependencies": {
     "@gmrchk/cli-testing-library": "^0.1.2",
     "@trigger.dev/tsconfig": "workspace:*",
+    "@types/dockerode": "^3.3.19",
     "@types/gradient-string": "^1.1.2",
     "@types/inquirer": "^9.0.3",
     "@types/jest": "^29.5.3",
+    "@types/jsonlines": "^0.1.2",
     "@types/node": "16",
     "@types/node-fetch": "^2.6.2",
+    "@types/semver": "^7.3.13",
+    "@types/tar": "^6.1.4",
     "jest": "^29.6.2",
     "rimraf": "^3.0.2",
     "ts-jest": "^29.1.1",
@@ -58,15 +62,20 @@
     "test": "jest"
   },
   "dependencies": {
+    "@swc/core": "^1.3.26",
+    "@trigger.dev/core": "workspace:*",
     "@types/degit": "^2.8.3",
     "chalk": "^5.2.0",
     "chokidar": "^3.5.3",
     "commander": "^9.4.1",
     "degit": "^2.8.4",
+    "dockerode": "^3.3.5",
     "dotenv": "^16.3.1",
+    "esbuild": "^0.19.2",
     "execa": "^7.0.0",
     "gradient-string": "^2.0.2",
     "inquirer": "^9.1.4",
+    "jsonlines": "^0.1.1",
     "localtunnel": "^2.0.2",
     "nanoid": "^4.0.2",
     "ngrok": "5.0.0-beta.2",
@@ -76,7 +85,10 @@
     "ora": "^6.1.2",
     "path-to-regexp": "^6.2.1",
     "posthog-node": "^3.1.1",
+    "semver": "^7.5.0",
     "simple-git": "^3.19.0",
+    "spawn-please": "^2.0.2",
+    "tar": "^6.2.0",
     "terminal-link": "^3.0.0",
     "tsconfck": "^2.1.2",
     "url": "^0.11.1",
@@ -85,4 +97,4 @@
   "engines": {
     "node": ">=16.8.0"
   }
-}
+}

packages/cli/src/cli/index.ts

@@ -10,6 +10,7 @@ import { telemetryClient } from "../telemetry/telemetry";
 import { getVersion } from "../utils/getVersion";
 import { updateCommand } from "../commands/update";
 import { sendEventCommand } from "../commands/sendEvent";
+import { deployCommand } from "../commands/deploy";
 
 export const program = new Command();
 
@@ -126,6 +127,21 @@ program
     }
   });
 
+program
+  .command("deploy")
+  .description("Deploy background tasks to Trigger.dev")
+  .argument("[path]", "The path to the project", ".")
+  .option("-e, --env-file <name>", "The name of the env file to load", ".env.local")
+  .option("-t, --tag <tag>", "The tag to use if the @trigger.dev/* packages are linked")
+  .version(getVersion(), "-v, --version", "Display the version number")
+  .action(async (path, options) => {
+    try {
+      await deployCommand(path, options);
+    } catch (e) {
+      throw e;
+    }
+  });
+
 export const promptTriggerUrl = async (): Promise<string> => {
   const { instanceType } = await inquirer.prompt<{
     instanceType: "cloud" | "self-hosted";

packages/cli/src/commands/deploy.ts

@@ -0,0 +1,390 @@
+import { CallExpression, Expression, parse } from "@swc/core";
+import { Visitor } from "@swc/core/Visitor.js";
+import type { DeployBackgroundTaskRequestBody } from "@trigger.dev/core";
+import childProcess from "child_process";
+import { build } from "esbuild";
+import pathModule from "node:path";
+import util from "util";
+import { z } from "zod";
+import { DockerWrapper } from "../utils/docker";
+import { downloadArchiveToTempDir } from "../utils/extractArchiveToTempDir";
+import { getTriggerApiDetails } from "../utils/getTriggerApiDetails";
+import { logger } from "../utils/logger";
+import { listPackageDependencies } from "../utils/packageManagers";
+import { resolvePath } from "../utils/parseNameAndPath";
+import { TriggerApi } from "../utils/triggerApi";
+
+const asyncExecFile = util.promisify(childProcess.execFile);
+
+export const DeployCommandOptionsSchema = z.object({
+  envFile: z.string(),
+  tag: z.string().optional(),
+});
+
+export type DevCommandOptions = z.infer<typeof DeployCommandOptionsSchema>;
+
+type DeployedArtifact = {
+  id: string;
+  hash: string;
+  image: string;
+  tag: string;
+  task: DeployBackgroundTaskRequestBody;
+};
+
+export async function deployCommand(path: string, anyOptions: any) {
+  const result = DeployCommandOptionsSchema.safeParse(anyOptions);
+
+  if (!result.success) {
+    logger.error(result.error.message);
+    return;
+  }
+  const options = result.data;
+
+  const resolvedPath = resolvePath(path);
+
+  const apiDetails = await getTriggerApiDetails(resolvedPath, options.envFile);
+
+  if (!apiDetails) {
+    logger.error("Could not find Trigger.dev API key");
+    return;
+  }
+
+  logger.info(`Deploying ${resolvedPath}...`);
+
+  // Find all files with .background.ts extension in the given path
+  const { stdout } = await asyncExecFile("find", [resolvedPath, "-name", "*.background.ts"], {
+    encoding: "utf-8",
+    cwd: resolvedPath,
+  });
+
+  const files = stdout
+    .split("\n")
+    .map((line) => line.trim())
+    .filter((line) => line.length > 0);
+
+  const entryPoints = files.map((path) => pathModule.relative(resolvedPath, path));
+
+  logger.info(`Found ${files.length} background tasks`);
+
+  // target is the process.version (e.g. v18.12.1) but we need to pass it as node18.12.1
+  const target = `node${process.version.replace("v", "")}`;
+
+  // Each file is an entry point and should be built into a separate bundle (not to be output to a file but instead to be sent to the server)
+  const bundle = await build({
+    entryPoints: entryPoints,
+    bundle: true,
+    format: "cjs",
+    platform: "node",
+    target,
+    write: false,
+    minify: false,
+    sourcemap: "external",
+    packages: "external",
+    metafile: true,
+    outdir: "dist",
+  });
+
+  logger.info(`Built bundle, extracting task IDs`);
+
+  const tasks: DeployBackgroundTaskRequestBody[] = [];
+
+  for (const entryPoint of entryPoints) {
+    const task = await gatherBackgroundTaskDeployment(resolvedPath, entryPoint, bundle, options);
+
+    if (task) {
+      tasks.push(task);
+    }
+  }
+
+  const apiClient = new TriggerApi(apiDetails.apiKey, apiDetails.apiUrl);
+
+  const artifacts: Array<DeployedArtifact> = [];
+
+  for (const task of tasks) {
+    const response = await apiClient.deployBackgroundTask(task);
+
+    if (!response.ok) {
+      logger.error(`Failed to deploy ${task.id}@${task.version}: ${response.error}`);
+    } else {
+      logger.info(`Deployed ${task.id}@${task.version}#${response.data.hash}`);
+
+      artifacts.push({
+        id: response.data.id,
+        hash: response.data.hash,
+        image: response.data.image,
+        tag: response.data.tag,
+        task,
+      });
+    }
+  }
+
+  // For each artifact, we need to build a docker image and push it to the registry
+  // The registry host is the TRIGGER_API_URL
+  for (const artifact of artifacts) {
+    const results = await buildAndPushDockerImage(apiDetails.apiUrl, apiDetails.apiKey, artifact);
+
+    if (results) {
+      await apiClient.createBackgroundTaskArtifactImage(artifact.id, {
+        digest: results.digest,
+        name: artifact.image,
+        tag: artifact.tag,
+        size: results.size,
+      });
+    }
+  }
+}
+
+async function buildAndPushDockerImage(
+  registryUrl: string,
+  apiKey: string,
+  artifact: DeployedArtifact
+) {
+  const registry = new URL(registryUrl).host;
+  const imageName = `${registry}/${artifact.image}:${artifact.tag}`;
+
+  const docker = new DockerWrapper({ docker: { socketPath: "/var/run/docker.sock" } });
+
+  logger.info(
+    `Building docker image ${imageName} for task ${artifact.task.id}@${artifact.task.version}...`
+  );
+
+  const artifactPath = await downloadArtifact(registryUrl, artifact);
+
+  const buildResults = await docker.buildImage(artifactPath, {
+    t: imageName,
+    dockerfile: "ctx/Dockerfile",
+    platform: "linux/amd64",
+    nocache: true,
+  });
+
+  if (!buildResults) {
+    logger.error(`Failed to build docker image ${imageName}`);
+    return;
+  }
+
+  logger.info(`Built docker image ${imageName}: ${JSON.stringify(buildResults)}`);
+
+  const pushResults = await docker.pushImage(imageName, {
+    authconfig: {
+      username: "x",
+      password: apiKey,
+      serveraddress: "https://index.docker.io/v1",
+    },
+  });
+
+  logger.info(`Pushed docker image ${imageName}: ${JSON.stringify(pushResults)}`);
+
+  return pushResults;
+}
+
+async function downloadArtifact(registryUrl: string, artifact: DeployedArtifact): Promise<string> {
+  const artifactArchiveUrl = `${registryUrl}/api/v1/background/artifacts/${artifact.id}.tgz`;
+
+  return await downloadArchiveToTempDir(artifactArchiveUrl, artifact.id);
+}
+
+async function gatherBackgroundTaskDeployment(
+  projectPath: string,
+  file: string,
+  bundle: TasksBundle,
+  options: DevCommandOptions
+): Promise<DeployBackgroundTaskRequestBody | undefined> {
+  const outputKey = Object.keys(bundle.metafile.outputs).find(
+    (o) => bundle.metafile.outputs[o]?.entryPoint === file
+  );
+
+  if (!outputKey) {
+    return;
+  }
+
+  const outputMetadata = bundle.metafile.outputs[outputKey];
+
+  if (!outputMetadata) {
+    return;
+  }
+
+  const outputPath = pathModule.join(projectPath, outputKey);
+
+  const outputFile = bundle.outputFiles.find((o) => o.path === outputPath);
+
+  if (!outputFile) {
+    return;
+  }
+
+  const outputSourcemapPath = pathModule.join(projectPath, `${outputKey}.map`);
+
+  const outputSourcemapFile = bundle.outputFiles.find((o) => o.path === outputSourcemapPath);
+
+  if (!outputSourcemapFile) {
+    return;
+  }
+
+  const taskInfo = await findTask(outputFile.text);
+
+  if (!taskInfo) {
+    return;
+  }
+
+  const dependencies: Record<string, string> = {};
+  const imports = new Set<string>();
+
+  for (const importMeta of outputMetadata.imports) {
+    if (
+      importMeta.kind === "require-call" &&
+      importMeta.external &&
+      !nodeBuiltIn(importMeta.path)
+    ) {
+      imports.add(importMeta.path);
+    }
+  }
+
+  const packageDependencies = await listPackageDependencies(projectPath, options.tag);
+
+  for (const importName of imports) {
+    const version = packageDependencies[importName];
+
+    if (version) {
+      dependencies[importName] = version;
+    }
+  }
+
+  logger.info(`Found ${Object.keys(dependencies).length} dependencies`);
+
+  return {
+    id: taskInfo.id,
+    version: taskInfo.version,
+    bundle: outputFile.text,
+    fileName: pathModule.basename(outputFile.path),
+    sourcemap: outputSourcemapFile.text,
+    dependencies,
+    nodeVersion: process.version,
+  };
+}
+
+function nodeBuiltIn(importName: string): boolean {
+  if (importName.startsWith("node:")) {
+    return true;
+  }
+
+  // Now we need to check for node built-in modules that don't use the node: prefix
+  // See https://nodejs.org/api/modules.html#modules_core_modules
+  const builtInModules = [
+    "assert",
+    "async_hooks",
+    "buffer",
+    "child_process",
+    "cluster",
+    "crypto",
+    "dgram",
+    "dns",
+    "domain",
+    "events",
+    "fs",
+    "http",
+    "http2",
+    "https",
+    "inspector",
+    "module",
+    "net",
+    "os",
+    "path",
+    "perf_hooks",
+    "process",
+    "punycode",
+    "querystring",
+    "readline",
+    "repl",
+    "stream",
+    "string_decoder",
+    "sys",
+    "timers",
+    "tls",
+    "trace_events",
+    "tty",
+    "url",
+    "util",
+    "v8",
+    "vm",
+    "zlib",
+  ];
+
+  return builtInModules.includes(importName);
+}
+
+class BackgroundTaskVisitor extends Visitor {
+  public id: string | null = null;
+  public version: string | null = null;
+
+  override visitCallExpression(n: CallExpression): Expression {
+    if (
+      n.type === "CallExpression" &&
+      n.callee.type === "MemberExpression" &&
+      n.callee.property.type === "Identifier" &&
+      n.callee.property.value === "defineBackgroundTask"
+    ) {
+      const firstArg = n.arguments[0];
+
+      if (firstArg && firstArg.expression.type === "ObjectExpression") {
+        const properties = firstArg.expression.properties;
+
+        properties.forEach((property) => {
+          if (property.type === "KeyValueProperty") {
+            const key = property.key;
+
+            if (key.type === "Identifier" && key.value === "id") {
+              const value = property.value;
+
+              if (value.type === "StringLiteral") {
+                this.id = value.value;
+              }
+            } else if (key.type === "Identifier" && key.value === "version") {
+              const value = property.value;
+
+              if (value.type === "StringLiteral") {
+                this.version = value.value;
+              }
+            }
+          }
+        });
+      }
+    }
+
+    return n;
+  }
+}
+
+async function findTask(code: string): Promise<{ id: string; version: string } | null> {
+  const ast = await parse(code);
+
+  const visitor = new BackgroundTaskVisitor();
+
+  visitor.visitProgram(ast);
+
+  if (typeof visitor.id === "string" && typeof visitor.version === "string") {
+    return {
+      id: visitor.id,
+      version: visitor.version,
+    };
+  }
+
+  return null;
+}
+
+type TasksBundle = Awaited<ReturnType<typeof buildTasks>>;
+
+async function buildTasks(files: Array<string>) {
+  return await build({
+    entryPoints: files,
+    bundle: true,
+    format: "cjs",
+    platform: "node",
+    target: "node18.12.1",
+    write: false,
+    minify: false,
+    sourcemap: "external",
+    packages: "external",
+    external: ["@trigger.dev/*"],
+    metafile: true,
+    outdir: "dist",
+  });
+}

packages/cli/src/utils/docker.ts

@@ -0,0 +1,130 @@
+import Docker, {
+  DockerOptions,
+  ImageBuildContext,
+  ImageBuildOptions,
+  ImagePushOptions,
+} from "dockerode";
+import { z } from "zod";
+
+const BuildAuxResultSchema = z.object({
+  aux: z.object({
+    ID: z.string(),
+  }),
+});
+
+const BuildStreamResultSchema = z.object({
+  stream: z.string(),
+});
+
+const PushAuxResultSchema = z.object({
+  aux: z.object({
+    Tag: z.string(),
+    Digest: z.string(),
+    Size: z.number(),
+  }),
+});
+
+export type DockerWrapperOptions = {
+  docker?: DockerOptions;
+};
+
+export class DockerWrapper {
+  readonly #docker: Docker;
+
+  constructor(options?: DockerWrapperOptions) {
+    this.#docker = new Docker(options?.docker);
+  }
+
+  // https://jsonhero.io/j/PzXccoO3lCSg
+  async buildImage(
+    context: ImageBuildContext | string,
+    options: ImageBuildOptions
+  ): Promise<{ digest: string; image: string } | undefined> {
+    const stream = await this.#docker.buildImage(context, options);
+
+    const buildResults = await new Promise<any[]>((resolve, reject) => {
+      this.#docker.modem.followProgress(stream, (err, res) => (err ? reject(err) : resolve(res)));
+    });
+
+    return this.#parseBuildResults(buildResults);
+  }
+
+  // https://jsonhero.io/j/BmJcRaX5rx1s
+  async pushImage(
+    imageName: string,
+    options?: ImagePushOptions
+  ): Promise<{ tag: string; digest: string; size: number } | undefined> {
+    const image = this.#docker.getImage(imageName);
+
+    const stream = await image.push(options);
+
+    const pushResults = await new Promise((resolve, reject) => {
+      this.#docker.modem.followProgress(stream, (err, res) => (err ? reject(err) : resolve(res)));
+    });
+
+    return this.#parsePushResults(pushResults);
+  }
+
+  #parsePushResults(
+    pushResults: unknown
+  ): { tag: string; digest: string; size: number } | undefined {
+    const resultsArray = z.array(z.any()).safeParse(pushResults);
+
+    if (!resultsArray.success) {
+      return;
+    }
+
+    const auxResult = findWithSchema(PushAuxResultSchema, resultsArray.data);
+
+    if (!auxResult) {
+      return undefined;
+    }
+
+    return {
+      tag: auxResult.aux.Tag,
+      digest: auxResult.aux.Digest,
+      size: auxResult.aux.Size,
+    };
+  }
+
+  #parseBuildResults(buildResults: any[]): { digest: string; image: string } | undefined {
+    const auxResult = findWithSchema(BuildAuxResultSchema, buildResults);
+
+    if (!auxResult) {
+      return undefined;
+    }
+
+    const lastResult = buildResults[buildResults.length - 1];
+    const parsedLastResult = BuildStreamResultSchema.safeParse(lastResult);
+
+    if (!parsedLastResult.success) {
+      return undefined;
+    }
+
+    const image = this.#parseImageFromStream(parsedLastResult.data.stream);
+
+    if (!image) {
+      return undefined;
+    }
+
+    return {
+      digest: auxResult.aux.ID,
+      image,
+    };
+  }
+
+  // Successfully tagged eric-webapp.trigger.dev/clm7ndmb60009dyeoe0edbf32-task-1:1.0.1
+  #parseImageFromStream(stream: string): string | undefined {
+    const match = stream.match(/Successfully tagged ([^:]+):([^ ]+)/);
+
+    if (!match) {
+      return undefined;
+    }
+
+    return match[1]?.trim();
+  }
+}
+
+function findWithSchema<T>(schema: z.Schema<T>, items: any[]): T | undefined {
+  return items.find((item) => (schema.safeParse(item).success ? true : false));
+}

packages/cli/src/utils/extractArchiveToTempDir.ts

@@ -0,0 +1,54 @@
+import { createWriteStream } from "node:fs";
+import { createTempDir } from "./fileSystem";
+import { pipeline } from "node:stream";
+import { promisify } from "node:util";
+import tar from "tar";
+
+const streamPipeline = promisify(pipeline);
+
+// A node.js function to download and extract a tarball to a temporary directory, and return the path to the directory.
+export async function extractArchiveToTempDir(url: string, prefix: string): Promise<string> {
+  const archiveTempDir = await createTempDir(`${prefix}-archive`);
+  const tempDir = await createTempDir(prefix);
+  const tempFilePath = `${archiveTempDir}/file.tgz`;
+
+  // Download the .tgz file
+  const response = await fetch(url);
+  if (!response.ok) {
+    throw new Error(`Failed to download .tgz file (${response.status} ${response.statusText})`);
+  }
+
+  if (!response.body) {
+    throw new Error("Failed to download .tgz file (no body)");
+  }
+
+  const fileStream = createWriteStream(tempFilePath);
+  // @ts-ignore
+  await streamPipeline(response.body, fileStream);
+
+  // Extract the contents to the temporary directory
+  await tar.x({ C: tempDir, file: tempFilePath });
+
+  return tempDir;
+}
+
+export async function downloadArchiveToTempDir(url: string, prefix: string): Promise<string> {
+  const archiveTempDir = await createTempDir(`${prefix}-archive`);
+  const tempFilePath = `${archiveTempDir}/file.tgz`;
+
+  // Download the .tgz file
+  const response = await fetch(url);
+  if (!response.ok) {
+    throw new Error(`Failed to download .tgz file (${response.status} ${response.statusText})`);
+  }
+
+  if (!response.body) {
+    throw new Error("Failed to download .tgz file (no body)");
+  }
+
+  const fileStream = createWriteStream(tempFilePath);
+  // @ts-ignore
+  await streamPipeline(response.body, fileStream);
+
+  return tempFilePath;
+}

packages/cli/src/utils/fileSystem.ts

@@ -1,6 +1,7 @@
-import fsModule, { writeFile } from "fs/promises";
-import fsSync from "fs";
-import pathModule from "path";
+import fsModule, { mkdtemp, writeFile } from "node:fs/promises";
+import fsSync from "node:fs";
+import pathModule from "node:path";
+import os from "node:os";
 
 // Creates a file at the given path, if the directory doesn't exist it will be created
 export async function createFile(path: string, contents: string): Promise<string> {
@@ -43,3 +44,33 @@ export function readJSONFileSync(path: string) {
 
   return JSON.parse(fileContents);
 }
+
+export async function createTempDir(prefix: string) {
+  return await mkdtemp(pathModule.join(os.tmpdir(), `${prefix}-`));
+}
+
+// Recursively list all the files in the directory, and returns an array a relative paths
+export async function listFilesInDir(dir: string): Promise<string[]> {
+  async function listFilesInDirRecursive(dir: string): Promise<string[]> {
+    const files = await fsModule.readdir(dir);
+
+    const filePaths = await Promise.all(
+      files.map(async (file) => {
+        const filePath = pathModule.join(dir, file);
+        const stat = await fsModule.stat(filePath);
+
+        if (stat.isDirectory()) {
+          return await listFilesInDirRecursive(filePath);
+        }
+
+        return filePath;
+      })
+    );
+
+    return filePaths.flat();
+  }
+
+  const allFiles = await listFilesInDirRecursive(dir);
+
+  return allFiles.map((file) => pathModule.relative(dir, file));
+}

packages/cli/src/utils/keyValueBy.ts

@@ -0,0 +1,39 @@
+type Index<T = any> = { [key: string]: T };
+type KeyValueGenerator<K, V, R> = (key: K, value: V, accum: Index<R>) => Index<R> | null;
+type ArrayKeyValueGenerator<T, R> = KeyValueGenerator<T, number, R>;
+type ObjectKeyValueGenerator<T, R> = KeyValueGenerator<string, T, R>;
+
+export function keyValueBy<T>(arr: T[]): Index<true>;
+export function keyValueBy<T, R>(
+  arr: T[],
+  keyValue: KeyValueGenerator<T, number, R>,
+  initialValue?: Index<R>
+): Index<R>;
+export function keyValueBy<T, R>(
+  obj: Index<T>,
+  keyValue: KeyValueGenerator<string, T, R>,
+  initialValue?: Index<R>
+): Index<R>;
+
+/** Generates an object from an array or object. Simpler than reduce or _.transform. The KeyValueGenerator passes (key, value) if the input is an object, and (value, i) if it is an array. The return object from each iteration is merged into the accumulated object. Return null to skip an item. */
+export function keyValueBy<T, R = true>(
+  input: T[] | Index<T>,
+  // if no keyValue is given, sets all values to true
+  keyValue?: ArrayKeyValueGenerator<T, R> | ObjectKeyValueGenerator<T, R>,
+  accum: Index<R> = {}
+): Index<R> {
+  const isArray = Array.isArray(input);
+  keyValue =
+    keyValue || ((key: T): Index<R> => ({ [key as unknown as string]: true as unknown as R }));
+  // considerably faster than Array.prototype.reduce
+  Object.entries(input || {}).forEach(([key, value], i) => {
+    const o = isArray
+      ? (keyValue as ArrayKeyValueGenerator<T, R>)(value, i, accum)
+      : (keyValue as ObjectKeyValueGenerator<T, R>)(key, value, accum);
+    Object.entries(o || {}).forEach((entry) => {
+      accum[entry[0]] = entry[1];
+    });
+  });
+
+  return accum;
+}

packages/cli/src/utils/packageManagers/index.ts

@@ -0,0 +1,276 @@
+import { getUserPackageManager } from "../getUserPkgManager";
+import spawn from "spawn-please";
+import { keyValueBy } from "../keyValueBy";
+import nodeSemver from "semver";
+import jsonlines from "jsonlines";
+
+export async function listPackageDependencies(
+  path: string,
+  tag: string | undefined = undefined
+): Promise<Record<string, string | undefined>> {
+  const packageManager = await getPackageManagerCommands(path);
+
+  const list = await packageManager.list({ cwd: path });
+
+  return Object.keys(list).reduce(
+    (acc, dependency) => {
+      const version = list[dependency];
+
+      if (!version) {
+        return acc;
+      }
+
+      if (dependency.startsWith("@trigger.dev/") && version.startsWith("link:")) {
+        acc[dependency] = tag ?? "latest";
+      } else {
+        acc[dependency] = version;
+      }
+
+      return acc;
+    },
+    {} as Record<string, string | undefined>
+  );
+}
+
+type PnpmList = {
+  path: string;
+  private: boolean;
+  dependencies: Record<
+    string,
+    {
+      from: string;
+      version: string;
+      resolved: string;
+    }
+  >;
+}[];
+
+async function getPackageManagerCommands(path: string): Promise<PackageManagerCommands> {
+  const packageManager = await getUserPackageManager(path);
+
+  switch (packageManager) {
+    case "npm":
+      return new NPMCommands();
+    case "pnpm":
+      return new PNPMCommands();
+    case "yarn":
+      return new YarnCommands();
+  }
+}
+
+type ListOptions = {
+  cwd?: string;
+  prefix?: string;
+  global?: boolean;
+};
+
+interface PackageManagerCommands {
+  list(options: ListOptions): Promise<Record<string, string | undefined>>;
+}
+
+class PNPMCommands implements PackageManagerCommands {
+  async list(options: ListOptions): Promise<Record<string, string | undefined>> {
+    const cmd = process.platform === "win32" ? "pnpm.cmd" : "pnpm";
+    const listOutput = await spawn(cmd, ["ls", "--depth", "1", "--json", "--long"], options);
+    const result = JSON.parse(listOutput) as PnpmList;
+
+    const list = keyValueBy(result[0]?.dependencies ?? {}, (name, { version }) => ({
+      [name]: version,
+    }));
+
+    return list;
+  }
+}
+
+type NpmOptions = {
+  location?: string;
+  prefix?: string;
+  registry?: string;
+};
+
+class NPMCommands implements PackageManagerCommands {
+  async list(options: ListOptions): Promise<Record<string, string | undefined>> {
+    const result = await this.#spawn(
+      ["ls", "--depth=0"],
+      {
+        ...(options.prefix ? { prefix: options.prefix } : null),
+      },
+      {
+        ...(options.cwd ? { cwd: options.cwd } : null),
+        rejectOnError: false,
+      }
+    );
+
+    const dependencies = this.#parseJson<{
+      dependencies: Record<string, { version?: string; required?: { version: string } }>;
+    }>(result, {
+      command: `npm${process.platform === "win32" ? ".cmd" : ""} ls --json${
+        options.global ? " --location=global" : ""
+      }${options.prefix ? " --prefix " + options.prefix : ""}`,
+    }).dependencies;
+
+    return keyValueBy(dependencies, (name, info) => ({
+      // unmet peer dependencies have a different structure
+      [name]: info.version || info.required?.version,
+    }));
+  }
+
+  /**
+   * Spawns npm with --json. Handles different commands for Window and Linux/OSX, and automatically converts --location=global to --global on node < 8.11.0.
+   *
+   * @param args
+   * @param [npmOptions={}]
+   * @param [spawnOptions={}]
+   * @returns
+   */
+  async #spawn(
+    args: string | string[],
+    npmOptions: NpmOptions = {},
+    spawnOptions: Record<string, any> = {}
+  ): Promise<any> {
+    const cmd = process.platform === "win32" ? "npm.cmd" : "npm";
+    args = Array.isArray(args) ? args : [args];
+
+    const fullArgs = args.concat(
+      npmOptions.location
+        ? (await this.#isGlobalDeprecated())
+          ? `--location=${npmOptions.location}`
+          : npmOptions.location === "global"
+          ? "--global"
+          : ""
+        : [],
+      npmOptions.prefix ? `--prefix=${npmOptions.prefix}` : [],
+      "--json"
+    );
+
+    return spawn(cmd, fullArgs, spawnOptions);
+  }
+
+  async #isGlobalDeprecated() {
+    const cmd = process.platform === "win32" ? "npm.cmd" : "npm";
+    const output = await spawn(cmd, ["--version"]);
+    const npmVersion = output.trim();
+    // --global was deprecated in npm v8.11.0.
+    return nodeSemver.valid(npmVersion) && nodeSemver.gte(npmVersion, "8.11.0");
+  }
+
+  /**
+   * Parse JSON and throw an informative error on failure.
+   *
+   * @param result Data to be parsed
+   * @param data
+   * @returns
+   */
+  #parseJson<R>(result: string, data: { command?: string; packageName?: string }): R {
+    let json;
+    try {
+      json = JSON.parse(result);
+    } catch (err) {
+      throw new Error(
+        `Expected JSON from "${data.command}".${
+          data.packageName ? ` There could be problems with the ${data.packageName} package.` : ""
+        } ${result ? "Instead received: " + result : "Received empty response."}`
+      );
+    }
+    return json as R;
+  }
+}
+
+interface YarnParsedDep {
+  version: string;
+  from: string;
+  required?: {
+    version: string;
+  };
+}
+
+class YarnCommands implements PackageManagerCommands {
+  async list(options: ListOptions): Promise<Record<string, string | undefined>> {
+    const jsonLines: string = await this.#spawn("list", options as Record<string, string>, {
+      ...(options.cwd ? { cwd: options.cwd } : {}),
+    });
+
+    const json: { dependencies: Record<string, YarnParsedDep> } = await this.#parseJsonLines(
+      jsonLines
+    );
+
+    const keyValues: Record<string, string | undefined> = keyValueBy<
+      YarnParsedDep,
+      string | undefined
+    >(json.dependencies, (name, info): { [key: string]: string | undefined } => ({
+      // unmet peer dependencies have a different structure
+      [name]: info.version || info.required?.version,
+    }));
+
+    return keyValues;
+  }
+
+  /**
+   * Spawn yarn requires a different command on Windows.
+   *
+   * @param args
+   * @param [yarnOptions={}]
+   * @param [spawnOptions={}]
+   * @returns
+   */
+  async #spawn(
+    args: string | string[],
+    yarnOptions: NpmOptions = {},
+    spawnOptions?: any
+  ): Promise<string> {
+    const cmd = process.platform === "win32" ? "yarn.cmd" : "yarn";
+
+    const fullArgs = [
+      ...(yarnOptions.location === "global" ? "global" : []),
+      ...(Array.isArray(args) ? args : [args]),
+      "--depth=0",
+      ...(yarnOptions.prefix ? `--prefix=${yarnOptions.prefix}` : []),
+      "--json",
+      "--no-progress",
+    ];
+
+    return spawn(cmd, fullArgs, spawnOptions);
+  }
+
+  /**
+   * Parse JSON lines and throw an informative error on failure.
+   *
+   * Note: although this is similar to the NPM parseJson() function we always return the
+   * same concrete-type here, for now.
+   *
+   * @param result    Output from `yarn list --json` to be parsed
+   */
+  #parseJsonLines(result: string): Promise<{ dependencies: Record<string, YarnParsedDep> }> {
+    return new Promise((resolve, reject) => {
+      const dependencies: Record<string, YarnParsedDep> = {};
+
+      const parser = jsonlines.parse();
+
+      parser.on("data", (d) => {
+        // only parse info data
+        // ignore error info, e.g. "Visit https://yarnpkg.com/en/docs/cli/list for documentation about this command."
+        if (d.type === "info" && !d.data.match(/^Visit/)) {
+          // parse package name and version number from info data, e.g. "nodemon@2.0.4" has binaries
+          const [, pkgName, pkgVersion] = d.data.match(/"(@?.*)@(.*)"/) || [];
+
+          dependencies[pkgName] = {
+            version: pkgVersion,
+            from: pkgName,
+          };
+        } else if (d.type === "error") {
+          reject(new Error(d.data));
+        }
+      });
+
+      parser.on("end", () => {
+        resolve({ dependencies });
+      });
+
+      parser.on("error", reject);
+
+      parser.write(result);
+
+      parser.end();
+    });
+  }
+}

packages/cli/src/utils/triggerApi.ts

@@ -1,3 +1,9 @@
+import {
+  CreateBackgroundTaskImageRequestBody,
+  CreateBackgroundTaskImageResponseBody,
+  DeployBackgroundTaskRequestBody,
+  DeployBackgroundTaskResponseBody,
+} from "@trigger.dev/core";
 import fetch from "node-fetch";
 import { z } from "zod";
 
@@ -18,6 +24,17 @@ export type EndpointData = {
   indexingHookIdentifier: string;
 };
 
+export type ApiResponse<TData> =
+  | {
+      ok: true;
+      data: TData;
+    }
+  | {
+      ok: false;
+      error: string;
+      retryable: boolean;
+    };
+
 export type EndpointResponse =
   | {
       ok: true;
@@ -55,7 +72,7 @@ export type WhoamiResponse = z.infer<typeof WhoamiResponseSchema>;
 export class TriggerApi {
   constructor(
     private apiKey: string,
-    private baseUrl: string
+    private baseUrl: string = "https://api.trigger.dev"
   ) {}
 
   async whoami(apiKey: string): Promise<WhoamiResponse | undefined> {
@@ -154,6 +171,123 @@ export class TriggerApi {
       data: data as any as EndpointData,
     };
   }
+
+  async deployBackgroundTask(
+    options: DeployBackgroundTaskRequestBody
+  ): Promise<ApiResponse<DeployBackgroundTaskResponseBody>> {
+    const response = await fetch(`${this.baseUrl}/api/v1/background/tasks`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${this.apiKey}`,
+      },
+      body: JSON.stringify(options),
+    });
+
+    if (!response.ok) {
+      const rawBody = await response.text();
+
+      if (typeof rawBody === "string") {
+        const rawJson = safeJsonParse(rawBody);
+
+        if (!rawJson) {
+          return {
+            ok: false,
+            error: "An unknown issue occurred deploying to Trigger.dev",
+            retryable: true,
+          };
+        }
+
+        const parsedJson = z.object({ error: z.string() }).safeParse(rawJson);
+
+        if (!parsedJson.success) {
+          return {
+            ok: false,
+            error: "An unknown issue occurred deploying to Trigger.dev",
+            retryable: true,
+          };
+        }
+
+        return {
+          ok: false,
+          error: parsedJson.data.error,
+          retryable: RETRYABLE_PATTERN.test(parsedJson.data.error),
+        };
+      } else {
+        return {
+          ok: false,
+          error: "An unknown issue occurred deploying to Trigger.dev",
+          retryable: true,
+        };
+      }
+    }
+
+    const data = await response.json();
+
+    return {
+      ok: true,
+      data: data as any as DeployBackgroundTaskResponseBody,
+    };
+  }
+
+  async createBackgroundTaskArtifactImage(
+    id: string,
+    options: CreateBackgroundTaskImageRequestBody
+  ): Promise<ApiResponse<CreateBackgroundTaskImageResponseBody>> {
+    const response = await fetch(`${this.baseUrl}/api/v1/background/artifacts/${id}/images`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${this.apiKey}`,
+      },
+      body: JSON.stringify(options),
+    });
+
+    if (!response.ok) {
+      const rawBody = await response.text();
+
+      if (typeof rawBody === "string") {
+        const rawJson = safeJsonParse(rawBody);
+
+        if (!rawJson) {
+          return {
+            ok: false,
+            error: "An unknown issue occurred creating image on Trigger.dev",
+            retryable: true,
+          };
+        }
+
+        const parsedJson = z.object({ error: z.string() }).safeParse(rawJson);
+
+        if (!parsedJson.success) {
+          return {
+            ok: false,
+            error: "An unknown issue occurred creating image on Trigger.dev",
+            retryable: true,
+          };
+        }
+
+        return {
+          ok: false,
+          error: parsedJson.data.error,
+          retryable: RETRYABLE_PATTERN.test(parsedJson.data.error),
+        };
+      } else {
+        return {
+          ok: false,
+          error: "An unknown issue occurred creating image on Trigger.dev",
+          retryable: true,
+        };
+      }
+    }
+
+    const data = await response.json();
+
+    return {
+      ok: true,
+      data: data as any as CreateBackgroundTaskImageResponseBody,
+    };
+  }
 }
 
 function safeJsonParse(raw: string | null | undefined): unknown {

packages/cli/tsconfig.json

@@ -47,7 +47,11 @@
     "useDefineForClassFields": true,
     "experimentalDecorators": true,
     "emitDecoratorMetadata": true,
-    "types": ["jest"]
+    "types": ["jest"],
+    "paths": {
+      "@trigger.dev/core/*": ["../core/src/*"],
+      "@trigger.dev/core": ["../core/src/index"]
+    }
   },
   "exclude": ["node_modules"]
 }

packages/core/src/logger.ts

@@ -30,6 +30,15 @@ export class Logger {
     this.#jsonReplacer = jsonReplacer;
   }
 
+  child(name: string): Logger {
+    return new Logger(
+      `${this.#name}:${name}`,
+      logLevels[this.#level],
+      this.#filteredKeys,
+      this.#jsonReplacer
+    );
+  }
+
   // Return a new Logger instance with the same name and a new log level
   // but filter out the keys from the log messages (at any level)
   filter(...keys: string[]) {

packages/core/src/schemas/api.ts

@@ -281,11 +281,27 @@ export const DynamicTriggerEndpointMetadataSchema = z.object({
 
 export type DynamicTriggerEndpointMetadata = z.infer<typeof DynamicTriggerEndpointMetadataSchema>;
 
+export const BackgroundTaskMetadataSchema = z.object({
+  id: z.string(),
+  name: z.string(),
+  version: z.string(),
+  enabled: z.boolean(),
+  cpu: z.number(),
+  memory: z.number(),
+  diskSizeInGB: z.number().optional(),
+  concurrency: z.number(),
+  region: z.string().optional(),
+  secrets: z.record(z.string()).optional(),
+});
+
+export type BackgroundTaskMetadata = z.infer<typeof BackgroundTaskMetadataSchema>;
+
 export const IndexEndpointResponseSchema = z.object({
   jobs: z.array(JobMetadataSchema),
   sources: z.array(SourceMetadataSchema),
   dynamicTriggers: z.array(DynamicTriggerEndpointMetadataSchema),
   dynamicSchedules: z.array(RegisterDynamicSchedulePayloadSchema),
+  backgroundTasks: z.array(BackgroundTaskMetadataSchema).optional(),
 });
 
 export type IndexEndpointResponse = z.infer<typeof IndexEndpointResponseSchema>;
@@ -593,7 +609,7 @@ export const RunTaskOptionsSchema = z.object({
   /** Allows you to link the Integration connection in the logs. This is handled automatically in integrations.  */
   connectionKey: z.string().optional(),
   /** An operation you want to perform on the Trigger.dev platform, current only "fetch" is supported. If you wish to `fetch` use [`io.backgroundFetch()`](https://trigger.dev/docs/sdk/io/backgroundfetch) instead. */
-  operation: z.enum(["fetch"]).optional(),
+  operation: z.enum(["fetch", "backgroundTask"]).optional(),
   /** A No Operation means that the code won't be executed. This is used internally to implement features like [io.wait()](https://trigger.dev/docs/sdk/io/wait).  */
   noop: z.boolean().default(false),
   redact: RedactSchema.optional(),
@@ -729,3 +745,64 @@ export const CreateExternalConnectionBodySchema = z.object({
 });
 
 export type CreateExternalConnectionBody = z.infer<typeof CreateExternalConnectionBodySchema>;
+
+export const SourceMapDefinitionSchema = z.object({
+  version: z.number(),
+  sources: z.array(z.string()),
+  mappings: z.string(),
+  sourcesContent: z.array(z.string()),
+  names: z.array(z.string()),
+});
+
+export const DeployBackgroundTaskRequestBodySchema = z.object({
+  id: z.string(),
+  version: z.string(),
+  fileName: z.string(),
+  bundle: z.string(),
+  sourcemap: z.string(),
+  dependencies: z.record(z.string()),
+  nodeVersion: z.string(),
+});
+
+export type DeployBackgroundTaskRequestBody = z.infer<typeof DeployBackgroundTaskRequestBodySchema>;
+
+export const DeployBackgroundTaskResponseBodySchema = z.object({
+  id: z.string(),
+  hash: z.string(),
+  image: z.string(),
+  tag: z.string(),
+  createdAt: z.coerce.date(),
+  updatedAt: z.coerce.date(),
+});
+
+export type DeployBackgroundTaskResponseBody = z.infer<
+  typeof DeployBackgroundTaskResponseBodySchema
+>;
+
+export const CreateBackgroundTaskImageRequestBodySchema = z.object({
+  name: z.string(),
+  tag: z.string(),
+  digest: z.string(),
+  size: z.number(),
+});
+
+export type CreateBackgroundTaskImageRequestBody = z.infer<
+  typeof CreateBackgroundTaskImageRequestBodySchema
+>;
+
+export const CreateBackgroundTaskImageResponseBodySchema = z.object({
+  id: z.string(),
+  backgroundTaskArtifactId: z.string(),
+  backgroundTaskId: z.string(),
+  name: z.string(),
+  tag: z.string(),
+  digest: z.string(),
+  size: z.number(),
+  provider: z.string(),
+  createdAt: z.coerce.date(),
+  updatedAt: z.coerce.date(),
+});
+
+export type CreateBackgroundTaskImageResponseBody = z.infer<
+  typeof CreateBackgroundTaskImageResponseBodySchema
+>;

packages/core/src/schemas/backgroundTasks.ts

@@ -0,0 +1,7 @@
+import { z } from "zod";
+
+export const BackgroundTaskOperationParamsSchema = z.object({
+  id: z.string(),
+  version: z.string(),
+  payload: z.any(),
+});

packages/core/src/schemas/index.ts

@@ -12,3 +12,4 @@ export * from "./fetch";
 export * from "./events";
 export * from "./runs";
 export * from "./addMissingVersionField";
+export * from "./backgroundTasks";

packages/database/prisma/migrations/20230905161917_add_background_tasks/migration.sql

@@ -0,0 +1,103 @@
+-- CreateTable
+CREATE TABLE "BackgroundTask" (
+    "id" TEXT NOT NULL,
+    "slug" TEXT NOT NULL,
+    "title" TEXT NOT NULL,
+    "organizationId" TEXT NOT NULL,
+    "projectId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "deletedAt" TIMESTAMP(3),
+
+    CONSTRAINT "BackgroundTask_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskVersion" (
+    "id" TEXT NOT NULL,
+    "version" TEXT NOT NULL,
+    "cpu" INTEGER NOT NULL DEFAULT 1,
+    "memory" INTEGER NOT NULL DEFAULT 256,
+    "concurrency" INTEGER NOT NULL DEFAULT 1,
+    "backgroundTaskId" TEXT NOT NULL,
+    "endpointId" TEXT NOT NULL,
+    "environmentId" TEXT NOT NULL,
+    "organizationId" TEXT NOT NULL,
+    "projectId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "BackgroundTaskVersion_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskAlias" (
+    "id" TEXT NOT NULL,
+    "name" TEXT NOT NULL DEFAULT 'latest',
+    "value" TEXT NOT NULL,
+    "versionId" TEXT NOT NULL,
+    "backgroundTaskId" TEXT NOT NULL,
+    "environmentId" TEXT NOT NULL,
+
+    CONSTRAINT "BackgroundTaskAlias_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskSecret" (
+    "id" TEXT NOT NULL,
+    "key" TEXT NOT NULL,
+    "secretReferenceId" TEXT NOT NULL,
+    "backgroundTaskVersionId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "BackgroundTaskSecret_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTask_projectId_slug_key" ON "BackgroundTask"("projectId", "slug");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskVersion_backgroundTaskId_version_environmentI_key" ON "BackgroundTaskVersion"("backgroundTaskId", "version", "environmentId");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskAlias_backgroundTaskId_environmentId_name_key" ON "BackgroundTaskAlias"("backgroundTaskId", "environmentId", "name");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskSecret_backgroundTaskVersionId_key_key" ON "BackgroundTaskSecret"("backgroundTaskVersionId", "key");
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTask" ADD CONSTRAINT "BackgroundTask_organizationId_fkey" FOREIGN KEY ("organizationId") REFERENCES "Organization"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTask" ADD CONSTRAINT "BackgroundTask_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskVersion" ADD CONSTRAINT "BackgroundTaskVersion_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskVersion" ADD CONSTRAINT "BackgroundTaskVersion_endpointId_fkey" FOREIGN KEY ("endpointId") REFERENCES "Endpoint"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskVersion" ADD CONSTRAINT "BackgroundTaskVersion_environmentId_fkey" FOREIGN KEY ("environmentId") REFERENCES "RuntimeEnvironment"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskVersion" ADD CONSTRAINT "BackgroundTaskVersion_organizationId_fkey" FOREIGN KEY ("organizationId") REFERENCES "Organization"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskVersion" ADD CONSTRAINT "BackgroundTaskVersion_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskAlias" ADD CONSTRAINT "BackgroundTaskAlias_versionId_fkey" FOREIGN KEY ("versionId") REFERENCES "BackgroundTaskVersion"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskAlias" ADD CONSTRAINT "BackgroundTaskAlias_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskAlias" ADD CONSTRAINT "BackgroundTaskAlias_environmentId_fkey" FOREIGN KEY ("environmentId") REFERENCES "RuntimeEnvironment"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskSecret" ADD CONSTRAINT "BackgroundTaskSecret_secretReferenceId_fkey" FOREIGN KEY ("secretReferenceId") REFERENCES "SecretReference"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskSecret" ADD CONSTRAINT "BackgroundTaskSecret_backgroundTaskVersionId_fkey" FOREIGN KEY ("backgroundTaskVersionId") REFERENCES "BackgroundTaskVersion"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/database/prisma/migrations/20230907093954_adding_background_task_artifact/migration.sql

@@ -0,0 +1,22 @@
+-- CreateTable
+CREATE TABLE "BackgroundTaskArtifact" (
+    "id" TEXT NOT NULL,
+    "version" TEXT NOT NULL,
+    "fileName" TEXT NOT NULL,
+    "hash" TEXT NOT NULL,
+    "bundle" TEXT NOT NULL,
+    "sourcemap" JSONB NOT NULL,
+    "nodeVersion" TEXT NOT NULL,
+    "dependencies" JSONB NOT NULL,
+    "backgroundTaskId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "BackgroundTaskArtifact_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskArtifact_backgroundTaskId_version_hash_key" ON "BackgroundTaskArtifact"("backgroundTaskId", "version", "hash");
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskArtifact" ADD CONSTRAINT "BackgroundTaskArtifact_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/database/prisma/migrations/20230912105319_add_more_background_task_models/migration.sql

@@ -0,0 +1,78 @@
+-- CreateEnum
+CREATE TYPE "BackgroundTaskProvider" AS ENUM ('FLY_IO');
+
+-- CreateEnum
+CREATE TYPE "BackgroundTaskMachineStatus" AS ENUM ('CREATED', 'STARTING', 'STARTED', 'STOPPING', 'STOPPED', 'DESTROYING', 'DESTROYED', 'REPLACING');
+
+-- CreateEnum
+CREATE TYPE "BackgroundTaskOperationStatus" AS ENUM ('PENDING', 'STARTED', 'SUCCESS', 'FAILURE');
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskImage" (
+    "id" TEXT NOT NULL,
+    "provider" "BackgroundTaskProvider" NOT NULL,
+    "name" TEXT NOT NULL,
+    "tag" TEXT NOT NULL,
+    "digest" TEXT NOT NULL,
+    "size" INTEGER NOT NULL,
+    "backgroundTaskId" TEXT NOT NULL,
+    "backgroundTaskArtifactId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "BackgroundTaskImage_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskMachine" (
+    "id" TEXT NOT NULL,
+    "externalId" TEXT NOT NULL,
+    "provider" "BackgroundTaskProvider" NOT NULL,
+    "data" JSONB NOT NULL,
+    "status" "BackgroundTaskMachineStatus" NOT NULL DEFAULT 'CREATED',
+    "backgroundTaskId" TEXT NOT NULL,
+    "backgroundTaskVersionId" TEXT NOT NULL,
+    "backgroundTaskImageId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "BackgroundTaskMachine_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskOperation" (
+    "id" TEXT NOT NULL,
+    "backgroundTaskId" TEXT NOT NULL,
+    "backgroundTaskVersionId" TEXT NOT NULL,
+    "payload" JSONB NOT NULL,
+    "output" JSONB,
+    "error" JSONB,
+    "status" "BackgroundTaskOperationStatus" NOT NULL DEFAULT 'PENDING',
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    "startedAt" TIMESTAMP(3),
+    "endedAt" TIMESTAMP(3),
+
+    CONSTRAINT "BackgroundTaskOperation_pkey" PRIMARY KEY ("id")
+);
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskImage" ADD CONSTRAINT "BackgroundTaskImage_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskImage" ADD CONSTRAINT "BackgroundTaskImage_backgroundTaskArtifactId_fkey" FOREIGN KEY ("backgroundTaskArtifactId") REFERENCES "BackgroundTaskArtifact"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachine" ADD CONSTRAINT "BackgroundTaskMachine_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachine" ADD CONSTRAINT "BackgroundTaskMachine_backgroundTaskVersionId_fkey" FOREIGN KEY ("backgroundTaskVersionId") REFERENCES "BackgroundTaskVersion"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachine" ADD CONSTRAINT "BackgroundTaskMachine_backgroundTaskImageId_fkey" FOREIGN KEY ("backgroundTaskImageId") REFERENCES "BackgroundTaskImage"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskOperation" ADD CONSTRAINT "BackgroundTaskOperation_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskOperation" ADD CONSTRAINT "BackgroundTaskOperation_backgroundTaskVersionId_fkey" FOREIGN KEY ("backgroundTaskVersionId") REFERENCES "BackgroundTaskVersion"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/database/prisma/migrations/20230912122347_add_unique_constraint_on_background_task_images/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - A unique constraint covering the columns `[backgroundTaskArtifactId,digest]` on the table `BackgroundTaskImage` will be added. If there are existing duplicate values, this will fail.
+
+*/
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskImage_backgroundTaskArtifactId_digest_key" ON "BackgroundTaskImage"("backgroundTaskArtifactId", "digest");

packages/database/prisma/migrations/20230912122608_rename_background_task_provider_enum/migration.sql

@@ -0,0 +1,20 @@
+/*
+  Warnings:
+
+  - Changed the type of `provider` on the `BackgroundTaskImage` table. No cast exists, the column would be dropped and recreated, which cannot be done if there is data, since the column is required.
+  - Changed the type of `provider` on the `BackgroundTaskMachine` table. No cast exists, the column would be dropped and recreated, which cannot be done if there is data, since the column is required.
+
+*/
+-- CreateEnum
+CREATE TYPE "BackgroundTaskProviderStrategy" AS ENUM ('FLY_IO');
+
+-- AlterTable
+ALTER TABLE "BackgroundTaskImage" DROP COLUMN "provider",
+ADD COLUMN     "provider" "BackgroundTaskProviderStrategy" NOT NULL;
+
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachine" DROP COLUMN "provider",
+ADD COLUMN     "provider" "BackgroundTaskProviderStrategy" NOT NULL;
+
+-- DropEnum
+DROP TYPE "BackgroundTaskProvider";

packages/database/prisma/migrations/20230912122727_add_unsupported_provider_strategy/migration.sql

@@ -0,0 +1,2 @@
+-- AlterEnum
+ALTER TYPE "BackgroundTaskProviderStrategy" ADD VALUE 'UNSUPPORTED';

packages/database/prisma/migrations/20230912130758_add_task_to_background_task_operation/migration.sql

@@ -0,0 +1,15 @@
+/*
+  Warnings:
+
+  - A unique constraint covering the columns `[taskId]` on the table `BackgroundTaskOperation` will be added. If there are existing duplicate values, this will fail.
+  - Added the required column `taskId` to the `BackgroundTaskOperation` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "BackgroundTaskOperation" ADD COLUMN     "taskId" TEXT NOT NULL;
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskOperation_taskId_key" ON "BackgroundTaskOperation"("taskId");
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskOperation" ADD CONSTRAINT "BackgroundTaskOperation_taskId_fkey" FOREIGN KEY ("taskId") REFERENCES "Task"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/database/prisma/migrations/20230912132333_add_waiting_on_image_to_operation_status/migration.sql

@@ -0,0 +1,2 @@
+-- AlterEnum
+ALTER TYPE "BackgroundTaskOperationStatus" ADD VALUE 'WAITING_ON_IMAGE';

packages/database/prisma/migrations/20230912132627_adding_pending_to_machine_status/migration.sql

@@ -0,0 +1,2 @@
+-- AlterEnum
+ALTER TYPE "BackgroundTaskMachineStatus" ADD VALUE 'PENDING';

packages/database/prisma/migrations/20230912133614_added_machine_pools/migration.sql

@@ -0,0 +1,45 @@
+/*
+  Warnings:
+
+  - Added the required column `poolId` to the `BackgroundTaskMachine` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterEnum
+ALTER TYPE "BackgroundTaskOperationStatus" ADD VALUE 'ASSIGNED_TO_POOL';
+
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachine" ADD COLUMN     "poolId" TEXT NOT NULL;
+
+-- AlterTable
+ALTER TABLE "BackgroundTaskOperation" ADD COLUMN     "poolId" TEXT;
+
+-- CreateTable
+CREATE TABLE "BackgroundTaskMachinePool" (
+    "id" TEXT NOT NULL,
+    "provider" "BackgroundTaskProviderStrategy" NOT NULL,
+    "imageId" TEXT NOT NULL,
+    "backgroundTaskVersionId" TEXT NOT NULL,
+    "backgroundTaskId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "BackgroundTaskMachinePool_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "BackgroundTaskMachinePool_backgroundTaskVersionId_imageId_key" ON "BackgroundTaskMachinePool"("backgroundTaskVersionId", "imageId");
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachine" ADD CONSTRAINT "BackgroundTaskMachine_poolId_fkey" FOREIGN KEY ("poolId") REFERENCES "BackgroundTaskMachinePool"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachinePool" ADD CONSTRAINT "BackgroundTaskMachinePool_imageId_fkey" FOREIGN KEY ("imageId") REFERENCES "BackgroundTaskImage"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachinePool" ADD CONSTRAINT "BackgroundTaskMachinePool_backgroundTaskVersionId_fkey" FOREIGN KEY ("backgroundTaskVersionId") REFERENCES "BackgroundTaskVersion"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskMachinePool" ADD CONSTRAINT "BackgroundTaskMachinePool_backgroundTaskId_fkey" FOREIGN KEY ("backgroundTaskId") REFERENCES "BackgroundTask"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "BackgroundTaskOperation" ADD CONSTRAINT "BackgroundTaskOperation_poolId_fkey" FOREIGN KEY ("poolId") REFERENCES "BackgroundTaskMachinePool"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/database/prisma/migrations/20230912142718_make_external_data_optional_on_task_machines/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachine" ALTER COLUMN "externalId" DROP NOT NULL,
+ALTER COLUMN "data" DROP NOT NULL;

packages/database/prisma/migrations/20230912152725_change_machine_status_default/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachine" ALTER COLUMN "status" SET DEFAULT 'PENDING';

packages/database/prisma/migrations/20230913115949_add_region_to_background_task_version/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "BackgroundTaskVersion" ADD COLUMN     "region" TEXT NOT NULL DEFAULT 'iad';

packages/database/prisma/migrations/20230913120657_add_region_to_pools/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - Added the required column `region` to the `BackgroundTaskMachinePool` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachinePool" ADD COLUMN     "region" TEXT NOT NULL;

packages/database/prisma/migrations/20230913120826_add_other_config_to_pools/migration.sql

@@ -0,0 +1,4 @@
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachinePool" ADD COLUMN     "concurrency" INTEGER NOT NULL DEFAULT 1,
+ADD COLUMN     "cpu" INTEGER NOT NULL DEFAULT 1,
+ADD COLUMN     "memory" INTEGER NOT NULL DEFAULT 256;

packages/database/prisma/migrations/20230913121002_add_disk_size_to_models/migration.sql

@@ -0,0 +1,5 @@
+-- AlterTable
+ALTER TABLE "BackgroundTaskMachinePool" ADD COLUMN     "diskSize" INTEGER NOT NULL DEFAULT 1;
+
+-- AlterTable
+ALTER TABLE "BackgroundTaskVersion" ADD COLUMN     "diskSize" INTEGER NOT NULL DEFAULT 1;

packages/database/prisma/schema.prisma

@@ -76,12 +76,14 @@ model Organization {
   events       EventRecord[]
   jobRuns      JobRun[]
 
-  projects         Project[]
-  members          OrgMember[]
-  invites          OrgMemberInvite[]
-  externalAccounts ExternalAccount[]
-  integrations     Integration[]
-  sources          TriggerSource[]
+  projects               Project[]
+  members                OrgMember[]
+  invites                OrgMemberInvite[]
+  externalAccounts       ExternalAccount[]
+  integrations           Integration[]
+  sources                TriggerSource[]
+  backgroundTasks        BackgroundTask[]
+  backgroundTaskVersions BackgroundTaskVersion[]
 }
 
 model ExternalAccount {
@@ -316,7 +318,10 @@ model RuntimeEnvironment {
   sources           TriggerSource[]
   eventDispatchers  EventDispatcher[]
   scheduleSources   ScheduleSource[]
-  ExternalAccount   ExternalAccount[]
+  externalAccounts  ExternalAccount[]
+
+  backgroundTaskVersions BackgroundTaskVersion[]
+  backgroundTaskAliases  BackgroundTaskAlias[]
 
   @@unique([projectId, slug, orgMemberId])
 }
@@ -346,6 +351,9 @@ model Project {
   events       EventRecord[]
   runs         JobRun[]
   sources      TriggerSource[]
+
+  backgroundTasks        BackgroundTask[]
+  backgroundTaskVersions BackgroundTaskVersion[]
 }
 
 model Endpoint {
@@ -374,6 +382,8 @@ model Endpoint {
   sources               TriggerSource[]
   indexings             EndpointIndex[]
 
+  backgroundTaskVersions BackgroundTaskVersion[]
+
   @@unique([environmentId, slug])
 }
 
@@ -807,9 +817,10 @@ model Task {
   runConnection   RunConnection? @relation(fields: [runConnectionId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   runConnectionId String?
 
-  children   Task[]            @relation("TaskParent")
-  executions JobRunExecution[]
-  attempts   TaskAttempt[]
+  children                Task[]                   @relation("TaskParent")
+  executions              JobRunExecution[]
+  attempts                TaskAttempt[]
+  backgroundTaskOperation BackgroundTaskOperation?
 
   @@unique([runId, idempotencyKey])
 }
@@ -859,6 +870,8 @@ model SecretReference {
   integrations   Integration[]
   triggerSources TriggerSource[]
 
+  backgroundTaskSecrets BackgroundTaskSecret[]
+
   createdAt DateTime @default(now())
   updatedAt DateTime @updatedAt
 }
@@ -1068,3 +1081,252 @@ model ApiIntegrationVote {
 
   @@unique([apiIdentifier, userId])
 }
+
+model BackgroundTask {
+  id    String @id @default(cuid())
+  slug  String
+  title String
+
+  organization   Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  organizationId String
+
+  project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  projectId String
+
+  versions  BackgroundTaskVersion[]
+  aliases   BackgroundTaskAlias[]
+  artifacts BackgroundTaskArtifact[]
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @default(now()) @updatedAt
+
+  deletedAt DateTime?
+
+  operations                BackgroundTaskOperation[]
+  machines                  BackgroundTaskMachine[]
+  images                    BackgroundTaskImage[]
+  BackgroundTaskMachinePool BackgroundTaskMachinePool[]
+
+  @@unique([projectId, slug])
+}
+
+model BackgroundTaskVersion {
+  id      String @id @default(cuid())
+  version String
+
+  cpu         Int    @default(1)
+  memory      Int    @default(256)
+  concurrency Int    @default(1)
+  diskSize    Int    @default(1)
+  region      String @default("iad")
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  endpoint   Endpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  endpointId String
+
+  environment   RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  environmentId String
+
+  organization   Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  organizationId String
+
+  project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  projectId String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  aliases    BackgroundTaskAlias[]
+  secrets    BackgroundTaskSecret[]
+  operations BackgroundTaskOperation[]
+  machines   BackgroundTaskMachine[]
+  pools      BackgroundTaskMachinePool[]
+
+  @@unique([backgroundTaskId, version, environmentId])
+}
+
+model BackgroundTaskAlias {
+  id    String @id @default(cuid())
+  name  String @default("latest")
+  value String
+
+  version   BackgroundTaskVersion @relation(fields: [versionId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  versionId String
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  environment   RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  environmentId String
+
+  @@unique([backgroundTaskId, environmentId, name])
+}
+
+model BackgroundTaskSecret {
+  id  String @id @default(cuid())
+  key String
+
+  secretReference   SecretReference @relation(fields: [secretReferenceId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  secretReferenceId String
+
+  backgroundTaskVersion   BackgroundTaskVersion @relation(fields: [backgroundTaskVersionId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskVersionId String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  @@unique([backgroundTaskVersionId, key])
+}
+
+model BackgroundTaskArtifact {
+  id      String @id @default(cuid())
+  version String
+
+  fileName     String
+  hash         String
+  bundle       String
+  sourcemap    Json
+  nodeVersion  String
+  dependencies Json
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  images BackgroundTaskImage[]
+
+  @@unique([backgroundTaskId, version, hash])
+}
+
+enum BackgroundTaskProviderStrategy {
+  FLY_IO
+  UNSUPPORTED
+}
+
+model BackgroundTaskImage {
+  id       String                         @id @default(cuid())
+  provider BackgroundTaskProviderStrategy
+  name     String
+  tag      String
+  digest   String
+  size     Int
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  backgroundTaskArtifact   BackgroundTaskArtifact @relation(fields: [backgroundTaskArtifactId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskArtifactId String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  machines BackgroundTaskMachine[]
+  pools    BackgroundTaskMachinePool[]
+
+  @@unique([backgroundTaskArtifactId, digest])
+}
+
+model BackgroundTaskMachine {
+  id String @id @default(cuid())
+
+  externalId String?
+  provider   BackgroundTaskProviderStrategy
+  data       Json?
+  status     BackgroundTaskMachineStatus    @default(PENDING)
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  backgroundTaskVersion   BackgroundTaskVersion @relation(fields: [backgroundTaskVersionId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskVersionId String
+
+  backgroundTaskImage   BackgroundTaskImage @relation(fields: [backgroundTaskImageId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskImageId String
+
+  pool   BackgroundTaskMachinePool @relation(fields: [poolId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  poolId String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+}
+
+enum BackgroundTaskMachineStatus {
+  PENDING
+  CREATED
+  STARTING
+  STARTED
+  STOPPING
+  STOPPED
+  DESTROYING
+  DESTROYED
+  REPLACING
+}
+
+model BackgroundTaskMachinePool {
+  id String @id @default(cuid())
+
+  provider BackgroundTaskProviderStrategy
+
+  region      String
+  cpu         Int    @default(1)
+  memory      Int    @default(256)
+  concurrency Int    @default(1)
+  diskSize    Int    @default(1)
+
+  image   BackgroundTaskImage @relation(fields: [imageId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  imageId String
+
+  backgroundTaskVersion   BackgroundTaskVersion @relation(fields: [backgroundTaskVersionId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskVersionId String
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  machines   BackgroundTaskMachine[]
+  operations BackgroundTaskOperation[]
+
+  @@unique([backgroundTaskVersionId, imageId])
+}
+
+model BackgroundTaskOperation {
+  id String @id @default(cuid())
+
+  backgroundTask   BackgroundTask @relation(fields: [backgroundTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskId String
+
+  backgroundTaskVersion   BackgroundTaskVersion @relation(fields: [backgroundTaskVersionId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  backgroundTaskVersionId String
+
+  task   Task   @relation(fields: [taskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  taskId String @unique
+
+  pool   BackgroundTaskMachinePool? @relation(fields: [poolId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  poolId String?
+
+  payload Json
+  output  Json?
+  error   Json?
+
+  status BackgroundTaskOperationStatus @default(PENDING)
+
+  createdAt DateTime  @default(now())
+  updatedAt DateTime  @updatedAt
+  startedAt DateTime?
+  endedAt   DateTime?
+}
+
+enum BackgroundTaskOperationStatus {
+  PENDING
+  WAITING_ON_IMAGE
+  ASSIGNED_TO_POOL
+  STARTED
+  SUCCESS
+  FAILURE
+}

packages/trigger-sdk/package.json

@@ -20,6 +20,7 @@
   },
   "scripts": {
     "clean": "rimraf dist",
+    "dev": "tsup --watch --dts-resolve",
     "build": "npm run clean && npm run build:tsup",
     "build:tsup": "tsup --dts-resolve",
     "typecheck": "tsc --noEmit"

packages/trigger-sdk/src/backgroundTask.ts

@@ -0,0 +1,105 @@
+import { BackgroundTaskMetadata, LogLevel } from "@trigger.dev/core";
+import { z } from "zod";
+import { TriggerClient } from "./triggerClient";
+import { slugifyId } from "./utils";
+import { runLocalStorage } from "./runLocalStorage";
+
+export type BackgroundTaskOptions<TPayload = any, TRunResult = any> = {
+  id: string;
+  name: string;
+  version: string;
+  schema?: z.Schema<TPayload>;
+  logLevel?: LogLevel;
+
+  cpu?: 1 | 2 | 4 | 8 | 16 | 32 | 64 | 128;
+  memory?: 256 | 512 | 1024 | 2048 | 4096 | 8192 | 16384 | 32768;
+  concurrency?: number;
+  region?: string;
+  diskSizeInGB?: number;
+
+  secrets?: {
+    [key: string]: string;
+  };
+
+  enabled?: boolean;
+  run: (payload: TPayload) => Promise<TRunResult>;
+};
+
+export class BackgroundTask<TPayload = any, TRunResult = any> {
+  readonly options: BackgroundTaskOptions<TPayload, TRunResult>;
+
+  client: TriggerClient;
+
+  constructor(client: TriggerClient, options: BackgroundTaskOptions<TPayload, TRunResult>) {
+    this.client = client;
+    this.options = options;
+    this.#validate();
+
+    client.attachBackgroundTask(this);
+  }
+
+  get id() {
+    return slugifyId(this.options.id);
+  }
+
+  get enabled() {
+    return typeof this.options.enabled === "boolean" ? this.options.enabled : true;
+  }
+
+  get name() {
+    return this.options.name;
+  }
+
+  get schema() {
+    return this.options.schema;
+  }
+
+  get version() {
+    return this.options.version;
+  }
+
+  get logLevel() {
+    return this.options.logLevel;
+  }
+
+  public async invoke(key: string | string[], payload: TPayload): Promise<TRunResult> {
+    if (!this.enabled) {
+      throw new Error(`Cannot invoke a disabled background task: ${this.id}`);
+    }
+
+    const runStore = runLocalStorage.getStore();
+
+    if (!runStore) {
+      throw new Error("Cannot invoke a background task outside of a job run");
+    }
+
+    const { io, ctx } = runStore;
+
+    return await io.backgroundTask(key, this.id, this.version, payload);
+  }
+
+  toJSON(): BackgroundTaskMetadata {
+    return {
+      id: this.id,
+      name: this.name,
+      version: this.version,
+      enabled: this.enabled,
+      cpu: this.options.cpu ?? 1,
+      memory: this.options.memory ?? 256,
+      region: this.options.region,
+      concurrency: this.options.concurrency ?? 1,
+      diskSizeInGB: this.options.diskSizeInGB,
+      secrets: this.options.secrets ?? {},
+    };
+  }
+
+  // Make sure the id is valid (must only contain alphanumeric characters and dashes)
+  // Make sure the version is valid (must be a valid semver version)
+  #validate() {
+    if (!this.version.match(/^(\d+)\.(\d+)\.(\d+)$/)) {
+      throw new Error(
+        `Invalid job version: "${this.version}". BackgroundTask versions must be valid semver versions.`
+      );
+    }
+  }
+}

packages/trigger-sdk/src/io.ts

@@ -199,6 +199,42 @@ export class IO {
     )) as TResponseData;
   }
 
+  /** `io.backgroundTask()` invokes a background task.
+   * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+   * @param taskId The id of the background task to invoke.
+   * @param payload The payload to send to the background task.
+   */
+  async backgroundTask<TResultData>(
+    key: string | any[],
+    id: string,
+    version: string,
+    payload: any
+  ): Promise<TResultData> {
+    return (await this.runTask(
+      key,
+      async (task) => {
+        return task.output;
+      },
+      {
+        name: `invoke ${id}`,
+        params: { payload, id, version },
+        operation: "backgroundTask",
+        icon: "background",
+        noop: false,
+        properties: [
+          {
+            label: "Task ID",
+            text: id,
+          },
+          {
+            label: "Version",
+            text: version,
+          },
+        ],
+      }
+    )) as TResultData;
+  }
+
   /** `io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name).
    * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
    * @param event The event to send. The event name must match the name of the event that your Jobs are listening for.

packages/trigger-sdk/src/runLocalStorage.ts

@@ -0,0 +1,10 @@
+import { IO } from "./io";
+import { TriggerContext } from "./types";
+import { TypedAsyncLocalStorage } from "./utils/typedAsyncLocalStorage";
+
+export type RunStore = {
+  io: IO;
+  ctx: TriggerContext;
+};
+
+export const runLocalStorage = new TypedAsyncLocalStorage<RunStore>();

packages/trigger-sdk/src/triggerClient.ts

@@ -42,6 +42,8 @@ import type {
   TriggerContext,
   TriggerPreprocessContext,
 } from "./types";
+import { BackgroundTask, BackgroundTaskOptions } from "./backgroundTask";
+import { runLocalStorage } from "./runLocalStorage";
 
 const registerSourceEvent: EventSpecification<RegisterSourceEventV2> = {
   name: REGISTER_SOURCE_EVENT_V2,
@@ -76,6 +78,7 @@ export class TriggerClient {
   #options: TriggerClientOptions;
   #registeredJobs: Record<string, Job<Trigger<EventSpecification<any>>, any>> = {};
   #registeredSources: Record<string, SourceMetadataV2> = {};
+  #registeredBackgroundTasks: Record<string, BackgroundTask<any>> = {};
   #registeredHttpSourceHandlers: Record<
     string,
     (
@@ -221,6 +224,9 @@ export class TriggerClient {
 
         const body: IndexEndpointResponse = {
           jobs: Object.values(this.#registeredJobs).map((job) => job.toJSON()),
+          backgroundTasks: Object.values(this.#registeredBackgroundTasks).map((task) =>
+            task.toJSON()
+          ),
           sources: Object.values(this.#registeredSources),
           dynamicTriggers: Object.values(this.#registeredDynamicTriggers).map((trigger) => ({
             id: trigger.id,
@@ -426,6 +432,10 @@ export class TriggerClient {
     job.trigger.attachToJob(this, job);
   }
 
+  attachBackgroundTask(task: BackgroundTask<any>): void {
+    this.#registeredBackgroundTasks[task.id] = task;
+  }
+
   attachDynamicTrigger(trigger: DynamicTrigger<any, any>): void {
     this.#registeredDynamicTriggers[trigger.id] = trigger;
 
@@ -648,11 +658,13 @@ export class TriggerClient {
     );
 
     try {
-      const output = await job.options.run(
-        job.trigger.event.parsePayload(body.event.payload ?? {}),
-        ioWithConnections,
-        context
-      );
+      const output = await runLocalStorage.runWith({ io, ctx: context }, () => {
+        return job.options.run(
+          job.trigger.event.parsePayload(body.event.payload ?? {}),
+          ioWithConnections,
+          context
+        );
+      });
 
       return { status: "SUCCESS", output };
     } catch (error) {
@@ -863,6 +875,12 @@ export class TriggerClient {
   >(options: JobOptions<TTrigger, TIntegrations>) {
     return new Job<TTrigger, TIntegrations>(this, options);
   }
+
+  defineBackgroundTask<TPayload = any, TRunResult = any>(
+    options: BackgroundTaskOptions<TPayload, TRunResult>
+  ) {
+    return new BackgroundTask<TPayload, TRunResult>(this, options);
+  }
 }
 
 function dynamicTriggerRegisterSourceJobId(id: string) {

packages/trigger-sdk/src/utils/typedAsyncLocalStorage.ts

@@ -0,0 +1,17 @@
+import { AsyncLocalStorage } from "async_hooks";
+
+export class TypedAsyncLocalStorage<T> {
+  private storage: AsyncLocalStorage<T>;
+
+  constructor() {
+    this.storage = new AsyncLocalStorage<T>();
+  }
+
+  runWith<R extends (...args: any[]) => Promise<any>>(context: T, fn: R): Promise<ReturnType<R>> {
+    return this.storage.run(context, fn);
+  }
+
+  getStore(): T | undefined {
+    return this.storage.getStore();
+  }
+}