From 6dea6e1c7d181d2f6138494af81d862fd376ed85 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sun, 11 Aug 2024 20:25:36 +0100 Subject: [PATCH] provide worker files as part of the worker creation on the server --- .../services/createBackgroundWorker.server.ts | 74 ++++++++++++++++++- packages/cli-v3/src/dev/workerRuntime.ts | 44 +++++++++++ packages/core/src/v3/schemas/resources.ts | 10 +++ .../migration.sql | 24 ++++++ .../migration.sql | 38 ++++++++++ packages/database/prisma/schema.prisma | 26 +++++++ 6 files changed, 213 insertions(+), 3 deletions(-) create mode 100644 packages/database/prisma/migrations/20240810090402_add_background_worker_file_model/migration.sql create mode 100644 packages/database/prisma/migrations/20240811185335_improve_background_worker_file_model/migration.sql diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index 34b747e76..5712c81a3 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -1,4 +1,8 @@ -import { CreateBackgroundWorkerRequestBody, TaskResource } from "@trigger.dev/core/v3"; +import { + BackgroundWorkerFileMetadata, + CreateBackgroundWorkerRequestBody, + TaskResource, +} from "@trigger.dev/core/v3"; import type { BackgroundWorker } from "@trigger.dev/database"; import { Prisma, PrismaClientOrTransaction } from "~/db.server"; import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; @@ -70,7 +74,19 @@ export class CreateBackgroundWorkerService extends BaseService { }, }); - await createBackgroundTasks(body.metadata.tasks, backgroundWorker, environment, this._prisma); + const tasksToBackgroundFiles = await createBackgroundFiles( + body.metadata.fileContents, + backgroundWorker, + environment, + this._prisma + ); + await createBackgroundTasks( + body.metadata.tasks, + backgroundWorker, + environment, + this._prisma, + tasksToBackgroundFiles + ); await syncDeclarativeSchedules( body.metadata.tasks, backgroundWorker, @@ -121,7 +137,8 @@ export async function createBackgroundTasks( tasks: TaskResource[], worker: BackgroundWorker, environment: AuthenticatedEnvironment, - prisma: PrismaClientOrTransaction + prisma: PrismaClientOrTransaction, + tasksToBackgroundFiles?: Map ) { for (const task of tasks) { try { @@ -138,6 +155,7 @@ export async function createBackgroundTasks( queueConfig: task.queue, machineConfig: task.machine, triggerSource: task.triggerSource === "schedule" ? "SCHEDULED" : "STANDARD", + fileId: tasksToBackgroundFiles?.get(task.id) ?? null, }, }); @@ -381,3 +399,53 @@ export async function syncDeclarativeSchedules( } } } + +export async function createBackgroundFiles( + files: Array | undefined, + worker: BackgroundWorker, + environment: AuthenticatedEnvironment, + prisma: PrismaClientOrTransaction +) { + // Maps from each taskId to the backgroundWorkerFileId + const results = new Map(); + + if (!files) { + return results; + } + + for (const file of files) { + const backgroundWorkerFile = await prisma.backgroundWorkerFile.upsert({ + where: { + projectId_contentHash: { + projectId: environment.projectId, + contentHash: file.contentHash, + }, + }, + create: { + friendlyId: generateFriendlyId("file"), + projectId: environment.projectId, + contentHash: file.contentHash, + filePath: file.filePath, + contents: Buffer.from(file.contents), + backgroundWorkers: { + connect: { + id: worker.id, + }, + }, + }, + update: { + backgroundWorkers: { + connect: { + id: worker.id, + }, + }, + }, + }); + + for (const taskId of file.taskIds) { + results.set(taskId, backgroundWorkerFile.id); + } + } + + return results; +} diff --git a/packages/cli-v3/src/dev/workerRuntime.ts b/packages/cli-v3/src/dev/workerRuntime.ts index 0aba11e51..3b27c162c 100644 --- a/packages/cli-v3/src/dev/workerRuntime.ts +++ b/packages/cli-v3/src/dev/workerRuntime.ts @@ -26,6 +26,9 @@ import { logger } from "../utilities/logger.js"; import { VERSION } from "../version.js"; import { BackgroundWorker, BackgroundWorkerCoordinator } from "./backgroundWorker.js"; import { getInstrumentedPackageNames } from "../build/instrumentation.js"; +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { createHash } from "node:crypto"; export interface WorkerRuntime { shutdown(): Promise; @@ -183,6 +186,11 @@ class DevWorkerRuntime implements WorkerRuntime { return; } + const fileContents = await this.#fetchTaskFiles( + backgroundWorker.manifest.tasks, + this.options.config.workingDir + ); + const backgroundWorkerBody: CreateBackgroundWorkerRequestBody = { localOnly: true, metadata: { @@ -190,6 +198,7 @@ class DevWorkerRuntime implements WorkerRuntime { cliPackageVersion: VERSION, tasks: backgroundWorker.manifest.tasks, contentHash: manifest.contentHash, + fileContents, }, supportsLazyAttempts: true, }; @@ -210,6 +219,41 @@ class DevWorkerRuntime implements WorkerRuntime { eventBus.emit("backgroundWorkerInitialized", backgroundWorker); } + async #fetchTaskFiles(tasks: TaskManifest[], workingDir: string) { + const tasksGroupedByFile: Record = {}; + + for (const task of tasks) { + if (!tasksGroupedByFile[task.filePath]) { + tasksGroupedByFile[task.filePath] = []; + } + + tasksGroupedByFile[task.filePath]!.push(task); + } + + const taskFiles: Array<{ + taskIds: string[]; + contents: string; + contentHash: string; + filePath: string; + }> = []; + + for (const [filePath, tasks] of Object.entries(tasksGroupedByFile)) { + const contents = await readFile(join(workingDir, filePath), "utf-8"); + const taskIds = tasks.map((task) => task.id); + const hasher = createHash("md5"); + hasher.update(contents); + + taskFiles.push({ + filePath, + taskIds, + contents, + contentHash: hasher.digest("hex"), + }); + } + + return taskFiles; + } + async #getEnvVars(): Promise> { const environmentVariablesResponse = await this.options.client.getEnvironmentVariables( this.options.config.project diff --git a/packages/core/src/v3/schemas/resources.ts b/packages/core/src/v3/schemas/resources.ts index 0acb6933b..228aad325 100644 --- a/packages/core/src/v3/schemas/resources.ts +++ b/packages/core/src/v3/schemas/resources.ts @@ -15,11 +15,21 @@ export const TaskResource = z.object({ export type TaskResource = z.infer; +export const BackgroundWorkerFileMetadata = z.object({ + filePath: z.string(), + contents: z.string(), + contentHash: z.string(), + taskIds: z.array(z.string()), +}); + +export type BackgroundWorkerFileMetadata = z.infer; + export const BackgroundWorkerMetadata = z.object({ packageVersion: z.string(), contentHash: z.string(), cliPackageVersion: z.string().optional(), tasks: z.array(TaskResource), + fileContents: z.array(BackgroundWorkerFileMetadata).optional(), }); export type BackgroundWorkerMetadata = z.infer; diff --git a/packages/database/prisma/migrations/20240810090402_add_background_worker_file_model/migration.sql b/packages/database/prisma/migrations/20240810090402_add_background_worker_file_model/migration.sql new file mode 100644 index 000000000..5b6948088 --- /dev/null +++ b/packages/database/prisma/migrations/20240810090402_add_background_worker_file_model/migration.sql @@ -0,0 +1,24 @@ +-- AlterTable +ALTER TABLE "BackgroundWorkerTask" ADD COLUMN "fileId" TEXT; + +-- CreateTable +CREATE TABLE "BackgroundWorkerFile" ( + "id" TEXT NOT NULL, + "friendlyId" TEXT NOT NULL, + "filePath" TEXT NOT NULL, + "contentHash" TEXT NOT NULL, + "contents" BYTEA NOT NULL, + "backgroundWorkerId" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "BackgroundWorkerFile_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "BackgroundWorkerFile_friendlyId_key" ON "BackgroundWorkerFile"("friendlyId"); + +-- AddForeignKey +ALTER TABLE "BackgroundWorkerFile" ADD CONSTRAINT "BackgroundWorkerFile_backgroundWorkerId_fkey" FOREIGN KEY ("backgroundWorkerId") REFERENCES "BackgroundWorker"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "BackgroundWorkerTask" ADD CONSTRAINT "BackgroundWorkerTask_fileId_fkey" FOREIGN KEY ("fileId") REFERENCES "BackgroundWorkerFile"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/packages/database/prisma/migrations/20240811185335_improve_background_worker_file_model/migration.sql b/packages/database/prisma/migrations/20240811185335_improve_background_worker_file_model/migration.sql new file mode 100644 index 000000000..7a6248a18 --- /dev/null +++ b/packages/database/prisma/migrations/20240811185335_improve_background_worker_file_model/migration.sql @@ -0,0 +1,38 @@ +/* + Warnings: + + - You are about to drop the column `backgroundWorkerId` on the `BackgroundWorkerFile` table. All the data in the column will be lost. + - A unique constraint covering the columns `[projectId,contentHash]` on the table `BackgroundWorkerFile` will be added. If there are existing duplicate values, this will fail. + - Added the required column `projectId` to the `BackgroundWorkerFile` table without a default value. This is not possible if the table is not empty. + +*/ +-- DropForeignKey +ALTER TABLE "BackgroundWorkerFile" DROP CONSTRAINT "BackgroundWorkerFile_backgroundWorkerId_fkey"; + +-- AlterTable +ALTER TABLE "BackgroundWorkerFile" DROP COLUMN "backgroundWorkerId", +ADD COLUMN "projectId" TEXT NOT NULL; + +-- CreateTable +CREATE TABLE "_BackgroundWorkerToBackgroundWorkerFile" ( + "A" TEXT NOT NULL, + "B" TEXT NOT NULL +); + +-- CreateIndex +CREATE UNIQUE INDEX "_BackgroundWorkerToBackgroundWorkerFile_AB_unique" ON "_BackgroundWorkerToBackgroundWorkerFile"("A", "B"); + +-- CreateIndex +CREATE INDEX "_BackgroundWorkerToBackgroundWorkerFile_B_index" ON "_BackgroundWorkerToBackgroundWorkerFile"("B"); + +-- CreateIndex +CREATE UNIQUE INDEX "BackgroundWorkerFile_projectId_contentHash_key" ON "BackgroundWorkerFile"("projectId", "contentHash"); + +-- AddForeignKey +ALTER TABLE "BackgroundWorkerFile" ADD CONSTRAINT "BackgroundWorkerFile_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "_BackgroundWorkerToBackgroundWorkerFile" ADD CONSTRAINT "_BackgroundWorkerToBackgroundWorkerFile_A_fkey" FOREIGN KEY ("A") REFERENCES "BackgroundWorker"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "_BackgroundWorkerToBackgroundWorkerFile" ADD CONSTRAINT "_BackgroundWorkerToBackgroundWorkerFile_B_fkey" FOREIGN KEY ("B") REFERENCES "BackgroundWorkerFile"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 758b71eab..c8194a39e 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -464,6 +464,7 @@ model Project { alerts ProjectAlert[] alertStorages ProjectAlertStorage[] bulkActionGroups BulkActionGroup[] + BackgroundWorkerFile BackgroundWorkerFile[] } enum ProjectVersion { @@ -1566,6 +1567,7 @@ model BackgroundWorker { tasks BackgroundWorkerTask[] attempts TaskRunAttempt[] lockedRuns TaskRun[] + files BackgroundWorkerFile[] deployment WorkerDeployment? @@ -1574,6 +1576,27 @@ model BackgroundWorker { @@unique([projectId, runtimeEnvironmentId, version]) } +model BackgroundWorkerFile { + id String @id @default(cuid()) + + friendlyId String @unique + + filePath String + contentHash String + contents Bytes + + project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) + projectId String + + backgroundWorkers BackgroundWorker[] + + tasks BackgroundWorkerTask[] + + createdAt DateTime @default(now()) + + @@unique([projectId, contentHash]) +} + model BackgroundWorkerTask { id String @id @default(cuid()) slug String @@ -1589,6 +1612,9 @@ model BackgroundWorkerTask { project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade) projectId String + file BackgroundWorkerFile? @relation(fields: [fileId], references: [id], onDelete: Cascade, onUpdate: Cascade) + fileId String? + runtimeEnvironment RuntimeEnvironment @relation(fields: [runtimeEnvironmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) runtimeEnvironmentId String