Add support for background task operations, starting with fetch

.changeset/modern-emus-suffer.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/github": patch
+"@trigger.dev/openai": patch
+"@trigger.dev/slack": patch
+---
+
+Added support for backgroundFetch

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

@@ -24,5 +24,6 @@ export function taskWithAttemptsToServerTask(
     parentId: task.parentId,
     attempts: task.attempts.length,
     idempotencyKey: task.idempotencyKey,
+    operation: task.operation,
   };
 }

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

@@ -22,6 +22,7 @@ import {
   UpdatingDuration,
 } from "../_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/RunCard";
 import { TaskStatusIcon } from "../_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/TaskStatus";
+import { sensitiveDataReplacer } from "~/services/logger";
 
 export const loader = async ({ request, params }: LoaderArgs) => {
   const userId = await requireUserId(request);
@@ -134,7 +135,9 @@ export default function Page() {
         <div className="mt-4 flex flex-col gap-2">
           <Header3>Input</Header3>
           {params ? (
-            <CodeBlock code={JSON.stringify(params, null, 2)} />
+            <CodeBlock
+              code={JSON.stringify(params, sensitiveDataReplacer, 2)}
+            />
           ) : (
             <Paragraph variant="small">No input</Paragraph>
           )}

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

@@ -222,12 +222,10 @@ export default function Page() {
                 Test run
               </span>
             )}
-            {(basicStatus === "COMPLETED" || basicStatus === "FAILED") && (
-              <RerunPopover
-                environmentType={run.environment.type}
-                status={basicStatus}
-              />
-            )}
+            <RerunPopover
+              environmentType={run.environment.type}
+              status={basicStatus}
+            />
           </PageButtons>
         </PageTitleRow>
         <PageInfoRow>

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

@@ -11,6 +11,7 @@ import { taskWithAttemptsToServerTask } from "~/models/task.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger";
 import { ulid } from "~/services/ulid.server";
+import { workerQueue } from "~/services/worker.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -230,6 +231,7 @@ export class RunTaskService {
           params: taskBody.params ?? undefined,
           properties: taskBody.properties ?? undefined,
           redact: taskBody.redact ?? undefined,
+          operation: taskBody.operation,
           style: taskBody.style ?? { style: "normal" },
           attempts: {
             create: {
@@ -244,6 +246,17 @@ export class RunTaskService {
         },
       });
 
+      if (task.status === "RUNNING" && typeof taskBody.operation === "string") {
+        // We need to schedule the operation
+        await workerQueue.enqueue(
+          "performTaskOperation",
+          {
+            id: task.id,
+          },
+          { tx, runAt: task.delayUntil ?? undefined }
+        );
+      }
+
       return task;
     });
 

apps/webapp/app/services/logger.ts

@@ -1,13 +1,37 @@
-import type { LogLevel } from "@trigger.dev/internal";
+import type { LogLevel, RedactString } from "@trigger.dev/internal";
 import { Logger } from "@trigger.dev/internal";
 
 export const logger = new Logger(
   "webapp",
-  (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel
+  (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
+  [],
+  sensitiveDataReplacer
 );
 
-export const projectLogger = logger.filter(
-  "latestCommit",
-  "dockerfile",
-  "dockerignore"
-);
+// Replaces redacted strings with "******".
+// For example, this object: {"Authorization":{"__redactedString":true,"strings":["Bearer ",""],"interpolations":["sk-1234"]}}
+// Would get stringified like so: {"Authorization": "Bearer ******"}
+export function sensitiveDataReplacer(key: string, value: any): any {
+  if (
+    typeof value === "object" &&
+    value !== null &&
+    value.__redactedString === true
+  ) {
+    return redactString(value);
+  }
+
+  return value;
+}
+
+function redactString(value: RedactString) {
+  let result = "";
+
+  for (let i = 0; i < value.strings.length; i++) {
+    result += value.strings[i];
+    if (i < value.interpolations.length) {
+      result += "********";
+    }
+  }
+
+  return result;
+}

apps/webapp/app/services/runs/performRunExecution.ts

@@ -393,32 +393,36 @@ export class PerformRunExecutionService {
         },
       });
 
-      const newJobExecution = await tx.jobRunExecution.create({
-        data: {
-          runId: run.id,
-          reason: "EXECUTE_JOB",
-          status: "PENDING",
-          retryLimit: EXECUTE_JOB_RETRY_LIMIT,
-          resumeTaskId: data.task.id,
-        },
-      });
+      // If the task has an operation, then the next performRunExecution will occur
+      // when that operation has finished
+      if (!data.task.operation) {
+        const newJobExecution = await tx.jobRunExecution.create({
+          data: {
+            runId: run.id,
+            reason: "EXECUTE_JOB",
+            status: "PENDING",
+            retryLimit: EXECUTE_JOB_RETRY_LIMIT,
+            resumeTaskId: data.task.id,
+          },
+        });
 
-      const graphileJob = await workerQueue.enqueue(
-        "performRunExecution",
-        {
-          id: newJobExecution.id,
-        },
-        { tx, runAt: data.task.delayUntil ?? undefined }
-      );
+        const graphileJob = await workerQueue.enqueue(
+          "performRunExecution",
+          {
+            id: newJobExecution.id,
+          },
+          { tx, runAt: data.task.delayUntil ?? undefined }
+        );
 
-      await tx.jobRunExecution.update({
-        where: {
-          id: newJobExecution.id,
-        },
-        data: {
-          graphileJobId: graphileJob.id,
-        },
-      });
+        await tx.jobRunExecution.update({
+          where: {
+            id: newJobExecution.id,
+          },
+          data: {
+            graphileJobId: graphileJob.id,
+          },
+        });
+      }
     });
   }
 

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

