Implement the disable job feature backend when indexing endpoints (#382)

Author: ericallam

.changeset/calm-crabs-kick.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/cli": patch
+---
+
+Added the send-event command

.changeset/honest-dingos-tap.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Support disabling jobs using the `enabled` flag

apps/webapp/app/services/endpoints/indexEndpoint.server.ts

@@ -7,10 +7,12 @@ import { logger } from "../logger.server";
 import { RegisterSourceService } from "../sources/registerSource.server";
 import { RegisterDynamicScheduleService } from "../triggers/registerDynamicSchedule.server";
 import { RegisterDynamicTriggerService } from "../triggers/registerDynamicTrigger.server";
+import { DisableJobService } from "../jobs/disableJob.server";
 
 export class IndexEndpointService {
   #prismaClient: PrismaClient;
   #registerJobService = new RegisterJobService();
+  #disableJobService = new DisableJobService();
   #registerSourceService = new RegisterSourceService();
   #registerDynamicTriggerService = new RegisterDynamicTriggerService();
   #registerDynamicScheduleService = new RegisterDynamicScheduleService();
@@ -57,23 +59,95 @@ export class IndexEndpointService {
       sources: 0,
       dynamicTriggers: 0,
       dynamicSchedules: 0,
+      disabledJobs: 0,
     };
 
+    const existingJobs = await this.#prismaClient.job.findMany({
+      where: {
+        projectId: endpoint.projectId,
+      },
+      include: {
+        aliases: {
+          where: {
+            name: "latest",
+            environmentId: endpoint.environmentId,
+          },
+          include: {
+            version: true,
+          },
+          take: 1,
+        },
+      },
+    });
+
     for (const job of jobs) {
       if (!job.enabled) {
-        continue;
+        const disabledJob = await this.#disableJobService
+          .call(endpoint, { slug: job.id, version: job.version })
+          .catch((error) => {
+            logger.error("Failed to disable job", {
+              endpointId: endpoint.id,
+              job,
+              error,
+            });
+
+            return;
+          });
+
+        if (disabledJob) {
+          indexStats.disabledJobs++;
+        }
+      } else {
+        try {
+          await this.#registerJobService.call(endpoint, job);
+
+          indexStats.jobs++;
+        } catch (error) {
+          logger.error("Failed to register job", {
+            endpointId: endpoint.id,
+            job,
+            error,
+          });
+        }
       }
+    }
 
