diff --git a/apps/webapp/app/models/jobConnection.server.ts b/apps/webapp/app/models/jobConnection.server.ts new file mode 100644 index 000000000..94209c105 --- /dev/null +++ b/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 +): Promise> { + const result: Record = {}; + + 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; +} diff --git a/apps/webapp/app/services/clientApi.server.ts b/apps/webapp/app/services/clientApi.server.ts index 2add31e20..3413ecf1e 100644 --- a/apps/webapp/app/services/clientApi.server.ts +++ b/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"; diff --git a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts b/apps/webapp/app/services/endpoints/endpointRegistered.server.ts index d243bc3d4..47313d88d 100644 --- a/apps/webapp/app/services/endpoints/endpointRegistered.server.ts +++ b/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 { + 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 { + // 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, }, }); } diff --git a/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts b/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts index cd9277e17..3ea724e06 100644 --- a/apps/webapp/app/services/endpoints/prepareJobInstance.server.ts +++ b/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) { diff --git a/apps/webapp/app/services/events/deliverEvent.server.ts b/apps/webapp/app/services/events/deliverEvent.server.ts index 011be68ce..50fb716ae 100644 --- a/apps/webapp/app/services/events/deliverEvent.server.ts +++ b/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"; diff --git a/apps/webapp/app/services/pulsarClient.server.ts b/apps/webapp/app/services/pulsarClient.server.ts deleted file mode 100644 index 24b4013ee..000000000 --- a/apps/webapp/app/services/pulsarClient.server.ts +++ /dev/null @@ -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, - }); -} diff --git a/apps/webapp/app/services/runs/startRun.server.ts b/apps/webapp/app/services/runs/startRun.server.ts index c09b6d8f9..7da0ac1ed 100644 --- a/apps/webapp/app/services/runs/startRun.server.ts +++ b/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; + const connections: Array = + 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) { diff --git a/examples/nextjs-example/src/pages/api/trigger.ts b/examples/nextjs-example/src/pages/api/trigger.ts index 501cf59f4..47d9aee4b 100644 --- a/examples/nextjs-example/src/pages/api/trigger.ts +++ b/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" }, diff --git a/examples/nextjs-example/tsconfig.json b/examples/nextjs-example/tsconfig.json index 2f99443e3..0c073ecf3 100644 --- a/examples/nextjs-example/tsconfig.json +++ b/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"], diff --git a/integrations/github/src/client.ts b/integrations/github/src/client.ts deleted file mode 100644 index 2b11f17b5..000000000 --- a/integrations/github/src/client.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { ClientFactory } from "@trigger.dev/sdk"; -import { Octokit } from "octokit"; - -export const clientFactory: ClientFactory> = ( - 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"], - }); -}; diff --git a/integrations/github/src/index.ts b/integrations/github/src/index.ts index 45acc31b9..2f35bce67 100644 --- a/integrations/github/src/index.ts +++ b/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( - "On Issue", - "issues", - options - ), + onIssue: buildRepoWebhookTrigger("On Issue", "issues", client), onIssueOpened: buildRepoWebhookTrigger( "On Issue Opened", "issues", - options, + client, { action: ["opened"], } @@ -48,15 +36,29 @@ export const github = (options?: { token: string }) => { onIssueComment: buildRepoWebhookTrigger( "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; }; function buildRepoWebhookTrigger( title: string, event: string, - options?: { token: string }, + client: Octokit, filter?: EventFilter ): (params: { repo: string }) => Trigger { return (params: { repo: string }) => @@ -77,7 +79,7 @@ function buildRepoWebhookTrigger( repo: params.repo, events: [event], }, - { token: options?.token } + client ), eventRule: { event, diff --git a/integrations/github/src/sources.ts b/integrations/github/src/sources.ts index d0a54db7e..35313f23c 100644 --- a/integrations/github/src/sources.ts +++ b/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("/"); diff --git a/integrations/github/src/tasks.ts b/integrations/github/src/tasks.ts index fe8c59b5d..9a3800ead 100644 --- a/integrations/github/src/tasks.ts +++ b/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, + 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, 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, + 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, 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, task, io ) => { diff --git a/integrations/slack/package.json b/integrations/slack/package.json new file mode 100644 index 000000000..bcbe65923 --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/integrations/slack/src/client.ts b/integrations/slack/src/client.ts new file mode 100644 index 000000000..2bb98f85c --- /dev/null +++ b/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> = ( + auth +) => { + console.log("Creating slack client", auth); + + return new WebClient(auth.accessToken); +}; diff --git a/integrations/slack/src/index.ts b/integrations/slack/src/index.ts new file mode 100644 index 000000000..5164e6599 --- /dev/null +++ b/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; +}; diff --git a/integrations/slack/src/metadata.ts b/integrations/slack/src/metadata.ts new file mode 100644 index 000000000..53903f376 --- /dev/null +++ b/integrations/slack/src/metadata.ts @@ -0,0 +1 @@ +export const metadata = { id: "slack", title: "Slack", icon: "slack" }; diff --git a/integrations/slack/src/tasks.ts b/integrations/slack/src/tasks.ts new file mode 100644 index 000000000..678d68452 --- /dev/null +++ b/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, + 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, + }, + ], + }; + }, +}); diff --git a/integrations/slack/tsconfig.json b/integrations/slack/tsconfig.json new file mode 100644 index 000000000..1b60e1db5 --- /dev/null +++ b/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"] +} diff --git a/integrations/slack/tsup.config.ts b/integrations/slack/tsup.config.ts new file mode 100644 index 000000000..483aba1d5 --- /dev/null +++ b/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"], + }, +]); diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index f80b1c9b5..c3987c458 100644 --- a/packages/internal/src/schemas/api.ts +++ b/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(), diff --git a/packages/internal/src/schemas/connections.ts b/packages/internal/src/schemas/connections.ts index 964adc894..762cf4ae9 100644 --- a/packages/internal/src/schemas/connections.ts +++ b/packages/internal/src/schemas/connections.ts @@ -8,39 +8,11 @@ export const ConnectionMetadataSchema = z.object({ export type ConnectionMetadata = z.infer; -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; - -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; - -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; diff --git a/packages/internal/src/schemas/triggers.ts b/packages/internal/src/schemas/triggers.ts index 166b7783c..b050f611b 100644 --- a/packages/internal/src/schemas/triggers.ts +++ b/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(), }); diff --git a/packages/trigger-sdk/src/connections.ts b/packages/trigger-sdk/src/connections.ts index 8d321c1d6..183afb20f 100644 --- a/packages/trigger-sdk/src/connections.ts +++ b/packages/trigger-sdk/src/connections.ts @@ -12,14 +12,15 @@ export type ClientFactory = (auth: ConnectionAuth) => TClientType; export type Connection< TClientType, - TTriggers extends Record>, TTasks extends Record> > = { usesLocalAuth: boolean; metadata: ConnectionMetadata; - clientFactory: ClientFactory; - triggers?: TTriggers; + clientFactory?: ClientFactory; + client?: TClientType; tasks?: TTasks; + id?: string; + [key: string]: any; }; export type ConnectionEvent = { @@ -28,7 +29,6 @@ export type ConnectionEvent = { }; export type AuthenticatedTask = { - clientFactory: ClientFactory; run: ( params: TParams, client: TClientType, @@ -39,7 +39,6 @@ export type AuthenticatedTask = { }; export function authenticatedTask(options: { - clientFactory: ClientFactory; run: ( params: TParams, client: TClientType, @@ -65,20 +64,26 @@ type ExtractTasks< [key in keyof TTasks]: ExtractRunFunction; }; -type ExtractClient> = { - client: ReturnType; -}; +type ExtractClient< + TClientFactory extends ClientFactory | undefined, + TClient extends any | undefined +> = TClientFactory extends ClientFactory + ? { client: TClientType } + : TClient extends any + ? { client: TClient } + : never; -type ExtractConnection> = - ExtractTasks & - ExtractClient; +type ExtractConnection> = ExtractTasks< + TConnection["tasks"] +> & + ExtractClient; type ExtractConnections< - TConnections extends Record> + TConnections extends Record> > = { [key in keyof TConnections]: ExtractConnection; }; export type IOWithConnections< - TConnections extends Record> + TConnections extends Record> > = IO & ExtractConnections; diff --git a/packages/trigger-sdk/src/externalSource.ts b/packages/trigger-sdk/src/externalSource.ts index e215da9df..2c7fab834 100644 --- a/packages/trigger-sdk/src/externalSource.ts +++ b/packages/trigger-sdk/src/externalSource.ts @@ -50,7 +50,7 @@ export type HandlerFunction< export type ExternalSourceOptions = { key: string; - localAuth?: ConnectionAuth; + usesLocalAuth: boolean; register: ( triggerClient: TriggerClient, auth?: ConnectionAuth @@ -99,11 +99,11 @@ export class ExternalSource } 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 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) { diff --git a/packages/trigger-sdk/src/index.ts b/packages/trigger-sdk/src/index.ts index 1fccd8f24..0d5c5bc76 100644 --- a/packages/trigger-sdk/src/index.ts +++ b/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"; diff --git a/packages/trigger-sdk/src/job.ts b/packages/trigger-sdk/src/job.ts index 7d44ca66c..6bfde7cd0 100644 --- a/packages/trigger-sdk/src/job.ts +++ b/packages/trigger-sdk/src/job.ts @@ -6,7 +6,7 @@ import type { TriggerContext } from "./types"; export type JobOptions< TEventType extends object = {}, - TConnections extends Record> = {} + TConnections extends Record> = {} > = { id: string; name: string; @@ -24,7 +24,7 @@ export type JobOptions< export class Job< TEventType extends object, - TConnections extends Record> + TConnections extends Record> > { readonly options: JobOptions; @@ -57,6 +57,7 @@ export class Job< key, metadata: connection.metadata, usesLocalAuth: connection.usesLocalAuth, + id: connection.id, }; }); } diff --git a/packages/trigger-sdk/src/keyValueStorage.ts b/packages/trigger-sdk/src/keyValueStorage.ts deleted file mode 100644 index 05e2f699b..000000000 --- a/packages/trigger-sdk/src/keyValueStorage.ts +++ /dev/null @@ -1,61 +0,0 @@ -import { TriggerKeyValueStorage } from "./types"; - -export type KvSetFunction = (operation: { - key: string; - namespace: string; - idempotencyKey: string; - value: any; -}) => Promise; -export type KvGetFunction = (operation: { - key: string; - namespace: string; - idempotencyKey: string; -}) => Promise; -export type KvDeleteFunction = (operation: { - key: string; - namespace: string; - idempotencyKey: string; -}) => Promise; - -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(key: string): Promise { - const operation = { - key, - namespace: this.namespace, - idempotencyKey: `get:${this.namespace}:${key}:${this.getCount++}`, - }; - - return this.onGet(operation); - } - set(key: string, value: T): Promise { - const operation = { - key, - namespace: this.namespace, - idempotencyKey: `set:${this.namespace}:${key}:${this.setCount++}`, - value, - }; - - return this.onSet(operation); - } - - delete(key: string): Promise { - const operation = { - key, - namespace: this.namespace, - idempotencyKey: `delete:${this.namespace}:${key}:${this.deleteCount++}`, - }; - - return this.onDelete(operation); - } -} diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 07679f70d..89236d51e 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/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, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7c1b4d622..187500a6a 100644 --- a/pnpm-lock.yaml +++ b/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'}