@@ -0,0 +1,373 @@
+import type { Task } from "@trigger.dev/database";
+import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts";
+import {
+  $transaction,
+  PrismaClient,
+  PrismaClientOrTransaction,
+  prisma,
+} from "~/db.server";
+import { workerQueue } from "../worker.server";
+import {
+  FetchOperationSchema,
+  FetchRequestInit,
+  FetchRetryOptions,
+  FetchRetryStrategy,
+  RedactString,
+  calculateRetryAt,
+} from "@trigger.dev/internal";
+import { safeJsonFromResponse } from "~/utils/json";
+import { logger } from "../logger";
+import { formatUnknownError } from "~/utils";
+
+type FoundTask = Awaited<ReturnType<typeof findTask>>;
+
+export class PerformTaskOperationService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(id: string) {
+    const task = await findTask(this.#prismaClient, id);
+
+    if (!task) {
+      return;
+    }
+
+    if (task.status === "COMPLETED" || task.status === "ERRORED") {
+      return await this.#resumeRunExecution(task, this.#prismaClient);
+    }
+
+    if (!task.operation) {
+      return await this.#resumeTask(task, null);
+    }
+
+    logger.debug("PerformTaskOperationService.call", { task });
+
+    switch (task.operation) {
+      case "fetch": {
+        const fetchOperation = FetchOperationSchema.safeParse(task.params);
+
+        if (!fetchOperation.success) {
+          return await this.#resumeTaskWithError(
+            task,
+            `Invalid fetch operation: ${fetchOperation.error.message}`
+          );
+        }
+
+        const { url, requestInit, retry } = fetchOperation.data;
+
+        const response = await fetch(url, {
+          method: requestInit?.method ?? "GET",
+          headers: normalizeHeaders(requestInit?.headers ?? {}),
+          body: requestInit?.body,
+        });
+
+        const jsonBody = await safeJsonFromResponse(response);
+
+        logger.debug("PerformTaskOperationService.call.fetch", {
+          url,
+          requestInit,
+          retry,
+          statusCode: response.status,
+          headers: Object.fromEntries(response.headers.entries()),
+          jsonBody,
+        });
+
+        if (!response.ok) {
+          const retryAt = this.#calculateRetryForResponse(
+            task,
+            retry,
+            response
+          );
+
+          if (retryAt) {
+            return await this.#retryTaskWithError(
+              task,
+              `Fetch failed with status ${response.status}`,
+              retryAt
+            );
+          }
+
+          // See if there is a json body
+          if (jsonBody) {
+            return await this.#resumeTaskWithError(task, jsonBody);
+          } else {
+            return await this.#resumeTaskWithError(task, {
+              message: `Fetch failed with status ${response.status}`,
+            });
+          }
+        }
+
+        return await this.#resumeTask(task, jsonBody);
+      }
+      default: {
+        await this.#resumeTaskWithError(task, {
+          message: `Unknown operation: ${task.operation}`,
+        });
+      }
+    }
+  }
+
+  #calculateRetryForResponse(
+    task: NonNullable<FoundTask>,
+    retry: FetchRetryOptions | undefined,
+    response: Response
+  ): Date | undefined {
+    if (!retry) {
+      return;
+    }
+
+    const strategy = this.#getRetryStrategyForStatusCode(
+      response.status,
+      retry
+    );
+
+    if (!strategy) {
+      return;
+    }
+
+    logger.debug("Calculating retry at for strategy", {
+      strategy,
+    });
+
+    switch (strategy.strategy) {
+      case "backoff": {
+        return calculateRetryAt(strategy, task.attempts.length - 1);
+      }
+      case "headers": {
+        const remaining = response.headers.get(strategy.remainingHeader);
+        const resetAt = response.headers.get(strategy.resetHeader);
+
+        if (
+          typeof remaining === "string" &&
+          typeof resetAt === "string" &&
+          remaining === "0"
+        ) {
+          return new Date(Number(resetAt) * 1000 + addJitterInMs());
+        }
+      }
+    }
+  }
+
+  #getRetryStrategyForStatusCode(
+    statusCode: number,
+    retry: FetchRetryOptions
+  ): FetchRetryStrategy | undefined {
+    const statusCodes = Object.keys(retry);
+
+    for (let i = 0; i < statusCodes.length; i++) {
+      const statusRange = statusCodes[i];
+      const strategy = retry[statusRange];
+
+      if (isStatusCodeInRange(statusCode, statusRange)) {
+        return strategy;
+      }
+    }
+
+    return;
+  }
+
+  async #retryTaskWithError(task: Task, error: string, retryAt: Date) {
+    await $transaction(this.#prismaClient, async (tx) => {
+      await tx.taskAttempt.updateMany({
+        where: {
+          taskId: task.id,
+          status: "PENDING",
+        },
+        data: {
+          status: "ERRORED",
+          error,
+        },
+      });
+
+      const currentMaxNumber = await tx.taskAttempt.aggregate({
+        where: { taskId: task.id },
+        _max: { number: true },
+      });
+
+      const newNumber = (currentMaxNumber._max.number ?? 0) + 1;
+
+      await tx.taskAttempt.create({
+        data: {
+          status: "PENDING",
+          taskId: task.id,
+          number: newNumber,
+          runAt: retryAt,
+        },
+      });
+
+      await workerQueue.enqueue(
+        "performTaskOperation",
+        {
+          id: task.id,
+        },
+        { tx, runAt: retryAt }
+      );
+    });
+  }
+
+  async #resumeTaskWithError(task: Task, output: any) {
+    await $transaction(this.#prismaClient, async (tx) => {
+      await tx.task.update({
+        where: { id: task.id },
+        data: {
+          status: "ERRORED",
+          completedAt: new Date(),
+          output,
+        },
+      });
+
+      await tx.taskAttempt.updateMany({
+        where: {
+          taskId: task.id,
+          status: "PENDING",
+        },
+        data: {
+          status: "ERRORED",
+          error: formatUnknownError(output),
+        },
+      });
+
+      await this.#resumeRunExecution(task, tx);
+    });
+  }
+
+  async #resumeTask(task: NonNullable<FoundTask>, output: any) {
+    await $transaction(this.#prismaClient, async (tx) => {
+      await tx.taskAttempt.updateMany({
+        where: {
+          taskId: task.id,
+          status: "PENDING",
+        },
+        data: {
+          status: "COMPLETED",
+        },
+      });
+
+      await tx.task.update({
+        where: { id: task.id },
+        data: {
+          status: "COMPLETED",
+          completedAt: new Date(),
+          output: output ? output : undefined,
+        },
+      });
+
+      await this.#resumeRunExecution(task, tx);
+    });
+  }
+
+  async #resumeRunExecution(task: Task, prisma: PrismaClientOrTransaction) {
+    await $transaction(prisma, async (tx) => {
+      const newJobExecution = await tx.jobRunExecution.create({
+        data: {
+          runId: task.runId,
+          reason: "EXECUTE_JOB",
+          status: "PENDING",
+          retryLimit: EXECUTE_JOB_RETRY_LIMIT,
+        },
+      });
+
+      const graphileJob = await workerQueue.enqueue(
+        "performRunExecution",
+        {
+          id: newJobExecution.id,
+        },
+        { tx }
+      );
+
+      await tx.jobRunExecution.update({
+        where: {
+          id: newJobExecution.id,
+        },
+        data: {
+          graphileJobId: graphileJob.id,
+        },
+      });
+    });
+  }
+}
+
+function normalizeHeaders(
+  headers: FetchRequestInit["headers"]
+): Record<string, string> {
+  if (!headers) {
+    return {};
+  }
+
+  return Object.fromEntries(
+    Object.entries(headers).map(([key, value]) => [
+      key,
+      typeof value === "string" ? value : hydrateRedactedString(value),
+    ])
+  );
+}
+
+function hydrateRedactedString(value: RedactString): string {
+  let result = "";
+
+  for (let i = 0; i < value.strings.length; i++) {
+    result += value.strings[i];
+    if (i < value.interpolations.length) {
+      result += value.interpolations[i];
+    }
+  }
+
+  return result;
+}
+
+async function findTask(prisma: PrismaClient, id: string) {
+  return prisma.task.findUnique({
+    where: { id },
+    include: {
+      attempts: true,
+    },
+  });
+}
+
+// Add a random number of ms between 0ms and 5000ms
+function addJitterInMs() {
+  return Math.floor(Math.random() * 5000);
+}
+
+/**
+ * Checks if a given status code falls within a given range.
+ * The range can be a single status code (e.g. "200"),
+ * a range of status codes (e.g. "500-599"),
+ * a range of status codes with a wildcard (e.g. "4xx" for any 4xx status code),
+ * or a list of status codes separated by commas (e.g. "401,403,404").
+ * Returns `true` if the status code falls within the range, and `false` otherwise.
+ */
+function isStatusCodeInRange(statusCode: number, statusRange: string): boolean {
+  if (statusRange === "all") {
+    return true;
+  }
+
+  if (statusRange.includes(",")) {
+    const statusCodes = statusRange.split(",").map((s) => s.trim());
+    return statusCodes.includes(statusCode.toString());
+  }
+
+  const [start, end] = statusRange.split("-");
+
+  if (end) {
+    return statusCode >= parseInt(start, 10) && statusCode <= parseInt(end, 10);
+  }
+
+  if (start.endsWith("xx")) {
+    const prefix = start.slice(0, -2);
+    const statusCodePrefix = Math.floor(statusCode / 100).toString();
+    return statusCodePrefix === prefix;
+  }
+
+  const statusCodeString = statusCode.toString();
+  const rangePrefix = start.slice(0, -1);
+
+  if (start.endsWith("x") && statusCodeString.startsWith(rangePrefix)) {
+    return true;
+  }
+
+  return statusCode === parseInt(start, 10);
+}

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

