diff --git a/apps/webapp/app/consts.ts b/apps/webapp/app/consts.ts index fc602a634..f2eafb63c 100644 --- a/apps/webapp/app/consts.ts +++ b/apps/webapp/app/consts.ts @@ -1,3 +1,4 @@ export const LIVE_ENVIRONMENT = "live"; export const DEV_ENVIRONMENT = "development"; export const MAX_LIVE_PROJECTS = 1; +export const DEFAULT_MAX_CONCURRENT_RUNS = 100; diff --git a/apps/webapp/app/platform/zodWorker.server.ts b/apps/webapp/app/platform/zodWorker.server.ts index c967cad4e..b05b3d678 100644 --- a/apps/webapp/app/platform/zodWorker.server.ts +++ b/apps/webapp/app/platform/zodWorker.server.ts @@ -15,7 +15,7 @@ import { logger } from "~/services/logger"; export type ZodTasks = { [K in keyof TConsumerSchema]: { - queueName?: string; + queueName?: string | ((payload: z.infer) => string); priority?: number; maxAttempts?: number; jobKeyMode?: "replace" | "preserve_run_at" | "unsafe_dedupe"; @@ -88,6 +88,10 @@ export class ZodWorker { ...task, }; + if (typeof task.queueName === "function") { + opts.queueName = task.queueName(payload); + } + const job = await this.#runner.addJob(identifier as string, payload, opts); logger.debug("Enqueued worker task", { diff --git a/apps/webapp/app/services/jobs/registerJob.server.ts b/apps/webapp/app/services/jobs/registerJob.server.ts index af1224fcb..fd754acd6 100644 --- a/apps/webapp/app/services/jobs/registerJob.server.ts +++ b/apps/webapp/app/services/jobs/registerJob.server.ts @@ -11,6 +11,7 @@ import type { LocalAuthConnectionConfig, TriggerMetadata, } from "@trigger.dev/internal"; +import { DEFAULT_MAX_CONCURRENT_RUNS } from "~/consts"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; import type { AuthenticatedEnvironment } from "../apiAuth.server"; @@ -45,8 +46,6 @@ export class RegisterJobService { jobResponse ); - // TODO: deliver internal event that will prepare the main trigger - await workerQueue.enqueue( "prepareJobInstance", { id: jobInstance.id }, @@ -140,6 +139,41 @@ export class RegisterJobService { }, }); + // Upsert the JobQueue + const queueName = + typeof metadata.queue === "string" + ? metadata.queue + : typeof metadata.queue === "object" + ? metadata.queue.name + : "default"; + + const jobQueue = await this.#prismaClient.jobQueue.upsert({ + where: { + environmentId_name: { + environmentId: environment.id, + name: queueName, + }, + }, + create: { + environment: { + connect: { + id: environment.id, + }, + }, + name: queueName, + maxJobs: + typeof metadata.queue === "object" + ? metadata.queue.maxConcurrent || DEFAULT_MAX_CONCURRENT_RUNS + : DEFAULT_MAX_CONCURRENT_RUNS, + }, + update: { + maxJobs: + typeof metadata.queue === "object" + ? metadata.queue.maxConcurrent || DEFAULT_MAX_CONCURRENT_RUNS + : DEFAULT_MAX_CONCURRENT_RUNS, + }, + }); + // Upsert the JobInstance const jobInstance = await this.#prismaClient.jobInstance.upsert({ where: { @@ -175,11 +209,21 @@ export class RegisterJobService { id: environment.projectId, }, }, + queue: { + connect: { + id: jobQueue.id, + }, + }, version: metadata.version, trigger: metadata.trigger, }, update: { trigger: metadata.trigger, + queue: { + connect: { + id: jobQueue.id, + }, + }, }, include: { connections: { diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index 466563e74..7300fad28 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/apps/webapp/app/services/runs/createRun.server.ts @@ -28,7 +28,13 @@ export class CreateRunService { }, }); - const execution = await this.#prismaClient.$transaction(async (prisma) => { + const jobQueue = await this.#prismaClient.jobQueue.findUniqueOrThrow({ + where: { + id: jobInstance.queueId, + }, + }); + + const run = await this.#prismaClient.$transaction(async (prisma) => { // Get the current max number for the given jobId const currentMaxNumber = await prisma.jobRun.aggregate({ where: { jobId: job.id }, @@ -49,14 +55,15 @@ export class CreateRunService { organization: { connect: { id: environment.organizationId } }, project: { connect: { id: environment.projectId } }, endpoint: { connect: { id: endpoint.id } }, + queue: { connect: { id: jobQueue.id } }, }, }); }); await workerQueue.enqueue("startRun", { - id: execution.id, + id: run.id, }); - return execution; + return run; } } diff --git a/apps/webapp/app/services/runs/postRun.server.ts b/apps/webapp/app/services/runs/postRun.server.ts index 246cec353..e7113291c 100644 --- a/apps/webapp/app/services/runs/postRun.server.ts +++ b/apps/webapp/app/services/runs/postRun.server.ts @@ -1,12 +1,12 @@ import type { CreateRunBody } from "@trigger.dev/internal"; import type { PrismaClient } from "~/db.server"; import { prisma } from "~/db.server"; -import { AuthenticatedEnvironment } from "../apiAuth.server"; +import type { AuthenticatedEnvironment } from "../apiAuth.server"; import { CreateRunService } from "./createRun.server"; export class PostRunService { #prismaClient: PrismaClient; - #createExecutionService = new CreateRunService(); + #createRunService = new CreateRunService(); constructor(prismaClient: PrismaClient = prisma) { this.#prismaClient = prismaClient; @@ -44,7 +44,7 @@ export class PostRunService { }, }); - return this.#createExecutionService.call({ + return this.#createRunService.call({ environment, job, jobInstance, diff --git a/apps/webapp/app/services/runs/resumeTask.server.ts b/apps/webapp/app/services/runs/resumeTask.server.ts index 757667b30..9f443ebb6 100644 --- a/apps/webapp/app/services/runs/resumeTask.server.ts +++ b/apps/webapp/app/services/runs/resumeTask.server.ts @@ -33,6 +33,7 @@ export class ResumeTaskService { }, }, }, + queue: true, }, }, }, @@ -85,8 +86,19 @@ export class ResumeTaskService { completedAt: new Date(), status: "SUCCESS", output: results.output ?? undefined, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, }, }); + + await workerQueue.enqueue("startQueuedRuns", { + id: run.queueId, + }); } if (results.task) { @@ -108,9 +120,39 @@ export class ResumeTaskService { completedAt: new Date(), status: "FAILURE", output: { message: error.message, stack: error.stack }, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, + }, + }); + } else { + await this.#prismaClient.jobRun.update({ + where: { id }, + data: { + completedAt: new Date(), + status: "FAILURE", + output: { + message: error instanceof Error ? error.message : "Unknown Error", + stack: error instanceof Error ? error.stack : undefined, + }, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, }, }); } + + await workerQueue.enqueue("startQueuedRuns", { + id: run.queueId, + }); } } } diff --git a/apps/webapp/app/services/runs/startQueuedRuns.server.ts b/apps/webapp/app/services/runs/startQueuedRuns.server.ts new file mode 100644 index 000000000..4bd5f6bc2 --- /dev/null +++ b/apps/webapp/app/services/runs/startQueuedRuns.server.ts @@ -0,0 +1,50 @@ +import type { PrismaClient } from "~/db.server"; +import { prisma } from "~/db.server"; +import { workerQueue } from "../worker.server"; + +export class StartQueuedRunsService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call(id: string) { + const queue = await this.#prismaClient.jobQueue.findUnique({ + where: { id }, + include: { + runs: { + where: { + status: "QUEUED", + }, + orderBy: { + queuedAt: "asc", + }, + take: 1, + }, + }, + }); + + if (!queue) { + return; + } + + if (queue.runs.length === 0) { + return; + } + + if (queue.jobCount >= queue.maxJobs) { + return; + } + + const run = queue.runs[0]; + + if (!run) { + return; + } + + await workerQueue.enqueue("startRun", { + id: run.id, + }); + } +} diff --git a/apps/webapp/app/services/runs/startRun.server.ts b/apps/webapp/app/services/runs/startRun.server.ts index 7da0ac1ed..70b9595be 100644 --- a/apps/webapp/app/services/runs/startRun.server.ts +++ b/apps/webapp/app/services/runs/startRun.server.ts @@ -5,6 +5,7 @@ import type { JobConnectionWithApiConnection } from "~/models/jobConnection.serv import { resolveJobConnections } from "~/models/jobConnection.server"; import { ClientApi, ClientApiError } from "../clientApi.server"; import { workerQueue } from "../worker.server"; +import { logger } from "../logger"; export class StartRunService { #prismaClient: PrismaClient; @@ -14,44 +15,98 @@ export class StartRunService { } public async call(id: string) { - const run = await this.#prismaClient.jobRun.findUniqueOrThrow({ - where: { id }, - include: { - jobInstance: { + const run = await this.#prismaClient.$transaction(async (tx) => { + const run = await tx.jobRun.findUnique({ + where: { id }, + include: { + queue: true, + }, + }); + + if (!run) { + return; + } + + if (run.status !== "PENDING" && run.status !== "QUEUED") { + return; + } + + // Check the JobQueue to make sure we can start the run + if (run.queue.jobCount >= run.queue.maxJobs) { + // Set the run status to QUEUED and return + return tx.jobRun.update({ + where: { id }, + data: { + status: "QUEUED", + queuedAt: new Date(), + }, include: { - endpoint: true, - job: true, - connections: { - include: { - apiConnection: { - include: { - dataReference: true, - }, + eventLog: true, + }, + }); + } else { + // Start the jobRun and increment the jobCount + return tx.jobRun.update({ + where: { id }, + data: { + status: "STARTED", + startedAt: new Date(), + queue: { + update: { + jobCount: { + increment: 1, }, }, - where: { - key: { not: "__trigger" }, + }, + }, + include: { + eventLog: true, + }, + }); + } + }); + + if (!run) { + logger.debug(`Run ${id} not found, aborting start run`, { id }); + + return; + } + + if (run.status === "QUEUED") { + logger.debug(`Run ${id} queued, aborting start run`, { id }); + + return; + } + + await workerQueue.enqueue("startQueuedRuns", { + id: run.queueId, + }); + + const jobInstance = await this.#prismaClient.jobInstance.findUniqueOrThrow({ + where: { id: run.jobInstanceId }, + include: { + endpoint: true, + job: true, + environment: true, + organization: true, + connections: { + include: { + apiConnection: { + include: { + dataReference: true, }, }, }, }, - environment: true, - eventLog: true, - organization: true, }, }); // If any of the connections are missing, we can't start the execution const connections: Array = - run.jobInstance.connections.filter( + jobInstance.connections.filter( (c) => c.apiConnection != null || c.usesLocalAuth ); - const client = new ClientApi( - run.environment.apiKey, - run.jobInstance.endpoint.url - ); - const startedAt = run.startedAt ?? new Date(); await this.#prismaClient.jobRun.update({ @@ -64,19 +119,24 @@ export class StartRunService { const event = ApiEventLogSchema.parse(run.eventLog); + const client = new ClientApi( + jobInstance.environment.apiKey, + jobInstance.endpoint.url + ); + try { const results = await client.executeJob({ event, job: { - id: run.jobInstance.job.slug, - version: run.jobInstance.version, + id: jobInstance.job.slug, + version: jobInstance.version, }, context: { id: run.id, - environment: run.environment.slug, - organization: run.organization.slug, + environment: jobInstance.environment.slug, + organization: jobInstance.organization.slug, isTest: run.isTest, - version: run.jobInstance.version, + version: jobInstance.version, startedAt, }, connections: await resolveJobConnections(connections), @@ -89,8 +149,19 @@ export class StartRunService { completedAt: new Date(), status: "SUCCESS", output: results.output ?? undefined, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, }, }); + + await workerQueue.enqueue("startQueuedRuns", { + id: run.queueId, + }); } if (results.task) { @@ -112,9 +183,39 @@ export class StartRunService { completedAt: new Date(), status: "FAILURE", output: { message: error.message, stack: error.stack }, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, + }, + }); + } else { + await this.#prismaClient.jobRun.update({ + where: { id }, + data: { + completedAt: new Date(), + status: "FAILURE", + output: { + message: error instanceof Error ? error.message : "Unknown Error", + stack: error instanceof Error ? error.stack : undefined, + }, + queue: { + update: { + jobCount: { + decrement: 1, + }, + }, + }, }, }); } + + await workerQueue.enqueue("startQueuedRuns", { + id: run.queueId, + }); } } } diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index 22f0e8d68..9e00b5aa3 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -11,6 +11,7 @@ import { ResumeTaskService } from "./runs/resumeTask.server"; import { StartRunService } from "./runs/startRun.server"; import { DeliverHttpSourceRequestService } from "./sources/deliverHttpSourceRequest.server"; import { PrepareTriggerVariantService } from "./endpoints/prepareTriggerVariant.server"; +import { StartQueuedRunsService } from "./runs/startQueuedRuns.server"; const workerCatalog = { organizationCreated: z.object({ id: z.string() }), @@ -42,6 +43,7 @@ const workerCatalog = { endpointId: z.string(), job: GetJobResponseSchema, }), + startQueuedRuns: z.object({ id: z.string() }), }; let workerQueue: ZodWorker; @@ -77,6 +79,15 @@ function getWorkerQueue() { }, schema: workerCatalog, tasks: { + startQueuedRuns: { + maxAttempts: 3, + queueName: (payload) => `queue:${payload.id}`, + handler: async (payload, job) => { + const service = new StartQueuedRunsService(); + + await service.call(payload.id); + }, + }, registerJob: { maxAttempts: 3, handler: async (payload, job) => { diff --git a/apps/webapp/package.json b/apps/webapp/package.json index 3698f551e..d1e901d44 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -63,7 +63,7 @@ "@nangohq/node": "^0.8.4", "@octokit/webhooks": "^10.4.0", "@octokit/webhooks-methods": "^3.0.2", - "@prisma/client": "^4.3.0", + "@prisma/client": "^4.13.0", "@radix-ui/react-dialog": "^1.0.3", "@radix-ui/react-popover": "^1.0.5", "@react-email/head": "^0.0.2", @@ -202,7 +202,7 @@ "postcss": "^8.4.14", "prettier": "^2.6.2", "prettier-plugin-tailwindcss": "^0.1.13", - "prisma": "^4.3.0", + "prisma": "^4.13.0", "react-date-range": "^1.4.0", "rimraf": "^3.0.2", "start-server-and-test": "^1.14.0", diff --git a/apps/webapp/prisma/migrations/20230505085501_add_queue_columns_to_job/migration.sql b/apps/webapp/prisma/migrations/20230505085501_add_queue_columns_to_job/migration.sql new file mode 100644 index 000000000..d180bc0db --- /dev/null +++ b/apps/webapp/prisma/migrations/20230505085501_add_queue_columns_to_job/migration.sql @@ -0,0 +1,3 @@ +-- AlterTable +ALTER TABLE "Job" ADD COLUMN "maxConcurrentRuns" INTEGER, +ADD COLUMN "queueName" TEXT; diff --git a/apps/webapp/prisma/migrations/20230505085546_move_queue_columns_to_job_instance/migration.sql b/apps/webapp/prisma/migrations/20230505085546_move_queue_columns_to_job_instance/migration.sql new file mode 100644 index 000000000..6531130b8 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230505085546_move_queue_columns_to_job_instance/migration.sql @@ -0,0 +1,14 @@ +/* + Warnings: + + - You are about to drop the column `maxConcurrentRuns` on the `Job` table. All the data in the column will be lost. + - You are about to drop the column `queueName` on the `Job` table. All the data in the column will be lost. + +*/ +-- AlterTable +ALTER TABLE "Job" DROP COLUMN "maxConcurrentRuns", +DROP COLUMN "queueName"; + +-- AlterTable +ALTER TABLE "JobInstance" ADD COLUMN "maxConcurrentRuns" INTEGER, +ADD COLUMN "queueName" TEXT; diff --git a/apps/webapp/prisma/migrations/20230505085931_add_queued_job_run_status/migration.sql b/apps/webapp/prisma/migrations/20230505085931_add_queued_job_run_status/migration.sql new file mode 100644 index 000000000..b19e7bd80 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230505085931_add_queued_job_run_status/migration.sql @@ -0,0 +1,2 @@ +-- AlterEnum +ALTER TYPE "JobRunStatus" ADD VALUE 'QUEUED'; diff --git a/apps/webapp/prisma/migrations/20230505091221_create_queue_model/migration.sql b/apps/webapp/prisma/migrations/20230505091221_create_queue_model/migration.sql new file mode 100644 index 000000000..1dd5a7fc5 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230505091221_create_queue_model/migration.sql @@ -0,0 +1,41 @@ +/* + Warnings: + + - You are about to drop the column `maxConcurrentRuns` on the `JobInstance` table. All the data in the column will be lost. + - You are about to drop the column `queueName` on the `JobInstance` table. All the data in the column will be lost. + - Added the required column `queueId` to the `JobInstance` table without a default value. This is not possible if the table is not empty. + - Added the required column `queueId` to the `JobRun` table without a default value. This is not possible if the table is not empty. + +*/ +-- AlterTable +ALTER TABLE "JobInstance" DROP COLUMN "maxConcurrentRuns", +DROP COLUMN "queueName", +ADD COLUMN "queueId" TEXT NOT NULL; + +-- AlterTable +ALTER TABLE "JobRun" ADD COLUMN "queueId" TEXT NOT NULL; + +-- CreateTable +CREATE TABLE "JobQueue" ( + "id" TEXT NOT NULL, + "name" TEXT NOT NULL, + "environmentId" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + "jobCount" INTEGER NOT NULL DEFAULT 0, + "maxJobs" INTEGER NOT NULL DEFAULT 100, + + CONSTRAINT "JobQueue_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "JobQueue_environmentId_name_key" ON "JobQueue"("environmentId", "name"); + +-- AddForeignKey +ALTER TABLE "JobInstance" ADD CONSTRAINT "JobInstance_queueId_fkey" FOREIGN KEY ("queueId") REFERENCES "JobQueue"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "JobQueue" ADD CONSTRAINT "JobQueue_environmentId_fkey" FOREIGN KEY ("environmentId") REFERENCES "RuntimeEnvironment"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "JobRun" ADD CONSTRAINT "JobRun_queueId_fkey" FOREIGN KEY ("queueId") REFERENCES "JobQueue"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/apps/webapp/prisma/migrations/20230505092917_add_queued_at_to_job_runs/migration.sql b/apps/webapp/prisma/migrations/20230505092917_add_queued_at_to_job_runs/migration.sql new file mode 100644 index 000000000..fac02caa1 --- /dev/null +++ b/apps/webapp/prisma/migrations/20230505092917_add_queued_at_to_job_runs/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "JobRun" ADD COLUMN "queuedAt" TIMESTAMP(3); diff --git a/apps/webapp/prisma/schema.prisma b/apps/webapp/prisma/schema.prisma index 3b392037d..5657459aa 100644 --- a/apps/webapp/prisma/schema.prisma +++ b/apps/webapp/prisma/schema.prisma @@ -181,6 +181,7 @@ model RuntimeEnvironment { requestDeliveries HttpSourceRequestDelivery[] jobEventRules JobEventRule[] jobAliases JobAlias[] + JobQueue JobQueue[] @@unique([projectId, slug, orgMemberId]) } @@ -1007,6 +1008,9 @@ model JobInstance { project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) projectId String + queue JobQueue @relation(fields: [queueId], references: [id]) + queueId String + ready Boolean @default(false) latest Boolean @default(false) @@ -1022,6 +1026,25 @@ model JobInstance { @@unique([jobId, version, endpointId]) } +model JobQueue { + id String @id @default(cuid()) + name String + + environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) + environmentId String + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + jobCount Int @default(0) + maxJobs Int @default(100) + + runs JobRun[] + instances JobInstance[] + + @@unique([environmentId, name]) +} + model JobTriggerVariant { id String @id @default(cuid()) slug String @@ -1172,8 +1195,12 @@ model JobRun { project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) projectId String + queue JobQueue @relation(fields: [queueId], references: [id]) + queueId String + createdAt DateTime @default(now()) updatedAt DateTime @updatedAt + queuedAt DateTime? startedAt DateTime? completedAt DateTime? @@ -1192,6 +1219,7 @@ model JobRun { enum JobRunStatus { PENDING + QUEUED STARTED SUCCESS FAILURE diff --git a/examples/nextjs-example/src/pages/api/trigger.ts b/examples/nextjs-example/src/pages/api/trigger.ts index 3ebe3d1fb..3a1e285a3 100644 --- a/examples/nextjs-example/src/pages/api/trigger.ts +++ b/examples/nextjs-example/src/pages/api/trigger.ts @@ -5,12 +5,12 @@ import { TriggerClient, } from "@trigger.dev/sdk"; import { github } from "@trigger.dev/github"; -import { slack } from "@trigger.dev/slack"; +import { slack as slackConnection } from "@trigger.dev/slack"; import type { NextApiRequest, NextApiResponse } from "next"; import { z } from "zod"; const gh = github({ id: "github" }); -const sl = slack({ id: "my-slack-new" }); +const slack = slackConnection({ id: "my-slack-new" }); const client = new TriggerClient("nextjs", { apiKey: process.env.TRIGGER_API_KEY, @@ -20,24 +20,43 @@ const client = new TriggerClient("nextjs", { }); new Job({ - id: "comment-on-new-issues", - name: "Comment on New GitHub issues", + id: "alert-on-new-github-issues", + name: "Alert on new GitHub issues", version: "0.1.1", - logLevel: "debug", connections: { - sl, + slack, }, trigger: gh.triggers.onIssueOpened({ repo: "ericallam/basic-starter-100k", }), run: async (event, io, ctx) => { - await io.sl.postMessage("Slack 📝", { + await io.slack.postMessage("Slack 📝", { text: `New Issue opened: ${event.issue.html_url}`, channel: "C04GWUTDC3W", }); }, }).attachTo(client); +new Job({ + id: "alert-on-new-github-issues-2", + name: "Alert on new GitHub issues 2", + version: "0.1.1", + trigger: gh.triggers.onIssueOpened({ + repo: "ericallam/basic-starter-100k", + }), + run: async (event, io, ctx) => {}, +}).attachTo(client); + +new Job({ + id: "alert-on-new-github-stars", + name: "Alert on new GitHub stars", + version: "0.1.1", + trigger: gh.triggers.onStar({ + repo: "ericallam/basic-starter-100k", + }), + run: async (event, io, ctx) => {}, +}).attachTo(client); + // const notifySlackONNewCommentsJob = new Job({ // id: "notify-slack-on-new-comments", // name: "Notify Slack on new GitHub comments", diff --git a/integrations/github/src/index.ts b/integrations/github/src/index.ts index eefae662b..28c0243cb 100644 --- a/integrations/github/src/index.ts +++ b/integrations/github/src/index.ts @@ -2,6 +2,7 @@ import { IssueCommentEvent, IssuesEvent, IssuesOpenedEvent, + StarEvent, } from "@octokit/webhooks-types"; import { Connection, @@ -71,11 +72,7 @@ function createTriggers(connection: Connection) { action: ["opened"], } ), - onIssueComment: buildRepoWebhookTrigger( - "On Issue Comment", - "issue_comment", - connection - ), + onStar: buildRepoWebhookTrigger("On Star", "star", connection), }; } diff --git a/packages/internal/src/schemas/api.ts b/packages/internal/src/schemas/api.ts index a1851cf73..55d9f881c 100644 --- a/packages/internal/src/schemas/api.ts +++ b/packages/internal/src/schemas/api.ts @@ -71,6 +71,13 @@ export const PongResponseSchema = z.object({ message: z.literal("PONG"), }); +export const QueueOptionsSchema = z.object({ + name: z.string(), + maxConcurrent: z.number().optional(), +}); + +export type QueueOptions = z.infer; + export const JobSchema = z.object({ id: z.string(), name: z.string(), @@ -78,6 +85,7 @@ export const JobSchema = z.object({ trigger: TriggerMetadataSchema, connections: z.record(ConnectionConfigSchema), internal: z.boolean().default(false), + queue: z.union([QueueOptionsSchema, z.string()]).optional(), }); export type JobMetadata = z.infer; diff --git a/packages/trigger-sdk/src/job.ts b/packages/trigger-sdk/src/job.ts index 757af20b3..06ca309bf 100644 --- a/packages/trigger-sdk/src/job.ts +++ b/packages/trigger-sdk/src/job.ts @@ -1,4 +1,9 @@ -import { ConnectionConfig, JobMetadata, LogLevel } from "@trigger.dev/internal"; +import { + ConnectionConfig, + JobMetadata, + LogLevel, + QueueOptions, +} from "@trigger.dev/internal"; import { Connection, IOWithConnections } from "./connections"; import { TriggerClient } from "./triggerClient"; import type { TriggerContext, Trigger, TriggerEventType } from "./types"; @@ -13,6 +18,7 @@ export type JobOptions< trigger: TTrigger; logLevel?: LogLevel; connections?: TConnections; + queue?: QueueOptions | string; run: ( event: TriggerEventType, @@ -110,6 +116,7 @@ export class Job< version: this.version, trigger: this.trigger.toJSON(), connections: this.connections, + queue: this.options.queue, internal, }; } diff --git a/packages/trigger-sdk/src/triggers/externalSource.ts b/packages/trigger-sdk/src/triggers/externalSource.ts index 460940531..cb7a2bc9a 100644 --- a/packages/trigger-sdk/src/triggers/externalSource.ts +++ b/packages/trigger-sdk/src/triggers/externalSource.ts @@ -92,7 +92,7 @@ export class ExternalSource< io: IOWithConnections<{ client: TConnection }>, ctx: TriggerContext ) { - await this.options.register(io, ctx); + return await this.options.register(io, ctx); } async handle( @@ -165,6 +165,10 @@ export class ExternalSourceEventTrigger< connections: { client: this.options.source.connection, }, + queue: { + name: `internal:${triggerClient.name}`, + maxConcurrent: 1, + }, run: async (event, io, ctx) => { return await this.options.source.register(io, ctx); }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b3fdb0375..2863e633f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -63,7 +63,7 @@ importers: '@octokit/webhooks': ^10.4.0 '@octokit/webhooks-methods': ^3.0.2 '@octokit/webhooks-types': ^6.10.0 - '@prisma/client': ^4.3.0 + '@prisma/client': ^4.13.0 '@radix-ui/react-dialog': ^1.0.3 '@radix-ui/react-popover': ^1.0.5 '@react-email/head': ^0.0.2 @@ -168,7 +168,7 @@ importers: prettier-plugin-tailwindcss: ^0.1.13 pretty-bytes: ^6.0.0 prism-react-renderer: ^1.3.5 - prisma: ^4.3.0 + prisma: ^4.13.0 prismjs: ^1.29.0 pulsar-client: 1.7.0 qs: ^6.11.0 @@ -229,7 +229,7 @@ importers: '@nangohq/node': 0.8.4 '@octokit/webhooks': 10.5.1 '@octokit/webhooks-methods': 3.0.2 - '@prisma/client': 4.8.1_prisma@4.8.1 + '@prisma/client': 4.13.0_prisma@4.13.0 '@radix-ui/react-dialog': 1.0.3_ib3m5ricvtkl2cll7qpr2f6lvq '@radix-ui/react-popover': 1.0.5_ib3m5ricvtkl2cll7qpr2f6lvq '@react-email/head': 0.0.2 @@ -367,7 +367,7 @@ importers: postcss: 8.4.21 prettier: 2.8.2 prettier-plugin-tailwindcss: 0.1.13_prettier@2.8.2 - prisma: 4.8.1 + prisma: 4.13.0 rimraf: 3.0.2 start-server-and-test: 1.15.2 tailwindcss: 3.1.8_aesdjsunmf4wiehhujt67my7tu @@ -4998,8 +4998,8 @@ packages: tslib: 2.4.1 dev: true - /@prisma/client/4.8.1_prisma@4.8.1: - resolution: {integrity: sha512-d4xhZhETmeXK/yZ7K0KcVOzEfI5YKGGEr4F5SBV04/MU4ncN/HcE28sy3e4Yt8UFW0ZuImKFQJE+9rWt9WbGSQ==} + /@prisma/client/4.13.0_prisma@4.13.0: + resolution: {integrity: sha512-YaiiICcRB2hatxsbnfB66uWXjcRw3jsZdlAVxmx0cFcTc/Ad/sKdHCcWSnqyDX47vAewkjRFwiLwrOUjswVvmA==} engines: {node: '>=14.17'} requiresBuild: true peerDependencies: @@ -5008,16 +5008,16 @@ packages: prisma: optional: true dependencies: - '@prisma/engines-version': 4.8.0-61.d6e67a83f971b175a593ccc12e15c4a757f93ffe - prisma: 4.8.1 + '@prisma/engines-version': 4.13.0-50.1e7af066ee9cb95cf3a403c78d9aab3e6b04f37a + prisma: 4.13.0 dev: false - /@prisma/engines-version/4.8.0-61.d6e67a83f971b175a593ccc12e15c4a757f93ffe: - resolution: {integrity: sha512-MHSOSexomRMom8QN4t7bu87wPPD+pa+hW9+71JnVcF3DqyyO/ycCLhRL1we3EojRpZxKvuyGho2REQsMCvxcJw==} + /@prisma/engines-version/4.13.0-50.1e7af066ee9cb95cf3a403c78d9aab3e6b04f37a: + resolution: {integrity: sha512-fsQlbkhPJf08JOzKoyoD9atdUijuGBekwoOPZC3YOygXEml1MTtgXVpnUNchQlRSY82OQ6pSGQ9PxUe4arcSLQ==} dev: false - /@prisma/engines/4.8.1: - resolution: {integrity: sha512-93tctjNXcIS+i/e552IO6tqw17sX8liivv8WX9lDMCpEEe3ci+nT9F+1oHtAafqruXLepKF80i/D20Mm+ESlOw==} + /@prisma/engines/4.13.0: + resolution: {integrity: sha512-HrniowHRZXHuGT9XRgoXEaP2gJLXM5RMoItaY2PkjvuZ+iHc0Zjbm/302MB8YsPdWozAPHHn+jpFEcEn71OgPw==} requiresBuild: true /@radix-ui/primitive/1.0.0: @@ -14907,13 +14907,13 @@ packages: react: 18.2.0 dev: false - /prisma/4.8.1: - resolution: {integrity: sha512-ZMLnSjwulIeYfaU1O6/LF6PEJzxN5par5weykxMykS9Z6ara/j76JH3Yo2AH3bgJbPN4Z6NeCK9s5fDkzf33cg==} + /prisma/4.13.0: + resolution: {integrity: sha512-L9mqjnSmvWIRCYJ9mQkwCtj4+JDYYTdhoyo8hlsHNDXaZLh/b4hR0IoKIBbTKxZuyHQzLopb/+0Rvb69uGV7uA==} engines: {node: '>=14.17'} hasBin: true requiresBuild: true dependencies: - '@prisma/engines': 4.8.1 + '@prisma/engines': 4.13.0 /prismjs/1.29.0: resolution: {integrity: sha512-Kx/1w86q/epKcmte75LNrEoT+lX8pBpavuAbvJWRXar7Hz8jrtF+e3vY751p0R8H9HdArwaCTNDDzHg/ScJK1Q==}