Integrate with the new ApiConnection system to authenticate with Slack using OAuth

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

@@ -0,0 +1,42 @@
+import type { JobConnection } from ".prisma/client";
+import type { ConnectionAuth } from "@trigger.dev/internal";
+import type { ApiConnectionWithSecretReference } from "~/services/externalApis/apiAuthenticationRepository.server";
+import { apiConnectionRepository } from "~/services/externalApis/apiAuthenticationRepository.server";
+
+export type JobConnectionWithApiConnection = JobConnection & {
+  apiConnection: ApiConnectionWithSecretReference | null;
+};
+
+export async function resolveJobConnections(
+  connections: Array<JobConnectionWithApiConnection>
+): Promise<Record<string, ConnectionAuth>> {
+  const result: Record<string, ConnectionAuth> = {};
+
+  for (const connection of connections) {
+    if (!connection.apiConnection) {
+      continue;
+    }
+
+    const response = await apiConnectionRepository.getCredentials(
+      connection.apiConnection
+    );
+
+    if (!response) {
+      continue;
+    }
+
+    if (result[connection.key]) {
+      throw new Error(
+        `Duplicate connection key ${connection.key} in job instance ${connection.jobInstanceId}`
+      );
+    }
+
+    result[connection.key] = {
+      type: "oauth2",
+      scopes: response.scopes,
+      accessToken: response.accessToken,
+    };
+  }
+
+  return result;
+}

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

@@ -1,20 +1,18 @@
-import {
+import type {
   ApiEventLog,
-  CachedTask,
   ConnectionAuth,
   ExecuteJobBody,
   HttpSourceRequest,
-  HttpSourceResponseSchema,
   PrepareForJobExecutionBody,
-  PrepareForJobExecutionResponseSchema,
-  ServerTask,
 } from "@trigger.dev/internal";
-import { ErrorWithStackSchema } from "@trigger.dev/internal";
 import {
   DeliverEventResponseSchema,
+  ErrorWithStackSchema,
   ExecuteJobResponseSchema,
   GetJobsResponseSchema,
+  HttpSourceResponseSchema,
   PongResponseSchema,
+  PrepareForJobExecutionResponseSchema,
 } from "@trigger.dev/internal";
 import { logger } from "./logger";
 

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

@@ -1,9 +1,15 @@
-import type { Endpoint, Job, JobConnection, JobInstance } from ".prisma/client";
+import type {
+  Endpoint,
+  Job,
+  JobConnection,
+  JobInstance,
+  ApiConnection,
+} from ".prisma/client";
 import type { ApiJob, ConnectionMetadata } from "@trigger.dev/internal";
 import semver from "semver";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
-import { AuthenticatedEnvironment } from "../apiAuth.server";
+import type { AuthenticatedEnvironment } from "../apiAuth.server";
 import { ClientApi } from "../clientApi.server";
 import { allConnectionsReady } from "../jobs/utils.server";
 import { logger } from "../logger";
@@ -83,7 +89,11 @@ export class EndpointRegisteredService {
         title: apiJob.name,
       },
       include: {
-        connections: true,
+        connections: {
+          include: {
+            apiConnection: true,
+          },
+        },
         instances: {
           where: {
             endpointId: endpoint.id,
@@ -152,7 +162,11 @@ export class EndpointRegisteredService {
         trigger: apiJob.trigger,
       },
       include: {
-        connections: true,
+        connections: {
+          include: {
+            apiConnection: true,
+          },
+        },
       },
     });
 
@@ -165,7 +179,8 @@ export class EndpointRegisteredService {
           jobInstance,
           "__trigger",
           apiJob.trigger.connection.metadata,
-          apiJob.trigger.connection.usesLocalAuth
+          apiJob.trigger.connection.usesLocalAuth,
+          apiJob.trigger.connection.id
         )
       );
     }
@@ -178,7 +193,8 @@ export class EndpointRegisteredService {
           jobInstance,
           connection.key,
           connection.metadata,
-          connection.usesLocalAuth
+          connection.usesLocalAuth,
+          connection.id
         )
       );
     }
@@ -264,31 +280,152 @@ export class EndpointRegisteredService {
   }
 
   async #upsertJobConnection(
-    job: Job & { connections: JobConnection[] },
-    jobInstance: JobInstance & { connections: JobConnection[] },
+    job: Job & {
+      connections: Array<
+        JobConnection & { apiConnection: ApiConnection | null }
+      >;
+    },
+    jobInstance: JobInstance & {
+      connections: Array<
+        JobConnection & { apiConnection: ApiConnection | null }
+      >;
+    },
     key: string,
     metadata: ConnectionMetadata,
-    usesLocalAuth: boolean
+    usesLocalAuth: boolean,
+    id?: string
   ): Promise<JobConnection> {
+    if (usesLocalAuth) {
+      return this.#upsertLocalAuthConnection(job, jobInstance, key, metadata);
+    }
+
+    if (!id) {
+      logger.debug("Missing connection id", {
+        key,
+        metadata,
+        usesLocalAuth,
+        job,
+      });
+
+      throw new Error("Missing connection id");
+    }
+
+    const apiConnection =
+      await this.#prismaClient.apiConnection.findUniqueOrThrow({
+        where: {
+          organizationId_slug: {
+            organizationId: job.organizationId,
+            slug: id,
+          },
+        },
+      });
+
     // Find existing connection in the job instance
     const existingInstanceConnection = jobInstance.connections.find(
       (connection) => connection.key === key
     );
 
     if (existingInstanceConnection) {
-      if (usesLocalAuth && existingInstanceConnection.apiConnectionId) {
-        // If the connection uses local auth, we need to delete the existing ApiConnection
-        return await this.#prismaClient.jobConnection.update({
-          where: {
-            id: existingInstanceConnection.id,
-          },
-          data: {
-            apiConnectionId: null,
-            usesLocalAuth: true,
-          },
-        });
-      }
+      return await this.#prismaClient.jobConnection.update({
+        where: {
+          id: existingInstanceConnection.id,
+        },
+        data: {
+          apiConnectionId: apiConnection.id,
+          usesLocalAuth: false,
+        },
+      });
+    }
 
