diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index f0a776ede..c92e706dc 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -1,6 +1,7 @@ import { PrismaClient, Prisma } from "@trigger.dev/database"; import invariant from "tiny-invariant"; import { z } from "zod"; +import { logger } from "./services/logger.server"; export type PrismaTransactionClient = Omit< PrismaClient, @@ -15,15 +16,54 @@ function isTransactionClient( return !("$transaction" in prisma); } -export function $transaction( +function isPrismaKnownError( + error: unknown +): error is Prisma.PrismaClientKnownRequestError { + return ( + typeof error === "object" && + error !== null && + "code" in error && + typeof error.code === "string" + ); +} + +export type PrismaTransactionOptions = { + /** The maximum amount of time (in ms) Prisma Client will wait to acquire a transaction from the database. The default value is 2000ms. */ + maxWait?: number; + + /** The maximum amount of time (in ms) the interactive transaction can run before being canceled and rolled back. The default value is 5000ms. */ + timeout?: number; + + /** Sets the transaction isolation level. By default this is set to the value currently configured in your database. */ + isolationLevel?: Prisma.TransactionIsolationLevel; +}; + +export async function $transaction( prisma: PrismaClientOrTransaction, - fn: (prisma: PrismaTransactionClient) => Promise -): Promise { + fn: (prisma: PrismaTransactionClient) => Promise, + options?: PrismaTransactionOptions +): Promise { if (isTransactionClient(prisma)) { return fn(prisma); } - return (prisma as PrismaClient).$transaction(fn); + try { + return await (prisma as PrismaClient).$transaction(fn, options); + } catch (error) { + if (isPrismaKnownError(error)) { + logger.debug("prisma.$transaction error", { + code: error.code, + meta: error.meta, + stack: error.stack, + message: error.message, + name: error.name, + }); + + return; + } + + throw error; + } } export { Prisma }; diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx index d497c03ea..523f81948 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.runs.$runParam/route.tsx @@ -43,7 +43,10 @@ import { useOrganization } from "~/hooks/useOrganizations"; import { usePathName } from "~/hooks/usePathName"; import { useProject } from "~/hooks/useProject"; import { JobRunStatus } from "~/models/job.server"; -import { redirectWithSuccessMessage } from "~/models/message.server"; +import { + redirectBackWithErrorMessage, + redirectWithSuccessMessage, +} from "~/models/message.server"; import { RunPresenter } from "~/presenters/RunPresenter.server"; import { ContinueRunService } from "~/services/runs/continueRun.server"; import { ReRunService } from "~/services/runs/reRun.server"; @@ -114,6 +117,10 @@ export const action: ActionFunction = async ({ request, params }) => { const rerunService = new ReRunService(); const run = await rerunService.call({ runId: runParam }); + if (!run) { + return redirectBackWithErrorMessage(request, "Unable to retry run"); + } + return redirectWithSuccessMessage( runDashboardPath( { slug: organizationSlug }, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx index 3dabc2c7a..7c4df5a1b 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx @@ -12,10 +12,7 @@ import { HowToRunATest } from "~/components/helpContent/HelpContentText"; import { Button, ButtonContent } from "~/components/primitives/Buttons"; import { Callout } from "~/components/primitives/Callout"; import { FormError } from "~/components/primitives/FormError"; -import { HelpTrigger } from "~/components/primitives/Help"; -import { HelpContent } from "~/components/primitives/Help"; -import { Help } from "~/components/primitives/Help"; -import { Paragraph } from "~/components/primitives/Paragraph"; +import { Help, HelpContent, HelpTrigger } from "~/components/primitives/Help"; import { Popover, PopoverContent } from "~/components/primitives/Popover"; import { Select, @@ -25,18 +22,16 @@ import { SelectTrigger, SelectValue, } from "~/components/primitives/Select"; -import { redirectWithSuccessMessage } from "~/models/message.server"; +import { + redirectBackWithErrorMessage, + redirectWithSuccessMessage, +} from "~/models/message.server"; import { TestJobPresenter } from "~/presenters/TestJobPresenter.server"; import { TestJobService } from "~/services/jobs/testJob.server"; import { requireUserId } from "~/services/session.server"; import { cn } from "~/utils/cn"; -import { formDataAsObject } from "~/utils/formData"; import { Handle } from "~/utils/handle"; -import { - JobParamsSchema, - jobTestPath, - runDashboardPath, -} from "~/utils/pathBuilder"; +import { JobParamsSchema, runDashboardPath } from "~/utils/pathBuilder"; export const loader = async ({ request, params }: LoaderArgs) => { const userId = await requireUserId(request); @@ -100,6 +95,13 @@ export const action: ActionFunction = async ({ request, params }) => { versionId: submission.value.versionId, }); + if (!run) { + return redirectBackWithErrorMessage( + request, + "Unable to start a test run: Something went wrong" + ); + } + return redirectWithSuccessMessage( runDashboardPath( { slug: organizationSlug }, diff --git a/apps/webapp/app/routes/api.v1.$endpointSlug.schedules.$id.registrations.ts b/apps/webapp/app/routes/api.v1.$endpointSlug.schedules.$id.registrations.ts index 11091582b..3e3e8e213 100644 --- a/apps/webapp/app/routes/api.v1.$endpointSlug.schedules.$id.registrations.ts +++ b/apps/webapp/app/routes/api.v1.$endpointSlug.schedules.$id.registrations.ts @@ -58,6 +58,10 @@ export async function action({ request, params }: ActionArgs) { id: parsedParams.data.id, }); + if (!registration) { + return json({ error: "Something went wrong" }, { status: 500 }); + } + return json( RegisterScheduleResponseBodySchema.parse({ id: registration.key, diff --git a/apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts b/apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts index bc981ffb9..2b42b5ff1 100644 --- a/apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts +++ b/apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts @@ -57,6 +57,10 @@ export async function action({ request, params }: ActionArgs) { key: parsedParams.data.key, }); + if (!registration) { + return json({ error: "Could not register trigger" }, { status: 500 }); + } + return json(registration); } catch (error) { if (error instanceof Error) { diff --git a/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts b/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts index 937e4ad12..d731ed467 100644 --- a/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts +++ b/apps/webapp/app/routes/api.v1.endpoints.$endpointSlug.index.ts @@ -63,13 +63,19 @@ export async function action({ request, params }: ActionArgs) { const service = new IndexEndpointService(); try { - const { data, ...index } = await service.call( + const indexing = await service.call( endpoint.id, "API", parsedBody.data.reason, parsedBody.data.data ); + if (!indexing) { + return json({ error: "Something went wrong" }, { status: 500 }); + } + + const { data, ...index } = indexing; + return json(index); } catch (error) { if (error instanceof Error) { diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts index 4e7bc2b4d..348cca2d8 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts @@ -149,6 +149,10 @@ export async function action({ request, params }: ActionArgs) { task, }); + if (!task) { + return json({ error: "Something went wrong" }, { status: 500 }); + } + return json(task); } catch (error) { if (error instanceof Error) { @@ -170,7 +174,7 @@ export class RunTaskService { runId: string, idempotencyKey: string, taskBody: RunTaskBodyOutput - ): Promise { + ): Promise { const task = await $transaction(this.#prismaClient, async (tx) => { const existingTask = await tx.task.findUnique({ where: { @@ -260,6 +264,6 @@ export class RunTaskService { return task; }); - return taskWithAttemptsToServerTask(task); + return task ? taskWithAttemptsToServerTask(task) : undefined; } } diff --git a/apps/webapp/app/services/endpoints/indexEndpoint.server.ts b/apps/webapp/app/services/endpoints/indexEndpoint.server.ts index 97363413a..72b65b3d3 100644 --- a/apps/webapp/app/services/endpoints/indexEndpoint.server.ts +++ b/apps/webapp/app/services/endpoints/indexEndpoint.server.ts @@ -51,90 +51,94 @@ export class IndexEndpointService { dynamicSchedules: 0, }; - return await $transaction(this.#prismaClient, async (tx) => { - for (const job of jobs) { - if (!job.enabled) { - continue; + return await $transaction( + this.#prismaClient, + async (tx) => { + for (const job of jobs) { + if (!job.enabled) { + continue; + } + + indexStats.jobs++; + + await workerQueue.enqueue( + "registerJob", + { + job, + endpointId: endpoint.id, + }, + { + queueName, + tx, + } + ); } - indexStats.jobs++; + for (const source of sources) { + indexStats.sources++; - await workerQueue.enqueue( - "registerJob", - { - job, - endpointId: endpoint.id, - }, - { - queueName, - tx, - } - ); - } + await workerQueue.enqueue( + "registerSource", + { + source, + endpointId: endpoint.id, + }, + { + queueName, + tx, + } + ); + } - for (const source of sources) { - indexStats.sources++; + for (const dynamicTrigger of dynamicTriggers) { + indexStats.dynamicTriggers++; - await workerQueue.enqueue( - "registerSource", - { - source, - endpointId: endpoint.id, - }, - { - queueName, - tx, - } - ); - } + await workerQueue.enqueue( + "registerDynamicTrigger", + { + dynamicTrigger, + endpointId: endpoint.id, + }, + { + queueName, + tx, + } + ); + } - for (const dynamicTrigger of dynamicTriggers) { - indexStats.dynamicTriggers++; + for (const dynamicSchedule of dynamicSchedules) { + indexStats.dynamicSchedules++; - await workerQueue.enqueue( - "registerDynamicTrigger", - { - dynamicTrigger, - endpointId: endpoint.id, - }, - { - queueName, - tx, - } - ); - } + await workerQueue.enqueue( + "registerDynamicSchedule", + { + dynamicSchedule, + endpointId: endpoint.id, + }, + { + queueName, + tx, + } + ); + } - for (const dynamicSchedule of dynamicSchedules) { - indexStats.dynamicSchedules++; - - await workerQueue.enqueue( - "registerDynamicSchedule", - { - dynamicSchedule, - endpointId: endpoint.id, - }, - { - queueName, - tx, - } - ); - } - - return await tx.endpointIndex.create({ - data: { - endpointId: endpoint.id, - stats: indexStats, + return await tx.endpointIndex.create({ data: { - jobs, - sources, - dynamicTriggers, - dynamicSchedules, + endpointId: endpoint.id, + stats: indexStats, + data: { + jobs, + sources, + dynamicTriggers, + dynamicSchedules, + }, + source, + sourceData, + reason, }, - source, - sourceData, - reason, - }, - }); - }); + }); + }, + { timeout: 15000 } + ); } } diff --git a/apps/webapp/app/services/events/deliverEvent.server.ts b/apps/webapp/app/services/events/deliverEvent.server.ts index 6a0fdc482..f3745a49b 100644 --- a/apps/webapp/app/services/events/deliverEvent.server.ts +++ b/apps/webapp/app/services/events/deliverEvent.server.ts @@ -13,76 +13,80 @@ export class DeliverEventService { } public async call(id: string) { - await $transaction(this.#prismaClient, async (tx) => { - const eventRecord = await tx.eventRecord.findUniqueOrThrow({ - where: { - id, - }, - include: { - environment: { - include: { - organization: true, - project: true, + await $transaction( + this.#prismaClient, + async (tx) => { + const eventRecord = await tx.eventRecord.findUniqueOrThrow({ + where: { + id, + }, + include: { + environment: { + include: { + organization: true, + project: true, + }, }, }, - }, - }); + }); - const possibleEventDispatchers = await tx.eventDispatcher.findMany({ - where: { - environmentId: eventRecord.environmentId, - event: eventRecord.name, - source: eventRecord.source, - enabled: true, - manual: false, - }, - }); + const possibleEventDispatchers = await tx.eventDispatcher.findMany({ + where: { + environmentId: eventRecord.environmentId, + event: eventRecord.name, + source: eventRecord.source, + enabled: true, + manual: false, + }, + }); - logger.debug("Found possible event dispatchers", { - possibleEventDispatchers, - eventRecord: eventRecord.id, - }); - - const matchingEventDispatchers = possibleEventDispatchers.filter( - (eventDispatcher) => - this.#evaluateEventRule(eventDispatcher, eventRecord) - ); - - if (matchingEventDispatchers.length === 0) { - logger.debug("No matching event dispatchers", { + logger.debug("Found possible event dispatchers", { + possibleEventDispatchers, eventRecord: eventRecord.id, }); - return; - } + const matchingEventDispatchers = possibleEventDispatchers.filter( + (eventDispatcher) => + this.#evaluateEventRule(eventDispatcher, eventRecord) + ); - logger.debug("Found matching event dispatchers", { - matchingEventDispatchers, - eventRecord: eventRecord.id, - }); + if (matchingEventDispatchers.length === 0) { + logger.debug("No matching event dispatchers", { + eventRecord: eventRecord.id, + }); - await Promise.all( - matchingEventDispatchers.map((eventDispatcher) => - workerQueue.enqueue( - "events.invokeDispatcher", - { - id: eventDispatcher.id, - eventRecordId: eventRecord.id, - }, - { tx } + return; + } + + logger.debug("Found matching event dispatchers", { + matchingEventDispatchers, + eventRecord: eventRecord.id, + }); + + await Promise.all( + matchingEventDispatchers.map((eventDispatcher) => + workerQueue.enqueue( + "events.invokeDispatcher", + { + id: eventDispatcher.id, + eventRecordId: eventRecord.id, + }, + { tx } + ) ) - ) - ); + ); - await tx.eventRecord.update({ - where: { - id: eventRecord.id, - }, - data: { - deliveredAt: new Date(), - }, - }); - }); + await tx.eventRecord.update({ + where: { + id: eventRecord.id, + }, + data: { + deliveredAt: new Date(), + }, + }); + }, + { timeout: 10000 } + ); } #evaluateEventRule( diff --git a/apps/webapp/app/services/externalApis/integrationConnectionCreated.server.ts b/apps/webapp/app/services/externalApis/integrationConnectionCreated.server.ts index 1ba6e55ea..b6f12eb7e 100644 --- a/apps/webapp/app/services/externalApis/integrationConnectionCreated.server.ts +++ b/apps/webapp/app/services/externalApis/integrationConnectionCreated.server.ts @@ -1,8 +1,8 @@ -import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server"; +import { MISSING_CONNECTION_RESOLVED_NOTIFICATION } from "@trigger.dev/internal"; +import { PrismaClientOrTransaction, prisma } from "~/db.server"; import { IngestSendEvent } from "../events/ingestSendEvent.server"; import { logger } from "../logger.server"; import { workerQueue } from "../worker.server"; -import { MISSING_CONNECTION_RESOLVED_NOTIFICATION } from "@trigger.dev/internal"; export class IntegrationConnectionCreatedService { #prismaClient: PrismaClientOrTransaction; @@ -14,9 +14,9 @@ export class IntegrationConnectionCreatedService { public async call(id: string) { logger.debug("IntegrationConnectionCreatedService.call", { id }); - return await $transaction(this.#prismaClient, async (tx) => { - // first, deliver the event through the dispatcher - const connection = await tx.integrationConnection.findUniqueOrThrow({ + // first, deliver the event through the dispatcher + const connection = + await this.#prismaClient.integrationConnection.findUniqueOrThrow({ where: { id, }, @@ -26,7 +26,8 @@ export class IntegrationConnectionCreatedService { }, }); - const missingConnection = await tx.missingConnection.findUnique({ + const missingConnection = + await this.#prismaClient.missingConnection.findUnique({ where: { integrationId_connectionType_accountIdentifier: { integrationId: connection.integrationId, @@ -55,68 +56,63 @@ export class IntegrationConnectionCreatedService { }, }); - if (!missingConnection) { - return; - } + if (!missingConnection) { + return; + } - if (missingConnection.resolved) { - return; - } + if (missingConnection.resolved) { + return; + } - const firstRun = missingConnection.runs[0]; + const firstRun = missingConnection.runs[0]; - if (!firstRun) { - return; - } + if (!firstRun) { + return; + } - const eventId = `${missingConnection.id}-resolved`; + const eventId = `${missingConnection.id}-resolved`; - const eventService = new IngestSendEvent(tx); + const eventService = new IngestSendEvent(); - await eventService.call(firstRun.environment, { - id: eventId, - name: MISSING_CONNECTION_RESOLVED_NOTIFICATION, - payload: { - id: missingConnection.id, - type: missingConnection.connectionType, - client: { - id: missingConnection.integration.slug, - title: missingConnection.integration.title, - scopes: missingConnection.integration.scopes, - createdAt: missingConnection.integration.createdAt, - updatedAt: missingConnection.integration.updatedAt, - }, - expiresAt: connection.expiresAt ?? undefined, - account: missingConnection.externalAccount - ? { - id: missingConnection.externalAccount.identifier, - metadata: missingConnection.externalAccount.metadata, - } - : undefined, + await eventService.call(firstRun.environment, { + id: eventId, + name: MISSING_CONNECTION_RESOLVED_NOTIFICATION, + payload: { + id: missingConnection.id, + type: missingConnection.connectionType, + client: { + id: missingConnection.integration.slug, + title: missingConnection.integration.title, + scopes: missingConnection.integration.scopes, + createdAt: missingConnection.integration.createdAt, + updatedAt: missingConnection.integration.updatedAt, }, - context: {}, - }); - - await tx.missingConnection.delete({ - where: { - id: missingConnection.id, - }, - }); - - for (const run of missingConnection.runs) { - logger.debug("[IntegrationConnectionCreatedService] restarting run", { - run, - }); - - // We need to start the run again - await workerQueue.enqueue( - "startRun", - { - id: run.id, - }, - { tx } - ); - } + expiresAt: connection.expiresAt ?? undefined, + account: missingConnection.externalAccount + ? { + id: missingConnection.externalAccount.identifier, + metadata: missingConnection.externalAccount.metadata, + } + : undefined, + }, + context: {}, }); + + await this.#prismaClient.missingConnection.delete({ + where: { + id: missingConnection.id, + }, + }); + + for (const run of missingConnection.runs) { + logger.debug("[IntegrationConnectionCreatedService] restarting run", { + run, + }); + + // We need to start the run again + await workerQueue.enqueue("startRun", { + id: run.id, + }); + } } } diff --git a/apps/webapp/app/services/runs/continueRun.server.ts b/apps/webapp/app/services/runs/continueRun.server.ts index 636ff34e5..d27e41df1 100644 --- a/apps/webapp/app/services/runs/continueRun.server.ts +++ b/apps/webapp/app/services/runs/continueRun.server.ts @@ -12,88 +12,92 @@ export class ContinueRunService { } public async call({ runId }: { runId: string }) { - return await $transaction(this.#prismaClient, async (tx) => { - const run = await tx.jobRun.findUniqueOrThrow({ - where: { id: runId }, - include: { - queue: true, - }, - }); - - if (!RESUMABLE_STATUSES.includes(run.status)) { - throw new Error("Run is not resumable"); - } - - if (run.queue.jobCount >= run.queue.maxJobs) { - await tx.jobRun.update({ + return await $transaction( + this.#prismaClient, + async (tx) => { + const run = await tx.jobRun.findUniqueOrThrow({ where: { id: runId }, - data: { - status: "QUEUED", - queuedAt: new Date(), - startedAt: null, - completedAt: null, - output: Prisma.DbNull, - timedOutAt: null, - timedOutReason: null, + include: { + queue: true, }, }); - } else { - await tx.jobRun.update({ - where: { id: runId }, - data: { - status: "STARTED", - queuedAt: null, - startedAt: new Date(), - completedAt: null, - output: Prisma.DbNull, - timedOutAt: null, - timedOutReason: null, - queue: { - update: { - jobCount: { - increment: 1, + + if (!RESUMABLE_STATUSES.includes(run.status)) { + throw new Error("Run is not resumable"); + } + + if (run.queue.jobCount >= run.queue.maxJobs) { + await tx.jobRun.update({ + where: { id: runId }, + data: { + status: "QUEUED", + queuedAt: new Date(), + startedAt: null, + completedAt: null, + output: Prisma.DbNull, + timedOutAt: null, + timedOutReason: null, + }, + }); + } else { + await tx.jobRun.update({ + where: { id: runId }, + data: { + status: "STARTED", + queuedAt: null, + startedAt: new Date(), + completedAt: null, + output: Prisma.DbNull, + timedOutAt: null, + timedOutReason: null, + queue: { + update: { + jobCount: { + increment: 1, + }, }, }, }, - }, - }); + }); - const execution = await tx.jobRunExecution.create({ - data: { - run: { - connect: { - id: runId, + const execution = await tx.jobRunExecution.create({ + data: { + run: { + connect: { + id: runId, + }, }, + status: "PENDING", + reason: "EXECUTE_JOB", + retryLimit: EXECUTE_JOB_RETRY_LIMIT, }, - status: "PENDING", - reason: "EXECUTE_JOB", - retryLimit: EXECUTE_JOB_RETRY_LIMIT, - }, - }); + }); - const job = await workerQueue.enqueue( - "performRunExecution", - { - id: execution.id, - }, - { tx } - ); + const job = await workerQueue.enqueue( + "performRunExecution", + { + id: execution.id, + }, + { tx } + ); - await tx.jobRunExecution.update({ - where: { id: execution.id }, - data: { - graphileJobId: job.id, - }, - }); + await tx.jobRunExecution.update({ + where: { id: execution.id }, + data: { + graphileJobId: job.id, + }, + }); - await workerQueue.enqueue( - "startQueuedRuns", - { - id: run.queueId, - }, - { tx } - ); - } - }); + await workerQueue.enqueue( + "startQueuedRuns", + { + id: run.queueId, + }, + { tx } + ); + } + }, + { timeout: 10000 } + ); } } diff --git a/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts b/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts index f4e7dbe7e..1054fe6fd 100644 --- a/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts +++ b/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts @@ -13,72 +13,82 @@ export class DeliverScheduledEventService { } public async call(id: string, payload: ScheduledPayload) { - return await $transaction(this.#prismaClient, async (tx) => { - // first, deliver the event through the dispatcher - const scheduleSource = await tx.scheduleSource.findUniqueOrThrow({ - where: { - id, - }, - include: { - dispatcher: true, - environment: { - include: { - organization: true, - project: true, - }, + return await $transaction( + this.#prismaClient, + async (tx) => { + // first, deliver the event through the dispatcher + const scheduleSource = await tx.scheduleSource.findUniqueOrThrow({ + where: { + id, }, - externalAccount: true, - }, - }); + include: { + dispatcher: true, + environment: { + include: { + organization: true, + project: true, + }, + }, + externalAccount: true, + }, + }); - if (!scheduleSource.active) { - return; - } - - const eventId = `${scheduleSource.id}:${payload.ts.getTime()}`; - - // false prevents send event from delivering the event to dispatchers - // since we are going to control that ourselves - const eventService = new IngestSendEvent(tx, false); - - const eventRecord = await eventService.call( - scheduleSource.environment, - { - id: eventId, - name: SCHEDULED_EVENT, - payload, - }, - { accountId: scheduleSource.externalAccount?.identifier }, - { - id: scheduleSource.key, - metadata: scheduleSource.metadata, + if (!scheduleSource.active) { + return; } - ); - const invokeDispatcherService = new InvokeDispatcherService(tx); + const eventId = `${scheduleSource.id}:${payload.ts.getTime()}`; - await invokeDispatcherService.call( - scheduleSource.dispatcher.id, - eventRecord.id - ); + // false prevents send event from delivering the event to dispatchers + // since we are going to control that ourselves + const eventService = new IngestSendEvent(tx, false); - logger.debug("updating lastEventTimestamp", { - id, - lastEventTimestamp: payload.ts, - }); + const eventRecord = await eventService.call( + scheduleSource.environment, + { + id: eventId, + name: SCHEDULED_EVENT, + payload, + }, + { accountId: scheduleSource.externalAccount?.identifier }, + { + id: scheduleSource.key, + metadata: scheduleSource.metadata, + } + ); - await tx.scheduleSource.update({ - where: { + if (!eventRecord) { + throw new Error( + `Unable to create an event record when delivering scheduled event for scheduleSource.id = ${scheduleSource.id}` + ); + } + + const invokeDispatcherService = new InvokeDispatcherService(tx); + + await invokeDispatcherService.call( + scheduleSource.dispatcher.id, + eventRecord.id + ); + + logger.debug("updating lastEventTimestamp", { id, - }, - data: { lastEventTimestamp: payload.ts, - }, - }); + }); - const nextScheduledEventService = new NextScheduledEventService(tx); + await tx.scheduleSource.update({ + where: { + id, + }, + data: { + lastEventTimestamp: payload.ts, + }, + }); - await nextScheduledEventService.call(scheduleSource.id); - }); + const nextScheduledEventService = new NextScheduledEventService(tx); + + await nextScheduledEventService.call(scheduleSource.id); + }, + { timeout: 10000 } + ); } } diff --git a/apps/webapp/app/services/sources/registerSource.server.ts b/apps/webapp/app/services/sources/registerSource.server.ts index bba9d875d..9294bc37e 100644 --- a/apps/webapp/app/services/sources/registerSource.server.ts +++ b/apps/webapp/app/services/sources/registerSource.server.ts @@ -64,7 +64,7 @@ export class RegisterSourceService { .filter(Boolean) .join(":"); - const { id, orphanedEvents } = await $transaction( + const source = await $transaction( this.#prismaClient, async (tx) => { const integration = await this.#findOrCreateIntegration( @@ -215,9 +215,16 @@ export class RegisterSourceService { id: triggerSource.id, orphanedEvents: Array.from(orphanedEvents), }; - } + }, + { timeout: 15000 } ); + if (!source) { + return; + } + + const { id, orphanedEvents } = source; + // We need to activate the source if: // 1. It's not active // 2. There are orphaned events diff --git a/apps/webapp/app/services/triggers/initializeTrigger.server.ts b/apps/webapp/app/services/triggers/initializeTrigger.server.ts index 0be540ebd..202cc0971 100644 --- a/apps/webapp/app/services/triggers/initializeTrigger.server.ts +++ b/apps/webapp/app/services/triggers/initializeTrigger.server.ts @@ -74,6 +74,10 @@ export class InitializeTriggerService { registrationMetadata: payload.metadata, }); + if (!registration) { + return; + } + await this.#sendEvent.call( environment, { diff --git a/apps/webapp/app/services/triggers/registerDynamicSchedule.server.ts b/apps/webapp/app/services/triggers/registerDynamicSchedule.server.ts index 9bfe50256..748440946 100644 --- a/apps/webapp/app/services/triggers/registerDynamicSchedule.server.ts +++ b/apps/webapp/app/services/triggers/registerDynamicSchedule.server.ts @@ -2,12 +2,7 @@ import { RegisterDynamicSchedulePayload, SCHEDULED_EVENT, } from "@trigger.dev/internal"; -import { - $transaction, - PrismaClient, - PrismaClientOrTransaction, -} from "~/db.server"; -import { prisma } from "~/db.server"; +import { PrismaClientOrTransaction, prisma } from "~/db.server"; export class RegisterDynamicScheduleService { #prismaClient: PrismaClientOrTransaction; @@ -20,90 +15,88 @@ export class RegisterDynamicScheduleService { endpointId: string, metadata: RegisterDynamicSchedulePayload ) { - await $transaction(this.#prismaClient, async (tx) => { - const dynamicTrigger = await tx.dynamicTrigger.upsert({ - where: { - endpointId_slug_type: { - endpointId: endpointId, - slug: metadata.id, - type: "SCHEDULE", - }, - }, - create: { + const dynamicTrigger = await this.#prismaClient.dynamicTrigger.upsert({ + where: { + endpointId_slug_type: { + endpointId: endpointId, slug: metadata.id, type: "SCHEDULE", - endpoint: { - connect: { - id: endpointId, - }, + }, + }, + create: { + slug: metadata.id, + type: "SCHEDULE", + endpoint: { + connect: { + id: endpointId, }, }, - update: {}, - include: { - jobs: true, - endpoint: true, - }, - }); + }, + update: {}, + include: { + jobs: true, + endpoint: true, + }, + }); - // Now we need to connect the jobs - const jobs = await tx.job.findMany({ - where: { - slug: { - in: metadata.jobs.map((job) => job.id), - }, - versions: { - some: { - endpointId, - }, + // Now we need to connect the jobs + const jobs = await this.#prismaClient.job.findMany({ + where: { + slug: { + in: metadata.jobs.map((job) => job.id), + }, + versions: { + some: { + endpointId, }, }, - }); + }, + }); - // Update all the jobs that are associated with this dynamic trigger - await tx.dynamicTrigger.update({ - where: { + // Update all the jobs that are associated with this dynamic trigger + await this.#prismaClient.dynamicTrigger.update({ + where: { + id: dynamicTrigger.id, + }, + data: { + jobs: { + connect: jobs.map((job) => ({ + id: job.id, + })), + disconnect: dynamicTrigger.jobs.filter( + (job) => !jobs.find((j) => j.id === job.id) + ), + }, + }, + }); + + await this.#prismaClient.eventDispatcher.upsert({ + where: { + dispatchableId_environmentId: { + dispatchableId: dynamicTrigger.id, + environmentId: dynamicTrigger.endpoint.environmentId, + }, + }, + create: { + event: SCHEDULED_EVENT, + source: "trigger.dev", + payloadFilter: {}, + contextFilter: {}, + environmentId: dynamicTrigger.endpoint.environmentId, + enabled: true, + dispatchable: { + type: "DYNAMIC_TRIGGER", id: dynamicTrigger.id, }, - data: { - jobs: { - connect: jobs.map((job) => ({ - id: job.id, - })), - disconnect: dynamicTrigger.jobs.filter( - (job) => !jobs.find((j) => j.id === job.id) - ), - }, + dispatchableId: dynamicTrigger.id, + manual: true, + }, + update: { + dispatchable: { + type: "DYNAMIC_TRIGGER", + id: dynamicTrigger.id, }, - }); - - const eventDispatcher = await tx.eventDispatcher.upsert({ - where: { - dispatchableId_environmentId: { - dispatchableId: dynamicTrigger.id, - environmentId: dynamicTrigger.endpoint.environmentId, - }, - }, - create: { - event: SCHEDULED_EVENT, - source: "trigger.dev", - payloadFilter: {}, - contextFilter: {}, - environmentId: dynamicTrigger.endpoint.environmentId, - enabled: true, - dispatchable: { - type: "DYNAMIC_TRIGGER", - id: dynamicTrigger.id, - }, - dispatchableId: dynamicTrigger.id, - manual: true, - }, - update: { - dispatchable: { - type: "DYNAMIC_TRIGGER", - id: dynamicTrigger.id, - }, - }, - }); + }, }); } } diff --git a/apps/webapp/app/services/triggers/registerTriggerSource.server.ts b/apps/webapp/app/services/triggers/registerTriggerSource.server.ts index 6281e571c..45e848623 100644 --- a/apps/webapp/app/services/triggers/registerTriggerSource.server.ts +++ b/apps/webapp/app/services/triggers/registerTriggerSource.server.ts @@ -32,7 +32,7 @@ export class RegisterTriggerSourceService { key: string; accountId?: string; registrationMetadata?: any; - }): Promise { + }): Promise { const endpoint = await this.#prismaClient.endpoint.findUniqueOrThrow({ where: { environmentId_slug: { @@ -53,97 +53,105 @@ export class RegisterTriggerSourceService { }, }); - return await $transaction(this.#prismaClient, async (tx) => { - const service = new RegisterSourceService(tx); + return await $transaction( + this.#prismaClient, + async (tx) => { + const service = new RegisterSourceService(tx); - const triggerSource = await service.call( - endpoint.id, - payload.source, - dynamicTrigger.id, - accountId, - { id: key, metadata: registrationMetadata } - ); + const triggerSource = await service.call( + endpoint.id, + payload.source, + dynamicTrigger.id, + accountId, + { id: key, metadata: registrationMetadata } + ); - const eventDispatcher = await tx.eventDispatcher.upsert({ - where: { - dispatchableId_environmentId: { + if (!triggerSource) { + return; + } + + const eventDispatcher = await tx.eventDispatcher.upsert({ + where: { + dispatchableId_environmentId: { + dispatchableId: triggerSource.id, + environmentId: environment.id, + }, + }, + create: { dispatchableId: triggerSource.id, environmentId: environment.id, + event: payload.rule.event, + source: payload.rule.source, + payloadFilter: payload.rule.payload, + contextFilter: payload.rule.context, + dispatchable: { + type: "DYNAMIC_TRIGGER", + id: dynamicTrigger.id, + }, }, - }, - create: { - dispatchableId: triggerSource.id, - environmentId: environment.id, - event: payload.rule.event, - source: payload.rule.source, - payloadFilter: payload.rule.payload, - contextFilter: payload.rule.context, - dispatchable: { - type: "DYNAMIC_TRIGGER", - id: dynamicTrigger.id, + update: { + event: payload.rule.event, + source: payload.rule.source, + payloadFilter: payload.rule.payload, + contextFilter: payload.rule.context, + dispatchable: { + type: "DYNAMIC_TRIGGER", + id: dynamicTrigger.id, + }, }, - }, - update: { - event: payload.rule.event, - source: payload.rule.source, - payloadFilter: payload.rule.payload, - contextFilter: payload.rule.context, - dispatchable: { - type: "DYNAMIC_TRIGGER", - id: dynamicTrigger.id, - }, - }, - }); + }); - const registration = await tx.dynamicTriggerRegistration.upsert({ - where: { - key_dynamicTriggerId: { + const registration = await tx.dynamicTriggerRegistration.upsert({ + where: { + key_dynamicTriggerId: { + key, + dynamicTriggerId: dynamicTrigger.id, + }, + }, + create: { key, dynamicTriggerId: dynamicTrigger.id, + sourceId: triggerSource.id, + eventDispatcherId: eventDispatcher.id, + metadata: registrationMetadata, }, - }, - create: { - key, - dynamicTriggerId: dynamicTrigger.id, - sourceId: triggerSource.id, - eventDispatcherId: eventDispatcher.id, - metadata: registrationMetadata, - }, - update: { - metadata: registrationMetadata, - }, - }); - - const secretStore = getSecretStore( - triggerSource.secretReference.provider, - { prismaClient: tx } - ); - - const { secret } = await secretStore.getSecretOrThrow( - z.object({ - secret: z.string(), - }), - triggerSource.secretReference.key - ); - - return { - id: registration.id, - source: { - key: triggerSource.key, - active: triggerSource.active, - params: triggerSource.params, - secret, - data: triggerSource.channelData as any, - channel: { - type: "HTTP", - url: `${env.APP_ORIGIN}/api/v1/sources/http/${triggerSource.id}`, + update: { + metadata: registrationMetadata, }, - clientId: triggerSource.integration.slug, - }, - events: triggerSource.events.map((e) => e.name), - missingEvents: [], - orphanedEvents: [], - }; - }); + }); + + const secretStore = getSecretStore( + triggerSource.secretReference.provider, + { prismaClient: tx } + ); + + const { secret } = await secretStore.getSecretOrThrow( + z.object({ + secret: z.string(), + }), + triggerSource.secretReference.key + ); + + return { + id: registration.id, + source: { + key: triggerSource.key, + active: triggerSource.active, + params: triggerSource.params, + secret, + data: triggerSource.channelData as any, + channel: { + type: "HTTP", + url: `${env.APP_ORIGIN}/api/v1/sources/http/${triggerSource.id}`, + }, + clientId: triggerSource.integration.slug, + }, + events: triggerSource.events.map((e) => e.name), + missingEvents: [], + orphanedEvents: [], + }; + }, + { timeout: 15000 } + ); } }