-      try {
-        await this.#registerJobService.call(endpoint, job);
+    // TODO: we need to do this for sources, dynamic triggers, and dynamic schedules
+    const missingJobs = existingJobs.filter((job) => {
+      return !jobs.find((j) => j.id === job.slug);
+    });
 
-        indexStats.jobs++;
-      } catch (error) {
-        logger.error("Failed to register job", {
-          endpointId: endpoint.id,
-          job,
-          error,
-        });
+    if (missingJobs.length > 0) {
+      logger.debug("Disabling missing jobs", {
+        endpointId: endpoint.id,
+        missingJobIds: missingJobs.map((job) => job.slug),
+      });
+
+      for (const job of missingJobs) {
+        const latestVersion = job.aliases[0]?.version;
+
+        if (!latestVersion) {
+          continue;
+        }
+
+        const disabledJob = await this.#disableJobService
+          .call(endpoint, {
+            slug: job.slug,
+            version: latestVersion.version,
+          })
+          .catch((error) => {
+            logger.error("Failed to disable job", {
+              endpointId: endpoint.id,
+              job,
+              error,
+            });
+
+            return;
+          });
+
+        if (disabledJob) {
+          indexStats.disabledJobs++;
+        }
       }
     }
 

apps/webapp/app/services/jobs/disableJob.server.ts

@@ -0,0 +1,103 @@
+import type { JobVersion } from "@trigger.dev/database";
+import type { PrismaClient } from "~/db.server";
+import { prisma } from "~/db.server";
+import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
+import type { AuthenticatedEnvironment } from "../apiAuth.server";
+import { DisableScheduleSourceService } from "../schedules/disableScheduleSource.server";
+
+export type DisableJobServiceOptions = {
+  slug: string;
+  version: string;
+};
+
+export class DisableJobService {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    endpointIdOrEndpoint: string | ExtendedEndpoint,
+    options: DisableJobServiceOptions
+  ) {
+    const endpoint =
+      typeof endpointIdOrEndpoint === "string"
+        ? await findEndpoint(endpointIdOrEndpoint)
+        : endpointIdOrEndpoint;
+
+    return this.#disableJob(endpoint.environment, options);
+  }
+
+  async #disableJob(
+    environment: AuthenticatedEnvironment,
+    options: DisableJobServiceOptions
+  ): Promise<JobVersion | undefined> {
+    // Find the job
+    const job = await this.#prismaClient.job.findUnique({
+      where: {
+        projectId_slug: {
+          projectId: environment.projectId,
+          slug: options.slug,
+        },
+      },
+    });
+
+    if (!job) {
+      return;
+    }
+
+    const jobVersion = await this.#prismaClient.jobVersion.findUnique({
+      where: {
+        jobId_version_environmentId: {
+          jobId: job.id,
+          version: options.version,
+          environmentId: environment.id,
+        },
+      },
+    });
+
+    if (!jobVersion) {
+      return;
+    }
+
+    if (jobVersion.status === "DISABLED") {
+      return;
+    }
+
+    // Upsert the JobVersion
+    const updatedJobVersion = await this.#prismaClient.jobVersion.update({
+      where: {
+        id: jobVersion.id,
+      },
+      data: {
+        status: "DISABLED",
+      },
+    });
+
+    await this.#disableEventDispatcher(updatedJobVersion);
+
+    return updatedJobVersion;
+  }
+
+  async #disableEventDispatcher(jobVersion: JobVersion) {
+    const eventDispatcher = await this.#prismaClient.eventDispatcher.update({
+      where: {
+        dispatchableId_environmentId: {
+          dispatchableId: jobVersion.jobId,
+          environmentId: jobVersion.environmentId,
+        },
+      },
+      data: {
+        enabled: false,
+      },
+    });
+
+    const service = new DisableScheduleSourceService();
+
+    await service.call({
+      key: jobVersion.jobId,
+      dispatcher: eventDispatcher,
+    });
+  }
+}

apps/webapp/app/services/jobs/registerJob.server.ts

@@ -236,8 +236,10 @@ export class RegisterJobService {
         eventSpecification,
         preprocessRuns: metadata.preprocessRuns,
         startPosition: "LATEST",
+        status: "ACTIVE",
       },
       update: {
+        status: "ACTIVE",
         startPosition: "LATEST",
         eventSpecification,
         preprocessRuns: metadata.preprocessRuns,
@@ -401,6 +403,7 @@ export class RegisterJobService {
               type: "JOB_VERSION",
               id: jobVersion.id,
             },
+            enabled: true,
           },
         });
 
@@ -444,6 +447,7 @@ export class RegisterJobService {
               type: "JOB_VERSION",
               id: jobVersion.id,
             },
+            enabled: true,
           },
         });
 

apps/webapp/app/services/schedules/disableScheduleSource.server.ts