+    // Find existing connection in the job
+    const existingJobConnection = job.connections.find(
+      (connection) => connection.key === key
+    );
+
+    if (existingJobConnection) {
+      return this.#prismaClient.jobConnection.create({
+        data: {
+          jobInstance: {
+            connect: {
+              id: jobInstance.id,
+            },
+          },
+          job: {
+            connect: {
+              id: job.id,
+            },
+          },
+          key,
+          connectionMetadata: existingJobConnection.connectionMetadata ?? {},
+          apiConnection: {
+            connect: {
+              id: apiConnection.id,
+            },
+          },
+          usesLocalAuth: false,
+        },
+      });
+    }
+
+    return this.#prismaClient.jobConnection.create({
+      data: {
+        jobInstance: {
+          connect: {
+            id: jobInstance.id,
+          },
+        },
+        job: {
+          connect: {
+            id: job.id,
+          },
+        },
+        key,
+        connectionMetadata: metadata,
+        apiConnection: {
+          connect: {
+            id: apiConnection.id,
+          },
+        },
+        usesLocalAuth,
+      },
+    });
+  }
+
+  async #upsertLocalAuthConnection(
+    job: Job & {
+      connections: Array<
+        JobConnection & { apiConnection: ApiConnection | null }
+      >;
+    },
+    jobInstance: JobInstance & {
+      connections: Array<
+        JobConnection & { apiConnection: ApiConnection | null }
+      >;
+    },
+    key: string,
+    metadata: ConnectionMetadata
+  ): Promise<JobConnection> {
+    // Find existing connection in the job instance
+    const existingInstanceConnection = jobInstance.connections.find(
+      (connection) => connection.key === key
+    );
+
+    if (
+      existingInstanceConnection &&
+      existingInstanceConnection.apiConnectionId
+    ) {
+      return await this.#prismaClient.jobConnection.update({
+        where: {
+          id: existingInstanceConnection.id,
+        },
+        data: {
+          apiConnectionId: null,
+          usesLocalAuth: true,
+        },
+      });
+    }
+
+    if (existingInstanceConnection) {
       return existingInstanceConnection;
     }
 
@@ -312,28 +449,11 @@ export class EndpointRegisteredService {
           },
           key,
           connectionMetadata: existingJobConnection.connectionMetadata ?? {},
-          apiConnection:
-            existingJobConnection.apiConnectionId && !usesLocalAuth
-              ? {
-                  connect: {
-                    id: existingJobConnection.apiConnectionId,
-                  },
-                }
-              : undefined,
-          usesLocalAuth,
+          usesLocalAuth: true,
         },
       });
     }
 
-    // Find existing ApiConnection in the org
-    const existingApiConnection =
-      await this.#prismaClient.apiConnection.findFirst({
-        where: {
-          apiIdentifier: metadata.id,
-          organizationId: job.organizationId,
-        },
-      });
-
     return this.#prismaClient.jobConnection.create({
       data: {
         jobInstance: {
@@ -348,15 +468,7 @@ export class EndpointRegisteredService {
         },
         key,
         connectionMetadata: metadata,
-        apiConnection:
-          existingApiConnection && !usesLocalAuth
-            ? {
-                connect: {
-                  id: existingApiConnection.id,
-                },
-              }
-            : undefined,
-        usesLocalAuth,
+        usesLocalAuth: true,
       },
     });
   }

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

@@ -1,6 +1,7 @@
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
 import { ClientApi } from "../clientApi.server";
