From 8b7dafcf2daab515dc5c016544446fcc5496976e Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 29 Dec 2022 15:29:49 +0000 Subject: [PATCH] WIP sending/receiving requests --- apps/coordinator/src/server.ts | 76 ++++++ .../app/models/integrationRequest.server.ts | 17 ++ .../app/services/messageBroker.server.ts | 94 ++++++- .../createIntegrationRequest.server.ts | 147 +++++++++++ .../performIntegrationRequest.server.ts | 247 ++++++++++++++++++ .../startIntegrationRequest.server.ts | 37 +++ .../requests/waitForConnection.server.ts | 34 +++ .../migration.sql | 30 +++ .../migration.sql | 41 +++ .../migration.sql | 2 + .../migration.sql | 2 + .../migration.sql | 14 + apps/webapp/prisma/schema.prisma | 103 +++++++- examples/send-to-slack/package.json | 19 ++ examples/send-to-slack/src/index.ts | 29 ++ examples/send-to-slack/tsconfig.json | 12 + packages/internal-bridge/src/schemas/host.ts | 19 ++ .../internal-bridge/src/schemas/server.ts | 15 +- packages/internal-integrations/src/headers.ts | 9 + packages/internal-integrations/src/index.ts | 1 + .../internal-integrations/src/slack/index.ts | 85 ++++++ .../src/slack/schemas.ts | 21 ++ packages/internal-integrations/src/types.ts | 16 ++ .../src/messages/catalogs/coordinator.ts | 4 +- .../src/messages/catalogs/platform.ts | 2 + .../schemas/finishIntegrationRequest.ts | 19 ++ .../schemas/initiateIntegrationRequest.ts | 49 ---- .../schemas/sendIntegrationRequest.ts | 16 ++ .../src/messages/sharedSchemas.ts | 19 +- packages/trigger-integrations/src/index.ts | 3 +- .../src/integrations/slack/index.ts | 32 +++ packages/trigger-sdk/package.json | 14 +- packages/trigger-sdk/src/client.ts | 144 +++++++--- packages/trigger-sdk/src/index.ts | 6 + packages/trigger-sdk/src/localStorage.ts | 27 ++ pnpm-lock.yaml | 129 +++++---- 36 files changed, 1375 insertions(+), 159 deletions(-) create mode 100644 apps/webapp/app/models/integrationRequest.server.ts create mode 100644 apps/webapp/app/services/requests/createIntegrationRequest.server.ts create mode 100644 apps/webapp/app/services/requests/performIntegrationRequest.server.ts create mode 100644 apps/webapp/app/services/requests/startIntegrationRequest.server.ts create mode 100644 apps/webapp/app/services/requests/waitForConnection.server.ts create mode 100644 apps/webapp/prisma/migrations/20221228155121_add_external_service_model/migration.sql create mode 100644 apps/webapp/prisma/migrations/20221229095721_add_integration_requests/migration.sql create mode 100644 apps/webapp/prisma/migrations/20221229100124_add_waiting_for_connection_status/migration.sql create mode 100644 apps/webapp/prisma/migrations/20221229113827_add_fetching_to_integration_request/migration.sql create mode 100644 apps/webapp/prisma/migrations/20221229121050_add_integration_responses/migration.sql create mode 100644 examples/send-to-slack/package.json create mode 100644 examples/send-to-slack/src/index.ts create mode 100644 examples/send-to-slack/tsconfig.json create mode 100644 packages/internal-integrations/src/headers.ts create mode 100644 packages/internal-integrations/src/slack/index.ts create mode 100644 packages/internal-integrations/src/slack/schemas.ts create mode 100644 packages/internal-platform/src/messages/schemas/finishIntegrationRequest.ts delete mode 100644 packages/internal-platform/src/messages/schemas/initiateIntegrationRequest.ts create mode 100644 packages/internal-platform/src/messages/schemas/sendIntegrationRequest.ts create mode 100644 packages/trigger-integrations/src/integrations/slack/index.ts create mode 100644 packages/trigger-sdk/src/localStorage.ts diff --git a/apps/coordinator/src/server.ts b/apps/coordinator/src/server.ts index a9dd088a4..33347fb11 100644 --- a/apps/coordinator/src/server.ts +++ b/apps/coordinator/src/server.ts @@ -89,6 +89,41 @@ export class TriggerServer { sender: HostRPCSchema, receiver: ServerRPCSchema, handlers: { + SEND_REQUEST: async (data) => { + if (!this.#triggerPublisher) { + // TODO: need to recover from this issue by trying to reconnect + return false; + } + + if (!this.#organizationId) { + // TODO: this should never really happen + throw new Error( + "Cannot complete workflow run without an organization ID" + ); + } + + if (!this.#workflowId) { + // TODO: this should never really happen + throw new Error("Cannot send log without a workflow ID"); + } + + const response = await this.#triggerPublisher.publish( + "SEND_INTEGRATION_REQUEST", + { + id: data.requestId, + service: data.service, + endpoint: data.endpoint, + params: data.params, + }, + { + "x-api-key": this.#apiKey, + "x-workflow-id": this.#workflowId, + "x-workflow-run-id": data.id, + } + ); + + return !!response; + }, SEND_EVENT: async (data) => { if (!this.#triggerPublisher) { // TODO: need to recover from this issue by trying to reconnect @@ -305,6 +340,47 @@ export class TriggerServer { subscriptionInitialPosition: "Earliest", }, handlers: { + FINISH_INTEGRATION_REQUEST: async (id, data, properties) => { + this.#logger.debug( + "Received finish integration request", + id, + data, + properties + ); + + if (!this.#serverRPC) { + throw new Error( + "Cannot finish integration request without an RPC connection" + ); + } + + // If the API keys don't match, then we should ignore it + // This ensures the workflow is triggered for the correct environment + if (properties["x-api-key"] !== this.#apiKey) { + return true; + } + + // If the workflow id is not the same as the workflow id + // that we are listening for, then we should ignore it + if (properties["x-workflow-id"] !== this.#workflowId) { + return true; + } + + const success = await this.#serverRPC.send("COMPLETE_REQUEST", { + id: data.id, + status: data.status, + response: data.response, + meta: { + workflowId: properties["x-workflow-id"], + organizationId: properties["x-org-id"], + environment: properties["x-env"], + apiKey: properties["x-api-key"], + runId: properties["x-workflow-run-id"], + }, + }); + + return success; + }, TRIGGER_WORKFLOW: async (id, data, properties) => { this.#logger.debug("Received trigger", id, data, properties); // If the API keys don't match, then we should ignore it diff --git a/apps/webapp/app/models/integrationRequest.server.ts b/apps/webapp/app/models/integrationRequest.server.ts new file mode 100644 index 000000000..88cea8965 --- /dev/null +++ b/apps/webapp/app/models/integrationRequest.server.ts @@ -0,0 +1,17 @@ +import { prisma } from "~/db.server"; +import type { IntegrationRequest } from ".prisma/client"; + +export type { IntegrationRequest }; + +export async function findIntegrationRequestById(id: string) { + return prisma.integrationRequest.findUnique({ + where: { + id, + }, + include: { + externalService: true, + step: true, + run: true, + }, + }); +} diff --git a/apps/webapp/app/services/messageBroker.server.ts b/apps/webapp/app/services/messageBroker.server.ts index 2b72ab818..f7dc58ce9 100644 --- a/apps/webapp/app/services/messageBroker.server.ts +++ b/apps/webapp/app/services/messageBroker.server.ts @@ -11,6 +11,7 @@ import type { Client as PulsarClient } from "pulsar-client"; import Pulsar from "pulsar-client"; import { z } from "zod"; import { env } from "~/env.server"; +import { findIntegrationRequestById } from "~/models/integrationRequest.server"; import { completeWorkflowRun, failWorkflowRun, @@ -22,17 +23,23 @@ import { } from "~/models/workflowRun.server"; import { DispatchEvent } from "./events/dispatch.server"; import { RegisterExternalSource } from "./externalSources/registerExternalSource.server"; +import { CreateIntegrationRequest } from "./requests/createIntegrationRequest.server"; +import { PerformIntegrationRequest } from "./requests/performIntegrationRequest.server"; +import { StartIntegrationRequest } from "./requests/startIntegrationRequest.server"; +import { WaitForConnection } from "./requests/waitForConnection.server"; let pulsarClient: PulsarClient; let triggerPublisher: ZodPublisher; let triggerSubscriber: ZodSubscriber; let internalPubSub: ZodPubSub; +let requestPubSub: ZodPubSub; declare global { var __pulsar_client__: typeof pulsarClient; var __trigger_publisher__: typeof triggerPublisher; var __trigger_subscriber__: typeof triggerSubscriber; var __internal_pub_sub__: typeof internalPubSub; + var __request_pub_sub__: typeof requestPubSub; } export async function init() { @@ -80,6 +87,15 @@ export async function init() { } internalPubSub = global.__internal_pub_sub__; } + + if (env.NODE_ENV === "production") { + requestPubSub = await createRequestPubSub(); + } else { + if (!global.__request_pub_sub__) { + global.__request_pub_sub__ = await createRequestPubSub(); + } + requestPubSub = global.__request_pub_sub__; + } } function createClient() { @@ -132,7 +148,19 @@ async function createTriggerSubscriber() { return true; }, - INITIATE_INTEGRATION_REQUEST: async (id, data, properties) => { + SEND_INTEGRATION_REQUEST: async (id, data, properties) => { + const service = new CreateIntegrationRequest(); + + const integrationRequest = await service.call( + properties["x-api-key"], + properties["x-workflow-run-id"], + data + ); + + internalPubSub.publish("INTEGRATION_REQUEST_CREATED", { + id: integrationRequest.id, + }); + return true; }, COMPLETE_WORKFLOW_RUN: async (id, data, properties) => { @@ -175,8 +203,47 @@ const InternalCatalog = { data: z.object({ id: z.string() }), properties: z.object({}), }, + INTEGRATION_REQUEST_CREATED: { + data: z.object({ id: z.string() }), + properties: z.object({}), + }, }; +const RequestCatalog = { + PERFORM_INTEGRATION_REQUEST: { + data: z.object({ id: z.string() }), + properties: z.object({}), + }, +}; + +async function createRequestPubSub() { + const pubSub = new ZodPubSub({ + client: pulsarClient, + topic: "persistent://public/default/internal-requests", + subscriberConfig: { + subscription: "webapp", + subscriptionType: "Shared", + }, + publisherConfig: { + sendTimeoutMs: 1000, + }, + schema: RequestCatalog, + handlers: { + PERFORM_INTEGRATION_REQUEST: async (id, data, properties) => { + const service = new PerformIntegrationRequest(); + + const success = await service.call(data.id); + + return success; + }, + }, + }); + + await pubSub.initialize(); + + return pubSub; +} + async function createInternalPubSub() { const pubSub = new ZodPubSub({ client: pulsarClient, @@ -190,6 +257,29 @@ async function createInternalPubSub() { }, schema: InternalCatalog, handlers: { + INTEGRATION_REQUEST_CREATED: async (id, data, properties) => { + const integrationRequest = await findIntegrationRequestById(data.id); + + if (!integrationRequest) { + return true; + } + + if (!integrationRequest.externalService.connectionId) { + const service = new WaitForConnection(); + await service.call( + integrationRequest, + integrationRequest.externalService, + integrationRequest.step, + integrationRequest.run + ); + return true; + } else { + const service = new StartIntegrationRequest(); + await service.call(integrationRequest, integrationRequest.step); + + return true; + } + }, EXTERNAL_SOURCE_UPSERTED: async (id, data, properties) => { const service = new RegisterExternalSource(); @@ -241,4 +331,4 @@ async function createInternalPubSub() { return pubSub; } -export { internalPubSub }; +export { internalPubSub, requestPubSub }; diff --git a/apps/webapp/app/services/requests/createIntegrationRequest.server.ts b/apps/webapp/app/services/requests/createIntegrationRequest.server.ts new file mode 100644 index 000000000..3d1f975fe --- /dev/null +++ b/apps/webapp/app/services/requests/createIntegrationRequest.server.ts @@ -0,0 +1,147 @@ +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import type { Organization } from "~/models/organization.server"; + +export class CreateIntegrationRequest { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + async call( + apiKey: string, + workflowRunId: string, + data: { + id: string; + service: string; + endpoint: string; + params?: any; + } + ) { + const environment = await this.#prismaClient.runtimeEnvironment.findUnique({ + where: { + apiKey, + }, + include: { + organization: true, + }, + }); + + if (!environment) { + throw new Error("Invalid API key"); + } + + const workflowRun = await this.#prismaClient.workflowRun.findUnique({ + where: { + id: workflowRunId, + }, + include: { + workflow: true, + }, + }); + + if (!workflowRun) { + throw new Error("Invalid workflow run ID"); + } + + if (workflowRun.workflow.organizationId !== environment.organizationId) { + throw new Error("Invalid workflow run ID"); + } + + // Find existing external service for this workflow and service + // If it doesn't exist, create it + + let externalService = await this.#prismaClient.externalService.findUnique({ + where: { + workflowId_slug: { + workflowId: workflowRun.workflowId, + slug: data.service, + }, + }, + }); + + if (!externalService) { + const existingConnection = await this.#findLatestExistingConnectionInOrg( + data.service, + environment.organization + ); + + externalService = await this.#prismaClient.externalService.create({ + data: { + workflowId: workflowRun.workflowId, + slug: data.service, // For now, we'll use the service name as the slug but this could change + service: data.service, + type: "HTTP_API", + connectionId: existingConnection?.id, + }, + }); + } else { + if (!externalService.connectionId) { + const existingConnection = + await this.#findLatestExistingConnectionInOrg( + data.service, + environment.organization + ); + + if (existingConnection) { + externalService = await this.#prismaClient.externalService.update({ + where: { + id: externalService.id, + }, + data: { + connectionId: existingConnection.id, + }, + }); + } + } + } + // Create the workflow run step + const workflowRunStep = await this.#prismaClient.workflowRunStep.create({ + data: { + runId: workflowRun.id, + type: "INTEGRATION_REQUEST", + input: data.params, + context: { + service: data.service, + endpoint: data.endpoint, + }, + status: "PENDING", + }, + }); + + // Create the integration request + const integrationRequest = + await this.#prismaClient.integrationRequest.create({ + data: { + id: data.id, + params: data.params, + endpoint: data.endpoint, + externalServiceId: externalService.id, + runId: workflowRun.id, + stepId: workflowRunStep.id, + status: "PENDING", + }, + }); + + return integrationRequest; + } + + async #findLatestExistingConnectionInOrg( + serviceIdentifier: string, + organization: Organization + ) { + const connection = await this.#prismaClient.aPIConnection.findFirst({ + where: { + organizationId: organization.id, + apiIdentifier: serviceIdentifier, + status: "CONNECTED", + }, + orderBy: { + createdAt: "desc", + }, + }); + + return connection; + } +} diff --git a/apps/webapp/app/services/requests/performIntegrationRequest.server.ts b/apps/webapp/app/services/requests/performIntegrationRequest.server.ts new file mode 100644 index 000000000..9fbb0cc2d --- /dev/null +++ b/apps/webapp/app/services/requests/performIntegrationRequest.server.ts @@ -0,0 +1,247 @@ +import type { NormalizedResponse } from "internal-integrations"; +import { slack } from "internal-integrations"; +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import type { IntegrationRequest } from "~/models/integrationRequest.server"; +import { pizzly } from "../pizzly.server"; + +export class PerformIntegrationRequest { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + async call(id: string): Promise { + const integrationRequest = + await this.#prismaClient.integrationRequest.findUnique({ + where: { id }, + include: { + externalService: { + include: { + connection: true, + }, + }, + }, + }); + + if (!integrationRequest) { + return false; + } + + if (!integrationRequest.externalService.connection) { + return false; + } + + const accessToken = await pizzly.accessToken( + integrationRequest.externalService.connection.apiIdentifier, + integrationRequest.externalService.connection.id + ); + + if (!accessToken) { + return false; + } + + const response = await this.#performRequest( + integrationRequest.externalService.connection.apiIdentifier, + accessToken, + integrationRequest + ); + + switch (statusCodeToType(response.statusCode)) { + case "informational": { + return this.#completeWithSuccess(integrationRequest, response); + } + case "success": { + return this.#completeWithSuccess(integrationRequest, response); + } + case "redirect": { + return this.#completeWithFailure(integrationRequest, response); + } + case "clientError": { + return this.#completeWithFailure(integrationRequest, response); + } + case "serverError": { + return this.#attemptRetry(integrationRequest, response); + } + default: { + return this.#unknownError(integrationRequest, response); + } + } + } + + async #completeWithSuccess( + integrationRequest: IntegrationRequest, + response: NormalizedResponse + ) { + await this.#createResponse(integrationRequest, response); + + await this.#prismaClient.integrationRequest.update({ + where: { + id: integrationRequest.id, + }, + data: { + status: "SUCCESS", + }, + }); + + await this.#prismaClient.workflowRunStep.update({ + where: { + id: integrationRequest.stepId, + }, + data: { + status: "SUCCESS", + output: response.body, + context: { + headers: response.headers, + statusCode: response.statusCode, + }, + finishedAt: new Date(), + }, + }); + + return true; + } + + async #completeWithFailure( + integrationRequest: IntegrationRequest, + response: NormalizedResponse + ) { + await this.#createResponse(integrationRequest, response); + + await this.#prismaClient.integrationRequest.update({ + where: { + id: integrationRequest.id, + }, + data: { + status: "ERROR", + }, + }); + + await this.#prismaClient.workflowRunStep.update({ + where: { + id: integrationRequest.stepId, + }, + data: { + status: "ERROR", + output: response.body, + context: { + headers: response.headers, + statusCode: response.statusCode, + }, + finishedAt: new Date(), + }, + }); + + return true; + } + + async #attemptRetry( + integrationRequest: IntegrationRequest, + response: NormalizedResponse + ) { + if (integrationRequest.retryCount >= 10) { + await this.#prismaClient.integrationRequest.update({ + where: { + id: integrationRequest.id, + }, + data: { + retryCount: { + increment: 1, + }, + }, + }); + + return this.#completeWithFailure(integrationRequest, response); + } + + await this.#createResponse(integrationRequest, response); + + await this.#prismaClient.integrationRequest.update({ + where: { + id: integrationRequest.id, + }, + data: { + status: "RETRYING", + retryCount: { + increment: 1, + }, + }, + }); + + return false; + } + + async #unknownError( + integrationRequest: IntegrationRequest, + response: NormalizedResponse + ) { + return false; + } + + async #createResponse( + integrationRequest: IntegrationRequest, + response: NormalizedResponse + ) { + const integrationResponse = + await this.#prismaClient.integrationResponse.create({ + data: { + request: { + connect: { + id: integrationRequest.id, + }, + }, + statusCode: response.statusCode, + headers: response.headers, + body: response.body, + }, + }); + + return integrationResponse; + } + + async #performRequest( + service: string, + accessToken: string, + integrationRequest: IntegrationRequest + ): Promise { + switch (service) { + case "slack": { + return slack.requests.perform({ + accessToken, + endpoint: integrationRequest.endpoint, + params: integrationRequest.params, + }); + } + default: { + throw new Error(`Unknown service: ${service}`); + } + } + } +} + +function statusCodeToType( + statusCode: number +): "informational" | "success" | "redirect" | "clientError" | "serverError" { + if (statusCode >= 100 && statusCode < 200) { + return "informational"; + } + + if (statusCode >= 200 && statusCode < 300) { + return "success"; + } + + if (statusCode >= 300 && statusCode < 400) { + return "redirect"; + } + + if (statusCode >= 400 && statusCode < 500) { + return "clientError"; + } + + if (statusCode >= 500 && statusCode < 600) { + return "serverError"; + } + + throw new Error(`Unknown status code: ${statusCode}`); +} diff --git a/apps/webapp/app/services/requests/startIntegrationRequest.server.ts b/apps/webapp/app/services/requests/startIntegrationRequest.server.ts new file mode 100644 index 000000000..4c221fea0 --- /dev/null +++ b/apps/webapp/app/services/requests/startIntegrationRequest.server.ts @@ -0,0 +1,37 @@ +import type { IntegrationRequest, WorkflowRunStep } from ".prisma/client"; +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import { requestPubSub } from "../messageBroker.server"; + +export class StartIntegrationRequest { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + async call(request: IntegrationRequest, step: WorkflowRunStep) { + await this.#prismaClient.integrationRequest.update({ + where: { + id: request.id, + }, + data: { + status: "FETCHING", + }, + }); + + await this.#prismaClient.workflowRunStep.update({ + where: { + id: step.id, + }, + data: { + status: "RUNNING", + startedAt: new Date(), + }, + }); + + requestPubSub.publish("PERFORM_INTEGRATION_REQUEST", { + id: request.id, + }); + } +} diff --git a/apps/webapp/app/services/requests/waitForConnection.server.ts b/apps/webapp/app/services/requests/waitForConnection.server.ts new file mode 100644 index 000000000..4e98aab8c --- /dev/null +++ b/apps/webapp/app/services/requests/waitForConnection.server.ts @@ -0,0 +1,34 @@ +import type { + ExternalService, + IntegrationRequest, + WorkflowRun, + WorkflowRunStep, +} from ".prisma/client"; +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; + +export class WaitForConnection { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + async call( + request: IntegrationRequest, + service: ExternalService, + step: WorkflowRunStep, + run: WorkflowRun + ) { + await this.#prismaClient.integrationRequest.update({ + where: { + id: request.id, + }, + data: { + status: "WAITING_FOR_CONNECTION", + }, + }); + + // TODO: Send user an email with a link to connect their account + } +} diff --git a/apps/webapp/prisma/migrations/20221228155121_add_external_service_model/migration.sql b/apps/webapp/prisma/migrations/20221228155121_add_external_service_model/migration.sql new file mode 100644 index 000000000..83d69e769 --- /dev/null +++ b/apps/webapp/prisma/migrations/20221228155121_add_external_service_model/migration.sql @@ -0,0 +1,30 @@ +-- CreateEnum +CREATE TYPE "ExternalServiceType" AS ENUM ('HTTP_API'); + +-- CreateEnum +CREATE TYPE "ExternalServiceStatus" AS ENUM ('CREATED', 'READY'); + +-- CreateTable +CREATE TABLE "ExternalService" ( + "id" TEXT NOT NULL, + "slug" TEXT NOT NULL, + "service" TEXT NOT NULL, + "workflowId" TEXT NOT NULL, + "connectionId" TEXT, + "type" "ExternalServiceType" NOT NULL, + "status" "ExternalServiceStatus" NOT NULL DEFAULT 'CREATED', + "readyAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "ExternalService_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "ExternalService_workflowId_slug_key" ON "ExternalService"("workflowId", "slug"); + +-- AddForeignKey +ALTER TABLE "ExternalService" ADD CONSTRAINT "ExternalService_workflowId_fkey" FOREIGN KEY ("workflowId") REFERENCES "Workflow"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "ExternalService" ADD CONSTRAINT "ExternalService_connectionId_fkey" FOREIGN KEY ("connectionId") REFERENCES "APIConnection"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20221229095721_add_integration_requests/migration.sql b/apps/webapp/prisma/migrations/20221229095721_add_integration_requests/migration.sql new file mode 100644 index 000000000..dbbd9e231 --- /dev/null +++ b/apps/webapp/prisma/migrations/20221229095721_add_integration_requests/migration.sql @@ -0,0 +1,41 @@ +-- CreateEnum +CREATE TYPE "IntegrationRequestStatus" AS ENUM ('PENDING', 'RETRYING', 'SUCCESS', 'ERROR'); + +-- CreateEnum +CREATE TYPE "WorkflowRunStepStatus" AS ENUM ('PENDING', 'RUNNING', 'SUCCESS', 'ERROR'); + +-- AlterEnum +ALTER TYPE "WorkflowRunStepType" ADD VALUE 'INTEGRATION_REQUEST'; + +-- AlterTable +ALTER TABLE "WorkflowRunStep" ADD COLUMN "status" "WorkflowRunStepStatus" NOT NULL DEFAULT 'PENDING'; + +-- CreateTable +CREATE TABLE "IntegrationRequest" ( + "id" TEXT NOT NULL, + "params" JSONB NOT NULL, + "endpoint" TEXT NOT NULL, + "externalServiceId" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + "status" "IntegrationRequestStatus" NOT NULL DEFAULT 'PENDING', + "runId" TEXT NOT NULL, + "stepId" TEXT NOT NULL, + "retryCount" INTEGER NOT NULL DEFAULT 0, + "error" JSONB, + "response" JSONB, + + CONSTRAINT "IntegrationRequest_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "IntegrationRequest_stepId_key" ON "IntegrationRequest"("stepId"); + +-- AddForeignKey +ALTER TABLE "IntegrationRequest" ADD CONSTRAINT "IntegrationRequest_externalServiceId_fkey" FOREIGN KEY ("externalServiceId") REFERENCES "ExternalService"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "IntegrationRequest" ADD CONSTRAINT "IntegrationRequest_runId_fkey" FOREIGN KEY ("runId") REFERENCES "WorkflowRun"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "IntegrationRequest" ADD CONSTRAINT "IntegrationRequest_stepId_fkey" FOREIGN KEY ("stepId") REFERENCES "WorkflowRunStep"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20221229100124_add_waiting_for_connection_status/migration.sql b/apps/webapp/prisma/migrations/20221229100124_add_waiting_for_connection_status/migration.sql new file mode 100644 index 000000000..fdb531227 --- /dev/null +++ b/apps/webapp/prisma/migrations/20221229100124_add_waiting_for_connection_status/migration.sql @@ -0,0 +1,2 @@ +-- AlterEnum +ALTER TYPE "IntegrationRequestStatus" ADD VALUE 'WAITING_FOR_CONNECTION'; diff --git a/apps/webapp/prisma/migrations/20221229113827_add_fetching_to_integration_request/migration.sql b/apps/webapp/prisma/migrations/20221229113827_add_fetching_to_integration_request/migration.sql new file mode 100644 index 000000000..bbfe7ae25 --- /dev/null +++ b/apps/webapp/prisma/migrations/20221229113827_add_fetching_to_integration_request/migration.sql @@ -0,0 +1,2 @@ +-- AlterEnum +ALTER TYPE "IntegrationRequestStatus" ADD VALUE 'FETCHING'; diff --git a/apps/webapp/prisma/migrations/20221229121050_add_integration_responses/migration.sql b/apps/webapp/prisma/migrations/20221229121050_add_integration_responses/migration.sql new file mode 100644 index 000000000..7f6681b5a --- /dev/null +++ b/apps/webapp/prisma/migrations/20221229121050_add_integration_responses/migration.sql @@ -0,0 +1,14 @@ +-- CreateTable +CREATE TABLE "IntegrationResponse" ( + "id" TEXT NOT NULL, + "requestId" TEXT NOT NULL, + "statusCode" INTEGER NOT NULL, + "body" JSONB NOT NULL, + "headers" JSONB NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "IntegrationResponse_pkey" PRIMARY KEY ("id") +); + +-- AddForeignKey +ALTER TABLE "IntegrationResponse" ADD CONSTRAINT "IntegrationResponse_requestId_fkey" FOREIGN KEY ("requestId") REFERENCES "IntegrationRequest"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index 4d63098b7..72a62ef9c 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -67,7 +67,8 @@ model APIConnection { organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade) organizationId String - externalSources ExternalSource[] + externalSources ExternalSource[] + externalServices ExternalService[] } enum APIConnectionType { @@ -117,8 +118,9 @@ model Workflow { externalSource ExternalSource? @relation(fields: [externalSourceId], references: [id], onDelete: Cascade, onUpdate: Cascade) externalSourceId String? - runs WorkflowRun[] - rules EventRule[] + runs WorkflowRun[] + rules EventRule[] + externalServices ExternalService[] service String @default("trigger") eventNames String[] @@ -202,6 +204,86 @@ enum ExternalSourceType { HTTP_POLLING } +model ExternalService { + id String @id @default(cuid()) + slug String + service String + + workflow Workflow @relation(fields: [workflowId], references: [id], onDelete: Cascade, onUpdate: Cascade) + workflowId String + + connection APIConnection? @relation(fields: [connectionId], references: [id], onDelete: Cascade, onUpdate: Cascade) + connectionId String? + + type ExternalServiceType + status ExternalServiceStatus @default(CREATED) + + readyAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + requests IntegrationRequest[] + + @@unique([workflowId, slug]) +} + +enum ExternalServiceType { + HTTP_API +} + +enum ExternalServiceStatus { + CREATED + READY +} + +model IntegrationRequest { + id String @id + + params Json + endpoint String + + externalService ExternalService @relation(fields: [externalServiceId], references: [id], onDelete: Cascade, onUpdate: Cascade) + externalServiceId String + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + status IntegrationRequestStatus @default(PENDING) + + run WorkflowRun @relation(fields: [runId], references: [id], onDelete: Cascade, onUpdate: Cascade) + runId String + + step WorkflowRunStep @relation(fields: [stepId], references: [id], onDelete: Cascade, onUpdate: Cascade) + stepId String @unique + + retryCount Int @default(0) + error Json? + response Json? + responses IntegrationResponse[] +} + +enum IntegrationRequestStatus { + PENDING + WAITING_FOR_CONNECTION + FETCHING + RETRYING + SUCCESS + ERROR +} + +model IntegrationResponse { + id String @id @default(cuid()) + + request IntegrationRequest @relation(fields: [requestId], references: [id], onDelete: Cascade, onUpdate: Cascade) + requestId String + + statusCode Int + body Json + headers Json + + createdAt DateTime @default(now()) +} + model TriggerEvent { id String @id @default(cuid()) service String @@ -258,7 +340,8 @@ model WorkflowRun { startedAt DateTime? finishedAt DateTime? - isTest Boolean @default(false) + isTest Boolean @default(false) + requests IntegrationRequest[] } enum WorkflowRunStatus { @@ -284,6 +367,17 @@ model WorkflowRunStep { startedAt DateTime? finishedAt DateTime? + + status WorkflowRunStepStatus @default(PENDING) + + integrationRequest IntegrationRequest? +} + +enum WorkflowRunStepStatus { + PENDING + RUNNING + SUCCESS + ERROR } enum WorkflowRunStepType { @@ -291,6 +385,7 @@ enum WorkflowRunStepType { LOG_MESSAGE DURABLE_DELAY CUSTOM_EVENT + INTEGRATION_REQUEST } //todo triggers are environment specific diff --git a/examples/send-to-slack/package.json b/examples/send-to-slack/package.json new file mode 100644 index 000000000..8151260c5 --- /dev/null +++ b/examples/send-to-slack/package.json @@ -0,0 +1,19 @@ +{ + "private": true, + "name": "@examples/send-to-slack", + "version": "0.0.1", + "description": "Send a message to slack when a customer creates a new custom domain", + "dependencies": { + "@trigger.dev/integrations": "workspace:*", + "@trigger.dev/sdk": "workspace:*", + "zod": "^3.20.2" + }, + "devDependencies": { + "@trigger.dev/tsconfig": "workspace:*", + "@types/node": "^18.11.9", + "tsx": "^3.12.0" + }, + "scripts": { + "dev": "tsx src/index.ts" + } +} \ No newline at end of file diff --git a/examples/send-to-slack/src/index.ts b/examples/send-to-slack/src/index.ts new file mode 100644 index 000000000..e2f5116e0 --- /dev/null +++ b/examples/send-to-slack/src/index.ts @@ -0,0 +1,29 @@ +import { Trigger, customEvent } from "@trigger.dev/sdk"; +import { slack } from "@trigger.dev/integrations"; +import { z } from "zod"; + +const trigger = new Trigger({ + id: "send-to-slack-on-new-domain", + name: "Send to Slack on new domain", + apiKey: "trigger_dev_zC25mKNn6c0q", + endpoint: "ws://localhost:8889/ws", + logLevel: "debug", + on: customEvent({ + name: "domain.created", + schema: z.object({ + id: z.string(), + customerId: z.string(), + domain: z.string(), + }), + }), + run: async (event, ctx) => { + const response = await slack.postMessage({ + channel: "test-integrations", + text: `New domain created: ${event.domain} by customer ${event.customerId}`, + }); + + return response; + }, +}); + +trigger.listen(); diff --git a/examples/send-to-slack/tsconfig.json b/examples/send-to-slack/tsconfig.json new file mode 100644 index 000000000..c3d429f86 --- /dev/null +++ b/examples/send-to-slack/tsconfig.json @@ -0,0 +1,12 @@ +{ + "extends": "@trigger.dev/tsconfig/node18.json", + "include": ["src/**/*.ts"], + "compilerOptions": { + "experimentalDecorators": true, + "emitDecoratorMetadata": true, + "lib": ["esnext", "dom"], + "outDir": "lib", + "moduleResolution": "node" + }, + "exclude": ["node_modules", "**/*.test.*"] +} diff --git a/packages/internal-bridge/src/schemas/host.ts b/packages/internal-bridge/src/schemas/host.ts index 530fc922c..85ce997b3 100644 --- a/packages/internal-bridge/src/schemas/host.ts +++ b/packages/internal-bridge/src/schemas/host.ts @@ -18,6 +18,25 @@ export const HostRPCSchema = { }), response: z.void().nullable(), }, + COMPLETE_REQUEST: { + request: z.object({ + id: z.string(), + status: z.enum(["SUCCESS", "FAILURE"]), + response: z.object({ + status: z.number(), + headers: z.record(z.string()), + body: z.string().optional(), + }), + meta: z.object({ + environment: z.string(), + workflowId: z.string(), + organizationId: z.string(), + apiKey: z.string(), + runId: z.string(), + }), + }), + response: z.boolean(), + }, }; export type HostRPC = typeof HostRPCSchema; diff --git a/packages/internal-bridge/src/schemas/server.ts b/packages/internal-bridge/src/schemas/server.ts index b5b4ebb8d..ef3ccdf5b 100644 --- a/packages/internal-bridge/src/schemas/server.ts +++ b/packages/internal-bridge/src/schemas/server.ts @@ -1,7 +1,20 @@ -import { CustomEventSchema, TriggerMetadataSchema } from "@trigger.dev/common-schemas"; +import { + CustomEventSchema, + TriggerMetadataSchema, +} from "@trigger.dev/common-schemas"; import { z } from "zod"; export const ServerRPCSchema = { + SEND_REQUEST: { + request: z.object({ + id: z.string(), + requestId: z.string(), + service: z.string(), + endpoint: z.string(), + params: z.any(), + }), + response: z.boolean(), + }, SEND_LOG: { request: z.object({ id: z.string(), diff --git a/packages/internal-integrations/src/headers.ts b/packages/internal-integrations/src/headers.ts new file mode 100644 index 000000000..000dbe530 --- /dev/null +++ b/packages/internal-integrations/src/headers.ts @@ -0,0 +1,9 @@ +export function normalizeHeaders(headers: Headers): Record { + const normalizedHeaders: Record = {}; + + headers.forEach((value, key) => { + normalizedHeaders[key.toLowerCase()] = value; + }); + + return normalizedHeaders; +} diff --git a/packages/internal-integrations/src/index.ts b/packages/internal-integrations/src/index.ts index 7cfe07e1e..ee2269bfd 100644 --- a/packages/internal-integrations/src/index.ts +++ b/packages/internal-integrations/src/index.ts @@ -1,2 +1,3 @@ export * as github from "./github"; +export * as slack from "./slack"; export * from "./types"; diff --git a/packages/internal-integrations/src/slack/index.ts b/packages/internal-integrations/src/slack/index.ts new file mode 100644 index 000000000..47b6da898 --- /dev/null +++ b/packages/internal-integrations/src/slack/index.ts @@ -0,0 +1,85 @@ +import { normalizeHeaders } from "../headers"; +import { + NormalizedResponse, + PerformRequestOptions, + RequestIntegration, +} from "../types"; +import { PostMessageResponseSchema, PostMessageBodySchema } from "./schemas"; + +export const schemas = { + PostMessageResponseSchema, + PostMessageBodySchema, +}; + +class SlackRequestIntegration implements RequestIntegration { + perform(options: PerformRequestOptions): Promise { + switch (options.endpoint) { + case "chat.postMessage": { + return this.#postMessage(options.accessToken, options.params); + } + default: { + throw new Error(`Unknown endpoint: ${options.endpoint}`); + } + } + } + + async #postMessage( + accessToken: string, + params: any + ): Promise { + const parsedParams = PostMessageBodySchema.parse(params); + + const channelId = await this.#findChannelId( + accessToken, + parsedParams.channel + ); + + const response = await fetch("https://slack.com/api/chat.postMessage", { + method: "POST", + headers: { + Authorization: `Bearer ${accessToken}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + ...parsedParams, + channel: channelId, + }), + }); + + return { + statusCode: response.status, + headers: normalizeHeaders(response.headers), + body: await response.json(), + }; + } + + // Will use the conversations.list API (using fetch) to find the channel ID + // unless the channel is already provided in the format of a channelID (for example: "D8572TUFR" or "C01BQJZLJGZ") + async #findChannelId( + accessToken: string, + channel: string + ): Promise { + if (channel.startsWith("C") || channel.startsWith("D")) { + return channel; + } + + const response = await fetch("https://slack.com/api/conversations.list", { + method: "GET", + headers: { + Authorization: `Bearer ${accessToken}`, + }, + }); + + if (!response.ok) { + throw new Error("Failed to fetch channels"); + } + + const { channels } = await response.json(); + + const channelInfo = channels.find((c: any) => c.name === channel); + + return channelInfo?.id; + } +} + +export const requests = new SlackRequestIntegration(); diff --git a/packages/internal-integrations/src/slack/schemas.ts b/packages/internal-integrations/src/slack/schemas.ts new file mode 100644 index 000000000..6aa8fa898 --- /dev/null +++ b/packages/internal-integrations/src/slack/schemas.ts @@ -0,0 +1,21 @@ +import { z } from "zod"; + +export const PostMessageResponseSchema = z.object({ + ok: z.boolean(), + channel: z.string(), + ts: z.string(), + message: z.object({ + text: z.string(), + username: z.string(), + bot_id: z.string(), + attachments: z.array(z.unknown()), + type: z.string(), + subtype: z.string(), + ts: z.string(), + }), +}); + +export const PostMessageBodySchema = z.object({ + channel: z.string(), + text: z.string(), +}); diff --git a/packages/internal-integrations/src/types.ts b/packages/internal-integrations/src/types.ts index 6ac578382..37ad945e3 100644 --- a/packages/internal-integrations/src/types.ts +++ b/packages/internal-integrations/src/types.ts @@ -10,6 +10,12 @@ export interface NormalizedRequest { searchParams: URLSearchParams; } +export interface NormalizedResponse { + body: any; + headers: Record; + statusCode: number; +} + export interface HandleWebhookOptions { request: NormalizedRequest; secret?: string; @@ -23,6 +29,16 @@ export interface ReceivedWebhook { context?: any; } +export type PerformRequestOptions = { + accessToken: string; + endpoint: string; + params: any; +}; + +export interface RequestIntegration { + perform: (options: PerformRequestOptions) => Promise; +} + export interface WebhookIntegration { keyForSource: (source: unknown) => string; registerWebhook: (config: WebhookConfig, source: unknown) => Promise; diff --git a/packages/internal-platform/src/messages/catalogs/coordinator.ts b/packages/internal-platform/src/messages/catalogs/coordinator.ts index 29de884dd..bc7e1c4c0 100644 --- a/packages/internal-platform/src/messages/catalogs/coordinator.ts +++ b/packages/internal-platform/src/messages/catalogs/coordinator.ts @@ -1,4 +1,4 @@ -import initiateIntegrationRequest from "../schemas/initiateIntegrationRequest"; +import sendIntegrationRequest from "../schemas/sendIntegrationRequest"; import startWorklowRun from "../schemas/startWorkflowRun"; import failWorkflowRun from "../schemas/failWorkflowRun"; import completeWorkflowRun from "../schemas/completeWorkflowRun"; @@ -7,7 +7,7 @@ import triggerCustomEvent from "../schemas/triggerCustomEvent"; import awaits from "../schemas/awaits"; const Catalog = { - ...initiateIntegrationRequest, + ...sendIntegrationRequest, ...startWorklowRun, ...failWorkflowRun, ...completeWorkflowRun, diff --git a/packages/internal-platform/src/messages/catalogs/platform.ts b/packages/internal-platform/src/messages/catalogs/platform.ts index 5e84093e7..f3232a9ac 100644 --- a/packages/internal-platform/src/messages/catalogs/platform.ts +++ b/packages/internal-platform/src/messages/catalogs/platform.ts @@ -1,7 +1,9 @@ import triggerWorkflow from "../schemas/triggerWorkflow"; +import finishIntegrationRequest from "../schemas/finishIntegrationRequest"; const Catalog = { ...triggerWorkflow, + ...finishIntegrationRequest, }; export default Catalog; diff --git a/packages/internal-platform/src/messages/schemas/finishIntegrationRequest.ts b/packages/internal-platform/src/messages/schemas/finishIntegrationRequest.ts new file mode 100644 index 000000000..2493e7623 --- /dev/null +++ b/packages/internal-platform/src/messages/schemas/finishIntegrationRequest.ts @@ -0,0 +1,19 @@ +import { z } from "zod"; +import { WorkflowRunEventPropertiesSchema } from "../sharedSchemas"; + +const Catalog = { + FINISH_INTEGRATION_REQUEST: { + data: z.object({ + id: z.string(), + status: z.enum(["SUCCESS", "FAILURE"]), + response: z.object({ + status: z.number(), + headers: z.record(z.string()), + body: z.string().optional(), + }), + }), + properties: WorkflowRunEventPropertiesSchema, + }, +}; + +export default Catalog; diff --git a/packages/internal-platform/src/messages/schemas/initiateIntegrationRequest.ts b/packages/internal-platform/src/messages/schemas/initiateIntegrationRequest.ts deleted file mode 100644 index 13f214a85..000000000 --- a/packages/internal-platform/src/messages/schemas/initiateIntegrationRequest.ts +++ /dev/null @@ -1,49 +0,0 @@ -import { z } from "zod"; -import { - WorkflowEventPropertiesSchema, - RetryOptionsSchema, -} from "../sharedSchemas"; - -export const IntegrationRequestOptionsSchema = z - .object({ - retry: RetryOptionsSchema.optional(), - }) - .optional(); - -export const IntegrationRequestInfoSchema = z.object({ - url: z.string(), - method: z.enum(["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"]), - headers: z.record(z.string()), - body: z.any(), - metadata: z - .object({ - id: z.string(), - name: z.string(), - description: z.string(), - }) - .optional(), -}); - -export type IntegrationRequestInfo = z.infer< - typeof IntegrationRequestInfoSchema ->; - -export const InitiateIntegrationRequestSchema = z.object({ - id: z.string(), - integrationId: z.string(), - requestInfo: IntegrationRequestInfoSchema, - options: z - .object({ - retry: RetryOptionsSchema.optional(), - }) - .optional(), -}); - -const Catalog = { - INITIATE_INTEGRATION_REQUEST: { - data: InitiateIntegrationRequestSchema, - properties: WorkflowEventPropertiesSchema, - }, -}; - -export default Catalog; diff --git a/packages/internal-platform/src/messages/schemas/sendIntegrationRequest.ts b/packages/internal-platform/src/messages/schemas/sendIntegrationRequest.ts new file mode 100644 index 000000000..ea8c72b85 --- /dev/null +++ b/packages/internal-platform/src/messages/schemas/sendIntegrationRequest.ts @@ -0,0 +1,16 @@ +import { z } from "zod"; +import { WorkflowSendRunEventPropertiesSchema } from "../sharedSchemas"; + +const Catalog = { + SEND_INTEGRATION_REQUEST: { + data: z.object({ + id: z.string(), + service: z.string(), + endpoint: z.string(), + params: z.any(), + }), + properties: WorkflowSendRunEventPropertiesSchema, + }, +}; + +export default Catalog; diff --git a/packages/internal-platform/src/messages/sharedSchemas.ts b/packages/internal-platform/src/messages/sharedSchemas.ts index 9d15bb88d..cf2426596 100644 --- a/packages/internal-platform/src/messages/sharedSchemas.ts +++ b/packages/internal-platform/src/messages/sharedSchemas.ts @@ -7,10 +7,17 @@ export const WorkflowEventPropertiesSchema = z.object({ "x-env": z.string(), }); -export const RetryOptionsSchema = z.object({ - retries: z.number().default(10), - factor: z.number().default(2), - minTimeout: z.number().default(1 * 1000), - maxTimeout: z.number().default(60 * 1000), - randomize: z.boolean().default(true), +export const WorkflowRunEventPropertiesSchema = + WorkflowEventPropertiesSchema.extend({ + "x-workflow-run-id": z.string(), + }); + +export const WorkflowSendEventPropertiesSchema = z.object({ + "x-workflow-id": z.string(), + "x-api-key": z.string(), }); + +export const WorkflowSendRunEventPropertiesSchema = + WorkflowSendEventPropertiesSchema.extend({ + "x-workflow-run-id": z.string(), + }); diff --git a/packages/trigger-integrations/src/index.ts b/packages/trigger-integrations/src/index.ts index 7df043435..a10f73ad4 100644 --- a/packages/trigger-integrations/src/index.ts +++ b/packages/trigger-integrations/src/index.ts @@ -1,3 +1,4 @@ import * as github from "./integrations/github"; +import * as slack from "./integrations/slack"; -export { github }; +export { github, slack }; diff --git a/packages/trigger-integrations/src/integrations/slack/index.ts b/packages/trigger-integrations/src/integrations/slack/index.ts new file mode 100644 index 000000000..82f910014 --- /dev/null +++ b/packages/trigger-integrations/src/integrations/slack/index.ts @@ -0,0 +1,32 @@ +import { getTriggerRun } from "@trigger.dev/sdk"; +import { z } from "zod"; +import { slack } from "internal-integrations"; + +export type PostMessageOptions = z.infer< + typeof slack.schemas.PostMessageBodySchema +>; + +export type PostMessageResponse = z.infer< + typeof slack.schemas.PostMessageResponseSchema +>; + +export async function postMessage( + options: PostMessageOptions +): Promise { + const run = getTriggerRun(); + + if (!run) { + throw new Error("Cannot call postMessage outside of a trigger run"); + } + + const response = await run.performRequest({ + service: "slack", + endpoint: "chat.postMessage", + params: options, + response: { + schema: slack.schemas.PostMessageResponseSchema, + }, + }); + + return response.body; +} diff --git a/packages/trigger-sdk/package.json b/packages/trigger-sdk/package.json index c80d750b4..2afbb6cb5 100644 --- a/packages/trigger-sdk/package.json +++ b/packages/trigger-sdk/package.json @@ -7,14 +7,25 @@ "files": [ "dist" ], + "exports": { + ".": { + "import": "./dist/index.js", + "require": "./dist/index.js" + }, + "./package.json": "./package.json", + "./internal": { + "import": "./dist/internal/index.js", + "require": "./dist/internal/index.js" + } + }, "devDependencies": { + "@trigger.dev/common-schemas": "workspace:*", "@trigger.dev/tsconfig": "workspace:*", "@types/debug": "^4.1.7", "@types/node": "^18.11.9", "@types/uuid": "^9.0.0", "@types/ws": "^8.5.3", "internal-bridge": "workspace:*", - "@trigger.dev/common-schemas": "workspace:*", "rimraf": "^3.0.2", "tsup": "^6.5.0", "tsx": "^3.12.1" @@ -28,6 +39,7 @@ "dependencies": { "debug": "^4.3.4", "evt": "^2.4.13", + "ulid": "^2.3.0", "uuid": "^9.0.0", "ws": "^8.11.0", "zod": "^3.20.2" diff --git a/packages/trigger-sdk/src/client.ts b/packages/trigger-sdk/src/client.ts index 42473cba4..e56514feb 100644 --- a/packages/trigger-sdk/src/client.ts +++ b/packages/trigger-sdk/src/client.ts @@ -12,6 +12,14 @@ import * as pkg from "../package.json"; import { Trigger, TriggerOptions } from "./trigger"; import { TriggerContext } from "./types"; import { ContextLogger } from "./logger"; +import { triggerRunLocalStorage } from "./localStorage"; +import { ulid } from "ulid"; + +type RequestResponse = { + body?: any; + headers: Record; + status: number; +}; export class TriggerClient { #trigger: Trigger; @@ -27,6 +35,14 @@ export class TriggerClient { #retryIntervalMs: number = 3000; #logger: Logger; + #responseCompleteCallbacks = new Map< + string, + { + resolve: (output: RequestResponse) => void; + reject: (err?: any) => void; + } + >(); + constructor(trigger: Trigger, options: TriggerOptions) { this.#trigger = trigger; this.#options = options; @@ -91,6 +107,25 @@ export class TriggerClient { sender: ServerRPCSchema, receiver: HostRPCSchema, handlers: { + COMPLETE_REQUEST: async (data) => { + const requestCallbacks = this.#responseCompleteCallbacks.get(data.id); + + if (!requestCallbacks) { + throw new Error( + `Could not find request callbacks for request ID ${data.id}` + ); + } + + const { resolve, reject } = requestCallbacks; + + if (data.status === "SUCCESS") { + resolve(data.response); + } else { + reject(new Error(`Request failed: ${data.response.status}`)); + } + + return true; + }, TRIGGER_WORKFLOW: async (data) => { console.log("TRIGGER_WORKFLOW", data); @@ -119,46 +154,83 @@ export class TriggerClient { const eventData = this.#options.on.schema.parse(data.trigger.input); - // TODO: handle this better - this.#trigger.options - .run(eventData, ctx) - .then((output) => { - return serverRPC.send("COMPLETE_WORKFLOW_RUN", { - id: data.id, - output: JSON.stringify(output), - workflowId: data.meta.workflowId, - }); - }) - .catch((anyError) => { - const parseAnyError = ( - error: any - ): { - name: string; - message: string; - stackTrace?: string; - } => { - if (error instanceof Error) { - return { - name: error.name, - message: error.message, - stackTrace: error.stack, - }; - } + triggerRunLocalStorage.run( + { + performRequest: async (options) => { + const requestId = ulid(); - return { - name: "UnknownError", - message: "An unknown error occurred", + const result = new Promise( + (resolve, reject) => { + this.#responseCompleteCallbacks.set(requestId, { + resolve, + reject, + }); + } + ); + + await serverRPC.send("SEND_REQUEST", { + id: data.id, + requestId, + service: options.service, + endpoint: options.endpoint, + params: options.params, + }); + + const response = await result; + + const parsedResponse = { + ok: true, + status: response.status, + headers: response.headers, + body: options.response.schema.parse(response.body), }; - }; - const error = parseAnyError(anyError); + return parsedResponse; + }, + }, + () => { + // TODO: handle this better + this.#trigger.options + .run(eventData, ctx) + .then((output) => { + return serverRPC.send("COMPLETE_WORKFLOW_RUN", { + id: data.id, + output: JSON.stringify(output), + workflowId: data.meta.workflowId, + }); + }) + .catch((anyError) => { + const parseAnyError = ( + error: any + ): { + name: string; + message: string; + stackTrace?: string; + } => { + if (error instanceof Error) { + return { + name: error.name, + message: error.message, + stackTrace: error.stack, + }; + } - return serverRPC.send("SEND_WORKFLOW_ERROR", { - id: data.id, - workflowId: data.meta.workflowId, - error, - }); - }); + return { + name: "UnknownError", + message: "An unknown error occurred", + }; + }; + + const error = parseAnyError(anyError); + + return serverRPC.send("SEND_WORKFLOW_ERROR", { + id: data.id, + workflowId: data.meta.workflowId, + error, + }); + }); + } + ); }, }, }); diff --git a/packages/trigger-sdk/src/index.ts b/packages/trigger-sdk/src/index.ts index b81f27778..09c221ffa 100644 --- a/packages/trigger-sdk/src/index.ts +++ b/packages/trigger-sdk/src/index.ts @@ -1,2 +1,8 @@ export * from "./trigger"; export * from "./events"; + +import { triggerRunLocalStorage } from "./localStorage"; + +export function getTriggerRun() { + return triggerRunLocalStorage.getStore(); +} diff --git a/packages/trigger-sdk/src/localStorage.ts b/packages/trigger-sdk/src/localStorage.ts new file mode 100644 index 000000000..b361e8c08 --- /dev/null +++ b/packages/trigger-sdk/src/localStorage.ts @@ -0,0 +1,27 @@ +import { AsyncLocalStorage } from "node:async_hooks"; +import { z } from "zod"; + +type PerformRequestOptions = { + service: string; + params: unknown; + endpoint: string; + response: { + schema: TSchema; + }; +}; + +type PerformRequestResponse = { + ok: boolean; + status: number; + headers: Record; + body: z.infer; +}; + +type TriggerRunLocalStorage = { + performRequest: ( + options: PerformRequestOptions + ) => Promise>; +}; + +export const triggerRunLocalStorage = + new AsyncLocalStorage(); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 43f7501ca..0cf72a7ec 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -208,7 +208,7 @@ importers: '@aws-sdk/client-s3': 3.226.0 '@aws-sdk/s3-request-presigner': 3.226.0 '@cfworker/json-schema': 1.12.5 - '@codemirror/autocomplete': 6.3.4_jvia4rcxqiacrvood3734bhyuy + '@codemirror/autocomplete': 6.3.4_4npvozs3agsv66jx2b7pfvr53q '@codemirror/commands': 6.1.2 '@codemirror/lang-javascript': 6.1.1 '@codemirror/lang-json': 6.0.1 @@ -234,7 +234,7 @@ importers: '@tailwindcss/forms': 0.5.3_tailwindcss@3.1.8 '@tanstack/react-table': 8.7.0_biqbaboplfbrettd7655fr4n2y '@trigger.dev/common-schemas': link:../../packages/common-schemas - '@uiw/react-codemirror': 4.17.1_c746qxthrd2ism2rvn4crnq5om + '@uiw/react-codemirror': 4.17.1_c6ric56h4625lhpbtenqifztqq bcryptjs: 2.4.3 classnames: 2.3.2 clsx: 1.2.1 @@ -415,6 +415,23 @@ importers: '@types/node': 18.11.15 tsx: 3.12.1 + examples/send-to-slack: + specifiers: + '@trigger.dev/integrations': workspace:* + '@trigger.dev/sdk': workspace:* + '@trigger.dev/tsconfig': workspace:* + '@types/node': ^18.11.9 + tsx: ^3.12.0 + zod: ^3.20.2 + dependencies: + '@trigger.dev/integrations': link:../../packages/trigger-integrations + '@trigger.dev/sdk': link:../../packages/trigger-sdk + zod: 3.20.2 + devDependencies: + '@trigger.dev/tsconfig': link:../../config-packages/tsconfig + '@types/node': 18.11.15 + tsx: 3.12.1 + examples/smoke-test: specifiers: '@trigger.dev/integrations': workspace:* @@ -543,12 +560,14 @@ importers: rimraf: ^3.0.2 tsup: ^6.5.0 tsx: ^3.12.1 + ulid: ^2.3.0 uuid: ^9.0.0 ws: ^8.11.0 zod: ^3.20.2 dependencies: debug: 4.3.4 evt: 2.4.13 + ulid: 2.3.0 uuid: 9.0.0 ws: 8.11.0 zod: 3.20.2 @@ -2910,13 +2929,12 @@ packages: resolution: {integrity: sha512-LDQvVDv5Kb50ny2s25Fhm3d9QSZimsoUGBsUioj6MC3qbMUCuC8GPIvk/M6IvXx3lYhAs0lwWUQLb+VIEUCECw==} dev: true - /@codemirror/autocomplete/6.3.4_jvia4rcxqiacrvood3734bhyuy: + /@codemirror/autocomplete/6.3.4_4npvozs3agsv66jx2b7pfvr53q: resolution: {integrity: sha512-irxKsTSjS0OkfMWWt9YxtNK97++/E+XIHfKnRpSVfZyHzda/amYF0BR+T8mMkrGQWidx2zApxHx08GT13egyQA==} peerDependencies: '@codemirror/language': ^6.0.0 '@codemirror/state': ^6.0.0 '@codemirror/view': ^6.0.0 - '@lezer/common': ^1.0.0 dependencies: '@codemirror/language': 6.3.1 '@codemirror/state': 6.1.4 @@ -2936,7 +2954,7 @@ packages: /@codemirror/lang-javascript/6.1.1: resolution: {integrity: sha512-F4+kiuC5d5dUSJmff96tJQwpEXs/tX/4bapMRnZWW6bHKK1Fx6MunTzopkCUWRa9bF87GPmb9m7Qtg7Yv8f3uQ==} dependencies: - '@codemirror/autocomplete': 6.3.4_jvia4rcxqiacrvood3734bhyuy + '@codemirror/autocomplete': 6.3.4_4npvozs3agsv66jx2b7pfvr53q '@codemirror/language': 6.3.1 '@codemirror/lint': 6.1.0 '@codemirror/state': 6.1.4 @@ -3660,7 +3678,7 @@ packages: eslint: 8.29.0 eslint-import-resolver-node: 0.3.6 eslint-import-resolver-typescript: 3.5.2_lt3hqehuojhfcbzgzqfngbtmrq - eslint-plugin-import: 2.26.0_i656iqvetrvx3ajhg4t6psfrl4 + eslint-plugin-import: 2.26.0_qfsg7upu5e4dqco5ntekgyqxwu eslint-plugin-jest: 26.9.0_gtacs36c3cng3fu32eiajkw5qm eslint-plugin-jest-dom: 4.0.3_eslint@8.29.0 eslint-plugin-jsx-a11y: 6.6.1_eslint@8.29.0 @@ -4749,18 +4767,17 @@ packages: eslint-visitor-keys: 3.3.0 dev: true - /@uiw/codemirror-extensions-basic-setup/4.17.1_yoq5blswu3ydocenanojwujrum: + /@uiw/codemirror-extensions-basic-setup/4.17.1_mldjzacanzbudgr2aukt2yvcyy: resolution: {integrity: sha512-lFH3gFPcpKDckaioYL2KonTYeeoP7gGtaDtDai7DV5UVEyuVPlkGukKCmHz6u0ol/Krs/RTbF4ylt8cDlBT1uA==} peerDependencies: '@codemirror/autocomplete': '>=6.0.0' '@codemirror/commands': '>=6.0.0' '@codemirror/language': '>=6.0.0' - '@codemirror/lint': '>=6.0.0' '@codemirror/search': '>=6.0.0' '@codemirror/state': '>=6.0.0' '@codemirror/view': '>=6.0.0' dependencies: - '@codemirror/autocomplete': 6.3.4_jvia4rcxqiacrvood3734bhyuy + '@codemirror/autocomplete': 6.3.4_4npvozs3agsv66jx2b7pfvr53q '@codemirror/commands': 6.1.2 '@codemirror/language': 6.3.1 '@codemirror/lint': 6.1.0 @@ -4769,14 +4786,11 @@ packages: '@codemirror/view': 6.6.0 dev: false - /@uiw/react-codemirror/4.17.1_c746qxthrd2ism2rvn4crnq5om: + /@uiw/react-codemirror/4.17.1_c6ric56h4625lhpbtenqifztqq: resolution: {integrity: sha512-ah7wFhvVW/uKbQR5D12AqDK51XGZaZI1WYO9/sraZzq32TGphL5BU3vcQd1P0YEZR6YLc23+KWNi2DCQ+EEAbA==} peerDependencies: - '@babel/runtime': '>=7.11.0' '@codemirror/state': '>=6.0.0' - '@codemirror/theme-one-dark': '>=6.0.0' '@codemirror/view': '>=6.0.0' - codemirror: '>=6.0.0' react: '>=16.8.0' react-dom: '>=16.8.0' dependencies: @@ -4785,14 +4799,13 @@ packages: '@codemirror/state': 6.1.4 '@codemirror/theme-one-dark': 6.1.0 '@codemirror/view': 6.6.0 - '@uiw/codemirror-extensions-basic-setup': 4.17.1_yoq5blswu3ydocenanojwujrum - codemirror: 6.0.1_@lezer+common@1.0.2 + '@uiw/codemirror-extensions-basic-setup': 4.17.1_mldjzacanzbudgr2aukt2yvcyy + codemirror: 6.0.1 react: 18.2.0 react-dom: 18.2.0_react@18.2.0 transitivePeerDependencies: - '@codemirror/autocomplete' - '@codemirror/language' - - '@codemirror/lint' - '@codemirror/search' dev: false @@ -5915,18 +5928,16 @@ packages: engines: {node: '>=0.10.0'} dev: false - /codemirror/6.0.1_@lezer+common@1.0.2: + /codemirror/6.0.1: resolution: {integrity: sha512-J8j+nZ+CdWmIeFIGXEFbFPtpiYacFMDR8GlHK3IyHQJMCaVRfGx9NT+Hxivv1ckLWPvNdZqndbr/7lVhrf/Svg==} dependencies: - '@codemirror/autocomplete': 6.3.4_jvia4rcxqiacrvood3734bhyuy + '@codemirror/autocomplete': 6.3.4_4npvozs3agsv66jx2b7pfvr53q '@codemirror/commands': 6.1.2 '@codemirror/language': 6.3.1 '@codemirror/lint': 6.1.0 '@codemirror/search': 6.2.3 '@codemirror/state': 6.1.4 '@codemirror/view': 6.6.0 - transitivePeerDependencies: - - '@lezer/common' dev: false /collection-visit/1.0.0: @@ -7166,7 +7177,7 @@ packages: eslint: 8.29.0 eslint-import-resolver-node: 0.3.6 eslint-import-resolver-typescript: 2.7.1_lt3hqehuojhfcbzgzqfngbtmrq - eslint-plugin-import: 2.26.0_eslint@8.29.0 + eslint-plugin-import: 2.26.0_dgd2m3r3ibazmk3pmfoyze3fka eslint-plugin-jsx-a11y: 6.6.1_eslint@8.29.0 eslint-plugin-react: 7.31.8_eslint@8.29.0 eslint-plugin-react-hooks: 4.6.0_eslint@8.29.0 @@ -7212,7 +7223,7 @@ packages: dependencies: debug: 4.3.4 eslint: 8.29.0 - eslint-plugin-import: 2.26.0_eslint@8.29.0 + eslint-plugin-import: 2.26.0_dgd2m3r3ibazmk3pmfoyze3fka glob: 7.2.3 is-glob: 4.0.3 resolve: 1.22.1 @@ -7231,7 +7242,7 @@ packages: debug: 4.3.4 enhanced-resolve: 5.12.0 eslint: 8.29.0 - eslint-plugin-import: 2.26.0_i656iqvetrvx3ajhg4t6psfrl4 + eslint-plugin-import: 2.26.0_qfsg7upu5e4dqco5ntekgyqxwu get-tsconfig: 4.2.0 globby: 13.1.2 is-core-module: 2.11.0 @@ -7241,7 +7252,7 @@ packages: - supports-color dev: true - /eslint-module-utils/2.7.4_jnakocfte2jywffz4vixv5kpsq: + /eslint-module-utils/2.7.4_457k6wn3tjkduxg6oi6e76gicy: resolution: {integrity: sha512-j4GT+rqzCoRKHwURX7pddtIPGySnX9Si/cgMI5ztrcqOPtk5dDEeZ34CQVPphnqkJytlc97Vuk05Um2mJ3gEQA==} engines: {node: '>=4'} peerDependencies: @@ -7266,11 +7277,12 @@ packages: debug: 3.2.7 eslint: 8.29.0 eslint-import-resolver-node: 0.3.6 + eslint-import-resolver-typescript: 2.7.1_lt3hqehuojhfcbzgzqfngbtmrq transitivePeerDependencies: - supports-color dev: true - /eslint-module-utils/2.7.4_uplb3bqnui63takc5j27khdnpm: + /eslint-module-utils/2.7.4_wbv6cezew2qbikiravago3ef2u: resolution: {integrity: sha512-j4GT+rqzCoRKHwURX7pddtIPGySnX9Si/cgMI5ztrcqOPtk5dDEeZ34CQVPphnqkJytlc97Vuk05Um2mJ3gEQA==} engines: {node: '>=4'} peerDependencies: @@ -7291,9 +7303,11 @@ packages: eslint-import-resolver-webpack: optional: true dependencies: + '@typescript-eslint/parser': 5.45.1_s5ps7njkmjlaqajutnox5ntcla debug: 3.2.7 eslint: 8.29.0 eslint-import-resolver-node: 0.3.6 + eslint-import-resolver-typescript: 3.5.2_lt3hqehuojhfcbzgzqfngbtmrq transitivePeerDependencies: - supports-color dev: true @@ -7318,37 +7332,7 @@ packages: regexpp: 3.2.0 dev: true - /eslint-plugin-import/2.26.0_eslint@8.29.0: - resolution: {integrity: sha512-hYfi3FXaM8WPLf4S1cikh/r4IxnO6zrhZbEGz2b660EJRbuxgpDS5gkCuYgGWg2xxh2rBuIr4Pvhve/7c31koA==} - engines: {node: '>=4'} - peerDependencies: - '@typescript-eslint/parser': '*' - eslint: ^2 || ^3 || ^4 || ^5 || ^6 || ^7.2.0 || ^8 - peerDependenciesMeta: - '@typescript-eslint/parser': - optional: true - dependencies: - array-includes: 3.1.6 - array.prototype.flat: 1.3.1 - debug: 2.6.9 - doctrine: 2.1.0 - eslint: 8.29.0 - eslint-import-resolver-node: 0.3.6 - eslint-module-utils: 2.7.4_uplb3bqnui63takc5j27khdnpm - has: 1.0.3 - is-core-module: 2.11.0 - is-glob: 4.0.3 - minimatch: 3.1.2 - object.values: 1.1.6 - resolve: 1.22.1 - tsconfig-paths: 3.14.1 - transitivePeerDependencies: - - eslint-import-resolver-typescript - - eslint-import-resolver-webpack - - supports-color - dev: true - - /eslint-plugin-import/2.26.0_i656iqvetrvx3ajhg4t6psfrl4: + /eslint-plugin-import/2.26.0_dgd2m3r3ibazmk3pmfoyze3fka: resolution: {integrity: sha512-hYfi3FXaM8WPLf4S1cikh/r4IxnO6zrhZbEGz2b660EJRbuxgpDS5gkCuYgGWg2xxh2rBuIr4Pvhve/7c31koA==} engines: {node: '>=4'} peerDependencies: @@ -7365,7 +7349,38 @@ packages: doctrine: 2.1.0 eslint: 8.29.0 eslint-import-resolver-node: 0.3.6 - eslint-module-utils: 2.7.4_jnakocfte2jywffz4vixv5kpsq + eslint-module-utils: 2.7.4_457k6wn3tjkduxg6oi6e76gicy + has: 1.0.3 + is-core-module: 2.11.0 + is-glob: 4.0.3 + minimatch: 3.1.2 + object.values: 1.1.6 + resolve: 1.22.1 + tsconfig-paths: 3.14.1 + transitivePeerDependencies: + - eslint-import-resolver-typescript + - eslint-import-resolver-webpack + - supports-color + dev: true + + /eslint-plugin-import/2.26.0_qfsg7upu5e4dqco5ntekgyqxwu: + resolution: {integrity: sha512-hYfi3FXaM8WPLf4S1cikh/r4IxnO6zrhZbEGz2b660EJRbuxgpDS5gkCuYgGWg2xxh2rBuIr4Pvhve/7c31koA==} + engines: {node: '>=4'} + peerDependencies: + '@typescript-eslint/parser': '*' + eslint: ^2 || ^3 || ^4 || ^5 || ^6 || ^7.2.0 || ^8 + peerDependenciesMeta: + '@typescript-eslint/parser': + optional: true + dependencies: + '@typescript-eslint/parser': 5.45.1_s5ps7njkmjlaqajutnox5ntcla + array-includes: 3.1.6 + array.prototype.flat: 1.3.1 + debug: 2.6.9 + doctrine: 2.1.0 + eslint: 8.29.0 + eslint-import-resolver-node: 0.3.6 + eslint-module-utils: 2.7.4_wbv6cezew2qbikiravago3ef2u has: 1.0.3 is-core-module: 2.11.0 is-glob: 4.0.3