@@ -28,6 +28,7 @@ import { IntegrationConnectionCreatedService } from "./externalApis/integrationC
 import { sendEmail } from "./email.server";
 import { DeliverEmailSchema } from "@/../../packages/emails/src";
 import { PerformRunExecutionService } from "./runs/performRunExecution";
+import { PerformTaskOperationService } from "./tasks/performTaskOperation.server";
 
 const workerCatalog = {
   organizationCreated: z.object({ id: z.string() }),
@@ -50,6 +51,9 @@ const workerCatalog = {
   performRunExecution: z.object({
     id: z.string(),
   }),
+  performTaskOperation: z.object({
+    id: z.string(),
+  }),
   runFinished: z.object({ id: z.string() }),
   deliverHttpSourceRequest: z.object({ id: z.string() }),
   refreshOAuthToken: z.object({
@@ -242,6 +246,15 @@ function getWorkerQueue() {
           await service.call(payload.id);
         },
       },
+      performTaskOperation: {
+        queueName: (payload) => `tasks:${payload.id}`,
+        maxAttempts: 3,
+        handler: async (payload, job) => {
+          const service = new PerformTaskOperationService();
+
+          await service.call(payload.id);
+        },
+      },
       scheduleEmail: {
         queueName: "internal-queue",
         priority: 100,

apps/webapp/app/utils.ts

@@ -1,4 +1,7 @@
-import { ErrorWithStack } from "@/../../packages/internal/src";
+import {
+  ErrorWithStack,
+  ErrorWithStackSchema,
+} from "@trigger.dev/internal";
 import type { RouteMatch } from "@remix-run/react";
 import { useMatches } from "@remix-run/react";
 import humanizeDuration from "humanize-duration";
@@ -198,3 +201,16 @@ export function formatError(
 
   return formatError(error, "short") + "\n" + error.stack;
 }
+
+export function formatUnknownError(
+  error: unknown,
+  style: "short" | "long" = "short"
+): string {
+  const parsedError = ErrorWithStackSchema.safeParse(error);
+
+  if (parsedError.success) {
+    return formatError(parsedError.data, style);
+  }
+
+  return "Unknown error";
+}

apps/webapp/app/utils/json.ts

@@ -20,3 +20,8 @@ export function safeJsonZodParse<T>(
 
   return schema.safeParse(parsed);
 }
+
+export async function safeJsonFromResponse(response: Response) {
+  const json = await response.text();
+  return safeJsonParse(json);
+}

examples/nextjs-example/src/trigger.ts

@@ -69,7 +69,45 @@ const dynamicSchedule = new DynamicSchedule(client, {
 
 const enabled = true;
 
-const CHAT_MODELS = ["gpt-3.5-turbo"];
+new Job(client, {
+  id: "test-background-fetch-retry",
+  name: "Test background fetch retry",
+  version: "0.0.1",
+  enabled,
+  trigger: eventTrigger({
+    name: "test.background-fetch",
+    schema: z.object({
+      url: z.string(),
+      method: z.string().optional(),
+      headers: z.record(z.string()).optional(),
+      body: z.any().optional(),
+      retry: z.any().optional(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    return await io.backgroundFetch<any>(
+      "fetch",
+      payload.url,
+      {
+        method: payload.method ?? "GET",
+        headers: payload.headers,
+        body: payload.body ? JSON.stringify(payload.body) : undefined,
+      },
+      payload.retry
+    );
+  },
+});
+
+const CHAT_MODELS = [
+  "gpt-3.5-turbo",
+  "gpt-3.5-turbo-0301",
+  "gpt-3.5-turbo-0613",
+  "gpt-3.5-turbo-16k",
+  "gpt-3.5-turbo-16k-0613",
+  "gpt-4",
+  "gpt-4-0314",
+  "gpt-4-0613",
+];
 
 new Job(client, {
   id: "openai-test",
@@ -81,6 +119,7 @@ new Job(client, {
     schema: z.object({
       model: z.string(),
       prompt: z.string(),
+      background: z.boolean().optional(),
     }),
   }),
   integrations: {
@@ -88,6 +127,23 @@ new Job(client, {
   },
   run: async (payload, io, ctx) => {
     if (CHAT_MODELS.includes(payload.model)) {
+      if (payload.background) {
+        const completion = await io.openai.backgroundCreateChatCompletion(
+          "✨",
+          {
+            model: payload.model,
+            messages: [
+              {
+                role: "user",
+                content: payload.prompt,
+              },
+            ],
+          }
+        );
+
+        return completion;
+      }
+
       const completion = await io.openai.createChatCompletion("✨", {
         model: payload.model,
         messages: [
@@ -101,6 +157,15 @@ new Job(client, {
       return completion;
     }
 
+    if (payload.background) {
+      const completion = await io.openai.backgroundCreateCompletion("✨", {
+        model: payload.model,
+        prompt: payload.prompt,
+      });
+
+      return completion;
+    }
+
     const completion = await io.openai.createCompletion("✨", {
       model: payload.model,
       prompt: payload.prompt,

integrations/github/src/index.ts

@@ -43,7 +43,7 @@ export class Github
   _orgTrigger: ReturnType<typeof createOrgTrigger>;
 
   constructor(private options: GithubIntegrationOptions) {
-    this.client = createConnectionFromOptions(options);
+    this.client = createClientFromOptions(options);
     this._repoSource = createRepoEventSource(this);
     this._orgSource = createOrgEventSource(this);
     this._repoTrigger = createRepoTrigger(this._repoSource);
@@ -73,7 +73,7 @@ export class Github
   }
 }
 
-function createConnectionFromOptions(
+function createClientFromOptions(
   options: GithubIntegrationOptions
 ): IntegrationClient<Octokit, typeof tasks> {
   if (options.token) {
@@ -89,6 +89,7 @@ function createConnectionFromOptions(
       usesLocalAuth: true,
       client,
       tasks,
+      auth: options.token,
     };
   }
 

integrations/openai/src/index.ts

@@ -1,16 +1,18 @@
 import { OpenAIApi, Configuration } from "openai";
 import type { IntegrationClient, TriggerIntegration } from "@trigger.dev/sdk";
-import { createChatCompletion, createCompletion } from "./tasks";
+import {
+  createChatCompletion,
+  createCompletion,
+  backgroundCreateCompletion,
+  backgroundCreateChatCompletion,
+} from "./tasks";
+import { OpenAIIntegrationOptions } from "./types";
 
 const tasks = {
   createCompletion,
   createChatCompletion,
-};
-
-export type OpenAIIntegrationOptions = {
-  id: string;
-  apiKey: string;
-  organization?: string;
+  backgroundCreateCompletion,
+  backgroundCreateChatCompletion,
 };
 
 export class OpenAI
@@ -28,6 +30,10 @@ export class OpenAI
           organization: options.organization,
         })
       ),
+      auth: {
+        apiKey: options.apiKey,
+        organization: options.organization,
+      },
     };
   }
 

integrations/openai/src/tasks.ts

@@ -1,26 +1,17 @@
 import type { AuthenticatedTask } from "@trigger.dev/sdk";
-import { CreateChatCompletionRequest, OpenAIApi } from "openai";
+import {
+  CreateChatCompletionRequest,
+  CreateCompletionRequest,
+  OpenAIApi,
+} from "openai";
+import { OpenAIIntegrationAuth } from "./types";
+import { redactString } from "@trigger.dev/sdk";
 
 type OpenAIClientType = InstanceType<typeof OpenAIApi>;
 
 export const createCompletion: AuthenticatedTask<
   OpenAIClientType,
-  {
-    model: string;
-    prompt: string | string[];
-    suffix?: string;
-    max_tokens?: number;
-    temperature?: number;
-    top_p?: number;
-    n?: number;
-    logprobs?: number;
-    echo?: boolean;
-    stop?: string | string[];
-    presence_penalty?: number;
-    frequency_penalty?: number;
-    best_of?: number;
-    user?: string;
-  },
+  CreateCompletionRequest,
   Awaited<ReturnType<OpenAIClientType["createCompletion"]>>["data"]
 > = {
   run: async (params, client) => {
@@ -41,17 +32,119 @@ export const createCompletion: AuthenticatedTask<
   },
 };
 
-export const createChatCompletion: AuthenticatedTask<
+type CreateCompletionResponseData = Awaited<
+  ReturnType<OpenAIClientType["createCompletion"]>
+>["data"];
+
+export const backgroundCreateCompletion: AuthenticatedTask<
   OpenAIClientType,
-  CreateChatCompletionRequest,
-  Awaited<ReturnType<OpenAIClientType["createChatCompletion"]>>["data"]
+  CreateCompletionRequest,
+  CreateCompletionResponseData,
+  OpenAIIntegrationAuth
 > = {
-  run: async (params, client) => {
-    return client.createChatCompletion(params).then((res) => res.data);
+  run: async (params, client, task, io, auth) => {
+    return io.backgroundFetch<CreateCompletionResponseData>(
+      "background",
+      "https://api.openai.com/v1/completions",
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          Authorization: redactString`Bearer ${auth.apiKey}`,
+          ...(auth.organization
+            ? { "OpenAI-Organization": auth.organization }
+            : {}),
+        },
+        body: JSON.stringify(params),
+      }
+    );
   },
   init: (params) => {
     return {
-      name: "Completion",
+      name: "Background Completion",
+      params,
+      icon: "openai",
+      properties: [
+        {
+          label: "model",
+          text: params.model,
+        },
+      ],
+    };
+  },
+};
+
+type CreateChatCompetionResponseData = Awaited<
+  ReturnType<OpenAIClientType["createChatCompletion"]>
+>["data"];
+
+export const createChatCompletion: AuthenticatedTask<
+  OpenAIClientType,
+  CreateChatCompletionRequest,
+  CreateChatCompetionResponseData
+> = {
+  run: async (params, client) => {
+    return client.createChatCompletion(params).then((res) => res.data);
+  },
+  init: (params) => {
+    return {
+      name: "Chat Completion",
+      params,
+      icon: "openai",
+      properties: [
+        {
+          label: "model",
+          text: params.model,
+        },
+      ],
+    };
+  },
+};
+
+export const backgroundCreateChatCompletion: AuthenticatedTask<
+  OpenAIClientType,
+  CreateChatCompletionRequest,
+  CreateChatCompetionResponseData,
+  OpenAIIntegrationAuth
+> = {
+  run: async (params, client, task, io, auth) => {
+    return io.backgroundFetch<CreateChatCompetionResponseData>(
+      "background",
+      "https://api.openai.com/v1/chat/completions",
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+          Authorization: redactString`Bearer ${auth.apiKey}`,
+          ...(auth.organization
+            ? { "OpenAI-Organization": auth.organization }
+            : {}),
+        },
+        body: JSON.stringify(params),
+      },
+      {
+        "500-599": {
+          strategy: "backoff",
+          limit: 5,
+          minTimeoutInMs: 1000,
+          maxTimeoutInMs: 30000,
+          factor: 1.8,
+          randomize: true,
+        },
+        "429": {
+          strategy: "backoff",
+          limit: 10,
+          minTimeoutInMs: 1000,
+          maxTimeoutInMs: 60000,
+          factor: 2,
+          randomize: true,
+        },
+      }
+    );
+  },
+  init: (params) => {
+    return {
+      name: "Background Chat Completion",
       params,
       icon: "openai",
       properties: [

integrations/openai/src/types.ts

@@ -0,0 +1,7 @@
+export type OpenAIIntegrationOptions = {
+  id: string;
+  apiKey: string;
+  organization?: string;
+};
+
+export type OpenAIIntegrationAuth = Omit<OpenAIIntegrationOptions, "id">;

packages/database/prisma/migrations/20230619100936_add_operation_to_task/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Task" ADD COLUMN     "operation" TEXT;

packages/database/prisma/schema.prisma

@@ -733,6 +733,7 @@ model Task {
   error       String?
   redact      Json?
   style       Json?
+  operation   String?
 
   startedAt   DateTime?
   completedAt DateTime?
@@ -989,4 +990,4 @@ model ApiIntegrationVote {
   updatedAt DateTime @updatedAt
 
   @@unique([apiIdentifier, userId])
-}
+}

packages/internal/src/index.ts

@@ -2,3 +2,4 @@ export * from "./logger";
 export * from "./schemas";
 export * from "./types";
 export * from "./utils";
+export * from "./retry";

packages/internal/src/logger.ts

@@ -8,23 +8,31 @@ export class Logger {
   #name: string;
   readonly #level: number;
   #filteredKeys: string[] = [];
+  #jsonReplacer?: (key: string, value: unknown) => unknown;
 
   constructor(
     name: string,
     level: LogLevel = "info",
-    filteredKeys: string[] = []
+    filteredKeys: string[] = [],
+    jsonReplacer?: (key: string, value: unknown) => unknown
   ) {
     this.#name = name;
     this.#level = logLevels.indexOf(
       (process.env.TRIGGER_LOG_LEVEL ?? level) as LogLevel
     );
     this.#filteredKeys = filteredKeys;
+    this.#jsonReplacer = 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[]) {
-    return new Logger(this.#name, logLevels[this.#level], keys);
+    return new Logger(
+      this.#name,
+      logLevels[this.#level],
+      keys,
+      this.#jsonReplacer
+    );
   }
 
   log(...args: any[]) {
@@ -64,10 +72,26 @@ export class Logger {
       ),
     };
 
-    console.debug(JSON.stringify(structuredLog, bigIntReplacer));
+    console.debug(
+      JSON.stringify(structuredLog, createReplacer(this.#jsonReplacer))
+    );
   }
 }
 
+function createReplacer(replacer?: (key: string, value: unknown) => unknown) {
+  return (key: string, value: unknown) => {
+    if (typeof value === "bigint") {
+      return value.toString();
+    }
+
+    if (replacer) {
+      return replacer(key, value);
+    }
+
+    return value;
+  };
+}
+
 // Replacer function for JSON.stringify that converts BigInts to strings
 function bigIntReplacer(_key: string, value: unknown) {
   if (typeof value === "bigint") {

packages/internal/src/retry.ts

@@ -0,0 +1,37 @@
+import { RetryOptions } from "./schemas";
+
+const DEFAULT_RETRY_OPTIONS = {
+  limit: 5,
+  factor: 1.8,
+  minTimeoutInMs: 1000,
+  maxTimeoutInMs: 60000,
+  randomize: true,
+} satisfies RetryOptions;
+
+export function calculateRetryAt(
+  retryOptions: RetryOptions,
+  attempts: number
+): Date | undefined {
+  const options = {
+    ...DEFAULT_RETRY_OPTIONS,
+    ...retryOptions,
+  };
+
+  const retryCount = attempts + 1;
+
+  if (retryCount >= options.limit) {
+    return;
+  }
+
+  const random = options.randomize ? Math.random() + 1 : 1;
+
+  let timeoutInMs = Math.round(
+    random *
+      Math.max(options.minTimeoutInMs, 1) *
+      Math.pow(options.factor, Math.max(attempts - 1, 0))
+  );
+
+  timeoutInMs = Math.min(timeoutInMs, options.maxTimeoutInMs);
+
+  return new Date(Date.now() + timeoutInMs);
+}

packages/internal/src/schemas/api.ts

@@ -376,13 +376,13 @@ export const CreateRunResponseBodySchema = z.discriminatedUnion("ok", [
 
 export type CreateRunResponseBody = z.infer<typeof CreateRunResponseBodySchema>;
 
-export const SecureStringSchema = z.object({
-  __secureString: z.literal(true),
+export const RedactStringSchema = z.object({
+  __redactedString: z.literal(true),
   strings: z.array(z.string()),
   interpolations: z.array(z.string()),
 });
 
-export type SecureString = z.infer<typeof SecureStringSchema>;
+export type RedactString = z.infer<typeof RedactStringSchema>;
 
 export const LogMessageSchema = z.object({
   level: z.enum(["DEBUG", "INFO", "WARN", "ERROR"]),
@@ -414,6 +414,7 @@ export const RunTaskOptionsSchema = z.object({
   icon: z.string().optional(),
   displayKey: z.string().optional(),
   noop: z.boolean().default(false),
+  operation: z.enum(["fetch"]).optional(),
   delayUntil: z.coerce.date().optional(),
   description: z.string().optional(),
   properties: z.array(DisplayPropertySchema).optional(),

packages/internal/src/schemas/fetch.ts

@@ -0,0 +1,48 @@
+import { z } from "zod";
+import { RedactStringSchema, RetryOptionsSchema } from "./api";
+
+export const FetchRetryHeadersStrategySchema = z.object({
+  strategy: z.literal("headers"),
+  limitHeader: z.string(),
+  remainingHeader: z.string(),
+  resetHeader: z.string(),
+});
+
+export type FetchRetryHeadersStrategy = z.infer<
+  typeof FetchRetryHeadersStrategySchema
+>;
+
+export const FetchRetryBackoffStrategySchema = RetryOptionsSchema.extend({
+  strategy: z.literal("backoff"),
+});
+
+export type FetchRetryBackoffStrategy = z.infer<
+  typeof FetchRetryBackoffStrategySchema
+>;
+
+export const FetchRetryStrategySchema = z.discriminatedUnion("strategy", [
+  FetchRetryHeadersStrategySchema,
+  FetchRetryBackoffStrategySchema,
+]);
+
+export type FetchRetryStrategy = z.infer<typeof FetchRetryStrategySchema>;
+
+export const FetchRequestInitSchema = z.object({
+  method: z.string().optional(),
+  headers: z.record(z.union([z.string(), RedactStringSchema])).optional(),
+  body: z.union([z.string(), z.instanceof(ArrayBuffer)]).optional(),
+});
+
+export type FetchRequestInit = z.infer<typeof FetchRequestInitSchema>;
+
+export const FetchRetryOptionsSchema = z.record(FetchRetryStrategySchema);
+
+export type FetchRetryOptions = z.infer<typeof FetchRetryOptionsSchema>;
+
+export const FetchOperationSchema = z.object({
+  url: z.string(),
+  requestInit: FetchRequestInitSchema.optional(),
+  retry: z.record(FetchRetryStrategySchema).optional(),
+});
+
+export type FetchOperation = z.infer<typeof FetchOperationSchema>;

packages/internal/src/schemas/index.ts

@@ -8,3 +8,4 @@ export * from "./properties";
 export * from "./integrations";
 export * from "./schedules";
 export * from "./notifications";
+export * from "./fetch";

packages/internal/src/schemas/tasks.ts

@@ -28,6 +28,7 @@ export const TaskSchema = z.object({
   error: z.string().optional().nullable(),
   parentId: z.string().optional().nullable(),
   style: StyleSchema.optional().nullable(),
+  operation: z.string().optional().nullable(),
 });
 
 export const ServerTaskSchema = TaskSchema.extend({

packages/trigger-sdk/src/index.ts

@@ -10,7 +10,7 @@ export * from "./io";
 export * from "./types";
 
 import { ServerTask } from "@trigger.dev/internal";
-import { SecureString } from "./types";
+import { RedactString } from "./types";
 export { isTriggerError } from "./errors";
 
 export type { NormalizedRequest, EventFilter } from "@trigger.dev/internal";
@@ -18,22 +18,22 @@ export type { NormalizedRequest, EventFilter } from "@trigger.dev/internal";
 export type Task = ServerTask;
 
 /*
- * This function is used to create a secure string that can be used in the headers of a fetch request.
+ * This function is used to create a redacted string that can be used in the headers of a fetch request.
  * It is used to prevent the string from being logged in trigger.dev.
  * You can use it like this:
  *
  * await ctx.fetch("https://example.com", {
  *  headers: {
- *    Authorization: secureString`Bearer ${ACCESS_TOKEN}`,
+ *    Authorization: redactString`Bearer ${ACCESS_TOKEN}`,
  *  },
  * })
  */
-export function secureString(
+export function redactString(
   strings: TemplateStringsArray,
   ...interpolations: string[]
-): SecureString {
+): RedactString {
   return {
-    __secureString: true,
+    __redactedString: true,
     strings: strings.raw as string[],
     interpolations,
   };

packages/trigger-sdk/src/integrations.ts

@@ -21,12 +21,13 @@ export interface TriggerIntegration<
 
 export type IntegrationClient<
   TClient,
-  TTasks extends Record<string, AuthenticatedTask<TClient, any, any>>
+  TTasks extends Record<string, AuthenticatedTask<TClient, any, any, any>>
 > =
   | {
       usesLocalAuth: true;
       client: TClient;
       tasks?: TTasks;
+      auth: any;
     }
   | {
       usesLocalAuth: false;
@@ -34,12 +35,18 @@ export type IntegrationClient<
       tasks?: TTasks;
     };
 
-export type AuthenticatedTask<TClient, TParams, TResult> = {
+export type AuthenticatedTask<
+  TClient,
+  TParams,
+  TResult,
+  TAuth = ConnectionAuth
+> = {
   run: (
     params: TParams,
     client: TClient,
     task: ServerTask,
-    io: IO
+    io: IO,
+    auth: TAuth
   ) => Promise<TResult>;
   init: (params: TParams) => RunTaskOptions;
   onError?: (
@@ -63,13 +70,14 @@ export function authenticatedTask<TClient, TParams, TResult>(options: {
 type ExtractRunFunction<T> = T extends AuthenticatedTask<
   any,
   infer TParams,
-  infer TResult
+  infer TResult,
+  infer TAuth
 >
   ? (key: string, params: TParams) => Promise<TResult>
   : never;
 
 type ExtractTasks<
-  TTasks extends Record<string, AuthenticatedTask<any, any, any>>
+  TTasks extends Record<string, AuthenticatedTask<any, any, any, any>>
 > = {
   [key in keyof TTasks]: ExtractRunFunction<TTasks[key]>;
 };

packages/trigger-sdk/src/io.ts

@@ -3,6 +3,8 @@ import {
   ConnectionAuth,
   CronOptions,
   ErrorWithStackSchema,
+  FetchRequestInit,
+  FetchRetryOptions,
   IntervalOptions,
   LogLevel,
   Logger,
@@ -123,6 +125,44 @@ export class IO {
     );
   }
 
+  async backgroundFetch<TResponseData>(
+    key: string | any[],
+    url: string,
+    requestInit?: FetchRequestInit,
+    retry?: FetchRetryOptions
+  ): Promise<TResponseData> {
+    const urlObject = new URL(url);
+
+    return (await this.runTask(
+      key,
+      {
+        name: `fetch ${urlObject.hostname}${urlObject.pathname}`,
+        params: { url, requestInit, retry },
+        operation: "fetch",
+        icon: "background",
+        noop: false,
+        properties: [
+          {
+            label: "url",
+            text: url,
+            url,
+          },
+          {
+            label: "method",
+            text: requestInit?.method ?? "GET",
+          },
+          {
+            label: "background",
+            text: "true",
+          },
+        ],
+      },
+      async (task) => {
+        return task.output;
+      }
+    )) as TResponseData;
+  }
+
   async sendEvent(
     key: string | any[],
     event: SendEvent,
@@ -350,7 +390,7 @@ export class IO {
       error: unknown,
       task: IOTask,
       io: IO
-    ) => { retryAt: Date; error?: Error } | undefined | void
+    ) => { retryAt: Date; error?: Error; jitter?: number } | undefined | void
   ): Promise<TResult> {
     const parentId = this._taskStorage.getStore()?.taskId;
 
@@ -414,6 +454,15 @@ export class IO {
       throw new ResumeWithTaskError(task);
     }
 
+    if (task.status === "RUNNING" && typeof task.operation === "string") {
+      this._logger.debug("Task running operation", {
+        idempotencyKey,
+        task,
+      });
+
+      throw new ResumeWithTaskError(task);
+    }
+
     const executeTask = async () => {
       try {
         const result = await callback(task, this);
@@ -429,6 +478,10 @@ export class IO {
 
         return result;
       } catch (error) {
+        if (isTriggerError(error)) {
+          throw error;
+        }
+
         if (onError) {
           const onErrorResult = onError(error, task, this);
 

packages/trigger-sdk/src/ioWithIntegrations.ts

@@ -23,18 +23,20 @@ export function createIOWithIntegrations<
 
   const connections = Object.entries(integrations).reduce(
     (acc, [connectionKey, integration]) => {
-      const connection = auths?.[connectionKey];
-      const client =
-        "client" in integration.client
-          ? integration.client.client
-          : connection
-          ? integration.client.clientFactory?.(connection)
-          : undefined;
+      let auth = auths?.[connectionKey];
+
+      const client = integration.client.usesLocalAuth
+        ? integration.client.client
+        : auth
+        ? integration.client.clientFactory?.(auth)
+        : undefined;
 
       if (!client) {
         return acc;
       }
 
+      auth = integration.client.usesLocalAuth ? integration.client.auth : auth;
+
       const ioConnection = {
         client,
       } as any;
@@ -42,7 +44,7 @@ export function createIOWithIntegrations<
       if (integration.client.tasks) {
         const tasks: Record<
           string,
-          AuthenticatedTask<any, any, any>
+          AuthenticatedTask<any, any, any, any>
         > = integration.client.tasks;
 
         Object.keys(tasks).forEach((taskName) => {
@@ -59,7 +61,7 @@ export function createIOWithIntegrations<
               key,
               options,
               async (ioTask) => {
-                return authenticatedTask.run(params, client, ioTask, io);
+                return authenticatedTask.run(params, client, ioTask, io, auth);
               },
               authenticatedTask.onError
             );

packages/trigger-sdk/src/retry.ts

@@ -1,37 +1,5 @@
 import type { RetryOptions } from "@trigger.dev/internal";
+import { calculateRetryAt } from "@trigger.dev/internal";
 
-const DEFAULT_RETRY_OPTIONS = {
-  limit: 5,
-  factor: 1.8,
-  minTimeoutInMs: 1000,
-  maxTimeoutInMs: 60000,
-  randomize: true,
-} satisfies RetryOptions;
-
-export function calculateRetryAt(
-  retryOptions: RetryOptions,
-  attempts: number
-): Date | undefined {
-  const options = {
-    ...DEFAULT_RETRY_OPTIONS,
-    ...retryOptions,
-  };
-
-  const retryCount = attempts + 1;
-
-  if (retryCount > options.limit) {
-    return;
-  }
-
-  const random = options.randomize ? Math.random() + 1 : 1;
-
-  let timeoutInMs = Math.round(
-    random *
-      Math.max(options.minTimeoutInMs, 1) *
-      Math.pow(options.factor, Math.max(attempts - 1, 0))
-  );
-
-  timeoutInMs = Math.min(timeoutInMs, options.maxTimeoutInMs);
-
-  return new Date(Date.now() + timeoutInMs);
-}
+export { calculateRetryAt };
+export type { RetryOptions };

packages/trigger-sdk/src/types.ts

@@ -2,14 +2,14 @@ import type {
   EventFilter,
   Logger,
   RuntimeEnvironmentType,
-  SecureString,
+  RedactString,
   TriggerMetadata,
 } from "@trigger.dev/internal";
 import { DisplayProperty } from "@trigger.dev/internal";
 import { Job } from "./job";
 import { TriggerClient } from "./triggerClient";
 
-export type { SecureString, Logger };
+export type { RedactString, Logger };
 
 export interface TriggerContext {
   job: { id: string; version: string };