+import { resolveJobConnections } from "~/models/jobConnection.server";
 
 export class PrepareJobInstanceService {
   #prismaClient: PrismaClient;
@@ -15,7 +16,15 @@ export class PrepareJobInstanceService {
         id,
       },
       include: {
-        connections: true,
+        connections: {
+          include: {
+            apiConnection: {
+              include: {
+                dataReference: true,
+              },
+            },
+          },
+        },
         job: true,
         endpoint: {
           include: {
@@ -33,7 +42,7 @@ export class PrepareJobInstanceService {
     const response = await client.prepareForJobExecution({
       id: jobInstance.job.slug,
       version: jobInstance.version,
-      connections: {}, // TODO: connections
+      connections: await resolveJobConnections(jobInstance.connections),
     });
 
     if (!response.ok) {

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

@@ -1,5 +1,6 @@
 import type { EventLog, JobEventRule } from ".prisma/client";
-import { EventFilter, EventFilterSchema } from "@trigger.dev/internal";
+import type { EventFilter } from "@trigger.dev/internal";
+import { EventFilterSchema } from "@trigger.dev/internal";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
 import { CreateRunService } from "../runs/createRun.server";

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

@@ -1,72 +0,0 @@
-import Pulsar, {
-  AuthenticationOauth2,
-  AuthenticationToken,
-} from "pulsar-client";
-import { env } from "~/env.server";
-
-export type ClientOptions = {
-  serviceUrl?: string;
-  token?: string;
-  operationTimeoutSeconds?: number;
-  ioThreads?: number;
-  messageListenerThreads?: number;
-  concurrentLookupRequest?: number;
-  useTls?: boolean;
-  tlsTrustCertsFilePath?: string;
-  tlsValidateHostname?: boolean;
-  tlsAllowInsecureConnection?: boolean;
-  statsIntervalInSeconds?: number;
-};
-
-export { Pulsar };
-
-export type PulsarMessage = Pulsar.Message;
-export type PulsarProducer = Pulsar.Producer;
-export type PulsarConsumer = Pulsar.Consumer;
-export type PulsarClient = Pulsar.Client;
-export type PulsarProducerConfig = Pulsar.ProducerConfig;
-export type PulsarConsumerConfig = Pulsar.ConsumerConfig;
-
-export function createPulsarClient(options?: ClientOptions): PulsarClient {
-  const opts = options || {};
-
-  const serviceUrl =
-    opts.serviceUrl || env.PULSAR_SERVICE_URL || "pulsar://localhost:6650";
-
-  const authentication = opts.token
-    ? new AuthenticationToken({ token: opts.token })
-    : oauth2AuthenticationFromEnv();
-
-  if (env.PULSAR_DEBUG) {
-    Pulsar.Client.setLogHandler((level, file, line, message) => {
-      console.log("[%s][%s:%d] %s", level, file, line, message);
-    });
-  }
-
-  console.log(`Connecting to pulsar instance at ${serviceUrl}...`);
-
-  return new Pulsar.Client({
-    ...opts,
-    serviceUrl,
-    authentication,
-  });
-}
-
-function oauth2AuthenticationFromEnv(): AuthenticationOauth2 | undefined {
-  const clientId = env.PULSAR_CLIENT_ID;
-  const clientSecret = env.PULSAR_CLIENT_SECRET;
-  const issuerUrl = env.PULSAR_ISSUER_URL;
-  const audience = env.PULSAR_AUDIENCE;
-
-  if (!clientId || !clientSecret || !issuerUrl || !audience) {
-    return undefined;
-  }
-
-  return new AuthenticationOauth2({
-    type: "sn_service_account",
-    client_id: clientId,
-    client_secret: clientSecret,
-    issuer_url: issuerUrl,
-    audience,
-  });
-}

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

@@ -1,7 +1,8 @@
-import type { ApiConnection, JobConnection } from ".prisma/client";
 import { ApiEventLogSchema } from "@trigger.dev/internal";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
+import type { JobConnectionWithApiConnection } from "~/models/jobConnection.server";
+import { resolveJobConnections } from "~/models/jobConnection.server";
 import { ClientApi, ClientApiError } from "../clientApi.server";
 import { workerQueue } from "../worker.server";
 
@@ -22,7 +23,11 @@ export class StartRunService {
             job: true,
             connections: {
               include: {
-                apiConnection: true,
+                apiConnection: {
+                  include: {
+                    dataReference: true,
+                  },
+                },
               },
               where: {
                 key: { not: "__trigger" },
@@ -37,9 +42,10 @@ export class StartRunService {
     });
 
     // If any of the connections are missing, we can't start the execution
-    const connections = run.jobInstance.connections.filter(
-      (c) => c.apiConnection != null || c.usesLocalAuth
-    ) as Array<JobConnection & { apiConnection?: ApiConnection }>;
+    const connections: Array<JobConnectionWithApiConnection> =
+      run.jobInstance.connections.filter(
+        (c) => c.apiConnection != null || c.usesLocalAuth
+      );
 
     const client = new ClientApi(
       run.environment.apiKey,
@@ -73,7 +79,7 @@ export class StartRunService {
           version: run.jobInstance.version,
           startedAt,
         },
-        connections: {}, // TODO: connections
+        connections: await resolveJobConnections(connections),
       });
 
       if (results.completed) {

examples/nextjs-example/src/pages/api/trigger.ts

@@ -5,10 +5,12 @@ import {
   TriggerClient,
 } from "@trigger.dev/sdk";
 import { github } from "@trigger.dev/github";
+import { slack } from "@trigger.dev/slack";
 import type { NextApiRequest, NextApiResponse } from "next";
 import { z } from "zod";
 
 const gh = github({ token: process.env.GITHUB_TOKEN! });
+const sl = slack({ id: "my-slack-new" });
 
 const client = new TriggerClient("nextjs", {
   apiKey: process.env.TRIGGER_API_KEY,
@@ -17,59 +19,6 @@ const client = new TriggerClient("nextjs", {
   logLevel: "debug",
 });
 
-// new Job({
-//   id: "comment-on-new-issues",
-//   name: "Comment on New GitHub issues",
-//   version: "0.1.1",
-//   logLevel: "debug",
-//   connections: {
-//     gh,
-//   },
-//   // issueEvent is a helper function that creates a trigger for a GitHub issue event webhook
-//   trigger: gh.onIssueOpened({
-//     repo: "ericallam/basic-starter-100k",
-//   }),
-
-//   run: async (event, io, ctx) => {
-//     // event is a GitHubIssueEvent
-//     const comment = await io.gh.createIssueComment("📝", {
-//       repo: event.repository.full_name,
-//       issueNumber: event.issue.number,
-//       body: "Hello from Trigger!",
-//     });
-
-//     // const token = await ctx.auth.gh
-
-//     const reaction = await io.runTask(
-//       "Add 🚀",
-//       {
-//         icon: "github",
-//         name: "addReaction",
-//         elements: [
-//           { label: "reaction", text: "🚀" },
-//           {
-//             label: "issue",
-//             text: event.issue.title,
-//             url: event.issue.html_url,
-//           },
-//         ],
-//         delayUntil: new Date(Date.now() + 1000 * 30), // 30 seconds from now
-//       },
-//       async (task) =>
-//         io.gh.client.rest.reactions
-//           .createForIssueComment({
-//             owner: event.repository.owner.login,
-//             repo: event.repository.name,
-//             comment_id: comment.id,
-//             content: "rocket",
-//           })
-//           .then((res) => res.data)
-//     );
-
-//     return reaction;
-//   },
-// }).registerWith(client);
-
 new Job({
   id: "comment-on-new-issues",
   name: "Comment on New GitHub issues",
@@ -77,13 +26,17 @@ new Job({
   logLevel: "debug",
   connections: {
     gh,
+    sl,
   },
-  // issueEvent is a helper function that creates a trigger for a GitHub issue event webhook
-  trigger: gh.onIssueOpened({
+  trigger: gh.triggers.onIssueOpened({
     repo: "ericallam/basic-starter-100k",
   }),
-
   run: async (event, io, ctx) => {
+    await io.sl.postMessage("Slack 📝", {
+      text: `New Issue opened: ${event.issue.html_url}`,
+      channel: "C04GWUTDC3W",
+    });
+
     await io.runTask(
       "Comment on Issue with a reaction",
       { name: "Parent Task" },

examples/nextjs-example/tsconfig.json

@@ -21,7 +21,9 @@
       "@trigger.dev/internal": ["../../packages/internal/src/index"],
       "@trigger.dev/internal/*": ["../../packages/internal/src/*"],
       "@trigger.dev/github": ["../../integrations/github/src/index"],
-      "@trigger.dev/github/*": ["../../integrations/github/src/*"]
+      "@trigger.dev/github/*": ["../../integrations/github/src/*"],
+      "@trigger.dev/slack": ["../../integrations/slack/src/index"],
+      "@trigger.dev/slack/*": ["../../integrations/slack/src/*"]
     }
   },
   "include": ["next-env.d.ts", "**/*.ts", "**/*.tsx"],

integrations/github/src/client.ts

@@ -1,17 +0,0 @@
-import { ClientFactory } from "@trigger.dev/sdk";
-import { Octokit } from "octokit";
-
-export const clientFactory: ClientFactory<InstanceType<typeof Octokit>> = (
-  auth
-) => {
-  if (auth.type === "basicAuth") {
-    throw new Error("Basic auth is not supported");
-  }
-
-  const token = auth.type === "apiKey" ? auth.apiKey : auth.accessToken;
-
-  return new Octokit({
-    auth: token,
-    baseUrl: (auth.additionalFields ?? {})["baseUrl"],
-  });
-};

integrations/github/src/index.ts

@@ -3,9 +3,9 @@ import {
   IssuesEvent,
   IssuesOpenedEvent,
 } from "@octokit/webhooks-types";
-import type { EventFilter } from "@trigger.dev/sdk";
+import type { Connection, EventFilter } from "@trigger.dev/sdk";
 import { ExternalSourceEventTrigger, Trigger } from "@trigger.dev/sdk/triggers";
-import { clientFactory as cf } from "./client";
+import { Octokit } from "octokit";
 import { metadata } from "./metadata";
 import { repositoryWebhookSource } from "./sources";
 import {
@@ -22,25 +22,13 @@ const tasks = {
   createIssueCommentWithReaction,
 };
 
-export const github = (options?: { token: string }) => {
-  const clientFactory = options?.token
-    ? () => cf({ type: "apiKey", apiKey: options.token })
-    : cf;
-
+function createTriggers(client: Octokit) {
   return {
-    metadata,
-    tasks,
-    usesLocalAuth: typeof options?.token === "string",
-    clientFactory,
-    onIssue: buildRepoWebhookTrigger<IssuesEvent>(
-      "On Issue",
-      "issues",
-      options
-    ),
+    onIssue: buildRepoWebhookTrigger<IssuesEvent>("On Issue", "issues", client),
     onIssueOpened: buildRepoWebhookTrigger<IssuesOpenedEvent>(
       "On Issue Opened",
       "issues",
-      options,
+      client,
       {
         action: ["opened"],
       }
@@ -48,15 +36,29 @@ export const github = (options?: { token: string }) => {
     onIssueComment: buildRepoWebhookTrigger<IssueCommentEvent>(
       "On Issue Comment",
       "issue_comment",
-      options
+      client
     ),
   };
+}
+
+export const github = (options: { token: string }) => {
+  const client = new Octokit({
+    auth: options.token,
+  });
+
+  return {
+    metadata,
+    tasks,
+    usesLocalAuth: true,
+    client,
+    triggers: createTriggers(client),
+  } satisfies Connection<Octokit, typeof tasks>;
 };
 
 function buildRepoWebhookTrigger<TEventType>(
   title: string,
   event: string,
-  options?: { token: string },
+  client: Octokit,
   filter?: EventFilter
 ): (params: { repo: string }) => Trigger<TEventType> {
   return (params: { repo: string }) =>
@@ -77,7 +79,7 @@ function buildRepoWebhookTrigger<TEventType>(
           repo: params.repo,
           events: [event],
         },
-        { token: options?.token }
+        client
       ),
       eventRule: {
         event,

integrations/github/src/sources.ts

@@ -1,6 +1,6 @@
 import { Webhooks } from "@octokit/webhooks";
 import { ExternalSource } from "@trigger.dev/sdk/externalSource";
-import { clientFactory } from "./client";
+import { Octokit } from "octokit";
 import { metadata } from "./metadata";
 
 type WebhookData = {
@@ -27,24 +27,21 @@ export function repositoryWebhookSource(
     events: string[];
     secret?: string;
   },
-  { token }: { token?: string } = {}
+  client: Octokit
 ) {
   // Create a stable key for this source so we only register it once
   const key = `github.repo.${params.repo}.webhook`;
 
   return new ExternalSource("http", metadata, {
+    usesLocalAuth: true,
     key,
-    localAuth: token ? { type: "apiKey", apiKey: token } : undefined,
     register: async (triggerClient, auth) => {
       if (!auth) {
         throw new Error("No auth provided");
       }
 
-      const client = clientFactory(auth);
-
       const httpSource = await triggerClient.registerHttpSource({
         key,
-        connectionId: auth.connectionId,
       });
 
       const [owner, repo] = params.repo.split("/");

integrations/github/src/tasks.ts

@@ -1,9 +1,12 @@
 import { authenticatedTask } from "@trigger.dev/sdk";
-import { clientFactory } from "./client";
+import { Octokit } from "octokit";
 
 export const createIssue = authenticatedTask({
-  clientFactory,
-  run: async (params: { title: string; repo: string }, client, task) => {
+  run: async (
+    params: { title: string; repo: string },
+    client: InstanceType<typeof Octokit>,
+    task
+  ) => {
     const [owner, repo] = params.repo.split("/");
 
     return client.rest.issues
@@ -33,10 +36,9 @@ export const createIssue = authenticatedTask({
 });
 
 export const createIssueComment = authenticatedTask({
-  clientFactory,
   run: async (
     params: { body: string; repo: string; issueNumber: number },
-    client,
+    client: InstanceType<typeof Octokit>,
     task
   ) => {
     const [owner, repo] = params.repo.split("/");
@@ -69,8 +71,11 @@ export const createIssueComment = authenticatedTask({
 });
 
 export const getRepo = authenticatedTask({
-  clientFactory,
-  run: async (params: { repo: string }, client, task) => {
+  run: async (
+    params: { repo: string },
+    client: InstanceType<typeof Octokit>,
+    task
+  ) => {
     const [owner, repo] = params.repo.split("/");
 
     const response = await client.rest.repos.get({
@@ -105,14 +110,13 @@ type ReactionContent =
   | "eyes";
 
 export const addIssueCommentReaction = authenticatedTask({
-  clientFactory,
   run: async (
     params: {
       repo: string;
       commentId: number;
       content: ReactionContent;
     },
-    client,
+    client: InstanceType<typeof Octokit>,
     task
   ) => {
     const [owner, repo] = params.repo.split("/");
@@ -175,7 +179,6 @@ export const addIssueCommentReaction = authenticatedTask({
 });
 
 export const createIssueCommentWithReaction = authenticatedTask({
-  clientFactory,
   run: async (
     params: {
       body: string;
@@ -183,7 +186,7 @@ export const createIssueCommentWithReaction = authenticatedTask({
       issueNumber: number;
       reaction: ReactionContent;
     },
-    client,
+    client: InstanceType<typeof Octokit>,
     task,
     io
   ) => {

integrations/slack/package.json

@@ -0,0 +1,31 @@
+{
+  "name": "@trigger.dev/slack",
+  "version": "0.2.0",
+  "description": "The official Slack integration for Trigger.dev",
+  "main": "./dist/index.js",
+  "types": "./dist/index.d.ts",
+  "publishConfig": {
+    "access": "public"
+  },
+  "files": [
+    "dist/index.js",
+    "dist/index.d.ts",
+    "dist/index.js.map"
+  ],
+  "devDependencies": {
+    "@trigger.dev/tsconfig": "workspace:*",
+    "@types/node": "18",
+    "rimraf": "^3.0.2",
+    "tsup": "^6.5.0"
+  },
+  "scripts": {
+    "clean": "rimraf dist",
+    "build": "npm run clean && npm run build:tsup",
+    "build:tsup": "tsup"
+  },
+  "dependencies": {
+    "@slack/web-api": "^6.8.1",
+    "@trigger.dev/sdk": "workspace:^1.0.0",
+    "zod": "^3.20.2"
+  }
+}

integrations/slack/src/client.ts

@@ -0,0 +1,10 @@
+import { ClientFactory } from "@trigger.dev/sdk";
+import { WebClient } from "@slack/web-api";
+
+export const clientFactory: ClientFactory<InstanceType<typeof WebClient>> = (
+  auth
+) => {
+  console.log("Creating slack client", auth);
+
+  return new WebClient(auth.accessToken);
+};

integrations/slack/src/index.ts

@@ -0,0 +1,23 @@
+import { WebClient } from "@slack/web-api";
+import type { Connection } from "@trigger.dev/sdk";
+import { clientFactory } from "./client";
+import { metadata } from "./metadata";
+import { postMessage } from "./tasks";
+
+const tasks = {
+  postMessage,
+};
+
+export type SlackIntegrationOptions = {
+  id: string;
+};
+
+export const slack = ({ id }: SlackIntegrationOptions) => {
+  return {
+    id,
+    metadata,
+    tasks,
+    usesLocalAuth: false,
+    clientFactory,
+  } satisfies Connection<WebClient, typeof tasks>;
+};

integrations/slack/src/metadata.ts

@@ -0,0 +1 @@
+export const metadata = { id: "slack", title: "Slack", icon: "slack" };

integrations/slack/src/tasks.ts

@@ -0,0 +1,29 @@
+import { authenticatedTask } from "@trigger.dev/sdk";
+import { clientFactory } from "./client";
+
+export const postMessage = authenticatedTask({
+  run: async (
+    params: { text: string; channel: string },
+    client: ReturnType<typeof clientFactory>,
+    task,
+    io
+  ) => {
+    return client.chat.postMessage({
+      text: params.text,
+      channel: params.channel,
+      link_names: true,
+    });
+  },
+  init: (params) => {
+    return {
+      name: "Post Message",
+      params,
+      elements: [
+        {
+          label: "Channel ID",
+          text: params.channel,
+        },
+      ],
+    };
+  },
+});

integrations/slack/tsconfig.json

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

integrations/slack/tsup.config.ts

@@ -0,0 +1,22 @@
+import { defineConfig } from "tsup";
+
+export default defineConfig([
+  {
+    name: "main",
+    entry: ["./src/index.ts"],
+    outDir: "./dist",
+    platform: "node",
+    format: ["cjs"],
+    legacyOutput: true,
+    sourcemap: true,
+    clean: true,
+    bundle: true,
+    splitting: false,
+    dts: true,
+    treeshake: {
+      preset: "smallest",
+    },
+    esbuildPlugins: [],
+    external: ["http", "https", "util", "events", "tty", "os", "timers"],
+  },
+]);

packages/internal/src/schemas/api.ts

@@ -81,6 +81,7 @@ export const JobSchema = z.object({
       key: z.string(),
       metadata: ConnectionMetadataSchema,
       usesLocalAuth: z.boolean().default(false),
+      id: z.string().optional(),
     })
   ),
   supportsPreparation: z.boolean(),

packages/internal/src/schemas/connections.ts

@@ -8,39 +8,11 @@ export const ConnectionMetadataSchema = z.object({
 
 export type ConnectionMetadata = z.infer<typeof ConnectionMetadataSchema>;
 
-const BaseConnectionAuthSchema = z.object({
-  additionalFields: z.record(z.string()).optional(),
-  connectionId: z.string().optional(),
-});
-
-export const ApiKeyConnectionAuthSchema = BaseConnectionAuthSchema.extend({
-  type: z.literal("apiKey"),
-  apiKey: z.string(),
-});
-
-export type ApiKeyConnectionAuth = z.infer<typeof ApiKeyConnectionAuthSchema>;
-
-export const OAuthConnectionAuthSchema = BaseConnectionAuthSchema.extend({
-  type: z.literal("oauth"),
+export const ConnectionAuthSchema = z.object({
+  type: z.enum(["oauth2"]),
   accessToken: z.string(),
+  scopes: z.array(z.string()).optional(),
+  additionalFields: z.record(z.string()).optional(),
 });
 
-export type OAuthConnectionAuth = z.infer<typeof OAuthConnectionAuthSchema>;
-
-export const BasicAuthConnectionAuthSchema = BaseConnectionAuthSchema.extend({
-  type: z.literal("basicAuth"),
-  username: z.string(),
-  password: z.string(),
-});
-
-export type BasicAuthConnectionAuth = z.infer<
-  typeof BasicAuthConnectionAuthSchema
->;
-
-export const ConnectionAuthSchema = z.discriminatedUnion("type", [
-  ApiKeyConnectionAuthSchema,
-  OAuthConnectionAuthSchema,
-  BasicAuthConnectionAuthSchema,
-]);
-
 export type ConnectionAuth = z.infer<typeof ConnectionAuthSchema>;

packages/internal/src/schemas/triggers.ts

@@ -18,6 +18,7 @@ export const TriggerMetadataSchema = z.object({
     .object({
       metadata: ConnectionMetadataSchema,
       usesLocalAuth: z.boolean(),
+      id: z.string().optional(),
     })
     .optional(),
 });

packages/trigger-sdk/src/connections.ts

@@ -12,14 +12,15 @@ export type ClientFactory<TClientType> = (auth: ConnectionAuth) => TClientType;
 
 export type Connection<
   TClientType,
-  TTriggers extends Record<string, Trigger<any>>,
   TTasks extends Record<string, AuthenticatedTask<TClientType, any, any>>
 > = {
   usesLocalAuth: boolean;
   metadata: ConnectionMetadata;
-  clientFactory: ClientFactory<TClientType>;
-  triggers?: TTriggers;
+  clientFactory?: ClientFactory<TClientType>;
+  client?: TClientType;
   tasks?: TTasks;
+  id?: string;
+  [key: string]: any;
 };
 
 export type ConnectionEvent<TParams, TEvent> = {
@@ -28,7 +29,6 @@ export type ConnectionEvent<TParams, TEvent> = {
 };
 
 export type AuthenticatedTask<TClientType, TParams, TResult> = {
-  clientFactory: ClientFactory<TClientType>;
   run: (
     params: TParams,
     client: TClientType,
@@ -39,7 +39,6 @@ export type AuthenticatedTask<TClientType, TParams, TResult> = {
 };
 
 export function authenticatedTask<TClientType, TParams, TResult>(options: {
-  clientFactory: ClientFactory<TClientType>;
   run: (
     params: TParams,
     client: TClientType,
@@ -65,20 +64,26 @@ type ExtractTasks<
   [key in keyof TTasks]: ExtractRunFunction<TTasks[key]>;
 };
 
-type ExtractClient<TClientFactory extends ClientFactory<any>> = {
-  client: ReturnType<TClientFactory>;
-};
+type ExtractClient<
+  TClientFactory extends ClientFactory<any> | undefined,
+  TClient extends any | undefined
+> = TClientFactory extends ClientFactory<infer TClientType>
+  ? { client: TClientType }
+  : TClient extends any
+  ? { client: TClient }
+  : never;
 
-type ExtractConnection<TConnection extends Connection<any, any, any>> =
-  ExtractTasks<TConnection["tasks"]> &
-    ExtractClient<TConnection["clientFactory"]>;
+type ExtractConnection<TConnection extends Connection<any, any>> = ExtractTasks<
+  TConnection["tasks"]
+> &
+  ExtractClient<TConnection["clientFactory"], TConnection["client"]>;
 
 type ExtractConnections<
-  TConnections extends Record<string, Connection<any, any, any>>
+  TConnections extends Record<string, Connection<any, any>>
 > = {
   [key in keyof TConnections]: ExtractConnection<TConnections[key]>;
 };
 
 export type IOWithConnections<
-  TConnections extends Record<string, Connection<any, any, any>>
+  TConnections extends Record<string, Connection<any, any>>
 > = IO & ExtractConnections<TConnections>;

packages/trigger-sdk/src/externalSource.ts

@@ -50,7 +50,7 @@ export type HandlerFunction<
 
 export type ExternalSourceOptions<TChannel extends ChannelNames> = {
   key: string;
-  localAuth?: ConnectionAuth;
+  usesLocalAuth: boolean;
   register: (
     triggerClient: TriggerClient,
     auth?: ConnectionAuth
@@ -99,11 +99,11 @@ export class ExternalSource<TChannel extends ChannelNames>
   }
 
   get usesLocalAuth() {
-    return typeof this.options.localAuth !== "undefined";
+    return this.options.usesLocalAuth;
   }
 
   async prepareForExecution(client: TriggerClient, auth?: ConnectionAuth) {
-    return this.options.register(client, auth ?? this.options.localAuth);
+    return this.options.register(client, auth);
   }
 
   async handler(
@@ -111,11 +111,7 @@ export class ExternalSource<TChannel extends ChannelNames>
     event: ExternalSourceChannelMap[TChannel]["event"],
     auth?: ConnectionAuth
   ) {
-    return this.options.handler(
-      triggerClient,
-      event,
-      auth ?? this.options.localAuth
-    );
+    return this.options.handler(triggerClient, event, auth);
   }
 
   eventElements(event: ApiEventLog) {

packages/trigger-sdk/src/index.ts

@@ -3,6 +3,7 @@ export * from "./job";
 export * from "./triggerClient";
 export * from "./connections";
 export * from "./externalSource";
+export * from "./io";
 
 import { SecureString } from "./types";
 

packages/trigger-sdk/src/job.ts

@@ -6,7 +6,7 @@ import type { TriggerContext } from "./types";
 
 export type JobOptions<
   TEventType extends object = {},
-  TConnections extends Record<string, Connection<any, any, any>> = {}
+  TConnections extends Record<string, Connection<any, any>> = {}
 > = {
   id: string;
   name: string;
@@ -24,7 +24,7 @@ export type JobOptions<
 
 export class Job<
   TEventType extends object,
-  TConnections extends Record<string, Connection<any, any, any>>
+  TConnections extends Record<string, Connection<any, any>>
 > {
   readonly options: JobOptions<TEventType, TConnections>;
 
@@ -57,6 +57,7 @@ export class Job<
         key,
         metadata: connection.metadata,
         usesLocalAuth: connection.usesLocalAuth,
+        id: connection.id,
       };
     });
   }

packages/trigger-sdk/src/keyValueStorage.ts

@@ -1,61 +0,0 @@
-import { TriggerKeyValueStorage } from "./types";
-
-export type KvSetFunction = (operation: {
-  key: string;
-  namespace: string;
-  idempotencyKey: string;
-  value: any;
-}) => Promise<void>;
-export type KvGetFunction = (operation: {
-  key: string;
-  namespace: string;
-  idempotencyKey: string;
-}) => Promise<any>;
-export type KvDeleteFunction = (operation: {
-  key: string;
-  namespace: string;
-  idempotencyKey: string;
-}) => Promise<any>;
-
-export class ContextKeyValueStorage implements TriggerKeyValueStorage {
-  getCount: number = 0;
-  setCount: number = 0;
-  deleteCount: number = 0;
-
-  constructor(
-    private namespace: string,
-    private onGet: KvGetFunction,
-    private onSet: KvSetFunction,
-    private onDelete: KvDeleteFunction
-  ) {}
-
-  get<T>(key: string): Promise<T | undefined> {
-    const operation = {
-      key,
-      namespace: this.namespace,
-      idempotencyKey: `get:${this.namespace}:${key}:${this.getCount++}`,
-    };
-
-    return this.onGet(operation);
-  }
-  set<T>(key: string, value: T): Promise<void> {
-    const operation = {
-      key,
-      namespace: this.namespace,
-      idempotencyKey: `set:${this.namespace}:${key}:${this.setCount++}`,
-      value,
-    };
-
-    return this.onSet(operation);
-  }
-
-  delete(key: string): Promise<void> {
-    const operation = {
-      key,
-      namespace: this.namespace,
-      idempotencyKey: `delete:${this.namespace}:${key}:${this.deleteCount++}`,
-    };
-
-    return this.onDelete(operation);
-  }
-}

packages/trigger-sdk/src/triggerClient.ts

@@ -1,6 +1,5 @@
 import {
   ApiEventLog,
-  ApiEventLogSchema,
   ConnectionAuth,
   ErrorWithMessage,
   ErrorWithStackSchema,
@@ -418,7 +417,12 @@ export class TriggerClient {
     const connections = Object.entries(jobConnections).reduce(
       (acc, [key, jobConnection]) => {
         const connection = executionConnections[key];
-        const client = jobConnection.clientFactory(connection);
+        const client =
+          jobConnection.client ?? jobConnection.clientFactory?.(connection);
+
+        if (!client) {
+          return acc;
+        }
 
         const ioConnection = {
           client,

pnpm-lock.yaml

@@ -475,6 +475,25 @@ importers:
       rimraf: 3.0.2
       tsup: 6.6.3
 
+  integrations/slack:
+    specifiers:
+      '@slack/web-api': ^6.8.1
+      '@trigger.dev/sdk': workspace:^1.0.0
+      '@trigger.dev/tsconfig': workspace:*
+      '@types/node': '18'
+      rimraf: ^3.0.2
+      tsup: ^6.5.0
+      zod: ^3.20.2
+    dependencies:
+      '@slack/web-api': 6.8.1
+      '@trigger.dev/sdk': link:../../packages/trigger-sdk
+      zod: 3.20.2
+    devDependencies:
+      '@trigger.dev/tsconfig': link:../../config-packages/tsconfig
+      '@types/node': 18.15.13
+      rimraf: 3.0.2
+      tsup: 6.6.3
+
   packages/create-trigger:
     specifiers:
       '@types/degit': ^2.8.3
@@ -5685,6 +5704,37 @@ packages:
     engines: {node: '>=10'}
     dev: true
 
+  /@slack/logger/3.0.0:
+    resolution: {integrity: sha512-DTuBFbqu4gGfajREEMrkq5jBhcnskinhr4+AnfJEk48zhVeEv3XnUKGIX98B74kxhYsIMfApGGySTn7V3b5yBA==}
+    engines: {node: '>= 12.13.0', npm: '>= 6.12.0'}
+    dependencies:
+      '@types/node': 18.15.13
+    dev: false
+
+  /@slack/types/2.8.0:
+    resolution: {integrity: sha512-ghdfZSF0b4NC9ckBA8QnQgC9DJw2ZceDq0BIjjRSv6XAZBXJdWgxIsYz0TYnWSiqsKZGH2ZXbj9jYABZdH3OSQ==}
+    engines: {node: '>= 12.13.0', npm: '>= 6.12.0'}
+    dev: false
+
+  /@slack/web-api/6.8.1:
+    resolution: {integrity: sha512-eMPk2S99S613gcu7odSw/LV+Qxr8A+RXvBD0GYW510wJuTERiTjP5TgCsH8X09+lxSumbDE88wvWbuFuvGa74g==}
+    engines: {node: '>= 12.13.0', npm: '>= 6.12.0'}
+    dependencies:
+      '@slack/logger': 3.0.0
+      '@slack/types': 2.8.0
+      '@types/is-stream': 1.1.0
+      '@types/node': 18.15.13
+      axios: 0.27.2
+      eventemitter3: 3.1.2
+      form-data: 2.5.1
+      is-electron: 2.2.0
+      is-stream: 1.1.0
+      p-queue: 6.6.2
+      p-retry: 4.6.2
+    transitivePeerDependencies:
+      - debug
+    dev: false
+
   /@swc/core-darwin-arm64/1.3.26:
     resolution: {integrity: sha512-FWWflBfKRYrUJtko2xiedC5XCa31O75IZZqnTWuLpe9g3C5tnUuF3M8LSXZS/dn6wprome1MhtG9GMPkSYkhkg==}
     engines: {node: '>=10'}
@@ -6124,6 +6174,12 @@ packages:
       ci-info: 3.7.1
     dev: false
 
+  /@types/is-stream/1.1.0:
+    resolution: {integrity: sha512-jkZatu4QVbR60mpIzjINmtS1ZF4a/FqdTUTBeQDVOQ2PYyidtwFKr0B5G6ERukKwliq+7mIXvxyppwzG5EgRYg==}
+    dependencies:
+      '@types/node': 18.15.13
+    dev: false
+
   /@types/istanbul-lib-coverage/2.0.4:
     resolution: {integrity: sha512-z/QT1XN4K4KYuslS23k62yDIDLwLFkzxOuMplDtObz0+y7VqJCaO2o+SPwHCvLFZh7xazvvoor2tA/hPz9ee7g==}
     dev: true
@@ -6313,6 +6369,10 @@ packages:
       '@types/node': 18.14.0
     dev: true
 
+  /@types/retry/0.12.0:
+    resolution: {integrity: sha512-wWKOClTTiizcZhXnPY4wikVAwmdYHp8q6DmC+EJUzAMsycb7HB32Kh9RN4+0gExjmPmZSAQjgURXIGATPegAvA==}
+    dev: false
+
   /@types/scheduler/0.16.2:
     resolution: {integrity: sha512-hppQEBDmlwhFAXKJX2KnWLYu5yMfi91yazPb2l+lbJiwW+wdo1gNeRA+3RgNSO39WYX2euey41KEwnqesU2Jew==}
 
@@ -10027,6 +10087,14 @@ packages:
     resolution: {integrity: sha512-tYUSVOGeQPKt/eC1ABfhHy5Xd96N3oIijJvN3O9+TsC28T5V9yX9oEfEK5faP0EFSNVOG97qtAS68GBrQB2hDg==}
     dev: true
 
+  /eventemitter3/3.1.2:
+    resolution: {integrity: sha512-tvtQIeLVHjDkJYnzf2dgVMxfuSGJeM/7UCG17TT4EumTfNtF+0nebF/4zWOIkCreAbtNqhGEboB6BWrwqNaw4Q==}
+    dev: false
+
+  /eventemitter3/4.0.7:
+    resolution: {integrity: sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==}
+    dev: false
+
   /events/3.3.0:
     resolution: {integrity: sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==}
     engines: {node: '>=0.8.x'}
@@ -10506,7 +10574,6 @@ packages:
       asynckit: 0.4.0
       combined-stream: 1.0.8
       mime-types: 2.1.35
-    dev: true
 
   /form-data/3.0.1:
     resolution: {integrity: sha512-RHkBKtLWUVwd7SqRIvCZMEvAMoGUp0XU+seQiZejj0COz3RI3hWP4sCv3gZWWLjJTd7rGwcsF5eKZGii0r/hbg==}
@@ -11638,6 +11705,10 @@ packages:
     hasBin: true
     dev: true
 
+  /is-electron/2.2.0:
+    resolution: {integrity: sha512-SpMppC2XR3YdxSzczXReBjqs2zGscWQpBIKqwXYBFic0ERaxNVgwLCHwOLZeESfdJQjX0RDvrJ1lBXX2ij+G1Q==}
+    dev: false
+
   /is-extendable/0.1.1:
     resolution: {integrity: sha512-5BMULNob1vgFX6EjQw5izWDxrecWK9AM72rugNr0TFldMOi0fj6Jk+zeKIt0xGj4cEfQIJth4w3OKWOJ4f+AFw==}
     engines: {node: '>=0.10.0'}
@@ -11813,6 +11884,11 @@ packages:
     dependencies:
       call-bind: 1.0.2
 
+  /is-stream/1.1.0:
+    resolution: {integrity: sha512-uQPm8kcs47jx38atAcWTVxyltQYoPT68y9aWYdV6yWXSyW8mzSat0TL6CiWdZeCdF3KrAvpVtnHbTv4RN+rqdQ==}
+    engines: {node: '>=0.10.0'}
+    dev: false
+
   /is-stream/2.0.1:
     resolution: {integrity: sha512-hFoiJiTl63nn+kstHGBtewWSKnQLpyb155KHheA1l39uvtO9nWIop1p3udqPcUd/xbF1VLMO4n7OI6p7RbngDg==}
     engines: {node: '>=8'}
@@ -14045,6 +14121,22 @@ packages:
       aggregate-error: 3.1.0
     dev: true
 
+  /p-queue/6.6.2:
+    resolution: {integrity: sha512-RwFpb72c/BhQLEXIZ5K2e+AhgNVmIejGlTgiB9MzZ0e93GRvqZ7uSi0dvRF7/XIXDeNkra2fNHBxTyPDGySpjQ==}
+    engines: {node: '>=8'}
+    dependencies:
+      eventemitter3: 4.0.7
+      p-timeout: 3.2.0
+    dev: false
+
+  /p-retry/4.6.2:
+    resolution: {integrity: sha512-312Id396EbJdvRONlngUx0NydfrIQ5lsYu0znKVUzVvArzEIt08V1qhtyESbGVd1FGX7UKtiFp5uwKZdM8wIuQ==}
+    engines: {node: '>=8'}
+    dependencies:
+      '@types/retry': 0.12.0
+      retry: 0.13.1
+    dev: false
+
   /p-timeout/3.2.0:
     resolution: {integrity: sha512-rhIwUycgwwKcP9yTOOFK/AKsAopjjCakVqLHePO3CC6Mir1Z99xT+R63jZxAT5lFZLa2inS5h+ZS2GvR99/FBg==}
     engines: {node: '>=8'}