@@ -0,0 +1,44 @@
+import type { EventDispatcher } from "@trigger.dev/database";
+import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { workerQueue } from "../worker.server";
+
+export class DisableScheduleSourceService {
+  #prismaClient: PrismaClientOrTransaction;
+
+  constructor(prismaClient: PrismaClientOrTransaction = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call({ key, dispatcher }: { key: string; dispatcher: EventDispatcher }) {
+    const scheduleSourceExists = await this.#prismaClient.scheduleSource.findUnique({
+      where: {
+        key_environmentId: {
+          key,
+          environmentId: dispatcher.environmentId,
+        },
+      },
+    });
+
+    if (!scheduleSourceExists) {
+      return;
+    }
+
+    return await $transaction(this.#prismaClient, async (tx) => {
+      const scheduleSource = await this.#prismaClient.scheduleSource.update({
+        where: {
+          key_environmentId: {
+            key,
+            environmentId: dispatcher.environmentId,
+          },
+        },
+        data: {
+          active: false,
+        },
+      });
+
+      await workerQueue.dequeue(`scheduled:${scheduleSource.id}`, { tx });
+
+      return scheduleSource;
+    });
+  }
+}

apps/webapp/app/services/schedules/nextScheduledEvent.server.ts

@@ -56,7 +56,6 @@ export class NextScheduledEventService {
         },
         {
           runAt: scheduleTime,
-          queueName: `scheduler:${scheduleSource.environmentId}`,
           tx,
           jobKey: `scheduled:${scheduleSource.id}`,
         }

apps/webapp/app/services/schedules/registerScheduleSource.server.ts

@@ -70,6 +70,7 @@ export class RegisterScheduleSourceService {
           },
           metadata: schedule.metadata ?? {},
           externalAccountId: externalAccount ? externalAccount.id : undefined,
+          active: environment.autoEnableInternalSources,
         },
       });
 

apps/webapp/app/services/schedules/unregisterSchedule.server.ts

@@ -1,8 +1,5 @@
-import { RegisterScheduleBody } from "@trigger.dev/core";
-import { $transaction, PrismaClient } from "~/db.server";
-import { prisma } from "~/db.server";
+import { PrismaClient, prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "../apiAuth.server";
-import { RegisterScheduleSourceService } from "./registerScheduleSource.server";
 
 export class UnregisterScheduleService {
   #prismaClient: PrismaClient;

docs/documentation/guides/cli.mdx

@@ -149,3 +149,23 @@ yarn dlx @trigger.dev/cli@latest whoami
 ```
 
 </CodeGroup>
+
+## send-event Command
+
+The `send-event` command will send an event to your Trigger.dev project. This is useful for testing your Trigger.dev project locally.
+
+<CodeGroup>
+
+```bash npm
+npx @trigger.dev/cli@latest send-event -n "event.name" -p "{ \"key\": \"value\" }"
+```
+
+```bash pnpm
+pnpm dlx @trigger.dev/cli@latest send-event -n "event.name" -p "{ \"key\": \"value\" }"
+```
+
+```bash yarn
+yarn dlx @trigger.dev/cli@latest send-event -n "event.name" -p "{ \"key\": \"value\" }"
+```
+
+</CodeGroup>

examples/job-catalog/package.json

@@ -10,6 +10,7 @@
     "supabase": "nodemon --watch src/supabase.ts -r tsconfig-paths/register -r dotenv/config src/supabase.ts",
     "supabase:types": "npx supabase gen types typescript --project-id $SUPABASE_PROJECT_ID --schema public --schema auth --schema storage > src/supabase-types.ts",
     "events": "nodemon --watch src/events.ts -r tsconfig-paths/register -r dotenv/config src/events.ts",
+    "schedules": "nodemon --watch src/schedules.ts -r tsconfig-paths/register -r dotenv/config src/schedules.ts",
     "stressTest": "nodemon --watch src/stressTest.ts -r tsconfig-paths/register -r dotenv/config src/stressTest.ts",
     "delays": "nodemon --watch src/delays.ts -r tsconfig-paths/register -r dotenv/config src/delays.ts",
     "dev:trigger": "trigger-cli dev --port 8080"

examples/job-catalog/src/events.ts

@@ -13,6 +13,7 @@ client.defineJob({
   id: "event-example-1",
   name: "Event Example 1",
   version: "1.0.0",
+  enabled: true,
   trigger: eventTrigger({
     name: "event.example",
   }),